fixup: bound the shared import by the dataflow's until and alias re-exports

The shared import synthesised its own bound one step past `as_of` instead of
passing the dataflow's `until`, and documented the result as serving
single-time dataflows only. The maintenance import passes `(as_of, until)` and
follows the trace when `until` is empty, and nothing about the shared import
needs a different bound. The router already sends only single-time dataflows to
the interactive runtime, so the two bounds were equal for every dataflow that
reached the import, and the narrowing bought nothing. The variant's single-time
note and the `Timestamp::MAX` special case go with it.

An export whose arrangement is an imported shared arrangement was
`unreachable!`. Nothing in the optimizer promises that a peek dataflow never
re-exports its import under the export key, and a panic on either runtime aborts
the process. Both export paths now alias the export's id to the imported
publication point, the shared analogue of the `Trace` arm. The alias has no
`TraceBundle`, so `report_frontiers` reads the frontier through the registry.

`import_shared_index` is renamed `import_published_index` so it no longer reads
as a permutation of the `import_index_shared` method that calls it.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VDm7opomJLxbNUEP3r9BLk

Author: antiguru

src/compute/src/compute_state.rs

@@ -1131,6 +1131,14 @@ impl<'a> ActiveComputeState<'a> {
                 traces.oks_mut().read_upper(&mut new_frontier);
             } else if let Some(frontier) = &collection.sink_write_frontier {
                 new_frontier.clone_from(&frontier.borrow());
+            } else if let Some(upper) = self
+                .compute_state
+                .sharing_registry
+                .published_upper(&id, self.timely_worker.index())
+            {
+                // An index that re-exports an imported shared arrangement has no trace of its own,
+                // only an alias to the published point. See `export_index`.
+                new_frontier.clone_from(&upper);
             } else {
                 error!(id = ?id, "collection without write frontier");
                 continue;

src/compute/src/render.rs

@@ -598,7 +598,7 @@ fn report_compacted_past(
 /// keeps the trace can downgrade it and the publisher compacts behind a long-lived import.
 ///
 /// Panics if the point's `since` is already beyond `as_of`, see [`report_compacted_past`].
-fn import_shared_index<'outer>(
+fn import_published_index<'outer>(
     outer: Scope<'outer, mz_repr::Timestamp>,
     registry: &ArrangementSharingRegistry,
     idx_id: GlobalId,
@@ -788,8 +788,7 @@ where
     ///
     /// Imports the published index as an arrangement, [`ArrangementFlavor::SharedTrace`], keyed and
     /// permuted as the plan expects, so a `Get` of `idx.on_id` and the joins and reduces below it
-    /// consume an arrangement rather than re-deriving one. The import is a snapshot at `as_of`, so
-    /// it serves single-time dataflows only.
+    /// consume an arrangement rather than re-deriving one.
     fn import_index_shared<'outer>(
         &mut self,
         outer: Scope<'outer, mz_repr::Timestamp>,
@@ -801,26 +800,13 @@ where
         start_signal: StartSignal,
     ) {
         let name = format!("Index({}, {:?})", idx.on_id, idx.key);
-        // Bound the snapshot to the single read time `as_of`. Interactive work is single-time, so the
-        // import's capability must drop once the shared trace seals past `as_of`, letting the one-shot
-        // result complete. `self.until` may be empty (unbounded) for a long-lived dependency, which a
-        // live `upper` never reaches, so it cannot serve as the snapshot bound.
-        //
-        // `try_step_forward` yields the frontier strictly greater than `as_of`. For an `as_of` at
-        // `Timestamp::MAX` there is no such finite time, so the element drops out and the bound is the
-        // empty (end-of-time) frontier, matching the semantics of "read the final state".
-        let snapshot_until = Antichain::from_iter(
-            self.as_of_frontier
-                .iter()
-                .filter_map(|t| t.try_step_forward()),
-        );
-        let (mut oks_arranged, errs_arranged, slot) = import_shared_index(
+        let (mut oks_arranged, errs_arranged, slot) = import_published_index(
             outer,
             &compute_state.sharing_registry,
             idx_id,
             &name,
             &self.as_of_frontier,
-            &snapshot_until,
+            &self.until,
         );
 
         // Attach the input probe to the replayed batch stream so hydration tracking observes it,
@@ -846,7 +832,7 @@ where
 
         // The slot Arc's strong count marks a live reader, so it must outlive the dataflow. The read
         // hold is not in here: it lives in the `Arranged`s the bundle above retains, so that a
-        // consumer can downgrade it. See `import_shared_index`.
+        // consumer can downgrade it. See `import_published_index`.
         tokens.insert(idx_id, Rc::new(slot));
     }
 }
@@ -931,15 +917,8 @@ impl<'g> Context<'g, mz_repr::Timestamp> {
                 }
                 compute_state.traces.set(idx_id, trace);
             }
-            Some(ArrangementFlavor::SharedTrace(..)) => {
-                // Only the interactive runtime produces `SharedTrace`, and only for imports it reads
-                // from the sharing registry. Its exports are transient query outputs, which are
-                // freshly rendered `Local` arrangements (a join/reduce output), never a direct
-                // re-export of an imported shared arrangement. The maintenance runtime's imports are
-                // `Local`/`Trace`. So an export can never observe a `SharedTrace` input.
-                unreachable!(
-                    "interactive runtime does not re-export an imported shared arrangement"
-                );
+            Some(ArrangementFlavor::SharedTrace(gid, _, _)) => {
+                alias_shared_reexport(compute_state, &self.scope, idx_id, gid);
             }
             None => {
                 println!("collection available: {:?}", bundle.collection.is_none());
@@ -1052,13 +1031,8 @@ where
                 }
                 compute_state.traces.set(idx_id, trace);
             }
-            Some(ArrangementFlavor::SharedTrace(..)) => {
-                // See `export_index`: only the interactive runtime produces `SharedTrace`, and its
-                // exports are freshly rendered `Local` query outputs, never a re-export of an
-                // imported shared arrangement, so an export can never observe this variant.
-                unreachable!(
-                    "interactive runtime does not re-export an imported shared arrangement"
-                );
+            Some(ArrangementFlavor::SharedTrace(gid, _, _)) => {
+                alias_shared_reexport(compute_state, &outer, idx_id, gid);
             }
             None => {
                 println!("collection available: {:?}", bundle.collection.is_none());
@@ -1102,6 +1076,30 @@ fn publish_reexport<'scope>(
     registry.publish(idx_id, &oks, &errs);
 }
 
+/// Publishes `idx_id` as an alias of `gid`'s publication point, for an export whose arrangement is
+/// an imported shared arrangement. The shared-arrangement analogue of [`publish_reexport`].
+///
+/// There is no `TraceBundle` to install for such an export, so `report_frontiers` reads its
+/// frontier through the registry instead.
+fn alias_shared_reexport<'scope, T: timely::progress::Timestamp>(
+    compute_state: &ComputeState,
+    scope: &Scope<'scope, T>,
+    idx_id: GlobalId,
+    gid: GlobalId,
+) {
+    // Only `import_published_index` creates slots, and a dataflow rendered here imports no transient
+    // id (see the multiplexer's routing), so no reader can have created `idx_id`'s slot ahead of
+    // this publisher and the alias always registers.
+    let aliased =
+        compute_state
+            .sharing_registry
+            .publish_alias(idx_id, gid, scope.index(), scope.peers());
+    assert!(
+        aliased,
+        "re-export {idx_id} of shared arrangement {gid} found a reader-created slot"
+    );
+}
+
 /// Information about bindings, tracked in `render_recursive_plan` and
 /// `render_plan`, to be passed to `render_letfree_plan`.
 ///

src/compute/src/render/context.rs

@@ -238,12 +238,8 @@ pub enum ArrangementFlavor<'scope, T: RenderTimestamp> {
     /// shared-trace primitive. Backed by `SharedTraceHandle`, so it is a real arrangement the plan
     /// can `Get`, not a re-derived collection. Only the interactive runtime produces this.
     ///
-    /// The `GlobalId` mirrors [`Self::Trace`]'s: it names the imported index. The interactive
-    /// runtime never re-exports maintained indexes, so it is never consumed by an export path.
-    ///
-    /// NOTE: the `TraceFrontier` bound on the imported handles is derived from the dataflow's
-    /// `as_of`, not from its `until`, so this variant serves single-time dataflows only. Nothing
-    /// checks that at the type level.
+    /// The `GlobalId` mirrors [`Self::Trace`]'s: it names the imported index, so an export of this
+    /// same arrangement can alias it instead of arranging again.
     SharedTrace(
         GlobalId,
         Arranged<'scope, SharedOksEnter<T>>,

src/compute/src/render/tests.rs

@@ -26,7 +26,7 @@ use crate::shared_trace::SharedOksFrontier;
 use crate::sharing::ArrangementSharingRegistry;
 use crate::typedefs::{ErrBatcher, ErrBuilder, ErrSpine, RowRowAgent, RowRowSpine};
 
-use super::import_shared_index;
+use super::import_published_index;
 use crate::server::ComputeRuntimeRole;
 
 fn test_rows() -> Vec<(Row, Row)> {
@@ -112,7 +112,7 @@ fn interactive_import_replays_rows_and_holds_at_as_of() {
         let probe = ProbeHandle::new();
         let (mut oks_trace, mut errs_trace) = worker.dataflow::<Timestamp, _, _>(|scope| {
             // `until` empty: no upper suppression, so the whole snapshot at `as_of` flows.
-            let (oks_arranged, errs_arranged, _slot) = import_shared_index(
+            let (oks_arranged, errs_arranged, _slot) = import_published_index(
                 scope.clone(),
                 &registry_in,
                 id,
@@ -236,7 +236,7 @@ fn tick(
 /// Mirrors the differential-dataflow primitive's own `import_hold_pins_then_releases`
 /// (`differential-dataflow/tests/sharing.rs`), which demonstrates the identical pin-then-release
 /// contract one layer down, directly on a bare `SharedTraceHandle` with no compute-level
-/// wrapping. This test drives the same `import_shared_index` primitive that
+/// wrapping. This test drives the same `import_published_index` primitive that
 /// `import_index_shared` calls in production, rather than re-deriving the contract from
 /// scratch.
 ///
@@ -272,7 +272,7 @@ fn interactive_import_hold_releases_on_drop() {
         // lets a consumer downgrade the hold as its frontier advances. The `stream`s are dropped,
         // as a consumer that only needs the trace would.
         let (oks_trace, errs_trace) = worker.dataflow::<Timestamp, _, _>(|scope| {
-            let (oks_arranged, errs_arranged, _slot) = import_shared_index(
+            let (oks_arranged, errs_arranged, _slot) = import_published_index(
                 scope.clone(),
                 &registry,
                 id,
@@ -368,7 +368,7 @@ fn interactive_import_holds_after_construction() {
         // handle it produced go out of scope with the builder, exactly as production does.
         let probe = ProbeHandle::new();
         worker.dataflow::<Timestamp, _, _>(|scope| {
-            let (oks_arranged, _errs_arranged, _slot) = import_shared_index(
+            let (oks_arranged, _errs_arranged, _slot) = import_published_index(
                 scope.clone(),
                 &registry,
                 id,
@@ -436,7 +436,7 @@ fn published_since_does_not_chase_reader_holds() {
         // A reader at the higher as_of. Its handles go out of scope with the builder; the
         // import operator's own hold remains.
         worker.dataflow::<Timestamp, _, _>(|scope| {
-            let (_o, _e, _slot) = import_shared_index(
+            let (_o, _e, _slot) = import_published_index(
                 scope.clone(),
                 &registry,
                 id,
@@ -501,7 +501,7 @@ fn import_reports_physical_within_chain_coverage() {
         );
 
         let mut trace = worker.dataflow::<Timestamp, _, _>(|scope| {
-            let (oks_arranged, _e, _slot) = import_shared_index(
+            let (oks_arranged, _e, _slot) = import_published_index(
                 scope.clone(),
                 &registry,
                 id,
@@ -553,7 +553,7 @@ fn interactive_import_hold_downgrades_while_live() {
             });
 
         let (mut oks_trace, mut errs_trace) = worker.dataflow::<Timestamp, _, _>(|scope| {
-            let (oks_arranged, errs_arranged, _slot) = import_shared_index(
+            let (oks_arranged, errs_arranged, _slot) = import_published_index(
                 scope.clone(),
                 &registry,
                 id,
@@ -623,7 +623,7 @@ fn interactive_import_hold_downgrades_while_live() {
 }
 
 /// A published slot's `since` may already sit above the dataflow's requested `as_of` if the
-/// controller offered an unreadable `as_of`, a protocol error: `import_shared_index` must
+/// controller offered an unreadable `as_of`, a protocol error: `import_published_index` must
 /// panic rather than let the read silently see coalesced data, mirroring the maintenance
 /// path's `compaction_frontier` assert in `import_index`.
 ///
@@ -663,7 +663,7 @@ fn import_asserts_since_at_most_as_of() {
 
         // Importing at `as_of` now finds a `since` already beyond it: the assert must panic.
         worker.dataflow::<Timestamp, _, _>(|scope| {
-            let _ = import_shared_index(
+            let _ = import_published_index(
                 scope.clone(),
                 &registry,
                 id,
@@ -720,7 +720,7 @@ fn standing_hold_pins_until_the_importing_runtime_applies() {
         // The queued create is now applied. It must import, and the rows it reads at `as_of` must
         // be the ones a read at `as_of` should see rather than a coalesced history.
         let (oks_trace, errs_trace) = worker.dataflow::<Timestamp, _, _>(|scope| {
-            let (oks_arranged, errs_arranged, _slot) = import_shared_index(
+            let (oks_arranged, errs_arranged, _slot) = import_published_index(
                 scope.clone(),
                 &registry,
                 id,

src/compute/src/shared_trace/publish.rs

@@ -122,6 +122,12 @@ where
         self.shared.upper()
     }
 
+    /// The published seal frontier.
+    pub(crate) fn upper(&self) -> Antichain<Tr::Time> {
+        let state = self.shared.state.lock().expect("shared trace poisoned");
+        state.upper.clone()
+    }
+
     /// Why this point would refuse an `as_of`. See [`Diagnostics`].
     pub(crate) fn diagnostics(&self) -> Diagnostics<Tr::Time> {
         Diagnostics {

src/compute/src/shared_trace/tests.rs

@@ -1038,7 +1038,7 @@ fn live_import_does_not_pin_merging() {
             );
             drop(arranged.trace);
         });
-        // Drop the minting handle, as `crate::render::import_shared_index` does: the import owns
+        // Drop the minting handle, as `crate::render::import_published_index` does: the import owns
         // its own clone, and a live mint would pin the floor at its own registration coverage and
         // mask what this test is about.
         drop(handle);

src/compute/src/sharing.rs

@@ -309,6 +309,17 @@ impl ArrangementSharingRegistry {
         }
     }
 
+    /// The `oks` seal frontier published for `id` on `worker_index`, if published.
+    pub(crate) fn published_upper(
+        &self,
+        id: &GlobalId,
+        worker_index: usize,
+    ) -> Option<Antichain<Timestamp>> {
+        let inner = self.lock();
+        let slot = inner.map.get(id)?.get(worker_index)?.as_ref()?;
+        Some(slot.oks.upper())
+    }
+
     /// Mints reader handles for `id` on `worker_index`, if published.
     pub(crate) fn handles(
         &self,