From fbbf7683d79a80cd0f2d7f184f7fa652d105d068 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Fri, 23 Oct 2026 22:34:11 +0000 Subject: [PATCH] fix: keep Text build sides on Spark by default and add Text conversion configs (#7123) * fix: default broadcast build side conversion to off and document invalid UTF-8 cases Converting a Text scan on a broadcast build side to Arrow decodes invalid UTF-8 to U+FFFD, so 1.2.0 could return different results from 1.1.0. Default spark.comet.convert.broadcastBuildSide.enabled to false and name the Text-source and CAST(binary AS STRING) cases in the compatibility guide. * test: move the #6008 build-side tests into CometBroadcastBuildSideSuite CometBroadcastBuildSideSuite turns the build-side bridge on for every test, so the tests no longer need the bridgeTest helper, and a test added later cannot miss the setting. The two tests that failed because they were not wrapped in bridgeTest now run with the bridge on, and the five that asserted nothing was bridged now exercise the bridge again. The test bodies are unchanged. * fix: keep Text build sides on Spark by default and add Text conversion configs Converting a Text scan to Arrow replaces bytes that are not valid UTF-8 with U+FFFD, so bridging a Text build side by default (#6786) could return different results from 1.1.0 (#7118). The other formats that reach the bridge either define strings as UTF-8 (ORC, Avro, Iceberg) or are decoded by Spark, so spark.comet.convert.broadcastBuildSide.enabled goes back to on by default, and Text gets its own configs: - spark.comet.convert.broadcastBuildSide.text.enabled (default false) bridges a Text build side. Without it, the scan records a fallback reason that names the config. - spark.comet.convert.text.enabled (default false) converts every Text scan, like the CSV, JSON and Parquet configs. The sparkToColumnar operator list still converts Text as before. The compatibility guide now names Text rather than CSV and JSON, no longer says that Spark requires Parquet strings to be valid UTF-8, adds Comet columnar shuffle as a path, and gives the configs that keep min/max over CAST(binary AS STRING) in Spark. * test: drop the QueryStageExec import that the test move left unused * fix: convert a Text build side only with spark.comet.convert.text.enabled Drop spark.comet.convert.broadcastBuildSide.text.enabled. A Text scan, on a build side or not, is now converted only by spark.comet.convert.text.enabled, like CSV, JSON and Parquet, and the build-side fallback reason names that config. Text goes through isConversionEnabled, so naming FileSourceScan or BatchScan in spark.comet.sparkToColumnar.supportedOperatorList still converts it but is deprecated. The #6008 build-side tests read ORC instead of Text, since a global Text switch would convert their probes and no-bridge assertions too. * docs: describe Text conversion through its own config only Update the Text, build-side and compatibility sections for the removal of spark.comet.convert.broadcastBuildSide.text.enabled, and note in the migration guide that converting Text through sparkToColumnar is deprecated. * fix: drop the unused SparkConf import and narrow the min/max workaround The suite no longer overrides sparkConf, so its SparkConf import failed Scalafix and the strict-warnings build. Reformat a CometExecRule comment that Spotless rejected. The three-config min/max workaround only keeps the decode out of the query when the cast is the aggregate's own argument. A cast in a projection below the aggregate runs natively and decodes first, so the guide now says so and keeps disabling Comet as the general answer. (cherry picked from commit 2b1c6a3938cbe1c4b2d0aa6383e64360569f877a) --- .../user-guide/latest/compatibility/index.md | 38 +- docs/source/user-guide/latest/datasources.md | 24 +- .../user-guide/latest/migration-guide.md | 3 + .../scala/org/apache/comet/CometConf.scala | 26 +- .../apache/comet/rules/CometExecRule.scala | 50 +- .../comet/CometBroadcastBuildSideSuite.scala | 722 +++++++++++++++++- .../apache/comet/exec/CometExecSuite.scala | 661 +--------------- 7 files changed, 859 insertions(+), 665 deletions(-) diff --git a/docs/source/user-guide/latest/compatibility/index.md b/docs/source/user-guide/latest/compatibility/index.md index cbff79d5afb..f93c7f1acad 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 9861eac54f0..044d8d192bc 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 24f38826463..2419192865b 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -245,6 +245,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) @@ -1342,10 +1354,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) @@ -1354,13 +1367,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 02e4cff1a44..37323e3dce7 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 @@ -193,6 +195,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 @@ -990,7 +1001,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) @@ -1460,8 +1471,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 @@ -1470,6 +1482,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 => @@ -1517,12 +1530,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 @@ -1551,7 +1580,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 @@ -1681,10 +1710,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 346b0b8cd91..c1d03d6a40b 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" @@ -5728,6 +5096,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()))