parthchandra commented on code in PR #6786:
URL: https://github.com/apache/datafusion-comet/pull/6786#discussion_r4263484615
##########
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:
Thank you!. Added `CometBroadcastBuildSideSuite` based on the probe
covering ORC and Iceberg build sides (the other formats the catch-all arms
bridge), plus a DSv2 case in CometExecSuite. Registered in both pr_build_*.yml.
##########
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:
You're right — it never fired (the bridge is behind the CometScanWrapper
leaf) and the rebuild was wrong for sink placeholders, unions and native scans.
Did what you suggested: dropped the revert and instead clear the tag and
convert the build side again. clearWastedBuildSideBridges
(CometExecRule.scala:1683) runs after the first transform, clears
BROADCAST_BUILD_SIDE_TAG from any scan still under a plain Spark
BroadcastExchangeExec, then transforms once more (CometExecRule.scala:999) —
same plan as without the feature, and the AQE stage pass no longer re-bridges.
Fixed the docstring (CometExecRule.scala:1533). Added FULL OUTER BNLJ tests
with filter, aggregate and union build sides in both AQE modes, each asserting
no CometSparkToColumnarExec under a Spark BroadcastExchangeExec
--
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]
