compute: import published indexes as a shared arrangement

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

@@ -168,6 +168,9 @@ use crate::logging::compute::{
 use crate::render::columnar::CollectionEdge;
 use crate::render::context::{ArrangementFlavor, Context};
 use crate::render::errors::DataflowErrorSer;
+use crate::server::ComputeRuntimeRole;
+use crate::shared_trace::{Diagnostics, SharedErrsFrontier, SharedOksFrontier};
+use crate::sharing::{ArrangementSharingRegistry, SharedIndexArrangement};
 use crate::typedefs::{ErrBatcher, ErrBuilder, ErrSpine, KeyBatcher, MzTimestamp};
 use mz_row_spine::{DatumSeq, RowRowBatcher, RowRowBuilder};
 
@@ -557,6 +560,88 @@ pub fn build_compute_dataflow(
     });
 }
 
+/// Reports a publication point refusing to serve `as_of`, and aborts.
+///
+/// A refusal is a protocol-ordering failure: the controller promises an index's `since` never
+/// passes the `as_of` of a dataflow importing it. The diagnostics say which side moved. A standing
+/// hold at the refusing `since` means this runtime had already applied the compaction, so the
+/// create was ordered behind it on this runtime's own stream. A standing hold below the `since`
+/// means the trace compacted past its bound.
+fn report_compacted_past(
+    idx_id: GlobalId,
+    part: &str,
+    as_of: &Antichain<mz_repr::Timestamp>,
+    since: &Antichain<mz_repr::Timestamp>,
+    diagnostics: Diagnostics<mz_repr::Timestamp>,
+) -> ! {
+    panic!(
+        "Index {idx_id} ({part}) has been allowed to compact beyond the dataflow as_of: \
+         since {:?}, as_of {:?}, standing hold {:?}",
+        since.elements(),
+        as_of.elements(),
+        diagnostics.standing_hold.elements(),
+    )
+}
+
+/// Imports the published `oks`/`errs` arrangements of `idx_id` into `outer` as a snapshot at
+/// `as_of` bounded by `until`, through [`crate::shared_trace::SharedTraceHandle::import_snapshot_at`].
+///
+/// Binds through [`ArrangementSharingRegistry::get_or_create`], so a dependency not yet published
+/// yields an unbacked point whose import produces nothing until a publisher adopts it. That is what
+/// lets every interactive dataflow build in command arrival order without deferring.
+///
+/// The returned slot must be retained for as long as the import is alive: its strong count is the
+/// registry's only measure of a live reader, since a handle holds only the inner `Arc<SharedTrace>`.
+/// The read hold is each returned `Arranged`'s own trace, registered at `as_of`, so a consumer that
+/// 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_published_index<'outer>(
+    outer: Scope<'outer, mz_repr::Timestamp>,
+    registry: &ArrangementSharingRegistry,
+    idx_id: GlobalId,
+    name: &str,
+    as_of: &Antichain<mz_repr::Timestamp>,
+    until: &Antichain<mz_repr::Timestamp>,
+) -> (
+    Arranged<'outer, SharedOksFrontier>,
+    Arranged<'outer, SharedErrsFrontier>,
+    Arc<SharedIndexArrangement>,
+) {
+    // Pairwise import reads publisher worker `i` from importer worker `i`. The primitive's import
+    // additionally asserts equal total peer counts.
+    let slot = registry.get_or_create(idx_id, outer.index(), outer.peers());
+
+    // `handle_at` checks the published `since` and registers the hold under one acquisition of the
+    // state lock, so the publisher cannot advance `since` between the check and the registration.
+    // A fresh placeholder's `since` is the minimum, so this succeeds for an unadopted slot.
+    let oks_handle = match slot.oks.handle_at(as_of) {
+        Ok(handle) => handle,
+        Err(since) => report_compacted_past(idx_id, "oks", as_of, &since, slot.oks.diagnostics()),
+    };
+    let errs_handle = match slot.errs.handle_at(as_of) {
+        Ok(handle) => handle,
+        Err(since) => report_compacted_past(idx_id, "errs", as_of, &since, slot.errs.diagnostics()),
+    };
+
+    // These handles' own registrations end with this function. The hold that outlives it is the one
+    // `import_snapshot_at` clones into each returned `Arranged`.
+    let oks_arranged = oks_handle.import_snapshot_at(
+        outer.clone(),
+        &format!("Shared{name}"),
+        as_of.clone(),
+        until.clone(),
+    );
+    let errs_arranged = errs_handle.import_snapshot_at(
+        outer,
+        &format!("SharedErr{name}"),
+        as_of.clone(),
+        until.clone(),
+    );
+
+    (oks_arranged, errs_arranged, slot)
+}
+
 // This implementation block allows child timestamps to vary from parent timestamps,
 // but requires the parent timestamp to be `repr::Timestamp`.
 impl<'g, T> Context<'g, T>
@@ -601,6 +686,21 @@ where
         snapshot_mode: SnapshotMode,
         start_signal: StartSignal,
     ) {
+        // The interactive runtime maintains no traces of its own. It imports the arrangements the
+        // maintenance runtime publishes into the per-process sharing registry.
+        if compute_state.role() == ComputeRuntimeRole::Interactive {
+            self.import_index_shared(
+                outer,
+                compute_state,
+                tokens,
+                input_probe,
+                idx_id,
+                idx,
+                start_signal,
+            );
+            return;
+        }
+
         if let Some(traces) = compute_state.traces.get_mut(&idx_id) {
             assert!(
                 PartialOrder::less_equal(&traces.compaction_frontier(), &self.as_of_frontier),
@@ -681,6 +781,58 @@ where
             );
         }
     }
+
+    /// The interactive-runtime counterpart to [`Self::import_index`].
+    ///
+    /// 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.
+    fn import_index_shared<'outer>(
+        &mut self,
+        outer: Scope<'outer, mz_repr::Timestamp>,
+        compute_state: &ComputeState,
+        tokens: &mut BTreeMap<GlobalId, Rc<dyn Any>>,
+        input_probe: probe::Handle<mz_repr::Timestamp>,
+        idx_id: GlobalId,
+        idx: &IndexDesc<LirScalarExpr>,
+        start_signal: StartSignal,
+    ) {
+        let name = format!("Index({}, {:?})", idx.on_id, idx.key);
+        let (mut oks_arranged, errs_arranged, slot) = import_published_index(
+            outer,
+            &compute_state.sharing_registry,
+            idx_id,
+            &name,
+            &self.as_of_frontier,
+            &self.until,
+        );
+
+        // Attach the input probe to the replayed batch stream so hydration tracking observes it,
+        // mirroring the maintenance import.
+        oks_arranged.stream = oks_arranged.stream.probe_with(&input_probe);
+
+        // Enter the dataflow scope and gate on the start signal, mirroring the maintenance Trace
+        // import's `.enter(self.scope).with_start_signal(..)`. The shared handle shares the
+        // maintenance arrangement's batch/cursor types, so the entered `Arranged` is a real
+        // arrangement `ArrangementFlavor::SharedTrace` can carry and downstream operators consume.
+        let ok_arranged = oks_arranged
+            .enter(self.scope)
+            .with_start_signal(start_signal.clone());
+        let err_arranged = errs_arranged
+            .enter(self.scope)
+            .with_start_signal(start_signal);
+
+        let bundle = CollectionBundle::from_expressions(
+            idx.key.clone(),
+            ArrangementFlavor::SharedTrace(idx_id, ok_arranged, err_arranged),
+        );
+        self.update_id(Id::Global(idx.on_id), bundle);
+
+        // 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_published_index`.
+        tokens.insert(idx_id, Rc::new(slot));
+    }
 }
 
 // This implementation block requires the scopes have the same timestamp as the trace manager.
@@ -763,6 +915,9 @@ impl<'g> Context<'g, mz_repr::Timestamp> {
                 }
                 compute_state.traces.set(idx_id, trace);
             }
+            Some(ArrangementFlavor::SharedTrace(gid, _, _)) => {
+                alias_shared_reexport(compute_state, &self.scope, idx_id, gid);
+            }
             None => {
                 println!("collection available: {:?}", bundle.collection.is_none());
                 println!(
@@ -874,6 +1029,9 @@ where
                 }
                 compute_state.traces.set(idx_id, trace);
             }
+            Some(ArrangementFlavor::SharedTrace(gid, _, _)) => {
+                alias_shared_reexport(compute_state, &outer, idx_id, gid);
+            }
             None => {
                 println!("collection available: {:?}", bundle.collection.is_none());
                 println!(
@@ -916,6 +1074,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`.
 ///
@@ -1525,6 +1707,9 @@ impl<'scope, T: RenderTimestamp + MaybeBucketByTime> Context<'scope, T> {
                     Trace(_, a, _) => {
                         a.stream = self.log_operator_hydration_inner(a.stream.clone(), lir_id);
                     }
+                    SharedTrace(_, a, _) => {
+                        a.stream = self.log_operator_hydration_inner(a.stream.clone(), lir_id);
+                    }
                 }
             }
             None => {
@@ -2093,3 +2278,6 @@ impl Pairer {
         (first, second)
     }
 }
+
+#[cfg(test)]
+mod tests;

src/compute/src/render/context.rs

@@ -54,6 +54,7 @@ use crate::extensions::reduce::MzReduce;
 use crate::render::columnar::CollectionEdge;
 use crate::render::errors::{DataflowErrorSer, ErrorLogger};
 use crate::render::{LinearJoinSpec, MaybeBucketByTime, RenderTimestamp};
+use crate::shared_trace::{SharedErrsEnter, SharedOksEnter};
 use crate::typedefs::{
     ErrAgent, ErrBatcher, ErrBuilder, ErrEnter, ErrSpine, RowRowAgent, RowRowEnter, RowRowSpine,
 };
@@ -233,6 +234,17 @@ pub enum ArrangementFlavor<'scope, T: RenderTimestamp> {
         Arranged<'scope, RowRowEnter<mz_repr::Timestamp, Diff, T>>,
         Arranged<'scope, ErrEnter<mz_repr::Timestamp, T>>,
     ),
+    /// A maintenance-runtime arrangement imported into the interactive runtime through the
+    /// 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, so an export of this
+    /// same arrangement can alias it instead of arranging again.
+    SharedTrace(
+        GlobalId,
+        Arranged<'scope, SharedOksEnter<T>>,
+        Arranged<'scope, SharedErrsEnter<T>>,
+    ),
 }
 
 impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
@@ -267,6 +279,10 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
                 oks.clone().as_collection(logic),
                 errs.clone().as_collection(|k, &()| k.clone()),
             ),
+            ArrangementFlavor::SharedTrace(_, oks, errs) => (
+                oks.clone().as_collection(logic),
+                errs.clone().as_collection(|k, &()| k.clone()),
+            ),
         }
     }
 
@@ -350,6 +366,18 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
                 let errs = errs.concat(mfp_errs.as_collection());
                 (oks, errs)
             }
+            ArrangementFlavor::SharedTrace(_, oks, errs) => {
+                let (oks, mfp_errs) = CollectionBundle::<T>::flat_map_core_fallible::<_, _, DCB, _>(
+                    oks.clone(),
+                    key,
+                    max_demand,
+                    logic,
+                    REFUEL,
+                );
+                let errs = errs.clone().as_collection(|k, &()| k.clone());
+                let errs = errs.concat(mfp_errs.as_collection());
+                (oks, errs)
+            }
         }
     }
 
@@ -395,6 +423,17 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
                 let errs = errs.clone().as_collection(|k, &()| k.clone());
                 (oks, errs)
             }
+            ArrangementFlavor::SharedTrace(_, oks, errs) => {
+                let oks = CollectionBundle::<T>::flat_map_core_ok::<_, _, DCB, _>(
+                    oks.clone(),
+                    key,
+                    max_demand,
+                    logic,
+                    REFUEL,
+                );
+                let errs = errs.clone().as_collection(|k, &()| k.clone());
+                (oks, errs)
+            }
         }
     }
 }
@@ -404,6 +443,7 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
         match self {
             ArrangementFlavor::Local(oks, _errs) => oks.stream.scope(),
             ArrangementFlavor::Trace(_gid, oks, _errs) => oks.stream.scope(),
+            ArrangementFlavor::SharedTrace(_gid, oks, _errs) => oks.stream.scope(),
         }
     }
 
@@ -419,6 +459,11 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
                 oks.clone().enter_region(region),
                 errs.clone().enter_region(region),
             ),
+            ArrangementFlavor::SharedTrace(gid, oks, errs) => ArrangementFlavor::SharedTrace(
+                *gid,
+                oks.clone().enter_region(region),
+                errs.clone().enter_region(region),
+            ),
         }
     }
 }
@@ -435,6 +480,11 @@ impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
                 oks.clone().leave_region(outer),
                 errs.clone().leave_region(outer),
             ),
+            ArrangementFlavor::SharedTrace(gid, oks, errs) => ArrangementFlavor::SharedTrace(
+                *gid,
+                oks.clone().leave_region(outer),
+                errs.clone().leave_region(outer),
+            ),
         }
     }
 }
@@ -565,8 +615,9 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
     /// reads is the consumer's choice, and a delta join reads both within one operator, so a
     /// binding's definition cannot know which form to collapse.
     ///
-    /// NOTE: Leaves imported arrangements (`ArrangementFlavor::Trace`) alone, whose error traces
-    /// this dataflow cannot rewrite in place. Their errors arrive bounded by the exporting
+    /// NOTE: Leaves imported arrangements (`ArrangementFlavor::Trace` and
+    /// `ArrangementFlavor::SharedTrace`) alone, whose error traces this dataflow cannot rewrite in
+    /// place. Their errors arrive bounded by the exporting
     /// dataflow's last level of sharing rather than collapsed to one, since nothing collapses at an
     /// export. A global read more than once within one dataflow is not collapsed either, because
     /// only local bindings reach this.
@@ -582,7 +633,9 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> {
                     let name = format!("Distinct errors[{key:?}]");
                     ArrangementFlavor::Local(oks, distinct_arranged_errs(errs, &name))
                 }
-                flavor @ ArrangementFlavor::Trace(..) => flavor,
+                flavor @ (ArrangementFlavor::Trace(..) | ArrangementFlavor::SharedTrace(..)) => {
+                    flavor
+                }
             };
             self.arranged.insert(key, flavor);
         }