compute: import published indexes as a shared arrangement

[antiguru]

Fifth of eight PRs splitting #37770. Stacks on #38389. Tracked by CPU-215.

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. The shared handle shares the RowRow/Err batch and cursor types of a maintenance TraceAgent, so the variant flows through the same generic bodies as Trace and a downstream Get receives a real arrangement with its key and permutation intact, rather than a collection the plan re-derives. 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.

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, since the publisher forwards the meet.

The variant and its constructor land together because a never-constructed variant of an enum inside the private render module trips dead_code, and CI lints with -D warnings. Reachable only when a runtime holds the Interactive role, which nothing constructs yet.

[def-]

QA LLM Review

1. MEDIUM -- Delta join's SharedTrace arms re-emit input errors that bundle_errs already seeded

src/compute/src/render/join/delta_join.rs:431 (and :785)

Details

The two new SharedTrace arms concatenate the input arrangement's own error collection into the errors they return, but render_delta_join already seeds inner_errs once per input from bundle_errs, which this PR extended with a SharedTrace arm too. An error carried by an imported shared index is therefore emitted 1 + (number of delta paths that read it) times instead of once, which is exactly the per-path multiplication the surrounding code is written to avoid.

The sibling arms deliberately drop the bundle's errors: Some(ArrangementFlavor::Local(oks, _errs)) at :359 and Some(ArrangementFlavor::Trace(_, oks, _errs)) at :383, likewise at :760/:768 in build_update_stream. The contract is stated on both helpers ("The returned error collection holds only the errors this stage produces. The errors bundle already carries are the caller's to propagate, once") and on bundle_errs itself at :294, which spells out the consequence: "Propagating them per path would multiply their multiplicities by the number of paths, and because a join's output is another join's input, those factors compound multiplicatively through a nested plan until the Diff overflows."

bundle_errs gained the matching arm at :323, so both sources are live for this flavor. Concretely, a three-way delta join over imported indexes on the interactive runtime reads each input as one path's source and two paths' lookup, so each input's pre-existing errors land in the output four times; the inflation is visible in mz_compute_error_counts for an index export and compounds through nested joins.

Fix: bind the errors away, as the Trace arms do.

-        Some(ArrangementFlavor::SharedTrace(_, oks, errs)) => {
+        Some(ArrangementFlavor::SharedTrace(_, oks, _errs)) => {
             let (oks, errs2) = if source_precedes_lookup {
...
-            (oks, errs2.concat(errs.as_collection(|k, _v| k.clone())))
+            (oks, errs2)
         }

and the same in build_update_stream:

-        Some(ArrangementFlavor::SharedTrace(_, oks, errs)) => {
-            let (oks, errs2) = build_update_stream_trace::<_, SharedOksEnter<_>>(
-                oks,
-                as_of,
-                source_relation,
-                initial_closure,
-            );
-            (oks, errs2.concat(errs.as_collection(|k, _v| k.clone())))
-        }
+        Some(ArrangementFlavor::SharedTrace(_, oks, _errs)) => {
+            build_update_stream_trace::<_, SharedOksEnter<_>>(
+                oks,
+                as_of,
+                source_relation,
+                initial_closure,
+            )
+        }

2. LOW -- import_shared_index's contract is attached to report_compacted_past

src/compute/src/render.rs:563

The doc comment block that opens at :563 runs straight into report_compacted_past's own summary at :605 without a break, so the entire import contract, including the "caller MUST retain [the slot] for as long as the import is alive" requirement, renders as documentation for the panic helper, and import_shared_index itself is left undocumented.

Details

Split the block so that /// Reports a publication point refusing to serve as_of, and aborts. starts report_compacted_past's own comment and the preceding paragraphs move onto import_shared_index.

Related: the NOTE: on the new variant at src/compute/src/render/context.rs:244 says "the imported handles carry no TraceFrontier/until bound". They do: SharedOksEnter<T> = TraceEnter<TraceFrontier<SharedOksHandle>, T>, structurally identical to RowRowEnter, and import_index_shared passes a real until (as_of stepped forward). The accurate caveat is that the bound is derived from as_of rather than from the dataflow's self.until, which is what a multi-time import would need.

[antiguru]

All three fixed in 6807118. The delta join's SharedTrace arms bind the input errors away as the Trace arms do, so bundle_errs remains the one place that propagates them. The import contract now sits on import_shared_index, and report_compacted_past has its own doc. The NOTE on the variant says what is true: the bound is derived from as_of rather than from until, so the variant serves single-time dataflows only.

Posted by Claude Code.

[def-]

QA LLM Review

1. MEDIUM -- interactive_import_holds_after_construction passes with the import's read hold deleted

src/compute/src/render/tests.rs:406

Both assertions in this test are satisfied by the standing hold alone, so the test passes unchanged when the import registers no read hold at all. I verified this by deleting the hold import_snapshot_at owns and re-running: all 35 tests across render::tests, sharing::tests and shared_trace::tests still pass. The regression the test names, a publisher compacting past a live import's as_of and feeding silently wrong rows to every consumer that keeps only the stream, currently has no coverage anywhere in the crate.

Details

SharedTrace::new_empty seeds logical_compaction with the standing hold's own contribution (MutableAntichain::from_elem(minimum), shared_trace/state.rs:208), and adopt advances it to the writer's frontier at adoption, which is {0} here. This test never calls note_standing_hold, so 0 stays in the accumulation, and published_logical_holds returns logical_compaction.frontier(), that is {0}, whether or not the import ever registered its own hold at as_of = {5}. Both !holds.is_empty() and less_equal(holds, as_of) are therefore true either way.

The sibling tests do not cover it either: interactive_import_hold_releases_on_drop and interactive_import_hold_downgrades_while_live retain the returned Arranged's trace, which registers a hold of its own, so they pass with the operator-owned hold gone. That hold is what protects the as_collection and reduce consumers, which drop the trace at build time, so it is exactly the path with no test.

Reproduction: patch let mut hold = Some(self.clone()); (shared_trace/handle.rs:293) to None, then cargo nextest run -p mz-compute --lib 'render::tests::' 'sharing::tests::' 'shared_trace::' → 35 passed.

Advancing the standing hold past as_of first leaves the import's own hold as the only thing the accumulation can report:

// The standing hold sits at the minimum and is itself one of the accumulated holds, so
// advance it past `as_of` first.
registry.note_standing_hold(id, 0, &Antichain::from_elem(Timestamp::from(10_u64)));
let holds = registry
    .published_logical_holds(&id, 0)
    .expect("still published");

With that line the test fails on the patched build (the import's hold must not have released past its own as_of: Antichain { elements: [10] }) and passes on this head; I ran both. Published::physical_holds() is an alternative measure, since it starts empty and only readers contribute to it.

[antiguru]

Macro is fine; but could the also be a vcall instead?

[antiguru]

Not without a branch in the per-record loop. TraceReader and Cursor carry GATs (Key<'a>, Val<'a>, Cursor<'a>), so neither is dyn-compatible. The alternative is an enum over the three trace types plus an enum over their cursors, with a match in every key, val, step_key, and so on, which is a dispatch per cursor operation inside the join's inner loop. The macro costs monomorphisation only, three new combinations of the nine. Keeping it.

Posted by Claude Code.

[antiguru]

I don't follow the reasoning: The bound should always be derived from the as-of, as we're interested in the data in (as_of, until[, right? What implies that this can only serve single-time dataflows?

[antiguru]

You are right, and the reasoning was wrong. The maintenance import passes (as_of, until) to import_frontier_core and follows the upper when until is empty. Nothing about the shared import needs a different bound, and the single-time restriction lives in the router, which admits only is_single_time() dataflows. So snapshot_until equalled self.until for every dataflow that reached this import, and the hardcoding bought nothing while forbidding a long-lived dependent later.

import_index_shared now passes self.until. The NOTE on the variant, the "serves single-time dataflows only" sentence, and the try_step_forward special case for Timestamp::MAX are gone. is_single_time() at MAX yields an empty until, so the import follows the trace to its terminal frontier there, the same as the maintenance import.

Posted by Claude Code.

[antiguru]

We have fn import_index_shared, and fn import_shared_index, which can be misread. Do we have more than one use of the import_shared_index free function? It's fine to keep if this is the cleanest, and the free function isn't pub.

[antiguru]

One production caller (import_index_shared) and the render tests. It is not pub. Renamed the free function to import_published_index, which is the registry's vocabulary and no longer reads as a permutation of the method's name.

Posted by Claude Code.

[antiguru]

I think this encodes an unwritten invariant: A peek against a re-export dataflow would be totally valid right now, but now wouldn't be anymore. I do not think the optimizer guarantees that it doesn't do no-op re-exports. What'd be the cost of re-exporting a shared trace?

[antiguru]

Agreed that nothing guaranteed it. Both unreachable! arms, here and in export_index_iterative, are now the shared-arrangement analogue of the Trace arm: alias_shared_reexport registers idx_id as an alias of gid's publication point through publish_alias, so a peek on idx_id reads the imported arrangement. Only import_published_index creates slots and a dataflow routed here imports no transient id, so no reader can have created idx_id's slot first, and the alias always registers, which is asserted.

The one thing the Trace arm has that this cannot is a TraceBundle to install, and report_frontiers reads the write frontier from it. The alias has no trace of its own, so report_frontiers falls back to the published point's upper through a new ArrangementSharingRegistry::published_upper, which is the same frontier the maintenance re-export reports. Everything else already worked: TraceManager::allow_compaction is a no-op on a missing id, Aliases::note defers to the target while it lives, and drop_collection removes the alias.

Cost of the re-export itself is a map insert. A protocol invariant would have to mirror what ArrangeBy does over the imported Get, and any simpler rule over-rejects a filter followed by a re-arrange under an import's key, so I did not pursue it.

#38393 gains two_runtime_reexport_interactive.spec, a query dataflow whose export is the arrangement it imports, which passes locally along with the four existing two-runtime specs.

Posted by Claude Code.