comphead opened a new issue, #7192:
URL: https://github.com/apache/datafusion-comet/issues/7192
### Describe the bug
A broadcast join fails when one string or binary column of the build side
holds more than 1 GiB in total. Spark runs the same query. Such a column still
fits a single `VarCharVector`, whose 32-bit offsets address up to 2 GiB.
```
org.apache.comet.shaded.arrow.vector.util.OversizedAllocationException:
Memory required for vector is (2147483648), which is overflow or more than max
allowed (2147483647). You could consider using
LargeVarCharVector/LargeVarBinaryVector for large strings/large bytes types
at
org.apache.comet.shaded.arrow.vector.BaseVariableWidthVector.checkDataBufferSize(BaseVariableWidthVector.java:461)
at
org.apache.comet.shaded.arrow.vector.BaseVariableWidthVector.reallocDataBuffer(BaseVariableWidthVector.java:579)
at
org.apache.comet.shaded.arrow.vector.BaseVariableWidthVector.reallocDataBuffer(BaseVariableWidthVector.java:557)
at
org.apache.comet.shaded.arrow.vector.util.VectorAppender.visit(VectorAppender.java:134)
at
org.apache.comet.shaded.arrow.vector.util.VectorAppender.visit(VectorAppender.java:45)
at
org.apache.comet.shaded.arrow.vector.BaseVariableWidthVector.accept(BaseVariableWidthVector.java:1496)
at
org.apache.comet.shaded.arrow.vector.util.VectorSchemaRootAppender.append(VectorSchemaRootAppender.java:68)
at
org.apache.comet.shaded.arrow.vector.util.VectorSchemaRootAppender.append(VectorSchemaRootAppender.java:83)
at
org.apache.spark.sql.comet.util.Utils$.$anonfun$coalesceBroadcastBatches$2(Utils.scala:412)
at
org.apache.spark.sql.comet.util.Utils$.$anonfun$coalesceBroadcastBatches$2$adapted(Utils.scala:383)
at scala.collection.ArrayOps$.foreach$extension(ArrayOps.scala:1324)
at
org.apache.spark.sql.comet.util.Utils$.coalesceBroadcastBatches(Utils.scala:383)
at
org.apache.spark.sql.comet.CometBroadcastExchangeExec.$anonfun$relationFuture$1(CometBroadcastExchangeExec.scala:151)
```
#6779 reported the same exception, but this build includes its fix (#6900)
and the cause is different.
`coalesceBroadcastBatches` appends every broadcast batch into one
`VectorSchemaRoot`. Arrow Java's `VectorAppender` grows a variable-width data
buffer by doubling it
([VectorAppender.java#L133-L135](https://github.com/apache/arrow-java/blob/v18.3.0/vector/src/main/java/org/apache/arrow/vector/util/VectorAppender.java#L133-L135),
[BaseVariableWidthVector.java#L546-L580](https://github.com/apache/arrow-java/blob/v18.3.0/vector/src/main/java/org/apache/arrow/vector/BaseVariableWidthVector.java#L546-L580)).
From 2^30 bytes the next step asks for 2^31, one byte over `MAX_BUFFER_SIZE`,
and throws. A coalesced string or binary column is therefore capped at 1 GiB,
half of what its offsets allow. Arrow Java `main` grows the buffer the same way.
Two things in Comet turn the cap into a query failure:
- The `catch` around the append
([Utils.scala#L422-L434](https://github.com/apache/datafusion-comet/blob/69111f68a5ff62904342910c445d191f53615aff/spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala#L422-L434))
falls back to the uncoalesced buffers only on `IllegalArgumentException`.
`OversizedAllocationException` is a `RuntimeException`, so it fails the
broadcast.
- `CometBroadcastExchangeExec` checks `maxBroadcastTableBytes` (8GB by
default) after coalescing, against the compressed result
([CometBroadcastExchangeExec.scala#L151-L163](https://github.com/apache/datafusion-comet/blob/69111f68a5ff62904342910c445d191f53615aff/spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastExchangeExec.scala#L151-L163)).
Nothing bounds the uncompressed size before the append.
Coalescing came in with #3703 (0.15.0), and every release branch since then
has this code. Before #3703 each batch was broadcast as its own buffer.
### Steps to reproduce
Spark 4.0 with a Comet build based on `main` at `d0f367dc02`, which includes
#6900. `coalesceBroadcastBatches` is unchanged on `main` since then. I ran
without Comet shuffle because the query has no shuffle.
```shell
spark-shell --master local[2] --driver-memory 8g --jars
comet-spark-spark4.0_2.13-1.2.0-SNAPSHOT.jar \
--conf spark.plugins=org.apache.spark.CometPlugin \
--conf spark.comet.shuffle.enabled=false \
--conf spark.memory.offHeap.enabled=true \
--conf spark.memory.offHeap.size=4g
```
```scala
val dir = "/tmp/broadcast-coalesce"
spark.range(1000).write.mode("overwrite").parquet(s"$dir/a")
spark.range(1200000)
.selectExpr("id", "concat(cast(id as string), repeat('x', 1000)) AS s")
.write.mode("overwrite").parquet(s"$dir/b")
spark.read.parquet(s"$dir/a").createOrReplaceTempView("a")
spark.read.parquet(s"$dir/b").createOrReplaceTempView("b")
spark.sql("SELECT /*+ BROADCAST(b) */ a.id, length(b.s) AS len FROM a JOIN b
ON a.id = b.id").collect()
```
Column `s` of `b` holds 1,207,288,890 bytes (1.12 GiB), and the query fails
with the exception above.
| rows in `b` | bytes in `s` | Comet | Comet disabled |
|---:|---:|---|---|
| 1,000,000 | 1,005,888,890 (0.94 GiB) | 1,000 rows | 1,000 rows |
| 1,200,000 | 1,207,288,890 (1.12 GiB) | `OversizedAllocationException` |
1,000 rows |
### Expected behavior
The query returns the same 1,000 rows as Spark, as it does for the 0.94 GiB
column.
### Additional context
A possible fix is to track the bytes of each variable-width vector, nested
children included, while coalescing, and to start a new output batch before any
of them would pass 2^30. `coalesceBroadcastBatches` already returns an
`Array[ChunkedByteBuffer]`, and its uncoalesced fallback returns many buffers,
so readers accept more than one. The check needs `Long` arithmetic in Comet,
because `VectorAppender` adds the two sizes as an `Int` and a sum that
overflows skips its growth loop. Catching `OversizedAllocationException` and
keeping the uncoalesced buffers would be a backstop.
Past 2 GiB a column no longer fits one array with 32-bit offsets, and
DataFusion's `HashJoinExec` concatenates its build side into one batch, so that
case is a separate limit.
Setting `spark.comet.exec.broadcastExchange.enabled=false` works around it.
The repro then runs the broadcast and the join in Spark and returns 1,000 rows.
#5538 adds a 1 GiB uncompressed budget that skips coalescing, but only on
its opt-in direct-read path (`nativeIpc = true`).
### Willingness to contribute
I can contribute a fix for this bug independently
--
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]
