comphead opened a new issue, #7486: URL: https://github.com/apache/datafusion-comet/issues/7486
### What is the problem the feature request solves? When the build side of a Comet broadcast reads the JVM columnar shuffle, string columns with few distinct values arrive dictionary-encoded, because that shuffle's writer encodes them when `preferDictionaryRatio` allows. `Utils.serializeBatches` keeps the dictionaries in the broadcast's Arrow IPC streams, and then: - `coalesceBroadcastBatches` leaves every stream as it is and logs "Unexpected dictionary-encoded column during BroadcastExchange coalescing; skipping coalesce" ([Utils.scala#L483-L497](https://github.com/apache/datafusion-comet/blob/7bf572efb7/spark/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala#L483-L497)). Its comment says this should not happen. - `SharedDecode.decode` gives up on such streams ([CometBroadcastValue.scala#L268](https://github.com/apache/datafusion-comet/blob/7bf572efb7/spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastValue.scala#L268)), so every task decodes every stream itself ([CometBroadcastValue.scala#L225](https://github.com/apache/datafusion-comet/blob/7bf572efb7/spark/src/main/scala/org/apache/spark/sql/comet/CometBroadcastValue.scala#L225)) and the sharing from #7016 does not apply. Reproduced with default settings on `main` at 7bf572efb7. `dimlow` holds 1M rows of `k BIGINT` and `s = concat('cat-', id % 16)`: ```sql SELECT /*+ BROADCAST(d) */ count(*), count(d.s) FROM factlow f JOIN (SELECT /*+ REPARTITION(8) */ k, s FROM dimlow) d ON f.k = d.k ``` | build side of `d` | streams | with dictionaries | coalesced batches | `SharedDecode.decode` | |---|---:|---:|---:|---| | Parquet scan of `dimlow` | 1 | 0 | 128 | shared | | `REPARTITION(8)`, default shuffle mode | 128 | 128 | 0 | `None`, each task decodes | | `REPARTITION(8, k)`, default shuffle mode | 128 | 128 | 0 | `None`, each task decodes | | `GROUP BY k` over the JVM columnar shuffle | 1 | 0 | 160 | shared | The driver logged the warning once for each broadcast that carried dictionaries. ### Describe the potential solution Decode dictionary-encoded columns before the broadcast is serialized, in `CometBroadcastExchangeExec.getByteArrayRdd`, which only broadcasts use and which runs on the executors. `CometVectorUtils.withDecodedVectors` already does this for the export to native. The streams then coalesce, and the tasks share one decode as they do for other broadcasts. ### Additional context Part of #6013. In 13 production event logs I checked, 4 Comet broadcasts ran and none skipped coalescing, so this was not seen there. That sample is small. ### 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]
