From 4eeb1f62c2e1f4e0c927f07b745b6e0e6d729873 Mon Sep 17 00:00:00 2001 From: Mihai Budiu Date: Wed, 8 Jul 2026 10:48:09 -0700 Subject: [PATCH 1/3] [manager] Set the memory limits from the local OS/container when they are not specified Signed-off-by: Mihai Budiu --- crates/adapters/src/controller.rs | 23 ++++++--- crates/adapters/src/server.rs | 15 +++++- crates/feldera-observability/Cargo.toml | 1 + crates/feldera-observability/src/lib.rs | 1 + crates/feldera-observability/src/system.rs | 39 +++++++++++++++ .../src/runner/local_runner.rs | 50 ++++++++++++++++++- docs.feldera.com/docs/operations/memory.md | 7 +++ python/tests/workloads/test_now.py | 2 +- 8 files changed, 127 insertions(+), 11 deletions(-) create mode 100644 crates/feldera-observability/src/system.rs diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index b9cd2600227..b815ff27866 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -5577,15 +5577,22 @@ 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 @@ -5593,8 +5600,8 @@ Using the Kubernetes limit as the RSS memory limit." && 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 index b6bc9ca73ca..5f33e7d2655 100644 --- a/crates/adapters/src/server.rs +++ b/crates/adapters/src/server.rs @@ -1024,7 +1024,7 @@ fn parse_config(config_file: impl AsRef) -> Result) -> Result sentry::integrations::actix::Sentry { pub mod fips; pub mod json_logging; +pub mod system; fn trace_header_value() -> Option { if !sentry_enabled() { diff --git a/crates/feldera-observability/src/system.rs b/crates/feldera-observability/src/system.rs new file mode 100644 index 00000000000..6ca57b07126 --- /dev/null +++ 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 { + 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 index b741b04fe80..949deef9951 100644 --- 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, n_hosts: usize) -> Option { + 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) { + 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 index 276d5c372e4..ea58bf92936 100644 --- 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 index c96189c6d91..3ca7d57f87f 100644 --- 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 From 5d17958e31adf2ac7b6c95bfeb85301cf9a5e04d Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Fri, 17 Jul 2026 23:34:35 -0700 Subject: [PATCH 2/3] [dbsp] Don't panic on balancer hints for unbalanced streams. `Balancer::set_hint` unwrapped the `stream_to_cluster` lookup, so a hint targeting a stream that is not an input to an adaptively balanced join panicked and killed the worker thread, crashing the pipeline. This could happen when adaptive joins were disabled altogether or the hint applied to a stream that wasn't part of an adaptive join (not sure if the compiler can ever generate such a hint). We now cleanly return an error instead of panicking. Signed-off-by: Leonid Ryzhyk --- .../dbsp/src/operator/dynamic/balance/balancer.rs | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/crates/dbsp/src/operator/dynamic/balance/balancer.rs b/crates/dbsp/src/operator/dynamic/balance/balancer.rs index e4156af223f..7bc28ff8a6b 100644 --- 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 From 8c8e8d3a083b6bb7e4304290ada961769be08ae3 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Sat, 18 Jul 2026 00:51:23 -0700 Subject: [PATCH 3/3] [sql] Skip balancer hints when adaptive joins are disabled. The compiler unconditionally generated balancing hints whether adaptive joins were enabled or not, which could cause a pipeline running without adaptive joins to fail. This commit modifies the generated code to ignore hints with a warning when adaptive joins are disabled in dev tweaks. Signed-off-by: Leonid Ryzhyk --- Cargo.lock | 1 + crates/dbsp/src/circuit/dbsp_handle.rs | 22 ++++++++++++++++++- crates/dbsp/src/circuit/runtime.rs | 8 +++++++ python/tests/workloads/test_now.py | 5 ++++- .../compiler/backend/rust/ToRustVisitor.java | 3 +-- .../backend/rust/multi/CircuitWriter.java | 3 +-- .../visitors/outer/CircuitPostfix.java | 13 +++++++++++ 7 files changed, 49 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f1e318a8ef7..7e3d27b4672 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5421,6 +5421,7 @@ dependencies = [ "reqwest 0.12.24", "sentry", "serde_json", + "sysinfo 0.38.4", "tracing", "tracing-subscriber", ] diff --git a/crates/dbsp/src/circuit/dbsp_handle.rs b/crates/dbsp/src/circuit/dbsp_handle.rs index caf51f4b869..3b5d2ef6b86 100644 --- 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 index 97243c0a868..822207ec004 100644 --- 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/python/tests/workloads/test_now.py b/python/tests/workloads/test_now.py index 3ca7d57f87f..a2a75c8e59a 100644 --- a/python/tests/workloads/test_now.py +++ b/python/tests/workloads/test_now.py @@ -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-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java index 22da616ce46..07752d34d41 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java +++ b/sql-to-dbsp-compiler/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/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/CircuitWriter.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/CircuitWriter.java index d436fecc869..31e982b0d66 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/CircuitWriter.java +++ b/sql-to-dbsp-compiler/SQL-compiler/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/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java index 4a0d7f5ee85..c70ae12be06 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java +++ b/sql-to-dbsp-compiler/SQL-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 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(); }