Allow row-based build sides (such as JDBC) to convert to Arrow for native BroadcastHashJoin

[mixermt]

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):

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:

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:

// 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.