You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "coderfender (via GitHub)" <gi...@apache.org> on 2026/05/25 17:20:09 UTC
[PR] feat: Broadcast nested join loop support [datafusion-comet]
coderfender opened a new pull request, #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #.
## Rationale for this change
Native support for BroadcastNestedLoopJoin to improve native performance.
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
## How are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Native Broadcast nested loop join support [datafusion-comet]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429#discussion_r3367815688
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1924,6 +1924,158 @@ trait CometHashJoin {
}
}
+case class CometBroadcastNestedLoopJoinExec(
Review Comment:
Verified DF's `NestedLoopJoinExec` uses the same `BuildProbeJoinMetrics` shape as `HashJoinExec` So instead of duplicating the metric map, renamed `hashJoinMetrics` to `joinMetrics` in `CometMetricNode.scala`, updated the two existing call sites (BHJ + SHJ case classes), and added a new override CometBroadcastNestedLoopJoinExec to leverage the same metric report function
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Native Broadcast nested loop join support [datafusion-comet]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429#discussion_r3374621045
##########
spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala:
##########
@@ -702,4 +705,102 @@ class CometJoinSuite extends CometTestBase {
}
}
}
+
+ test("BroadcastNestedLoopJoin with unequal filter") {
Review Comment:
Thank you.
1. Added LeftOuter + BuildLeft, FullOuter tests which should fall back to spark.
2. Also added nulls to input data for all the test cases to verify null join key condition as well
3. Broadcast reuse was a great idea ! I added a test to verify that as well
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Native Broadcast nested loop join support [datafusion-comet]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429#issuecomment-4571942209
Investigating test failures
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Native Broadcast nested loop join support [datafusion-comet]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429#discussion_r3367816692
##########
spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala:
##########
@@ -38,7 +38,10 @@ class CometJoinSuite extends CometTestBase {
override protected def test(testName: String, testTags: Tag*)(testFun: => Any)(implicit
pos: Position): Unit = {
super.test(testName, testTags: _*) {
- withSQLConf(CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true") {
+ withSQLConf(
+ CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_BROADCAST_NESTED_LOOP_JOIN_ENABLED.key -> "true",
+ CometConf.getOperatorAllowIncompatConfigKey("BroadcastNestedLoopJoinExec") -> "true") {
Review Comment:
Addressed this. Thank you
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Native Broadcast nested loop join support [datafusion-comet]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429#issuecomment-4569151592
Benchmarks (local M5 Max) :
```
range join (BETWEEN): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
Spark 5257 5275 26 0.2 5013.0 1.0X
Comet 368 375 7 2.9 350.8 14.3X
Running benchmark: inequality join (>)
Running case: Spark
Stopped after 2 iterations, 8921 ms
Running case: Comet
Stopped after 6 iterations, 2122 ms
OpenJDK 64-Bit Server VM 17.0.16+8-LTS on Mac OS X 26.3.2
Apple M5 Pro
inequality join (>): Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
Spark 4448 4461 18 0.2 4241.9 1.0X
Comet 346 354 6 3.0 329.8 12.9X
Running benchmark: left outer non-equi
Running case: Spark
Stopped after 2 iterations, 11028 ms
Running case: Comet
Stopped after 6 iterations, 2246 ms
OpenJDK 64-Bit Server VM 17.0.16+8-LTS on Mac OS X 26.3.2
Apple M5 Pro
left outer non-equi: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
------------------------------------------------------------------------------------------------------------------------
Spark 5513 5514 2 0.2 5257.7 1.0X
Comet 362 374 8 2.9 345.5 15.2X
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Native Broadcast nested loop join support [datafusion-comet]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429#issuecomment-4685564352
Thank you for the approval @andygrove
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Native Broadcast nested loop join support [datafusion-comet]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender merged PR #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Native Broadcast nested loop join support [datafusion-comet]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429#issuecomment-4569171005
TODO : Build vs Probe swap to DF's BNLJ op
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Native Broadcast nested loop join support [datafusion-comet]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429#issuecomment-4538939463
Bunch TPCDS queries seem to be using BNLJ warranting golden file regen
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Native Broadcast nested loop join support [datafusion-comet]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on code in PR #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429#discussion_r3367795619
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1924,6 +1924,158 @@ trait CometHashJoin {
}
}
+case class CometBroadcastNestedLoopJoinExec(
+ override val nativeOp: Operator,
+ override val originalPlan: SparkPlan,
+ override val output: Seq[Attribute],
+ override val outputOrdering: Seq[SortOrder],
+ joinType: JoinType,
+ condition: Option[Expression],
+ buildSide: BuildSide,
+ override val left: SparkPlan,
+ override val right: SparkPlan,
+ override val serializedPlanOpt: SerializedPlan)
+ extends CometBinaryExec {
+
+ // Mirror Spark's BroadcastNestedLoopJoinExec: output partitioning derives from the streamed
+ // (non-broadcast) side. Reading from live left/right rather than originalPlan keeps this
+ // correct after an AQE child swap.
+ private def streamedPlan: SparkPlan = buildSide match {
+ case BuildLeft => right
+ case BuildRight => left
+ }
+
+ override def outputPartitioning: Partitioning = streamedPlan.outputPartitioning
+
+ override def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan =
+ this.copy(left = newLeft, right = newRight)
+
+ override def stringArgs: Iterator[Any] =
+ Iterator(joinType, buildSide, condition, left, right)
+
+ override def equals(obj: Any): Boolean = {
+ obj match {
+ case other: CometBroadcastNestedLoopJoinExec =>
+ this.output == other.output &&
+ this.joinType == other.joinType &&
+ this.condition == other.condition &&
+ this.buildSide == other.buildSide &&
+ this.left == other.left &&
+ this.right == other.right &&
+ this.serializedPlanOpt == other.serializedPlanOpt
+ case _ =>
+ false
+ }
+ }
+
+ override def hashCode(): Int =
+ Objects.hashCode(output, joinType, condition, buildSide, left, right)
+}
+
+object CometBroadcastNestedLoopJoinExec extends CometOperatorSerde[BroadcastNestedLoopJoinExec] {
+
+ /**
+ * Get the optional Comet configuration entry that is used to enable or disable native support
+ * for this operator.
+ */
+ override def enabledConfig: Option[ConfigEntry[Boolean]] = {
+ Some(CometConf.COMET_EXEC_BROADCAST_NESTED_LOOP_JOIN_ENABLED)
+ }
+
+ private val broadcastBuildReplicationReason =
+ "BNLJ combinations that emit per-build-row results need a cross-partition merge that" +
+ " DataFusion's NestedLoopJoinExec does not provide. Affects: LeftOuter+BuildLeft," +
+ " RightOuter+BuildRight, FullOuter, LeftSemi+BuildLeft, LeftAnti+BuildLeft."
+
+ override def getSupportLevel(op: BroadcastNestedLoopJoinExec): SupportLevel =
Review Comment:
Agreed. I was initially gating the join behind the config in the test suite to figure out the support matrix through testing. I have since then removed the now redundant config
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Native Broadcast nested loop join support [datafusion-comet]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429#issuecomment-4664284736
Thank you very much for the approval @mbutrovich
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Native Broadcast nested loop join support [datafusion-comet]
Posted by "coderfender (via GitHub)" <gi...@apache.org>.
coderfender commented on PR #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429#issuecomment-4666143254
Tagging @parthchandra @comphead @andygrove for additional review
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Native Broadcast nested loop join support [datafusion-comet]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #4429:
URL: https://github.com/apache/datafusion-comet/pull/4429#discussion_r3363491497
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1924,6 +1924,158 @@ trait CometHashJoin {
}
}
+case class CometBroadcastNestedLoopJoinExec(
Review Comment:
Should this define a `metrics` map like `CometMetricNode.hashJoinMetrics` does for BHJ (and `sortMergeJoinMetrics` for SMJ)? As written it inherits `baselineMetrics`, so the native NLJ's build/probe/join timing metrics won't surface in the Spark UI.
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1924,6 +1924,158 @@ trait CometHashJoin {
}
}
+case class CometBroadcastNestedLoopJoinExec(
+ override val nativeOp: Operator,
+ override val originalPlan: SparkPlan,
+ override val output: Seq[Attribute],
+ override val outputOrdering: Seq[SortOrder],
+ joinType: JoinType,
+ condition: Option[Expression],
+ buildSide: BuildSide,
+ override val left: SparkPlan,
+ override val right: SparkPlan,
+ override val serializedPlanOpt: SerializedPlan)
+ extends CometBinaryExec {
+
+ // Mirror Spark's BroadcastNestedLoopJoinExec: output partitioning derives from the streamed
+ // (non-broadcast) side. Reading from live left/right rather than originalPlan keeps this
+ // correct after an AQE child swap.
+ private def streamedPlan: SparkPlan = buildSide match {
+ case BuildLeft => right
+ case BuildRight => left
+ }
+
+ override def outputPartitioning: Partitioning = streamedPlan.outputPartitioning
+
+ override def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan =
+ this.copy(left = newLeft, right = newRight)
+
+ override def stringArgs: Iterator[Any] =
+ Iterator(joinType, buildSide, condition, left, right)
+
+ override def equals(obj: Any): Boolean = {
+ obj match {
+ case other: CometBroadcastNestedLoopJoinExec =>
+ this.output == other.output &&
+ this.joinType == other.joinType &&
+ this.condition == other.condition &&
+ this.buildSide == other.buildSide &&
+ this.left == other.left &&
+ this.right == other.right &&
+ this.serializedPlanOpt == other.serializedPlanOpt
+ case _ =>
+ false
+ }
+ }
+
+ override def hashCode(): Int =
+ Objects.hashCode(output, joinType, condition, buildSide, left, right)
+}
+
+object CometBroadcastNestedLoopJoinExec extends CometOperatorSerde[BroadcastNestedLoopJoinExec] {
+
+ /**
+ * Get the optional Comet configuration entry that is used to enable or disable native support
+ * for this operator.
+ */
+ override def enabledConfig: Option[ConfigEntry[Boolean]] = {
+ Some(CometConf.COMET_EXEC_BROADCAST_NESTED_LOOP_JOIN_ENABLED)
+ }
+
+ private val broadcastBuildReplicationReason =
+ "BNLJ combinations that emit per-build-row results need a cross-partition merge that" +
+ " DataFusion's NestedLoopJoinExec does not provide. Affects: LeftOuter+BuildLeft," +
+ " RightOuter+BuildRight, FullOuter, LeftSemi+BuildLeft, LeftAnti+BuildLeft."
+
+ override def getSupportLevel(op: BroadcastNestedLoopJoinExec): SupportLevel =
Review Comment:
The support matrix here lines up exactly with Spark's `BroadcastNestedLoopJoinExec.outputPartitioning` supported set, and the reasoning in `broadcastBuildReplicationReason` is spot on. Every case returns either `Compatible(None)` or `Unsupported(...)`, never `Incompatible`. Since `isOperatorEnabled` only consults the allow-incompat config on the `Incompatible` branch, the `getOperatorAllowIncompatConfigKey("BroadcastNestedLoopJoinExec")` line in the `CometJoinSuite` override is dead. Is that intentional, or leftover?
##########
spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala:
##########
@@ -702,4 +705,102 @@ class CometJoinSuite extends CometTestBase {
}
}
}
+
+ test("BroadcastNestedLoopJoin with unequal filter") {
Review Comment:
The compatible matrix is well covered. Could we add a couple of result-only `checkSparkAnswer` cases for the unsupported combinations (`FullOuter`, `LeftOuter` with `BuildLeft`) to confirm they fall back to Spark and still match? A broadcast-reuse case (same relation feeding two BNLJs under AQE) would also be reassuring given reuse is handled generically rather than in this operator. Minor: no coverage for a join condition referencing nullable or null-producing columns, and no left-outer-without-condition case.
##########
spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala:
##########
@@ -38,7 +38,10 @@ class CometJoinSuite extends CometTestBase {
override protected def test(testName: String, testTags: Tag*)(testFun: => Any)(implicit
pos: Position): Unit = {
super.test(testName, testTags: _*) {
- withSQLConf(CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true") {
+ withSQLConf(
+ CometConf.COMET_EXEC_SHUFFLE_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_BROADCAST_NESTED_LOOP_JOIN_ENABLED.key -> "true",
+ CometConf.getOperatorAllowIncompatConfigKey("BroadcastNestedLoopJoinExec") -> "true") {
Review Comment:
Similar to the other comment in operators.scala. This allow-incompat config is never read for BNLJ. `getSupportLevel` only ever returns `Compatible(None)` or `Unsupported(...)`, never `Incompatible`, and `isOperatorEnabled` only consults allow-incompat on the `Incompatible` branch. Can this be dropped, or was an `Incompatible` case intended?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org