feldera · mihaibudiu · Jul 8, 2026 · Jul 18, 2026 · Jul 18, 2026 · snkas
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
@@ -5577,24 +5577,31 @@ impl ControllerInit {
         if max_rss_mb.is_none()
             && let Some(memory_mb_max) = &pipeline_config.global.resources.memory_mb_max
         {
-            warn!(
-                "RSS memory limit ('max_rss_mb') is not set, but a Kubernetes memory limit \
+            if feldera_observability::system::running_in_kubernetes() {
+                warn!(
+                    "RSS memory limit ('max_rss_mb') is not set, but a Kubernetes pod memory limit \
 ('resources.memory_mb_max' = {memory_mb_max} MB) is configured. \
-Using the Kubernetes limit as the RSS memory limit."
-            );
+Using the pod limit as the RSS memory limit."
+                );
+            } else {
+                info!(
+                    "RSS memory limit ('max_rss_mb') is not set; using 'resources.memory_mb_max' \
+({memory_mb_max} MB) as the RSS memory limit."
+                );
+            }
             max_rss_mb = Some(*memory_mb_max);
         } else if max_rss_mb.is_none() && pipeline_config.global.resources.memory_mb_max.is_none() {
             warn!(
-                "RSS memory limit ('max_rss_mb') is not set, and no Kubernetes memory limit \
+                "RSS memory limit ('max_rss_mb') is not set, and no deployment memory limit \
 ('resources.memory_mb_max') is configured. We recommend configuring at least one of these settings to avoid out-of-memory failures."
             );
         } else if let Some(max_rss_mb) = max_rss_mb
             && let Some(memory_mb_max) = &pipeline_config.global.resources.memory_mb_max
             && max_rss_mb > *memory_mb_max
         {
             warn!(
-                "RSS memory limit ('max_rss_mb') is set to {max_rss_mb} MB exceeds the Kubernetes memory limit \
-('resources.memory_mb_max' = {memory_mb_max} MB) is configured. This will likely cause out-of-memory failures."
+                "RSS memory limit ('max_rss_mb' = {max_rss_mb} MB) exceeds the deployment memory limit \
+('resources.memory_mb_max' = {memory_mb_max} MB). This will likely cause out-of-memory failures."
             );
         }
 
@@ -8415,7 +8422,7 @@ impl ControllerInner {
                 }
                 TransactionState::Started {
                     tid,
-                    start,
+                    start: _,
                     processed_records,
                 } if next.is_none() || next != open => {
                     // The next step belongs to a different transaction, or the

diff --git a/crates/adapters/src/server.rs b/crates/adapters/src/server.rs
@@ -1024,7 +1024,7 @@ fn parse_config(config_file: impl AsRef<Path>) -> Result<PipelineConfig, Control
     }
 
     let path = config_file.as_ref();
-    let config = if path.extension() == Some(OsStr::new("json")) {
+    let mut config = if path.extension() == Some(OsStr::new("json")) {
         parse_json_config(path)
     } else {
         let json_path = path.with_extension("json");
@@ -1035,6 +1035,19 @@ fn parse_config(config_file: impl AsRef<Path>) -> Result<PipelineConfig, Control
         }
     }?;
 
+    // This is run in the pipeline process itself, so on a multihost deployment every
+    // host derives the limit from its own machine.
+    if config.global.effective_memory_mb().is_none()
+        && let Some(available_mb) = observability::system::total_memory_megabyte()
+    {
+        // The logger is not running yet.
+        eprintln!(
+            "No memory limit configured ('max_rss_mb' or 'resources.memory_mb_max'); \
+using this host's available memory: 'max_rss_mb' = {available_mb} MB."
+        );
+        config.global.max_rss_mb = Some(available_mb);
+    }
+
     eprintln!(
         "Pipeline configuration loaded successfully: {}",
         config.display_summary()

diff --git a/crates/dbsp/src/circuit/dbsp_handle.rs b/crates/dbsp/src/circuit/dbsp_handle.rs
@@ -39,7 +39,7 @@ use std::{
     thread::Result as ThreadResult,
     time::Instant,
 };
-use tracing::{debug, info};
+use tracing::{debug, info, warn};
 use uuid::Uuid;
 
 #[cfg(doc)]
@@ -2385,6 +2385,26 @@ impl DBSPHandle {
         Ok(())
     }
 
+    /// Whether adaptive joins are enabled for this circuit
+    /// (`dev_tweaks.adaptive_joins`). When disabled, the circuit uses plain hash
+    /// joins that never register with the balancer, so balancer hints have no effect.
+    ///
+    /// Compiler-generated circuit code gates its `set_balancer_hint` calls on this;
+    /// when disabled it logs a warning that the hints are being skipped (the generated
+    /// crate cannot log directly).
+    pub fn adaptive_joins_enabled(&self) -> bool {
+        let enabled = self
+            .runtime
+            .as_ref()
+            .is_some_and(|runtime| runtime.runtime().adaptive_joins());
+        if !enabled {
+            warn!(
+                "ignoring balancer hints: adaptive joins are disabled (dev_tweaks.adaptive_joins)"
+            );
+        }
+        enabled
+    }
+
     pub fn set_balancer_hint_by_global_id(
         &mut self,
         global_node_id: &GlobalNodeId,

diff --git a/crates/dbsp/src/circuit/runtime.rs b/crates/dbsp/src/circuit/runtime.rs
@@ -1088,6 +1088,14 @@ impl Runtime {
         self.inner().allow_input_during_commit
     }
 
+    /// Whether adaptive (dynamically balanced) joins are enabled, per
+    /// `dev_tweaks.adaptive_joins`. Unlike [`Self::with_dev_tweaks`], this reads the
+    /// runtime's configured tweaks, so it is valid off the worker threads (e.g. on
+    /// the thread that owns the `DBSPHandle`).
+    pub fn adaptive_joins(&self) -> bool {
+        self.inner().dev_tweaks.adaptive_joins()
+    }
+
     /// Returns the worker index as a string.
     ///
     /// This is useful for metric labels.

diff --git a/crates/dbsp/src/operator/dynamic/balance/balancer.rs b/crates/dbsp/src/operator/dynamic/balance/balancer.rs
@@ -776,6 +776,9 @@ impl BalancerInner {
 
     /// Set a hint for a stream.
     ///
+    /// Fails with `NotRegisteredWithBalancer` if the stream is not managed by the balancer,
+    /// i.e., it is not an input to an adaptively balanced join.
+    ///
     /// Fails if the hint specifies a fixed policy that cannot be enforced in the current state, i.e.,
     /// it either contradicts the set of existing hints for other streams (e.g., we refuse to join
     /// two streams both of which have Broadcast policies) or the policy cannot be enforced during the
@@ -789,7 +792,15 @@ impl BalancerInner {
         //     Runtime::worker_index(),
         //     hint
         // );
-        let cluster_index = *self.stream_to_cluster.get(&node_id).unwrap();
+        // A stream is absent from `stream_to_cluster` when it is not an input to an
+        // adaptively balanced join (e.g. an input to a plain stream join, such as a
+        // `now()` cross join, or a join the compiler could not make adaptive). Setting
+        // a policy hint on such a stream is meaningless, so reject it with an error
+        // instead of panicking and taking down the worker thread.
+        let cluster_index = *self
+            .stream_to_cluster
+            .get(&node_id)
+            .ok_or(BalancerError::NotRegisteredWithBalancer(node_id))?;
 
         // Apply the hint to the stored hints, remembering the previous value so we
         // can roll back if the hint turns out to be unenforceable. The hint must be

diff --git a/crates/feldera-observability/Cargo.toml b/crates/feldera-observability/Cargo.toml
@@ -20,3 +20,4 @@ serde_json = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["env-filter", "json"] }
 colored = { workspace = true }
+sysinfo = { workspace = true }
diff --git a/crates/feldera-observability/src/lib.rs b/crates/feldera-observability/src/lib.rs
@@ -69,6 +69,7 @@ pub fn actix_middleware() -> sentry::integrations::actix::Sentry {
 
 pub mod fips;
 pub mod json_logging;
+pub mod system;
 
 fn trace_header_value() -> Option<String> {
     if !sentry_enabled() {

diff --git a/crates/feldera-observability/src/system.rs b/crates/feldera-observability/src/system.rs
@@ -0,0 +1,39 @@
+//! System introspection utilities.
+
+/// True when this process runs inside a Kubernetes pod.
+///
+/// Kubernetes injects `KUBERNETES_SERVICE_HOST` into every pod, and the
+/// pipeline manager rejects user-supplied `KUBERNETES_*` variables in the
+/// pipeline's `env` config.
+pub fn running_in_kubernetes() -> bool {
+    std::env::var_os("KUBERNETES_SERVICE_HOST").is_some()
+}
+
+/// Returns the memory, in megabytes (MB), of the current system, or `None`
+/// when the operating system does not report it.
+///
+/// This is the min of the host's available memory and the cgroup budget
+/// (what a container or Kubernetes pod memory limit sets).
+pub fn total_memory_megabyte() -> Option<u64> {
+    let mut system = sysinfo::System::new();
+    system.refresh_memory();
+    let mut available = system.total_memory();
+    if let Some(limits) = system.cgroup_limits() {
+        available = available.min(limits.total_memory);
+    }
+    let mb = available / 1_000_000;
+    // `sysinfo` reports 0 on platforms it does not support.
+    if mb > 0 { Some(mb) } else { None }
+}
+
+#[cfg(test)]
+mod tests {
+    /// Guards the `sysinfo` call sequence (`cgroup_limits` panics when memory
+    /// was not refreshed first).
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn reports_nonzero_memory() {
+        let mb = super::total_memory_megabyte().expect("Linux reports available memory");
+        assert!(mb > 0, "available memory must be positive, got {mb} MB");
+    }
+}
diff --git a/crates/pipeline-manager/src/runner/local_runner.rs b/crates/pipeline-manager/src/runner/local_runner.rs
@@ -13,9 +13,10 @@ use crate::runner::error::RunnerError;
 use crate::runner::pipeline_executor::{PipelineExecutor, ProvisionStatus};
 use crate::runner::pipeline_logs::{LogMessage, LogsSender};
 use async_trait::async_trait;
+use feldera_observability::system::total_memory_megabyte;
 use feldera_observability::ReqwestTracingExt;
 use feldera_types::config::{
-    PipelineConfig, PipelineConfigProgramInfo, StorageCacheConfig, StorageConfig,
+    PipelineConfig, PipelineConfigProgramInfo, RuntimeConfig, StorageCacheConfig, StorageConfig,
 };
 use feldera_types::runtime_status::{BootstrapConfig, RuntimeDesiredStatus};
 use reqwest::StatusCode;
@@ -791,6 +792,13 @@ impl LocalRunner {
         base_config.outputs = program_info.outputs;
         base_config.program_ir = program_info.program_ir;
 
+        // The memory budget is per host, and all host processes share this
+        // machine, so each host gets an equal share of the available memory.
+        apply_default_memory_limit(
+            &mut base_config.global,
+            per_host_share(total_memory_megabyte(), n_hosts),
+        );
+
         // Retrieve the pipeline binary once; every host process runs the same
         // executable.
         let binary_file_path = self
@@ -1105,6 +1113,10 @@ impl PipelineExecutor for LocalRunner {
         deployment_config.outputs = program_info.outputs;
         deployment_config.program_ir = program_info.program_ir;
 
+        // Unlike Kubernetes, a local deployment has no pod memory limit for
+        // the pipeline to fall back on, so derive one from this machine.
+        apply_default_memory_limit(&mut deployment_config.global, total_memory_megabyte());
+
         // Write config as YAML and JSON
         //
         // Newer pipelines will read the JSON, older ones will read the YAML.
@@ -1390,6 +1402,42 @@ impl PipelineExecutor for LocalRunner {
     }
 }
 
+/// Splits `available_mb` evenly among `n_hosts` colocated host processes;
+/// `None` when the share rounds down to nothing.
+fn per_host_share(available_mb: Option<u64>, n_hosts: usize) -> Option<u64> {
+    available_mb
+        .map(|mb| mb / n_hosts.max(1) as u64)
+        .filter(|mb| *mb > 0)
+}
+
+/// Sets `global.resources.memory_mb_max` to `available_mb` when the pipeline
+/// has no memory budget.
+fn apply_default_memory_limit(global: &mut RuntimeConfig, available_mb: Option<u64>) {
+    if global.effective_memory_mb().is_none() {
+        if let Some(available_mb) = available_mb {
+            info!(
+                "pipeline has no memory limit ('max_rss_mb' or 'resources.memory_mb_max'): \
+defaulting 'resources.memory_mb_max' to {available_mb} MB"
+            );
+            global.resources.memory_mb_max = Some(available_mb);
+        }
+    }
+}
+
+#[cfg(test)]
+mod memory_limit_tests {
+    use super::per_host_share;
+
+    /// Colocated hosts split the machine evenly; a share that rounds down to
+    /// nothing must yield `None`, not a 0 MB limit.
+    #[test]
+    fn hosts_split_available_memory() {
+        assert_eq!(per_host_share(Some(4000), 4), Some(1000));
+        assert_eq!(per_host_share(Some(3), 4), None);
+        assert_eq!(per_host_share(None, 4), None);
+    }
+}
+
 #[cfg(test)]
 mod multihost_tests {
     use super::{

diff --git a/docs.feldera.com/docs/operations/memory.md b/docs.feldera.com/docs/operations/memory.md
@@ -13,6 +13,13 @@ If `max_rss_mb` is not set but `memory_mb_max` is configured in the `resources`
 [Runtime configuration], the latter is used as the effective memory cap. We **strongly recommend**
 setting at least one of these parameters to prevent out-of-memory failures.
 
+On single-host deployments (for example, in Docker), if neither
+parameter is set, `resources.memory_mb_max` is set to the smaller
+value of host memory size, and container memory limit.  In a
+multi-host deployment each pipeline process that finds neither
+parameter set derives `max_rss_mb` from its own host's available
+memory at startup.
+
 When either `max_rss_mb` or `resources.memory_mb_max` is configured, the pipeline reports its current
 memory pressure level (`low`, `moderate`, `high`, or `critical`) via the
 [`memory_pressure` metric](/operations/metrics). High or critical pressure indicates that memory usage

diff --git a/python/tests/workloads/test_now.py b/python/tests/workloads/test_now.py
@@ -88,7 +88,7 @@ def test_now(self):
             ViewSpec(
                 "v",
                 """
-                select
+                select /*+ BROADCAST(v_now) */
                     t.*,
                     v_now.now_ts as now_ts
                 from t, v_now
@@ -101,7 +101,10 @@ def test_now(self):
             tables,
             views,
             # This test uses >2GB of storage in the ad hoc query.
-            resources=Resources(storage_mb_max=8192),
+            resources=Resources(
+                storage_mb_max=8192,
+                memory_mb_max=1024,
+            ),
         )
 
         pipeline.start()

diff --git a/.../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java b/.../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java
@@ -303,8 +303,7 @@ public static void registerPreAndPostprocessors(DBSPCompiler compiler, IIndentSt
     }
 
     void emitBalancerHints() {
-        for (var hint: this.postfix.balancerHints)
-            hint.emit(this.builder);
+        CircuitPostfix.emitBalancerHints(this.builder, this.postfix.balancerHints);
     }
 
     @Override

diff --git a/...ompiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/CircuitWriter.java b/...ompiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/CircuitWriter.java
@@ -220,8 +220,7 @@ void writeCircuit(DBSPCompiler compiler, DBSPCircuit circuit) {
     }
 
     void emitBalancerHints() {
-        for (var hint: this.materializations.balancerHints)
-            hint.emit(this.builder());
+        CircuitPostfix.emitBalancerHints(this.builder(), this.materializations.balancerHints);
     }
 
     @Override

diff --git a/...L-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java b/...L-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java
@@ -81,6 +81,19 @@ public void recordHint(DBSPBinaryOperator operator, JoinStrategy strategy) {
         this.balancerHints.add(hint);
     }
 
+    /** Emit the recorded balancer hints, gated on adaptive joins being enabled.
+     * When adaptive joins are disabled (`dev_tweaks.adaptive_joins`), the streams a
+     * hint references are never registered with the balancer; `adaptive_joins_enabled`
+     * then returns false (logging a warning) and the hints are skipped. */
+    public static void emitBalancerHints(IIndentStream builder, List<BalancerHint> hints) {
+        if (hints.isEmpty())
+            return;
+        builder.append("if circuit.adaptive_joins_enabled() {").increase().newline();
+        for (BalancerHint hint: hints)
+            hint.emit(builder);
+        builder.decrease().append("}").newline();
+    }
+
     public void clearRegions() {
         this.regionsCreated.clear();
     }