diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index a2bf55981ba8a..5422b6855c8af 100644 --- a/src/compute/src/compute_state.rs +++ b/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; diff --git a/src/compute/src/render.rs b/src/compute/src/render.rs index b9623d508800b..ba93f26f29584 100644 --- a/src/compute/src/render.rs +++ b/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, + since: &Antichain, + diagnostics: Diagnostics, +) -> ! { + 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`. +/// 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, + until: &Antichain, +) -> ( + Arranged<'outer, SharedOksFrontier>, + Arranged<'outer, SharedErrsFrontier>, + Arc, +) { + // 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>, + input_probe: probe::Handle, + idx_id: GlobalId, + idx: &IndexDesc, + 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; diff --git a/src/compute/src/render/context.rs b/src/compute/src/render/context.rs index e3a6351ec857e..6c0031be90452 100644 --- a/src/compute/src/render/context.rs +++ b/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>, Arranged<'scope, ErrEnter>, ), + /// 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>, + Arranged<'scope, SharedErrsEnter>, + ), } 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::::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::::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); } diff --git a/src/compute/src/render/join/delta_join.rs b/src/compute/src/render/join/delta_join.rs index cfac33a55887a..3a15109352223 100644 --- a/src/compute/src/render/join/delta_join.rs +++ b/src/compute/src/render/join/delta_join.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,31 @@ where }; (oks, errs2) } + // As `Trace`, over the trace type the interactive runtime imports. + 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) + } None => panic!("Arrangement promised by the planner is absent!"), } } @@ -743,6 +772,16 @@ where initial_closure, ) } + // As `Trace`, over the trace type the interactive runtime imports. + Some(ArrangementFlavor::SharedTrace(_, oks, _errs)) => { + let (oks, errs2) = build_update_stream_trace::<_, SharedOksEnter<_>>( + oks, + as_of, + source_relation, + initial_closure, + ); + (oks, errs2) + } None => panic!("Arrangement promised by the planner is absent!"), } } diff --git a/src/compute/src/render/join/linear_join.rs b/src/compute/src/render/join/linear_join.rs index 35e1a77abc9e9..34091f06747be 100644 --- a/src/compute/src/render/join/linear_join.rs +++ b/src/compute/src/render/join/linear_join.rs @@ -42,6 +42,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, RowRowSpine}; /// Available linear join implementations. @@ -195,6 +196,8 @@ enum JoinedFlavor<'scope, T: RenderTimestamp> { Local(Arranged<'scope, RowRowAgent>), /// An imported arrangement. Trace(Arranged<'scope, RowRowEnter>), + /// A shared-trace arrangement imported into the interactive runtime. + SharedTrace(Arranged<'scope, SharedOksEnter>), } 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<_, _>>( - 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::, 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::, 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<_, _, _>>( - 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) } }, } diff --git a/src/compute/src/render/tests.rs b/src/compute/src/render/tests.rs new file mode 100644 index 0000000000000..9b58236f85a05 --- /dev/null +++ b/src/compute/src/render/tests.rs @@ -0,0 +1,784 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +use std::sync::mpsc; + +use differential_dataflow::input::{Input, InputSession}; +use differential_dataflow::operators::arrange::Arranged; +use differential_dataflow::trace::TraceReader; +use mz_repr::{Datum, Diff, GlobalId, Row, Timestamp}; +use mz_row_spine::{DatumSeq, RowRowBatcher, RowRowBuilder}; +use mz_timely_util::columnation::ColumnationChunker; +use timely::dataflow::operators::capture::Extract; +use timely::dataflow::operators::{Capture, Probe}; +use timely::dataflow::{ProbeHandle, Scope}; +use timely::progress::Antichain; + +use crate::extensions::arrange::{KeyCollection, MzArrange}; +use crate::shared_trace::PublishArrangement; +use crate::shared_trace::SharedOksFrontier; +use crate::sharing::ArrangementSharingRegistry; +use crate::typedefs::{ErrBatcher, ErrBuilder, ErrSpine, RowRowAgent, RowRowSpine}; + +use super::import_published_index; +use crate::server::ComputeRuntimeRole; + +fn test_rows() -> Vec<(Row, Row)> { + vec![ + ( + Row::pack_slice(&[Datum::Int32(1)]), + Row::pack_slice(&[Datum::String("a")]), + ), + ( + Row::pack_slice(&[Datum::Int32(2)]), + Row::pack_slice(&[Datum::String("b")]), + ), + ] +} + +/// Publishes `rows` as a `(RowRow oks, Err errs)` index into `registry` under `id` on worker 0 +/// of `scope`. The updates are written at time 0 and sealed by advancing the inputs to 1. +/// +/// The `InputSession` handles drop at the end of this call, buffering the sealed updates for the +/// worker to process on later steps, mirroring `sharing.rs`'s `publish_index_into`. Returns the +/// trace agents: the point closes with the trace, and the trace lives as long as an agent does, so +/// the caller keeps them for as long as it reads. +fn publish_index( + scope: Scope<'_, Timestamp>, + registry: &ArrangementSharingRegistry, + id: GlobalId, + rows: Vec<(Row, Row)>, +) -> ( + RowRowAgent, + crate::typedefs::ErrAgent, +) { + let (mut oks_input, oks_collection) = scope.new_collection::<(Row, Row), Diff>(); + let oks = oks_collection.mz_arrange::< + ColumnationChunker<_>, + RowRowBatcher<_, _>, + RowRowBuilder<_, _>, + RowRowSpine<_, _>, + >("test oks"); + + let (mut errs_input, errs_collection) = + scope.new_collection::(); + let errs = KeyCollection::from(errs_collection) + .mz_arrange::, ErrBatcher<_, _>, ErrBuilder<_, _>, ErrSpine<_, _>>( + "test errs", + ); + + let slot = registry.get_or_create(id, 0, 1); + PublishArrangement::adopt(&oks, &slot.oks, || {}); + PublishArrangement::adopt(&errs, &slot.errs, || {}); + registry.notify(id, 0); + + for (k, v) in rows { + oks_input.update((k, v), Diff::ONE); + } + oks_input.advance_to(Timestamp::from(1_u64)); + oks_input.flush(); + errs_input.advance_to(Timestamp::from(1_u64)); + errs_input.flush(); + (oks.trace.clone(), errs.trace.clone()) +} + +/// The interactive import path imports a maintenance-published arrangement into a second +/// dataflow as a static `as_of` snapshot via `SharedTraceHandle::import_snapshot_at`, +/// reconstructing the same rows, and registers a read hold at the importing dataflow's `as_of`. +#[mz_ore::test] +fn interactive_import_replays_rows_and_holds_at_as_of() { + let id = GlobalId::User(1); + let rows = test_rows(); + let mut expected: Vec<(Row, Row)> = rows.clone(); + expected.sort(); + + // `as_of` beyond the publish-time `since` (0), so a correct hold advance is observable: the + // freshly minted handle's hold starts at `since` (0) and must be advanced to `as_of` (1). + let as_of = Antichain::from_elem(Timestamp::from(1_u64)); + let registry = ArrangementSharingRegistry::new(); + + let (capture_tx, capture_rx) = mpsc::channel(); + let registry_in = registry.clone(); + let as_of_in = as_of.clone(); + + timely::execute_directly(move |worker| { + // Maintenance runtime: publish the index into the shared registry. + let _keep = worker.dataflow::(|scope| { + publish_index(scope, ®istry_in, id, rows.clone()) + }); + + // Interactive runtime: a temporary dataflow imports the published arrangement via the + // new path and captures the reconstructed rows. + let probe = ProbeHandle::new(); + let (mut oks_trace, mut errs_trace) = worker.dataflow::(|scope| { + // `until` empty: no upper suppression, so the whole snapshot at `as_of` flows. + let (oks_arranged, errs_arranged, _slot) = import_published_index( + scope.clone(), + ®istry_in, + id, + "Index", + &as_of_in, + &Antichain::new(), + ); + + let collected = Arranged::::flat_map_batches( + oks_arranged.stream, + |k: DatumSeq, v: DatumSeq| { + let key = Row::pack_slice(&k.into_iter().collect::>()); + let val = Row::pack_slice(&v.into_iter().collect::>()); + [(key, val)] + }, + ); + collected.inner.probe_with(&probe).capture_into(capture_tx); + (oks_arranged.trace, errs_arranged.trace) + }); + + // The read hold is the `Arranged`'s own trace, and it sits at the dataflow's `as_of`, not + // the publish-time `since`. + assert_eq!(oks_trace.get_logical_compaction(), as_of_in.borrow()); + assert_eq!(errs_trace.get_logical_compaction(), as_of_in.borrow()); + + // Drive both dataflows until the imported-and-reconstructed output has sealed time 0. + while probe.less_than(&Timestamp::from(1_u64)) { + worker.step(); + } + }); + + let mut found: Vec<(Row, Row)> = capture_rx + .extract() + .into_iter() + .flat_map(|(_, data)| data) + .filter(|(_, _, diff)| diff.is_positive()) + .map(|((k, v), _, _)| (k, v)) + .collect(); + found.sort(); + assert_eq!(found, expected); +} + +/// Like [`publish_index`], but also returns the writer-side `oks` `InputSession` and a plain +/// `TraceAgent` clone of the `oks` trace (not a `SharedTraceHandle`), so a test can keep +/// publishing after the initial seal and force compaction directly on the writer. Mirrors the +/// `writer` handle in the differential-dataflow primitive's own `import_hold_pins_then_releases` +/// (`differential-dataflow/tests/sharing.rs`), which drives the writer side of the identical +/// pin-then-release scenario one layer down. +fn publish_index_with_writer( + scope: Scope<'_, Timestamp>, + registry: &ArrangementSharingRegistry, + id: GlobalId, + rows: Vec<(Row, Row)>, +) -> ( + InputSession, + InputSession, + RowRowAgent, + crate::typedefs::ErrAgent, +) { + let (mut oks_input, oks_collection) = scope.new_collection::<(Row, Row), Diff>(); + let oks = oks_collection.mz_arrange::< + ColumnationChunker<_>, + RowRowBatcher<_, _>, + RowRowBuilder<_, _>, + RowRowSpine<_, _>, + >("test oks"); + let oks_writer = oks.trace.clone(); + + let (mut errs_input, errs_collection) = + scope.new_collection::(); + let errs = KeyCollection::from(errs_collection) + .mz_arrange::, ErrBatcher<_, _>, ErrBuilder<_, _>, ErrSpine<_, _>>( + "test errs", + ); + + let slot = registry.get_or_create(id, 0, 1); + PublishArrangement::adopt(&oks, &slot.oks, || {}); + PublishArrangement::adopt(&errs, &slot.errs, || {}); + registry.notify(id, 0); + + for (k, v) in rows { + oks_input.update((k, v), Diff::ONE); + } + oks_input.advance_to(Timestamp::from(1_u64)); + oks_input.flush(); + errs_input.advance_to(Timestamp::from(1_u64)); + errs_input.flush(); + + (oks_input, errs_input, oks_writer, errs.trace.clone()) +} + +/// Feeds `oks_input` a filler update at `at`, advances it to `next`, and steps `worker` a few +/// times, mirroring the `tick` helper in `differential-dataflow`'s own `sharing.rs` test suite. +/// A reader's hold reaches the trace on the arrange operator's next activation, which an idle +/// dataflow never gets, so tests tick after moving one. +fn tick( + worker: &mut timely::worker::Worker, + oks_input: &mut InputSession, + at: Timestamp, + next: Timestamp, +) { + oks_input.advance_to(at); + oks_input.update( + ( + Row::pack_slice(&[Datum::Int32(-1)]), + Row::pack_slice(&[Datum::String("tick")]), + ), + Diff::ONE, + ); + oks_input.advance_to(next); + oks_input.flush(); + for _ in 0..20 { + worker.step(); + } +} + +/// The interactive import's read hold pins the maintenance trace at `as_of` only while it is +/// alive: once the importing dataflow drops, and with it every registration the import made, the +/// trace is free to compact past `as_of`, which it could not do before the drop. +/// +/// 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_published_index` primitive that +/// `import_index_shared` calls in production, rather than re-deriving the contract from +/// scratch. +/// +/// Staging this end-to-end through the real `ComputeState`/`TraceManager`, as the maintenance +/// `import_index` path would, is not practical in this harness: there is no controller driving +/// frontier advancement, so nothing would ever request compaction past `as_of` for real (the +/// same limitation that keeps the since-gate tests elsewhere in this crate on +/// `execute_directly` plus a directly-driven writer, rather than a full coordinator). The +/// closest observable proxy is used instead: a writer-side compaction request advanced directly +/// on the published trace, exactly as `import_hold_pins_then_releases` does, with the assertion +/// made through `SharedTraceHandle::snapshot_at` (a real read against the shared trace's actual +/// `since`, not a count or a flag). +#[mz_ore::test] +fn interactive_import_hold_releases_on_drop() { + let id = GlobalId::User(1); + let rows = test_rows(); + let as_of_time = Timestamp::from(1_u64); + let as_of = Antichain::from_elem(as_of_time); + let registry = ArrangementSharingRegistry::new(); + + timely::execute_directly(move |worker| { + // Maintenance runtime: publish the index, keeping the `oks` `InputSession` (so we can + // tick the dataflow afterward) and a plain writer trace handle (so we can request + // compaction on it directly, as a controller would) alive across the whole closure. + let (mut oks_input, _errs_input, mut oks_writer, _errs_keep) = worker + .dataflow::(|scope| { + publish_index_with_writer(scope, ®istry, id, rows.clone()) + }); + + // Interactive runtime: import at `as_of`, exactly as `import_index_shared` does. The read + // hold is each `Arranged`'s own `trace`, so those are what is kept here. Production keeps + // them the same way, inside the `CollectionBundle` the import is bound into, which is what + // 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::(|scope| { + let (oks_arranged, errs_arranged, _slot) = import_published_index( + scope.clone(), + ®istry, + id, + "Index", + &as_of, + &Antichain::new(), + ); + (oks_arranged.trace, errs_arranged.trace) + }); + + // The controller requests compaction well past `as_of`, and both runtimes apply it: the + // writer handle advances, which the trace mirrors into the published `since` at once, and + // `note_standing_hold` advances the importing runtime's own position, exactly as + // `handle_allow_compaction` does on each side. The `since` stays pinned to `as_of` here by + // the live reader hold. + let target = Antichain::from_elem(Timestamp::from(10_u64)); + registry.note_standing_hold(id, 0, &target); + oks_writer.set_logical_compaction(target.borrow()); + oks_writer.set_physical_compaction(target.borrow()); + tick( + worker, + &mut oks_input, + Timestamp::from(5_u64), + Timestamp::from(6_u64), + ); + + // The live interactive-import hold still pins the trace at `as_of`: a read there still + // succeeds despite the writer's request. The probe handle is minted only to read and is + // dropped immediately, so the hold it registers at the current `since` cannot outlive this + // scope and confound the release assertion below. + { + let (probe_oks, _probe_errs) = registry.handles(&id, 0).expect("still published"); + assert!( + probe_oks.snapshot_at(&as_of_time).is_some(), + "the live interactive-import hold must keep `as_of` readable" + ); + } + + // Drop the import's traces, as happens when the interactive dataflow and the + // `CollectionBundle` holding its arrangements drop. With no reader hold left, the next tick + // lets the publisher's forwarded `since` follow the writer's request. + drop(oks_trace); + drop(errs_trace); + tick( + worker, + &mut oks_input, + Timestamp::from(11_u64), + Timestamp::from(12_u64), + ); + + // The trace compacted past `as_of`: a fresh handle (minted only now, so it introduces no + // new hold at `as_of`) can no longer read there. + let (released_oks, _released_errs) = registry.handles(&id, 0).expect("still published"); + assert!( + released_oks.snapshot_at(&as_of_time).is_none(), + "after the hold drops, the trace must be free to compact past `as_of`" + ); + }); +} + +/// A stream-only import still holds the shared trace after dataflow construction ends. +/// +/// This is the regression that matters for anything long-lived on the interactive runtime. The +/// hold that a consumer keeps is the returned `Arranged`'s own trace, and only `mz_join_core` +/// keeps one: it moves its input traces into its operator. `as_collection` and the reduce path +/// take the stream and drop the handle, and the `CollectionBundle` holding it lives in the +/// build-time `Context`, which dies when `build_compute_dataflow` returns. So without a hold owned +/// by the import's own source operator there is no registration left once the dataflow is built, +/// the publisher falls back to the writer-driven frontier, and it compacts straight past the +/// `as_of` the dataflow is still reading at. +/// +/// The assertion is on `Published::logical_holds` rather than on a read, because a read cannot +/// tell "a hold exists at `f`" from "no hold exists and the publisher is forwarding `f` from the +/// fallback". Those two look identical from outside and are the whole difference here. +#[mz_ore::test] +fn interactive_import_holds_after_construction() { + let id = GlobalId::User(1); + let rows = test_rows(); + // `as_of` beyond the published seal, so the import cannot acknowledge past it and downgrade + // the hold away. That keeps the assertion about the hold's existence rather than its value. + let as_of = Antichain::from_elem(Timestamp::from(5_u64)); + let registry = ArrangementSharingRegistry::new(); + + timely::execute_directly(move |worker| { + let (mut oks_input, _errs_input, _oks_writer, _errs_keep) = worker + .dataflow::(|scope| { + publish_index_with_writer(scope, ®istry, id, rows.clone()) + }); + + // Build an interactive import whose only consumer is the batch stream, and let every + // handle it produced go out of scope with the builder, exactly as production does. + let probe = ProbeHandle::new(); + worker.dataflow::(|scope| { + let (oks_arranged, _errs_arranged, _slot) = import_published_index( + scope.clone(), + ®istry, + id, + "Index", + &as_of, + &Antichain::new(), + ); + let collected = Arranged::::flat_map_batches( + oks_arranged.stream, + |k: DatumSeq, _v: DatumSeq| [Row::pack_slice(&k.into_iter().collect::>())], + ); + collected.inner.probe_with(&probe); + }); + + // Run both dataflows, so the import registers its queue and drains what is published. + // `tick` advances the input, so each call needs a fresh, larger time. It stops at 3, + // leaving the published seal below the `as_of` of 5. + tick( + worker, + &mut oks_input, + Timestamp::from(1_u64), + Timestamp::from(2_u64), + ); + tick( + worker, + &mut oks_input, + Timestamp::from(2_u64), + Timestamp::from(3_u64), + ); + + let holds = registry + .published_logical_holds(&id, 0) + .expect("still published"); + assert!( + !holds.is_empty(), + "a built import must leave a read hold behind, else the publisher compacts past its \ + as_of as soon as the controller allows it" + ); + assert!( + timely::PartialOrder::less_equal(&holds, &as_of), + "the import's hold must not have released past its own as_of: {holds:?}" + ); + }); +} + +/// The published `since` must not chase the readers' own holds. +/// +/// Before the controller's first `AllowCompaction` there is no writer-driven floor, and if the +/// publisher falls back to its own agent hold it closes a feedback loop: it drives that hold up +/// from the meet of the reader holds every activation, so the published `since` climbs to wherever +/// the readers are. A later read at an earlier time is then refused, and it is a read the +/// controller has allowed nothing against. +#[mz_ore::test] +fn published_since_does_not_chase_reader_holds() { + let id = GlobalId::User(1); + let rows = test_rows(); + let high = Antichain::from_elem(Timestamp::from(2_u64)); + let low = Antichain::from_elem(Timestamp::from(1_u64)); + let registry = ArrangementSharingRegistry::new(); + + timely::execute_directly(move |worker| { + let (mut oks_input, _errs_input, _w, _errs_keep) = + worker.dataflow::(|scope| { + publish_index_with_writer(scope, ®istry, id, rows.clone()) + }); + // 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::(|scope| { + let (_o, _e, _slot) = import_published_index( + scope.clone(), + ®istry, + id, + "Index", + &high, + &Antichain::new(), + ); + }); + for t in 1..4 { + tick( + worker, + &mut oks_input, + Timestamp::from(t), + Timestamp::from(t + 1), + ); + } + + // The writer has compacted nothing, so a read at the lower time is still legal. + let (probe_oks, _) = registry.handles(&id, 0).expect("published"); + let since = probe_oks.frontiers().0; + assert!( + timely::PartialOrder::less_equal(&since, &low), + "published since {:?} chased the reader's as_of; a legal read at {:?} would be \ + refused even though the controller allowed no compaction", + since.elements(), + low.elements() + ); + }); +} + +/// An import's reported physical compaction must not lead the published chain's coverage. +/// +/// `mz_join_core` asserts exactly this at start-up, against the coverage it derives from +/// `map_batches`, and differential's own `join_core` carries the same assert. An `as_of` may +/// legitimately lead the coverage: an import over a placeholder whose publisher has not adopted it +/// yet sees an empty chain, and a read at a timestamp beyond the index's seal leads it too. +/// Reporting the `as_of` here therefore aborts the worker on a correct import, and under shared +/// fate that takes the process with it. +#[mz_ore::test] +fn import_reports_physical_within_chain_coverage() { + let id = GlobalId::User(1); + let rows = test_rows(); + let as_of = Antichain::from_elem(Timestamp::from(5_u64)); + let registry = ArrangementSharingRegistry::new(); + + timely::execute_directly(move |worker| { + let (mut oks_input, _errs_input, _w, _errs_keep) = + worker.dataflow::(|scope| { + publish_index_with_writer(scope, ®istry, id, rows.clone()) + }); + tick( + worker, + &mut oks_input, + Timestamp::from(1_u64), + Timestamp::from(2_u64), + ); + tick( + worker, + &mut oks_input, + Timestamp::from(2_u64), + Timestamp::from(3_u64), + ); + + let mut trace = worker.dataflow::(|scope| { + let (oks_arranged, _e, _slot) = import_published_index( + scope.clone(), + ®istry, + id, + "Index", + &as_of, + &Antichain::new(), + ); + oks_arranged.trace + }); + + // Exactly `mz_join_core`'s start-up computation. + use differential_dataflow::trace::BatchReader; + let mut coverage = Antichain::from_elem(Timestamp::MIN); + trace.map_batches(|b| coverage.clone_from(b.upper())); + let physical = trace.get_physical_compaction().to_owned(); + assert!( + timely::PartialOrder::less_equal(&physical, &coverage), + "mz_join_core would panic: physical {:?} leads coverage {:?}", + physical.elements(), + coverage.elements() + ); + }); +} + +/// A live import's hold can be downgraded, so the publisher compacts behind a long-lived reader +/// rather than staying pinned at its `as_of` for the reader's whole life. +/// +/// This is what a join on the interactive runtime does: `mz_join_core` calls +/// `set_logical_compaction` on each input trace as the other input's frontier advances, and +/// `set_physical_compaction` as it acknowledges batches. An unbounded interactive dataflow that +/// could not downgrade would pin the maintenance index at the `as_of` it started from, so the +/// publisher could never compact for as long as the dataflow ran. +/// +/// The hold has to be the `Arranged`'s own trace for this to work. A separate hold token retained +/// beside it would defeat the downgrade entirely, since the publisher forwards the *meet* of the +/// registered holds and a hold nobody downgrades is a floor under every hold that is. +#[mz_ore::test] +fn interactive_import_hold_downgrades_while_live() { + let id = GlobalId::User(1); + let rows = test_rows(); + let as_of_time = Timestamp::from(1_u64); + let as_of = Antichain::from_elem(as_of_time); + let registry = ArrangementSharingRegistry::new(); + + timely::execute_directly(move |worker| { + let (mut oks_input, _errs_input, mut oks_writer, _errs_keep) = worker + .dataflow::(|scope| { + publish_index_with_writer(scope, ®istry, id, rows.clone()) + }); + + let (mut oks_trace, mut errs_trace) = worker.dataflow::(|scope| { + let (oks_arranged, errs_arranged, _slot) = import_published_index( + scope.clone(), + ®istry, + id, + "Index", + &as_of, + &Antichain::new(), + ); + (oks_arranged.trace, errs_arranged.trace) + }); + + // The controller allows compaction well past `as_of`, both runtimes apply it, and the + // writer applies it to the trace. + let target = Antichain::from_elem(Timestamp::from(10_u64)); + registry.note_standing_hold(id, 0, &target); + oks_writer.set_logical_compaction(target.borrow()); + oks_writer.set_physical_compaction(target.borrow()); + tick( + worker, + &mut oks_input, + Timestamp::from(5_u64), + Timestamp::from(6_u64), + ); + + // Still pinned: the import has not downgraded, so `as_of` stays readable. + { + let (probe_oks, _probe_errs) = registry.handles(&id, 0).expect("still published"); + assert!( + probe_oks.snapshot_at(&as_of_time).is_some(), + "an import that has not downgraded must keep `as_of` readable" + ); + } + + // The consumer downgrades, as a join does once its other input has advanced. The traces + // stay alive throughout, which is the point: this is a downgrade, not a release. + oks_trace.set_logical_compaction(target.borrow()); + oks_trace.set_physical_compaction(target.borrow()); + errs_trace.set_logical_compaction(target.borrow()); + errs_trace.set_physical_compaction(target.borrow()); + assert_eq!( + oks_trace.get_logical_compaction(), + target.borrow(), + "the downgrade must be reflected in what the handle reports holding" + ); + tick( + worker, + &mut oks_input, + Timestamp::from(11_u64), + Timestamp::from(12_u64), + ); + + // The publisher followed the downgrade: `as_of` is no longer readable even though the + // import is still live and still holding at the downgraded frontier. + let (compacted_oks, _compacted_errs) = registry.handles(&id, 0).expect("still published"); + assert!( + compacted_oks.snapshot_at(&as_of_time).is_none(), + "after the downgrade, the publisher must compact past the original `as_of`" + ); + assert!( + compacted_oks + .snapshot_at(&Timestamp::from(10_u64)) + .is_some(), + "the downgraded frontier must still be readable" + ); + drop((oks_trace, errs_trace)); + }); +} + +/// 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_published_index` must +/// panic rather than let the read silently see coalesced data, mirroring the maintenance +/// path's `compaction_frontier` assert in `import_index`. +/// +/// Advances the writer's compaction well past `as_of` with no reader hold registered yet, the +/// same `publish_without_readers_does_not_pin_compaction` scenario `shared_trace.rs` covers, +/// so a freshly minted handle's hold starts at the already-advanced `since`. Importing at +/// `as_of` afterward must panic. +#[mz_ore::test] +#[should_panic(expected = "since")] +fn import_asserts_since_at_most_as_of() { + let id = GlobalId::User(1); + let rows = test_rows(); + let as_of = Antichain::from_elem(Timestamp::from(1_u64)); + let registry = ArrangementSharingRegistry::new(); + + timely::execute_directly(move |worker| { + let (mut oks_input, _errs_input, mut oks_writer, _errs_keep) = worker + .dataflow::(|scope| { + publish_index_with_writer(scope, ®istry, id, rows.clone()) + }); + + // The controller advances compaction well past `as_of`, with no reader hold registered + // yet, and both runtimes apply it. The publisher then advances the published `since` past + // `as_of` on the next tick: no reader hold pins it, and the standing hold has moved with + // the writer floor. + let target = Antichain::from_elem(Timestamp::from(10_u64)); + registry.note_standing_hold(id, 0, &target); + oks_writer.set_logical_compaction(target.borrow()); + oks_writer.set_physical_compaction(target.borrow()); + tick( + worker, + &mut oks_input, + Timestamp::from(5_u64), + Timestamp::from(6_u64), + ); + + // Importing at `as_of` now finds a `since` already beyond it: the assert must panic. + worker.dataflow::(|scope| { + let _ = import_published_index( + scope.clone(), + ®istry, + id, + "Index", + &as_of, + &Antichain::new(), + ); + }); + }); +} + +/// The standing hold keeps `as_of` importable while the importing runtime is behind. +/// +/// This is [`import_asserts_since_at_most_as_of`] with one difference: the importing runtime has +/// not applied the controller's compaction. That is the state the two-runtime split makes +/// reachable, and it is not a protocol error. The controller can create a dataflow at `as_of`, +/// drop it (a cancelled peek releases its read hold), and allow compaction, all before the runtime +/// rendering that dataflow has applied the create. From the controller's side nothing is wrong. +/// The create is still queued, so no reader hold exists to pin the arrangement, and the writer +/// floor alone would let the publisher compact straight past the `as_of` the queued create is +/// about to read at. +/// +/// Asserting the import *succeeds* is the point. The sibling test asserts the panic that a genuine +/// protocol error produces, so between them a mechanism that pinned nothing, or one that pinned +/// everything, fails one of the two. +#[mz_ore::test] +fn standing_hold_pins_until_the_importing_runtime_applies() { + let id = GlobalId::User(1); + let rows = test_rows(); + let as_of_time = Timestamp::from(1_u64); + let as_of = Antichain::from_elem(as_of_time); + let registry = ArrangementSharingRegistry::new(); + + timely::execute_directly(move |worker| { + let (mut oks_input, _errs_input, mut oks_writer, _errs_keep) = worker + .dataflow::(|scope| { + publish_index_with_writer(scope, ®istry, id, rows.clone()) + }); + + // The maintenance runtime applies `AllowCompaction(10)` in full: the writer floor moves and + // its own trace handle compacts. The interactive runtime has not applied the broadcast copy + // of that command, so its standing hold does not move. + let target = Antichain::from_elem(Timestamp::from(10_u64)); + oks_writer.set_logical_compaction(target.borrow()); + oks_writer.set_physical_compaction(target.borrow()); + tick( + worker, + &mut oks_input, + Timestamp::from(5_u64), + Timestamp::from(6_u64), + ); + + // 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::(|scope| { + let (oks_arranged, errs_arranged, _slot) = import_published_index( + scope.clone(), + ®istry, + id, + "Index", + &as_of, + &Antichain::new(), + ); + (oks_arranged.trace, errs_arranged.trace) + }); + + // Scoped: the probe registers a hold of its own at the current `since`, which would pin the + // arrangement at `as_of` and make the release assertion below pass for the wrong reason. + { + let (probe_oks, _probe_errs) = registry.handles(&id, 0).expect("still published"); + assert!( + probe_oks.snapshot_at(&as_of_time).is_some(), + "the standing hold must keep `as_of` readable while the importing runtime is behind" + ); + } + + // Once that runtime applies the compaction, the bound lifts. The live import's own hold + // takes over from here, which is what the sibling hold tests cover. + registry.note_standing_hold(id, 0, &target); + drop((oks_trace, errs_trace)); + tick( + worker, + &mut oks_input, + Timestamp::from(11_u64), + Timestamp::from(12_u64), + ); + let (released_oks, _released_errs) = registry.handles(&id, 0).expect("still published"); + assert!( + released_oks.snapshot_at(&as_of_time).is_none(), + "with the standing hold advanced and no reader left, the arrangement must compact" + ); + }); +} + +/// A two-runtime process's maintenance runtime publishes into the sharing registry. Its +/// interactive peer reads only from the registry, so publication is what keeps interactive +/// peeks from blocking until they time out. +#[mz_ore::test] +fn maintenance_role_publishes() { + assert!(ComputeRuntimeRole::Maintenance.publishes()); +} + +/// A two-runtime process's interactive runtime publishes its transient query outputs into the +/// sharing registry, so a result peek served from the registry can read the output and receive +/// its seal notifications. +#[mz_ore::test] +fn interactive_role_publishes() { + assert!(ComputeRuntimeRole::Interactive.publishes()); +} + +/// The `Solo` (single-runtime) role has no registry peer, so it does not publish. +#[mz_ore::test] +fn solo_role_does_not_publish() { + assert!(!ComputeRuntimeRole::Solo.publishes()); +} diff --git a/src/compute/src/render/threshold.rs b/src/compute/src/render/threshold.rs index df4b5b1bf05d6..250d6e4c73894 100644 --- a/src/compute/src/render/threshold.rs +++ b/src/compute/src/render/threshold.rs @@ -24,6 +24,7 @@ use crate::extensions::arrange::{KeyCollection, MzArrange}; use crate::extensions::reduce::MzReduce; use crate::render::RenderTimestamp; use crate::render::context::{ArrangementFlavor, CollectionBundle, Context}; +use crate::shared_trace::SharedOksEnter; use crate::typedefs::{ErrBatcher, ErrBuilder, RowRowAgent, RowRowEnter, RowRowSpine}; /// Thresholds a dataflow-local ok arrangement, keeping rows with a positive count. @@ -68,6 +69,27 @@ fn threshold_trace<'scope, T: RenderTimestamp>( arrangement.mz_reduce_abelian::<_, RowRowBuilder, RowRowSpine, _>(name, logic) } +/// Thresholds a shared-trace ok arrangement, keeping rows with a positive count. +/// +/// Concrete-spine counterpart to [`threshold_trace`] for the shared-trace input the interactive +/// runtime imports. See [`threshold_local`] for why the input trace type must be concrete here. +fn threshold_shared_trace<'scope, T: RenderTimestamp>( + arrangement: Arranged<'scope, SharedOksEnter>, + name: &str, +) -> Arranged<'scope, RowRowAgent> { + let logic = move |_key: DatumSeq<'_>, s: &[(DatumSeq<'_>, Diff)], t: &mut Vec<(Row, Diff)>| { + for (record, count) in s.iter() { + if count.is_positive() { + t.push(( + > as Cursor>::owned_val(*record), + *count, + )); + } + } + }; + arrangement.mz_reduce_abelian::<_, RowRowBuilder, RowRowSpine, _>(name, logic) +} + /// Build a dataflow to threshold the input data. /// /// This implementation maintains rows in the output, i.e. all rows that have a count greater than @@ -93,6 +115,15 @@ pub fn build_threshold_basic<'scope, T: RenderTimestamp>( ); CollectionBundle::from_expressions(key, ArrangementFlavor::Local(oks, errs)) } + ArrangementFlavor::SharedTrace(_, oks, errs) => { + let oks = threshold_shared_trace(oks, "Threshold shared trace"); + let errs: KeyCollection<_, _, _> = errs.as_collection(|k, _| k.clone()).into(); + let errs = errs + .mz_arrange::, ErrBatcher<_, _>, ErrBuilder<_, _>, _>( + "Arrange threshold basic err", + ); + CollectionBundle::from_expressions(key, ArrangementFlavor::Local(oks, errs)) + } } } diff --git a/src/compute/src/shared_trace.rs b/src/compute/src/shared_trace.rs index bb21c4e7829e7..9914121e22c32 100644 --- a/src/compute/src/shared_trace.rs +++ b/src/compute/src/shared_trace.rs @@ -23,13 +23,6 @@ //! like any trace handle, from any thread. [`SharedTraceHandle::import_snapshot_at`] replays the //! shared arrangement into another scope. -// TODO(CPU-215): drop both once `crate::sharing` calls this module. Nothing in the crate does yet, -// so every item here reads as dead and the re-exports below as unused. The alternative, exporting -// the module publicly to keep the analysis quiet, would ship a crate-internal primitive on the -// public surface. Comments throughout name `crate::sharing` for the same reason: it is the caller -// this module is built for. -#![allow(dead_code, unused_imports)] - mod handle; mod publish; diff --git a/src/compute/src/shared_trace/publish.rs b/src/compute/src/shared_trace/publish.rs index cbe376ef2eba0..049b70c68f612 100644 --- a/src/compute/src/shared_trace/publish.rs +++ b/src/compute/src/shared_trace/publish.rs @@ -23,10 +23,9 @@ use crate::shared_trace::handle::SharedTraceHandle; /// Why a publication point refused an `as_of`. /// /// Read off the point rather than off a handle, so a failure path registers no hold on its way to a -/// panic, and so the caller reports the point that actually refused rather than a sibling. +/// panic, and so the caller reports the point that actually refused rather than a sibling. The +/// refusing `since` is already in [`Published::handle_at`]'s `Err`, so it is not repeated here. pub(crate) struct Diagnostics { - /// The published `since`, the meet of the writer's own compaction frontier and every hold. - pub(crate) since: Antichain, /// The frontier the importing runtime has applied. /// /// A refusal with this AT the refusing `since` means that runtime had already applied the @@ -125,7 +124,6 @@ where /// Why this point would refuse an `as_of`. See [`Diagnostics`]. pub(crate) fn diagnostics(&self) -> Diagnostics { Diagnostics { - since: self.shared.since(), standing_hold: self.standing_hold(), } } diff --git a/src/compute/src/shared_trace/tests.rs b/src/compute/src/shared_trace/tests.rs index 62eab33758654..1ba5ee5d0f537 100644 --- a/src/compute/src/shared_trace/tests.rs +++ b/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); diff --git a/src/compute/src/sharing.rs b/src/compute/src/sharing.rs index cf3c489b4e00c..c802cb3d0e7eb 100644 --- a/src/compute/src/sharing.rs +++ b/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> { + 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, diff --git a/src/compute/src/sharing/tests.rs b/src/compute/src/sharing/tests.rs index f62868633a893..400b54cb61b16 100644 --- a/src/compute/src/sharing/tests.rs +++ b/src/compute/src/sharing/tests.rs @@ -13,7 +13,7 @@ use std::thread; use std::time::{Duration, Instant}; use differential_dataflow::input::Input; -use differential_dataflow::trace::Cursor; +use differential_dataflow::trace::{BatchReader, Cursor, TraceReader}; use mz_repr::{Datum, Row}; use mz_row_spine::{RowRowBatcher, RowRowBuilder}; use mz_timely_util::columnation::ColumnationChunker; @@ -23,6 +23,7 @@ use timely::dataflow::operators::{Capture, Probe}; use timely::progress::Antichain; use crate::extensions::arrange::{KeyCollection, MzArrange}; +use crate::render::context::ArrangementFlavor; use crate::render::errors::DataflowErrorSer; use crate::shared_trace::PublishArrangement; use crate::typedefs::{ErrBatcher, ErrBuilder}; @@ -427,41 +428,39 @@ fn publish_join_input( id: GlobalId, updates: &[Update], seal: u64, -) -> impl FnMut(&mut timely::worker::Worker) + use<> { +) -> JoinInput { let registry_in = registry.clone(); let updates = updates.to_vec(); - let (mut oks_input, mut errs_input, keep) = worker.dataflow::(move |scope| { - let (oks_input, oks_collection) = scope.new_collection::<(Row, Row), Diff>(); - let oks = oks_collection.mz_arrange::< - ColumnationChunker<_>, - RowRowBatcher<_, _>, - RowRowBuilder<_, _>, - RowRowSpine<_, _>, - >("input oks"); + let (mut oks_input, mut errs_input, (oks, errs)) = + worker.dataflow::(move |scope| { + let (oks_input, oks_collection) = scope.new_collection::<(Row, Row), Diff>(); + let oks = oks_collection.mz_arrange::< + ColumnationChunker<_>, + RowRowBatcher<_, _>, + RowRowBuilder<_, _>, + RowRowSpine<_, _>, + >("input oks"); - let (errs_input, errs_collection) = scope.new_collection::(); - let errs = KeyCollection::from(errs_collection).mz_arrange::< - ColumnationChunker<_>, - ErrBatcher<_, _>, - ErrBuilder<_, _>, - ErrSpine<_, _>, - >("input errs"); + let (errs_input, errs_collection) = scope.new_collection::(); + let errs = KeyCollection::from(errs_collection).mz_arrange::< + ColumnationChunker<_>, + ErrBatcher<_, _>, + ErrBuilder<_, _>, + ErrSpine<_, _>, + >("input errs"); - registry_in.publish(id, &oks, &errs); - ( - oks_input, - errs_input, - (oks.trace.clone(), errs.trace.clone()), - ) - }); + registry_in.publish(id, &oks, &errs); + ( + oks_input, + errs_input, + (oks.trace.clone(), errs.trace.clone()), + ) + }); - // Distinct update times in order. Insert each time's updates, then advance and step, so the - // publisher seals and appends one batch per time rather than one batch for everything. let mut times: Vec = updates.iter().map(|&(_, _, t, _)| t).collect(); times.sort_unstable(); times.dedup(); - for &t in × { oks_input.advance_to(Timestamp::from(t)); for &(k, v, ut, d) in &updates { @@ -470,8 +469,6 @@ fn publish_join_input( } } oks_input.flush(); - // Step so the arrange operator observes this frontier and the publisher appends the - // sealed batch to importer queues before the next time is loaded. for _ in 0..16 { worker.step(); } @@ -481,23 +478,51 @@ fn publish_join_input( errs_input.advance_to(Timestamp::from(seal)); errs_input.flush(); - // Return a closure that keeps the input handles and the trace agents alive and continues - // stepping. Dropping the handles would drop the inputs and let the dataflow drain to the empty - // frontier, and dropping the agents would drop the trace, either closing the publication before - // the importer has read it. - // - // Each call also advances the inputs to a fresh filler time, so the arrange operator keeps - // activating the way a live index's does in production. The filler times carry no updates, so - // they add empty seal-only batches and advance `upper` without changing any accumulation. - let mut filler = seal; - move |worker: &mut timely::worker::Worker| { - let _keep = &keep; - filler += 1; - oks_input.advance_to(Timestamp::from(filler)); - oks_input.flush(); - errs_input.advance_to(Timestamp::from(filler)); - errs_input.flush(); + JoinInput { + oks_input, + errs_input, + oks, + errs, + filler: seal, + } +} + +/// A published join input: its inputs, its trace agents, and the filler clock its ticks advance. +/// +/// The inputs keep the dataflow from draining to the empty frontier, and the agents keep the trace +/// alive, either of which would close the publication before an importer has read it. Production +/// keeps the agents in the trace manager. +struct JoinInput { + oks_input: differential_dataflow::input::InputSession, + errs_input: differential_dataflow::input::InputSession, + oks: crate::typedefs::RowRowAgent, + errs: crate::typedefs::ErrAgent, + filler: u64, +} + +impl JoinInput { + /// Advances the inputs to a fresh filler time and steps the worker once, so the arrange + /// operators keep activating the way a live index's do in production. The filler times carry + /// no updates, so they add empty seal-only batches and advance `upper` without changing any + /// accumulation. The agents' physical compaction follows the upper, as the trace manager's + /// maintenance does, so the spines may merge. + fn tick(&mut self, worker: &mut timely::worker::Worker) { + self.filler += 1; + let upper = Antichain::from_elem(Timestamp::from(self.filler)); + self.oks_input.advance_to(Timestamp::from(self.filler)); + self.oks_input.flush(); + self.errs_input.advance_to(Timestamp::from(self.filler)); + self.errs_input.flush(); worker.step(); + self.oks.set_physical_compaction(upper.borrow()); + self.errs.set_physical_compaction(upper.borrow()); + } + + /// Compacts both arrangements to `frontier` on their own agents, as `handle_allow_compaction` + /// does through the trace manager. + fn allow_compaction(&mut self, frontier: &Antichain) { + self.oks.set_logical_compaction(frontier.borrow()); + self.errs.set_logical_compaction(frontier.borrow()); } } @@ -570,8 +595,8 @@ fn join_over_imported_arrangements_matches_direct() { let seal_ts = Timestamp::from(seal); let mut steps = 0; while probe.less_than(&seal_ts) { - keep_a(worker); - keep_b(worker); + keep_a.tick(worker); + keep_b.tick(worker); worker.step(); steps += 1; assert!(steps < 10_000, "join did not seal through {seal_ts:?}"); @@ -730,7 +755,7 @@ fn join_over_point_adopted_late_matches_direct() { // Step with A still unadopted. The import holds A's frontier at the minimum, // so the join frontier cannot pass 0 and no output is produced. for _ in 0..64 { - keep_b(worker); + keep_b.tick(worker); worker.step(); } assert!( @@ -748,7 +773,7 @@ fn join_over_point_adopted_late_matches_direct() { let mut steps = 0; while probe.less_than(&seal_ts) { keep_a(worker); - keep_b(worker); + keep_b.tick(worker); worker.step(); steps += 1; assert!( @@ -1028,3 +1053,404 @@ fn sync_activator_fires_cross_thread() { let _ = activator.activate(); worker.join().expect("worker thread panicked"); } + +/// Consolidates a captured `(Row, Timestamp, Diff)` stream per `(row, time)`, dropping entries +/// whose accumulated diff is zero, and returns them sorted. Shared by the assertions below. +fn consolidate_capture( + rx: mpsc::Receiver< + timely::dataflow::operators::capture::Event>, + >, +) -> Vec<(Row, Timestamp, Diff)> { + let got: Vec<(Row, Timestamp, Diff)> = rx + .extract() + .into_iter() + .flat_map(|(_, data)| data) + .collect(); + let mut consolidated: BTreeMap<(Row, Timestamp), Diff> = BTreeMap::new(); + for (row, time, diff) in got { + *consolidated.entry((row, time)).or_insert(Diff::ZERO) += diff; + } + consolidated + .into_iter() + .filter(|(_, d)| !d.is_zero()) + .map(|((row, t), d)| (row, t, d)) + .collect() +} + +/// Exercises [`ArrangementFlavor::SharedTrace`], the render variant that carries a +/// maintenance-published index imported into the interactive runtime *as an arrangement*. +/// +/// Two `RowRow` indexes are published, imported through `SharedTraceHandle::import_snapshot_at` +/// as a static `as_of` snapshot, entered into a region, and wrapped in +/// `ArrangementFlavor::SharedTrace`, exactly as `import_index_shared` does with its +/// `.enter(self.scope)`. Because the import is a snapshot at `as_of`, every update is coalesced +/// to `as_of`, so key 1's insert and retraction cancel. The flavor is then consumed two ways, +/// standing in for the two downstream operator families that matter: +/// +/// * REDUCE input surface: `ArrangementFlavor::as_collection` reconstructs rows through the +/// render's generic arrangement body, the same surface `as_specific_collection` feeds a +/// reduce. The reconstructed `(key, value)` rows must equal the published rows coalesced at +/// `as_of`. +/// * JOIN surface: the two flavors' arrangements are joined with `join_core`, the differential +/// surface the linear join's `DifferentialDataflow` path calls. The output must equal the +/// direct join. +/// +/// Both consume the imported shared arrangement AS an arrangement, never re-deriving it from a +/// collection. That is the property the `SharedTrace` variant exists to preserve, and the +/// property the prior `CollectionBundle::from_collections` degradation broke. +#[mz_ore::test] +fn shared_trace_flavor_feeds_join_and_reduce() { + let id_a = GlobalId::User(1); + let id_b = GlobalId::User(2); + + // Same inputs as `join_over_imported_arrangements_matches_direct`: key 1 inserted then + // retracted, plus keys 2 and 3, joined against one value per key. + let a: Vec = vec![ + (1, "a", 0, 1), + (2, "b", 0, 1), + (3, "c", 1, 1), + (1, "a", 2, -1), + ]; + let b: Vec = vec![(1, "x", 0, 1), (2, "y", 1, 1), (3, "z", 2, 1)]; + let seal = 3; + // Read as of `seal - 1`, one tick below the sealed upper `{seal}`: `import_snapshot_at` + // emits only once `upper` is strictly beyond `as_of` (as `snapshot_at` does). All input + // times (0, 1, 2) are at or below `as_of`, so they coalesce to it and key 1 cancels. + let as_of_ts = Timestamp::from(seal - 1); + + // The interactive import is a static snapshot at `as_of`, so every update is coalesced to + // `as_of`: all times advance to `as_of_ts` and cancel there. Key 1's insert and retraction + // therefore net to zero, so it appears in neither the join nor the reduce output. + let coalesce_at = |rows: Vec<(Row, Timestamp, Diff)>| -> Vec<(Row, Timestamp, Diff)> { + let mut out: BTreeMap = BTreeMap::new(); + for (row, _time, diff) in rows { + *out.entry(row).or_insert(Diff::ZERO) += diff; + } + let mut v: Vec<_> = out + .into_iter() + .filter(|(_, d)| !d.is_zero()) + .map(|(row, d)| (row, as_of_ts, d)) + .collect(); + v.sort(); + v + }; + + let expected_join_rows = coalesce_at(expected_join(&a, &b)); + + // Reduce-surface oracle: `a`'s updates coalesced at `as_of` into the two-column + // `(key, value)` rows that `as_collection` reconstructs. + let expected_reduce_rows = coalesce_at( + a.iter() + .map(|&(k, v, t, d)| { + ( + Row::pack_slice(&[Datum::Int64(k), Datum::String(v)]), + Timestamp::from(t), + Diff::from(d), + ) + }) + .collect(), + ); + + let (join_tx, join_rx) = mpsc::channel(); + let (reduce_tx, reduce_rx) = mpsc::channel(); + + timely::execute_directly(move |worker| { + let registry = ArrangementSharingRegistry::new(); + + // Maintenance side: publish both indexes, sealing several batches each. + let mut keep_a = publish_join_input(®istry, worker, id_a, &a, seal); + let mut keep_b = publish_join_input(®istry, worker, id_b, &b, seal); + + let worker_index = worker.index(); + let (oks_a, errs_a) = registry.handles(&id_a, worker_index).expect("A published"); + let (oks_b, errs_b) = registry.handles(&id_b, worker_index).expect("B published"); + + let join_probe = ProbeHandle::new(); + let reduce_probe = ProbeHandle::new(); + worker.dataflow::(|scope| { + // Import each index as a static snapshot at `as_of`, with no upper suppression, the + // interactive single-time read path (`import_index_shared`). + let as_of = Antichain::from_elem(as_of_ts); + let until = Antichain::new(); + let arr_a = + oks_a.import_snapshot_at(scope.clone(), "import A", as_of.clone(), until.clone()); + let err_a = errs_a.import_snapshot_at( + scope.clone(), + "import A errs", + as_of.clone(), + until.clone(), + ); + let arr_b = + oks_b.import_snapshot_at(scope.clone(), "import B", as_of.clone(), until.clone()); + let err_b = errs_b.import_snapshot_at(scope.clone(), "import B errs", as_of, until); + + scope.region_named("SharedTraceFlavor", |inner| { + // Enter the region and wrap as `SharedTrace`, mirroring `import_index_shared`. + let flavor_a = + ArrangementFlavor::SharedTrace(id_a, arr_a.enter(inner), err_a.enter(inner)); + let flavor_b = + ArrangementFlavor::SharedTrace(id_b, arr_b.enter(inner), err_b.enter(inner)); + + // REDUCE surface: reconstruct A's rows through the flavor's generic body. + #[allow(deprecated)] + let (oks_coll, _errs_coll) = flavor_a.as_collection(); + oks_coll + .inner + .probe_with(&reduce_probe) + .capture_into(reduce_tx.clone()); + + // JOIN surface: join the two flavors' arrangements. Extracting them by matching + // the variant proves the flavor holds real arrangements the join consumes. + let (join_a, join_b) = match (&flavor_a, &flavor_b) { + ( + ArrangementFlavor::SharedTrace(_, a, _), + ArrangementFlavor::SharedTrace(_, b, _), + ) => (a.clone(), b.clone()), + _ => unreachable!("both flavors constructed as SharedTrace above"), + }; + let joined = join_a.join_core(join_b, |key, v1, v2| { + let row = + Row::pack(key.into_iter().chain(v1.into_iter()).chain(v2.into_iter())); + Some(row) + }); + joined + .inner + .probe_with(&join_probe) + .capture_into(join_tx.clone()); + }); + }); + + // Step until both operators have sealed through the seal frontier, keeping the + // publisher inputs alive so their publication points stay open. + let seal_ts = Timestamp::from(seal); + let mut steps = 0; + while join_probe.less_than(&seal_ts) || reduce_probe.less_than(&seal_ts) { + keep_a.tick(worker); + keep_b.tick(worker); + worker.step(); + steps += 1; + assert!(steps < 10_000, "dataflow did not seal through {seal_ts:?}"); + } + }); + + assert_eq!( + consolidate_capture(join_rx), + expected_join_rows, + "join over SharedTrace flavor diverged from the direct join" + ); + assert_eq!( + consolidate_capture(reduce_rx), + expected_reduce_rows, + "as_collection over SharedTrace flavor diverged from the published rows" + ); +} + +/// A join and a reduce over an arrangement imported at a stale `as_of`, where the publisher's +/// spine has folded the history below it into fewer, larger batches. +/// +/// This is the regime production reads in and no other test reaches. The other join and reduce +/// tests publish four updates and read at `as_of = 0`, so their chains are one batch per time and +/// no merge ever precedes the read time. Here sixteen times are published and the controller then +/// allows compaction to the read time, which is what raises the published `since` and lets the +/// spine fold the batches below it together. +/// +/// The test asserts both halves of the shape rather than assuming them. A merge must have +/// happened, so the import really does seed from a folded chain. And a batch must straddle the +/// `as_of`, because that is the case being covered: an import does not cut at its `as_of`, it is +/// seeded with the whole chain and wrapped in `TraceFrontier`, which advances times instead of +/// cutting. The join and reduce output is the observable, so a straddling batch mishandled would +/// show up as updates at times not before the cut, double counted. +#[mz_ore::test] +fn stale_as_of_import_over_merged_chain_matches_direct() { + let id_a = GlobalId::User(1); + let id_b = GlobalId::User(2); + + // Sixteen distinct times, four keys cycling, so every key accumulates several updates and + // the publisher seals sixteen batches for the spine to merge. + let times = 16u64; + let mut a: Vec = Vec::new(); + let mut b: Vec = Vec::new(); + for t in 0..times { + let key = i64::try_from(t % 4).expect("small") + 1; + a.push((key, "a", t, 1)); + b.push((key, "x", t, 1)); + } + // Retract key 1's first insert at a time still below `as_of`, so the stale read must + // coalesce the pair away rather than report both. + a.push((1, "a", 3, -1)); + let seal = times; + // Read from the middle of the history, far enough below the seal that the batches around it + // have been merged over. + let as_of_ts = Timestamp::from(times / 2); + + // The import advances times at or below `as_of` up to it and leaves later times alone, so + // the oracle is the direct computation over the same advanced updates. + let advance = |updates: &[Update]| -> Vec { + updates + .iter() + .map(|&(k, v, t, d)| (k, v, t.max(u64::from(as_of_ts)), d)) + .collect() + }; + let a_advanced = advance(&a); + let b_advanced = advance(&b); + + let expected_join_rows = expected_join(&a_advanced, &b_advanced); + + // Reduce-surface oracle: A's advanced updates as the two-column `(key, value)` rows that + // `as_collection` reconstructs, consolidated per `(row, time)`. + let expected_reduce_rows = { + let mut out: BTreeMap<(Row, Timestamp), Diff> = BTreeMap::new(); + for &(k, v, t, d) in &a_advanced { + let row = Row::pack_slice(&[Datum::Int64(k), Datum::String(v)]); + *out.entry((row, Timestamp::from(t))).or_insert(Diff::ZERO) += Diff::from(d); + } + let mut v: Vec<_> = out + .into_iter() + .filter(|(_, d)| !d.is_zero()) + .map(|((row, t), d)| (row, t, d)) + .collect(); + v.sort(); + v + }; + + let (join_tx, join_rx) = mpsc::channel(); + let (reduce_tx, reduce_rx) = mpsc::channel(); + + timely::execute_directly(move |worker| { + let registry = ArrangementSharingRegistry::new(); + + // Publish both indexes to completion BEFORE any importer registers. + let mut keep_a = publish_join_input(®istry, worker, id_a, &a, seal); + let mut keep_b = publish_join_input(®istry, worker, id_b, &b, seal); + for _ in 0..64 { + keep_a.tick(worker); + keep_b.tick(worker); + } + + // The controller allows compaction up to the read time, which the writer applies to the + // arrangements as `handle_allow_compaction` does through the trace manager. That raises the + // published `since`, so the spine may coalesce the history below the read time. No importer + // has registered yet, so nothing holds the spines' physical frontiers down and they are free + // to fold those batches together. The extra ticks give them activations to do so. + let allow = Antichain::from_elem(as_of_ts); + keep_a.allow_compaction(&allow); + keep_b.allow_compaction(&allow); + for _ in 0..64 { + keep_a.tick(worker); + keep_b.tick(worker); + } + + let worker_index = worker.index(); + let (oks_a, errs_a) = registry.handles(&id_a, worker_index).expect("A published"); + let (oks_b, errs_b) = registry.handles(&id_b, worker_index).expect("B published"); + + // First half of the premise: the spine folded batches, so the import seeds from a merged + // chain rather than from the one-batch-per-time shape the other tests cover. Each + // published time seals its own `[t, t+1)` batch, so a batch spanning more than one time + // can only come from a merge. + // + // Second half: a batch *does* straddle `as_of`, which is the case this fixture exists to + // cover. An import does not cut at `as_of`, it is seeded with the whole chain and wrapped + // in `TraceFrontier`, which advances times instead. So a straddling batch is harmless and + // the observable is the join and reduce output below, which must still match the direct + // computation. Asserting the straddle rather than its absence keeps this test as the + // detector for a publisher that holds physical compaction down collectively again. + let mut merged = false; + let mut straddles_as_of = false; + oks_a.map_batches(|batch| { + let lower = batch.lower().elements().first().copied(); + let upper = batch.upper().elements().first().copied(); + if let (Some(lower), Some(upper)) = (lower, upper) { + if upper.saturating_sub(lower) > Timestamp::from(1_u64) { + merged = true; + } + if lower < as_of_ts && as_of_ts < upper { + straddles_as_of = true; + } + } + }); + assert!( + merged, + "no published batch spans more than one time; the spine did not merge and the test \ + is not exercising the merged-chain cut" + ); + assert!( + straddles_as_of, + "no published batch straddles as_of {as_of_ts:?}, so this fixture is not reaching \ + the case it exists for: an import whose `as_of` falls inside a batch. If this \ + fires, the publisher has gone back to holding physical compaction down to a \ + collective floor such as the published `since`, which stops the spine merging \ + across `as_of` at all" + ); + + let join_probe = ProbeHandle::new(); + let reduce_probe = ProbeHandle::new(); + worker.dataflow::(|scope| { + let as_of = Antichain::from_elem(as_of_ts); + let until = Antichain::new(); + let arr_a = + oks_a.import_snapshot_at(scope.clone(), "import A", as_of.clone(), until.clone()); + let err_a = errs_a.import_snapshot_at( + scope.clone(), + "import A errs", + as_of.clone(), + until.clone(), + ); + let arr_b = + oks_b.import_snapshot_at(scope.clone(), "import B", as_of.clone(), until.clone()); + let err_b = errs_b.import_snapshot_at(scope.clone(), "import B errs", as_of, until); + + scope.region_named("SharedTraceFlavor", |inner| { + let flavor_a = + ArrangementFlavor::SharedTrace(id_a, arr_a.enter(inner), err_a.enter(inner)); + let flavor_b = + ArrangementFlavor::SharedTrace(id_b, arr_b.enter(inner), err_b.enter(inner)); + + #[allow(deprecated)] + let (oks_coll, _errs_coll) = flavor_a.as_collection(); + oks_coll + .inner + .probe_with(&reduce_probe) + .capture_into(reduce_tx.clone()); + + let (join_a, join_b) = match (&flavor_a, &flavor_b) { + ( + ArrangementFlavor::SharedTrace(_, a, _), + ArrangementFlavor::SharedTrace(_, b, _), + ) => (a.clone(), b.clone()), + _ => unreachable!("both flavors constructed as SharedTrace above"), + }; + let joined = join_a.join_core(join_b, |key, v1, v2| { + let row = + Row::pack(key.into_iter().chain(v1.into_iter()).chain(v2.into_iter())); + Some(row) + }); + joined + .inner + .probe_with(&join_probe) + .capture_into(join_tx.clone()); + }); + }); + + let seal_ts = Timestamp::from(seal); + let mut steps = 0; + while join_probe.less_than(&seal_ts) || reduce_probe.less_than(&seal_ts) { + keep_a.tick(worker); + keep_b.tick(worker); + worker.step(); + steps += 1; + assert!(steps < 10_000, "dataflow did not seal through {seal_ts:?}"); + } + }); + + assert_eq!( + consolidate_capture(join_rx), + expected_join_rows, + "join over a merged chain read at a stale as_of diverged from the direct join" + ); + assert_eq!( + consolidate_capture(reduce_rx), + expected_reduce_rows, + "as_collection over a merged chain read at a stale as_of diverged from the published rows" + ); +}