deps: build against differential-dataflow master-next and timely master

Cargo.toml

@@ -686,6 +686,12 @@ iceberg-storage-opendal = { git = "https://github.com/MaterializeInc/iceberg-rus
 # Hopefully we can upstream these changes eventually.
 # Additionally we keep the apache crates updated to latest versions.
 duckdb = { git = "https://github.com/MaterializeInc/duckdb-rs.git", rev = "752c7efe2582" }
+# Build against the next differential-dataflow, which pins timely by git in turn.
+#
+# `branch` is not reproducible: pin a `rev` (or a release) before this merges.
+differential-dataflow = { git = "https://github.com/TimelyDataflow/differential-dataflow", branch = "master-next" }
+differential-dogs3 = { git = "https://github.com/TimelyDataflow/differential-dataflow", branch = "master-next" }
+timely = { git = "https://github.com/TimelyDataflow/timely-dataflow" }
 
 
 # BEGIN LINT CONFIG

src/compute/benches/columnar_merge_batcher_row.rs

@@ -28,17 +28,17 @@
 //! consume the same pre-built [`Column<Tuple>`] inputs so the chunker
 //! sees identical input shape.
 
-use std::mem::size_of;
-
 use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
-use differential_dataflow::trace::Batcher;
-use differential_dataflow::trace::implementations::merge_batcher::MergeBatcher;
+use differential_dataflow::batcher::Batcher;
+use differential_dataflow::trace::implementations::merge_batcher::Merger;
 use mz_ore::cast::{CastFrom, CastLossy, ReinterpretCast};
 use mz_repr::{Datum, Row};
 use mz_timely_util::columnar::Column;
 use mz_timely_util::columnar::batcher::{Chunker, ColumnChunker, ColumnMerger};
 use mz_timely_util::columnation::{ColInternalMerger, ColumnationStack};
+use mz_timely_util::operator::ConsolidatingBatcher;
 use rand::{Rng, SeedableRng, rngs::StdRng};
+use std::mem::size_of;
 use timely::container::ContainerBuilder;
 use timely::container::PushInto;
 use timely::progress::Antichain;
@@ -48,18 +48,6 @@ type Time = u64;
 type Diff = i64;
 type Tuple = (Data, Time, Diff);
 
-/// Legacy path: input is `Column<Tuple>`, chunker produces
-/// `ColumnationStack<Tuple>` chunks, merger operates on those.
-type ColumnationBatcher = MergeBatcher<ColInternalMerger<Data, Time, Diff>>;
-/// Chunker feeding [`ColumnationBatcher`].
-type ColumnationBatcherChunker = Chunker<ColumnationStack<Tuple>>;
-
-/// All-`Column` path: input is `Column<Tuple>`, chunker produces
-/// `Column<Tuple>` chunks, merger operates on those.
-type ColumnBatcher = MergeBatcher<ColumnMerger<Data, Time, Diff>>;
-/// Chunker feeding [`ColumnBatcher`].
-type ColumnBatcherChunker = ColumnChunker<Tuple>;
-
 /// Per-side payload-byte targets. Element counts are derived from
 /// [`ROW_PAYLOAD_BYTES`]. Same shape as [`columnar_merger_row`] so
 /// numbers can be cross-referenced.
@@ -193,31 +181,21 @@ fn rounds_to_columns(rounds: &[Vec<Tuple>]) -> Vec<Column<Tuple>> {
         .collect()
 }
 
-/// Run a fresh `B` over a clone of `prebuilt_rounds`, pushing each round
-/// then sealing at `+inf` so all data flows through extract. Generic over
-/// `B::Output` because the columnation path produces `ColumnationStack`
-/// chunks while the column path produces `Column` chunks; the no-op
-/// builder accommodates either.
-fn drive_batcher<B, Chu>(prebuilt_rounds: &[Column<Tuple>])
+/// Run a fresh `Chu`/`M` batcher over a clone of `prebuilt_rounds`, inserting each round then
+/// extracting at `+inf` so all data flows through the merge ladder.
+fn drive_batcher<Chu, M>(prebuilt_rounds: &[Column<Tuple>])
 where
-    B: Batcher<Time = Time>,
-    Chu: ContainerBuilder<Container = B::Output> + for<'a> PushInto<&'a mut Column<Tuple>>,
-    B::Output: 'static,
+    M: Merger<Time = Time>,
+    Chu: ContainerBuilder<Container = M::Chunk> + for<'a> PushInto<&'a mut Column<Tuple>>,
+    ConsolidatingBatcher<Chu, M>: Batcher<Column<Tuple>, Time = Time>,
 {
-    let mut batcher = B::new(None, 0);
-    let mut chunker = Chu::default();
+    let mut batcher = ConsolidatingBatcher::<Chu, M>::new(None, 0);
     for round in prebuilt_rounds {
         let mut col = round.clone();
-        chunker.push_into(&mut col);
-        while let Some(chunk) = chunker.extract() {
-            batcher.push_into(std::mem::take(chunk));
-        }
-    }
-    while let Some(chunk) = chunker.finish() {
-        batcher.push_into(std::mem::take(chunk));
+        batcher.insert(&mut col);
     }
     let upper = Antichain::from_elem(Time::MAX);
-    let _ = batcher.seal(upper);
+    let _ = batcher.extract(upper.borrow());
 }
 
 fn bench_batcher(c: &mut Criterion) {
@@ -266,7 +244,10 @@ fn bench_batcher(c: &mut Criterion) {
                 bencher.iter_batched(
                     || prebuilt.clone(),
                     |rounds| {
-                        drive_batcher::<ColumnationBatcher, ColumnationBatcherChunker>(&rounds)
+                        drive_batcher::<
+                            Chunker<ColumnationStack<Tuple>>,
+                            ColInternalMerger<Data, Time, Diff>,
+                        >(&rounds)
                     },
                     BatchSize::LargeInput,
                 );
@@ -275,7 +256,11 @@ fn bench_batcher(c: &mut Criterion) {
             group.bench_with_input(BenchmarkId::new("column", &id), &(), |bencher, _| {
                 bencher.iter_batched(
                     || prebuilt.clone(),
-                    |rounds| drive_batcher::<ColumnBatcher, ColumnBatcherChunker>(&rounds),
+                    |rounds| {
+                        drive_batcher::<ColumnChunker<Tuple>, ColumnMerger<Data, Time, Diff>>(
+                            &rounds,
+                        )
+                    },
                     BatchSize::LargeInput,
                 );
             });

src/compute/src/arrangement/manager.rs

@@ -9,16 +9,15 @@
 
 //! Management of arrangements across dataflows.
 
-use std::any::Any;
-use std::collections::BTreeMap;
-use std::rc::Rc;
-use std::time::Instant;
-
 use differential_dataflow::lattice::antichain_join;
 use differential_dataflow::operators::arrange::{Arranged, ShutdownButton, TraceAgent};
-use differential_dataflow::trace::TraceReader;
 use differential_dataflow::trace::wrappers::frontier::TraceFrontier;
+use differential_dataflow::trace::{Span, TraceReader};
 use mz_repr::{Diff, GlobalId, Timestamp};
+use std::any::Any;
+use std::collections::BTreeMap;
+use std::rc::Rc;
+use std::time::Instant;
 use timely::PartialOrder;
 use timely::dataflow::Scope;
 use timely::dataflow::operators::CapabilitySet;
@@ -165,6 +164,13 @@ where
     type Time = Tr::Time;
     type Batch = Tr::Batch;
 
+    fn spans_through(
+        &mut self,
+        upper: AntichainRef<Self::Time>,
+    ) -> Option<Vec<Span<Self::Time, Self::Batch>>> {
+        self.trace.spans_through(upper)
+    }
+
     fn batches_through(&mut self, upper: AntichainRef<Self::Time>) -> Option<Vec<Self::Batch>> {
         self.trace.batches_through(upper)
     }
@@ -204,8 +210,8 @@ where
         self.trace.get_physical_compaction()
     }
 
-    fn map_batches<F: FnMut(&Self::Batch)>(&self, f: F) {
-        self.trace.map_batches(f)
+    fn map_spans<F: FnMut(&Span<Self::Time, Self::Batch>)>(&self, f: F) {
+        self.trace.map_spans(f)
     }
 }
 

src/compute/src/compute_state.rs

@@ -5,15 +5,6 @@
 
 //! Worker-local state for compute timely instances.
 
-use std::any::Any;
-use std::cell::RefCell;
-use std::cmp::Ordering;
-use std::collections::{BTreeMap, BTreeSet, VecDeque};
-use std::num::NonZeroUsize;
-use std::rc::Rc;
-use std::sync::Arc;
-use std::time::{Duration, Instant};
-
 use differential_dataflow::Hashable;
 use differential_dataflow::lattice::Lattice;
 use differential_dataflow::trace::TraceReader;
@@ -56,6 +47,14 @@ use mz_storage_types::sources::SourceData;
 use mz_storage_types::time_dependence::TimeDependence;
 use mz_txn_wal::operator::TxnsContext;
 use mz_txn_wal::txn_cache::TxnsCache;
+use std::any::Any;
+use std::cell::RefCell;
+use std::cmp::Ordering;
+use std::collections::{BTreeMap, BTreeSet, VecDeque};
+use std::num::NonZeroUsize;
+use std::rc::Rc;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
 use timely::dataflow::operators::probe;
 use timely::order::PartialOrder;
 use timely::progress::frontier::Antichain;
@@ -67,6 +66,7 @@ use uuid::Uuid;
 use crate::arrangement::manager::{TraceBundle, TraceManager};
 use crate::compute_state::peek_budget::InlineBudget;
 use crate::compute_state::peek_metrics::{IndexPeekMetrics, PeekWalkMetrics};
+
 pub(crate) use crate::compute_state::peek_offload::PeekPermits;
 use crate::compute_state::peek_offload::{OffloadConfig, OffloadedPeek};
 use crate::compute_state::peek_scan::{
@@ -76,7 +76,7 @@ use crate::logging;
 use crate::logging::compute::{CollectionLogging, ComputeEvent, PeekEvent};
 use crate::logging::initialize::LoggingTraces;
 use crate::metrics::{CollectionMetrics, WorkerMetrics};
-use crate::render::{LinearJoinSpec, StartSignal};
+use crate::render::StartSignal;
 use crate::server::{ComputeInstanceContext, ResponseSender};
 
 mod error_scan;
@@ -221,7 +221,6 @@ pub struct ComputeState {
     /// Max size in bytes of any result.
     max_result_size: u64,
     /// Specification for rendering linear joins.
-    pub linear_join_spec: LinearJoinSpec,
     /// Metrics for this worker.
     pub metrics: WorkerMetrics,
     /// A process-global handle to tracing configuration.
@@ -337,7 +336,6 @@ impl ComputeState {
             txns_ctx,
             command_history,
             max_result_size: u64::MAX,
-            linear_join_spec: Default::default(),
             metrics,
             tracing_handle,
             context,
@@ -390,8 +388,6 @@ impl ComputeState {
 
         let config = &self.worker_config;
 
-        self.linear_join_spec = LinearJoinSpec::from_config(config);
-
         if ENABLE_LGALLOC.get(config) {
             if let Some(path) = &self.context.scratch_directory {
                 let clear_bytes = LGALLOC_SLOW_CLEAR_BYTES.get(config);

src/compute/src/compute_state/error_scan.rs

@@ -6,12 +6,13 @@
 //! An index peek's walk over its error trace, the phase that runs before
 //! [`PeekResultIterator`](super::peek_result_iterator::PeekResultIterator) walks the ok trace.
 
-use std::time::{Duration, Instant};
-
+use differential_dataflow::trace::cursor::cursor_list;
 use differential_dataflow::trace::{Cursor, TraceReader};
 use mz_compute_client::protocol::response::PeekError;
 use mz_repr::{Diff, GlobalId, Timestamp};
+use std::time::{Duration, Instant};
 use timely::order::PartialOrder;
+use timely::progress::Antichain;
 use tracing::error;
 
 use crate::arrangement::manager::PaddedTrace;
@@ -55,7 +56,10 @@ impl ErrorScan {
     /// supply through [`ErrorScan::set_row_iteration_limit`] before each step.
     pub(super) fn new(errs: &mut ErrsHandle) -> Self {
         let scan_start = Instant::now();
-        let (cursor, storage) = errs.cursor();
+        let batches = errs
+            .batches_through(Antichain::new().borrow())
+            .expect("trace is not compacted beyond the empty frontier");
+        let (cursor, storage) = cursor_list(batches);
         let mut scan = Self::from_cursor(cursor, storage);
         scan.scan_time = scan_start.elapsed();
         scan