comphead commented on code in PR #3703:
URL: https://github.com/apache/datafusion-comet/pull/3703#discussion_r2942836335
##########
common/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala:
##########
@@ -252,6 +255,101 @@ object Utils extends CometTypeShim {
new ArrowReaderIterator(Channels.newChannel(ins), source)
}
+ /**
+ * Coalesces many small ChunkedByteBuffers (one per source partition) into a
single
+ * ChunkedByteBuffer. Without coalescing, each consumer task in a broadcast
hash join
+ * deserializes N separate Arrow IPC streams (one per source partition),
which dominates
+ * build-side time when partition counts are high (e.g. 200+ partitions in
TPC-H Q18).
+ *
+ * We decode and append all source batches into one VectorSchemaRoot on the
driver, then
+ * re-serialize once via ArrowStreamWriter. This is done on the driver (not
per-task) so the
+ * cost is paid once rather than once per consumer partition.
+ */
+ def coalesceBroadcastBatches(
+ input: Iterator[ChunkedByteBuffer]): (Array[ChunkedByteBuffer], Long,
Long) = {
+ val buffers = input.filterNot(_.size == 0).toArray
+ if (buffers.isEmpty) {
+ return (Array.empty, 0L, 0L)
+ }
+
+ val allocator = org.apache.comet.CometArrowAllocator
+ .newChildAllocator("broadcast-coalesce", 0, Long.MaxValue)
+ try {
+ var targetRoot: VectorSchemaRoot = null
+ var totalRows = 0L
+ var batchCount = 0
+
+ try {
+ for (bytes <- buffers) {
+ val codec = CompressionCodec.createCodec(SparkEnv.get.conf)
Review Comment:
can we reuse the codec?
--
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]
