mixermt opened a new issue, #6056:
URL: https://github.com/apache/datafusion-comet/issues/6056
### What is the problem the feature request solves?
When joining a large native dataset (e.g., Parquet or Iceberg) with a small
row-based dimension table (such as a MySQL / PostgreSQL JDBC table, or any
Spark plan containing non-native expressions/UDFs), Comet refuses to convert
`BroadcastExchangeExec` to `CometBroadcastExchangeExec`.
#### Root Cause in Code
In `CometExecRule.scala` (lines 467-468):
```scala
case b: BroadcastExchangeExec if
b.children.forall(_.isInstanceOf[CometNativeExec]) =>
convertToComet(b, CometBroadcastExchangeExec).getOrElse(b)
```
Because JDBC is a Spark row-based source (`RowDataSourceScanExec`),
`b.children.forall(_.isInstanceOf[CometNativeExec])` evaluates to `false`.
Furthermore, `CometBroadcastExchangeExec.scala` (lines 112-116) assumes
columnar execution:
```scala
private def getByteArrayRdd(plan: SparkPlan): RDD[(Long, ChunkedByteBuffer)]
= {
plan.executeColumnar().mapPartitionsInternal { iter =>
Utils.serializeBatches(iter)
}
}
```
#### Fallback Cascade
1. The build side remains a Spark `BroadcastExchangeExec` (producing
row-based `HashedRelation`).
2. `BroadcastHashJoinExec` cannot convert to `CometBroadcastHashJoinExec`
because it requires native broadcast input.
3. Comet is forced to insert `CometColumnarToRow` on the probe stream to
feed the Spark JVM join.
4. All downstream operations (such as `HashAggregateExec`) also fall back to
Spark JVM.
5. In production pipelines with large probe datasets, this JVM fallback
causes severe GC overhead and degrades performance significantly (e.g., turning
an aggregation that should take minutes into hours of JVM task execution).
*(Note: While #6008 discusses unsupported Text scan leaves, this issue
addresses the broader pattern of row-based data sources like JDBC and Spark
subtrees feeding broadcast joins).*
### Describe the potential solution
Support adapting row-based broadcast build sides to Arrow format so
`CometBroadcastHashJoinExec` can remain native.
Instead of requiring every leaf and intermediate operator on the build side
to be `CometNativeExec`, provide a row-to-Arrow bridge at the
`BroadcastExchange` boundary:
1. **Driver-Side / Exchange-Level Row-to-Arrow Conversion:**
In `CometExecRule.scala`, allow `BroadcastExchangeExec` with row-based
children to convert to `CometBroadcastExchangeExec` when the downstream join is
a candidate for `CometBroadcastHashJoinExec`.
2. In `CometBroadcastExchangeExec`, if `child` is not columnar
(`!child.supportsColumnar`), consume its rows via `child.execute()` and convert
them to Arrow RecordBatches before broadcast (e.g., using Spark's
`ArrowConverters` or bridging via `CometSparkToColumnarExec`).
3. Since broadcast tables are bounded by
`spark.sql.autoBroadcastJoinThreshold` (default 10MB), the CPU and memory cost
of converting a few megabytes of rows to Arrow is trivial, while the benefit of
keeping the multi-gigabyte/terabyte probe stream and downstream
joins/aggregates in native code is enormous.
### Additional context
**Reproducer scenario:**
```scala
// Native large Iceberg / Parquet fact table
val fact = spark.read.table("large_iceberg_events")
// Small dimension table from MySQL via JDBC
val dim = spark.read.format("jdbc")
.option("url", "jdbc:mysql://...")
.option("dbtable", "dim_countries")
.load()
// Broadcast join
val result = fact.join(broadcast(dim), "country_id")
.groupBy("country_name")
.count()
```
*Current behavior:* `BroadcastExchangeExec`, `BroadcastHashJoinExec`, and
`HashAggregateExec` all fall back to JVM.
*Expected behavior:* `dim` is converted to Arrow at the broadcast exchange,
allowing `CometBroadcastExchangeExec`, `CometBroadcastHashJoinExec`, and
`CometHashAggregateExec` to execute natively.
--
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]
