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>

Author: antiguru

src/compute/src/render.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,23 @@ 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. The interactive
+    /// runtime never re-exports maintained indexes, so it is never consumed by an export path.
+    ///
+    /// NOTE: no compile-time guard prevents an `until`-carrying multi-time dataflow from importing
+    /// this variant. Unlike [`Self::Trace`], the imported handles carry no `TraceFrontier`/`until`
+    /// bound, so safety rests on the Interactive=peek-only invariant (peeks are single-time and
+    /// need no `until`). A future SUBSCRIBE-on-interactive migration must add the
+    /// `TraceFrontier`/`until` bound here first.
+    SharedTrace(
+        GlobalId,
+        Arranged<'scope, SharedOksEnter<T>>,
+        Arranged<'scope, SharedErrsEnter<T>>,
+    ),
 }
 
 impl<'scope, T: RenderTimestamp> ArrangementFlavor<'scope, T> {
@@ -267,6 +285,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 +372,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 +429,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 +449,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 +465,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 +486,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 +621,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 +639,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);
         }

src/compute/src/render/context.rs

@@ -41,6 +41,7 @@ use timely::progress::Antichain;
 use crate::render::RenderTimestamp;
 use crate::render::context::{ArrangementFlavor, CollectionBundle, Context};
 use crate::render::errors::DataflowErrorSer;
+use crate::shared_trace::SharedOksEnter;
 use crate::typedefs::{RowRowAgent, RowRowEnter};
 
 impl<'scope, T: RenderTimestamp> Context<'scope, T> {
@@ -319,6 +320,9 @@ fn bundle_errs<'scope, T: RenderTimestamp>(
             ArrangementFlavor::Trace(_id, _oks, errs) => {
                 errs.clone().as_collection(|k, _v| k.clone())
             }
+            ArrangementFlavor::SharedTrace(_id, _oks, errs) => {
+                errs.clone().as_collection(|k, _v| k.clone())
+            }
         };
         collected.push(errs);
     }
@@ -400,6 +404,32 @@ where
             };
             (oks, errs2)
         }
+        // A maintenance index the interactive runtime imported from the sharing registry. Same
+        // shape as `Trace`, only the trace type differs (`SharedOksEnter`).
+        Some(ArrangementFlavor::SharedTrace(_, oks, errs)) => {
+            let (oks, errs2) = if source_precedes_lookup {
+                build_halfjoin_trace::<_, SharedOksEnter<_>, _>(
+                    updates,
+                    oks,
+                    prev_key,
+                    prev_thinning,
+                    |t1, t2| t1.le(t2),
+                    closure,
+                    config_set,
+                )
+            } else {
+                build_halfjoin_trace::<_, SharedOksEnter<_>, _>(
+                    updates,
+                    oks,
+                    prev_key,
+                    prev_thinning,
+                    |t1, t2| t1.lt(t2),
+                    closure,
+                    config_set,
+                )
+            };
+            (oks, errs2.concat(errs.as_collection(|k, _v| k.clone())))
+        }
         None => panic!("Arrangement promised by the planner is absent!"),
     }
 }
@@ -743,6 +773,17 @@ where
                 initial_closure,
             )
         }
+        // A maintenance index the interactive runtime imported from the sharing registry. Same
+        // shape as `Trace`, only the trace type differs (`SharedOksEnter`).
+        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())))
+        }
         None => panic!("Arrangement promised by the planner is absent!"),
     }
 }

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

@@ -41,6 +41,7 @@ use crate::render::RenderTimestamp;
 use crate::render::context::{ArrangementFlavor, CollectionBundle, Context};
 use crate::render::errors::DataflowErrorSer;
 use crate::render::join::mz_join_core::mz_join_core;
+use crate::shared_trace::SharedOksEnter;
 use crate::typedefs::{RowRowAgent, RowRowEnter};
 use mz_row_spine::{RowRowBuilder, RowRowColPagedBuilder, RowRowSpine};
 
@@ -195,6 +196,8 @@ enum JoinedFlavor<'scope, T: RenderTimestamp> {
     Local(Arranged<'scope, RowRowAgent<T, Diff>>),
     /// An imported arrangement.
     Trace(Arranged<'scope, RowRowEnter<mz_repr::Timestamp, Diff, T>>),
+    /// A shared-trace arrangement imported into the interactive runtime.
+    SharedTrace(Arranged<'scope, SharedOksEnter<T>>),
 }
 
 impl<'scope, T> Context<'scope, T>
@@ -238,6 +241,10 @@ where
                 errors.push(errs.as_collection(|k, _v| k.clone()).enter_region(inner));
                 JoinedFlavor::Trace(oks.enter_region(inner))
             }
+            (Some(ArrangementFlavor::SharedTrace(_gid, oks, errs)), None) => {
+                errors.push(errs.as_collection(|k, _v| k.clone()).enter_region(inner));
+                JoinedFlavor::SharedTrace(oks.enter_region(inner))
+            }
             (_, initial_closure) => {
                 // TODO: extract closure from the first stage in the join plan, should it exist.
                 // TODO: apply that closure in `flat_map_ref` rather than calling `.collection`.
@@ -421,52 +428,57 @@ where
             .arrangement(&lookup_key[..])
             .expect("Arrangement absent despite explicit construction");
 
+        // The nine `(stream flavor) x (lookup flavor)` combinations differ only in the two trace
+        // types handed to the generic `differential_join_inner` and the two arrangement values
+        // consumed. This local macro spells one combination. The `SharedTrace` rows exist so an
+        // interactive-runtime join over imported indexes type-checks. At runtime a dataflow's
+        // arrangements are all one runtime's flavor, so the mixed rows never fire, but exhaustive
+        // matching requires them.
+        macro_rules! join {
+            ($stream:expr, $stream_tr:ty, $lookup:expr, $lookup_tr:ty, $errs1:expr) => {{
+                let (oks, errs2) = self
+                    .differential_join_inner::<$stream_tr, $lookup_tr>($stream, $lookup, closure);
+                errors.push($errs1.as_collection(|k, _v| k.clone()));
+                errors.extend(errs2);
+                oks
+            }};
+        }
+
         match joined {
             JoinedFlavor::Collection(_) => {
                 unreachable!("JoinedFlavor::VecCollection variant avoided at top of method");
             }
             JoinedFlavor::Local(local) => match arrangement {
                 ArrangementFlavor::Local(oks, errs1) => {
-                    let (oks, errs2) = self
-                        .differential_join_inner::<RowRowAgent<_, _>, RowRowAgent<_, _>>(
-                            local, oks, closure,
-                        );
-
-                    errors.push(errs1.as_collection(|k, _v| k.clone()));
-                    errors.extend(errs2);
-                    oks
+                    join!(local, RowRowAgent<_, _>, oks, RowRowAgent<_, _>, errs1)
                 }
                 ArrangementFlavor::Trace(_gid, oks, errs1) => {
-                    let (oks, errs2) = self
-                        .differential_join_inner::<RowRowAgent<_, _>, RowRowEnter<_, _, _>>(
-                            local, oks, closure,
-                        );
-
-                    errors.push(errs1.as_collection(|k, _v| k.clone()));
-                    errors.extend(errs2);
-                    oks
+                    join!(local, RowRowAgent<_, _>, oks, RowRowEnter<_, _, _>, errs1)
+                }
+                ArrangementFlavor::SharedTrace(_gid, oks, errs1) => {
+                    join!(local, RowRowAgent<_, _>, oks, SharedOksEnter<_>, errs1)
                 }
             },
             JoinedFlavor::Trace(trace) => match arrangement {
                 ArrangementFlavor::Local(oks, errs1) => {
-                    let (oks, errs2) = self
-                        .differential_join_inner::<RowRowEnter<_, _, _>, RowRowAgent<_, _>>(
-                            trace, oks, closure,
-                        );
-
-                    errors.push(errs1.as_collection(|k, _v| k.clone()));
-                    errors.extend(errs2);
-                    oks
+                    join!(trace, RowRowEnter<_, _, _>, oks, RowRowAgent<_, _>, errs1)
+                }
+                ArrangementFlavor::Trace(_gid, oks, errs1) => {
+                    join!(trace, RowRowEnter<_, _, _>, oks, RowRowEnter<_, _, _>, errs1)
+                }
+                ArrangementFlavor::SharedTrace(_gid, oks, errs1) => {
+                    join!(trace, RowRowEnter<_, _, _>, oks, SharedOksEnter<_>, errs1)
+                }
+            },
+            JoinedFlavor::SharedTrace(trace) => match arrangement {
+                ArrangementFlavor::Local(oks, errs1) => {
+                    join!(trace, SharedOksEnter<_>, oks, RowRowAgent<_, _>, errs1)
                 }
                 ArrangementFlavor::Trace(_gid, oks, errs1) => {
-                    let (oks, errs2) = self
-                        .differential_join_inner::<RowRowEnter<_, _, _>, RowRowEnter<_, _, _>>(
-                            trace, oks, closure,
-                        );
-
-                    errors.push(errs1.as_collection(|k, _v| k.clone()));
-                    errors.extend(errs2);
-                    oks
+                    join!(trace, SharedOksEnter<_>, oks, RowRowEnter<_, _, _>, errs1)
+                }
+                ArrangementFlavor::SharedTrace(_gid, oks, errs1) => {
+                    join!(trace, SharedOksEnter<_>, oks, SharedOksEnter<_>, errs1)
                 }
             },
         }