andygrove commented on code in PR #6786:
URL: https://github.com/apache/datafusion-comet/pull/6786#discussion_r4241895250
##########
spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala:
##########
@@ -301,6 +301,366 @@ class CometExecSuite extends CometTestBase {
}
}
+ test("broadcast build side with unsupported text source goes native
(#6008)") {
Review Comment:
Here it is. It runs as its own suite under
`spark/src/test/scala/org/apache/comet/`. Each query goes through
`checkSparkAnswer`, and the printed line shows whether the bridge and the
native join were planned. All three tests pass on `3738210b84` with 3.4, and
the first two also pass on 4.1. For the PR, asserting the bridge and the
`CometBroadcastHashJoinExec` instead of printing them should be enough.
<details><summary><code>Pr6786BuildSideProbeSuite.scala</code></summary>
```scala
package org.apache.comet
import org.apache.spark.sql.{CometTestBase, DataFrame}
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.internal.SQLConf
// Probes from the PR #6786 review: ORC and Iceberg build sides that Comet
cannot scan natively,
// bridged by spark.comet.convert.broadcastBuildSide.enabled. Each query is
compared with Spark
// through checkSparkAnswer, and the summary line shows whether the bridge
and the native join
// were planned.
class Pr6786BuildSideProbeSuite extends CometTestBase with
CometIcebergTestBase {
/** Every node, descending through AQE, stages and subqueries, but not
into reused exchanges. */
private def allNodes(p: SparkPlan): Seq[SparkPlan] = {
val kids = p match {
case a: AdaptiveSparkPlanExec => Seq(a.executedPlan)
case s: QueryStageExec => Seq(s.plan)
case _: ReusedExchangeExec => Seq.empty
case other => other.children
}
val subs = p.expressions.flatMap(_.collect { case e:
ExecSubqueryExpression => e.plan })
Seq(p) ++ kids.flatMap(allNodes) ++ subs.flatMap(allNodes)
}
private def summary(df: DataFrame): String = {
df.collect()
val nodes = allNodes(df.queryExecution.executedPlan)
s"bridges=${nodes.count(_.isInstanceOf[CometSparkToColumnarExec])} " +
s"cometBHJ=${nodes.count(_.isInstanceOf[CometBroadcastHashJoinExec])}
" +
s"icebergNative=${nodes.count(_.isInstanceOf[CometIcebergNativeScanExec])}"
}
private def dppSummary(df: DataFrame): String = {
df.collect()
val nodes = allNodes(df.queryExecution.executedPlan)
val dppTrue = nodes.exists(_.expressions.exists(_.exists {
case DynamicPruningExpression(Literal.TrueLiteral) => true
case _ => false
}))
val dppLive = nodes.exists(_.expressions.exists(_.exists {
case DynamicPruningExpression(e) if e != Literal.TrueLiteral => true
case _ => false
}))
summary(df) + s" dppLive=$dppLive dppTrue=$dppTrue " +
s"reusedX=${nodes.count(_.isInstanceOf[ReusedExchangeExec])}"
}
test("ORC build side, all types, non-UTC session (#6008)") {
withTempDir { dir =>
val fact = s"${dir.getAbsolutePath}/fact.parquet"
val dim = s"${dir.getAbsolutePath}/dim.orc"
withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
spark.range(0, 2000).selectExpr("CAST(id % 60 AS INT) AS k", "id AS
v").write.parquet(fact)
spark
.range(0, 60, 1, 3)
.selectExpr(
"CAST(id AS INT) AS k",
"IF(id % 7 = 0, NULL, CAST(id AS BYTE)) AS b",
"IF(id % 7 = 1, NULL, CAST(id AS SHORT)) AS s",
"IF(id % 7 = 2, NULL, id * 1000000000L) AS l",
"IF(id % 7 = 3, NULL, id % 2 = 0) AS bo",
"IF(id % 7 = 4, NULL, CAST(id AS FLOAT) / 3) AS f",
"CASE WHEN id % 7 = 5 THEN NULL WHEN id = 6 THEN CAST('NaN' AS
DOUBLE) " +
"WHEN id = 8 THEN -0.0D ELSE id / 3.0D END AS d",
"IF(id % 7 = 6, NULL, CAST(id * 1.25 AS DECIMAL(10,2))) AS dec1",
"IF(id % 5 = 0, NULL, CAST(id AS DECIMAL(38,10)) / 7) AS dec2",
"IF(id % 5 = 1, NULL, CONCAT('s', id, 'é')) AS str",
"IF(id % 5 = 2, NULL, CAST(CONCAT('b', id) AS BINARY)) AS bin",
"IF(id % 5 = 3, NULL, DATE_ADD(DATE'1960-01-01', CAST(id * 300
AS INT))) AS dt",
"IF(id % 5 = 4, NULL, TIMESTAMP_MICROS(id * 8640000000000L -
900000000000000L)) AS ts",
"IF(id % 9 = 0, NULL, " +
"CAST(TIMESTAMP_MICROS(id * 8640000000000L - 900000000000000L)
AS TIMESTAMP_NTZ)) " +
"AS ntz",
"IF(id % 6 = 0, NULL, named_struct('a', CAST(id AS INT), 'b', " +
"IF(id % 4 = 0, NULL, CONCAT('x', id)))) AS st",
"CASE WHEN id % 6 = 1 THEN NULL WHEN id % 6 = 2 THEN
CAST(array() AS ARRAY<STRING>) " +
"ELSE array(CONCAT('e', id), NULL, 'z') END AS arr",
"CASE WHEN id % 6 = 3 THEN NULL WHEN id % 6 = 4 THEN " +
"CAST(map() AS MAP<STRING,STRING>) " +
"ELSE map(CONCAT('k', id), IF(id % 2 = 0, NULL, 'v')) END AS
m")
.write
.orc(dim)
}
val query =
"""SELECT /*+ BROADCAST(d) */ f.v, d.*, hour(d.ts) AS h, CAST(d.ts
AS STRING) AS tss,
| CAST(d.ntz AS STRING) AS ntzs, date_trunc('DAY', d.ts) AS tday
|FROM fact_o f JOIN dim_o d ON f.k = d.k""".stripMargin
for {
aqe <- Seq("false", "true")
v1List <- Seq("avro,csv,json,kafka,orc,parquet,text", "parquet")
(vec, nestedVec) <- Seq(("true", "true"), ("true", "false"),
("false", "false"))
} {
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe,
SQLConf.USE_V1_SOURCE_LIST.key -> v1List,
SQLConf.ORC_VECTORIZED_READER_ENABLED.key -> vec,
SQLConf.ORC_VECTORIZED_READER_NESTED_COLUMN_ENABLED.key ->
nestedVec,
SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
spark.read.parquet(fact).createOrReplaceTempView("fact_o")
spark.read.orc(dim).createOrReplaceTempView("dim_o")
// scalastyle:off println
println(s"ORC aqe=$aqe v1=$v1List vec=$vec nested=$nestedVec: " +
summary(sql(query)))
// scalastyle:on println
checkSparkAnswer(sql(query))
}
}
}
}
test("Iceberg build side that Comet's native reader declines (#6008)") {
assume(icebergAvailable, "Iceberg not available in classpath")
withTempIcebergDir { warehouseDir =>
withTempDir { dir =>
val fact = s"${dir.getAbsolutePath}/fact.parquet"
withSQLConf(
"spark.sql.catalog.pc" -> "org.apache.iceberg.spark.SparkCatalog",
"spark.sql.catalog.pc.type" -> "hadoop",
"spark.sql.catalog.pc.warehouse" -> warehouseDir.getAbsolutePath) {
withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
spark
.range(0, 2000)
.selectExpr("CAST(id % 40 AS INT) AS k", "id AS v")
.write
.parquet(fact)
for ((name, props) <- Seq(
"dim_orc" -> "'write.format.default'='orc'",
"dim_pq" -> ("'format-version'='2',
'write.delete.mode'='merge-on-read', " +
"'write.update.mode'='merge-on-read'"))) {
spark.sql(s"""CREATE TABLE pc.db.$name (
| k INT, s STRING, d DECIMAL(10,2), ts TIMESTAMP,
| st STRUCT<a: INT, b: STRING>, arr
ARRAY<STRING>,
| m MAP<STRING, STRING>, region STRING)
|USING iceberg PARTITIONED BY (region)
|TBLPROPERTIES ($props)""".stripMargin)
spark
.range(0, 40, 1, 2)
.selectExpr(
"CAST(id AS INT) AS k",
"IF(id % 5 = 0, NULL, CONCAT('s', id)) AS s",
"IF(id % 6 = 0, NULL, CAST(id * 1.5 AS DECIMAL(10,2))) AS
d",
"IF(id % 7 = 0, NULL, " +
"TIMESTAMP_MICROS(id * 8640000000000L -
900000000000000L)) AS ts",
"IF(id % 4 = 0, NULL, named_struct('a', CAST(id AS INT),
'b', " +
"IF(id % 3 = 0, NULL, CONCAT('b', id)))) AS st",
"CASE WHEN id % 4 = 1 THEN NULL WHEN id % 4 = 2 THEN " +
"CAST(array() AS ARRAY<STRING>) ELSE array(CONCAT('e',
id), NULL) END AS arr",
"IF(id % 3 = 1, NULL, map(CONCAT('k', id), IF(id % 2 = 0,
NULL, 'v'))) AS m",
"IF(id % 2 = 0, 'east', 'west') AS region")
.writeTo(s"pc.db.$name")
.append()
}
spark.sql("DELETE FROM pc.db.dim_pq WHERE k % 9 = 0")
}
spark.read.parquet(fact).createOrReplaceTempView("ice_fact")
val queries = Seq(
"orc" -> ("SELECT /*+ BROADCAST(d) */ f.v, d.* FROM ice_fact f "
+
"JOIN pc.db.dim_orc d ON f.k = d.k"),
"pq-mor" -> ("SELECT /*+ BROADCAST(d) */ f.v, d.* FROM ice_fact
f " +
"JOIN pc.db.dim_pq d ON f.k = d.k"),
"snapshots" -> ("SELECT /*+ BROADCAST(s) */ f.v, s.operation,
s.summary FROM " +
"ice_fact f JOIN pc.db.dim_pq.snapshots s ON f.k =
CAST(s.snapshot_id % 2 AS INT)"))
for {
(name, query) <- queries
native <- Seq("true", "false")
aqe <- Seq("false", "true")
} {
withSQLConf(
CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> native,
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe,
SQLConf.SESSION_LOCAL_TIMEZONE.key -> "Asia/Kolkata") {
// scalastyle:off println
println(s"ICEBERG $name icebergNative=$native aqe=$aqe:
${summary(sql(query))}")
// scalastyle:on println
checkSparkAnswer(sql(query))
}
}
}
}
}
}
test("Spark 3.4: native Iceberg fact, text dimension, AQE DPP (#6008)") {
assume(icebergAvailable, "Iceberg not available in classpath")
withTempIcebergDir { warehouseDir =>
withTempDir { dir =>
val dim = s"${dir.getAbsolutePath}/dim.txt"
withSQLConf(
"spark.sql.catalog.pc" -> "org.apache.iceberg.spark.SparkCatalog",
"spark.sql.catalog.pc.type" -> "hadoop",
"spark.sql.catalog.pc.warehouse" -> warehouseDir.getAbsolutePath) {
withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
spark.sql("""CREATE TABLE pc.db.fact (v BIGINT, fk STRING)
|USING iceberg PARTITIONED BY (fk)""".stripMargin)
spark
.range(0, 1000, 1, 2)
.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.read.text(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 ON f.fk = d.value""".stripMargin
for {
aqe <- Seq("true", "false")
feature <- Seq("true", "false")
} {
withSQLConf(
SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> aqe,
SQLConf.DYNAMIC_PARTITION_PRUNING_ENABLED.key -> "true",
SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key ->
"true",
"spark.comet.convert.broadcastBuildSide.enabled" -> feature) {
// scalastyle:off println
val df = sql(query)
val dpp = dppSummary(df) // runs the query
val factRows =
allNodes(df.queryExecution.executedPlan).collect {
case s: CometIcebergNativeScanExec =>
s.metrics("output_rows").value
}
// With DPP live the fact scan reads 400 of its 1000 rows.
println(s"DPP aqe=$aqe feature=$feature: $dpp
factRows=${factRows.mkString(",")}")
// scalastyle:on println
checkSparkAnswer(sql(query))
}
}
}
}
}
}
}
```
</details>
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -1518,6 +1535,126 @@ 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.
+ */
+ private def shouldBridgeUnsupportedScan(conf: SQLConf, op: SparkPlan):
Boolean =
+ isSparkToArrowEnabled(conf, op) || isBroadcastBuildSideLeaf(op)
+
+ /**
+ * 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
+ * a `CometBroadcastExchangeExec`, so bridging the build side would only add
a row to Arrow copy
+ * for no native join. This does not also check the join serdes
+ * (COMET_EXEC_BROADCAST_HASH_JOIN_ENABLED /
COMET_EXEC_BROADCAST_NESTED_LOOP_JOIN_ENABLED): if
+ * the exchange converts but the join does not, the broadcast arm returns
the original plan and
+ * EliminateRedundantTransitions removes the now-unused bridge, so a
leftover copy never reaches
Review Comment:
I don't think `revertBridgedBroadcasts` ever fires. During the transform the
bridge is `CometScanWrapper(nativeOp, CometSparkToColumnarExec(scan))`, and
`CometScanWrapper` is a leaf, so
`b.exists(_.isInstanceOf[CometSparkToColumnarExec])` never sees it. On
`3738210b84` the `FULL OUTER` and build-left `LEFT OUTER` nested loop joins
still end up with `BroadcastExchange <- CometColumnarToRow <- CometFilter <-
CometSparkRowToColumnar <- FileScan text`, with AQE on and off. The same
happens when the hash join declines because an operator on the probe side stays
on Spark. I used an `explode` with `spark.comet.exec.explode.enabled=false` to
get one. With DPP on top, that query scans the text dimension twice and builds
two broadcasts. With the config off it scans it once and reuses the broadcast.
I tried making the check look inside `CometScanWrapper`, and that isn't
enough on its own. With AQE off, a filtered dimension goes back to plain Spark
and DPP reuses the broadcast again. With AQE on, the stage gets prepared again
without the join above it, and the scan still has `BROADCAST_BUILD_SIDE_TAG`,
so the bridge is back in the final plan. `revertCometToSpark` also rebuilds
both a `CometSinkPlaceHolder` and the Comet node inside it from the same
`originalPlan`. With an aggregate on the build side, that gives a Spark
`Exchange` over a `CometExchange` over a Spark partial aggregate, which fails
at runtime with `ClassCastException: OffHeapColumnVector cannot be cast to
CometVector`. With a `UNION ALL` on the build side, planning fails with
`Incorrect number of children`. It also skips the `sparkFallback` overrides in
`CometNativeScanExec`, `CometIcebergNativeScanExec` and `CometLocalTopKExec`.
Could the rule remove the tag from the scans under a join that stays on
Spark and convert that build side again, instead of rebuilding the Spark nodes?
That should give exactly the plan we get today, and with the tag gone the AQE
stage pass won't bridge the scan again. Could you also add tests for a `FULL
OUTER` nested loop join in both AQE modes, with a filter, an aggregate and a
union on the build side, that assert there's no `CometSparkToColumnarExec`
under a Spark `BroadcastExchangeExec`? A DPP query where the join declines
could assert the `ReusedExchangeExec`. The docstring here on
`broadcastBuildSideBridgeEnabled` also still says
`EliminateRedundantTransitions` removes the unused bridge and that the join
configs aren't checked.
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -1518,6 +1535,126 @@ 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.
+ */
+ private def shouldBridgeUnsupportedScan(conf: SQLConf, op: SparkPlan):
Boolean =
+ isSparkToArrowEnabled(conf, op) || isBroadcastBuildSideLeaf(op)
+
+ /**
+ * 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
+ * a `CometBroadcastExchangeExec`, so bridging the build side would only add
a row to Arrow copy
+ * for no native join. This does not also check the join serdes
+ * (COMET_EXEC_BROADCAST_HASH_JOIN_ENABLED /
COMET_EXEC_BROADCAST_NESTED_LOOP_JOIN_ENABLED): if
+ * the exchange converts but the join does not, the broadcast arm returns
the original plan and
+ * EliminateRedundantTransitions removes the now-unused bridge, so a
leftover copy never reaches
+ * runtime. See issue #6008.
+ */
+ private def broadcastBuildSideBridgeEnabled: Boolean =
+ CometConf.COMET_SPARK_TO_ARROW_BROADCAST_BUILD_SIDE_ENABLED.get(conf) &&
+ CometConf.COMET_EXEC_BROADCAST_EXCHANGE_ENABLED.get(conf)
+
+ /**
+ * True when `op` is a scan on a broadcast join's build side and build-side
bridging is enabled.
+ * `tagBroadcastBuildSideLeaves` only sets the tag when bridging is enabled,
but the tag lives
+ * on the physical node and can survive onto a reused instance, so re-check
here: a stale tag
+ * must not bridge a scan once the feature is turned off. A tagged scan is
bridged to Arrow even
+ * when the general sparkToColumnar path is off, letting the build branch
and the join run
+ * natively. See issue #6008.
+ */
+ private def isBroadcastBuildSideLeaf(op: SparkPlan): Boolean =
+ broadcastBuildSideBridgeEnabled &&
+ op.getTagValue(CometExecRule.BROADCAST_BUILD_SIDE_TAG).isDefined
+
+ /**
+ * 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
+ * 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
+ * selective filter above the scan means the bridge copies the pre-filter
scan output. See issue
+ * #6008.
+ *
+ * The bridge is applied ONLY when the probe side is already natively
scannable, i.e. the join
+ * can actually become a fully native CometBroadcastHashJoinExec. Bridging a
build side under a
+ * join that stays on Spark is pure overhead and, worse, it rewrites the
build broadcast's
+ * subtree to Comet and breaks Spark's DPP broadcast reuse (the DPP subquery
keeps an unbridged
+ * copy of the same scan, so the two exchanges no longer share a canonical
form). CometScanRule
+ * has already run by this point, so a native probe scan is a Comet scan and
an unsupported
+ * probe scan is still a plain Spark scan - see `hasOnlyNativeScans`.
+ *
+ * Only `FileSourceScanExec` / `BatchScanExec` are tagged - the scan types
the per-format arms
+ * of shouldApplySparkToColumnar handle. A natively scannable file (e.g.
Parquet) is normally
+ * already a `CometScanExec` here; if native scan conversion did not take,
its
+ * `FileSourceScanExec` is still tagged, but the tag is harmless because the
per-format arm
+ * decides that scan before isBroadcastBuildSideLeaf is ever consulted. DPP
subquery broadcasts
+ * live in expressions, not `children`, so `foreach` never reaches them -
that scope is
+ * intentional.
+ */
+ private def tagBroadcastBuildSideLeaves(plan: SparkPlan): Unit = {
+ if (!broadcastBuildSideBridgeEnabled) {
+ return
+ }
+ plan.foreach {
+ case j: BroadcastHashJoinExec => tagBuildIfProbeNative(j.buildSide,
j.left, j.right)
+ case j: BroadcastNestedLoopJoinExec =>
tagBuildIfProbeNative(j.buildSide, j.left, j.right)
+ case _ =>
+ }
+ }
+
+ /** Tag the build side's file-source scans, but only if the probe side is
natively scannable. */
+ private def tagBuildIfProbeNative(
+ buildSide: BuildSide,
+ left: SparkPlan,
+ right: SparkPlan): Unit = {
+ val (buildPlan, probePlan) = buildSide match {
+ case BuildLeft => (left, right)
+ case BuildRight => (right, left)
+ }
+ if (hasOnlyNativeScans(probePlan)) {
Review Comment:
Thanks, this works now. On 3.4.3 with AQE the fact scan reads 400 rows again
and the broadcast is reused. The new DPP test is skipped below 3.5, though, so
nothing would catch it if this came back. Could you add a 3.4 test with a
native Iceberg fact partitioned on the join key and a text dimension, filtered
with something Spark can't infer onto the fact, that asserts the fact's DPP
filter isn't `true`? The last test in the probe suite I posted in the ORC
thread does this.
##########
spark/src/main/scala/org/apache/comet/rules/CometPlanAdaptiveDynamicPruningFilters.scala:
##########
@@ -351,6 +360,26 @@ case object CometPlanAdaptiveDynamicPruningFilters
}
}
+ /**
+ * True when `plan` (a DPP subquery's build, taken from an
AdaptiveSparkPlanExec.executedPlan)
+ * produces native Comet columnar output, so it is safe as a
CometBroadcastExchangeExec child.
+ * Unwraps AQE stage wrappers and any columnar->row transition (mirrors the
stripping in
+ * CometExecRule.rewriteInSubqueryPlan), then checks for a CometNativeExec.
See issue #6008.
+ */
+ private def isNativeBuildSide(plan: SparkPlan): Boolean = {
+ val stripped = stripAQEPlan(plan) match {
+ case c2r: CometNativeColumnarToRowExec => c2r.child
+ case c2r: CometColumnarToRowExec => c2r.child
+ case WholeStageCodegenExec(c2r: CometColumnarToRowExec) =>
+ c2r.child match {
+ case InputAdapter(child) => child
+ case other => other
+ }
+ case other => other
+ }
+ stripped.isInstanceOf[CometNativeExec]
Review Comment:
Thanks, the union dimension gets its `ReusedExchange` back under AQE on 4.1.
Could you add a test so it stays that way? An AQE DPP query whose dimension is
a `UNION ALL` of two filtered Parquet tables, asserting a `ReusedExchangeExec`
under the DPP subquery, fails with the old `CometNativeExec` check. The
docstring above `isNativeBuildSide` also still says it checks for a
`CometNativeExec`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
