From 67608e433ef36a3a981b8a7d75dc152846c9b150 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Fri, 24 Jul 2026 10:41:33 -0600 Subject: [PATCH 1/6] 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/6] 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/6] 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 f4a6bcc2d7d87b9e631f08007c66e091b4a2aa6b Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 27 Jul 2026 10:52:39 -0600 Subject: [PATCH 4/6] Update ballista/executor/src/config.rs Co-authored-by: Martin Grigorov --- ballista/executor/src/config.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ballista/executor/src/config.rs b/ballista/executor/src/config.rs index 1071b345d7..6ba9910cb6 100644 --- a/ballista/executor/src/config.rs +++ b/ballista/executor/src/config.rs @@ -176,7 +176,7 @@ pub struct Config { 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: 128MB). + /// Maximum size of outgoing gRPC messages in bytes (default: 128MiB). #[arg( long, default_value_t = 134217728, From 70da7db60da4fd9f3fb80fa1a250778a48d2ae2b Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 27 Jul 2026 10:52:56 -0600 Subject: [PATCH 5/6] Update ballista/executor/src/config.rs Co-authored-by: Martin Grigorov --- ballista/executor/src/config.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ballista/executor/src/config.rs b/ballista/executor/src/config.rs index 6ba9910cb6..013ee31139 100644 --- a/ballista/executor/src/config.rs +++ b/ballista/executor/src/config.rs @@ -169,7 +169,7 @@ 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: 128MB). + /// Maximum size of incoming gRPC messages in bytes (default: 128MiB). #[arg( long, default_value_t = 134217728, From fa18302e7c3c310346f60c06ffe92657257177dd Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Thu, 30 Jul 2026 09:40:49 -0600 Subject: [PATCH 6/6] refactor: keep 16 MiB default, make gRPC max message size consistently configurable Revert the default bump and instead close the gaps that made the ceiling unconfigurable: the scheduler's outbound gRPC clients now honour a `--grpc-client-max-message-size` flag, and `ExecutorManager` applies that limit to the tonic codec so it is no longer silently dropped. --- ballista/core/src/config.rs | 7 ++- ballista/core/src/utils.rs | 2 +- ballista/executor/src/config.rs | 8 +-- ballista/executor/src/executor_process.rs | 4 +- ballista/scheduler/src/config.rs | 12 ++--- .../scheduler/src/state/executor_manager.rs | 54 ++++++++++++++++++- docs/source/user-guide/configs.md | 2 +- 7 files changed, 70 insertions(+), 19 deletions(-) diff --git a/ballista/core/src/config.rs b/ballista/core/src/config.rs index 84ec3a8bbe..58e5901639 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((128 * 1024 * 1024).to_string())), + Some((16 * 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,7 +934,7 @@ mod tests { #[test] fn default_config() -> Result<()> { let config = BallistaConfig::default(); - assert_eq!(134217728, config.grpc_client_max_message_size()); + assert_eq!(16777216, config.grpc_client_max_message_size()); Ok(()) } @@ -950,8 +950,7 @@ mod tests { 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?)", + 16 * 1024 * 1024, ); let r = config diff --git a/ballista/core/src/utils.rs b/ballista/core/src/utils.rs index 6ce842b82c..34b5b05f10 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: 128 * 1024 * 1024, + max_message_size: 16 * 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 013ee31139..9dbd4c78cd 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: 128MiB). + /// Maximum size of incoming gRPC messages in bytes (default: 16MiB). #[arg( long, - default_value_t = 134217728, + default_value_t = 16777216, 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: 128MiB). + /// Maximum size of outgoing gRPC messages in bytes (default: 16MiB). #[arg( long, - default_value_t = 134217728, + default_value_t = 16777216, 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 e70397ecb1..091a94c5f9 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: 134217728, - grpc_max_encoding_message_size: 134217728, + grpc_max_decoding_message_size: 16777216, + grpc_max_encoding_message_size: 16777216, 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 459835427a..4bdea02830 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 = 134217728, + default_value_t = 16777216, 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 = 134217728, + default_value_t = 16777216, help = "The maximum size of an encoded message at the grpc server side." )] pub grpc_server_max_encoding_message_size: u32, @@ -165,7 +165,7 @@ pub struct Config { /// succeed with big encoded plans. #[arg( long, - default_value_t = 134217728, + default_value_t = 16777216, help = "The maximum size of a message sent by the scheduler's outbound gRPC clients (in bytes)." )] pub grpc_client_max_message_size: u32, @@ -317,9 +317,9 @@ 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: 134217728, - grpc_server_max_encoding_message_size: 134217728, - grpc_client_max_message_size: 134217728, + grpc_server_max_decoding_message_size: 16777216, + grpc_server_max_encoding_message_size: 16777216, + grpc_client_max_message_size: 16777216, executor_timeout_seconds: 180, expire_dead_executor_interval_seconds: 15, override_config_producer: None, diff --git a/ballista/scheduler/src/state/executor_manager.rs b/ballista/scheduler/src/state/executor_manager.rs index 910350cd40..2f0b162b03 100644 --- a/ballista/scheduler/src/state/executor_manager.rs +++ b/ballista/scheduler/src/state/executor_manager.rs @@ -561,7 +561,13 @@ impl ExecutorManager { } let connection = endpoint.connect().await?; - let client = ExecutorGrpcClient::new(connection); + // Message-size limits are tonic codec settings, not `Endpoint` + // settings, so `create_grpc_client_endpoint` cannot apply them. + // Without this the configured `max_message_size` is silently + // ignored and task assignment falls back to tonic's own defaults. + let client = ExecutorGrpcClient::new(connection) + .max_encoding_message_size(grpc_client_config.max_message_size) + .max_decoding_message_size(grpc_client_config.max_message_size); { self.clients.insert(executor_id.to_owned(), client.clone()); @@ -590,3 +596,49 @@ impl ExecutorManager { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::test_utils::test_cluster_context; + use ballista_core::extension::SessionConfigExt; + use datafusion::prelude::SessionConfig; + + #[test] + fn grpc_client_max_message_size_flag_reaches_client_config() { + let config = Arc::new( + SchedulerConfig::default() + .with_grpc_client_max_message_size(64 * 1024 * 1024), + ); + let manager = + ExecutorManager::new(test_cluster_context().cluster_state(), config); + + assert_eq!( + manager.grpc_client_config.max_message_size, + 64 * 1024 * 1024 + ); + } + + #[test] + fn config_producer_still_wins_over_the_flag() { + let config = Arc::new( + SchedulerConfig::default() + .with_grpc_client_max_message_size(64 * 1024 * 1024) + .with_override_config_producer(Arc::new(|| { + let mut session_config = SessionConfig::new_with_ballista(); + session_config + .options_mut() + .set("ballista.client.grpc_max_message_size", "33554432") + .expect("valid setting"); + session_config + })), + ); + let manager = + ExecutorManager::new(test_cluster_context().cluster_state(), config); + + assert_eq!( + manager.grpc_client_config.max_message_size, + 32 * 1024 * 1024 + ); + } +} diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 8622b50b7a..c40f1a582c 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 | 134217728 | Configuration for max message size in gRPC clients | +| ballista.client.grpc_max_message_size | UInt64 | 16777216 | 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. |