Columnar merge batcher

Signed-off-by: Moritz Hoffmann <[email protected]>

Author: antiguru

Cargo.lock

@@ -18,7 +18,7 @@ bytesize = "1.3.0"
 chrono = { version = "0.4.39", default-features = false, features = ["std"] }
 dec = "0.4.8"
 derivative = "2.2.0"
-differential-dataflow = "0.15.1"
+differential-dataflow = "0.15.2"
 enum-kinds = "0.5.1"
 fail = { version = "0.5.1", features = ["failpoints"] }
 futures = "0.3.31"

src/adapter/Cargo.toml

@@ -18,7 +18,7 @@ bytesize = "1.3.0"
 chrono = { version = "0.4.39", default-features = false, features = ["std"] }
 clap = { version = "4.5.23", features = ["derive"] }
 derivative = "2.2.0"
-differential-dataflow = "0.15.1"
+differential-dataflow = "0.15.2"
 futures = "0.3.31"
 ipnet = "2.11.0"
 itertools = "0.14.0"

src/catalog/Cargo.toml

@@ -13,7 +13,7 @@ workspace = true
 anyhow = "1.0.98"
 async-trait = "0.1.88"
 crossbeam-channel = "0.5.15"
-differential-dataflow = "0.15.1"
+differential-dataflow = "0.15.2"
 futures = "0.3.31"
 lgalloc = "0.5.0"
 mz-cluster-client = { path = "../cluster-client" }

src/cluster/Cargo.toml

@@ -16,7 +16,7 @@ bytesize = "1.3.0"
 chrono = { version = "0.4.39", default-features = false, features = ["std"] }
 crossbeam-channel = "0.5.15"
 derivative = "2.2.0"
-differential-dataflow = "0.15.1"
+differential-dataflow = "0.15.2"
 futures = "0.3.31"
 http = "1.2.0"
 mz-build-info = { path = "../build-info" }

src/compute-client/Cargo.toml

@@ -12,7 +12,7 @@ workspace = true
 [dependencies]
 columnar = "0.5.0"
 columnation = "0.1.0"
-differential-dataflow = "0.15.1"
+differential-dataflow = "0.15.2"
 itertools = "0.14.0"
 mz-dyncfg = { path = "../dyncfg" }
 mz-expr = { path = "../expr" }

src/compute-types/Cargo.toml

@@ -16,8 +16,8 @@ bytesize = "1.3.0"
 columnar = "0.5.0"
 crossbeam-channel = "0.5.15"
 dec = { version = "0.4.8", features = ["serde"] }
-differential-dataflow = "0.15.1"
-differential-dogs3 = "0.1.12"
+differential-dataflow = "0.15.2"
+differential-dogs3 = "0.1.13"
 futures = "0.3.31"
 itertools = "0.14.0"
 lgalloc = "0.5"

src/compute/Cargo.toml

@@ -36,7 +36,7 @@ use crate::logging::{
     DifferentialLog, EventQueue, LogCollection, LogVariant, SharedLoggingState,
     consolidate_and_pack,
 };
-use crate::row_spine::RowRowBuilder;
+use crate::row_spine::RowRowBuilderColumn;
 use crate::typedefs::{KeyBatcher, RowRowSpine};
 
 /// The return type of [`construct`].
@@ -169,7 +169,7 @@ pub(super) fn construct<G: Scope<Timestamp = Timestamp>>(
             let variant = LogVariant::Differential(variant);
             if config.index_logs.contains_key(&variant) {
                 let trace = collection
-                    .mz_arrange_core::<_, Col2ValBatcher<_, _, _, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
+                    .mz_arrange_core::<_, Col2ValBatcher<_, _, _, _>, RowRowBuilderColumn<_, _>, RowRowSpine<_, _>>(
                         ExchangeCore::<ColumnBuilder<_>, _>::new_core(columnar_exchange::<mz_repr::Row, mz_repr::Row, Timestamp, mz_repr::Diff>),
                         &format!("Arrange {variant:?}"),
                     )

src/compute/src/logging/differential.rs

@@ -14,6 +14,7 @@ use std::convert::TryInto;
 use std::rc::Rc;
 use std::time::Duration;
 
+use columnar::Columnar;
 use mz_compute_client::logging::LoggingConfig;
 use mz_ore::cast::CastFrom;
 use mz_repr::{Datum, Diff, Row, Timestamp};
@@ -26,7 +27,7 @@ use timely::dataflow::channels::pact::ExchangeCore;
 use crate::extensions::arrange::MzArrangeCore;
 use crate::logging::initialize::ReachabilityEvent;
 use crate::logging::{EventQueue, LogCollection, LogVariant, TimelyLog, consolidate_and_pack};
-use crate::row_spine::RowRowBuilder;
+use crate::row_spine::RowRowBuilderColumn;
 use crate::typedefs::RowRowSpine;
 
 /// The return type of [`construct`].
@@ -83,14 +84,14 @@ pub(super) fn construct<G: Scope<Timestamp = Timestamp>>(
             TimelyLog::Reachability,
             move |((datum, ()), time, diff), packer, session| {
                 let (update_type, operator_id, source, port, ts) = datum;
-                let update_type = if *update_type { "source" } else { "target" };
+                let update_type = if update_type { "source" } else { "target" };
                 let data = packer.pack_slice(&[
-                    Datum::UInt64(u64::cast_from(*operator_id)),
+                    Datum::UInt64(u64::cast_from(operator_id)),
                     Datum::UInt64(u64::cast_from(worker_id)),
-                    Datum::UInt64(u64::cast_from(*source)),
-                    Datum::UInt64(u64::cast_from(*port)),
+                    Datum::UInt64(u64::cast_from(source)),
+                    Datum::UInt64(u64::cast_from(port)),
                     Datum::String(update_type),
-                    Datum::from(*ts),
+                    Datum::from(Timestamp::into_owned(ts)),
                 ]);
                 session.give((data, time, diff));
             }
@@ -100,7 +101,7 @@ pub(super) fn construct<G: Scope<Timestamp = Timestamp>>(
         for variant in logs_active {
             if config.index_logs.contains_key(&variant) {
                 let trace = updates
-                    .mz_arrange_core::<_, Col2ValBatcher<_, _, _, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
+                    .mz_arrange_core::<_, Col2ValBatcher<_, _, _, _>, RowRowBuilderColumn<_, _>, RowRowSpine<_, _>>(
                         ExchangeCore::<ColumnBuilder<_>, _>::new_core(columnar_exchange::<Row, Row, Timestamp, Diff>),
                         &format!("Arrange {variant:?}"),
                     )

src/compute/src/logging/reachability.rs

@@ -39,7 +39,7 @@ use crate::extensions::arrange::MzArrangeCore;
 use crate::logging::compute::{ComputeEvent, DataflowShutdown};
 use crate::logging::{EventQueue, LogVariant, TimelyLog};
 use crate::logging::{LogCollection, SharedLoggingState, consolidate_and_pack};
-use crate::row_spine::RowRowBuilder;
+use crate::row_spine::RowRowBuilderColumn;
 use crate::typedefs::{KeyBatcher, KeyValBatcher, RowRowSpine};
 
 /// The return type of [`construct`].
@@ -312,7 +312,7 @@ pub(super) fn construct<G: Scope<Timestamp = Timestamp>>(
             let variant = LogVariant::Timely(variant);
             if config.index_logs.contains_key(&variant) {
                 let trace = collection
-                    .mz_arrange_core::<_, Col2ValBatcher<_, _, _, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
+                    .mz_arrange_core::<_, Col2ValBatcher<_, _, _, _>, RowRowBuilderColumn<_, _>, RowRowSpine<_, _>>(
                         ExchangeCore::<ColumnBuilder<_>, _>::new_core(columnar_exchange::<mz_repr::Row, mz_repr::Row, Timestamp, Diff>),
                         &format!("Arrange {variant:?}"),
                     )

src/compute/src/logging/timely.rs

@@ -48,7 +48,7 @@ use crate::compute_state::{ComputeState, HydrationEvent};
 use crate::extensions::arrange::{KeyCollection, MzArrange, MzArrangeCore};
 use crate::render::errors::ErrorLogger;
 use crate::render::{LinearJoinSpec, RenderTimestamp};
-use crate::row_spine::{DatumSeq, RowRowBuilder};
+use crate::row_spine::{DatumSeq, RowRowBuilderColumn};
 use crate::typedefs::{
     ErrAgent, ErrBatcher, ErrBuilder, ErrEnter, ErrSpine, RowRowAgent, RowRowEnter, RowRowSpine,
 };
@@ -942,7 +942,7 @@ where
                 },
             );
         let oks = oks
-            .mz_arrange_core::<_, Col2ValBatcher<_, _,_, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
+            .mz_arrange_core::<_, Col2ValBatcher<_, _,_, _>, RowRowBuilderColumn<_, _>, RowRowSpine<_, _>>(
                 ExchangeCore::<ColumnBuilder<_>, _>::new_core(columnar_exchange::<Row, Row, S::Timestamp, Diff>),name
             );
         (oks, errs.as_collection())

src/compute/src/render/context.rs

@@ -39,7 +39,7 @@ use crate::extensions::arrange::MzArrangeCore;
 use crate::render::RenderTimestamp;
 use crate::render::context::{ArrangementFlavor, CollectionBundle, Context, ShutdownToken};
 use crate::render::join::mz_join_core::mz_join_core;
-use crate::row_spine::{RowRowBuilder, RowRowSpine};
+use crate::row_spine::{RowRowBuilderColumn, RowRowSpine};
 use crate::typedefs::{RowRowAgent, RowRowEnter};
 
 /// Available linear join implementations.
@@ -400,7 +400,7 @@ where
             errors.push(errs.as_collection());
 
             let arranged = keyed
-                .mz_arrange_core::<_, Col2ValBatcher<_, _,_, _>, RowRowBuilder<_, _>, RowRowSpine<_, _>>(
+                .mz_arrange_core::<_, Col2ValBatcher<_, _,_, _>, RowRowBuilderColumn<_, _>, RowRowSpine<_, _>>(
                     ExchangeCore::<ColumnBuilder<_>, _>::new_core(columnar_exchange::<Row, Row, S::Timestamp, Diff>),"JoinStage"
                 );
             joined = JoinedFlavor::Local(arranged);

src/compute/src/render/join/linear_join.rs

@@ -11,8 +11,8 @@ pub use self::container::DatumContainer;
 pub use self::container::DatumSeq;
 pub use self::offset_opt::OffsetOptimized;
 pub use self::spines::{
-    RowBatcher, RowBuilder, RowRowBatcher, RowRowBuilder, RowRowSpine, RowSpine, RowValBatcher,
-    RowValBuilder, RowValSpine,
+    RowBatcher, RowBuilder, RowRowBatcher, RowRowBuilder, RowRowBuilderColumn, RowRowSpine,
+    RowSpine, RowValBatcher, RowValBuilder, RowValSpine,
 };
 use differential_dataflow::trace::implementations::OffsetList;
 
@@ -28,6 +28,7 @@ mod spines {
     use differential_dataflow::trace::implementations::spine_fueled::Spine;
     use differential_dataflow::trace::rc_blanket_impls::RcBuilder;
     use mz_repr::Row;
+    use mz_timely_util::containers::ColumnValBuilder;
 
     use crate::row_spine::{DatumContainer, OffsetOptimized};
     use crate::typedefs::{KeyBatcher, KeyValBatcher};
@@ -36,6 +37,8 @@ mod spines {
     pub type RowRowBatcher<T, R> = KeyValBatcher<Row, Row, T, R>;
     pub type RowRowBuilder<T, R> =
         RcBuilder<OrdValBuilder<RowRowLayout<((Row, Row), T, R)>, TimelyStack<((Row, Row), T, R)>>>;
+    pub type RowRowBuilderColumn<T, R> =
+        RcBuilder<ColumnValBuilder<RowRowLayout<((Row, Row), T, R)>>>;
 
     pub type RowValSpine<V, T, R> = Spine<Rc<OrdValBatch<RowValLayout<((Row, V), T, R)>>>>;
     pub type RowValBatcher<V, T, R> = KeyValBatcher<Row, V, T, R>;

src/compute/src/row_spine.rs

@@ -12,7 +12,7 @@ workspace = true
 [dependencies]
 async-trait = "0.1.88"
 bytes = { version = "1.10.1" }
-differential-dataflow = "0.15.1"
+differential-dataflow = "0.15.2"
 futures = "0.3.31"
 itertools = { version = "0.14.0" }
 mz-ore = { path = "../ore", features = ["process"] }

src/durable-cache/Cargo.toml

@@ -26,7 +26,7 @@ chrono = { version = "0.4.39", default-features = false, features = ["std"] }
 chrono-tz = { version = "0.8.1", features = ["serde", "case-insensitive"] }
 crc32fast = "1.4.2"
 csv = "1.3.1"
-differential-dataflow = "0.15.1"
+differential-dataflow = "0.15.2"
 dec = "0.4.8"
 derivative = "2.2.0"
 encoding = "0.2.0"

src/expr/Cargo.toml

@@ -20,7 +20,7 @@ byteorder = "1.4.3"
 bytes = "1.10.1"
 chrono = { version = "0.4.39", default-features = false, features = ["std"] }
 clap = { version = "4.5.23", features = ["derive"] }
-differential-dataflow = "0.15.1"
+differential-dataflow = "0.15.2"
 itertools = "0.14.0"
 maplit = "1.0.2"
 mz-avro = { path = "../avro", features = ["snappy"] }

src/interchange/Cargo.toml

@@ -29,7 +29,7 @@ columnation = { version = "0.1.0", optional = true }
 columnar = { version = "0.5.0", optional = true }
 compact_bytes = { version = "0.1.4", optional = true }
 ctor = { version = "0.4.2", optional = true }
-differential-dataflow = { version = "0.15.1", optional = true }
+differential-dataflow = { version = "0.15.2", optional = true }
 derivative = { version = "2.2.0" }
 either = "1.15.0"
 futures = { version = "0.3.31", optional = true }

src/ore/Cargo.toml

@@ -23,7 +23,7 @@ async-trait = "0.1.88"
 axum = "0.7.5"
 bytes = { version = "1.10.1", features = ["serde"] }
 clap = { version = "4.5.23", features = ["derive", "env"] }
-differential-dataflow = "0.15.1"
+differential-dataflow = "0.15.2"
 futures = "0.3.31"
 humantime = "2.2.0"
 mz-dyncfg = { path = "../dyncfg" }

src/persist-cli/Cargo.toml

@@ -35,7 +35,7 @@ async-stream = "0.3.6"
 async-trait = "0.1.88"
 bytes = { version = "1.10.1", features = ["serde"] }
 clap = { version = "4.5.23", features = ["derive"] }
-differential-dataflow = "0.15.1"
+differential-dataflow = "0.15.2"
 futures = "0.3.31"
 futures-util = "0.3"
 h2 = "0.4.10"

src/persist-client/Cargo.toml

@@ -37,7 +37,7 @@ azure_core = "0.21.0"
 base64 = "0.22.1"
 bytes = "1.10.1"
 deadpool-postgres = "0.10.3"
-differential-dataflow = "0.15.1"
+differential-dataflow = "0.15.2"
 fail = { version = "0.5.1", features = ["failpoints"] }
 futures-util = "0.3.31"
 itertools = "0.14.0"

src/persist/Cargo.toml

@@ -36,7 +36,7 @@ columnation = "0.1.0"
 chrono = { version = "0.4.39", default-features = false, features = ["serde", "std"] }
 compact_bytes = "0.1.4"
 dec = "0.4.8"
-differential-dataflow = "0.15.1"
+differential-dataflow = "0.15.2"
 enum-kinds = "0.5.1"
 hex = "0.4.3"
 itertools = "0.14.0"
@@ -79,6 +79,7 @@ tracing = { version = "0.1.37" }
 # for the tracing feature
 tracing-subscriber = { version = "0.3.19", default-features = false, optional = true }
 workspace-hack = { version = "0.0.0", path = "../workspace-hack", optional = true }
+bytemuck = { version = "1.23.0", features = ["derive"] }
 
 [dev-dependencies]
 criterion = { version = "0.6.0" }