diff --git a/Cargo.lock b/Cargo.lock index 629bcc067ae1e..c9d5350de753c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3313,8 +3313,7 @@ checksum = "56254986775e3233ffa9c4d7d3faaf6d36a2c09d30b20687e9f88bc8bafc16c8" [[package]] name = "differential-dataflow" version = "0.25.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2677690d1dfbf4d8427eef2fdcf350ab0b258d3be20d01dd37dca07ca6f50bcc" +source = "git+https://github.com/TimelyDataflow/differential-dataflow?branch=master-next#8fd9cae7631fb4baad912be39f39a835fcee61ba" dependencies = [ "columnar", "columnation", @@ -3328,8 +3327,7 @@ dependencies = [ [[package]] name = "differential-dogs3" version = "0.25.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "52c9eee4acc936da45f096424617c91df981c619a92f3f8b4b2a324db7969b34" +source = "git+https://github.com/TimelyDataflow/differential-dataflow?branch=master-next#8fd9cae7631fb4baad912be39f39a835fcee61ba" dependencies = [ "differential-dataflow", "serde", @@ -11091,7 +11089,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck", - "itertools 0.14.0", + "itertools 0.10.5", "log", "multimap", "petgraph", @@ -11112,7 +11110,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.10.5", "proc-macro2", "quote", "syn 2.0.119", @@ -13523,8 +13521,7 @@ dependencies = [ [[package]] name = "timely" version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ced05836c8e471cc85b5a12fbbe9a72568d98ea3b5a79e4a965918bc002108d9" +source = "git+https://github.com/TimelyDataflow/timely-dataflow#4eb3ee09681c209ebcaeced73548a9fbf8e1c71e" dependencies = [ "bincode", "byteorder", @@ -13543,14 +13540,12 @@ dependencies = [ [[package]] name = "timely_bytes" version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ad328384fad249cbe20f742d842ee9e3d2c6086bcde6277fb11bb154f21fb92e" +source = "git+https://github.com/TimelyDataflow/timely-dataflow#4eb3ee09681c209ebcaeced73548a9fbf8e1c71e" [[package]] name = "timely_communication" version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fe397622e1c1957f3486efa0f706908b245c6ee73e7634ddaaf72f48bb3687dc" +source = "git+https://github.com/TimelyDataflow/timely-dataflow#4eb3ee09681c209ebcaeced73548a9fbf8e1c71e" dependencies = [ "byteorder", "columnar", @@ -13564,14 +13559,12 @@ dependencies = [ [[package]] name = "timely_container" version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "de8c518a66e00ef8af9e9fae7e01b7d2628b970948f32855dfc89e503403ff5a" +source = "git+https://github.com/TimelyDataflow/timely-dataflow#4eb3ee09681c209ebcaeced73548a9fbf8e1c71e" [[package]] name = "timely_logging" version = "0.31.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ae9187f19aec95c932196b95c1981708090d9c35213ae266d7bac0f32432886" +source = "git+https://github.com/TimelyDataflow/timely-dataflow#4eb3ee09681c209ebcaeced73548a9fbf8e1c71e" dependencies = [ "timely_container", ] diff --git a/Cargo.toml b/Cargo.toml index 27422005f8eb2..7c12acc1e9781 100644 --- a/Cargo.toml +++ b/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 diff --git a/src/compute/benches/columnar_merge_batcher_row.rs b/src/compute/benches/columnar_merge_batcher_row.rs index d0e7a1cab09a2..fa1f3150eb147 100644 --- a/src/compute/benches/columnar_merge_batcher_row.rs +++ b/src/compute/benches/columnar_merge_batcher_row.rs @@ -28,17 +28,17 @@ //! consume the same pre-built [`Column`] 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`, chunker produces -/// `ColumnationStack` chunks, merger operates on those. -type ColumnationBatcher = MergeBatcher>; -/// Chunker feeding [`ColumnationBatcher`]. -type ColumnationBatcherChunker = Chunker>; - -/// All-`Column` path: input is `Column`, chunker produces -/// `Column` chunks, merger operates on those. -type ColumnBatcher = MergeBatcher>; -/// Chunker feeding [`ColumnBatcher`]. -type ColumnBatcherChunker = ColumnChunker; - /// 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]) -> Vec> { .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(prebuilt_rounds: &[Column]) +/// 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(prebuilt_rounds: &[Column]) where - B: Batcher