andygrove opened a new issue, #5527:
URL: https://github.com/apache/datafusion-comet/issues/5527
Following up on PR #5513 (part of #5352). I was reading through the new
byte-admission code and found that the two new configs cannot both hold at
their shipped defaults, and that the consequence for a large row is a failed
job rather than a slow one.
`spark.comet.shuffle.rss.maxFrameBytes` defaults to 64 MiB and
`spark.comet.shuffle.rss.maxInFlightBytes` defaults to 256 MiB. In
`RssPartitionWriter::push_batch_within_limit`, `admitted_frame_limit` is
computed as `(reservation_limit - ipc_scratch) / 3`, and `ipc_scratch` already
charges `original_size * 4`. So admitting a frame of `S` bytes needs roughly
`7S` of the executor-wide budget, which means the largest admissible row at the
defaults is about 36 MiB. The configured 64 MiB maximum is not reachable.
I reproduced the cutover with a small test driving the real Rust writer, one
`Utf8` column, one row per batch, `maxFrameBytes` set to 64 MiB and
`maxInFlightBytes` set to 256 MiB:
```
row= 30 MiB -> PUSHED
row= 34 MiB -> PUSHED
row= 36 MiB -> PUSHED
row= 38 MiB -> FAILED: Remote shuffle frame exceeds its configured maximum:
a single row exceeds 36346226 bytes
row= 50 MiB -> FAILED: Remote shuffle frame exceeds its configured maximum:
a single row exceeds 19569010 bytes
row= 63 MiB -> FAILED: Remote shuffle frame exceeds its configured maximum:
a single row exceeds 1393693 bytes
```
The failure itself worries me more than the accounting. `push_split_batch`
halves rows and bottoms out at a single row, so a row above the effective cap
can never be pushed at all. Spark supports strings and arrays up to 2 GiB, and
both local Comet shuffle and stock Celeborn shuffle handle a 70 MiB row without
complaint, so enabling native Celeborn shuffle converts a query that works
today into one that fails.
It also fails after doing damage. With a batch holding one small row and one
70 MiB row, the small row's frame was already pushed to Celeborn before the
error:
```
result=Err("Remote shuffle frame for a single row and encoding workspace
exceed the byte
admission budget of 268435456 bytes") frames_pushed=1 bytes=540
```
Because `CelebornShufflePusherFactory.rejectRetriedAttempt` throws
`FetchFailedException` for every attempt with `attemptNumber > 0`, each failure
costs a full map-stage rerun, and the rerun produces the same batch. That
terminates at `spark.stage.maxConsecutiveAttempts`, so the user sees four full
stage reruns and then a failed job.
A few questions on how to resolve this. Should the two defaults be made
consistent, either by raising the default `maxInFlightBytes` to roughly seven
times `maxFrameBytes` or by validating the pair at config time so an
unreachable combination fails at startup instead of mid-stage? And should a row
that cannot fit fall back to ordinary Spark shuffle for that stage rather than
failing the query?
Separately, the error messages are hard to act on. For a 63 MiB row the
message reports `a single row exceeds 1393693 bytes`, which is a derived
residual rather than any value the user configured. Naming
`spark.comet.shuffle.rss.maxInFlightBytes` and the multiple it needs would make
this diagnosable from a log line.
--
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]
