andygrove opened a new issue, #2172:
URL: https://github.com/apache/datafusion-ballista/issues/2172
**Is your feature request related to a problem or challenge? Please describe
what you are trying to do.**
Running TPC-H (e.g. Q21/Q22) on a distributed cluster can fail with:
```
DataFusion error: Execution error: Status { code: OutOfRange,
message: "Error, encoded message length too large: found 19282335 bytes,
the limit is: 16777216 bytes" }
```
This is a shuffle-time failure: a `ShuffleReaderExec` fetches an
intermediate partition over Arrow Flight, and the producing executor's Flight
server tries to encode a single Arrow IPC batch larger than the 16 MB gRPC
frame limit, so the encode fails and the task (and job) dies.
Today the only workaround is to hand-raise **three** separate 16 MB limits,
across every process, to the same larger value:
- `ballista.client.grpc_max_message_size` (client session config, no CLI
flag) — governs the executor's shuffle-read *decode* limit, propagated via
`GrpcClientConfig`
- scheduler `--grpc-server-max-{en,de}coding-message-size`
- executor `--grpc-server-max-{en,de}coding-message-size` (this one also
drives the executor's Arrow Flight server encode limit)
Raising only a subset just moves the failure (e.g. raising the executor
encode limit lets the sender serialize the batch, but the receiving executor
then rejects it on decode because the client session config is still 16 MB).
This is confusing to diagnose and easy to get wrong. See #2165.
**Describe the solution you'd like**
Bound the encoded IPC message size on the shuffle *write*/serve path so a
single Flight message can never exceed the transport frame limit. Rather than
relying on the batch happening to fit under a user-tuned cap, the shuffle
producer should split oversized record batches into slices whose encoded size
stays under the configured limit, so shuffle transfer works regardless of
intermediate batch width/size and without any per-deployment tuning.
**Describe alternatives you've considered**
- Documenting the three limits and requiring users to raise all of them
together. Works, but fragile and a poor default experience (this issue is a
symptom of that).
- Raising the default limits. Only pushes the cliff further out; a large
enough intermediate batch still hits it.
- Deriving/propagating a single effective max-message-size so the three
knobs can't drift apart. Improves the workaround but still caps total shuffle
message size rather than removing the constraint.
**Additional context**
Relevant wiring:
- Executor Flight server encode limit:
`ballista/executor/src/executor_process.rs` (`flight_server_task`,
`.max_encoding_message_size(...)`)
- Executor shuffle-read client decode limit:
`ballista/core/src/execution_plans/shuffle_reader.rs` /
`ballista/executor/src/client_pool.rs` (`config.max_message_size`, from
`ballista.client.grpc_max_message_size`)
- Client/session limit definition: `ballista/core/src/config.rs`
(`BALLISTA_CLIENT_GRPC_MAX_MESSAGE_SIZE`)
Reported failure: #2165.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
