adapter: add SystemVar linear_join_yielding

This commit adds a new `SystemVar` called `linear_join_yielding` that
can be used to control the yielding behavior of linear joins rendered by
the compute layer.

Author: teskje

src/adapter/src/flags.rs

@@ -10,23 +10,30 @@
 use std::time::Duration;
 
 use mz_compute_client::protocol::command::ComputeParameters;
+use mz_compute_types::dataflows::YieldSpec;
 use mz_orchestrator::scheduling_config::{ServiceSchedulingConfig, ServiceTopologySpreadConfig};
 use mz_ore::cast::CastFrom;
 use mz_ore::error::ErrorExt;
 use mz_persist_client::cfg::{PersistParameters, RetryParameters};
 use mz_service::params::GrpcClientParameters;
-use mz_sql::session::vars::SystemVars;
+use mz_sql::session::vars::{SystemVars, DEFAULT_LINEAR_JOIN_YIELDING};
 use mz_storage_types::parameters::{
     StorageMaxInflightBytesConfig, StorageParameters, UpsertAutoSpillConfig,
 };
 use mz_tracing::params::TracingParameters;
 
 /// Return the current compute configuration, derived from the system configuration.
 pub fn compute_config(config: &SystemVars) -> ComputeParameters {
+    let linear_join_yielding = config.linear_join_yielding();
+    let linear_join_yielding = parse_yield_spec(linear_join_yielding).unwrap_or_else(|| {
+        tracing::error!("invalid `linear_join_yielding` config: {linear_join_yielding}");
+        parse_yield_spec(&DEFAULT_LINEAR_JOIN_YIELDING).expect("default is valid")
+    });
+
     ComputeParameters {
         max_result_size: Some(config.max_result_size()),
         dataflow_max_inflight_bytes: Some(config.dataflow_max_inflight_bytes()),
-        linear_join_yielding: None,
+        linear_join_yielding: Some(linear_join_yielding),
         enable_mz_join_core: Some(config.enable_mz_join_core()),
         enable_jemalloc_profiling: Some(config.enable_jemalloc_profiling()),
         persist: persist_config(config),
@@ -35,6 +42,22 @@ pub fn compute_config(config: &SystemVars) -> ComputeParameters {
     }
 }
 
+fn parse_yield_spec(s: &str) -> Option<YieldSpec> {
+    let parts: Vec<_> = s.split(':').collect();
+    match &parts[..] {
+        ["work", amount] => {
+            let amount = amount.parse().ok()?;
+            Some(YieldSpec::ByWork(amount))
+        }
+        ["time", millis] => {
+            let millis = millis.parse().ok()?;
+            let duration = Duration::from_millis(millis);
+            Some(YieldSpec::ByTime(duration))
+        }
+        _ => None,
+    }
+}
+
 /// Return the current storage configuration, derived from the system configuration.
 pub fn storage_config(config: &SystemVars) -> StorageParameters {
     StorageParameters {

src/sql/src/session/vars.rs

@@ -1289,6 +1289,16 @@ const ENABLE_MZ_JOIN_CORE: ServerVar<bool> = ServerVar {
     internal: true,
 };
 
+pub static DEFAULT_LINEAR_JOIN_YIELDING: Lazy<String> = Lazy::new(|| "work:1000000".into());
+static LINEAR_JOIN_YIELDING: Lazy<ServerVar<String>> = Lazy::new(|| ServerVar {
+    name: UncasedStr::new("linear_join_yielding"),
+    value: &DEFAULT_LINEAR_JOIN_YIELDING,
+    description:
+        "The yielding behavior compute rendering should apply for linear join operators. Either \
+         'work:<amount>' or 'time:<milliseconds>'.",
+    internal: true,
+});
+
 pub const ENABLE_DEFAULT_CONNECTION_VALIDATION: ServerVar<bool> = ServerVar {
     name: UncasedStr::new("enable_default_connection_validation"),
     value: &true,
@@ -2428,6 +2438,7 @@ impl SystemVars {
             .with_var(&KEEP_N_SOURCE_STATUS_HISTORY_ENTRIES)
             .with_var(&KEEP_N_SINK_STATUS_HISTORY_ENTRIES)
             .with_var(&ENABLE_MZ_JOIN_CORE)
+            .with_var(&LINEAR_JOIN_YIELDING)
             .with_var(&ENABLE_STORAGE_SHARD_FINALIZATION)
             .with_var(&ENABLE_CONSOLIDATE_AFTER_UNION_NEGATE)
             .with_var(&ENABLE_DEFAULT_CONNECTION_VALIDATION)
@@ -3079,6 +3090,11 @@ impl SystemVars {
         *self.expect_value(&ENABLE_MZ_JOIN_CORE)
     }
 
+    /// Returns the `linear_join_yielding` configuration parameter.
+    pub fn linear_join_yielding(&self) -> &String {
+        self.expect_value(&LINEAR_JOIN_YIELDING)
+    }
+
     /// Returns the `enable_storage_shard_finalization` configuration parameter.
     pub fn enable_storage_shard_finalization(&self) -> bool {
         *self.expect_value(&ENABLE_STORAGE_SHARD_FINALIZATION)
@@ -4744,6 +4760,7 @@ pub fn is_tracing_var(name: &str) -> bool {
 pub fn is_compute_config_var(name: &str) -> bool {
     name == MAX_RESULT_SIZE.name()
         || name == DATAFLOW_MAX_INFLIGHT_BYTES.name()
+        || name == LINEAR_JOIN_YIELDING.name()
         || name == ENABLE_MZ_JOIN_CORE.name()
         || name == ENABLE_JEMALLOC_PROFILING.name()
         || is_persist_config_var(name)