From 60008a2697ef546e5760bfaa29f4f7d2eb9f3b71 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Tue, 21 Jul 2026 14:53:48 +0200 Subject: [PATCH 1/4] compute: Emit dataflow import outputs as the columnar edge Flip the three import producers to the columnar edge via a `vec_to_columnar` leaf-encode at the boundary: source imports in the recursive and non-recursive scopes, and the `SnapshotMode::Exclude` index import. These read row-shaped data from persist or a filtered trace, so encoding at the boundary is the accepted leaf-encode, symmetric to the join outputs and the reduce/join Vec-internal pattern. The batches are already consolidated and the prior `from_collections` was non-consolidating, so a plain `vec_to_columnar` (non-consolidating) is correct; no consolidating builder is added. The persist/trace decode machinery is untouched. `from_collections` -> `from_edge`; err stays `Vec`. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/compute/src/render.rs | 24 ++++++++++++++++++------ 1 file changed, 18 insertions(+), 6 deletions(-) diff --git a/src/compute/src/render.rs b/src/compute/src/render.rs index 4bd99df87abc9..5c008c26253f5 100644 --- a/src/compute/src/render.rs +++ b/src/compute/src/render.rs @@ -166,7 +166,7 @@ use crate::extensions::temporal_bucket::TemporalBucketing; use crate::logging::compute::{ ComputeEvent, DataflowGlobal, LirMapping, LirMetadata, LogDataflowErrors, OperatorHydration, }; -use crate::render::columnar::CollectionEdge; +use crate::render::columnar::{CollectionEdge, vec_to_columnar}; use crate::render::context::{ArrangementFlavor, Context}; use crate::render::errors::DataflowErrorSer; use crate::typedefs::{ErrBatcher, ErrBuilder, ErrSpine, KeyBatcher, MzTimestamp}; @@ -375,8 +375,11 @@ pub fn build_compute_dataflow( ); for (id, (oks, errs)) in imported_sources.into_iter() { - let bundle = crate::render::CollectionBundle::from_collections( - oks.enter(region), + // Imports read row-shaped data from persist; encode it to the + // columnar edge at the boundary. The batches are already + // consolidated, so this leaf-encode is non-consolidating. + let bundle = crate::render::CollectionBundle::from_edge( + CollectionEdge::Columnar(vec_to_columnar(oks.enter(region))), errs.enter(region), ); // Associate collection bundle with the source identifier. @@ -475,8 +478,11 @@ pub fn build_compute_dataflow( ); for (id, (oks, errs)) in imported_sources.into_iter() { - let bundle = crate::render::CollectionBundle::from_collections( - oks.enter_region(region), + // Imports read row-shaped data from persist; encode it to the + // columnar edge at the boundary. The batches are already + // consolidated, so this leaf-encode is non-consolidating. + let bundle = crate::render::CollectionBundle::from_edge( + CollectionEdge::Columnar(vec_to_columnar(oks.enter_region(region))), errs.enter_region(region), ); // Associate collection bundle with the source identifier. @@ -668,7 +674,13 @@ where start_signal, |e, _| e.clone(), ); - CollectionBundle::from_collections(oks, errs) + // The filtered index collection is row-shaped; encode it to + // the columnar edge at the boundary. It is already + // consolidated, so this leaf-encode is non-consolidating. + CollectionBundle::from_edge( + CollectionEdge::Columnar(vec_to_columnar(oks)), + errs, + ) } }; self.update_id(Id::Global(idx.on_id), bundle); From fa26564236e5fbab78cca35a2433321da8a379ee Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Wed, 22 Jul 2026 10:16:31 +0200 Subject: [PATCH 2/4] test: update relations.slt golden for import leaf-encode The import boundary now encodes outputs to the columnar edge, inserting a VecToColumnar operator. Reflect the new conversion channel and the Vec->Column flip on the arrange input in the introspection golden. Co-Authored-By: Claude Opus 4.8 --- test/sqllogictest/introspection/relations.slt | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/test/sqllogictest/introspection/relations.slt b/test/sqllogictest/introspection/relations.slt index 5a36a7cee42a8..c000d2f64bb10 100644 --- a/test/sqllogictest/introspection/relations.slt +++ b/test/sqllogictest/introspection/relations.slt @@ -65,9 +65,10 @@ FormArrangementKey Concatenate alloc::vec::Vec<(mz_compute::render::errors::Da InputRegion:␠materialize.public.test_primary_idx BuildRegion:␠materialize.public.test_primary_idx alloc::vec::Vec<(mz_compute::render::errors::DataflowErrorSer,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> InputRegion:␠materialize.public.test_primary_idx BuildRegion:␠materialize.public.test_primary_idx alloc::vec::Vec<(mz_repr::row::Row,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> LimitProgress(Dataflow:␠materialize.public.test_primary_idx) Probe alloc::vec::Vec<(mz_repr::row::Row,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> -LogOperatorHydration␠(1) FormArrangementKey alloc::vec::Vec<(mz_repr::row::Row,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> +LogOperatorHydration␠(1) FormArrangementKey mz_timely_util::columnar::Column<(mz_repr::row::Row,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> OkErr SuppressEarlyProgress alloc::vec::Vec<(mz_repr::row::Row,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> SuppressEarlyProgress LimitProgress(Dataflow:␠materialize.public.test_primary_idx) alloc::vec::Vec<(mz_repr::row::Row,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> +VecToColumnar BuildingObject(User(2)) mz_timely_util::columnar::Column<(mz_repr::row::Row,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> decode_backpressure_probe(u1) Feedback alloc::vec::Vec expire_stream_at(materialize.public.test_primary_idx_export_index_errs) LogDataflowErrorsStream alloc::vec::Vec)>>>> expire_stream_at(materialize.public.test_primary_idx_export_index_oks) InspectBatch alloc::vec::Vec)>>>> @@ -97,11 +98,12 @@ GROUP BY type; 2 alloc::vec::Vec<(usize,␠mz_persist_client::fetch::ExchangeableBatchPart)> 2 alloc::vec::Vec> 4 alloc::vec::Vec<(core::result::Result,␠(mz_repr::timestamp::Timestamp,␠mz_storage_operators::persist_source::Subtime),␠mz_ore::overflowing::Overflowing)> +4 mz_timely_util::columnar::Column<(mz_repr::row::Row,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> 5 alloc::vec::Vec 5 alloc::vec::Vec)>>>> 6 alloc::vec::Vec<(mz_compute::render::errors::DataflowErrorSer,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> +6 alloc::vec::Vec<(mz_repr::row::Row,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> 6 alloc::vec::Vec)>>>> -9 alloc::vec::Vec<(mz_repr::row::Row,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> query TTT rowsort SELECT mdod_from.name AS from_name, From 5bdb47a38cef023ab999227eaa7a2a2bcfdce157 Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Wed, 22 Jul 2026 18:18:33 +0200 Subject: [PATCH 3/4] style: split structuring semicolons in import leaf-encode comments Comment-only, no behavior change. Co-Authored-By: Claude Opus 4.8 --- src/compute/src/render.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/compute/src/render.rs b/src/compute/src/render.rs index 5c008c26253f5..c5db18966abc7 100644 --- a/src/compute/src/render.rs +++ b/src/compute/src/render.rs @@ -375,7 +375,7 @@ pub fn build_compute_dataflow( ); for (id, (oks, errs)) in imported_sources.into_iter() { - // Imports read row-shaped data from persist; encode it to the + // Imports read row-shaped data from persist. Encode it to the // columnar edge at the boundary. The batches are already // consolidated, so this leaf-encode is non-consolidating. let bundle = crate::render::CollectionBundle::from_edge( @@ -478,7 +478,7 @@ pub fn build_compute_dataflow( ); for (id, (oks, errs)) in imported_sources.into_iter() { - // Imports read row-shaped data from persist; encode it to the + // Imports read row-shaped data from persist. Encode it to the // columnar edge at the boundary. The batches are already // consolidated, so this leaf-encode is non-consolidating. let bundle = crate::render::CollectionBundle::from_edge( @@ -674,7 +674,7 @@ where start_signal, |e, _| e.clone(), ); - // The filtered index collection is row-shaped; encode it to + // The filtered index collection is row-shaped. Encode it to // the columnar edge at the boundary. It is already // consolidated, so this leaf-encode is non-consolidating. CollectionBundle::from_edge( From 83904f95eacc46065208eaa0744bc3dc0315a20e Mon Sep 17 00:00:00 2001 From: Moritz Hoffmann Date: Thu, 10 Sep 2026 20:20:41 +0200 Subject: [PATCH 4/4] doc: tighten the import leaf-encode comments State the property that makes the encode non-consolidating and drop the restatement of what the call does. Co-Authored-By: Claude Opus 5 (1M context) --- src/compute/src/render.rs | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/src/compute/src/render.rs b/src/compute/src/render.rs index c5db18966abc7..bba77989e45fd 100644 --- a/src/compute/src/render.rs +++ b/src/compute/src/render.rs @@ -375,9 +375,8 @@ pub fn build_compute_dataflow( ); for (id, (oks, errs)) in imported_sources.into_iter() { - // Imports read row-shaped data from persist. Encode it to the - // columnar edge at the boundary. The batches are already - // consolidated, so this leaf-encode is non-consolidating. + // Persist batches are row-shaped and already consolidated, so the + // encode here is non-consolidating. let bundle = crate::render::CollectionBundle::from_edge( CollectionEdge::Columnar(vec_to_columnar(oks.enter(region))), errs.enter(region), @@ -478,9 +477,8 @@ pub fn build_compute_dataflow( ); for (id, (oks, errs)) in imported_sources.into_iter() { - // Imports read row-shaped data from persist. Encode it to the - // columnar edge at the boundary. The batches are already - // consolidated, so this leaf-encode is non-consolidating. + // Persist batches are row-shaped and already consolidated, so the + // encode here is non-consolidating. let bundle = crate::render::CollectionBundle::from_edge( CollectionEdge::Columnar(vec_to_columnar(oks.enter_region(region))), errs.enter_region(region), @@ -674,9 +672,8 @@ where start_signal, |e, _| e.clone(), ); - // The filtered index collection is row-shaped. Encode it to - // the columnar edge at the boundary. It is already - // consolidated, so this leaf-encode is non-consolidating. + // The filtered index collection is row-shaped and already + // consolidated, so the encode here is non-consolidating. CollectionBundle::from_edge( CollectionEdge::Columnar(vec_to_columnar(oks)), errs,