diff --git a/src/compute/src/extensions/arrange.rs b/src/compute/src/extensions/arrange.rs index f63e9ebfe08c5..5aa383e0f11fa 100644 --- a/src/compute/src/extensions/arrange.rs +++ b/src/compute/src/extensions/arrange.rs @@ -8,7 +8,8 @@ // by the Apache License, Version 2.0. use std::collections::BTreeMap; -use std::rc::{Rc, Weak}; +use std::rc::Rc; +use std::sync::{Arc, Weak}; use differential_dataflow::difference::Semigroup; use differential_dataflow::lattice::Lattice; @@ -17,6 +18,7 @@ use differential_dataflow::operators::arrange::{Arranged, TraceAgent}; use differential_dataflow::trace::implementations::spine_fueled::Spine; use differential_dataflow::trace::{Batch, Batcher, Builder, Trace, TraceReader}; use differential_dataflow::{Collection, Data, ExchangeData, Hashable, VecCollection}; +use mz_row_spine::ArcBatch; use timely::Container; use timely::container::{ContainerBuilder, PushInto}; use timely::dataflow::Stream; @@ -240,10 +242,14 @@ pub trait ArrangementSize { /// * `arranged`: The arrangement to inspect. /// * `logic`: Closure that calculates the heap size/capacity/allocations for a batch. The return /// value are size and capacity in bytes, and number of allocations, all in absolute values. +/// +/// Batch-size logging identifies each batch by the address of its backing allocation and holds a +/// weak reference to it, so it needs the `Arc` underlying the spine's [`ArcBatch`] batches; +/// `batch.0` reaches straight through the newtype to it. fn log_arrangement_size_inner<'scope, B, L>( - arranged: Arranged<'scope, TraceAgent>>>, + arranged: Arranged<'scope, TraceAgent>>>, mut logic: L, -) -> Arranged<'scope, TraceAgent>>> +) -> Arranged<'scope, TraceAgent>>> where B: Batch + 'static, L: FnMut(&B) -> (usize, usize, usize) + 'static, @@ -282,14 +288,14 @@ where input.for_each(|time, data| { for batch in data.iter() { batches - .entry(Rc::as_ptr(batch)) - .or_insert_with(|| (Rc::downgrade(batch), logic(batch))); + .entry(Arc::as_ptr(&batch.0)) + .or_insert_with(|| (Arc::downgrade(&batch.0), logic(&batch.0))); } output.session(&time).give_container(data); }); let Some(trace) = trace.upgrade() else { // Invariant: `batches` holds no entries once the trace is gone. Each entry's - // `Weak` keeps its batch's `RcBox` allocation reserved, and the `retain` below + // `Weak` keeps its batch's `ArcInner` allocation reserved, and the `retain` below // that would drop it is unreachable on this path, so the entries have to go // here. The upgrade cannot start succeeding again, hence clearing on every // activation that takes this path also covers batches that arrive on the @@ -300,8 +306,8 @@ where trace.borrow().trace().map_batches(|batch| { batches - .entry(Rc::as_ptr(batch)) - .or_insert_with(|| (Rc::downgrade(batch), logic(batch))); + .entry(Arc::as_ptr(&batch.0)) + .or_insert_with(|| (Arc::downgrade(&batch.0), logic(&batch.0))); }); let (mut size, mut capacity, mut allocations) = (0, 0, 0); diff --git a/src/compute/src/typedefs.rs b/src/compute/src/typedefs.rs index 3b7f33af5ea68..8414445e1c945 100644 --- a/src/compute/src/typedefs.rs +++ b/src/compute/src/typedefs.rs @@ -29,32 +29,29 @@ pub use crate::typedefs::spines::{ColKeySpine, ColValSpine}; pub use mz_row_spine::{RowRowSpine, RowSpine, RowValBatcher, RowValSpine}; pub(crate) mod spines { - use std::rc::Rc; - use columnation::Columnation; use differential_dataflow::trace::implementations::ord_neu::{ OrdKeyBatch, OrdKeyBuilder, OrdValBatch, OrdValBuilder, }; use differential_dataflow::trace::implementations::spine_fueled::Spine; use differential_dataflow::trace::implementations::{Layout, Update}; - use differential_dataflow::trace::rc_blanket_impls::RcBuilder; use mz_timely_util::columnation::ColumnationStack; - use mz_row_spine::OffsetOptimized; + use mz_row_spine::{ArcBatch, ArcBuilder, OffsetOptimized}; use crate::typedefs::{KeyBatcher, KeyValBatcher}; /// A spine for generic keys and values. - pub type ColValSpine = Spine>>>; + pub type ColValSpine = Spine>>>; pub type ColValBatcher = KeyValBatcher; pub type ColValBuilder = - RcBuilder, ColumnationStack<((K, V), T, R)>>>; + ArcBuilder, ColumnationStack<((K, V), T, R)>>>; /// A spine for generic keys - pub type ColKeySpine = Spine>>>; + pub type ColKeySpine = Spine>>>; pub type ColKeyBatcher = KeyBatcher; pub type ColKeyBuilder = - RcBuilder, ColumnationStack<((K, ()), T, R)>>>; + ArcBuilder, ColumnationStack<((K, ()), T, R)>>>; /// A layout based on chunked timely stacks pub struct MzStack { diff --git a/src/row-spine/Cargo.toml b/src/row-spine/Cargo.toml index 998f8dbc92f61..9ce7b79fab11c 100644 --- a/src/row-spine/Cargo.toml +++ b/src/row-spine/Cargo.toml @@ -19,5 +19,8 @@ mz-repr = { path = "../repr" } mz-timely-util = { path = "../timely-util", default-features = false } timely.workspace = true +[dev-dependencies] +mz-ore = { path = "../ore", default-features = false, features = ["columnation"] } + [features] default = ["mz-ore/default", "mz-timely-util/default"] diff --git a/src/row-spine/src/arc_batch.rs b/src/row-spine/src/arc_batch.rs new file mode 100644 index 0000000000000..5d37b8dd836aa --- /dev/null +++ b/src/row-spine/src/arc_batch.rs @@ -0,0 +1,266 @@ +// 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. + +//! An `Arc`-backed batch newtype whose contents can be shared across timely runtimes. +//! +//! Differential's default spines reference-count their batches with `Rc`, which is worker-local. +//! Sharing an arrangement with another runtime (a reader on a different worker thread) needs the +//! batches behind an `Arc` so a batch whose contents are `Send + Sync` can be read from that other +//! thread. +//! +//! The blanket `impl Trait for Arc` that would express this lives outside this crate: both `Arc` +//! and differential's `Batch`/`Builder`/`Merger`/`Cursor` traits are foreign, so the orphan rule +//! forbids it here. [`ArcBatch`] is a local newtype around `Arc` that carries those impls +//! instead. The impls delegate straight through to the inner batch, so `ArcBatch` behaves +//! exactly like `B` except that its handle is atomically reference counted. +//! +//! This mirrors differential's own `rc_blanket_impls` (for `Rc`), swapping in `Arc`. Keeping it +//! here as a newtype lets cross-thread arrangement sharing build against a released +//! differential-dataflow, with no differential-side `Arc` batch impls required. + +use std::sync::Arc; + +use differential_dataflow::trace::{ + Batch, BatchReader, Builder, Cursor, Description, Merger, Navigable, +}; +use timely::progress::{Antichain, frontier::AntichainRef}; + +/// An `Arc`-backed batch, shareable across threads when `B`'s contents are `Send + Sync`. +/// +/// A transparent newtype around `Arc`. Cloning shares the underlying batch, exactly like the +/// `Rc`-backed default, but with atomic reference counting. +pub struct ArcBatch(pub Arc); + +// Hand-written rather than derived: `#[derive(Clone)]` would bound `B: Clone`, but `Arc` is +// `Clone` for any `B` (it clones the handle, not the batch). The derived bound would make +// `ArcBatch: Clone` fail for a non-`Clone` batch such as `OrdValBatch`, which in turn breaks +// `Spine>: TraceReader`. +impl Clone for ArcBatch { + fn clone(&self) -> Self { + ArcBatch(Arc::clone(&self.0)) + } +} + +impl ArcBatch { + /// Wraps a batch in an `Arc`. + pub fn new(batch: B) -> Self { + ArcBatch(Arc::new(batch)) + } +} + +impl std::ops::Deref for ArcBatch { + type Target = B; + fn deref(&self) -> &B { + &self.0 + } +} + +impl Navigable for ArcBatch { + type Cursor = ArcBatchCursor; + fn cursor(&self) -> Self::Cursor { + // Disambiguate to the inner batch's cursor, reached through the `Deref`, so the wrapper's + // `Cursor` is `B`'s rather than any impl that might exist on `Arc` itself. + ArcBatchCursor::new(::cursor(&self.0)) + } +} + +impl BatchReader for ArcBatch { + type Time = B::Time; + fn len(&self) -> usize { + self.0.len() + } + fn description(&self) -> &Description { + self.0.description() + } +} + +/// Cursor over an [`ArcBatch`], delegating to the inner batch's cursor. +pub struct ArcBatchCursor { + cursor: C, +} + +impl ArcBatchCursor { + fn new(cursor: C) -> Self { + ArcBatchCursor { cursor } + } +} + +impl Cursor for ArcBatchCursor { + type Storage = ArcBatch; + + type Key<'a> = C::Key<'a>; + type ValOwn = C::ValOwn; + type Val<'a> = C::Val<'a>; + type Time = C::Time; + type TimeGat<'a> = C::TimeGat<'a>; + type Diff = C::Diff; + type DiffGat<'a> = C::DiffGat<'a>; + type KeyContainer = C::KeyContainer; + type ValContainer = C::ValContainer; + type TimeContainer = C::TimeContainer; + type DiffContainer = C::DiffContainer; + + #[inline] + fn key_valid(&self, storage: &Self::Storage) -> bool { + self.cursor.key_valid(&storage.0) + } + #[inline] + fn val_valid(&self, storage: &Self::Storage) -> bool { + self.cursor.val_valid(&storage.0) + } + + #[inline] + fn key<'a>(&self, storage: &'a Self::Storage) -> Self::Key<'a> { + self.cursor.key(&storage.0) + } + #[inline] + fn val<'a>(&self, storage: &'a Self::Storage) -> Self::Val<'a> { + self.cursor.val(&storage.0) + } + + #[inline] + fn get_key<'a>(&self, storage: &'a Self::Storage) -> Option> { + self.cursor.get_key(&storage.0) + } + #[inline] + fn get_val<'a>(&self, storage: &'a Self::Storage) -> Option> { + self.cursor.get_val(&storage.0) + } + + #[inline] + fn map_times, Self::DiffGat<'_>)>( + &mut self, + storage: &Self::Storage, + logic: L, + ) { + self.cursor.map_times(&storage.0, logic) + } + + #[inline] + fn step_key(&mut self, storage: &Self::Storage) { + self.cursor.step_key(&storage.0) + } + #[inline] + fn seek_key(&mut self, storage: &Self::Storage, key: Self::Key<'_>) { + self.cursor.seek_key(&storage.0, key) + } + + #[inline] + fn step_val(&mut self, storage: &Self::Storage) { + self.cursor.step_val(&storage.0) + } + #[inline] + fn seek_val(&mut self, storage: &Self::Storage, val: Self::Val<'_>) { + self.cursor.seek_val(&storage.0, val) + } + + #[inline] + fn rewind_keys(&mut self, storage: &Self::Storage) { + self.cursor.rewind_keys(&storage.0) + } + #[inline] + fn rewind_vals(&mut self, storage: &Self::Storage) { + self.cursor.rewind_vals(&storage.0) + } +} + +impl Batch for ArcBatch { + type Merger = ArcMerger; + fn empty(lower: Antichain, upper: Antichain) -> Self { + ArcBatch::new(B::empty(lower, upper)) + } +} + +/// Builds [`ArcBatch`]es, delegating to the inner batch's builder. +pub struct ArcBuilder { + builder: B, +} + +impl Builder for ArcBuilder { + type Input = B::Input; + type Time = B::Time; + type Output = ArcBatch; + fn with_capacity(keys: usize, vals: usize, upds: usize) -> Self { + ArcBuilder { + builder: B::with_capacity(keys, vals, upds), + } + } + fn push(&mut self, input: &mut Self::Input) { + self.builder.push(input) + } + fn done(self, description: Description) -> ArcBatch { + ArcBatch::new(self.builder.done(description)) + } + fn seal(chain: &mut Vec, description: Description) -> Self::Output { + ArcBatch::new(B::seal(chain, description)) + } +} + +/// Merges [`ArcBatch`]es, delegating to the inner batch's merger. +pub struct ArcMerger { + merger: B::Merger, +} + +impl Merger> for ArcMerger { + fn new( + source1: &ArcBatch, + source2: &ArcBatch, + compaction_frontier: AntichainRef, + ) -> Self { + ArcMerger { + merger: B::begin_merge(&source1.0, &source2.0, compaction_frontier), + } + } + fn work(&mut self, source1: &ArcBatch, source2: &ArcBatch, fuel: &mut isize) { + self.merger.work(&source1.0, &source2.0, fuel) + } + fn done(self) -> ArcBatch { + ArcBatch::new(self.merger.done()) + } +} + +#[cfg(test)] +mod tests { + use differential_dataflow::trace::cursor::Cursor; + use differential_dataflow::trace::implementations::ord_neu::OrdValBatcher; + use differential_dataflow::trace::{Batcher, Builder, Navigable}; + use timely::container::PushInto; + use timely::progress::Antichain; + + use crate::ArcOrdValBuilder; + + /// An `ArcBatch`'s cursor can be constructed and read from a thread other than the one that + /// built it, proving the newtype's batches are usable across a thread boundary. This is the + /// property that lets [`crate::ArcOrdValSpine`] (and the `RowRow`/`Err` spines built on + /// [`ArcBatch`]) back a cross-runtime shared trace; the default `Rc`-backed spines are + /// worker-local by design and do not have it. + /// + /// Mirrors differential-dataflow's own `tests/trace.rs` cross-thread batch read, over the local + /// [`ArcBatch`] newtype. + #[mz_ore::test] + fn arc_batch_reads_from_other_thread() { + fn assert_send_sync(_: &T) {} + + let mut batcher = OrdValBatcher::::new(None, 0); + batcher.push_into(vec![((1, 2), 0, 1), ((2, 3), 1, 1)]); + let (mut chain, description) = batcher.seal(Antichain::from_elem(2)); + let batch = ArcOrdValBuilder::::seal(&mut chain, description); + + assert_send_sync(&batch); + + let read = std::thread::spawn(move || { + let mut cursor = batch.cursor(); + cursor.to_vec(&batch, |k| *k, |v| *v) + }) + .join() + .expect("reader thread panicked"); + + assert_eq!(read, vec![((1, 2), vec![(0, 1)]), ((2, 3), vec![(1, 1)])]); + } +} diff --git a/src/row-spine/src/lib.rs b/src/row-spine/src/lib.rs index 3c2a14d2e524d..f7495ddc8dbfb 100644 --- a/src/row-spine/src/lib.rs +++ b/src/row-spine/src/lib.rs @@ -13,14 +13,18 @@ //! allocations, as well as a `dictionary` encoding wrapper that is able to rewrite //! the byte slices to use spare tags in each column to reference common values. +pub use self::arc_batch::{ArcBatch, ArcBuilder}; pub use self::dictionary::DatumContainer; pub use self::dictionary::DatumSeq; pub use self::offset_opt::OffsetOptimized; pub use self::spines::{ - RowBatcher, RowBuilder, RowRowBatcher, RowRowBuilder, RowRowColPagedBuilder, RowRowSpine, - RowSpine, RowValBatcher, RowValBuilder, RowValSpine, ValRowBatcher, ValRowBuilder, - ValRowColPagedBuilder, ValRowSpine, + ArcOrdKeyBuilder, ArcOrdKeySpine, ArcOrdValBuilder, ArcOrdValSpine, RowBatcher, RowBuilder, + RowRowBatcher, RowRowBuilder, RowRowColPagedBuilder, RowRowSpine, RowSpine, RowValBatcher, + RowValBuilder, RowValSpine, ValRowBatcher, ValRowBuilder, ValRowColPagedBuilder, ValRowSpine, }; + +mod arc_batch; + use differential_dataflow::trace::implementations::OffsetList; /// Enable per-column dictionary compression in row containers. @@ -29,18 +33,19 @@ pub static DICTIONARY_COMPRESSION: std::sync::atomic::AtomicBool = /// Spines specialized to contain `Row` types in keys and values. mod spines { - use std::rc::Rc; - use columnation::Columnation; use differential_dataflow::trace::implementations::Layout; use differential_dataflow::trace::implementations::Update; + use differential_dataflow::trace::implementations::Vector; use differential_dataflow::trace::implementations::merge_batcher::MergeBatcher; - use differential_dataflow::trace::implementations::ord_neu::{OrdKeyBatch, OrdValBatch}; + use differential_dataflow::trace::implementations::ord_neu::{ + OrdKeyBatch, OrdKeyBuilder, OrdValBatch, OrdValBuilder, + }; use differential_dataflow::trace::implementations::spine_fueled::Spine; - use differential_dataflow::trace::rc_blanket_impls::RcBuilder; use mz_repr::Row; use mz_timely_util::columnation::{ColInternalMerger, ColumnationStack}; + use crate::arc_batch::{ArcBatch, ArcBuilder}; use crate::{DatumContainer, OffsetOptimized}; /// Batcher matching `mz_compute::typedefs::KeyValBatcher`, redeclared @@ -48,9 +53,9 @@ mod spines { type KeyValBatcher = MergeBatcher>; type KeyBatcher = KeyValBatcher; - pub type RowRowSpine = Spine>>>; + pub type RowRowSpine = Spine>>>; pub type RowRowBatcher = KeyValBatcher; - pub type RowRowBuilder = RcBuilder>; + pub type RowRowBuilder = ArcBuilder>; /// `RowRowBuilder` variant that consumes [`Column`] chunks. Pairs with /// [`Col2ValPagedBatcher`] for the spillable arrange path. Installs a @@ -61,21 +66,21 @@ mod spines { /// [`Col2ValPagedBatcher`]: mz_timely_util::columnar::Col2ValPagedBatcher /// [`Column`]: mz_timely_util::columnar::Column pub type RowRowColPagedBuilder = - RcBuilder>; + ArcBuilder>; - pub type RowValSpine = Spine>>>; + pub type RowValSpine = Spine>>>; pub type RowValBatcher = KeyValBatcher; pub type RowValBuilder = - RcBuilder>; + ArcBuilder>; - pub type RowSpine = Spine>>>; + pub type RowSpine = Spine>>>; pub type RowBatcher = KeyBatcher; - pub type RowBuilder = RcBuilder>; + pub type RowBuilder = ArcBuilder>; - pub type ValRowSpine = Spine>>>; + pub type ValRowSpine = Spine>>>; pub type ValRowBatcher = KeyValBatcher; pub type ValRowBuilder = - RcBuilder>; + ArcBuilder>; /// `ValRowBuilder` variant that consumes [`Column`] chunks. Pairs with /// `Col2ValPagedBatcher` for the spillable arrange path where @@ -86,7 +91,20 @@ mod spines { /// /// [`Column`]: mz_timely_util::columnar::Column pub type ValRowColPagedBuilder = - RcBuilder>; + ArcBuilder>; + + /// A generic `Arc`-backed key/value spine, for callers outside `mz_compute` that need an + /// arrangement over non-`Row`-specialized types. The `Arc` handle rides on the local + /// [`ArcBatch`] newtype, so no differential-side `Arc` batch impls are required. + pub type ArcOrdValSpine = Spine>>>; + /// Generic `Arc`-backed key-only spine. See [`ArcOrdValSpine`]. + pub type ArcOrdKeySpine = Spine>>>; + /// Builder pairing with [`ArcOrdValSpine`]. + pub type ArcOrdValBuilder = + ArcBuilder, Vec<((K, V), T, R)>>>; + /// Builder pairing with [`ArcOrdKeySpine`]. + pub type ArcOrdKeyBuilder = + ArcBuilder, Vec<((K, ()), T, R)>>>; /// A layout based on timely stacks pub struct RowRowLayout> { @@ -155,10 +173,28 @@ mod spines { #[cfg(test)] mod tests { use crate::DatumContainer; + use crate::spines::{RowLayout, RowRowLayout, RowValLayout}; use differential_dataflow::trace::implementations::BatchContainer; + use differential_dataflow::trace::implementations::ord_neu::{OrdKeyBatch, OrdValBatch}; use mz_repr::adt::date::Date; use mz_repr::adt::interval::Interval; - use mz_repr::{Datum, Row, SqlScalarType}; + use mz_repr::{Datum, Diff, Row, SqlScalarType, Timestamp}; + use mz_timely_util::columnation::ColumnationStack; + + fn assert_send_sync() {} + + /// The batch types backing our spines must stay `Send + Sync`, so that batches + /// can be shared across threads (for example behind an `Arc`) to serve reads + /// from outside the worker that maintains the trace. This holds because the + /// backing containers bottom out in `Vec`s, lgalloc regions, and `CompactBytes`, + /// all of which are thread-safe. + #[mz_ore::test] + fn batches_are_send_sync() { + assert_send_sync::>>(); + assert_send_sync::>>(); + assert_send_sync::>>(); + assert_send_sync::>(); + } #[mz_ore::test] #[cfg_attr(miri, ignore)] // unsupported operation: integer-to-pointer casts and `ptr::with_exposed_provenance` are not supported diff --git a/src/storage/src/render/sinks.rs b/src/storage/src/render/sinks.rs index e27212700e331..c0d44acf083ec 100644 --- a/src/storage/src/render/sinks.rs +++ b/src/storage/src/render/sinks.rs @@ -14,12 +14,11 @@ use std::time::{Duration, Instant}; use differential_dataflow::operators::arrange::{Arrange, Arranged, TraceAgent}; use differential_dataflow::trace::TraceReader; -use differential_dataflow::trace::implementations::ord_neu::{ - OrdValBatcher, OrdValSpine, RcOrdValBuilder, -}; +use differential_dataflow::trace::implementations::ord_neu::OrdValBatcher; use differential_dataflow::{AsCollection, Hashable, VecCollection}; use mz_persist_client::operators::shard_source::SnapshotMode; use mz_repr::{Datum, Diff, GlobalId, Row, Timestamp}; +use mz_row_spine::{ArcOrdValBuilder, ArcOrdValSpine}; use mz_storage_operators::persist_source; use mz_storage_types::controller::CollectionMetadata; use mz_storage_types::errors::DataflowError; @@ -35,7 +34,7 @@ use crate::storage_state::StorageState; /// The concrete trace type produced internally when arranging a sink's input. /// The sink never sees this directly — only the batches flowing through it — /// but it's the anchor for the batch type in [`SinkBatchStream`]. -pub(crate) type SinkTrace = TraceAgent, Row, Timestamp, Diff>>; +pub(crate) type SinkTrace = TraceAgent, Row, Timestamp, Diff>>; /// Stream of arrangement batches handed to [`SinkRender::render_sink`]. /// @@ -148,7 +147,7 @@ fn arrange_sink_input<'scope>( // Allow access to `arrange_named` because we cannot access Mz's wrapper // from here. TODO(database-issues#5046): Revisit with cluster unification. #[allow(clippy::disallowed_methods)] - let Arranged {stream, trace: _} = keyed.arrange_named::, RcOrdValBuilder<_, _, _, _>, OrdValSpine<_, _, _, _>>("Arrange Sink"); + let Arranged {stream, trace: _} = keyed.arrange_named::, ArcOrdValBuilder<_, _, _, _>, ArcOrdValSpine<_, _, _, _>>("Arrange Sink"); stream } diff --git a/src/storage/src/sink/iceberg.rs b/src/storage/src/sink/iceberg.rs index aed8ad817272f..315ba1428e822 100644 --- a/src/storage/src/sink/iceberg.rs +++ b/src/storage/src/sink/iceberg.rs @@ -134,6 +134,7 @@ use mz_persist_client::Diagnostics; use mz_persist_client::write::WriteHandle; use mz_persist_types::codec_impls::UnitSchema; use mz_repr::{Diff, GlobalId, Row, Timestamp}; +use mz_row_spine::ArcBatch; use mz_storage_types::StorageDiff; use mz_storage_types::configuration::StorageConfiguration; use mz_storage_types::controller::CollectionMetadata; @@ -1588,7 +1589,7 @@ fn write_data_files<'scope, H: EnvelopeHandler + 'static>( // Rows can arrive before their batch description due to dataflow parallelism. // Stash them until we know which batch they belong to. // Keyed by the lower bound (per arrangement batch) of the rows. - let mut stashed_rows: VecDeque>> = VecDeque::new(); + let mut stashed_rows: VecDeque>> = VecDeque::new(); // Track batches currently being written. When a row arrives, we check if it belongs // to an in-flight batch. When frontiers advance to a batch's upper, we close the @@ -1682,7 +1683,7 @@ fn write_data_files<'scope, H: EnvelopeHandler + 'static>( last_input_bounds = Some((rows.lower().clone(), rows.upper().clone())); - stashed_rows.push_back(Rc::clone(rows)); + stashed_rows.push_back(rows.clone()); } } Event::Progress(frontier) => { @@ -1829,7 +1830,7 @@ type BatchDescription = (Antichain, Antichain); /// are in order and non-overlapping. async fn with_ready_batches( input_frontier: Antichain, - input_batches: &mut VecDeque>>, + input_batches: &mut VecDeque>>, output_frontier: Antichain, output_batches: &mut VecDeque<(BatchDescription, W)>, mut write_rows: Write, @@ -2148,9 +2149,9 @@ mod tests { /// An input batch with the given bounds. The pairing logic under test /// only looks at bounds, so the batch holds no data. - fn input(lower: u64, upper: Option) -> Rc { + fn input(lower: u64, upper: Option) -> ArcBatch { let (lower, upper) = span(lower, upper); - Rc::new(TestBatch::empty(lower, upper)) + ArcBatch(Arc::new(TestBatch::empty(lower, upper))) } #[derive(Debug, PartialEq)] @@ -2164,7 +2165,7 @@ mod tests { /// sequence of calls it made. async fn run( input_frontier: Antichain, - input_batches: &mut VecDeque>, + input_batches: &mut VecDeque>, output_frontier: Antichain, output_batches: &mut VecDeque<(BatchDescription, ())>, ) -> Vec { diff --git a/test/sqllogictest/introspection/relations.slt b/test/sqllogictest/introspection/relations.slt index a1f31e8090c53..b8acec4390bb0 100644 --- a/test/sqllogictest/introspection/relations.slt +++ b/test/sqllogictest/introspection/relations.slt @@ -49,11 +49,11 @@ FROM ON mdco.to_operator_id = mdod_to.id WHERE mdod_to.dataflow_name LIKE '%test_primary_idx' ---- -ArrangeBy[[Column(0,␠"a"),␠Column(1,␠"b")]] ArrangementSize alloc::vec::Vec)>>>> -ArrangeBy[[Column(0,␠"a"),␠Column(1,␠"b")]]-errors ArrangementSize alloc::vec::Vec)>>>> -ArrangementSize LogOperatorHydration␠(2) alloc::vec::Vec)>>>> -BuildingObject(User(2)) expire_stream_at(materialize.public.test_primary_idx_export_index_errs) alloc::vec::Vec)>>>> -BuildingObject(User(2)) expire_stream_at(materialize.public.test_primary_idx_export_index_oks) alloc::vec::Vec)>>>> +ArrangeBy[[Column(0,␠"a"),␠Column(1,␠"b")]] ArrangementSize alloc::vec::Vec)>>>> +ArrangeBy[[Column(0,␠"a"),␠Column(1,␠"b")]]-errors ArrangementSize alloc::vec::Vec)>>>> +ArrangementSize LogOperatorHydration␠(2) alloc::vec::Vec)>>>> +BuildingObject(User(2)) expire_stream_at(materialize.public.test_primary_idx_export_index_errs) alloc::vec::Vec)>>>> +BuildingObject(User(2)) expire_stream_at(materialize.public.test_primary_idx_export_index_oks) alloc::vec::Vec)>>>> Concatenate FlatMap alloc::vec::Vec<(mz_compute::render::errors::DataflowErrorSer,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> Exchange FlatMap alloc::vec::Vec<(u64,␠mz_txn_wal::txn_read::DataRemapEntry)> Feedback persist_source_backpressure(backpressure(u1)) alloc::vec::Vec @@ -69,8 +69,8 @@ LogOperatorHydration␠(1) FormArrangementKey alloc::vec::Vec<(mz_repr::row::R 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)> 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)>>>> +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)>>>> granular_backpressure(u1) shard_source_descs_return(u1) alloc::vec::Vec granular_backpressure(u1) txns_progress_frontiers(u1) alloc::vec::Vec<(core::result::Result,␠(mz_repr::timestamp::Timestamp,␠mz_storage_operators::persist_source::Subtime),␠mz_ore::overflowing::Overflowing)> persist_source::decode_and_mfp(u1) InspectBatch alloc::vec::Vec<(core::result::Result,␠(mz_repr::timestamp::Timestamp,␠mz_storage_operators::persist_source::Subtime),␠mz_ore::overflowing::Overflowing)> @@ -97,10 +97,10 @@ 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)> -5 alloc::vec::Vec)>>>> 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)>>>> +6 alloc::vec::Vec)>>>> 9 alloc::vec::Vec<(mz_repr::row::Row,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> query TTT rowsort @@ -115,38 +115,38 @@ FROM ON mdco.to_operator_id = mdod_to.id WHERE mdod_to.dataflow_name = 'Dataflow: logging' ---- -Arrange␠Compute(ArrangementHeapAllocations) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Compute(ArrangementHeapCapacity) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Compute(ArrangementHeapSize) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Compute(DataflowCurrent) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Compute(DataflowGlobal) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Compute(ErrorCount) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Compute(FrontierCurrent) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Compute(HydrationTime) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Compute(ImportFrontierCurrent) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Compute(LirMapping) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Compute(OperatorHydrationStatus) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Compute(PeekCurrent) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Compute(PeekDuration) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Differential(ArrangementBatches) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Differential(ArrangementRecords) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Differential(BatcherAllocations) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Differential(BatcherCapacity) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Differential(BatcherRecords) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Differential(BatcherSize) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Differential(Sharing) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠PrometheusMetrics ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Timely(Addresses) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Timely(BatchesReceived) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Timely(BatchesSent) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Timely(Channels) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Timely(Elapsed) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Timely(Histogram) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Timely(MessagesReceived) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Timely(MessagesSent) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Timely(Operates) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Timely(Parks) ArrangementSize alloc::vec::Vec)>>>> -Arrange␠Timely(Reachability) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(ArrangementHeapAllocations) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(ArrangementHeapCapacity) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(ArrangementHeapSize) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(DataflowCurrent) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(DataflowGlobal) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(ErrorCount) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(FrontierCurrent) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(HydrationTime) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(ImportFrontierCurrent) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(LirMapping) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(OperatorHydrationStatus) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(PeekCurrent) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Compute(PeekDuration) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Differential(ArrangementBatches) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Differential(ArrangementRecords) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Differential(BatcherAllocations) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Differential(BatcherCapacity) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Differential(BatcherRecords) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Differential(BatcherSize) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Differential(Sharing) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠PrometheusMetrics ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Timely(Addresses) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Timely(BatchesReceived) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Timely(BatchesSent) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Timely(Channels) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Timely(Elapsed) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Timely(Histogram) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Timely(MessagesReceived) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Timely(MessagesSent) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Timely(Operates) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Timely(Parks) ArrangementSize alloc::vec::Vec)>>>> +Arrange␠Timely(Reachability) ArrangementSize alloc::vec::Vec)>>>> Compute␠Logging␠Demux Arrange␠Compute(ArrangementHeapAllocations) mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> Compute␠Logging␠Demux Arrange␠Compute(ArrangementHeapCapacity) mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> Compute␠Logging␠Demux Arrange␠Compute(ArrangementHeapSize) mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> @@ -245,7 +245,7 @@ GROUP BY type; 1 mz_timely_util::columnar::Column<(core::time::Duration,␠(usize,␠alloc::vec::Vec<(usize,␠usize,␠bool,␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)>))> 1 mz_timely_util::columnar::Column<(core::time::Duration,␠mz_compute::logging::compute::ComputeEvent)> 3 alloc::vec::Vec<(core::time::Duration,␠timely::logging::TimelyEvent)> -32 alloc::vec::Vec)>>>> +32 alloc::vec::Vec)>>>> 32 mz_timely_util::columnar::Column<((mz_repr::row::Row,␠mz_repr::row::Row),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> 4 alloc::vec::Vec<((mz_compute::logging::timely::MessageDatum,␠()),␠mz_repr::timestamp::Timestamp,␠mz_ore::overflowing::Overflowing)> 4 alloc::vec::Vec)>>>