diff --git a/LICENSE b/LICENSE index 0cc0ff1ff8409..f0e5e32d33ff2 100644 --- a/LICENSE +++ b/LICENSE @@ -13,7 +13,7 @@ Business Source License 1.1 Licensor: Materialize, Inc. -Licensed Work: Materialize Version 20260911 +Licensed Work: Materialize Version 20260912 The Licensed Work is © 2026 Materialize, Inc. Additional Use Grant: Within a single installation of the Licensed Work, @@ -39,7 +39,7 @@ Additional Use Grant: Within a single installation of the Licensed Work, whose definitions are controlled by such third parties. -Change Date: September 11, 2030 +Change Date: September 12, 2030 Change License: Apache License, Version 2.0 diff --git a/ci/mkpipeline.py b/ci/mkpipeline.py index 909a2d7de7ebd..ddaf8d78370f5 100644 --- a/ci/mkpipeline.py +++ b/ci/mkpipeline.py @@ -759,33 +759,47 @@ def permit_rerunning_successful_steps(pipeline: Any) -> None: ) +# Failures that are the infrastructure's fault rather than the step's, so every +# step retries them regardless of what else it asks for. +SHARED_AUTOMATIC_RETRIES = [ + { + "exit_status": -1, + "signal_reason": "none", + "limit": 2, + }, + { + "signal_reason": "agent_stop", # Stopped by OS + "limit": 2, + }, + { + "exit_status": 128, # Temporary Github/GHCR/DockerHub connection issue + "limit": 2, + }, + { + "exit_status": 199, # Rust ICE https://github.com/rust-lang/rust/issues/148581 + "limit": 2, + }, +] + + def set_retry_on_agent_lost(pipeline: Any) -> None: for step in steps(pipeline): if "trigger" in step or "wait" in step or "group" in step or "block" in step: continue - retry = step.setdefault("retry", {}) - if "automatic" in retry: + automatic = step.setdefault("retry", {}).setdefault("automatic", []) + # A step retrying on every exit status already covers these. + if any(rule.get("exit_status") == "*" for rule in automatic): continue - retry.setdefault("automatic", []).extend( - [ - { - "exit_status": -1, - "signal_reason": "none", - "limit": 2, - }, - { - "signal_reason": "agent_stop", # Stopped by OS - "limit": 2, - }, - { - "exit_status": 128, # Temporary Github/GHCR/DockerHub connection issue - "limit": 2, - }, - { - "exit_status": 199, # Rust ICE https://github.com/rust-lang/rust/issues/148581 - "limit": 2, - }, - ] + # Merge rather than skip a step that declares its own rules: skipping + # lets a narrow override such as `exit_status: 1` drop every shared rule + # and hard-fail the step on the infrastructure classes they exist for. + # Nothing surfaces that, and the mzcompose plugin goes out of its way to + # propagate 128 and 199 so these rules can match. + declared = {(r.get("exit_status"), r.get("signal_reason")) for r in automatic} + automatic.extend( + dict(rule) + for rule in SHARED_AUTOMATIC_RETRIES + if (rule.get("exit_status"), rule.get("signal_reason")) not in declared ) diff --git a/doc/user/content/sql/system-catalog/mz_internal.md b/doc/user/content/sql/system-catalog/mz_internal.md index d36965fa807ed..0778a125b2121 100644 --- a/doc/user/content/sql/system-catalog/mz_internal.md +++ b/doc/user/content/sql/system-catalog/mz_internal.md @@ -227,6 +227,7 @@ usage. | `disk_bytes` | [`uint8`] | Approximate disk usage, in bytes. | `heap_bytes` | [`uint8`] | Approximate heap (RAM + swap) usage, in bytes. | `heap_limit` | [`uint8`] | Available heap (RAM + swap) space, in bytes. +| `swap_bytes` | [`uint8`] | Approximate swap usage, in bytes. ## `mz_cluster_replica_metrics_history` @@ -252,6 +253,7 @@ history is retained across restarts. | `occurred_at` | [`timestamp with time zone`] | Wall-clock timestamp at which the event occurred. | `heap_bytes` | [`uint8`] | Approximate heap (RAM + swap) usage, in bytes. | `heap_limit` | [`uint8`] | Available heap (RAM + swap) space, in bytes. +| `swap_bytes` | [`uint8`] | Approximate swap usage, in bytes. ## `mz_cluster_replica_statuses` @@ -298,6 +300,7 @@ At this time, we do not make any guarantees about the exactness or freshness of | `memory_percent` | [`double precision`] | Approximate RAM usage, in percent of the total allocation. | `disk_percent` | [`double precision`] | Approximate disk usage, in percent of the total allocation. | `heap_percent` | [`double precision`] | Approximate heap (RAM + swap) usage, in percent of the total allocation. +| `swap_percent` | [`double precision`] | Approximate swap usage, in percent of the total heap allocation. ## `mz_cluster_replica_utilization_history` @@ -316,6 +319,7 @@ At this time, we do not make any guarantees about the exactness or freshness of | `memory_percent` | [`double precision`] | Approximate RAM usage, in percent of the total allocation. | `disk_percent` | [`double precision`] | Approximate disk usage, in percent of the total allocation. | `heap_percent` | [`double precision`] | Approximate heap (RAM + swap) usage, in percent of the total allocation. +| `swap_percent` | [`double precision`] | Approximate swap usage, in percent of the total heap allocation. | `occurred_at` | [`timestamp with time zone`] | Wall-clock timestamp at which the event occurred. ## `mz_cluster_replica_history` diff --git a/src/adapter/src/catalog/open/builtin_schema_migration.rs b/src/adapter/src/catalog/open/builtin_schema_migration.rs index 34653cd13aad0..e21d3da4179bb 100644 --- a/src/adapter/src/catalog/open/builtin_schema_migration.rs +++ b/src/adapter/src/catalog/open/builtin_schema_migration.rs @@ -462,6 +462,16 @@ static MIGRATIONS: LazyLock> = LazyLock::new(|| { MZ_INTERNAL_SCHEMA, "mz_object_dependencies", ), + // Adding the `swap_bytes` column to `mz_cluster_replica_metrics_history` + // is a backward-compatible append, so the shard schema can evolve in + // place. See the NOTE above: this version must stay at the workspace's + // current dev version until the change ships. + MigrationStep::evolution( + "26.43.0-dev.0", + CatalogItemType::Source, + MZ_INTERNAL_SCHEMA, + "mz_cluster_replica_metrics_history", + ), ] }); diff --git a/src/catalog/src/builtin/mz_internal.rs b/src/catalog/src/builtin/mz_internal.rs index 006492fbdf5d2..ccde5b89bd957 100644 --- a/src/catalog/src/builtin/mz_internal.rs +++ b/src/catalog/src/builtin/mz_internal.rs @@ -2734,6 +2734,7 @@ pub static MZ_CLUSTER_REPLICA_METRICS_HISTORY: LazyLock = "Approximate heap (RAM + swap) usage, in bytes.", ), ("heap_limit", "Available heap (RAM + swap) space, in bytes."), + ("swap_bytes", "Approximate swap usage, in bytes."), ]), is_retained_metrics_object: false, access: vec![PUBLIC_SELECT], @@ -2752,6 +2753,7 @@ pub static MZ_CLUSTER_REPLICA_METRICS: LazyLock = LazyLock::new(|| .with_column("disk_bytes", SqlScalarType::UInt64.nullable(true)) .with_column("heap_bytes", SqlScalarType::UInt64.nullable(true)) .with_column("heap_limit", SqlScalarType::UInt64.nullable(true)) + .with_column("swap_bytes", SqlScalarType::UInt64.nullable(true)) .with_key(vec![0, 1]) .finish(), column_comments: BTreeMap::from_iter([ @@ -2768,6 +2770,7 @@ pub static MZ_CLUSTER_REPLICA_METRICS: LazyLock = LazyLock::new(|| "Approximate heap (RAM + swap) usage, in bytes.", ), ("heap_limit", "Available heap (RAM + swap) space, in bytes."), + ("swap_bytes", "Approximate swap usage, in bytes."), ]), sql: " SELECT @@ -2778,7 +2781,8 @@ SELECT memory_bytes, disk_bytes, heap_bytes, - heap_limit + heap_limit, + swap_bytes FROM mz_internal.mz_cluster_replica_metrics_history JOIN mz_cluster_replicas r ON r.id = replica_id ORDER BY replica_id, process_id, occurred_at DESC", @@ -2805,6 +2809,7 @@ ORDER BY replica_id, process_id, occurred_at DESC", ("disk_bytes", SemanticType::ByteCount), ("heap_bytes", SemanticType::ByteCount), ("heap_limit", SemanticType::ByteCount), + ("swap_bytes", SemanticType::ByteCount), ] }, }), @@ -5421,6 +5426,7 @@ pub static MZ_CLUSTER_REPLICA_UTILIZATION: LazyLock = LazyLock::new .with_column("memory_percent", SqlScalarType::Float64.nullable(true)) .with_column("disk_percent", SqlScalarType::Float64.nullable(true)) .with_column("heap_percent", SqlScalarType::Float64.nullable(true)) + .with_column("swap_percent", SqlScalarType::Float64.nullable(true)) .finish(), column_comments: BTreeMap::from_iter([ ("replica_id", "The ID of a cluster replica."), @@ -5441,6 +5447,10 @@ pub static MZ_CLUSTER_REPLICA_UTILIZATION: LazyLock = LazyLock::new "heap_percent", "Approximate heap (RAM + swap) usage, in percent of the total allocation.", ), + ( + "swap_percent", + "Approximate swap usage, in percent of the total heap allocation.", + ), ]), sql: " SELECT @@ -5449,7 +5459,8 @@ SELECT m.cpu_nano_cores::float8 / NULLIF(s.cpu_nano_cores, 0) * 100 AS cpu_percent, m.memory_bytes::float8 / NULLIF(s.memory_bytes, 0) * 100 AS memory_percent, m.disk_bytes::float8 / NULLIF(s.disk_bytes, 0) * 100 AS disk_percent, - m.heap_bytes::float8 / NULLIF(m.heap_limit, 0) * 100 AS heap_percent + m.heap_bytes::float8 / NULLIF(m.heap_limit, 0) * 100 AS heap_percent, + m.swap_bytes::float8 / NULLIF(m.heap_limit, 0) * 100 AS swap_percent FROM mz_catalog.mz_cluster_replicas AS r JOIN mz_catalog.mz_cluster_replica_sizes AS s ON r.size = s.size @@ -5486,6 +5497,7 @@ pub static MZ_CLUSTER_REPLICA_UTILIZATION_HISTORY: LazyLock = .with_column("memory_percent", SqlScalarType::Float64.nullable(true)) .with_column("disk_percent", SqlScalarType::Float64.nullable(true)) .with_column("heap_percent", SqlScalarType::Float64.nullable(true)) + .with_column("swap_percent", SqlScalarType::Float64.nullable(true)) .with_column( "occurred_at", SqlScalarType::TimestampTz { precision: None }.nullable(false), @@ -5510,6 +5522,10 @@ pub static MZ_CLUSTER_REPLICA_UTILIZATION_HISTORY: LazyLock = "heap_percent", "Approximate heap (RAM + swap) usage, in percent of the total allocation.", ), + ( + "swap_percent", + "Approximate swap usage, in percent of the total heap allocation.", + ), ( "occurred_at", "Wall-clock timestamp at which the event occurred.", @@ -5523,6 +5539,7 @@ SELECT m.memory_bytes::float8 / NULLIF(s.memory_bytes, 0) * 100 AS memory_percent, m.disk_bytes::float8 / NULLIF(s.disk_bytes, 0) * 100 AS disk_percent, m.heap_bytes::float8 / NULLIF(m.heap_limit, 0) * 100 AS heap_percent, + m.swap_bytes::float8 / NULLIF(m.heap_limit, 0) * 100 AS swap_percent, m.occurred_at FROM mz_catalog.mz_cluster_replicas AS r diff --git a/src/compute/src/extensions/temporal_bucket.rs b/src/compute/src/extensions/temporal_bucket.rs index 13a0f45009987..6f24cf081683d 100644 --- a/src/compute/src/extensions/temporal_bucket.rs +++ b/src/compute/src/extensions/temporal_bucket.rs @@ -9,73 +9,81 @@ //! Utilities and stream extensions for temporal bucketing. -use columnar::{Columnar, Index, Len}; +use std::hash::Hash; + +use columnar::{Columnar, Index, Len, Push}; use differential_dataflow::Hashable; use differential_dataflow::difference::Semigroup; use differential_dataflow::lattice::Lattice; use differential_dataflow::trace::Batcher; use mz_timely_util::columnar::Column; use mz_timely_util::columnar::batcher::ColumnChunker; +use mz_timely_util::columnar::builder::ColumnBuilder; +use mz_timely_util::columnar::columnar_exchange_data; use mz_timely_util::columnar::merge_batcher::ColumnMergeBatcher; -use mz_timely_util::temporal::{Bucket, BucketChain, BucketTimestamp}; +use mz_timely_util::temporal::{Bucket, BucketChain, BucketRange, BucketTimestamp}; use timely::Accountable; -use timely::container::PushInto; -use timely::dataflow::channels::pact::Exchange; +use timely::container::{CapacityContainerBuilder, PushInto}; +use timely::dataflow::channels::pact::{Exchange, ExchangeCore}; use timely::dataflow::operators::Operator; use timely::dataflow::{Stream, StreamVec}; use timely::order::TotalOrder; use timely::progress::{Antichain, PathSummary, Timestamp}; -use timely::{ContainerBuilder, ExchangeData, PartialOrder}; +use timely::{ExchangeData, PartialOrder}; use crate::typedefs::MzData; /// Sort outstanding updates into a [`BucketChain`], and reveal data not in advance of the input /// frontier. Retains a capability at the last input frontier to retain the right to produce data /// at times between the last input frontier and the current input frontier. -pub trait TemporalBucketing<'scope, T: Timestamp, O> { +pub trait TemporalBucketing<'scope, T: Timestamp>: Sized { /// Construct a new stream that stores updates into a [`BucketChain`] and reveals data /// not in advance of the frontier. Data that is within `threshold` distance of the input /// frontier or the `as_of` is passed through without being stored in the chain. - fn bucket( - self, - as_of: Antichain, - threshold: T::Summary, - ) -> Stream<'scope, T, CB::Container> - where - CB: ContainerBuilder + PushInto; + /// + /// The output container matches the input's, so a caller keeps whichever + /// representation it had. + fn bucket(self, as_of: Antichain, threshold: T::Summary) -> Self; } /// Implementation for streams in scopes where timestamps define a total order. -impl<'scope, T, D> TemporalBucketing<'scope, T, (D, T, mz_repr::Diff)> - for StreamVec<'scope, T, (D, T, mz_repr::Diff)> +impl<'scope, T, D> TemporalBucketing<'scope, T> for Stream<'scope, T, Column<(D, T, mz_repr::Diff)>> where T: Timestamp + Default + ExchangeData + MzData + BucketTimestamp + TotalOrder + Lattice, + for<'a> columnar::Ref<'a, T>: Copy + Ord, D: ExchangeData + MzData + Ord + Clone + std::fmt::Debug + Hashable, + for<'a> columnar::Ref<'a, D>: Copy + Ord + Hash, + for<'a> columnar::Ref<'a, mz_repr::Diff>: Ord, + for<'a> <(D, T, mz_repr::Diff) as Columnar>::Container: + Push>, { - fn bucket( - self, - as_of: Antichain, - threshold: T::Summary, - ) -> Stream<'scope, T, CB::Container> - where - CB: ContainerBuilder + PushInto<(D, T, mz_repr::Diff)>, - { + fn bucket(self, as_of: Antichain, threshold: T::Summary) -> Self { let scope = self.scope(); let logger = scope .worker() .logger_for("differential/arrange") .map(Into::into); - let pact = Exchange::new(|(d, _, _): &(D, T, mz_repr::Diff)| d.hashed().into()); - self.unary_frontier::(pact, "Temporal delay", |cap, info| { + type CB = CapacityContainerBuilder>; + + let pact = ExchangeCore::, _>::new_core( + columnar_exchange_data::, + ); + self.unary_frontier::, _, _, _>(pact, "Temporal delay", |cap, info| { let mut chain = BucketChain::new(MergeBatcherWrapper::new(logger, info.global_id)); let activator = scope.activator_for(info.address); // Cap tracking the lower bound of potentially outstanding data. let mut cap = Some(cap); - // Buffer for data to be inserted into the chain. - let mut buffer = Vec::new(); + // Holds one bucket's worth of updates on the way into the chain. + // Reused across activations for its allocation. + let mut buffer: Column<(D, T, mz_repr::Diff)> = Default::default(); + // Reused input permutation, ordered by time. + let mut permutation: Vec = Vec::new(); + // Reused so reading a record's time does not allocate: an iterative `T` + // owns a `PointStamp`'s allocation. + let mut time_buf = T::minimum(); move |(input, frontier), output| { // The upper frontier is the join of the input frontier and the `as_of` frontier, @@ -93,41 +101,55 @@ where input.for_each_time(|time, data| { let mut session = output.session_with_builder(&time); for data in data { - // Skip data that is about to be revealed. - let pass_through = data.extract_if(.., |(_, t, _)| !upper.less_equal(t)); - session.give_iterator(pass_through); - - // Sort data by time, then drain it into a buffer that contains data for a - // single bucket. We scan the data for ranges of time that fall into the same - // bucket so we can push batches of data at once. - data.sort_unstable_by(|(_, t, _), (_, t2, _)| t.cmp(t2)); - - let mut drain = data.drain(..); - if let Some((datum, time, diff)) = drain.next() { - let mut range = chain.range_of(&time).expect("Must exist"); - buffer.push((datum, time, diff)); - for (datum, time, diff) in drain { - // If we have a range, check if the time is not within it. - if !range.contains(&time) { - // If the time is outside the range, push the current buffer - // to the chain and reset the range. - if !buffer.is_empty() { - let bucket = - chain.find_mut(&range.start).expect("Must exist"); - bucket.push_container(&mut buffer); - buffer.clear(); - } - range = chain.range_of(&time).expect("Must exist"); - } - buffer.push((datum, time, diff)); + let borrowed = data.borrow(); + + // Pass through data about to be revealed, and retain the + // index of everything the chain has to hold. Only the + // retained records need ordering, and in steady state the + // pass-through share is the larger one. + permutation.clear(); + for index in 0..borrowed.len() { + let update = borrowed.get(index); + time_buf.copy_from(update.1); + if upper.less_equal(&time_buf) { + permutation.push(index); + } else { + session.give(update); } + } - // Handle leftover data in the buffer. - if !buffer.is_empty() { - let bucket = chain.find_mut(&range.start).expect("Must exist"); - bucket.push_container(&mut buffer); - buffer.clear(); + // Order the retained records by time so each bucket's + // records land contiguously below. Sorting indices keeps + // the records in place. + permutation.sort_unstable_by_key(|index| borrowed.get(*index).1); + + // The range `buffer`'s contents belong to, `None` while empty. + let mut buffered_range = None; + for index in permutation.drain(..) { + let update = borrowed.get(index); + time_buf.copy_from(update.1); + + // Ship the buffer whenever the bucket changes, which + // the time order makes a single transition per bucket. + let contained = match &buffered_range { + Some(range) => BucketRange::contains(range, &time_buf), + None => false, + }; + if !contained { + if let Some(range) = buffered_range.take() { + let bucket = chain.find_mut(&range.start).expect("Must exist"); + bucket.push_container(&mut buffer); + } + buffered_range = + Some(chain.range_of(&time_buf).expect("Must exist")); } + buffer.push_into(update); + } + + // Handle leftover data in the buffer. + if let Some(range) = buffered_range.take() { + let bucket = chain.find_mut(&range.start).expect("Must exist"); + bucket.push_container(&mut buffer); } } }); @@ -136,16 +158,10 @@ where let peeled = chain.peel(upper.borrow()); if let Some(cap) = cap.as_ref() { let mut session = output.session_with_builder(cap); - for chunk in peeled.into_iter().flat_map(|x| x.done()) { - // The columnar merge batcher hands back `Column` chunks; the output - // builder consumes owned `(D, T, Diff)` tuples, so reconstitute each - // record from its columnar reference. - session.give_iterator( - chunk - .borrow() - .into_index_iter() - .map(<(D, T, mz_repr::Diff)>::into_owned), - ); + // The chain hands back `Column` chunks already in the output's + // shape, so each one moves as a container. + for mut chunk in peeled.into_iter().flat_map(|x| x.done()) { + session.give_container(&mut chunk); } } else { // If we don't have a cap, we should not have any data to reveal. @@ -178,13 +194,146 @@ where } } +/// Implementation for `Vec` streams in scopes where timestamps define a total order. +/// +/// A caller whose consumer wants owned records keeps a `Vec`-native operator, because +/// staging the whole stream through a column would copy every pass-through record and +/// allocate it again on the way out. Only records that enter the chain are encoded, which +/// they were anyway: the chain's batcher is columnar. The reduce key-value path is the one +/// such caller, and this implementation goes away once its consumer reads columns. +impl<'scope, T, D> TemporalBucketing<'scope, T> for StreamVec<'scope, T, (D, T, mz_repr::Diff)> +where + T: Timestamp + Default + ExchangeData + MzData + BucketTimestamp + TotalOrder + Lattice, + D: ExchangeData + MzData + Ord + Clone + std::fmt::Debug + Hashable, + for<'a> <(D, T, mz_repr::Diff) as Columnar>::Container: Push<&'a (D, T, mz_repr::Diff)>, +{ + fn bucket(self, as_of: Antichain, threshold: T::Summary) -> Self { + let scope = self.scope(); + let logger = scope + .worker() + .logger_for("differential/arrange") + .map(Into::into); + + let pact = Exchange::new(|(d, _, _): &(D, T, mz_repr::Diff)| d.hashed().into()); + self.unary_frontier::>, _, _, _>( + pact, + "Temporal delay", + |cap, info| { + let mut chain = BucketChain::new(MergeBatcherWrapper::new(logger, info.global_id)); + let activator = scope.activator_for(info.address); + + // Cap tracking the lower bound of potentially outstanding data. + let mut cap = Some(cap); + + // Staging column for the records of one bucket. The chain's batcher is + // columnar, so a stored record is encoded either way. + let mut buffer: Column<(D, T, mz_repr::Diff)> = Default::default(); + + move |(input, frontier), output| { + // The upper frontier is the join of the input frontier and the `as_of` + // frontier, with the `threshold` summary applied to it. + let mut upper = Antichain::new(); + for time1 in &frontier.frontier() { + for time2 in as_of.elements() { + // TODO: Use `join_assign` if we ever use a timestamp with allocations. + if let Some(time) = threshold.results_in(&time1.join(time2)) { + upper.insert(time); + } + } + } + + input.for_each_time(|time, data| { + let mut session = output.session_with_builder(&time); + for data in data { + // Skip data that is about to be revealed. + let pass_through = + data.extract_if(.., |(_, t, _)| !upper.less_equal(t)); + session.give_iterator(pass_through); + + // Sort data by time, then drain it into a buffer that contains data + // for a single bucket. We scan the data for ranges of time that fall + // into the same bucket so we can push batches of data at once. + data.sort_unstable_by(|(_, t, _), (_, t2, _)| t.cmp(t2)); + + let mut drain = data.drain(..); + if let Some(update) = drain.next() { + let mut range = chain.range_of(&update.1).expect("Must exist"); + buffer.push_into(&update); + for update in drain { + // If we have a range, check if the time is not within it. + if !range.contains(&update.1) { + // If the time is outside the range, push the current + // buffer to the chain and reset the range. + if !buffer.is_empty() { + let bucket = + chain.find_mut(&range.start).expect("Must exist"); + bucket.push_container(&mut buffer); + } + range = chain.range_of(&update.1).expect("Must exist"); + } + buffer.push_into(&update); + } + + // Handle leftover data in the buffer. + if !buffer.is_empty() { + let bucket = chain.find_mut(&range.start).expect("Must exist"); + bucket.push_container(&mut buffer); + } + } + } + }); + + // Check for data that is ready to be revealed. + let peeled = chain.peel(upper.borrow()); + if let Some(cap) = cap.as_ref() { + let mut session = output.session_with_builder(cap); + for chunk in peeled.into_iter().flat_map(|x| x.done()) { + session.give_iterator( + chunk + .borrow() + .into_index_iter() + .map(<(D, T, mz_repr::Diff)>::into_owned), + ); + } + } else { + // If we don't have a cap, we should not have any data to reveal. + assert!( + peeled + .into_iter() + .flat_map(|x| x.done()) + .all(|chunk| chunk.record_count() == 0), + "Unexpected data revealed without a cap." + ); + } + + // Downgrade the cap to the current input frontier. + if frontier.is_empty() || upper.is_empty() { + cap = None; + } else if let Some(cap) = cap.as_mut() { + // TODO: This assumes that the time is total ordered. + cap.downgrade(&upper[0]); + } + + // Maintain the bucket chain by restoring it with fuel. + let mut fuel = 1_000_000; + chain.restore(&mut fuel); + if fuel <= 0 { + // If we run out of fuel, we activate the operator to continue processing. + activator.activate(); + } + } + }, + ) + } +} + /// A wrapper around [`ColumnMergeBatcher`] that implements the bucketing API. /// /// This is the same columnar-native merge batcher (`Col2ValPagedBatcher`) the /// default arrangement uses, so the bucket chain and arrangements share a single /// merge-batcher implementation. The batcher consumes pre-chunked, consolidated /// [`Column`] input, so this wrapper carries a [`ColumnChunker`] that sorts and -/// consolidates the `Vec` input into the [`Column`] chunks the batcher consumes. +/// consolidates the input columns into the chunks the batcher consumes. struct MergeBatcherWrapper where D: MzData + Ord + Clone, @@ -203,10 +352,10 @@ where T: MzData + Ord + PartialOrder + Clone + Default + Timestamp, R: MzData + Semigroup + Default + 'static + for<'a> Semigroup>, for<'a> columnar::Ref<'a, R>: Ord, - for<'a> ::Container: columnar::Push>, - for<'a> ::Container: columnar::Push>, - for<'a> ::Container: columnar::Push<&'a R>, - for<'a> <(D, T, R) as Columnar>::Container: columnar::Push<&'a (D, T, R)>, + for<'a> ::Container: Push>, + for<'a> ::Container: Push>, + for<'a> ::Container: Push<&'a R>, + for<'a> <(D, T, R) as Columnar>::Container: Push<&'a (D, T, R)>, { /// Construct a new `MergeBatcherWrapper` with the given logger and operator ID. fn new(logger: Option, operator_id: usize) -> Self { @@ -219,19 +368,14 @@ where } /// Consolidate `buffer` through the chunker and feed any complete chunks to - /// the batcher. Empties `buffer`, retaining its capacity. - fn push_container(&mut self, buffer: &mut Vec<(D, T, R)>) { + /// the batcher. Leaves `buffer` empty, retaining its allocation. + fn push_container(&mut self, buffer: &mut Column<(D, T, R)>) { use timely::container::{ContainerBuilder as _, PushInto as _}; if buffer.is_empty() { return; } - // The chunker consumes `Column` input, so stage the `Vec` updates into a - // raw `Column` first; the chunker then sorts and consolidates them. - let mut raw: Column<(D, T, R)> = Default::default(); - for update in buffer.drain(..) { - raw.push_into(&update); - } - self.chunker.push_into(&mut raw); + self.chunker.push_into(buffer); + buffer.clear(); while let Some(chunk) = self.chunker.extract() { self.inner.push_into(std::mem::take(chunk)); } @@ -259,33 +403,27 @@ where T: MzData + Ord + PartialOrder + Clone + Default + 'static + BucketTimestamp, R: MzData + Semigroup + Default + 'static + for<'a> Semigroup>, for<'a> columnar::Ref<'a, R>: Ord, - for<'a> ::Container: columnar::Push>, - for<'a> ::Container: columnar::Push>, - for<'a> ::Container: columnar::Push<&'a R>, - for<'a> <(D, T, R) as Columnar>::Container: columnar::Push<&'a (D, T, R)>, + for<'a> ::Container: Push>, + for<'a> ::Container: Push>, + for<'a> ::Container: Push<&'a R>, + for<'a> <(D, T, R) as Columnar>::Container: Push<&'a (D, T, R)>, { type Timestamp = T; fn split(mut self, timestamp: &Self::Timestamp, fuel: &mut i64) -> (Self, Self) { - // The implementation isn't tuned for performance: we round-trip the sealed - // chunks back through a `Vec` and re-chunk them into the lower batcher. + // Re-chunks the sealed chunks into the lower batcher rather than splitting the + // batcher's chains in place, so the chunker sorts and re-pushes every record, + // which is what the per-record `fuel` charge below accounts for. No record is + // reconstituted as an owned tuple on the way. + // // TODO: Split the batcher's chains directly without re-chunking. self.flush(); let upper = Antichain::from_elem(timestamp.clone()); let mut lower = Self::new(self.logger.clone(), self.operator_id); - let mut buffer = Vec::new(); let (chain, _description) = self.inner.seal(upper); - for chunk in chain { + for mut chunk in chain { *fuel = fuel.saturating_sub(chunk.record_count()); - // TODO: Avoid this cloning. - buffer.extend( - chunk - .borrow() - .into_index_iter() - .map(<(D, T, R)>::into_owned), - ); - lower.push_container(&mut buffer); - buffer.clear(); + lower.push_container(&mut chunk); } (lower, self) } diff --git a/src/compute/src/render.rs b/src/compute/src/render.rs index dd7ffdf40bbb1..ae60942908a5a 100644 --- a/src/compute/src/render.rs +++ b/src/compute/src/render.rs @@ -110,6 +110,7 @@ use std::rc::{Rc, Weak}; use std::sync::Arc; use std::task::Poll; +use ::columnar::{Columnar as ColumnarData, Index as ColumnarIndex, Push as ColumnarPush}; use differential_dataflow::dynamic::pointstamp::PointStamp; use differential_dataflow::lattice::Lattice; use differential_dataflow::operators::arrange::Arranged; @@ -117,7 +118,7 @@ use differential_dataflow::operators::arrange::ShutdownButton; use differential_dataflow::operators::iterate::Variable; use differential_dataflow::trace::cursor::{BatchCursor, BatchDiff, BatchKey, BatchVal}; use differential_dataflow::trace::{BatchReader, Cursor, Navigable, TraceReader}; -use differential_dataflow::{AsCollection, Data, VecCollection}; +use differential_dataflow::{AsCollection, Collection, Data, VecCollection}; use futures::FutureExt; use futures::channel::oneshot; use itertools::Itertools; @@ -140,12 +141,12 @@ use mz_repr::fixed_length::ExtendDatums; use mz_repr::{Datum, DatumVec, Diff, GlobalId, ReprRelationType, Row, RowArena, SharedRow}; use mz_storage_operators::persist_source; use mz_storage_types::controller::CollectionMetadata; +use mz_timely_util::columnar::Column; use mz_timely_util::columnation::ColumnationChunker; use mz_timely_util::operator::{CollectionExt, StreamExt}; use mz_timely_util::probe::{Handle as MzProbeHandle, ProbeNotify}; use mz_timely_util::scope_label::ScopeExt; use timely::PartialOrder; -use timely::container::CapacityContainerBuilder; use timely::dataflow::channels::pact::Pipeline; use timely::dataflow::operators::core::to_stream::ToStreamBuilder; use timely::dataflow::operators::vec::ToStream; @@ -292,22 +293,24 @@ pub fn build_compute_dataflow( // Note: For correctness, we require that sources only emit times advanced by // `dataflow.as_of`. `persist_source` is documented to provide this guarantee. - let (mut ok_stream, err_stream, token) = - persist_source::persist_source::( - inner, - *source_id, - Arc::clone(&compute_state.persist_clients), - &compute_state.txns_ctx, - import.desc.storage_metadata.clone(), - read_schema, - dataflow.as_of.clone(), - snapshot_mode, - until.clone(), - mfp.as_mut(), - compute_state.dataflow_max_inflight_bytes(), - start_signal.clone().into_send_future(), - ErrorHandler::Halt("compute_import"), - ); + let (mut ok_stream, err_stream, token) = persist_source::persist_source::< + DataflowErrorSer, + ConsolidatingColumnBuilder, + >( + inner, + *source_id, + Arc::clone(&compute_state.persist_clients), + &compute_state.txns_ctx, + import.desc.storage_metadata.clone(), + read_schema, + dataflow.as_of.clone(), + snapshot_mode, + until.clone(), + mfp.as_mut(), + compute_state.dataflow_max_inflight_bytes(), + start_signal.clone().into_send_future(), + ErrorHandler::Halt("compute_import"), + ); // If `mfp` is non-identity, we need to apply what remains. // For the moment, assert that it is either trivial or `None`. @@ -377,10 +380,8 @@ pub fn build_compute_dataflow( ); for (id, (oks, errs)) in imported_sources.into_iter() { - // Persist batches are row-shaped and already consolidated, so the - // encode here is non-consolidating. let bundle = crate::render::CollectionBundle::from_edge( - vec_to_columnar(oks.enter(region)), + oks.enter(region), errs.enter(region), ); // Associate collection bundle with the source identifier. @@ -479,10 +480,8 @@ pub fn build_compute_dataflow( ); for (id, (oks, errs)) in imported_sources.into_iter() { - // Persist batches are row-shaped and already consolidated, so the - // encode here is non-consolidating. let bundle = crate::render::CollectionBundle::from_edge( - vec_to_columnar(oks.enter_region(region)), + oks.enter_region(region), errs.enter_region(region), ); // Associate collection bundle with the source identifier. @@ -1431,14 +1430,11 @@ impl<'scope, T: RenderTimestamp + MaybeBucketByTime> Context<'scope, T> { .get(&self.config_set) .try_into() .expect("must fit"); - // Temporal bucketing is `Vec`-internal, so decode in and - // encode out, keeping this Union input a columnar edge. - let os = columnar_to_vec(os); - vec_to_columnar(T::maybe_apply_temporal_bucketing( + T::maybe_apply_temporal_bucketing( os.inner, self.as_of_frontier.clone(), summary, - )) + ) } else { os }; @@ -1627,8 +1623,30 @@ pub trait RenderTimestamp: MzTimestamp + Default + Refines { /// general property of a render timestamp, so the dispatch lives in its own trait. /// Total-ordered timestamps perform real bucketing; partially-ordered timestamps /// (e.g. `Product<…>` in iterative scopes) implement this as a no-op. -pub trait MaybeBucketByTime: Timestamp { +pub trait MaybeBucketByTime: Timestamp + ColumnarData { + /// Buckets a columnar dataflow edge, keeping it columnar. fn maybe_apply_temporal_bucketing<'scope, D>( + stream: Stream<'scope, Self, Column<(D, Self, Diff)>>, + as_of: Antichain, + summary: mz_repr::Timestamp, + ) -> Collection<'scope, Self, Column<(D, Self, Diff)>> + where + D: differential_dataflow::ExchangeData + + crate::typedefs::MzData + + differential_dataflow::Hashable + + ColumnarData, + for<'a> ::columnar::Ref<'a, D>: Copy + Ord + std::hash::Hash, + for<'a> ::columnar::Ref<'a, Self>: Copy + Ord, + for<'a> ::columnar::Ref<'a, Diff>: Ord, + for<'a> <(D, Self, Diff) as ColumnarData>::Container: + ColumnarPush<&'a (D, Self, Diff)> + ColumnarPush<::columnar::Ref<'a, (D, Self, Diff)>>; + + /// Buckets a `Vec` stream, keeping it `Vec`. + /// + /// For a consumer that re-encodes what it reads, where a `Vec` hands it moved + /// allocations rather than copied bytes. The reduce key-value path is the one + /// such caller, since its bucketed output feeds an arrangement. + fn maybe_apply_temporal_bucketing_vec<'scope, D>( stream: StreamVec<'scope, Self, (D, Self, Diff)>, as_of: Antichain, summary: mz_repr::Timestamp, @@ -1636,7 +1654,13 @@ pub trait MaybeBucketByTime: Timestamp { where D: differential_dataflow::ExchangeData + crate::typedefs::MzData - + differential_dataflow::Hashable; + + differential_dataflow::Hashable + + ColumnarData, + for<'a> ::columnar::Ref<'a, D>: Copy + Ord + std::hash::Hash, + for<'a> ::columnar::Ref<'a, Self>: Copy + Ord, + for<'a> ::columnar::Ref<'a, Diff>: Ord, + for<'a> <(D, Self, Diff) as ColumnarData>::Container: + ColumnarPush<&'a (D, Self, Diff)> + ColumnarPush<::columnar::Ref<'a, (D, Self, Diff)>>; } impl RenderTimestamp for mz_repr::Timestamp { @@ -1662,6 +1686,25 @@ impl RenderTimestamp for mz_repr::Timestamp { impl MaybeBucketByTime for mz_repr::Timestamp { fn maybe_apply_temporal_bucketing<'scope, D>( + stream: Stream<'scope, Self, Column<(D, Self, Diff)>>, + as_of: Antichain, + summary: mz_repr::Timestamp, + ) -> Collection<'scope, Self, Column<(D, Self, Diff)>> + where + D: differential_dataflow::ExchangeData + + crate::typedefs::MzData + + differential_dataflow::Hashable + + ColumnarData, + for<'a> ::columnar::Ref<'a, D>: Copy + Ord + std::hash::Hash, + for<'a> ::columnar::Ref<'a, Self>: Copy + Ord, + for<'a> ::columnar::Ref<'a, Diff>: Ord, + for<'a> <(D, Self, Diff) as ColumnarData>::Container: + ColumnarPush<&'a (D, Self, Diff)> + ColumnarPush<::columnar::Ref<'a, (D, Self, Diff)>>, + { + stream.bucket(as_of, summary).as_collection() + } + + fn maybe_apply_temporal_bucketing_vec<'scope, D>( stream: StreamVec<'scope, Self, (D, Self, Diff)>, as_of: Antichain, summary: mz_repr::Timestamp, @@ -1669,11 +1712,15 @@ impl MaybeBucketByTime for mz_repr::Timestamp { where D: differential_dataflow::ExchangeData + crate::typedefs::MzData - + differential_dataflow::Hashable, + + differential_dataflow::Hashable + + ColumnarData, + for<'a> ::columnar::Ref<'a, D>: Copy + Ord + std::hash::Hash, + for<'a> ::columnar::Ref<'a, Self>: Copy + Ord, + for<'a> ::columnar::Ref<'a, Diff>: Ord, + for<'a> <(D, Self, Diff) as ColumnarData>::Container: + ColumnarPush<&'a (D, Self, Diff)> + ColumnarPush<::columnar::Ref<'a, (D, Self, Diff)>>, { - stream - .bucket::>(as_of, summary) - .as_collection() + stream.bucket(as_of, summary).as_collection() } } @@ -1708,6 +1755,26 @@ impl RenderTimestamp for Product> { impl MaybeBucketByTime for Product> { fn maybe_apply_temporal_bucketing<'scope, D>( + stream: Stream<'scope, Self, Column<(D, Self, Diff)>>, + _as_of: Antichain, + _summary: mz_repr::Timestamp, + ) -> Collection<'scope, Self, Column<(D, Self, Diff)>> + where + D: differential_dataflow::ExchangeData + + crate::typedefs::MzData + + differential_dataflow::Hashable + + ColumnarData, + for<'a> ::columnar::Ref<'a, D>: Copy + Ord + std::hash::Hash, + for<'a> ::columnar::Ref<'a, Self>: Copy + Ord, + for<'a> ::columnar::Ref<'a, Diff>: Ord, + for<'a> <(D, Self, Diff) as ColumnarData>::Container: + ColumnarPush<&'a (D, Self, Diff)> + ColumnarPush<::columnar::Ref<'a, (D, Self, Diff)>>, + { + // TODO: Implement bucketing on outer timestamp for iterative scopes. + stream.as_collection() + } + + fn maybe_apply_temporal_bucketing_vec<'scope, D>( stream: StreamVec<'scope, Self, (D, Self, Diff)>, _as_of: Antichain, _summary: mz_repr::Timestamp, @@ -1715,7 +1782,13 @@ impl MaybeBucketByTime for Product> { where D: differential_dataflow::ExchangeData + crate::typedefs::MzData - + differential_dataflow::Hashable, + + differential_dataflow::Hashable + + ColumnarData, + for<'a> ::columnar::Ref<'a, D>: Copy + Ord + std::hash::Hash, + for<'a> ::columnar::Ref<'a, Self>: Copy + Ord, + for<'a> ::columnar::Ref<'a, Diff>: Ord, + for<'a> <(D, Self, Diff) as ColumnarData>::Container: + ColumnarPush<&'a (D, Self, Diff)> + ColumnarPush<::columnar::Ref<'a, (D, Self, Diff)>>, { // TODO: Implement bucketing on outer timestamp for iterative scopes. stream.as_collection() @@ -1865,7 +1938,7 @@ fn suppress_early_progress<'scope, T: Timestamp, D>( as_of: Antichain, ) -> Stream<'scope, T, D> where - D: Data + timely::Container, + D: timely::Container + Clone, { stream.unary_frontier(Pipeline, "SuppressEarlyProgress", |default_cap, _info| { let mut early_cap = Some(default_cap); @@ -1932,13 +2005,44 @@ trait LimitProgress { ) -> Self; } +/// Reads the times of the records a container holds, for [`LimitProgress`]. +trait RecordTimes { + /// Call `f` once per record, with that record's time. + fn for_each_time(&self, f: impl FnMut(mz_repr::Timestamp)); +} + +impl RecordTimes for Vec<(D, mz_repr::Timestamp, R)> { + fn for_each_time(&self, mut f: impl FnMut(mz_repr::Timestamp)) { + for (_, time, _) in self { + f(*time); + } + } +} + +impl RecordTimes for Column<(D, mz_repr::Timestamp, R)> +where + D: ColumnarData, + R: ColumnarData, + (D, mz_repr::Timestamp, R): ColumnarData< + Container = ( + D::Container, + ::Container, + R::Container, + ), + >, +{ + fn for_each_time(&self, mut f: impl FnMut(mz_repr::Timestamp)) { + for time in self.borrow().1.into_index_iter() { + f(time); + } + } +} + // TODO: We could make this generic over a `T` that can be converted to and from a u64 millisecond // number. -impl<'scope, D, R> LimitProgress - for StreamVec<'scope, mz_repr::Timestamp, (D, mz_repr::Timestamp, R)> +impl<'scope, C> LimitProgress for Stream<'scope, mz_repr::Timestamp, C> where - D: Clone + 'static, - R: Clone + 'static, + C: timely::Container + Clone + RecordTimes, { fn limit_progress( self, @@ -1961,10 +2065,10 @@ where move |(input, frontier), output| { input.for_each(|cap, data| { - for time in data - .iter() - .flat_map(|(_, time, _)| u64::from(time).checked_add(slack_ms)) - { + data.for_each_time(|time| { + let Some(time) = u64::from(time).checked_add(slack_ms) else { + return; + }; // `slack_ms == 0` means no rounding; otherwise round up to the next // multiple of `slack_ms`. Avoids a divide-by-zero panic when the // operator is configured without slack. @@ -1976,7 +2080,7 @@ where if !upper.less_than(&rounded_time.into()) { pending_times.insert(rounded_time.into()); } - } + }); output.session(&cap).give_container(data); if retained_cap.as_ref().is_none_or(|c| { !c.time().less_than(cap.time()) && !upper.less_than(cap.time()) diff --git a/src/compute/src/render/context.rs b/src/compute/src/render/context.rs index 4ce4ad6bec604..dcda673932f39 100644 --- a/src/compute/src/render/context.rs +++ b/src/compute/src/render/context.rs @@ -51,7 +51,7 @@ use timely::progress::{Antichain, Timestamp}; use crate::compute_state::ComputeState; use crate::extensions::arrange::{ArrangementBatcher, KeyCollection, MzArrange, MzArrangeCore}; use crate::extensions::reduce::MzReduce; -use crate::render::columnar::{CollectionEdge, columnar_to_vec, flat_map_datums, vec_to_columnar}; +use crate::render::columnar::{CollectionEdge, flat_map_datums}; use crate::render::errors::{DataflowErrorSer, ErrorLogger}; use crate::render::{LinearJoinSpec, MaybeBucketByTime, RenderTimestamp}; use crate::typedefs::{ @@ -1095,12 +1095,8 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { .try_into() .expect("must fit"); bucketed = true; - // Temporal bucketing is `Vec`-internal, so decode in and encode out. - vec_to_columnar(T::maybe_apply_temporal_bucketing( - columnar_to_vec(oks).inner, - as_of.clone(), - summary, - )) + // Temporal bucketing is columnar throughout, so no round trip here. + T::maybe_apply_temporal_bucketing(oks.inner, as_of.clone(), summary) } else { oks }; @@ -1132,13 +1128,8 @@ impl<'scope, T: RenderTimestamp> CollectionBundle<'scope, T> { .try_into() .expect("must fit"); bucketed = true; - // Temporal bucketing is `Vec`-internal, so decode in and encode out. - let oks = columnar_to_vec(oks); - vec_to_columnar(T::maybe_apply_temporal_bucketing( - oks.inner, - as_of.clone(), - summary, - )) + // Temporal bucketing is columnar throughout, so no round trip here. + T::maybe_apply_temporal_bucketing(oks.inner, as_of.clone(), summary) } else { oks }; @@ -1472,7 +1463,7 @@ mod tests { use timely::dataflow::operators::capture::{Event, Extract}; use super::*; - use crate::render::columnar::vec_to_columnar; + use crate::render::columnar::{columnar_to_vec, vec_to_columnar}; type OkUpdate = ((Row, Row), Timestamp, Diff); type ErrUpdate = (DataflowErrorSer, Timestamp, Diff); diff --git a/src/compute/src/render/join/linear_join.rs b/src/compute/src/render/join/linear_join.rs index 0b960c688614f..53cd8fa218dec 100644 --- a/src/compute/src/render/join/linear_join.rs +++ b/src/compute/src/render/join/linear_join.rs @@ -36,8 +36,11 @@ use mz_timely_util::columnar::{ Col2ValBatcher, Col2ValColBatcher, Col2ValPagedBatcher, columnar_exchange, }; use mz_timely_util::operator::{CollectionExt, StreamExt}; +use timely::ContainerBuilder; +use timely::container::{CapacityContainerBuilder, PushInto}; use timely::dataflow::channels::pact::{ExchangeCore, Pipeline}; use timely::dataflow::operators::OkErr; +use timely::dataflow::operators::generic::Operator; use timely::dataflow::{Scope, Stream}; use crate::extensions::arrange::{ArrangementBatcher, MzArrangeCore}; @@ -100,15 +103,21 @@ impl LinearJoinSpec { } } - /// Render a join operator according to this specification. - fn render<'s, T, Tr1, Tr2, L, I>( + /// Render a join operator according to this specification, assembling its + /// output through `CB`. + /// + /// The `DifferentialDataflow` implementation builds its own `Vec` output and + /// cannot be handed a container builder, so that arm re-encodes through `CB`. + /// The `Materialize` implementation writes `CB` directly. + fn render<'s, T, Tr1, Tr2, L, I, CB>( &self, arranged1: Arranged<'s, Tr1>, arranged2: Arranged<'s, Tr2>, result: L, - ) -> VecCollection<'s, T, I::Item, Diff> + ) -> Stream<'s, T, CB::Container> where T: Lattice + timely::progress::Timestamp, + CB: ContainerBuilder + PushInto<(I::Item, T, Diff)> + 'static, Tr1: TraceReader + Clone + 'static, Tr2: TraceReader + Clone + 'static, BatchCursor: Cursor