DNM: upsert v2 hydration optimizations and payload-separated state

Cargo.toml

@@ -342,6 +342,7 @@ deadpool-postgres = "0.10.3"
 dec = "0.4.9"
 derivative = "2.2.0"
 differential-dataflow = "0.25.0"
+differential-dataflow-next = { package = "differential-dataflow", git = "https://github.com/DAlperin/differential-dataflow", rev = "3659671246e0b5635a8128cc27d65f12ecf7dda5", default-features = false }
 differential-dogs3 = "0.25.0"
 digest = "0.10.7"
 dirs = "6.0.0"
@@ -526,6 +527,7 @@ tikv-jemalloc-ctl = { version = "0.6", features = ["stats", "use_std"] }
 tikv-jemallocator = { version = "0.6", features = ["background_threads", "profiling", "stats", "unprefixed_malloc_on_supported_platforms"] }
 time = "0.3.17"
 timely = "0.31.0"
+timely-next = { package = "timely", git = "https://github.com/TimelyDataflow/timely-dataflow", default-features = false }
 tokio = { version = "1.52.3", features = ["full", "test-util"] }
 tokio-metrics = "0.5.0"
 tokio-native-tls = "0.3.1"

deny.toml

@@ -16,6 +16,14 @@ multiple-versions = "deny"
 # Materialize-maintained fork that avoids the duplicated transitive
 # dependencies.
 skip = [
+    # The DNM async-merge prototype consumes two DD/Timely APIs.
+    { name = "differential-dataflow", version = "=0.25.1" },
+    { name = "timely", version = "=0.31.0" },
+    { name = "timely_bytes", version = "=0.31.0" },
+    { name = "timely_communication", version = "=0.31.0" },
+    { name = "timely_container", version = "=0.31.0" },
+    { name = "timely_logging", version = "=0.31.0" },
+
     # arrayvec had a significant API change in 0.7
     { name = "arrayvec", version = "0.5.2" },
     # One-time exception for base64 due to its prevalence in the crate graph.
@@ -367,6 +375,11 @@ license-files = [
 [sources]
 unknown-git = "deny"
 unknown-registry = "deny"
+# DNM-only exceptions for the async-merge prototype, pinned in Cargo.lock.
+allow-git = [
+    "https://github.com/DAlperin/differential-dataflow",
+    "https://github.com/TimelyDataflow/timely-dataflow",
+]
 # Do not allow non-MaterializeInc Git repositories here! Git repositories must
 # be owned by the MaterializeInc organization so that maintainership is shared
 # amongst Materialize employees and so that historical versions of Materialize

doc/user/data/metrics.yml

@@ -556,6 +556,22 @@ metrics:
   help: Evicted chunks re-admitted by an admitting read stealing the slot of a clean backed victim of the same size class.
   source: src/timely-util/src/pool_config/metrics.rs
   visibility: internal
+- name: mz_column_pool_async_reads_in_flight
+  help: Submitted pool reads that have not released their concurrency permit.
+  source: src/timely-util/src/pool_config/metrics.rs
+  visibility: internal
+- name: mz_column_pool_async_reads_total
+  help: Pool reads submitted to the blocking executor.
+  source: src/timely-util/src/pool_config/metrics.rs
+  visibility: internal
+- name: mz_column_pool_cold_inserts_total
+  help: Chunks inserted directly into an extent by caller request.
+  source: src/timely-util/src/pool_config/metrics.rs
+  visibility: internal
+- name: mz_column_pool_direct_extent_inserts_total
+  help: Inserts written directly to an extent because resident admission was full.
+  source: src/timely-util/src/pool_config/metrics.rs
+  visibility: internal
 - name: mz_column_pool_eager_backs_total
   help: Chunks eagerly compressed to compressed-but-resident by idle spill threads; their later eviction is a pure page release.
   source: src/timely-util/src/pool_config/metrics.rs

misc/python/materialize/mzcompose/__init__.py

@@ -364,6 +364,26 @@ def get_variable_system_parameters(
             "true",
             ["true", "false"],
         ),
+        VariableSystemParameter(
+            "enable_column_chunk_direct_compressed_output",
+            "true",
+            ["true", "false"],
+        ),
+        VariableSystemParameter(
+            "enable_upsert_payload_stash",
+            "true",
+            ["true", "false"],
+        ),
+        VariableSystemParameter(
+            "enable_upsert_async_merges",
+            "true",
+            ["true", "false"],
+        ),
+        VariableSystemParameter(
+            "enable_upsert_async_reads",
+            "true",
+            ["true", "false"],
+        ),
         VariableSystemParameter(
             "enable_upsert_v2",
             "false",

misc/python/materialize/parallel_workload/action.py

@@ -3179,6 +3179,11 @@ def __init__(
         ]
         self.flags_with_values["enable_upsert_paged_spill"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_upsert_chunked_stash"] = BOOLEAN_FLAG_VALUES
+        self.flags_with_values["enable_upsert_payload_stash"] = BOOLEAN_FLAG_VALUES
+        self.flags_with_values["enable_upsert_async_reads"] = BOOLEAN_FLAG_VALUES
+        self.flags_with_values["enable_column_chunk_direct_compressed_output"] = (
+            BOOLEAN_FLAG_VALUES
+        )
         self.flags_with_values["column_chunk_compress_min_depth"] = [
             "0",  # compress every spilled body
             "1",  # the default: fresh chunks store uncompressed

src/cluster/src/client.rs

@@ -10,6 +10,7 @@
 //! An interactive cluster server.
 
 use std::fmt;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use std::thread::Thread;
 
@@ -18,6 +19,8 @@ use async_trait::async_trait;
 use differential_dataflow::trace::ExertionLogic;
 use futures::future;
 use mz_cluster_client::client::{TimelyConfig, TryIntoProtocolNonce};
+use mz_ore::metric;
+use mz_ore::metrics::{ComputedUIntGauge, MetricsRegistry};
 use mz_service::client::{GenericClient, Partitionable, Partitioned};
 use mz_service::local::LocalClient;
 use timely::WorkerConfig;
@@ -225,6 +228,7 @@ pub trait ClusterSpec: Clone + Send + Sync + 'static {
             // largest to the smallest layer.
 
             let arc: ExertionLogic = Arc::new(move |layers| {
+                EXERT_POLICY_CALLS.fetch_add(1, Ordering::Relaxed);
                 let mut prop = config.arrangement_exert_proportionality;
 
                 // Layers are ordered from largest to smallest.
@@ -238,12 +242,14 @@ pub trait ClusterSpec: Clone + Send + Sync + 'static {
                 for (_idx, count, len) in layers {
                     if count > 1 {
                         // Found an in-progress merge that we should continue.
+                        EXERT_POLICY_MERGE_GRANTS.fetch_add(1, Ordering::Relaxed);
                         return merge_effort;
                     }
 
                     if !first && prop > 0 && len > 0 {
                         // Found a non-empty batch within `arrangement_exert_proportionality` of
                         // the largest one.
+                        EXERT_POLICY_CONSOLIDATION_GRANTS.fetch_add(1, Ordering::Relaxed);
                         return merge_effort;
                     }
 
@@ -284,6 +290,29 @@ pub trait ClusterSpec: Clone + Send + Sync + 'static {
     }
 }
 
+// The exertion policy runs on every arrangement's scheduling turn, so these
+// process-wide counts describe how often traces of every implementation were
+// offered optional effort, and for which reason.
+static EXERT_POLICY_CALLS: AtomicU64 = AtomicU64::new(0);
+static EXERT_POLICY_MERGE_GRANTS: AtomicU64 = AtomicU64::new(0);
+static EXERT_POLICY_CONSOLIDATION_GRANTS: AtomicU64 = AtomicU64::new(0);
+
+/// Register gauges for the arrangement exertion policy's decisions.
+pub fn register_exert_policy_metrics(registry: &MetricsRegistry) {
+    let _: ComputedUIntGauge = registry.register_computed_gauge(
+        metric!(name: "mz_arrangement_exert_policy_calls_total", help: "Arrangement maintenance turns that consulted the exertion policy."),
+        || EXERT_POLICY_CALLS.load(Ordering::Relaxed),
+    );
+    let _: ComputedUIntGauge = registry.register_computed_gauge(
+        metric!(name: "mz_arrangement_exert_policy_grants_total", help: "Exertion policy grants, by reason.", const_labels: {"reason" => "active_merge"}),
+        || EXERT_POLICY_MERGE_GRANTS.load(Ordering::Relaxed),
+    );
+    let _: ComputedUIntGauge = registry.register_computed_gauge(
+        metric!(name: "mz_arrangement_exert_policy_grants_total", help: "Exertion policy grants, by reason.", const_labels: {"reason" => "consolidation"}),
+        || EXERT_POLICY_CONSOLIDATION_GRANTS.load(Ordering::Relaxed),
+    );
+}
+
 mod alloc {
     /// A Timely communication refill function that uses lgalloc.
     ///

src/compute-types/src/dyncfgs.rs

@@ -134,6 +134,15 @@ pub const COLUMN_CHUNK_COMPRESS_MIN_DEPTH: Config<u32> = Config::new(
     ParameterScope::Replica,
 );
 
+/// Bypass resident pool slots when committing compression-eligible chunk bodies.
+pub const ENABLE_COLUMN_CHUNK_DIRECT_COMPRESSED_OUTPUT: Config<bool> = Config::new(
+    "enable_column_chunk_direct_compressed_output",
+    false,
+    "Compress spilled chunk output directly into extents without resident slot admission. \
+     Applies at the compression depth floor and above, at each spill.",
+    ParameterScope::Replica,
+);
+
 /// Resident-bytes budget fraction for chunk spilling. Two consumers read
 /// it: the column pager's tiered policy multiplies it against the
 /// announced memory limit, and the buffer pool (`mz_ore::pool`)
@@ -852,4 +861,5 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&COLUMN_PAGED_BATCHER_EAGER_BACKING)
         .add(&COLUMN_PAGED_BATCHER_POOL_RSS_TARGET_FRACTION)
         .add(&COLUMN_CHUNK_COMPRESS_MIN_DEPTH)
+        .add(&ENABLE_COLUMN_CHUNK_DIRECT_COMPRESSED_OUTPUT)
 }

src/compute/src/compute_state.rs

@@ -581,6 +581,9 @@ impl ComputeState {
             let compress_min_depth =
                 u8::try_from(COLUMN_CHUNK_COMPRESS_MIN_DEPTH.get(config)).unwrap_or(u8::MAX);
             mz_timely_util::columnar::chunk::set_compress_min_depth(compress_min_depth);
+            mz_timely_util::columnar::chunk::set_direct_compressed_output(
+                ENABLE_COLUMN_CHUNK_DIRECT_COMPRESSED_OUTPUT.get(config),
+            );
         }
 
         // Remember the maintenance interval locally to avoid reading it from the config set on

src/compute/src/server.rs

@@ -175,6 +175,7 @@ pub async fn serve(
         mz_timely_util::column_pager::tiered_policy(),
     );
     mz_timely_util::pool_config::metrics::register(metrics_registry);
+    mz_cluster::client::register_exert_policy_metrics(metrics_registry);
 
     let config = Config {
         persist_clients,

src/controller-types/src/dyncfgs.rs

@@ -76,6 +76,17 @@ pub const ARRANGEMENT_EXERT_PROPORTIONALITY: Config<u32> = Config::new(
     ParameterScope::Replica,
 );
 
+/// The proportionality halves per layer below the largest, so 16 lets the
+/// policy request consolidation of layers within four levels of the largest
+/// batch, the same reach compute uses. Larger values let every small published
+/// batch be lifted into the largest batch on idle turns.
+pub const STORAGE_ARRANGEMENT_EXERT_PROPORTIONALITY: Config<u32> = Config::new(
+    "storage_arrangement_exert_proportionality",
+    16,
+    "Storage arrangement maintenance proportionality. Zero disables optional maintenance. Applies when a replica is provisioned.",
+    ParameterScope::Replica,
+);
+
 pub const ENABLE_PAUSED_CLUSTER_READHOLD_DOWNGRADE: Config<bool> = Config::new(
     "enable_paused_cluster_readhold_downgrade",
     true,
@@ -94,5 +105,6 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&ENABLE_TIMELY_ZERO_COPY_LGALLOC)
         .add(&TIMELY_ZERO_COPY_LIMIT)
         .add(&ARRANGEMENT_EXERT_PROPORTIONALITY)
+        .add(&STORAGE_ARRANGEMENT_EXERT_PROPORTIONALITY)
         .add(&ENABLE_PAUSED_CLUSTER_READHOLD_DOWNGRADE)
 }

src/controller/src/clusters.rs

@@ -26,7 +26,8 @@ use mz_compute_client::logging::LogVariant;
 use mz_compute_types::config::{ComputeReplicaConfig, ComputeReplicaLogging};
 use mz_controller_types::dyncfgs::{
     ARRANGEMENT_EXERT_PROPORTIONALITY, CONTROLLER_PAST_GENERATION_REPLICA_CLEANUP_RETRY_INTERVAL,
-    ENABLE_TIMELY_ZERO_COPY, ENABLE_TIMELY_ZERO_COPY_LGALLOC, TIMELY_ZERO_COPY_LIMIT,
+    ENABLE_TIMELY_ZERO_COPY, ENABLE_TIMELY_ZERO_COPY_LGALLOC,
+    STORAGE_ARRANGEMENT_EXERT_PROPORTIONALITY, TIMELY_ZERO_COPY_LIMIT,
 };
 use mz_controller_types::{ClusterId, ReplicaId};
 use mz_orchestrator::NamespacedOrchestrator;
@@ -704,18 +705,18 @@ impl Controller {
         let persist_pubsub_url = self.persist_pubsub_url.clone();
         let secrets_args = self.secrets_args.to_flags();
 
-        // TODO(teskje): use the same values as for compute?
-        let storage_proto_timely_config = TimelyConfig {
-            arrangement_exert_proportionality: 1337,
-            ..Default::default()
-        };
         // These configure the replica's process rather than environmentd's, so
         // they are `ParameterScope::Replica` and must be read through this
         // replica's scoped overrides. They are baked into the process
         // configuration at provisioning time, so a later change to either the
         // environment-wide value or the override reaches the replica only when
         // it is next provisioned.
         let overrides = self.replica_dyncfg_overrides.get(&replica_id);
+        let storage_proto_timely_config = TimelyConfig {
+            arrangement_exert_proportionality: STORAGE_ARRANGEMENT_EXERT_PROPORTIONALITY
+                .get_with_overrides(&self.dyncfg, overrides),
+            ..Default::default()
+        };
         let compute_proto_timely_config = TimelyConfig {
             arrangement_exert_proportionality: ARRANGEMENT_EXERT_PROPORTIONALITY
                 .get_with_overrides(&self.dyncfg, overrides),