feat: make gRPC max message size consistently configurable

[andygrove]

Which issue does this PR close?

Related to #2168.

Rationale for this change

At SF1000 several TPC-H plans (Q11, Q21, Q22) encode above the 16 MiB gRPC message-size ceiling and fail with:

Status { code: OutOfRange, message: "Error, decoded message length too large:
        found ~22 MB, the limit is: 16777216 bytes" }

The obvious remedy is to raise the limit, but today that isn't reliably possible. Raising the runtime flags on the scheduler/executor pods (--grpc-server-max-*-message-size=...) only lifts the server-side receive limits, and the scheduler-to-executor path ignores the setting entirely:

• ExecutorManager::new builds a GrpcClientConfig but falls back to GrpcClientConfig::default() (16 MiB) with no way to override it short of writing a config producer in Rust.
• Worse, ExecutorManager::get_client then constructs ExecutorGrpcClient::new(connection) without ever applying max_message_size. Message-size limits are tonic codec settings rather than Endpoint settings, so create_grpc_client_endpoint can't apply them, and the value that was carefully computed is silently dropped. Every other client construction in the codebase (executor_process.rs, executor_server.rs, standalone.rs, flight_proxy_service.rs) does apply it — this one path did not.

So a user hitting the ceiling on task assignment has no supported knob to turn.

An earlier revision of this PR raised the defaults to 128 MiB. Per review feedback from @martin-g that a large default makes it harder to notice messages growing too heavy (and works against optimisations like #1853), the default stays at 16 MiB. This PR now only makes the ceiling consistently configurable, so operators who need a larger one can set it.

What changes are included in this PR?

• ballista/scheduler/src/config.rs — new --grpc-client-max-message-size flag plus the matching SchedulerConfig field and with_grpc_client_max_message_size builder, defaulting to 16 MiB (unchanged behaviour).
• ballista/scheduler/src/state/executor_manager.rs:
  • ExecutorManager::new uses that flag when no override_config_producer is wired. An explicit config producer still wins, so embedders keep full control.
  • get_client applies max_encoding_message_size / max_decoding_message_size to ExecutorGrpcClient, so the configured limit actually takes effect (per @phillipleblanc's review).
• Doc comments on the executor's --grpc-server-max-*-message-size flags now state the default in MiB consistently (@martin-g's suggestions, with the value left at 16 MiB).

No default value changes, so docs/source/user-guide/configs.md is unchanged.

New tests:

• ballista-scheduler: --grpc-client-max-message-size reaches ExecutorManager's client config, and an override_config_producer still takes precedence over it.
• ballista-core: round-trip test that ballista.client.grpc_max_message_size, set on a SessionConfig via options_mut().set(...), reaches BallistaConfig::grpc_client_max_message_size() unchanged. That path (options.set → ExtensionOptions → get_usize_setting) had no existing coverage, and it's the one Python clients rely on via their cluster_config overrides.

Are there any user-facing changes?

One new scheduler CLI flag, --grpc-client-max-message-size (default 16 MiB, i.e. what the scheduler already used). No behaviour changes for existing configurations, and no breaking API changes.

[milenkovicm]

There is also outstanding pr which helps reducing message size #1853

[phillipleblanc]

Looks good to me, one thing to check is whether we want to wire this into create_grpc_client_endpoint to pass along the max message size value.

i.e.

   SchedulerConfig.grpc_client_max_message_size
           v
   GrpcClientConfig.max_message_size
           v
   create_grpc_client_endpoint(config) // here
           v
   ExecutorGrpcClient::new(channel)

create_grpc_client_endpoint reads settings such as:

• connection timeout
• keepalive interval
• HTTP/2 window sizes

But it never reads config.max_message_size. Message-size limits are Tonic codec settings, not HTTP/2 Endpoint settings.

It would need something like:

  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);

[martin-g]

I wonder whether 128MiB is too much ?!
Having a lower value led to an optimisation like #1853.
With a big value like 128MiB it will be harder to notice that some message(s) became too heavy.

[andygrove]

Thanks both, I've reworked this based on your feedback and updated the title and description.

@martin-g fair point about losing the signal that messages are getting heavy. I've dropped the default bump entirely, so everything stays at 16 MiB and optimisations like #1853 still get noticed when a message grows too big. The PR is now only about making the ceiling configurable where it wasn't, so operators who genuinely need a bigger one can raise it. I've also taken your MiB wording suggestions, just with the value left at 16MiB.

@phillipleblanc good catch, and it turned out to be worse than "not wired". ExecutorManager was computing grpc_client_config.max_message_size and then dropping it on the floor, because get_client built the client with a plain ExecutorGrpcClient::new(connection). It now applies both max_encoding_message_size and max_decoding_message_size, with a comment explaining that these are tonic codec settings so create_grpc_client_endpoint cannot apply them. I checked every other client construction in the tree (executor_process.rs, executor_server.rs, standalone.rs, flight_proxy_service.rs, client.rs) and they all set the limits already, so this path was the only inconsistent one.

The net diff against main is now purely additive. One new scheduler flag, --grpc-client-max-message-size, defaulting to the 16 MiB the scheduler already used, plus tests that the flag reaches the client config and that an override_config_producer still takes precedence over it.