parthchandra commented on code in PR #6786:
URL: https://github.com/apache/datafusion-comet/pull/6786#discussion_r4265484822
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -1480,6 +1508,194 @@ 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,
`clearWastedBuildSideBridges` clears the tag
+ * after the transform and a second transform rebuilds the Spark build side,
so a leftover copy
+ * never reaches runtime. See issue #6008.
+ */
+ private def broadcastBuildSideBridgeEnabled: Boolean =
+ CometConf.COMET_CONVERT_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,
+ CometConf.COMET_EXEC_BROADCAST_HASH_JOIN_ENABLED)
+ case j: BroadcastNestedLoopJoinExec =>
+ tagBuildIfProbeNative(
+ j.buildSide,
+ j.left,
+ j.right,
+ CometConf.COMET_EXEC_BROADCAST_NESTED_LOOP_JOIN_ENABLED)
+ case _ =>
+ }
+ }
+
+ /**
+ * Tag the build side's file-source scans, but only when the join looks like
it can become a
+ * native Comet broadcast join: the probe side is natively scannable and the
join's own serde is
+ * enabled. This is a best-effort gate that avoids inserting a bridge we
would only tear down
+ * again - it cannot see every reason a join stays on Spark (an unsupported
operator above a
+ * native probe scan, or a join type the serde declines), so
`clearWastedBuildSideBridges` is
+ * the backstop that undoes any bridge left under a Spark
`BroadcastExchangeExec` after the
+ * transform. Skipping here still matters because a bridge under a Spark
broadcast breaks
+ * Spark's DPP broadcast reuse until it is torn down (the DPP subquery keeps
an unbridged copy).
+ * We skip when:
+ * - the join's own serde is disabled (`joinEnabled`), so it will stay on
Spark; or
+ * - the build-side `BroadcastExchangeExec` carries
`SKIP_COMET_BROADCAST_TAG`, which
+ * `CometSpark34AqeDppFallbackRule` sets on Spark 3.4 to keep the
broadcast Spark-native so
+ * `PlanAdaptiveDynamicPruningFilters` can match it (bridging it would
turn DPP off). See
+ * issue #6008.
+ */
+ private def tagBuildIfProbeNative(
+ buildSide: BuildSide,
+ left: SparkPlan,
+ right: SparkPlan,
+ joinEnabled: ConfigEntry[Boolean]): Unit = {
+ if (!joinEnabled.get(conf)) {
+ return
+ }
+ val (buildPlan, probePlan) = buildSide match {
+ case BuildLeft => (left, right)
+ case BuildRight => (right, left)
+ }
+ val skipForDpp = buildPlan match {
+ case b: BroadcastExchangeExec =>
+ b.getTagValue(CometExecRule.SKIP_COMET_BROADCAST_TAG).isDefined
+ case _ => false
+ }
+ if (!skipForDpp && hasOnlyNativeScans(probePlan)) {
+ tagBuildSideScans(buildPlan)
+ }
+ }
+
+ /**
+ * Tag the broadcast dimension's own file-source scans, descending through
its filters, projects
+ * and aggregations (incl. the aggregate's shuffle) but stopping at a nested
join. A nested
+ * join's inputs are not the bounded broadcast dimension - its streamed side
can be far larger
+ * than the broadcast output - so bridging them would copy a large input to
Arrow for no reason.
+ * That nested join is tagged by its own pre-pass visit based on its own
probe. See issue #6008.
+ */
+ private def tagBuildSideScans(plan: SparkPlan): Unit = plan match {
+ case scan @ (_: FileSourceScanExec | _: BatchScanExec) =>
+ scan.setTagValue(CometExecRule.BROADCAST_BUILD_SIDE_TAG, ())
+ case _: BroadcastHashJoinExec | _: BroadcastNestedLoopJoinExec | _:
ShuffledHashJoinExec |
+ _: SortMergeJoinExec =>
+ case other => other.children.foreach(tagBuildSideScans)
+ }
+
+ /**
+ * True when `plan` has at least one scan and none of its scan leaves is an
unsupported plain
+ * Spark scan - i.e. every scan is already a Comet scan (CometScanRule ran
before this rule).
+ * Used to decide whether a broadcast join's probe side can feed a native
join. A probe that
+ * itself contains an unsupported scan (e.g. a DSv2
`InMemoryTableWithV2Filter` fact) keeps the
+ * join on Spark, so bridging its build side would only break DPP reuse.
+ */
+ private def hasOnlyNativeScans(plan: SparkPlan): Boolean = {
+ var sawNativeScan = false
+ var sawUnsupportedScan = false
+ plan.foreach {
+ case _: CometScanExec | _: CometNativeScanExec | _: CometBatchScanExec =>
+ sawNativeScan = true
+ case _: FileSourceScanExec | _: BatchScanExec =>
+ sawUnsupportedScan = true
+ case _ =>
+ }
+ sawNativeScan && !sawUnsupportedScan
+ }
+
+ /**
+ * After the main transform, a #6008 build-side bridge is left under a plain
Spark
+ * `BroadcastExchangeExec` only when its broadcast join did not go native -
an unsupported join
+ * type (e.g. a full outer `BroadcastNestedLoopJoinExec`), a build side the
serde declined, or a
+ * probe that is not fully native (e.g. an unsupported operator above a
native scan). Clear
+ * `BROADCAST_BUILD_SIDE_TAG` from every scan under such a broadcast so a
second transform
+ * rebuilds that build side exactly as it would without the feature, and so
a later AQE stage
+ * pass - which prepares the build stage on its own, without the join above
it - does not bridge
+ * the scan again. Returns true if any tag was cleared.
+ *
+ * Scans under a `CometBroadcastExchangeExec` keep their tag: that join went
native and wants
Review Comment:
Right, the walk didn't match the comment. clearScanTags now stops at
CometBroadcastExchangeExec, so a nested native broadcast join inside a
Spark-staying build keeps its bridge.
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -1480,6 +1508,194 @@ 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,
`clearWastedBuildSideBridges` clears the tag
+ * after the transform and a second transform rebuilds the Spark build side,
so a leftover copy
+ * never reaches runtime. See issue #6008.
+ */
+ private def broadcastBuildSideBridgeEnabled: Boolean =
+ CometConf.COMET_CONVERT_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,
+ CometConf.COMET_EXEC_BROADCAST_HASH_JOIN_ENABLED)
+ case j: BroadcastNestedLoopJoinExec =>
+ tagBuildIfProbeNative(
+ j.buildSide,
+ j.left,
+ j.right,
+ CometConf.COMET_EXEC_BROADCAST_NESTED_LOOP_JOIN_ENABLED)
+ case _ =>
+ }
+ }
+
+ /**
+ * Tag the build side's file-source scans, but only when the join looks like
it can become a
+ * native Comet broadcast join: the probe side is natively scannable and the
join's own serde is
+ * enabled. This is a best-effort gate that avoids inserting a bridge we
would only tear down
+ * again - it cannot see every reason a join stays on Spark (an unsupported
operator above a
+ * native probe scan, or a join type the serde declines), so
`clearWastedBuildSideBridges` is
+ * the backstop that undoes any bridge left under a Spark
`BroadcastExchangeExec` after the
+ * transform. Skipping here still matters because a bridge under a Spark
broadcast breaks
+ * Spark's DPP broadcast reuse until it is torn down (the DPP subquery keeps
an unbridged copy).
+ * We skip when:
+ * - the join's own serde is disabled (`joinEnabled`), so it will stay on
Spark; or
+ * - the build-side `BroadcastExchangeExec` carries
`SKIP_COMET_BROADCAST_TAG`, which
+ * `CometSpark34AqeDppFallbackRule` sets on Spark 3.4 to keep the
broadcast Spark-native so
+ * `PlanAdaptiveDynamicPruningFilters` can match it (bridging it would
turn DPP off). See
+ * issue #6008.
+ */
+ private def tagBuildIfProbeNative(
+ buildSide: BuildSide,
+ left: SparkPlan,
+ right: SparkPlan,
+ joinEnabled: ConfigEntry[Boolean]): Unit = {
+ if (!joinEnabled.get(conf)) {
+ return
+ }
+ val (buildPlan, probePlan) = buildSide match {
+ case BuildLeft => (left, right)
+ case BuildRight => (right, left)
+ }
+ val skipForDpp = buildPlan match {
+ case b: BroadcastExchangeExec =>
+ b.getTagValue(CometExecRule.SKIP_COMET_BROADCAST_TAG).isDefined
+ case _ => false
+ }
+ if (!skipForDpp && hasOnlyNativeScans(probePlan)) {
+ tagBuildSideScans(buildPlan)
+ }
+ }
+
+ /**
+ * Tag the broadcast dimension's own file-source scans, descending through
its filters, projects
+ * and aggregations (incl. the aggregate's shuffle) but stopping at a nested
join. A nested
+ * join's inputs are not the bounded broadcast dimension - its streamed side
can be far larger
+ * than the broadcast output - so bridging them would copy a large input to
Arrow for no reason.
+ * That nested join is tagged by its own pre-pass visit based on its own
probe. See issue #6008.
+ */
+ private def tagBuildSideScans(plan: SparkPlan): Unit = plan match {
+ case scan @ (_: FileSourceScanExec | _: BatchScanExec) =>
+ scan.setTagValue(CometExecRule.BROADCAST_BUILD_SIDE_TAG, ())
+ case _: BroadcastHashJoinExec | _: BroadcastNestedLoopJoinExec | _:
ShuffledHashJoinExec |
+ _: SortMergeJoinExec =>
+ case other => other.children.foreach(tagBuildSideScans)
+ }
+
+ /**
+ * True when `plan` has at least one scan and none of its scan leaves is an
unsupported plain
+ * Spark scan - i.e. every scan is already a Comet scan (CometScanRule ran
before this rule).
+ * Used to decide whether a broadcast join's probe side can feed a native
join. A probe that
+ * itself contains an unsupported scan (e.g. a DSv2
`InMemoryTableWithV2Filter` fact) keeps the
+ * join on Spark, so bridging its build side would only break DPP reuse.
+ */
+ private def hasOnlyNativeScans(plan: SparkPlan): Boolean = {
+ var sawNativeScan = false
+ var sawUnsupportedScan = false
+ plan.foreach {
+ case _: CometScanExec | _: CometNativeScanExec | _: CometBatchScanExec =>
+ sawNativeScan = true
+ case _: FileSourceScanExec | _: BatchScanExec =>
+ sawUnsupportedScan = true
+ case _ =>
+ }
+ sawNativeScan && !sawUnsupportedScan
+ }
+
+ /**
+ * After the main transform, a #6008 build-side bridge is left under a plain
Spark
+ * `BroadcastExchangeExec` only when its broadcast join did not go native -
an unsupported join
+ * type (e.g. a full outer `BroadcastNestedLoopJoinExec`), a build side the
serde declined, or a
+ * probe that is not fully native (e.g. an unsupported operator above a
native scan). Clear
+ * `BROADCAST_BUILD_SIDE_TAG` from every scan under such a broadcast so a
second transform
+ * rebuilds that build side exactly as it would without the feature, and so
a later AQE stage
+ * pass - which prepares the build stage on its own, without the join above
it - does not bridge
+ * the scan again. Returns true if any tag was cleared.
+ *
+ * Scans under a `CometBroadcastExchangeExec` keep their tag: that join went
native and wants
+ * the bridged build. 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
+ def clearScanTags(node: SparkPlan): Unit = {
+ if (node.getTagValue(CometExecRule.BROADCAST_BUILD_SIDE_TAG).isDefined) {
+ node.unsetTagValue(CometExecRule.BROADCAST_BUILD_SIDE_TAG)
+ cleared = true
Review Comment:
Good catch — fixed. clearScanTags now only counts a clear when the tagged
scan is under a CometSparkToColumnarExec, so a tagged-but-unbridged
CSV/JSON/Parquet scan no longer triggers the second transform. Matches what you
measured.Good catch — fixed. clearScanTags now only counts a clear when the
tagged scan is under a CometSparkToColumnarExec, so a tagged-but-unbridged
CSV/JSON/Parquet scan no longer triggers the second transform. Matches what you
measured.
##########
spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala:
##########
@@ -307,6 +307,592 @@ 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("Spark 3.4 AQE DPP with a text dimension on the broadcast build side
(#6008)") {
Review Comment:
Agreed — on 3.4 the V1 fact stays on Spark so nothing is tagged. Renamed and
recommented it as the V1-fallback case (feature stays out of the way, DPP stays
live), and noted the Iceberg test in CometBroadcastBuildSideSuite covers the
skip-tag path with a native fact.
--
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]
