From 67608e433ef36a3a981b8a7d75dc152846c9b150 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Fri, 24 Jul 2026 10:41:33 -0600 Subject: [PATCH 1/4] feat: bump default gRPC max message size 16 MiB -> 128 MiB MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit At SF1000, some TPC-H plans (Q11, Q21, Q22) encode above the 16 MiB default gRPC message-size ceiling, so submitting them fails with: Status { code: OutOfRange, message: "Error, decoded message length too large: found ~22 MB, the limit is: 16777216 bytes" } Raising the runtime flags on scheduler and executor pods only lifts the server-side receive limits. Several paths still use `GrpcClientConfig::default()` (16 MiB) unconditionally — most notably the scheduler-to-executor client in `ExecutorManager::new` — so task-assignment RPCs with large plans reject even when the servers were raised. Bump every default that fed a 16 MiB ceiling to 128 MiB: - ballista/core/src/config.rs — `BALLISTA_CLIENT_GRPC_MAX_MESSAGE_SIZE` default (SessionConfig-driven client path). - ballista/core/src/utils.rs — `GrpcClientConfig::default()` `max_message_size` (used by scheduler-to-executor, flight proxy, executor client pool default lookups). - ballista/scheduler/src/config.rs and ballista/executor/src/config.rs — both CLI flags (`--grpc-server-max-{decoding,encoding}-message-size`) and the matching struct defaults. - ballista/executor/src/executor_process.rs — matching ExecutorProcessConfig defaults. The value is a ceiling, not a preallocation, so bumping it has no runtime cost for messages below the old limit. Adds a round-trip test confirming that `ballista.client.grpc_max_message_size` set on a SessionConfig reaches `BallistaConfig::grpc_client_max_message_size()` unchanged, since that path (options.set -> ExtensionOptions -> get_usize_setting) had no existing coverage. Updates the `default_config` unit test to reflect the new default. Signed-off-by: Andy Grove --- ballista/core/src/config.rs | 31 +++++++++++++++++++++-- ballista/core/src/utils.rs | 2 +- ballista/executor/src/config.rs | 8 +++--- ballista/executor/src/executor_process.rs | 4 +-- ballista/scheduler/src/config.rs | 8 +++--- 5 files changed, 40 insertions(+), 13 deletions(-) diff --git a/ballista/core/src/config.rs b/ballista/core/src/config.rs index fd3060297d..84ec3a8bbe 100644 --- a/ballista/core/src/config.rs +++ b/ballista/core/src/config.rs @@ -212,7 +212,7 @@ static CONFIG_ENTRIES: LazyLock> = LazyLock::new(|| ConfigEntry::new(BALLISTA_CLIENT_GRPC_MAX_MESSAGE_SIZE.to_string(), "Configuration for max message size in gRPC clients".to_string(), DataType::UInt64, - Some((16 * 1024 * 1024).to_string())), + Some((128 * 1024 * 1024).to_string())), ConfigEntry::new(BALLISTA_CLIENT_GRPC_CONNECT_TIMEOUT_SECONDS.to_string(), "Connection timeout for gRPC client in seconds".to_string(), DataType::UInt64, @@ -934,10 +934,37 @@ mod tests { #[test] fn default_config() -> Result<()> { let config = BallistaConfig::default(); - assert_eq!(16777216, config.grpc_client_max_message_size()); + assert_eq!(134217728, config.grpc_client_max_message_size()); Ok(()) } + #[test] + fn cluster_config_plumbs_grpc_max_message_size() { + // Sanity check that setting `ballista.client.grpc_max_message_size` + // via `SessionConfig::options_mut().set(...)` — the path Python + // clients' `cluster_config` overrides use — actually reaches the + // reader consulted by `distributed_query`. + use crate::extension::SessionConfigExt; + use datafusion::prelude::SessionConfig; + + let mut config = SessionConfig::new_with_ballista(); + assert_eq!( + config.ballista_config().grpc_client_max_message_size(), + 128 * 1024 * 1024, + "unexpected default (did the bump land?)", + ); + + let r = config + .options_mut() + .set("ballista.client.grpc_max_message_size", "67108864"); + assert!(r.is_ok(), "set failed: {r:?}"); + + assert_eq!( + config.ballista_config().grpc_client_max_message_size(), + 64 * 1024 * 1024, + ); + } + // The default must stay non-zero: `0` disables the fit check, and since AQE // no longer consults `prefer_hash_join`, a disabled check would leave the // Partitioned arm unconditionally on hash join and make SortMergeJoin — the diff --git a/ballista/core/src/utils.rs b/ballista/core/src/utils.rs index 34b5b05f10..6ce842b82c 100644 --- a/ballista/core/src/utils.rs +++ b/ballista/core/src/utils.rs @@ -107,7 +107,7 @@ impl Default for GrpcClientConfig { tcp_keepalive_seconds: 3600, http2_keepalive_interval_seconds: 300, use_tls: false, - max_message_size: 16 * 1024 * 1024, + max_message_size: 128 * 1024 * 1024, io_retries_times: 3, io_retry_wait_time_ms: 3000, initial_connection_window_size: 67108864, diff --git a/ballista/executor/src/config.rs b/ballista/executor/src/config.rs index 2b0ffe608d..1071b345d7 100644 --- a/ballista/executor/src/config.rs +++ b/ballista/executor/src/config.rs @@ -169,17 +169,17 @@ pub struct Config { help = "Tracing log rotation policy." )] pub log_rotation_policy: ballista_core::config::LogRotationPolicy, - /// Maximum size of incoming gRPC messages in bytes (default: 16MB). + /// Maximum size of incoming gRPC messages in bytes (default: 128MB). #[arg( long, - default_value_t = 16777216, + default_value_t = 134217728, help = "The maximum size of a decoded message at the grpc server side." )] pub grpc_server_max_decoding_message_size: u32, - /// Maximum size of outgoing gRPC messages in bytes (default: 16MB). + /// Maximum size of outgoing gRPC messages in bytes (default: 128MB). #[arg( long, - default_value_t = 16777216, + default_value_t = 134217728, help = "The maximum size of an encoded message at the grpc server side." )] pub grpc_server_max_encoding_message_size: u32, diff --git a/ballista/executor/src/executor_process.rs b/ballista/executor/src/executor_process.rs index 091a94c5f9..e70397ecb1 100644 --- a/ballista/executor/src/executor_process.rs +++ b/ballista/executor/src/executor_process.rs @@ -375,8 +375,8 @@ impl Default for ExecutorProcessConfig { log_rotation_policy: Default::default(), job_data_ttl_seconds: 604800, job_data_clean_up_interval_seconds: 0, - grpc_max_decoding_message_size: 16777216, - grpc_max_encoding_message_size: 16777216, + grpc_max_decoding_message_size: 134217728, + grpc_max_encoding_message_size: 134217728, grpc_server_config: Default::default(), executor_heartbeat_interval_seconds: 60, metric_collection_policy: ExecutorMetricCollectionPolicy::default(), diff --git a/ballista/scheduler/src/config.rs b/ballista/scheduler/src/config.rs index 940888714b..2e7b81ea5f 100644 --- a/ballista/scheduler/src/config.rs +++ b/ballista/scheduler/src/config.rs @@ -148,14 +148,14 @@ pub struct Config { /// Maximum size of decoded gRPC messages. #[arg( long, - default_value_t = 16777216, + default_value_t = 134217728, help = "The maximum size of a decoded message at the grpc server side." )] pub grpc_server_max_decoding_message_size: u32, /// Maximum size of encoded gRPC messages. #[arg( long, - default_value_t = 16777216, + default_value_t = 134217728, help = "The maximum size of an encoded message at the grpc server side." )] pub grpc_server_max_encoding_message_size: u32, @@ -305,8 +305,8 @@ impl Default for SchedulerConfig { job_resubmit_interval_ms: None, executor_termination_grace_period: 0, scheduler_event_expected_processing_duration: 0, - grpc_server_max_decoding_message_size: 16777216, - grpc_server_max_encoding_message_size: 16777216, + grpc_server_max_decoding_message_size: 134217728, + grpc_server_max_encoding_message_size: 134217728, executor_timeout_seconds: 180, expire_dead_executor_interval_seconds: 15, override_config_producer: None, From e2827d5f90d5dc1841ddb5e6b1fdeee9ac49b336 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Fri, 24 Jul 2026 10:55:17 -0600 Subject: [PATCH 2/4] docs: regenerate configs.md for the bumped default Signed-off-by: Andy Grove --- docs/source/user-guide/configs.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index c40f1a582c..8622b50b7a 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -109,7 +109,7 @@ standard DataFusion settings. | ballista.cache.noop | Boolean | true | Disable default cache node extension | | ballista.client.grpc_connect_timeout_seconds | UInt64 | 20 | Connection timeout for gRPC client in seconds | | ballista.client.grpc_http2_keepalive_interval_seconds | UInt64 | 300 | HTTP/2 keep-alive interval for gRPC client in seconds | -| ballista.client.grpc_max_message_size | UInt64 | 16777216 | Configuration for max message size in gRPC clients | +| ballista.client.grpc_max_message_size | UInt64 | 134217728 | Configuration for max message size in gRPC clients | | ballista.client.grpc_tcp_keepalive_seconds | UInt64 | 3600 | TCP keep-alive interval for gRPC client in seconds | | ballista.client.grpc_timeout_seconds | UInt64 | 20 | Request timeout for gRPC client in seconds | | ballista.client.initial_connection_window_size | UInt64 | 67108864 | HTTP/2 initial connection-level flow-control window for gRPC data-plane clients, in bytes. Should be >= the shuffle governor byte budget so the governor, not the transport window, is the binding backpressure. 0 leaves the tonic default. | From 26d6238398248006215d8e22f3b22a95644015b6 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Fri, 24 Jul 2026 11:24:36 -0600 Subject: [PATCH 3/4] feat(scheduler): --grpc-client-max-message-size configurable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bumping GrpcClientConfig::default().max_message_size to 128 MiB helps most cases but leaves no runtime knob for the scheduler's outbound gRPC client to executors — the client the task-assignment RPC actually flows through. Its config was only overridable via a Rust-side `override_config_producer`, which end users of the shipped binary can't set. Add `--grpc-client-max-message-size` on the scheduler CLI (default 134217728 == the new default), and use it to override `GrpcClientConfig::max_message_size` in `ExecutorManager::new` when no `override_config_producer` is wired. The producer path still takes precedence so embedders keep full control. Users can now scale beyond 128 MiB without recompiling. Signed-off-by: Andy Grove --- ballista/scheduler/src/config.rs | 20 +++++++++++++++++++ .../scheduler/src/state/executor_manager.rs | 11 +++++++++- 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/ballista/scheduler/src/config.rs b/ballista/scheduler/src/config.rs index 2e7b81ea5f..459835427a 100644 --- a/ballista/scheduler/src/config.rs +++ b/ballista/scheduler/src/config.rs @@ -159,6 +159,16 @@ pub struct Config { help = "The maximum size of an encoded message at the grpc server side." )] pub grpc_server_max_encoding_message_size: u32, + /// Maximum size of messages sent by the scheduler's outbound gRPC clients + /// (e.g. task assignment to executors). Should be at least as large as the + /// executor's `--grpc-server-max-decoding-message-size` for those RPCs to + /// succeed with big encoded plans. + #[arg( + long, + default_value_t = 134217728, + help = "The maximum size of a message sent by the scheduler's outbound gRPC clients (in bytes)." + )] + pub grpc_client_max_message_size: u32, /// Timeout in seconds before marking an executor as dead. #[arg( long, @@ -256,6 +266,8 @@ pub struct SchedulerConfig { pub grpc_server_max_decoding_message_size: u32, /// The maximum size of an encoded message at the grpc server side. pub grpc_server_max_encoding_message_size: u32, + /// The maximum size of a message sent by the scheduler's outbound gRPC clients. + pub grpc_client_max_message_size: u32, /// The executor timeout in seconds. It should be longer than executor's heartbeat intervals. pub executor_timeout_seconds: u64, /// The interval to check expired or dead executors @@ -307,6 +319,7 @@ impl Default for SchedulerConfig { scheduler_event_expected_processing_duration: 0, grpc_server_max_decoding_message_size: 134217728, grpc_server_max_encoding_message_size: 134217728, + grpc_client_max_message_size: 134217728, executor_timeout_seconds: 180, expire_dead_executor_interval_seconds: 15, override_config_producer: None, @@ -426,6 +439,12 @@ impl SchedulerConfig { self } + /// Sets the maximum message size for the scheduler's outbound gRPC clients. + pub fn with_grpc_client_max_message_size(mut self, value: u32) -> Self { + self.grpc_client_max_message_size = value; + self + } + /// Sets a custom config producer. pub fn with_override_config_producer( mut self, @@ -539,6 +558,7 @@ impl TryFrom for SchedulerConfig { .grpc_server_max_decoding_message_size, grpc_server_max_encoding_message_size: opt .grpc_server_max_encoding_message_size, + grpc_client_max_message_size: opt.grpc_client_max_message_size, executor_timeout_seconds: opt.executor_timeout_seconds, expire_dead_executor_interval_seconds: opt .expire_dead_executor_interval_seconds, diff --git a/ballista/scheduler/src/state/executor_manager.rs b/ballista/scheduler/src/state/executor_manager.rs index d8a8331778..910350cd40 100644 --- a/ballista/scheduler/src/state/executor_manager.rs +++ b/ballista/scheduler/src/state/executor_manager.rs @@ -79,13 +79,22 @@ impl ExecutorManager { cluster_state: Arc, config: Arc, ) -> Self { + // Prefer an explicit override_config_producer if the embedder wired one, + // so a full BallistaConfig (with all its grpc-client knobs) still takes + // precedence. Otherwise, use `default()` but override + // `max_message_size` from the scheduler's `grpc_client_max_message_size` + // CLI flag so users can raise the ceiling for outbound task-assignment + // RPCs without having to write a config-producer in Rust. let grpc_client_config = if let Some(config_producer) = &config.override_config_producer { let session_config = config_producer(); let ballista_config = session_config.ballista_config(); GrpcClientConfig::from(&ballista_config) } else { - GrpcClientConfig::default() + GrpcClientConfig { + max_message_size: config.grpc_client_max_message_size as usize, + ..GrpcClientConfig::default() + } }; Self { cluster_state, From 341abea3166318c19126b01ac0a44ced75949641 Mon Sep 17 00:00:00 2001 From: Martin Tzvetanov Grigorov Date: Mon, 27 Jul 2026 11:08:48 +0300 Subject: [PATCH 4/4] Add AI config files --- .cursor/rules.md | 5 +++++ .gemini/rules.md | 5 +++++ AGENTS.md | 5 +++++ CLAUDE.md | 5 +++++ 4 files changed, 20 insertions(+) create mode 100644 .cursor/rules.md create mode 100644 .gemini/rules.md create mode 100644 AGENTS.md create mode 100644 CLAUDE.md diff --git a/.cursor/rules.md b/.cursor/rules.md new file mode 100644 index 0000000000..4ee2be4ad3 --- /dev/null +++ b/.cursor/rules.md @@ -0,0 +1,5 @@ +When making a code review never link to any issue or a pull request +in any GitHub repository! + +Do not review AI agents' config files like CLAUDE.md, AGENTS.md or rules.md! + diff --git a/.gemini/rules.md b/.gemini/rules.md new file mode 100644 index 0000000000..4ee2be4ad3 --- /dev/null +++ b/.gemini/rules.md @@ -0,0 +1,5 @@ +When making a code review never link to any issue or a pull request +in any GitHub repository! + +Do not review AI agents' config files like CLAUDE.md, AGENTS.md or rules.md! + diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000000..4ee2be4ad3 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,5 @@ +When making a code review never link to any issue or a pull request +in any GitHub repository! + +Do not review AI agents' config files like CLAUDE.md, AGENTS.md or rules.md! + diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000000..4ee2be4ad3 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,5 @@ +When making a code review never link to any issue or a pull request +in any GitHub repository! + +Do not review AI agents' config files like CLAUDE.md, AGENTS.md or rules.md! +