diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md index be55a72085e..fdf621ac8cb 100644 --- a/docs/source/user-guide/latest/compatibility/index.md +++ b/docs/source/user-guide/latest/compatibility/index.md @@ -97,9 +97,41 @@ underlying bytes rather than on the rendered text: bytes) but `true` in Comet. Equality, joins, grouping, ordering, and byte-based string functions such as `contains` can therefore disagree with Spark when non-UTF-8 bytes are involved. -Both differences require string data that is not valid UTF-8, which does not occur for text read from -Parquet or produced by string expressions. Consistent handling of invalid UTF-8 across all native -string paths is tracked by +Both differences require string data that is not valid UTF-8. Comet's native Parquet scan never +produces such strings (it rejects them, see below), and neither do most string expressions. They can +reach Comet in these ways, which are the ones to check before relying on a result: + +- **Text files.** Spark's Text reader keeps the bytes of each line as they are, so a Text file in an + encoding other than UTF-8, such as Latin-1, holds strings that are not valid UTF-8. Comet decodes + them when it converts a Text scan to Arrow, which it does only when + `spark.comet.convert.text.enabled=true` (see [Data sources](../datasources.md#text)), or, as a + deprecated alternative, when `spark.comet.sparkToColumnar.enabled=true` and + `spark.comet.sparkToColumnar.supportedOperatorList` names the scan. For example, with a Text file + holding the bytes `1 caf\xE9`, `hex(value)` returns `3120636166EFBFBD` instead of Spark's + `3120636166E9`, and `length` of `2 na\xEFve` is 7 instead of 5. Spark decodes CSV files as it reads + them, so they never hold such strings, and its JSON reader treats a line with such bytes as + malformed, so only the corrupt record column keeps them. Spark's ORC and Avro readers keep stored + bytes without checking them, but those formats define strings as UTF-8, so only a malformed file is + affected. +- **Comet columnar shuffle.** When Comet shuffles rows that come from a Spark operator, such as a + Text scan or a Spark partial aggregate, it converts them to Arrow and decodes their strings in the + same way. It does this with the default `spark.comet.shuffle.mode=auto`, so, for example, + repartitioning a Latin-1 Text table can return `U+FFFD` where Spark keeps the original bytes. + Setting `spark.comet.shuffle.mode=native` keeps such shuffles on Spark. +- **`CAST(binary AS STRING)`.** Comet decodes the bytes where Spark keeps them. Because `min` and `max` + over strings run natively, a cast below them is decoded before the comparison, and the two can swap. + With `b` holding `X'80'` and `X'C3A9'`, Spark returns `min(CAST(b AS STRING))` = the `0x80` string + and `max` = `é`, whereas Comet returns `min` = `é` and `max` = `U+FFFD`. This is the ordering + difference above. If the cast input can hold bytes that are not valid UTF-8, disable Comet for the + query. When the cast is the direct argument of `min` or `max`, a narrower alternative is to set + `spark.comet.expression.Min.enabled=false`, `spark.comet.expression.Max.enabled=false` and + `spark.comet.shuffle.mode=native`. All three are needed: with only the first two, the aggregate + runs in Spark, but Comet's columnar shuffle still decodes its partial results. These settings are + not enough when the cast is in a projection below the aggregate, for example + `SELECT min(s) FROM (SELECT CAST(b AS STRING) AS s FROM t)`, because Comet runs that projection + natively and decodes the bytes before the aggregate sees them. + +Consistent handling of invalid UTF-8 across all native string paths is tracked by [#4764](https://github.com/apache/datafusion-comet/issues/4764). Separately, Comet's native Parquet scan currently rejects string columns whose stored bytes are not diff --git a/docs/source/user-guide/latest/datasources.md b/docs/source/user-guide/latest/datasources.md index f687d3ab3b3..b9754029911 100644 --- a/docs/source/user-guide/latest/datasources.md +++ b/docs/source/user-guide/latest/datasources.md @@ -49,6 +49,15 @@ converted into Arrow format, allowing the Comet pipeline to take over after that Comet does not provide a Rust-based JSON scan, but when `spark.comet.convert.json.enabled` is enabled, data is immediately converted into Arrow format, allowing the Comet pipeline to take over after that. +### Text + +Comet does not provide a Rust-based Text scan, but when `spark.comet.convert.text.enabled` is enabled, data from Spark's +Text reader is immediately converted into Arrow format, allowing the Comet pipeline to take over after that. Spark's Text +reader keeps the bytes of each line as they are, so a file in an encoding other than UTF-8, such as Latin-1, holds strings +that are not valid UTF-8. Arrow strings must be valid UTF-8, so the conversion replaces those bytes with `U+FFFD`, and +results that depend on the original bytes can differ from Spark. Enable it only for Text files that are UTF-8. See +[Strings with non-UTF-8 bytes](compatibility/index.md#strings-with-non-utf-8-bytes). + ### Other Spark inputs Comet can also convert the output of these Spark inputs to Arrow format, so that the operators @@ -67,12 +76,15 @@ because the row it converts has no columns. such as JDBC tables, which Spark scans with `RowDataSourceScanExec`. To convert any other leaf operator, such as the scan of a Data Source V2 connector or of a file -format other than Parquet, JSON and CSV, set `spark.comet.sparkToColumnar.enabled=true` and name the -operator in `spark.comet.sparkToColumnar.supportedOperatorList` by its Spark class name without the -`Exec` suffix, such as `BatchScan` or `FileSourceScan`. +format other than Parquet, JSON, CSV and Text, set `spark.comet.sparkToColumnar.enabled=true` and name +the operator in `spark.comet.sparkToColumnar.supportedOperatorList` by its Spark class name without the +`Exec` suffix, such as `BatchScan` or `FileSourceScan`. Naming either to convert a Text scan still +works but is deprecated, like naming `Range`, `InMemoryTableScan`, `RDDScan`, `OneRowRelation` or +`RowDataSourceScan`, and will stop working in a future major release. Use +`spark.comet.convert.text.enabled` instead. One case is converted automatically: when the build (broadcast) side of a broadcast join is a file -or V2 scan Comet cannot read natively - for example a small Text lookup table - and every scan on +or V2 scan Comet cannot read natively - for example a small ORC lookup table - and every scan on the probe side is already native, Comet converts that build-side scan to Arrow so the whole join runs natively over the large probe. This is controlled by `spark.comet.convert.broadcastBuildSide.enabled`, which is **on by default**, unlike the switches @@ -81,6 +93,10 @@ the scan can make the copied volume larger, and with dynamic partition pruning t a second time for the pruning subquery. Set `spark.comet.convert.broadcastBuildSide.enabled=false` to keep such build sides on Spark. +Parquet, JSON, CSV and Text build sides are converted only by their own switches above. A Text file +can hold bytes that are not valid UTF-8 (see [Text](#text)), so a Text lookup table stays on Spark +unless `spark.comet.convert.text.enabled` is set. + ### Spark-to-Comet conversion types Spark-to-Comet conversion supports `ARRAY` and `MAP` with binary diff --git a/docs/source/user-guide/latest/migration-guide.md b/docs/source/user-guide/latest/migration-guide.md index d48cc116834..41cae7545a8 100644 --- a/docs/source/user-guide/latest/migration-guide.md +++ b/docs/source/user-guide/latest/migration-guide.md @@ -86,6 +86,9 @@ it. `spark.comet.convert.oneRowRelation.enabled` is on by default, so such a que without either setting, and logs no warning. The list remains the way to convert other leaf operators, such as the scan of a Data Source V2 connector. +Converting a Text scan with `FileSourceScan` or `BatchScan` in the list is deprecated in the same +way, and logs the same warning. Use `spark.comet.convert.text.enabled` instead. + ### Iceberg Writes Comet's Iceberg write path is now switched by one setting, `spark.comet.write.iceberg.enabled`, diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index d982dd4cc28..2cc80d30291 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -227,6 +227,18 @@ object CometConf extends ShimCometConf { .booleanConf .createWithDefault(false) + val COMET_CONVERT_FROM_TEXT_ENABLED: ConfigEntry[Boolean] = + conf("spark.comet.convert.text.enabled") + .category(CATEGORY_EXEC) + .doc( + "When enabled, data from Spark (non-native) Text v1 and v2 scans will be converted to " + + "Arrow format. Arrow strings must be valid UTF-8, so Comet replaces bytes that are " + + "not valid UTF-8, for example in a Latin-1 file, with U+FFFD, and results that " + + "depend on the original bytes can differ from Spark. Enable this only for Text " + + "sources that are UTF-8. See the compatibility guide.") + .booleanConf + .createWithDefault(false) + val COMET_CONVERT_FROM_RANGE_ENABLED: ConfigEntry[Boolean] = conf("spark.comet.convert.range.enabled") .category(CATEGORY_EXEC) @@ -1257,10 +1269,11 @@ object CometConf extends ShimCometConf { s"`${COMET_SPARK_TO_ARROW_ENABLED.key}` is true, each named by its Spark class name " + "without the `Exec` suffix. It is for operators that have no `spark.comet.convert` " + "config of their own, such as `BatchScan` for a Data Source V2 connector, or " + - "`FileSourceScan` for file formats other than Parquet, JSON and CSV. Naming `Range`, " + - "`InMemoryTableScan`, `RDDScan`, `OneRowRelation` or `RowDataSourceScan` is " + - "deprecated and will stop working in a future major release. Use their " + - "`spark.comet.convert` configs instead.") + "`FileSourceScan` for file formats other than Parquet, JSON, CSV and Text. Naming " + + "`Range`, `InMemoryTableScan`, `RDDScan`, `OneRowRelation` or `RowDataSourceScan`, " + + "or naming `FileSourceScan` or `BatchScan` to convert a Text scan, is deprecated " + + "and will stop working in a future major release. Use their `spark.comet.convert` " + + "configs instead.") .stringConf .toSequence .createWithDefault(Nil) @@ -1269,13 +1282,14 @@ object CometConf extends ShimCometConf { conf("spark.comet.convert.broadcastBuildSide.enabled") .category(CATEGORY_EXEC) .doc("When enabled, Comet converts a file or V2 scan that it cannot scan natively (for " + - "example a Text source) to Arrow when that scan feeds a broadcast join's build side and " + + "example an ORC source) to Arrow when that scan feeds a broadcast join's build side and " + "every scan on the probe side is already native, so the broadcast join runs natively " + "over the large probe. The build side is usually small (auto-broadcasts are capped by " + "the broadcast threshold), so the row to Arrow copy is usually cheap - though an " + "explicit BROADCAST hint or a selective filter above the scan can make it larger. With " + "dynamic partition pruning the dimension is also read a second time for the pruning " + - "subquery. Set this to false to keep such build sides on Spark.") + "subquery. Parquet, JSON, CSV and Text scans are converted only by their own " + + "`spark.comet.convert` configs. Set this to false to keep such build sides on Spark.") .booleanConf .createWithDefault(true) diff --git a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala index a9e3f442f7c..6a03efe2a54 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala @@ -45,10 +45,12 @@ import org.apache.spark.sql.execution.datasources.{InsertIntoHadoopFsRelationCom import org.apache.spark.sql.execution.datasources.csv.CSVFileFormat import org.apache.spark.sql.execution.datasources.json.JsonFileFormat import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat +import org.apache.spark.sql.execution.datasources.text.TextFileFormat import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, V2CommandExec} import org.apache.spark.sql.execution.datasources.v2.csv.CSVScan import org.apache.spark.sql.execution.datasources.v2.json.JsonScan import org.apache.spark.sql.execution.datasources.v2.parquet.ParquetScan +import org.apache.spark.sql.execution.datasources.v2.text.TextScan import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, BroadcastExchangeLike, Exchange, ReusedExchangeExec, ShuffleExchangeExec, ShuffleExchangeLike} import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, ShuffledHashJoinExec, SortMergeJoinExec} import org.apache.spark.sql.execution.window.WindowExec @@ -192,6 +194,15 @@ object CometExecRule { "Spark to Arrow conversion is not compatible with input_file_name, " + "input_file_block_start, or input_file_block_length" + /** + * Why Comet does not bridge a Text scan on a broadcast build side. A Text file can hold bytes + * that are not valid UTF-8, and the conversion replaces them with U+FFFD. See issue #7118. + */ + private val TEXT_BUILD_SIDE_FALLBACK_REASON: String = + "Comet converts a Text scan on a broadcast build side to Arrow only when " + + s"${CometConf.COMET_CONVERT_FROM_TEXT_ENABLED.key}=true, because a Text " + + "file can hold bytes that are not valid UTF-8" + /** * Tag set on a scan leaf that sits on a broadcast join's build side. When * `spark.comet.convert.broadcastBuildSide.enabled` is on, such a scan is bridged to Arrow with @@ -989,7 +1000,7 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false) tagUnsafePartialAggregates(planWithJoinRewritten) // Tag leaves on broadcast join build sides so shouldApplySparkToColumnar can bridge an - // unsupported build-side scan (e.g. Text) to Arrow and let the join run natively (#6008). + // unsupported build-side scan (e.g. ORC) to Arrow and let the join run natively (#6008). tagBroadcastBuildSideLeaves(planWithJoinRewritten) var transformed = transform(planWithJoinRewritten) @@ -1459,8 +1470,9 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false) case _: CSVFileFormat => CometConf.COMET_CONVERT_FROM_CSV_ENABLED.get(conf) case _: JsonFileFormat => CometConf.COMET_CONVERT_FROM_JSON_ENABLED.get(conf) case _: ParquetFileFormat => CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.get(conf) - // Unsupported format (e.g. Text): bridge it if the general opt-in is on or it feeds a - // broadcast build side. The per-format arms above keep honoring their own opt-outs. + case _: TextFileFormat => shouldConvertTextScan(conf, op) + // Unsupported format (e.g. ORC): bridge it if the general opt-in is on or it feeds a + // broadcast build side. The per-format arms above keep honoring their own switches. case _ => shouldBridgeUnsupportedScan(conf, op) } // Convert Spark DS v2 scan to Arrow format @@ -1469,6 +1481,7 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false) case _: CSVScan => CometConf.COMET_CONVERT_FROM_CSV_ENABLED.get(conf) case _: JsonScan => CometConf.COMET_CONVERT_FROM_JSON_ENABLED.get(conf) case _: ParquetScan => CometConf.COMET_CONVERT_FROM_PARQUET_ENABLED.get(conf) + case _: TextScan => shouldConvertTextScan(conf, op) case _ => shouldBridgeUnsupportedScan(conf, op) } case _: RangeExec => @@ -1516,12 +1529,28 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false) /** * An unsupported file-source scan is bridged to Arrow when the general sparkToColumnar opt-in * covers it, or when it feeds a broadcast join's build side (see issue #6008). This is only - * reached from the unsupported-format arms, so the per-format `COMET_CONVERT_FROM_*` opt-outs - * for CSV/JSON/Parquet still take precedence. + * reached from the unsupported-format arms, so the per-format `COMET_CONVERT_FROM_*` switches + * for CSV/JSON/Parquet/Text still take precedence. */ private def shouldBridgeUnsupportedScan(conf: SQLConf, op: SparkPlan): Boolean = isSparkToArrowEnabled(conf, op) || isBroadcastBuildSideLeaf(op) + /** + * A Text scan is converted only by `spark.comet.convert.text.enabled`, like CSV, JSON and + * Parquet, and so is a Text scan on a broadcast build side: a Text file can hold bytes that are + * not valid UTF-8, for example a Latin-1 file, and the conversion replaces them with U+FFFD, so + * results that depend on the original bytes would differ from Spark. A Text build side that the + * bridge would otherwise convert records why it did not, so the explain output names the + * config. See issue #7118. + */ + private def shouldConvertTextScan(conf: SQLConf, op: SparkPlan): Boolean = + isConversionEnabled(conf, op, CometConf.COMET_CONVERT_FROM_TEXT_ENABLED) || { + if (isBroadcastBuildSideLeaf(op)) { + withFallbackReason(op, CometExecRule.TEXT_BUILD_SIDE_FALLBACK_REASON) + } + false + } + /** * True when the build-side auto-bridge should run: its own config is on and Comet broadcast * exchange conversion is enabled. Without the latter the `BroadcastExchangeExec` cannot become @@ -1550,7 +1579,7 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false) /** * Tag the file-source scan leaves on a broadcast join's build side so - * shouldApplySparkToColumnar can bridge an unsupported build-side scan (e.g. a Text scan) to + * shouldApplySparkToColumnar can bridge an unsupported build-side scan (e.g. an ORC scan) to * Arrow, letting the whole join run natively. The build side is usually small (auto-broadcasts * are capped by the broadcast threshold), so the row to Arrow copy is usually cheap - but that * is a heuristic, not a bound: an explicit BROADCAST hint can force a larger build side, and a @@ -1680,10 +1709,11 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false) * Two guards keep the walk honest. It stops at a `CometBroadcastExchangeExec` - a nested * broadcast join that went native keeps its bridge. And it only clears a tag (and so asks for a * second transform) when the tagged scan sits under a `CometSparkToColumnarExec`, i.e. it was - * actually bridged. A CSV/JSON/Parquet scan whose per-format arm left it on Spark still carries - * the tag but was never bridged, so touching it would force a pointless second transform. The - * descent enters `CometScanWrapper`'s wrapped `originalPlan` by hand because it is a leaf, so a - * plain `foreach` would never reach the bridged scan inside it. See issue #6008. + * actually bridged. A CSV/JSON/Parquet/Text scan whose per-format arm left it on Spark still + * carries the tag but was never bridged, so touching it would force a pointless second + * transform. The descent enters `CometScanWrapper`'s wrapped `originalPlan` by hand because it + * is a leaf, so a plain `foreach` would never reach the bridged scan inside it. See issue + * #6008. */ private def clearWastedBuildSideBridges(plan: SparkPlan): Boolean = { var cleared = false diff --git a/spark/src/test/scala/org/apache/comet/CometBroadcastBuildSideSuite.scala b/spark/src/test/scala/org/apache/comet/CometBroadcastBuildSideSuite.scala index da7bf08f5d7..a1cb9ffdec2 100644 --- a/spark/src/test/scala/org/apache/comet/CometBroadcastBuildSideSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometBroadcastBuildSideSuite.scala @@ -19,20 +19,30 @@ package org.apache.comet -import org.apache.spark.sql.CometTestBase +import java.io.File +import java.nio.charset.StandardCharsets.UTF_8 +import java.nio.file.Files + +import org.apache.spark.sql.{CometTestBase, DataFrame, Row} import org.apache.spark.sql.catalyst.expressions.{DynamicPruningExpression, Literal} import org.apache.spark.sql.comet._ import org.apache.spark.sql.execution._ import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, QueryStageExec} import org.apache.spark.sql.execution.exchange.ReusedExchangeExec +import org.apache.spark.sql.execution.joins.BroadcastNestedLoopJoinExec import org.apache.spark.sql.internal.SQLConf -// ORC and Iceberg build sides that Comet cannot scan natively, bridged by -// spark.comet.convert.broadcastBuildSide.enabled (from the PR #6786 review of issue #6008). Each -// query is compared with Spark through checkSparkAnswer, and every combination asserts that the -// broadcast join ran natively (bridged when the scan is unsupported). See issue #6008. +import org.apache.comet.CometSparkSessionExtensions.isSpark35Plus + +// Tests for the broadcast build-side bridge, spark.comet.convert.broadcastBuildSide.enabled (issue +// #6008). A build-side scan that Comet cannot read natively, such as ORC or an Iceberg table +// the native reader declines, is converted to Arrow so that the broadcast join runs natively. Each +// query is compared with Spark, and the tests assert where the bridge must and must not appear. class CometBroadcastBuildSideSuite extends CometTestBase with CometIcebergTestBase { + // Most tests use an ORC build side. Text, Parquet, JSON and CSV scans are converted only by their + // own spark.comet.convert configs, so they would not exercise the bridge (see issue #7118). + /** Every node, descending through AQE, stages and subqueries, but not into reused exchanges. */ private def allNodes(p: SparkPlan): Seq[SparkPlan] = { val kids = p match { @@ -204,11 +214,11 @@ class CometBroadcastBuildSideSuite extends CometTestBase with CometIcebergTestBa } } - test("Spark 3.4: native Iceberg fact, text dimension, AQE DPP (#6008)") { + test("Spark 3.4: native Iceberg fact, ORC dimension, AQE DPP (#6008)") { assume(icebergAvailable, "Iceberg not available in classpath") withTempIcebergDir { warehouseDir => withTempDir { dir => - val dim = s"${dir.getAbsolutePath}/dim.txt" + val dim = s"${dir.getAbsolutePath}/dim.orc" withSQLConf( "spark.sql.catalog.pc" -> "org.apache.iceberg.spark.SparkCatalog", "spark.sql.catalog.pc.type" -> "hadoop", @@ -221,9 +231,9 @@ class CometBroadcastBuildSideSuite extends CometTestBase with CometIcebergTestBa .selectExpr("id AS v", "CAST(id % 10 AS STRING) AS fk") .writeTo("pc.db.fact") .append() - spark.range(0, 10).selectExpr("CAST(id AS STRING) AS value").write.text(dim) + spark.range(0, 10).selectExpr("CAST(id AS STRING) AS value").write.orc(dim) } - spark.read.text(dim).createOrReplaceTempView("ice_dim") + spark.read.orc(dim).createOrReplaceTempView("ice_dim") val query = """SELECT /*+ BROADCAST(d) */ f.v FROM pc.db.fact f JOIN |(SELECT value FROM ice_dim WHERE CAST(value AS INT) % 3 = 0) d @@ -249,4 +259,698 @@ class CometBroadcastBuildSideSuite extends CometTestBase with CometIcebergTestBa } } } + + test("broadcast build side with unsupported ORC source goes native (#6008)") { + // A tiny lookup table read from an ORC file (which Comet cannot scan natively) sits on the + // build side of a broadcast join over a large native probe. Without the auto-bridge the whole + // join stays on Spark. With spark.comet.convert.broadcastBuildSide.enabled (default on) the + // ORC leaf is bridged to Arrow so the broadcast and the join run natively. + assert(CometConf.COMET_CONVERT_BROADCAST_BUILD_SIDE_ENABLED.defaultValue.contains(true)) + withTempDir { dir => + val probePath = s"${dir.getAbsolutePath}/probe.parquet" + val allowPath = s"${dir.getAbsolutePath}/allow.orc" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(0, 100) + .selectExpr("CAST(id AS STRING) AS key", "id AS v") + .write + .parquet(probePath) + spark + .range(0, 5) + .selectExpr("CAST(id AS STRING) AS value") + .write + .orc(allowPath) + } + + // Cover both the AQE and non-AQE planning paths - the tag pre-pass runs per stage under AQE. + for (aqe <- Seq(false, true)) { + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) { + spark.read.parquet(probePath).createOrReplaceTempView("probe_6008") + spark.read.orc(allowPath).createOrReplaceTempView("allow_6008") + val query = + "SELECT /*+ BROADCAST(a) */ p.v FROM probe_6008 p JOIN allow_6008 a ON p.key = a.value" + + // Default: the ORC build side is bridged to Arrow and the join goes native. The plan is + // fully native (the bridge counts as a native leaf), so use checkSparkAnswerAndOperator - + // checkSparkAnswer would fail under COMET_STRICT_TESTING on a fully-native plan. + val (_, cometPlan) = checkSparkAnswerAndOperator(sql(query)) + assert( + collect(cometPlan) { case s: CometSparkToColumnarExec => s }.nonEmpty, + s"aqe=$aqe: expected a CometSparkToColumnarExec on the ORC build side:\n" + + cometPlan.treeString) + assert( + collect(cometPlan) { case b: CometBroadcastExchangeExec => b }.nonEmpty, + s"aqe=$aqe: expected a CometBroadcastExchangeExec:\n${cometPlan.treeString}") + assert( + collect(cometPlan) { case j: CometBroadcastHashJoinExec => j }.nonEmpty, + s"aqe=$aqe: expected a CometBroadcastHashJoinExec:\n${cometPlan.treeString}") + + // Off: the unsupported ORC leaf blocks native execution of the broadcast and the join. + withSQLConf(CometConf.COMET_CONVERT_BROADCAST_BUILD_SIDE_ENABLED.key -> "false") { + val (_, cometPlanOff) = checkSparkAnswer(sql(query)) + assert( + collect(cometPlanOff) { case s: CometSparkToColumnarExec => s }.isEmpty, + s"aqe=$aqe: expected no CometSparkToColumnarExec when disabled:\n" + + cometPlanOff.treeString) + assert( + collect(cometPlanOff) { case b: CometBroadcastExchangeExec => b }.isEmpty, + s"aqe=$aqe: expected no CometBroadcastExchangeExec when disabled:\n" + + cometPlanOff.treeString) + assert( + collect(cometPlanOff) { case j: CometBroadcastHashJoinExec => j }.isEmpty, + s"aqe=$aqe: expected no CometBroadcastHashJoinExec when disabled:\n" + + cometPlanOff.treeString) + } + } + } + } + } + + test("a Text build side keeps Spark's bytes unless Text conversion is on (#7118)") { + // The bridge is on by default, but a Text scan is converted only by + // spark.comet.convert.text.enabled, like Parquet, JSON and CSV. A Text file can hold bytes that + // are not valid UTF-8, here a Latin-1 "é" (0xE9) and "ï" (0xEF), and converting them to Arrow + // replaces each ill-formed sequence with U+FFFD. By default the join stays on Spark and keeps + // the bytes, as in 1.1.0, and the scan records why it was not converted. + assert(CometConf.COMET_CONVERT_BROADCAST_BUILD_SIDE_ENABLED.defaultValue.contains(true)) + assert(CometConf.COMET_CONVERT_FROM_TEXT_ENABLED.defaultValue.contains(false)) + withTempDir { dir => + val factPath = s"${dir.getAbsolutePath}/fact.parquet" + val dimFile = new File(dir, "dim.txt") + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(0, 20) + .selectExpr("CAST(id % 2 + 1 AS STRING) AS k", "id AS v") + .write + .parquet(factPath) + } + Files.write( + dimFile.toPath, + "1 caf".getBytes(UTF_8) ++ Array(0xe9.toByte) ++ "\n2 na".getBytes(UTF_8) ++ + Array(0xef.toByte) ++ "ve\n".getBytes(UTF_8)) + val query = + """SELECT /*+ BROADCAST(d) */ f.v, hex(d.value) AS h, length(d.value) AS n + |FROM fact_7118 f JOIN dim_7118 d ON f.k = substring(d.value, 1, 1)""".stripMargin + def hexAndLength(df: DataFrame): Set[Row] = df.select("h", "n").collect().toSet + def bridged(df: DataFrame): Boolean = + collect(df.queryExecution.executedPlan) { case s: CometSparkToColumnarExec => s }.nonEmpty + for { + aqe <- Seq("false", "true") + // V1 FileSourceScanExec, then V2 BatchScanExec (TextScan). + v1List <- Seq("parquet,text", "parquet") + } { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe, + SQLConf.USE_V1_SOURCE_LIST.key -> v1List) { + spark.read.parquet(factPath).createOrReplaceTempView("fact_7118") + spark.read.text(dimFile.getAbsolutePath).createOrReplaceTempView("dim_7118") + val ctx = s"aqe=$aqe v1=$v1List" + + checkSparkAnswerAndFallbackReason( + sql(query), + "Comet converts a Text scan on a broadcast build side to Arrow only when") + val kept = sql(query) + assert(hexAndLength(kept) == Set(Row("3120636166E9", 6), Row("32206E61EF7665", 5)), ctx) + assert(!bridged(kept), s"$ctx: the Text build side must stay on Spark") + + // With spark.comet.convert.text.enabled the build side is converted, and each + // ill-formed byte comes back as U+FFFD (EF BF BD), which changes `hex` and `length`. + // The compatibility guide quotes these values. + withSQLConf(CometConf.COMET_CONVERT_FROM_TEXT_ENABLED.key -> "true") { + val df = sql(query) + assert( + hexAndLength(df) == Set(Row("3120636166EFBFBD", 6), Row("32206E61EFBFBD7665", 7)), + ctx) + assert(bridged(df), s"$ctx: expected spark.comet.convert.text.enabled to convert") + } + } + } + } + } + + test("broadcast build side bridge is gated on broadcast-exchange conversion (#6008)") { + // The bridge is pointless when Comet broadcast exchange conversion is off: the broadcast (and + // so the join) cannot go native. With COMET_EXEC_BROADCAST_EXCHANGE_ENABLED=false the ORC + // build side must not be bridged even though broadcastBuildSide is on. + withTempDir { dir => + val probePath = s"${dir.getAbsolutePath}/probe.parquet" + val allowPath = s"${dir.getAbsolutePath}/allow.orc" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(0, 100) + .selectExpr("CAST(id AS STRING) AS key", "id AS v") + .write + .parquet(probePath) + spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.orc(allowPath) + } + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + CometConf.COMET_CONVERT_BROADCAST_BUILD_SIDE_ENABLED.key -> "true", + CometConf.COMET_EXEC_BROADCAST_EXCHANGE_ENABLED.key -> "false") { + spark.read.parquet(probePath).createOrReplaceTempView("probe_gate_6008") + spark.read.orc(allowPath).createOrReplaceTempView("allow_gate_6008") + val (_, cometPlan) = checkSparkAnswer(sql(""" + |SELECT /*+ BROADCAST(a) */ p.v + |FROM probe_gate_6008 p JOIN allow_gate_6008 a ON p.key = a.value + |""".stripMargin)) + assert( + collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, + "with broadcast-exchange conversion off the build side must not be bridged:\n" + + cometPlan.treeString) + } + } + } + + test("broadcast build side scan below an aggregate is bridged, shuffle stage is not (#6008)") { + // A file scan can sit BELOW build-side operators. Broadcasting an aggregated ORC table puts a + // ORC FileSourceScanExec under a partial aggregate + shuffle on the build side. The tag + // pre-pass walks the whole broadcast subtree, so the deep ORC scan must be bridged to Arrow, + // yet no query stage (e.g. the build-side shuffle under AQE) may be wrapped in the bridge. + withTempDir { dir => + val allowPath = s"${dir.getAbsolutePath}/allow.orc" + val probePath = s"${dir.getAbsolutePath}/probe.parquet" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark.range(0, 20).selectExpr("CAST(id % 5 AS STRING) AS value").write.orc(allowPath) + spark + .range(0, 100) + .selectExpr("CAST(id % 5 AS STRING) AS key", "id AS v") + .write + .parquet(probePath) + } + spark.read.orc(allowPath).createOrReplaceTempView("allow_agg_6008") + spark.read.parquet(probePath).createOrReplaceTempView("probe_agg_6008") + val query = """ + |SELECT /*+ BROADCAST(d) */ p.v, d.n + |FROM probe_agg_6008 p + |JOIN (SELECT value AS k, COUNT(*) AS n FROM allow_agg_6008 GROUP BY value) d + | ON p.key = d.k + |""".stripMargin + for (aqe <- Seq(false, true)) { + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString, + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> (10 * 1024 * 1024).toString) { + // The plan is fully native (the bridged ORC scan under the aggregate is a native leaf), + // so checkSparkAnswerAndOperator also pins that the ORC scan did get bridged - an + // un-bridged Spark scan would read as non-native and fail here. It is also strict-testing + // safe, unlike checkSparkAnswer on a fully-native plan. + val (_, cometPlan) = checkSparkAnswerAndOperator(sql(query)) + // `collect` descends AQE query stages, so the bridge on the deep ORC scan is reachable + // under AQE too (it lives in the build-side shuffle stage). Assert it in both modes - and + // because checkSparkAnswerAndOperator only inspects the initial plan under AQE, this is + // the only thing that checks the final AQE plan. + assert( + collect(cometPlan) { case s: CometSparkToColumnarExec => s }.nonEmpty, + s"aqe=$aqe: expected the build-side ORC scan to be bridged:\n${cometPlan.treeString}") + // A query stage (the build-side shuffle) must never be wrapped in the bridge. + collect(cometPlan) { case s: CometSparkToColumnarExec => s.child }.foreach { child => + assert( + !child.isInstanceOf[QueryStageExec], + s"aqe=$aqe: a QueryStageExec must not be bridged to Arrow:\n${cometPlan.treeString}") + } + } + } + } + } + + test("per-format opt-out still wins on a broadcast build side (#6008)") { + // A CSV build side stays a Spark FileSourceScanExec and is tagged, but the CSV arm of + // shouldApplySparkToColumnar honors COMET_CONVERT_FROM_CSV_ENABLED and never calls the + // broadcast-build-side bypass. So with convert-from-csv off, the CSV build side must NOT be + // bridged and the join must stay on Spark, even with broadcastBuildSide enabled. + withTempDir { dir => + val allowPath = s"${dir.getAbsolutePath}/allow.csv" + val probePath = s"${dir.getAbsolutePath}/probe.parquet" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.csv(allowPath) + spark + .range(0, 100) + .selectExpr("CAST(id AS STRING) AS key", "id AS v") + .write + .parquet(probePath) + } + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + CometConf.COMET_CONVERT_FROM_CSV_ENABLED.key -> "false", + CometConf.COMET_CONVERT_BROADCAST_BUILD_SIDE_ENABLED.key -> "true") { + spark.read.csv(allowPath).createOrReplaceTempView("allow_csv_6008") + spark.read.parquet(probePath).createOrReplaceTempView("probe_csv_6008") + val (_, cometPlan) = checkSparkAnswer(sql(""" + |SELECT /*+ BROADCAST(a) */ p.v + |FROM probe_csv_6008 p JOIN allow_csv_6008 a ON p.key = a._c0 + |""".stripMargin)) + assert( + collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, + "a CSV build side with convert-from-csv disabled must not be bridged:\n" + + cometPlan.treeString) + } + } + } + + test("broadcast build side with an unsupported DSv2 scan goes native (#6008)") { + // Exercises the BatchScanExec arm: an ORC scan read as DSv2 on the build side is bridged so + // the join goes native. Keep parquet on the V1 list so the probe stays a native CometScan; + // drop orc from the list so it is read as a DSv2 BatchScan/OrcScan. + withTempDir { dir => + val allowPath = s"${dir.getAbsolutePath}/allow.orc" + val probePath = s"${dir.getAbsolutePath}/probe.parquet" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.orc(allowPath) + spark + .range(0, 100) + .selectExpr("CAST(id AS STRING) AS key", "id AS v") + .write + .parquet(probePath) + } + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", + SQLConf.USE_V1_SOURCE_LIST.key -> "parquet") { + spark.read.orc(allowPath).createOrReplaceTempView("allow_v2_6008") + spark.read.parquet(probePath).createOrReplaceTempView("probe_v2_6008") + // Fully native (bridged DSv2 ORC build side + native parquet probe + native join), so use + // checkSparkAnswerAndOperator - strict-testing safe, unlike checkSparkAnswer here. + val (_, cometPlan) = checkSparkAnswerAndOperator(sql(""" + |SELECT /*+ BROADCAST(a) */ p.v + |FROM probe_v2_6008 p JOIN allow_v2_6008 a ON p.key = a.value + |""".stripMargin)) + assert( + collect(cometPlan) { case s: CometSparkToColumnarExec => s }.nonEmpty, + s"expected the DSv2 ORC build side to be bridged:\n${cometPlan.treeString}") + assert( + collect(cometPlan) { case j: CometBroadcastHashJoinExec => j }.nonEmpty, + s"expected the join to go native over the DSv2 build side:\n${cometPlan.treeString}") + } + } + } + + test("broadcast join with an unsupported probe stays on Spark (#6008)") { + // Both sides are unsupported ORC sources. The probe is not natively scannable, so the join + // cannot go native - and because the probe is not native, the build side must NOT be bridged + // either (bridging it would be useless and would break Spark's broadcast/DPP reuse). This pins + // the probe-native gate that fixed the DynamicPartitionPruningV2 regression. + withTempDir { dir => + val probePath = s"${dir.getAbsolutePath}/probe.orc" + val allowPath = s"${dir.getAbsolutePath}/allow.orc" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark.range(0, 100).selectExpr("CAST(id AS STRING) AS value").write.orc(probePath) + spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.orc(allowPath) + } + for (aqe <- Seq(false, true)) { + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) { + spark.read.orc(probePath).createOrReplaceTempView("probe_orc_6008") + spark.read.orc(allowPath).createOrReplaceTempView("allow_orc_6008") + val (_, cometPlan) = checkSparkAnswer(sql(""" + |SELECT /*+ BROADCAST(a) */ p.value + |FROM probe_orc_6008 p JOIN allow_orc_6008 a ON p.value = a.value + |""".stripMargin)) + // The non-native probe must leave the build side un-bridged (the probe-native gate). + assert( + collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, + s"aqe=$aqe: a non-native probe must leave the build side un-bridged:\n" + + cometPlan.treeString) + assert( + collect(cometPlan) { case j: CometBroadcastHashJoinExec => j }.isEmpty, + s"aqe=$aqe: the join must stay on Spark when the probe is not native:\n" + + cometPlan.treeString) + assert( + collect(cometPlan) { case b: CometBroadcastExchangeExec => b }.isEmpty, + s"aqe=$aqe: the broadcast must stay on Spark when the join does not convert:\n" + + cometPlan.treeString) + } + } + } + } + + test("broadcast build side on the left (BuildLeft) with an ORC source goes native (#6008)") { + // All other tests broadcast the right table (BuildRight). Put the small ORC table first and + // hint it so it becomes the left build side, exercising the BuildLeft arm of tagBuildIfProbeNative. + withTempDir { dir => + val probePath = s"${dir.getAbsolutePath}/probe.parquet" + val allowPath = s"${dir.getAbsolutePath}/allow.orc" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(0, 100) + .selectExpr("CAST(id AS STRING) AS key", "id AS v") + .write + .parquet(probePath) + spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.orc(allowPath) + } + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { + spark.read.orc(allowPath).createOrReplaceTempView("allow_bl_6008") + spark.read.parquet(probePath).createOrReplaceTempView("probe_bl_6008") + val (_, cometPlan) = checkSparkAnswerAndOperator(sql(""" + |SELECT /*+ BROADCAST(a) */ p.v + |FROM allow_bl_6008 a JOIN probe_bl_6008 p ON a.value = p.key + |""".stripMargin)) + assert( + collect(cometPlan) { case s: CometSparkToColumnarExec => s }.nonEmpty, + s"expected the ORC build side to be bridged:\n${cometPlan.treeString}") + assert( + collect(cometPlan) { case j: CometBroadcastHashJoinExec => j }.nonEmpty, + s"expected a CometBroadcastHashJoinExec:\n${cometPlan.treeString}") + } + } + } + + test("broadcast nested loop join build side with an ORC source goes native (#6008)") { + // Exercises the BroadcastNestedLoopJoinExec arm of tagBroadcastBuildSideLeaves: an inequality + // join forces a BNLJ; the ORC build side is bridged so the BNLJ runs natively. + withTempDir { dir => + val probePath = s"${dir.getAbsolutePath}/probe.parquet" + val allowPath = s"${dir.getAbsolutePath}/allow.orc" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark.range(0, 100).selectExpr("id AS v", "CAST(id AS INT) AS k").write.parquet(probePath) + spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.orc(allowPath) + } + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { + spark.read.orc(allowPath).createOrReplaceTempView("allow_bnlj_6008") + spark.read.parquet(probePath).createOrReplaceTempView("probe_bnlj_6008") + val (_, cometPlan) = checkSparkAnswer(sql(""" + |SELECT /*+ BROADCAST(a) */ p.v + |FROM probe_bnlj_6008 p JOIN allow_bnlj_6008 a ON p.k > CAST(a.value AS INT) + |""".stripMargin)) + assert( + collect(cometPlan) { case s: CometSparkToColumnarExec => s }.nonEmpty, + s"expected the ORC build side to be bridged:\n${cometPlan.treeString}") + assert( + collect(cometPlan) { case j: CometBroadcastNestedLoopJoinExec => j }.nonEmpty, + s"expected a CometBroadcastNestedLoopJoinExec:\n${cometPlan.treeString}") + } + } + } + + test("AQE DPP with an ORC dimension on the broadcast build side runs natively (#6008)") { + // An ORC dimension on a broadcast build side joined to a PARTITIONED Parquet fact triggers DPP. + // The build-side bridge makes the join a CometBroadcastHashJoinExec; the DPP subquery's own + // build (the ORC scan) is not bridged, so CometPlanAdaptiveDynamicPruningFilters must NOT wrap + // it in a CometBroadcastExchangeExec (that crashed with "Comet execution only takes Arrow + // Arrays"). It falls back to a Spark DPP broadcast while the join stays native. See issue #6008. + assume(isSpark35Plus, "Native AQE DPP requires Spark 3.5+") + withSQLConf( + SQLConf.USE_V1_SOURCE_LIST.key -> "parquet", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true", + SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true") { + withTempDir { dir => + val factPath = s"${dir.getAbsolutePath}/fact.parquet" + val dimPath = s"${dir.getAbsolutePath}/dim.orc" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(0, 100) + .selectExpr("CAST(id % 10 AS STRING) AS fk", "id AS v") + .write + .partitionBy("fk") + .parquet(factPath) + spark.range(0, 10).selectExpr("CAST(id AS STRING) AS value").write.orc(dimPath) + } + spark.read.parquet(factPath).createOrReplaceTempView("dpp_fact_orc") + spark.read.orc(dimPath).createOrReplaceTempView("dpp_dim_orc") + // CAST(d.value AS INT) % 2 = 0 is a filter Spark cannot infer onto f.fk, so the fact is + // pruned only via a live DPP subquery (a filter Spark can push, like d.value < '5', would + // prune the fact statically and the test would pass even with DPP disabled). + val df = sql(""" + |SELECT /*+ BROADCAST(d) */ f.v + |FROM dpp_fact_orc f JOIN dpp_dim_orc d ON f.fk = d.value + |WHERE CAST(d.value AS INT) % 2 = 0 + |""".stripMargin) + // Must not crash, and must match Spark. + val (_, cometPlan) = checkSparkAnswer(df) + // The join still goes native over the bridged build side. + assert( + collect(cometPlan) { case j: CometBroadcastHashJoinExec => j }.nonEmpty, + s"expected the join to run natively:\n${cometPlan.treeString}") + // DPP is still live - the fact scan keeps a broadcast pruning subquery (Spark-side here, + // because the unbridged ORC dimension build is not native). + val dppSubqueries = collectWithSubqueries(cometPlan) { + case s: SubqueryBroadcastExec => s + case s: CometSubqueryBroadcastExec => s + } + assert( + dppSubqueries.nonEmpty, + s"expected a live DPP broadcast subquery on the fact scan:\n${cometPlan.treeString}") + } + } + } + + test("broadcast join that stays on Spark leaves no build-side bridge (#6008)") { + // A non-equi FULL OUTER broadcast join is a BroadcastNestedLoopJoinExec whose join type Comet + // declines (getSupportLevel is Unsupported for FullOuter). The probe is native Parquet, so the + // build-side gate still tags and bridges the ORC build - but because the join cannot go + // native, clearWastedBuildSideBridges must undo that bridge after the transform, leaving the + // plan exactly as it would be without the feature: a Spark BNLJ over a Spark broadcast with no + // CometSparkToColumnarExec. Regression test for the first-round revert that never fired. #6008. + withTempDir { dir => + val probePath = s"${dir.getAbsolutePath}/probe.parquet" + val buildPath = s"${dir.getAbsolutePath}/build.orc" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark.range(0, 100).selectExpr("CAST(id AS INT) AS k", "id AS v").write.parquet(probePath) + spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.orc(buildPath) + } + for (aqe <- Seq(false, true)) { + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) { + spark.read.parquet(probePath).createOrReplaceTempView("probe_fo_6008") + spark.read.orc(buildPath).createOrReplaceTempView("build_fo_6008") + val (_, cometPlan) = checkSparkAnswer(sql(""" + |SELECT /*+ BROADCAST(a) */ p.v, a.value + |FROM probe_fo_6008 p + |FULL OUTER JOIN (SELECT value FROM build_fo_6008 WHERE LENGTH(value) > 0) a + |ON p.k > CAST(a.value AS INT) + |""".stripMargin)) + assert( + collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, + s"aqe=$aqe: a Spark-staying join must leave no build-side bridge:\n" + + cometPlan.treeString) + assert( + collect(cometPlan) { case b: CometBroadcastExchangeExec => b }.isEmpty, + s"aqe=$aqe: the broadcast must stay on Spark:\n${cometPlan.treeString}") + assert( + collect(cometPlan) { case j: BroadcastNestedLoopJoinExec => j }.nonEmpty, + s"aqe=$aqe: full outer join must stay a Spark BNLJ:\n${cometPlan.treeString}") + } + } + } + } + + test("Spark-staying broadcast join with an aggregate build side leaves no bridge (#6008)") { + // Same declined FULL OUTER BNLJ, but the ORC build side is aggregated. The bridge is inserted + // under the aggregate (and its CometSinkPlaceHolder) and then cleared. The first-round revert + // rebuilt the Spark nodes from originalPlan and crashed here with a ClassCastException; the + // tag-clear-and-retransform approach must give a clean Spark plan and the right answer. #6008. + withTempDir { dir => + val probePath = s"${dir.getAbsolutePath}/probe.parquet" + val buildPath = s"${dir.getAbsolutePath}/build.orc" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark.range(0, 100).selectExpr("CAST(id AS INT) AS k", "id AS v").write.parquet(probePath) + spark.range(0, 20).selectExpr("CAST(id % 5 AS STRING) AS value").write.orc(buildPath) + } + for (aqe <- Seq(false, true)) { + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) { + spark.read.parquet(probePath).createOrReplaceTempView("probe_foa_6008") + spark.read.orc(buildPath).createOrReplaceTempView("build_foa_6008") + val (_, cometPlan) = checkSparkAnswer(sql(""" + |SELECT /*+ BROADCAST(a) */ p.v, a.n + |FROM probe_foa_6008 p + |FULL OUTER JOIN + | (SELECT CAST(value AS INT) AS gk, COUNT(*) AS n + | FROM build_foa_6008 GROUP BY value) a + |ON p.k > a.gk + |""".stripMargin)) + assert( + collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, + s"aqe=$aqe: aggregate build side must leave no bridge:\n${cometPlan.treeString}") + assert( + collect(cometPlan) { case b: CometBroadcastExchangeExec => b }.isEmpty, + s"aqe=$aqe: the broadcast must stay on Spark:\n${cometPlan.treeString}") + } + } + } + } + + test("Spark-staying broadcast join with a union build side leaves no bridge (#6008)") { + // Same declined FULL OUTER BNLJ, but the build side is a UNION ALL of two ORC scans. The + // first-round revert rebuilt each Comet node from originalPlan and failed here with "Incorrect + // number of children"; tag-clear-and-retransform must produce a clean Spark plan. See #6008. + withTempDir { dir => + val probePath = s"${dir.getAbsolutePath}/probe.parquet" + val buildPath1 = s"${dir.getAbsolutePath}/build1.orc" + val buildPath2 = s"${dir.getAbsolutePath}/build2.orc" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark.range(0, 100).selectExpr("CAST(id AS INT) AS k", "id AS v").write.parquet(probePath) + spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.orc(buildPath1) + spark.range(5, 10).selectExpr("CAST(id AS STRING) AS value").write.orc(buildPath2) + } + for (aqe <- Seq(false, true)) { + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) { + spark.read.parquet(probePath).createOrReplaceTempView("probe_fou_6008") + spark.read.orc(buildPath1).createOrReplaceTempView("build_fou1_6008") + spark.read.orc(buildPath2).createOrReplaceTempView("build_fou2_6008") + val (_, cometPlan) = checkSparkAnswer(sql(""" + |SELECT /*+ BROADCAST(a) */ p.v, a.value + |FROM probe_fou_6008 p + |FULL OUTER JOIN + | (SELECT value FROM build_fou1_6008 UNION ALL SELECT value FROM build_fou2_6008) a + |ON p.k > CAST(a.value AS INT) + |""".stripMargin)) + assert( + collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, + s"aqe=$aqe: union build side must leave no bridge:\n${cometPlan.treeString}") + assert( + collect(cometPlan) { case b: CometBroadcastExchangeExec => b }.isEmpty, + s"aqe=$aqe: the broadcast must stay on Spark:\n${cometPlan.treeString}") + } + } + } + } + + test("a native broadcast join nested in a Spark-staying join keeps its bridge (#6008)") { + // The outer FULL OUTER BNLJ stays on Spark, but its build side is itself a broadcast hash join + // of a native Parquet table with a bridged ORC table. clearWastedBuildSideBridges must stop at + // the inner CometBroadcastExchangeExec so that native join keeps its bridge, rather than walk + // through it and revert the inner scan. Pins the stop-at-CometBroadcastExchangeExec case. + withTempDir { dir => + val probePath = s"${dir.getAbsolutePath}/probe.parquet" + val factPath = s"${dir.getAbsolutePath}/fact.parquet" + val dimPath = s"${dir.getAbsolutePath}/dim.orc" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark.range(0, 20).selectExpr("CAST(id AS INT) AS ok", "id AS x").write.parquet(probePath) + spark.range(0, 100).selectExpr("CAST(id AS INT) AS k", "id AS v").write.parquet(factPath) + spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.orc(dimPath) + } + for (aqe <- Seq(false, true)) { + withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) { + spark.read.parquet(probePath).createOrReplaceTempView("nest_outer") + spark.read.parquet(factPath).createOrReplaceTempView("nest_fact") + spark.read.orc(dimPath).createOrReplaceTempView("nest_dim") + val (_, cometPlan) = checkSparkAnswer(sql(""" + |SELECT /*+ BROADCAST(ib) */ o.x, ib.v + |FROM nest_outer o + |FULL OUTER JOIN ( + | SELECT /*+ BROADCAST(t) */ p.k AS k, p.v AS v + | FROM nest_fact p JOIN nest_dim t ON p.k = CAST(t.value AS INT) + |) ib + |ON o.ok > ib.k + |""".stripMargin)) + // Inner join runs natively and keeps its ORC bridge. + assert( + collect(cometPlan) { case j: CometBroadcastHashJoinExec => j }.nonEmpty, + s"aqe=$aqe: the inner join should be a CometBroadcastHashJoinExec:\n" + + cometPlan.treeString) + assert( + collect(cometPlan) { case s: CometSparkToColumnarExec => s }.nonEmpty, + s"aqe=$aqe: the inner join must keep its ORC bridge:\n${cometPlan.treeString}") + // Outer full outer join stays on Spark. + assert( + collect(cometPlan) { case j: BroadcastNestedLoopJoinExec => j }.nonEmpty, + s"aqe=$aqe: the outer full outer join must stay a Spark BNLJ:\n${cometPlan.treeString}") + } + } + } + } + + test("Spark 3.4 AQE DPP with a V1 Parquet fact and an ORC dimension (#6008)") { + // On Spark 3.4, CometScanRule.transformV1Scan has an "AQE Dynamic Partition Pruning requires + // Spark 3.5+" fallback that keeps a V1 scan with an AQE DPP filter on Spark. So this Parquet + // fact is not native, hasOnlyNativeScans is false, and the build side is never tagged or + // bridged. The feature stays out of the way on 3.4: DPP stays live and the answer matches + // Spark. The SKIP_COMET_BROADCAST_TAG path needs a native fact; the Iceberg test in + // CometBroadcastBuildSideSuite covers it. See #6008. + assume(!isSpark35Plus, "This test covers the Spark 3.4 V1-fact DPP fallback") + withSQLConf( + SQLConf.USE_V1_SOURCE_LIST.key -> "parquet", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true", + SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true") { + withTempDir { dir => + val factPath = s"${dir.getAbsolutePath}/fact.parquet" + val dimPath = s"${dir.getAbsolutePath}/dim.orc" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(0, 100) + .selectExpr("CAST(id % 10 AS STRING) AS fk", "id AS v") + .write + .partitionBy("fk") + .parquet(factPath) + spark.range(0, 10).selectExpr("CAST(id AS STRING) AS value").write.orc(dimPath) + } + spark.read.parquet(factPath).createOrReplaceTempView("dpp34_fact_orc") + spark.read.orc(dimPath).createOrReplaceTempView("dpp34_dim_orc") + // CAST(d.value AS INT) % 2 = 0 is a filter Spark cannot infer onto f.fk, so the fact is + // pruned only via a live DPP subquery. + val df = sql(""" + |SELECT /*+ BROADCAST(d) */ f.v + |FROM dpp34_fact_orc f JOIN dpp34_dim_orc d ON f.fk = d.value + |WHERE CAST(d.value AS INT) % 2 = 0 + |""".stripMargin) + val (_, cometPlan) = checkSparkAnswer(df) + // The non-native V1 fact means the build side is never tagged, so nothing is bridged. + assert( + collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, + s"3.4: the ORC dimension must not be bridged:\n${cometPlan.treeString}") + // DPP is still live - the fact scan keeps a broadcast pruning subquery. + val dppSubqueries = collectWithSubqueries(cometPlan) { + case s: SubqueryBroadcastExec => s + case s: CometSubqueryBroadcastExec => s + } + assert( + dppSubqueries.nonEmpty, + s"3.4: expected a live DPP broadcast subquery on the fact scan:\n${cometPlan.treeString}") + } + } + } + + test("AQE DPP reuses the broadcast for a UNION ALL dimension (#6008)") { + // The DPP subquery's build is a CometUnionExec - a columnar CometPlan, not a CometNativeExec. + // isNativeBuildSide must accept it so CometPlanAdaptiveDynamicPruningFilters converts the + // subquery's SubqueryBroadcastExec and Spark reuses the join's CometBroadcastExchangeExec as a + // ReusedExchangeExec. The earlier CometNativeExec-only check dropped the reuse. See #6008. + assume(isSpark35Plus, "Native AQE DPP requires Spark 3.5+") + withSQLConf( + SQLConf.USE_V1_SOURCE_LIST.key -> "parquet", + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true", + SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true") { + withTempDir { dir => + val factPath = s"${dir.getAbsolutePath}/fact.parquet" + val dim1Path = s"${dir.getAbsolutePath}/dim1.parquet" + val dim2Path = s"${dir.getAbsolutePath}/dim2.parquet" + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .range(0, 100) + .selectExpr("CAST(id % 10 AS STRING) AS fk", "id AS v") + .write + .partitionBy("fk") + .parquet(factPath) + spark + .range(0, 10) + .selectExpr("CAST(id AS STRING) AS value", "id AS n") + .write + .parquet(dim1Path) + spark + .range(10, 20) + .selectExpr("CAST(id AS STRING) AS value", "id AS n") + .write + .parquet(dim2Path) + } + spark.read.parquet(factPath).createOrReplaceTempView("dpp_fact_u") + spark.read.parquet(dim1Path).createOrReplaceTempView("dpp_dim1_u") + spark.read.parquet(dim2Path).createOrReplaceTempView("dpp_dim2_u") + val df = sql(""" + |SELECT /*+ BROADCAST(d) */ f.v + |FROM dpp_fact_u f JOIN + | (SELECT value FROM dpp_dim1_u WHERE n % 2 = 0 + | UNION ALL SELECT value FROM dpp_dim2_u WHERE n % 2 = 0) d + |ON f.fk = d.value + |""".stripMargin) + val (_, cometPlan) = checkSparkAnswer(df) + // The union dimension broadcast goes native. + assert( + collectWithSubqueries(cometPlan) { case b: CometBroadcastExchangeExec => b }.nonEmpty, + "expected a native CometBroadcastExchangeExec for the union dimension:\n" + + cometPlan.treeString) + // DPP reuses that broadcast rather than building a second one. + assert( + collectWithSubqueries(cometPlan) { case r: ReusedExchangeExec => r }.nonEmpty, + s"expected a ReusedExchangeExec under the DPP subquery:\n${cometPlan.treeString}") + } + } + } } diff --git a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala index 4ae13863fb5..c26cf8b7d08 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala +++ b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala @@ -39,7 +39,7 @@ import org.apache.spark.sql.comet._ import org.apache.spark.sql.comet.execution.shuffle.{CometColumnarShuffle, CometNativeShuffle, CometShuffleExchangeExec} import org.apache.spark.sql.connector.catalog.InMemoryTableCatalog import org.apache.spark.sql.execution._ -import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AQEShuffleReadExec, BroadcastQueryStageExec, LogicalQueryStage, QueryStageExec} +import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AQEShuffleReadExec, BroadcastQueryStageExec, LogicalQueryStage} import org.apache.spark.sql.execution.columnar.{CometInMemoryRelationHelper, InMemoryTableScanExec} import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat import org.apache.spark.sql.execution.exchange.{BroadcastExchangeExec, BroadcastExchangeLike, ReusedExchangeExec, ShuffleExchangeExec, ShuffleExchangeLike} @@ -307,638 +307,6 @@ class CometExecSuite extends CometTestBase { } } - test("broadcast build side with unsupported text source goes native (#6008)") { - // A tiny lookup table read from a Text file (which Comet cannot scan natively) sits on the - // build side of a broadcast join over a large native probe. Without the auto-bridge the whole - // join stays on Spark. With spark.comet.convert.broadcastBuildSide.enabled (default on) - // the text leaf is bridged to Arrow so the broadcast and the join run natively. - assert(CometConf.COMET_CONVERT_BROADCAST_BUILD_SIDE_ENABLED.defaultValue.contains(true)) - withTempDir { dir => - val probePath = s"${dir.getAbsolutePath}/probe.parquet" - val allowPath = s"${dir.getAbsolutePath}/allow.txt" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark - .range(0, 100) - .selectExpr("CAST(id AS STRING) AS key", "id AS v") - .write - .parquet(probePath) - spark - .range(0, 5) - .selectExpr("CAST(id AS STRING) AS value") - .write - .text(allowPath) - } - - // Cover both the AQE and non-AQE planning paths - the tag pre-pass runs per stage under AQE. - for (aqe <- Seq(false, true)) { - withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) { - spark.read.parquet(probePath).createOrReplaceTempView("probe_6008") - spark.read.text(allowPath).createOrReplaceTempView("allow_6008") - val query = - "SELECT /*+ BROADCAST(a) */ p.v FROM probe_6008 p JOIN allow_6008 a ON p.key = a.value" - - // Default: the text build side is bridged to Arrow and the join goes native. The plan is - // fully native (the bridge counts as a native leaf), so use checkSparkAnswerAndOperator - - // checkSparkAnswer would fail under COMET_STRICT_TESTING on a fully-native plan. - val (_, cometPlan) = checkSparkAnswerAndOperator(sql(query)) - assert( - collect(cometPlan) { case s: CometSparkToColumnarExec => s }.nonEmpty, - s"aqe=$aqe: expected a CometSparkToColumnarExec on the text build side:\n" + - cometPlan.treeString) - assert( - collect(cometPlan) { case b: CometBroadcastExchangeExec => b }.nonEmpty, - s"aqe=$aqe: expected a CometBroadcastExchangeExec:\n${cometPlan.treeString}") - assert( - collect(cometPlan) { case j: CometBroadcastHashJoinExec => j }.nonEmpty, - s"aqe=$aqe: expected a CometBroadcastHashJoinExec:\n${cometPlan.treeString}") - - // Off: the unsupported text leaf blocks native execution of the broadcast and the join. - withSQLConf(CometConf.COMET_CONVERT_BROADCAST_BUILD_SIDE_ENABLED.key -> "false") { - val (_, cometPlanOff) = checkSparkAnswer(sql(query)) - assert( - collect(cometPlanOff) { case s: CometSparkToColumnarExec => s }.isEmpty, - s"aqe=$aqe: expected no CometSparkToColumnarExec when disabled:\n" + - cometPlanOff.treeString) - assert( - collect(cometPlanOff) { case b: CometBroadcastExchangeExec => b }.isEmpty, - s"aqe=$aqe: expected no CometBroadcastExchangeExec when disabled:\n" + - cometPlanOff.treeString) - assert( - collect(cometPlanOff) { case j: CometBroadcastHashJoinExec => j }.isEmpty, - s"aqe=$aqe: expected no CometBroadcastHashJoinExec when disabled:\n" + - cometPlanOff.treeString) - } - } - } - } - } - - test("broadcast build side bridge is gated on broadcast-exchange conversion (#6008)") { - // The bridge is pointless when Comet broadcast exchange conversion is off: the broadcast (and - // so the join) cannot go native. With COMET_EXEC_BROADCAST_EXCHANGE_ENABLED=false the text - // build side must not be bridged even though broadcastBuildSide is on. - withTempDir { dir => - val probePath = s"${dir.getAbsolutePath}/probe.parquet" - val allowPath = s"${dir.getAbsolutePath}/allow.txt" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark - .range(0, 100) - .selectExpr("CAST(id AS STRING) AS key", "id AS v") - .write - .parquet(probePath) - spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.text(allowPath) - } - withSQLConf( - SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", - CometConf.COMET_CONVERT_BROADCAST_BUILD_SIDE_ENABLED.key -> "true", - CometConf.COMET_EXEC_BROADCAST_EXCHANGE_ENABLED.key -> "false") { - spark.read.parquet(probePath).createOrReplaceTempView("probe_gate_6008") - spark.read.text(allowPath).createOrReplaceTempView("allow_gate_6008") - val (_, cometPlan) = checkSparkAnswer(sql(""" - |SELECT /*+ BROADCAST(a) */ p.v - |FROM probe_gate_6008 p JOIN allow_gate_6008 a ON p.key = a.value - |""".stripMargin)) - assert( - collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, - "with broadcast-exchange conversion off the build side must not be bridged:\n" + - cometPlan.treeString) - } - } - } - - test("broadcast build side scan below an aggregate is bridged, shuffle stage is not (#6008)") { - // A file scan can sit BELOW build-side operators. Broadcasting an aggregated Text table puts a - // Text FileSourceScanExec under a partial aggregate + shuffle on the build side. The tag - // pre-pass walks the whole broadcast subtree, so the deep text scan must be bridged to Arrow, - // yet no query stage (e.g. the build-side shuffle under AQE) may be wrapped in the bridge. - withTempDir { dir => - val allowPath = s"${dir.getAbsolutePath}/allow.txt" - val probePath = s"${dir.getAbsolutePath}/probe.parquet" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark.range(0, 20).selectExpr("CAST(id % 5 AS STRING) AS value").write.text(allowPath) - spark - .range(0, 100) - .selectExpr("CAST(id % 5 AS STRING) AS key", "id AS v") - .write - .parquet(probePath) - } - spark.read.text(allowPath).createOrReplaceTempView("allow_agg_6008") - spark.read.parquet(probePath).createOrReplaceTempView("probe_agg_6008") - val query = """ - |SELECT /*+ BROADCAST(d) */ p.v, d.n - |FROM probe_agg_6008 p - |JOIN (SELECT value AS k, COUNT(*) AS n FROM allow_agg_6008 GROUP BY value) d - | ON p.key = d.k - |""".stripMargin - for (aqe <- Seq(false, true)) { - withSQLConf( - SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString, - SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> (10 * 1024 * 1024).toString) { - // The plan is fully native (the bridged text scan under the aggregate is a native leaf), - // so checkSparkAnswerAndOperator also pins that the text scan did get bridged - an - // un-bridged Spark scan would read as non-native and fail here. It is also strict-testing - // safe, unlike checkSparkAnswer on a fully-native plan. - val (_, cometPlan) = checkSparkAnswerAndOperator(sql(query)) - // `collect` descends AQE query stages, so the bridge on the deep text scan is reachable - // under AQE too (it lives in the build-side shuffle stage). Assert it in both modes - and - // because checkSparkAnswerAndOperator only inspects the initial plan under AQE, this is - // the only thing that checks the final AQE plan. - assert( - collect(cometPlan) { case s: CometSparkToColumnarExec => s }.nonEmpty, - s"aqe=$aqe: expected the build-side text scan to be bridged:\n${cometPlan.treeString}") - // A query stage (the build-side shuffle) must never be wrapped in the bridge. - collect(cometPlan) { case s: CometSparkToColumnarExec => s.child }.foreach { child => - assert( - !child.isInstanceOf[QueryStageExec], - s"aqe=$aqe: a QueryStageExec must not be bridged to Arrow:\n${cometPlan.treeString}") - } - } - } - } - } - - test("per-format opt-out still wins on a broadcast build side (#6008)") { - // A CSV build side stays a Spark FileSourceScanExec and is tagged, but the CSV arm of - // shouldApplySparkToColumnar honors COMET_CONVERT_FROM_CSV_ENABLED and never calls the - // broadcast-build-side bypass. So with convert-from-csv off, the CSV build side must NOT be - // bridged and the join must stay on Spark, even with broadcastBuildSide enabled. - withTempDir { dir => - val allowPath = s"${dir.getAbsolutePath}/allow.csv" - val probePath = s"${dir.getAbsolutePath}/probe.parquet" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.csv(allowPath) - spark - .range(0, 100) - .selectExpr("CAST(id AS STRING) AS key", "id AS v") - .write - .parquet(probePath) - } - withSQLConf( - SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", - CometConf.COMET_CONVERT_FROM_CSV_ENABLED.key -> "false", - CometConf.COMET_CONVERT_BROADCAST_BUILD_SIDE_ENABLED.key -> "true") { - spark.read.csv(allowPath).createOrReplaceTempView("allow_csv_6008") - spark.read.parquet(probePath).createOrReplaceTempView("probe_csv_6008") - val (_, cometPlan) = checkSparkAnswer(sql(""" - |SELECT /*+ BROADCAST(a) */ p.v - |FROM probe_csv_6008 p JOIN allow_csv_6008 a ON p.key = a._c0 - |""".stripMargin)) - assert( - collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, - "a CSV build side with convert-from-csv disabled must not be bridged:\n" + - cometPlan.treeString) - } - } - } - - test("broadcast build side with an unsupported DSv2 scan goes native (#6008)") { - // Exercises the BatchScanExec arm: an unsupported-format scan (Text read as DSv2) on the build - // side is bridged so the join goes native. Keep parquet on the V1 list so the probe stays a - // native CometScan; drop text from the list so it is read as a DSv2 BatchScan/TextScan. - withTempDir { dir => - val allowPath = s"${dir.getAbsolutePath}/allow.txt" - val probePath = s"${dir.getAbsolutePath}/probe.parquet" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.text(allowPath) - spark - .range(0, 100) - .selectExpr("CAST(id AS STRING) AS key", "id AS v") - .write - .parquet(probePath) - } - withSQLConf( - SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false", - SQLConf.USE_V1_SOURCE_LIST.key -> "parquet") { - spark.read.text(allowPath).createOrReplaceTempView("allow_v2_6008") - spark.read.parquet(probePath).createOrReplaceTempView("probe_v2_6008") - // Fully native (bridged DSv2 text build side + native parquet probe + native join), so use - // checkSparkAnswerAndOperator - strict-testing safe, unlike checkSparkAnswer here. - val (_, cometPlan) = checkSparkAnswerAndOperator(sql(""" - |SELECT /*+ BROADCAST(a) */ p.v - |FROM probe_v2_6008 p JOIN allow_v2_6008 a ON p.key = a.value - |""".stripMargin)) - assert( - collect(cometPlan) { case s: CometSparkToColumnarExec => s }.nonEmpty, - s"expected the DSv2 text build side to be bridged:\n${cometPlan.treeString}") - assert( - collect(cometPlan) { case j: CometBroadcastHashJoinExec => j }.nonEmpty, - s"expected the join to go native over the DSv2 build side:\n${cometPlan.treeString}") - } - } - } - - test("broadcast join with an unsupported probe stays on Spark (#6008)") { - // Both sides are unsupported Text sources. The probe is not natively scannable, so the join - // cannot go native - and because the probe is not native, the build side must NOT be bridged - // either (bridging it would be useless and would break Spark's broadcast/DPP reuse). This pins - // the probe-native gate that fixed the DynamicPartitionPruningV2 regression. - withTempDir { dir => - val probePath = s"${dir.getAbsolutePath}/probe.txt" - val allowPath = s"${dir.getAbsolutePath}/allow.txt" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark.range(0, 100).selectExpr("CAST(id AS STRING) AS value").write.text(probePath) - spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.text(allowPath) - } - for (aqe <- Seq(false, true)) { - withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) { - spark.read.text(probePath).createOrReplaceTempView("probe_txt_6008") - spark.read.text(allowPath).createOrReplaceTempView("allow_txt_6008") - val (_, cometPlan) = checkSparkAnswer(sql(""" - |SELECT /*+ BROADCAST(a) */ p.value - |FROM probe_txt_6008 p JOIN allow_txt_6008 a ON p.value = a.value - |""".stripMargin)) - // The non-native probe must leave the build side un-bridged (the probe-native gate). - assert( - collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, - s"aqe=$aqe: a non-native probe must leave the build side un-bridged:\n" + - cometPlan.treeString) - assert( - collect(cometPlan) { case j: CometBroadcastHashJoinExec => j }.isEmpty, - s"aqe=$aqe: the join must stay on Spark when the probe is not native:\n" + - cometPlan.treeString) - assert( - collect(cometPlan) { case b: CometBroadcastExchangeExec => b }.isEmpty, - s"aqe=$aqe: the broadcast must stay on Spark when the join does not convert:\n" + - cometPlan.treeString) - } - } - } - } - - test("broadcast build side on the left (BuildLeft) with a text source goes native (#6008)") { - // All other tests broadcast the right table (BuildRight). Put the small text table first and - // hint it so it becomes the left build side, exercising the BuildLeft arm of tagBuildIfProbeNative. - withTempDir { dir => - val probePath = s"${dir.getAbsolutePath}/probe.parquet" - val allowPath = s"${dir.getAbsolutePath}/allow.txt" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark - .range(0, 100) - .selectExpr("CAST(id AS STRING) AS key", "id AS v") - .write - .parquet(probePath) - spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.text(allowPath) - } - withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { - spark.read.text(allowPath).createOrReplaceTempView("allow_bl_6008") - spark.read.parquet(probePath).createOrReplaceTempView("probe_bl_6008") - val (_, cometPlan) = checkSparkAnswerAndOperator(sql(""" - |SELECT /*+ BROADCAST(a) */ p.v - |FROM allow_bl_6008 a JOIN probe_bl_6008 p ON a.value = p.key - |""".stripMargin)) - assert( - collect(cometPlan) { case s: CometSparkToColumnarExec => s }.nonEmpty, - s"expected the text build side to be bridged:\n${cometPlan.treeString}") - assert( - collect(cometPlan) { case j: CometBroadcastHashJoinExec => j }.nonEmpty, - s"expected a CometBroadcastHashJoinExec:\n${cometPlan.treeString}") - } - } - } - - test("broadcast nested loop join build side with a text source goes native (#6008)") { - // Exercises the BroadcastNestedLoopJoinExec arm of tagBroadcastBuildSideLeaves: an inequality - // join forces a BNLJ; the text build side is bridged so the BNLJ runs natively. - withTempDir { dir => - val probePath = s"${dir.getAbsolutePath}/probe.parquet" - val allowPath = s"${dir.getAbsolutePath}/allow.txt" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark.range(0, 100).selectExpr("id AS v", "CAST(id AS INT) AS k").write.parquet(probePath) - spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.text(allowPath) - } - withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") { - spark.read.text(allowPath).createOrReplaceTempView("allow_bnlj_6008") - spark.read.parquet(probePath).createOrReplaceTempView("probe_bnlj_6008") - val (_, cometPlan) = checkSparkAnswer(sql(""" - |SELECT /*+ BROADCAST(a) */ p.v - |FROM probe_bnlj_6008 p JOIN allow_bnlj_6008 a ON p.k > CAST(a.value AS INT) - |""".stripMargin)) - assert( - collect(cometPlan) { case s: CometSparkToColumnarExec => s }.nonEmpty, - s"expected the text build side to be bridged:\n${cometPlan.treeString}") - assert( - collect(cometPlan) { case j: CometBroadcastNestedLoopJoinExec => j }.nonEmpty, - s"expected a CometBroadcastNestedLoopJoinExec:\n${cometPlan.treeString}") - } - } - } - - test("AQE DPP with a text dimension on the broadcast build side runs natively (#6008)") { - // A Text dimension on a broadcast build side joined to a PARTITIONED Parquet fact triggers DPP. - // The build-side bridge makes the join a CometBroadcastHashJoinExec; the DPP subquery's own - // build (the text scan) is not bridged, so CometPlanAdaptiveDynamicPruningFilters must NOT wrap - // it in a CometBroadcastExchangeExec (that crashed with "Comet execution only takes Arrow - // Arrays"). It falls back to a Spark DPP broadcast while the join stays native. See issue #6008. - assume(isSpark35Plus, "Native AQE DPP requires Spark 3.5+") - withSQLConf( - SQLConf.USE_V1_SOURCE_LIST.key -> "parquet", - SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", - SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true", - SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true") { - withTempDir { dir => - val factPath = s"${dir.getAbsolutePath}/fact.parquet" - val dimPath = s"${dir.getAbsolutePath}/dim.txt" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark - .range(0, 100) - .selectExpr("CAST(id % 10 AS STRING) AS fk", "id AS v") - .write - .partitionBy("fk") - .parquet(factPath) - spark.range(0, 10).selectExpr("CAST(id AS STRING) AS value").write.text(dimPath) - } - spark.read.parquet(factPath).createOrReplaceTempView("dpp_fact_txt") - spark.read.text(dimPath).createOrReplaceTempView("dpp_dim_txt") - // CAST(d.value AS INT) % 2 = 0 is a filter Spark cannot infer onto f.fk, so the fact is - // pruned only via a live DPP subquery (a filter Spark can push, like d.value < '5', would - // prune the fact statically and the test would pass even with DPP disabled). - val df = sql(""" - |SELECT /*+ BROADCAST(d) */ f.v - |FROM dpp_fact_txt f JOIN dpp_dim_txt d ON f.fk = d.value - |WHERE CAST(d.value AS INT) % 2 = 0 - |""".stripMargin) - // Must not crash, and must match Spark. - val (_, cometPlan) = checkSparkAnswer(df) - // The join still goes native over the bridged build side. - assert( - collect(cometPlan) { case j: CometBroadcastHashJoinExec => j }.nonEmpty, - s"expected the join to run natively:\n${cometPlan.treeString}") - // DPP is still live - the fact scan keeps a broadcast pruning subquery (Spark-side here, - // because the unbridged text dimension build is not native). - val dppSubqueries = collectWithSubqueries(cometPlan) { - case s: SubqueryBroadcastExec => s - case s: CometSubqueryBroadcastExec => s - } - assert( - dppSubqueries.nonEmpty, - s"expected a live DPP broadcast subquery on the fact scan:\n${cometPlan.treeString}") - } - } - } - - test("broadcast join that stays on Spark leaves no build-side bridge (#6008)") { - // A non-equi FULL OUTER broadcast join is a BroadcastNestedLoopJoinExec whose join type Comet - // declines (getSupportLevel is Unsupported for FullOuter). The probe is native Parquet, so the - // build-side gate still tags and bridges the text build - but because the join cannot go - // native, clearWastedBuildSideBridges must undo that bridge after the transform, leaving the - // plan exactly as it would be without the feature: a Spark BNLJ over a Spark broadcast with no - // CometSparkToColumnarExec. Regression test for the first-round revert that never fired. #6008. - withTempDir { dir => - val probePath = s"${dir.getAbsolutePath}/probe.parquet" - val buildPath = s"${dir.getAbsolutePath}/build.txt" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark.range(0, 100).selectExpr("CAST(id AS INT) AS k", "id AS v").write.parquet(probePath) - spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.text(buildPath) - } - for (aqe <- Seq(false, true)) { - withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) { - spark.read.parquet(probePath).createOrReplaceTempView("probe_fo_6008") - spark.read.text(buildPath).createOrReplaceTempView("build_fo_6008") - val (_, cometPlan) = checkSparkAnswer(sql(""" - |SELECT /*+ BROADCAST(a) */ p.v, a.value - |FROM probe_fo_6008 p - |FULL OUTER JOIN (SELECT value FROM build_fo_6008 WHERE LENGTH(value) > 0) a - |ON p.k > CAST(a.value AS INT) - |""".stripMargin)) - assert( - collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, - s"aqe=$aqe: a Spark-staying join must leave no build-side bridge:\n" + - cometPlan.treeString) - assert( - collect(cometPlan) { case b: CometBroadcastExchangeExec => b }.isEmpty, - s"aqe=$aqe: the broadcast must stay on Spark:\n${cometPlan.treeString}") - assert( - collect(cometPlan) { case j: BroadcastNestedLoopJoinExec => j }.nonEmpty, - s"aqe=$aqe: full outer join must stay a Spark BNLJ:\n${cometPlan.treeString}") - } - } - } - } - - test("Spark-staying broadcast join with an aggregate build side leaves no bridge (#6008)") { - // Same declined FULL OUTER BNLJ, but the text build side is aggregated. The bridge is inserted - // under the aggregate (and its CometSinkPlaceHolder) and then cleared. The first-round revert - // rebuilt the Spark nodes from originalPlan and crashed here with a ClassCastException; the - // tag-clear-and-retransform approach must give a clean Spark plan and the right answer. #6008. - withTempDir { dir => - val probePath = s"${dir.getAbsolutePath}/probe.parquet" - val buildPath = s"${dir.getAbsolutePath}/build.txt" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark.range(0, 100).selectExpr("CAST(id AS INT) AS k", "id AS v").write.parquet(probePath) - spark.range(0, 20).selectExpr("CAST(id % 5 AS STRING) AS value").write.text(buildPath) - } - for (aqe <- Seq(false, true)) { - withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) { - spark.read.parquet(probePath).createOrReplaceTempView("probe_foa_6008") - spark.read.text(buildPath).createOrReplaceTempView("build_foa_6008") - val (_, cometPlan) = checkSparkAnswer(sql(""" - |SELECT /*+ BROADCAST(a) */ p.v, a.n - |FROM probe_foa_6008 p - |FULL OUTER JOIN - | (SELECT CAST(value AS INT) AS gk, COUNT(*) AS n - | FROM build_foa_6008 GROUP BY value) a - |ON p.k > a.gk - |""".stripMargin)) - assert( - collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, - s"aqe=$aqe: aggregate build side must leave no bridge:\n${cometPlan.treeString}") - assert( - collect(cometPlan) { case b: CometBroadcastExchangeExec => b }.isEmpty, - s"aqe=$aqe: the broadcast must stay on Spark:\n${cometPlan.treeString}") - } - } - } - } - - test("Spark-staying broadcast join with a union build side leaves no bridge (#6008)") { - // Same declined FULL OUTER BNLJ, but the build side is a UNION ALL of two text scans. The - // first-round revert rebuilt each Comet node from originalPlan and failed here with "Incorrect - // number of children"; tag-clear-and-retransform must produce a clean Spark plan. See #6008. - withTempDir { dir => - val probePath = s"${dir.getAbsolutePath}/probe.parquet" - val buildPath1 = s"${dir.getAbsolutePath}/build1.txt" - val buildPath2 = s"${dir.getAbsolutePath}/build2.txt" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark.range(0, 100).selectExpr("CAST(id AS INT) AS k", "id AS v").write.parquet(probePath) - spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.text(buildPath1) - spark.range(5, 10).selectExpr("CAST(id AS STRING) AS value").write.text(buildPath2) - } - for (aqe <- Seq(false, true)) { - withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) { - spark.read.parquet(probePath).createOrReplaceTempView("probe_fou_6008") - spark.read.text(buildPath1).createOrReplaceTempView("build_fou1_6008") - spark.read.text(buildPath2).createOrReplaceTempView("build_fou2_6008") - val (_, cometPlan) = checkSparkAnswer(sql(""" - |SELECT /*+ BROADCAST(a) */ p.v, a.value - |FROM probe_fou_6008 p - |FULL OUTER JOIN - | (SELECT value FROM build_fou1_6008 UNION ALL SELECT value FROM build_fou2_6008) a - |ON p.k > CAST(a.value AS INT) - |""".stripMargin)) - assert( - collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, - s"aqe=$aqe: union build side must leave no bridge:\n${cometPlan.treeString}") - assert( - collect(cometPlan) { case b: CometBroadcastExchangeExec => b }.isEmpty, - s"aqe=$aqe: the broadcast must stay on Spark:\n${cometPlan.treeString}") - } - } - } - } - - test("a native broadcast join nested in a Spark-staying join keeps its bridge (#6008)") { - // The outer FULL OUTER BNLJ stays on Spark, but its build side is itself a broadcast hash join - // of a native Parquet table with a bridged text table. clearWastedBuildSideBridges must stop at - // the inner CometBroadcastExchangeExec so that native join keeps its bridge, rather than walk - // through it and revert the inner scan. Pins the stop-at-CometBroadcastExchangeExec case. - withTempDir { dir => - val probePath = s"${dir.getAbsolutePath}/probe.parquet" - val factPath = s"${dir.getAbsolutePath}/fact.parquet" - val dimPath = s"${dir.getAbsolutePath}/dim.txt" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark.range(0, 20).selectExpr("CAST(id AS INT) AS ok", "id AS x").write.parquet(probePath) - spark.range(0, 100).selectExpr("CAST(id AS INT) AS k", "id AS v").write.parquet(factPath) - spark.range(0, 5).selectExpr("CAST(id AS STRING) AS value").write.text(dimPath) - } - for (aqe <- Seq(false, true)) { - withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe.toString) { - spark.read.parquet(probePath).createOrReplaceTempView("nest_outer") - spark.read.parquet(factPath).createOrReplaceTempView("nest_fact") - spark.read.text(dimPath).createOrReplaceTempView("nest_dim") - val (_, cometPlan) = checkSparkAnswer(sql(""" - |SELECT /*+ BROADCAST(ib) */ o.x, ib.v - |FROM nest_outer o - |FULL OUTER JOIN ( - | SELECT /*+ BROADCAST(t) */ p.k AS k, p.v AS v - | FROM nest_fact p JOIN nest_dim t ON p.k = CAST(t.value AS INT) - |) ib - |ON o.ok > ib.k - |""".stripMargin)) - // Inner join runs natively and keeps its text bridge. - assert( - collect(cometPlan) { case j: CometBroadcastHashJoinExec => j }.nonEmpty, - s"aqe=$aqe: the inner join should be a CometBroadcastHashJoinExec:\n" + - cometPlan.treeString) - assert( - collect(cometPlan) { case s: CometSparkToColumnarExec => s }.nonEmpty, - s"aqe=$aqe: the inner join must keep its text bridge:\n${cometPlan.treeString}") - // Outer full outer join stays on Spark. - assert( - collect(cometPlan) { case j: BroadcastNestedLoopJoinExec => j }.nonEmpty, - s"aqe=$aqe: the outer full outer join must stay a Spark BNLJ:\n${cometPlan.treeString}") - } - } - } - } - - test("Spark 3.4 AQE DPP with a V1 Parquet fact and a text dimension (#6008)") { - // On Spark 3.4, CometScanRule.transformV1Scan has an "AQE Dynamic Partition Pruning requires - // Spark 3.5+" fallback that keeps a V1 scan with an AQE DPP filter on Spark. So this Parquet - // fact is not native, hasOnlyNativeScans is false, and the build side is never tagged or - // bridged. The feature stays out of the way on 3.4: DPP stays live and the answer matches - // Spark. The SKIP_COMET_BROADCAST_TAG path needs a native fact; the Iceberg test in - // CometBroadcastBuildSideSuite covers it. See #6008. - assume(!isSpark35Plus, "This test covers the Spark 3.4 V1-fact DPP fallback") - withSQLConf( - SQLConf.USE_V1_SOURCE_LIST.key -> "parquet", - SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", - SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true", - SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true") { - withTempDir { dir => - val factPath = s"${dir.getAbsolutePath}/fact.parquet" - val dimPath = s"${dir.getAbsolutePath}/dim.txt" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark - .range(0, 100) - .selectExpr("CAST(id % 10 AS STRING) AS fk", "id AS v") - .write - .partitionBy("fk") - .parquet(factPath) - spark.range(0, 10).selectExpr("CAST(id AS STRING) AS value").write.text(dimPath) - } - spark.read.parquet(factPath).createOrReplaceTempView("dpp34_fact_txt") - spark.read.text(dimPath).createOrReplaceTempView("dpp34_dim_txt") - // CAST(d.value AS INT) % 2 = 0 is a filter Spark cannot infer onto f.fk, so the fact is - // pruned only via a live DPP subquery. - val df = sql(""" - |SELECT /*+ BROADCAST(d) */ f.v - |FROM dpp34_fact_txt f JOIN dpp34_dim_txt d ON f.fk = d.value - |WHERE CAST(d.value AS INT) % 2 = 0 - |""".stripMargin) - val (_, cometPlan) = checkSparkAnswer(df) - // The non-native V1 fact means the build side is never tagged, so nothing is bridged. - assert( - collect(cometPlan) { case s: CometSparkToColumnarExec => s }.isEmpty, - s"3.4: the text dimension must not be bridged:\n${cometPlan.treeString}") - // DPP is still live - the fact scan keeps a broadcast pruning subquery. - val dppSubqueries = collectWithSubqueries(cometPlan) { - case s: SubqueryBroadcastExec => s - case s: CometSubqueryBroadcastExec => s - } - assert( - dppSubqueries.nonEmpty, - s"3.4: expected a live DPP broadcast subquery on the fact scan:\n${cometPlan.treeString}") - } - } - } - - test("AQE DPP reuses the broadcast for a UNION ALL dimension (#6008)") { - // The DPP subquery's build is a CometUnionExec - a columnar CometPlan, not a CometNativeExec. - // isNativeBuildSide must accept it so CometPlanAdaptiveDynamicPruningFilters converts the - // subquery's SubqueryBroadcastExec and Spark reuses the join's CometBroadcastExchangeExec as a - // ReusedExchangeExec. The earlier CometNativeExec-only check dropped the reuse. See #6008. - assume(isSpark35Plus, "Native AQE DPP requires Spark 3.5+") - withSQLConf( - SQLConf.USE_V1_SOURCE_LIST.key -> "parquet", - SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", - SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true", - SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true") { - withTempDir { dir => - val factPath = s"${dir.getAbsolutePath}/fact.parquet" - val dim1Path = s"${dir.getAbsolutePath}/dim1.parquet" - val dim2Path = s"${dir.getAbsolutePath}/dim2.parquet" - withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark - .range(0, 100) - .selectExpr("CAST(id % 10 AS STRING) AS fk", "id AS v") - .write - .partitionBy("fk") - .parquet(factPath) - spark - .range(0, 10) - .selectExpr("CAST(id AS STRING) AS value", "id AS n") - .write - .parquet(dim1Path) - spark - .range(10, 20) - .selectExpr("CAST(id AS STRING) AS value", "id AS n") - .write - .parquet(dim2Path) - } - spark.read.parquet(factPath).createOrReplaceTempView("dpp_fact_u") - spark.read.parquet(dim1Path).createOrReplaceTempView("dpp_dim1_u") - spark.read.parquet(dim2Path).createOrReplaceTempView("dpp_dim2_u") - val df = sql(""" - |SELECT /*+ BROADCAST(d) */ f.v - |FROM dpp_fact_u f JOIN - | (SELECT value FROM dpp_dim1_u WHERE n % 2 = 0 - | UNION ALL SELECT value FROM dpp_dim2_u WHERE n % 2 = 0) d - |ON f.fk = d.value - |""".stripMargin) - val (_, cometPlan) = checkSparkAnswer(df) - // The union dimension broadcast goes native. - assert( - collectWithSubqueries(cometPlan) { case b: CometBroadcastExchangeExec => b }.nonEmpty, - "expected a native CometBroadcastExchangeExec for the union dimension:\n" + - cometPlan.treeString) - // DPP reuses that broadcast rather than building a second one. - assert( - collectWithSubqueries(cometPlan) { case r: ReusedExchangeExec => r }.nonEmpty, - s"expected a ReusedExchangeExec under the DPP subquery:\n${cometPlan.treeString}") - } - } - } - test("non-AQE DPP: BHJ works with CometNativeScanExec") { withTempDir { path => val factPath = s"${path.getAbsolutePath}/fact.parquet" @@ -5701,6 +5069,33 @@ class CometExecSuite extends CometTestBase { } } + test("read Text file") { + withTempPath { dir => + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark.range(0, 10).selectExpr("CAST(id AS STRING)").write.text(dir.toString) + } + Seq("", "text").foreach { v1List => + withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> v1List) { + spark.read.text(dir.toString).createOrReplaceTempView("tbl") + val query = "SELECT CAST(value AS INT) + 1 FROM tbl" + // Off by default, because a Text file can hold bytes that are not valid UTF-8. + val (_, plan) = checkSparkAnswer(query) + assert(collect(plan) { case c: CometSparkToColumnarExec => c }.isEmpty) + withSQLConf(CometConf.COMET_CONVERT_FROM_TEXT_ENABLED.key -> "true") { + checkSparkAnswerAndOperator(query) + } + // The general switch still converts Text, as it did before Text had a config (deprecated). + val operatorList = CometConf.COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST.key + withSQLConf( + CometConf.COMET_SPARK_TO_ARROW_ENABLED.key -> "true", + operatorList -> "FileSourceScan,BatchScan") { + checkSparkAnswerAndOperator(query) + } + } + } + } + } + test("Supported file formats for CometScanExec") { assert(CometScanExec.isFileFormatSupported(new ParquetFileFormat()))