Text-format source on a broadcast build side blocks native `BroadcastHashJoin`

[parthchandra]

What is the problem the feature request solves?

Comet doesn't natively scan the Text file format ([COMET: Unsupported file format Text]). In isolation that's fine, but when the unsupported source sits on the build side of a broadcast join, it cascades: the build branch (text scan → transforms → BroadcastExchange) stays on Spark, so the exchange never becomes a CometBroadcastExchange, so the BroadcastHashJoin can't go native — even though the large probe side is a fully native scan+filter. A tiny lookup table read from text disqualifies acceleration of the entire join over the big stream.

Describe the potential solution

The core problem isn't that Text is unsupported — it's that an unsupported leaf on a broadcast build side vetoes native execution of the whole join, including the large probe stream. One possible fix is to give the build branch a columnar bridge so it can reach the BroadcastExchange as native. There are two options

Option A — auto-insert SparkToColumnar at unsupported leaves on broadcast build sides

CometSparkToColumnarExec wraps a Spark leaf and is delivered as a CometScanWrapper (a CometNativeExec), so operators above it (explode/Filter/Project) can convert and the exchange becomes a CometBroadcastExchange. Today it's off by default and its sparkToColumnar.supportedOperatorList excludes FileSourceScan, so this path is never hit

Change: when an unsupported leaf scan feeds a broadcast build side and nothing else bloc ks the branch, insert CometSparkToColumnarExec at that leaf automatically. This is safe precisely because the build side is broadcast — it's bounded and small, so the row→Arrow copy cost is negligible, and the payoff is a native join over the (large) probe input.

Guards:

• Only on broadcast build sides (or gate by a size/row threshold) to avoid converting large row inputs where the copy would dominate.
• Keep the existing isSchemaSupported check.
• Respect a config, e.g. spark.comet.sparkToColumnar.broadcastBuildSide.enabled (default true), plus the existing opt-in for the general case.

As an interim, this is already achievable by config — document it:

--conf spark.comet.sparkToColumnar.enabled=true
--conf spark.comet.sparkToColumnar.supportedOperatorList=Range,InMemoryTableScan,RDDScan,OneRowRelation,FileSourceScan

Option B — native Text scan

Add a minimal native Text reader (single value: string column, line-delimited), mirroring the CSV native scan path, gated by e.g. spark.comet.scan.text.enabled. This removes the row→columnar copy entirely and gives native I/O on the build side, not just a bridge. More work, but it's the general fix and closes the "unsupported file format" gap directly.

Note / follow-up

Whatever the bridge, plan-time conversion of the build chain still hinges on the transforms above the leaf converting (e.g. explode over a Scala UDF returning array<string> — supported at plan time via codegen dispatch). Runtime execution of a native Explode over a JvmScalarUdf child should be validated separately; it's orthogonal to enabling the columnar bridge but worth testing in the same change.

Additional context

Steps to reproduce

Join a large Parquet/Iceberg fact against a small allowlist loaded from a text file:

val big   = spark.read.format("iceberg").load("db.events")          // native Comet scan
val allow = spark.read.text("s3a://example-bucket/config/allow.txt") // Text -> Spark only
             .select(explode(to_hosts($"value")).as("host"), $"value".as("pattern"))
big.join(broadcast(allow), big("host") === allow("host"))
   .where(rlike(big("url"), allow("pattern")))

Explain: only CometIcebergNativeScan + CometFilter on the probe side convert; Generate, Filter, Project, BroadcastExchange, and the BroadcastHashJoin all remain Spark. Summary: accelerated 2 of 12 operators (16%).

Expected

One of:

1. Native scan support for Text (even a minimal single-value-column reader), or
2. Make spark.comet.sparkToColumnar.enabled cover file-source leaves like Text by default (or document adding FileSourceScan to sparkToColumnar.supportedOperatorList), so a small build side can be adapted to Arrow and let the join go native.

Additional context

• With sparkToColumnar.enabled=true and FileSourceScan added to the supported list, the build chain (SparkToColumnar leaf → explode → filter → project) does convert at plan time, which flips the exchange and join native — but this is non-obvious and off by default.
• Impact is huge: a small text lookup on the build side blocks native execution of a join over a very large probe input.