Skip to content

Commit f92f260

Browse files
antiguruclaude
andcommitted
compute: import published indexes as a shared arrangement
`ArrangementFlavor` gains a `SharedTrace` variant carrying arrangements backed by `TraceFrontier<SharedTraceHandle>`, and `import_index_shared` constructs it on the interactive runtime instead of reading the local `TraceManager`, which holds nothing there. Because the shared handle shares the `RowRow`/`Err` batch and cursor types of a maintenance `TraceAgent`, the variant flows through the same generic bodies as `Trace`, so a downstream `Get` receives a real arrangement with its key and permutation intact rather than a collection the plan has to re-derive. Joins, delta joins, thresholds, and hydration logging gain the corresponding arm. The import is a static snapshot at `as_of`, bounded one step past it rather than by `self.until`. Interactive work is single-time, so the capability must drop once the shared trace seals past `as_of` for the one-shot result to complete, and an unbounded `until` never gets there. An `as_of` at `Timestamp::MAX` yields the empty bound, which is the right reading of "the final state". The read hold is the returned `Arranged`'s own trace rather than a separate token, so consumers can downgrade it as their frontiers advance. A token nobody downgrades would be a floor under every hold that is downgraded, because the publisher forwards the meet. The two export paths reject `SharedTrace` as unreachable. Only the interactive runtime produces the variant, and its exports are freshly rendered query outputs, never a re-export of an import. Both `import_index_shared` and the variant are reachable only when a runtime holds the `Interactive` role, which nothing constructs yet. The two `sharing` tests that build this variant return here, along with the `consolidate_capture` helper they share. `render`'s tests move out of line to `render/tests.rs`, per the convention in `src/compute/AGENTS.md`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1 parent dffb179 commit f92f260

7 files changed

Lines changed: 1638 additions & 47 deletions

File tree

src/compute/src/render.rs

Lines changed: 273 additions & 11 deletions
Large diffs are not rendered by default.

src/compute/src/render/context.rs

Lines changed: 62 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@ use crate::extensions::reduce::MzReduce;
5252
use crate::render::columnar::CollectionEdge;
5353
use crate::render::errors::{DataflowErrorSer, ErrorLogger};
5454
use crate::render::{LinearJoinSpec, MaybeBucketByTime, RenderTimestamp};
55+
use crate::shared_trace::{SharedErrsEnter, SharedOksEnter};
5556
use crate::typedefs::{
5657
ErrAgent, ErrBatcher, ErrBuilder, ErrEnter, ErrSpine, RowRowAgent, RowRowEnter, RowRowSpine,
5758
};
@@ -231,6 +232,23 @@ pub enum ArrangementFlavor<'scope, T: RenderTimestamp> {
231232
Arranged<'scope, RowRowEnter<mz_repr::Timestamp, Diff, T>>,
232233
Arranged<'scope, ErrEnter<mz_repr::Timestamp, T>>,
233234
),
235+
/// A maintenance-runtime arrangement imported into the interactive runtime through the
236+
/// shared-trace primitive. Backed by `SharedTraceHandle`, so it is a real arrangement the plan
237+
/// can `Get`, not a re-derived collection. Only the interactive runtime produces this.
238+
///
239+
/// The `GlobalId` mirrors [`Self::Trace`]'s: it names the imported index. The interactive
240+
/// runtime never re-exports maintained indexes, so it is never consumed by an export path.
241+
///
242+
/// NOTE: no compile-time guard prevents an `until`-carrying multi-time dataflow from importing
243+
/// this variant. Unlike [`Self::Trace`], the imported handles carry no `TraceFrontier`/`until`
244+
/// bound, so safety rests on the Interactive=peek-only invariant (peeks are single-time and
245+
/// need no `until`). A future SUBSCRIBE-on-interactive migration must add the
246+
/// `TraceFrontier`/`until` bound here first.
247+
SharedTrace(
248+
GlobalId,
249+
Arranged<'scope, SharedOksEnter<T>>,
250+
Arranged<'scope, SharedErrsEnter<T>>,
251+
),
234252
}
235253

236254
impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
@@ -265,6 +283,10 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
265283
oks.clone().as_collection(logic),
266284
errs.clone().as_collection(|k, &()| k.clone()),
267285
),
286+
ArrangementFlavor::SharedTrace(_, oks, errs) => (
287+
oks.clone().as_collection(logic),
288+
errs.clone().as_collection(|k, &()| k.clone()),
289+
),
268290
}
269291
}
270292

@@ -348,6 +370,18 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
348370
let errs = errs.concat(mfp_errs.as_collection());
349371
(oks, errs)
350372
}
373+
ArrangementFlavor::SharedTrace(_, oks, errs) => {
374+
let (oks, mfp_errs) = CollectionBundle::<T>::flat_map_core_fallible::<_, _, DCB, _>(
375+
oks.clone(),
376+
key,
377+
max_demand,
378+
logic,
379+
REFUEL,
380+
);
381+
let errs = errs.clone().as_collection(|k, &()| k.clone());
382+
let errs = errs.concat(mfp_errs.as_collection());
383+
(oks, errs)
384+
}
351385
}
352386
}
353387

@@ -393,6 +427,17 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
393427
let errs = errs.clone().as_collection(|k, &()| k.clone());
394428
(oks, errs)
395429
}
430+
ArrangementFlavor::SharedTrace(_, oks, errs) => {
431+
let oks = CollectionBundle::<T>::flat_map_core_ok::<_, _, DCB, _>(
432+
oks.clone(),
433+
key,
434+
max_demand,
435+
logic,
436+
REFUEL,
437+
);
438+
let errs = errs.clone().as_collection(|k, &()| k.clone());
439+
(oks, errs)
440+
}
396441
}
397442
}
398443
}
@@ -402,6 +447,7 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
402447
match self {
403448
ArrangementFlavor::Local(oks, _errs) => oks.stream.scope(),
404449
ArrangementFlavor::Trace(_gid, oks, _errs) => oks.stream.scope(),
450+
ArrangementFlavor::SharedTrace(_gid, oks, _errs) => oks.stream.scope(),
405451
}
406452
}
407453

@@ -417,6 +463,11 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
417463
oks.clone().enter_region(region),
418464
errs.clone().enter_region(region),
419465
),
466+
ArrangementFlavor::SharedTrace(gid, oks, errs) => ArrangementFlavor::SharedTrace(
467+
*gid,
468+
oks.clone().enter_region(region),
469+
errs.clone().enter_region(region),
470+
),
420471
}
421472
}
422473
}
@@ -433,6 +484,11 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
433484
oks.clone().leave_region(outer),
434485
errs.clone().leave_region(outer),
435486
),
487+
ArrangementFlavor::SharedTrace(gid, oks, errs) => ArrangementFlavor::SharedTrace(
488+
*gid,
489+
oks.clone().leave_region(outer),
490+
errs.clone().leave_region(outer),
491+
),
436492
}
437493
}
438494
}
@@ -563,8 +619,9 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
563619
/// reads is the consumer's choice, and a delta join reads both within one operator, so a
564620
/// binding's definition cannot know which form to collapse.
565621
///
566-
/// NOTE: Leaves imported arrangements (`ArrangementFlavor::Trace`) alone, whose error traces
567-
/// this dataflow cannot rewrite in place. Their errors arrive bounded by the exporting
622+
/// NOTE: Leaves imported arrangements (`ArrangementFlavor::Trace` and
623+
/// `ArrangementFlavor::SharedTrace`) alone, whose error traces this dataflow cannot rewrite in
624+
/// place. Their errors arrive bounded by the exporting
568625
/// dataflow's last level of sharing rather than collapsed to one, since nothing collapses at an
569626
/// export. A global read more than once within one dataflow is not collapsed either, because
570627
/// only local bindings reach this.
@@ -580,7 +637,9 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
580637
let name = format!("Distinct errors[{key:?}]");
581638
ArrangementFlavor::Local(oks, distinct_arranged_errs(errs, &name))
582639
}
583-
flavor @ ArrangementFlavor::Trace(..) => flavor,
640+
flavor @ (ArrangementFlavor::Trace(..) | ArrangementFlavor::SharedTrace(..)) => {
641+
flavor
642+
}
584643
};
585644
self.arranged.insert(key, flavor);
586645
}

src/compute/src/render/join/delta_join.rs

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ use timely::progress::Antichain;
4141
use crate::render::RenderTimestamp;
4242
use crate::render::context::{ArrangementFlavor, CollectionBundle, Context};
4343
use crate::render::errors::DataflowErrorSer;
44+
use crate::shared_trace::SharedOksEnter;
4445
use crate::typedefs::{RowRowAgent, RowRowEnter};
4546

4647
impl<'scope, T: RenderTimestamp> Context<'scope, T> {
@@ -319,6 +320,9 @@ fn bundle_errs<'scope, T: RenderTimestamp>(
319320
ArrangementFlavor::Trace(_id, _oks, errs) => {
320321
errs.clone().as_collection(|k, _v| k.clone())
321322
}
323+
ArrangementFlavor::SharedTrace(_id, _oks, errs) => {
324+
errs.clone().as_collection(|k, _v| k.clone())
325+
}
322326
};
323327
collected.push(errs);
324328
}
@@ -400,6 +404,32 @@ where
400404
};
401405
(oks, errs2)
402406
}
407+
// A maintenance index the interactive runtime imported from the sharing registry. Same
408+
// shape as `Trace`, only the trace type differs (`SharedOksEnter`).
409+
Some(ArrangementFlavor::SharedTrace(_, oks, errs)) => {
410+
let (oks, errs2) = if source_precedes_lookup {
411+
build_halfjoin_trace::<_, SharedOksEnter<_>, _>(
412+
updates,
413+
oks,
414+
prev_key,
415+
prev_thinning,
416+
|t1, t2| t1.le(t2),
417+
closure,
418+
config_set,
419+
)
420+
} else {
421+
build_halfjoin_trace::<_, SharedOksEnter<_>, _>(
422+
updates,
423+
oks,
424+
prev_key,
425+
prev_thinning,
426+
|t1, t2| t1.lt(t2),
427+
closure,
428+
config_set,
429+
)
430+
};
431+
(oks, errs2.concat(errs.as_collection(|k, _v| k.clone())))
432+
}
403433
None => panic!("Arrangement promised by the planner is absent!"),
404434
}
405435
}
@@ -743,6 +773,17 @@ where
743773
initial_closure,
744774
)
745775
}
776+
// A maintenance index the interactive runtime imported from the sharing registry. Same
777+
// shape as `Trace`, only the trace type differs (`SharedOksEnter`).
778+
Some(ArrangementFlavor::SharedTrace(_, oks, errs)) => {
779+
let (oks, errs2) = build_update_stream_trace::<_, SharedOksEnter<_>>(
780+
oks,
781+
as_of,
782+
source_relation,
783+
initial_closure,
784+
);
785+
(oks, errs2.concat(errs.as_collection(|k, _v| k.clone())))
786+
}
746787
None => panic!("Arrangement promised by the planner is absent!"),
747788
}
748789
}

src/compute/src/render/join/linear_join.rs

Lines changed: 44 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ use crate::render::RenderTimestamp;
4141
use crate::render::context::{ArrangementFlavor, CollectionBundle, Context};
4242
use crate::render::errors::DataflowErrorSer;
4343
use crate::render::join::mz_join_core::mz_join_core;
44+
use crate::shared_trace::SharedOksEnter;
4445
use crate::typedefs::{RowRowAgent, RowRowEnter};
4546
use mz_row_spine::{RowRowBuilder, RowRowColPagedBuilder, RowRowSpine};
4647

@@ -195,6 +196,8 @@ enum JoinedFlavor<'scope, T: RenderTimestamp> {
195196
Local(Arranged<'scope, RowRowAgent<T, Diff>>),
196197
/// An imported arrangement.
197198
Trace(Arranged<'scope, RowRowEnter<mz_repr::Timestamp, Diff, T>>),
199+
/// A shared-trace arrangement imported into the interactive runtime.
200+
SharedTrace(Arranged<'scope, SharedOksEnter<T>>),
198201
}
199202

200203
impl<'scope, T> Context<'scope, T>
@@ -238,6 +241,10 @@ where
238241
errors.push(errs.as_collection(|k, _v| k.clone()).enter_region(inner));
239242
JoinedFlavor::Trace(oks.enter_region(inner))
240243
}
244+
(Some(ArrangementFlavor::SharedTrace(_gid, oks, errs)), None) => {
245+
errors.push(errs.as_collection(|k, _v| k.clone()).enter_region(inner));
246+
JoinedFlavor::SharedTrace(oks.enter_region(inner))
247+
}
241248
(_, initial_closure) => {
242249
// TODO: extract closure from the first stage in the join plan, should it exist.
243250
// TODO: apply that closure in `flat_map_ref` rather than calling `.collection`.
@@ -415,52 +422,57 @@ where
415422
.arrangement(&lookup_key[..])
416423
.expect("Arrangement absent despite explicit construction");
417424

425+
// The nine `(stream flavor) x (lookup flavor)` combinations differ only in the two trace
426+
// types handed to the generic `differential_join_inner` and the two arrangement values
427+
// consumed. This local macro spells one combination. The `SharedTrace` rows exist so an
428+
// interactive-runtime join over imported indexes type-checks. At runtime a dataflow's
429+
// arrangements are all one runtime's flavor, so the mixed rows never fire, but exhaustive
430+
// matching requires them.
431+
macro_rules! join {
432+
($stream:expr, $stream_tr:ty, $lookup:expr, $lookup_tr:ty, $errs1:expr) => {{
433+
let (oks, errs2) = self
434+
.differential_join_inner::<$stream_tr, $lookup_tr>($stream, $lookup, closure);
435+
errors.push($errs1.as_collection(|k, _v| k.clone()));
436+
errors.extend(errs2);
437+
oks
438+
}};
439+
}
440+
418441
match joined {
419442
JoinedFlavor::Collection(_) => {
420443
unreachable!("JoinedFlavor::VecCollection variant avoided at top of method");
421444
}
422445
JoinedFlavor::Local(local) => match arrangement {
423446
ArrangementFlavor::Local(oks, errs1) => {
424-
let (oks, errs2) = self
425-
.differential_join_inner::<RowRowAgent<_, _>, RowRowAgent<_, _>>(
426-
local, oks, closure,
427-
);
428-
429-
errors.push(errs1.as_collection(|k, _v| k.clone()));
430-
errors.extend(errs2);
431-
oks
447+
join!(local, RowRowAgent<_, _>, oks, RowRowAgent<_, _>, errs1)
432448
}
433449
ArrangementFlavor::Trace(_gid, oks, errs1) => {
434-
let (oks, errs2) = self
435-
.differential_join_inner::<RowRowAgent<_, _>, RowRowEnter<_, _, _>>(
436-
local, oks, closure,
437-
);
438-
439-
errors.push(errs1.as_collection(|k, _v| k.clone()));
440-
errors.extend(errs2);
441-
oks
450+
join!(local, RowRowAgent<_, _>, oks, RowRowEnter<_, _, _>, errs1)
451+
}
452+
ArrangementFlavor::SharedTrace(_gid, oks, errs1) => {
453+
join!(local, RowRowAgent<_, _>, oks, SharedOksEnter<_>, errs1)
442454
}
443455
},
444456
JoinedFlavor::Trace(trace) => match arrangement {
445457
ArrangementFlavor::Local(oks, errs1) => {
446-
let (oks, errs2) = self
447-
.differential_join_inner::<RowRowEnter<_, _, _>, RowRowAgent<_, _>>(
448-
trace, oks, closure,
449-
);
450-
451-
errors.push(errs1.as_collection(|k, _v| k.clone()));
452-
errors.extend(errs2);
453-
oks
458+
join!(trace, RowRowEnter<_, _, _>, oks, RowRowAgent<_, _>, errs1)
459+
}
460+
ArrangementFlavor::Trace(_gid, oks, errs1) => {
461+
join!(trace, RowRowEnter<_, _, _>, oks, RowRowEnter<_, _, _>, errs1)
462+
}
463+
ArrangementFlavor::SharedTrace(_gid, oks, errs1) => {
464+
join!(trace, RowRowEnter<_, _, _>, oks, SharedOksEnter<_>, errs1)
465+
}
466+
},
467+
JoinedFlavor::SharedTrace(trace) => match arrangement {
468+
ArrangementFlavor::Local(oks, errs1) => {
469+
join!(trace, SharedOksEnter<_>, oks, RowRowAgent<_, _>, errs1)
454470
}
455471
ArrangementFlavor::Trace(_gid, oks, errs1) => {
456-
let (oks, errs2) = self
457-
.differential_join_inner::<RowRowEnter<_, _, _>, RowRowEnter<_, _, _>>(
458-
trace, oks, closure,
459-
);
460-
461-
errors.push(errs1.as_collection(|k, _v| k.clone()));
462-
errors.extend(errs2);
463-
oks
472+
join!(trace, SharedOksEnter<_>, oks, RowRowEnter<_, _, _>, errs1)
473+
}
474+
ArrangementFlavor::SharedTrace(_gid, oks, errs1) => {
475+
join!(trace, SharedOksEnter<_>, oks, SharedOksEnter<_>, errs1)
464476
}
465477
},
466478
}

0 commit comments

Comments
 (0)