parthchandra opened a new issue, #7276:
URL: https://github.com/apache/datafusion-comet/issues/7276
Part of #7267.
## Problem
`CometBroadcastExchangeExec` collects the broadcast side and calls
`Utils.coalesceBroadcastBatches` (`CometBroadcastExchangeExec.scala:151`,
defined
at `spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala:379`).
That
method appends every broadcast batch into one Arrow Java `VectorSchemaRoot`
and
writes it as a single IPC batch. Its scaladoc (`Utils.scala:362-377`) gives
the
reason: it avoids decoding up to W*P tiny IPC streams on every task (for
example
~200K batches).
Executors decode it once and share the decoded copy across tasks
(`CometBroadcastValue` / `SharedDecode`, gated by
`spark.comet.exec.broadcastExchange.shareDecoded.enabled`, default true). It
is
still one batch. It reaches the native plan through `wrapColumnarBatchRDD`
and
`ScanExec`, and neither slices it.
So the broadcast side ignores `spark.comet.batchSize`. Its size is limited
only by
Spark's broadcast row and byte limits, and the byte check runs after the
merge
(`CometBroadcastExchangeExec.scala:157-163`).
## How it fails
Worked out from the Arrow Java 18.3.0 source; not reproduced yet:
- `VectorAppender` grows a variable-width data buffer by doubling it.
Doubling from
2^30 asks for 2^31, which is over the 2^31-1 maximum and throws
`OversizedAllocationException`. So the merge fails once one string/binary
column
passes about 1 GiB, not 2 GiB.
- The catch block only handles `IllegalArgumentException`
(`Utils.scala:426`). The
exception escapes and the query fails on the driver, with no fallback to
Spark.
- Possible edge case (unsure): the new capacity is computed as an int sum.
If the
merged size plus one incoming batch passes 2^31-1, the sum can go
negative, the
grow step is skipped, and the copy could write past the end of the buffer.
This
needs one source batch with about 1 GiB in one column, so it is rare. Needs
confirming.
The broadcast exchange is the join's direct child, so no native filter or
projection sees the merged batch. If the batches were kept separate, the
native
hash join would still concatenate its build side and fail at 2 GiB
(sub-issue 6).
Fixing this issue alone moves the failure from about 1 GiB to 2 GiB and
gives a
clear error.
Related: #7257 proposes grouping small native shuffle batches before
broadcast
serialization, bounded by rows and bytes. Any byte cap here should match it.
## Proposed fix
- Cap each merged batch by bytes as well as rows: start a new root before any
variable-width column passes about 1 GiB. This keeps the benefit of
merging tiny
batches.
- Catch `OversizedAllocationException` and either stop merging (keep the
remaining
batches separate) or fall back to Spark with a clear reason.
- Confirm or rule out the negative-capacity edge case in `VectorAppender`.
## Tests
- Scala test: coalesce batches whose combined string column is over the byte
cap
(use a small test cap), and check several batches come out with all rows.
- SQL-file test: broadcast hash join with a binary column of values a few
hundred
KB in size on the broadcast side, compared with Spark.
--
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]
