You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/16 20:27:33 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-6810-be65c57fba3879f086c355011b311c5dfa6b2ef7 created (now 783c4a4d48)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6810-be65c57fba3879f086c355011b311c5dfa6b2ef7
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 783c4a4d48 fix: fall back to Spark for NOT IN over an ordered streamed side (#6810)
This branch includes the following new commits:
new 783c4a4d48 fix: fall back to Spark for NOT IN over an ordered streamed side (#6810)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: fix: fall back to Spark for NOT IN over an ordered streamed side (#6810)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6810-be65c57fba3879f086c355011b311c5dfa6b2ef7
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 783c4a4d487334d098251b0241f1c0c78fc1597b
Author: dustin <dw...@users.noreply.github.com>
AuthorDate: Fri Oct 16 20:27:11 2026 +0000
fix: fall back to Spark for NOT IN over an ordered streamed side (#6810)
* fix: sort a native hash join's output when the probe order it reports is not kept
Spark reports a hash join's streamed-side ordering and may drop a sort
above the join on that basis. DataFusion keeps unmatched probe rows of a
right join in order only when it sees the probe input sorted, which it
does not when that order comes from outside the native plan. The new
HashJoin output_ordering field carries the ordering, and the native
planner adds a sort above the join when the join does not already
produce it.
* fix: send a hash join's reported ordering to the native planner
For LeftOuter with BuildRight and RightOuter with BuildLeft, Spark reports
the streamed side's ordering, and DataFusion runs the join as a right join
that may emit unmatched probe rows after the matched ones. The serde now
sends that ordering so the native planner can restore it, and falls back
to Spark when the ordering cannot be sorted natively.
* test: a native sort-merge join below a hash join adds no extra sort
The ordering the merge join produces reaches the hash join through the
swapped right join and its projection, so the planner keeps it without
sorting again.
* fix: keep the streamed order of a null-aware anti join and widen the ordering tests
A null-aware anti join runs unswapped, so DataFusion builds on Spark's
streamed side and concatenates its batches in reverse; it now sends its
ordering too and gets the same native sort. The tests cover multi-column
and multi-batch orderings, the expression fallback, a natively sorted
probe, join metrics with a sort on top, and the join class per hint. The
compatibility guide describes how hash joins keep the order.
* test: tighten the hash join ordering tests and correct the NOT IN docs
The null-aware anti join test now checks that its build spans several
batches, a metric check that held for any batch size is gone, and a test
comment claims only what the test checks. The guides say a NOT IN join is
sorted too, and that only a NOT IN under OR, planned as an existence
nested loop join, falls back to Spark.
* test: a null-aware anti join's native output is sorted across build batches
DataFusion builds a NOT IN join on Spark's streamed side and, on its hash
path, concatenates those batches in reverse, so the planner sorts the
output. The test feeds two sparse-key batches and checks the order.
* docs: say only a NOT IN combined with OR skips the native existence join
A plain NOT IN runs as a native null-aware anti join; only one combined
with another predicate through OR becomes an existence join that falls
back to Spark.
* refactor: share the planner's sort-unless-ordered helper with the sort aggregate
sort_unless_ordered now takes physical sort expressions. The hash join
builds them at its call site, and the sort aggregate arm calls the same
helper instead of its own ordering_satisfy check and SortExec wrap.
* test: check the hash join ordering field for every shape that sends one
The streamed order matrix and the null-aware anti join test assert that
the native op carries an output ordering. The separate test keeps only
the inner join check, since its outer half repeated a matrix case.
* test: keep the hash join probe order through runtime filter wrappers
With runtime filters on, a probe-side filter or projection becomes CometFilterExec or CometProjectionExec before the join is built. Check that both keep the sort's ordering, so an outer hash join that reports its streamed ordering still adds no sort.
* fix: fall back to Spark for NOT IN over an ordered streamed side
Comet runs a null-aware anti join unswapped in CollectLeft mode, so
DataFusion builds on Spark's streamed side. On its hash map path the
build batches are concatenated in reverse, and the unmatched rows come
out in reverse batch order. Spark has already dropped any sort above
the join because the join reports the streamed side's ordering, so
sortWithinPartitions returns unsorted partitions and a sort aggregate
above the join splits groups.
Sorting the native output instead needed about four times the memory
pool under fair_unified, so fall back to Spark when a null-aware anti
join's streamed side reports an ordering, until the DataFusion fix
(apache/datafusion#26226) reaches Comet. This drops the HashJoin
output_ordering field and the planner sort from this branch.
---
.../user-guide/latest/compatibility/operators.md | 8 +
docs/source/user-guide/latest/operators.md | 5 +-
.../main/scala/org/apache/comet/CometConf.scala | 3 +-
.../org/apache/spark/sql/comet/operators.scala | 12 ++
.../org/apache/comet/exec/CometJoinSuite.scala | 178 ++++++++++++++++++++-
5 files changed, 201 insertions(+), 5 deletions(-)
diff --git a/docs/source/user-guide/latest/compatibility/operators.md b/docs/source/user-guide/latest/compatibility/operators.md
index 1b9a61bf7f..8312fb520a 100644
--- a/docs/source/user-guide/latest/compatibility/operators.md
+++ b/docs/source/user-guide/latest/compatibility/operators.md
@@ -109,6 +109,14 @@ rows by the grouping keys only, and Spark and Comet can leave rows with equal ke
orders, so a group with more than one candidate value can return a different value than Spark.
Both results are valid under Spark's semantics for these functions.
+## Hash Join
+
+Spark plans `NOT IN` as a null-aware anti join, and the native hash join builds it on Spark's
+streamed side without keeping that side's order. When Spark reports the streamed side as ordered,
+for example the output of a window, a sort-merge join, or a cached table sorted within its
+partitions, Comet falls back to Spark for the join, so that a sort Spark removed above it on the
+strength of that order is not lost.
+
## Window Functions
Comet runs `WindowExec` natively and it is enabled by default (`spark.comet.exec.window.enabled`). A broad set of
diff --git a/docs/source/user-guide/latest/operators.md b/docs/source/user-guide/latest/operators.md
index aa4fd58967..566ec058f1 100644
--- a/docs/source/user-guide/latest/operators.md
+++ b/docs/source/user-guide/latest/operators.md
@@ -98,14 +98,15 @@ omitted from the tables below and may be reconsidered based on demand:
| Operator | Status | Notes |
| ----------------------------- | ------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| `BroadcastHashJoinExec` | ✅ | |
+| `BroadcastHashJoinExec` | ✅ | Falls back to Spark for a `NOT IN` (null-aware anti join) whose streamed side reports an ordering. |
| `ShuffledHashJoinExec` | ✅ | |
| `SortMergeJoinExec` | ✅ | Supports `BINARY` join keys. Nested-type (struct, array, map) and collated-string join keys fall back to Spark. |
| `BroadcastNestedLoopJoinExec` | ✅ | Falls back to Spark when the preserved side is broadcast (for example LEFT OUTER with BROADCAST on the left) ([#4429](https://github.com/apache/datafusion-comet/pull/4429)). |
`ExistenceJoin` (produced when `EXISTS` / `IN` is combined with another predicate via `OR`) runs
natively on `BroadcastHashJoinExec` and `ShuffledHashJoinExec`. Existence sort-merge joins, residual
-(non-equi) join conditions, computed (non-column) join keys, and `NOT IN` fall back to Spark.
+(non-equi) join conditions, and computed (non-column) join keys fall back to Spark. So does `NOT IN`
+combined with `OR`, which Spark plans as an existence `BroadcastNestedLoopJoinExec`.
## Exchanges
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala
index 4cf228e47e..a12654bf83 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -338,7 +338,8 @@ object CometConf extends ShimCometConf {
.doc(
"Whether to enable native ExistenceJoin, produced when EXISTS / NOT EXISTS / IN is " +
"combined with another predicate via OR. Sort-merge joins, residual (non-equi) join " +
- "conditions, computed (non-column) join keys, and NOT IN fall back to Spark.")
+ "conditions, computed (non-column) join keys, and NOT IN combined with OR fall back " +
+ "to Spark.")
.booleanConf
.createWithDefault(true)
val COMET_EXEC_AGGREGATE_ENABLED: ConfigEntry[Boolean] =
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
index 3ee11e4978..87268b5fc6 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala
@@ -2853,6 +2853,18 @@ trait CometHashJoin {
return None
}
+ // DataFusion builds a null-aware anti join on Spark's streamed side, and on its hash map path
+ // it emits the build batches last to first, so the output loses the ordering Spark reports
+ // for the join. Remove this fallback once Comet uses a DataFusion release that fixes
+ // https://github.com/apache/datafusion/issues/26226.
+ if (isNullAwareAntiJoin && join.outputOrdering.nonEmpty) {
+ withFallbackReason(
+ join,
+ "null-aware anti-join over an ordered streamed side is not supported natively, " +
+ "because the native join does not keep that order")
+ return None
+ }
+
val condition = join.condition.map { cond =>
val condProto = exprToProto(cond, join.left.output ++ join.right.output)
if (condProto.isEmpty) {
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala
index d96d4f7b8e..d0e1438ee4 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala
@@ -36,11 +36,12 @@ import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
import org.apache.spark.sql.catalyst.plans.Inner
import org.apache.spark.sql.catalyst.plans.logical.Join
import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometFilterExec, CometHashJoinExec, CometNativeScanExec, CometProjectExec, CometSortExec, CometSortMergeJoinExec, CometUnionExec, CometWindowExec}
-import org.apache.spark.sql.execution.{ColumnarToRowTransition, InputAdapter, LocalTableScanExec, SortExec, SparkPlan, WholeStageCodegenExec}
+import org.apache.spark.sql.execution.{ColumnarToRowTransition, InputAdapter, LocalTableScanExec, SortExec, SparkPlan, SQLExecution, WholeStageCodegenExec}
import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, AQEShuffleReadExec, QueryStageExec}
+import org.apache.spark.sql.execution.aggregate.SortAggregateExec
import org.apache.spark.sql.execution.datasources.SchemaColumnConvertNotSupportedException
import org.apache.spark.sql.execution.exchange.{ReusedExchangeExec, ShuffleExchangeLike}
-import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{ArrayType, IntegerType, MetadataBuilder, StructField, StructType}
@@ -1323,6 +1324,179 @@ class CometJoinSuite extends CometTestBase {
}
}
+ // Spark plans NOT IN as a null-aware anti join that streams the left side and reports that
+ // side's ordering, so Spark drops a sort above the join when the side arrives sorted. Here that
+ // side is a cached table, sorted within its partitions, that reaches the native plan through
+ // Comet's JVM to native conversion. Comet runs the join with that side as DataFusion's build
+ // side. Sparse keys make DataFusion build a hash map, which takes the batches last to first;
+ // dense integer keys would take its array map path instead, which keeps them in order.
+ private def withSortedCache(data: DataFrame)(f: => Unit): Unit = {
+ withSQLConf(
+ CometConf.COMET_CONVERT_FROM_IN_MEMORY_CACHE_ENABLED.key -> "true",
+ CometConf.COMET_BATCH_SIZE.key -> "64",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10485760",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "2") {
+ val cached = data.repartition(2, data("_1")).sortWithinPartitions("_1").cache()
+ try {
+ cached.count()
+ withTempView("streamed") {
+ cached.createOrReplaceTempView("streamed")
+ f
+ }
+ } finally {
+ cached.unpersist()
+ }
+ }
+ }
+
+ private def withNotInTables(f: => Unit): Unit = {
+ withParquetTable((0 until 10000).map(i => (i % 100 * 1000, i)), "big") {
+ withParquetTable((0 until 10).map(i => (i * 10000, i)), "small") {
+ f
+ }
+ }
+ }
+
+ /**
+ * Checks the query's rows against Spark, that no sort runs above a broadcast hash join, and
+ * that every partition is sorted by the query's first column, an integer. Returns the Comet
+ * plan.
+ */
+ private def checkSortedPartitions(query: => DataFrame): SparkPlan = {
+ val (_, plan) = checkSparkAnswer(query)
+ // Spark dropped the sort above the join, so the order comes from the plan below it.
+ val sorts = collect(plan) { case s @ (_: SortExec | _: CometSortExec) => s }
+ assert(
+ sorts.forall(_.find {
+ case _: BroadcastHashJoinExec | _: CometBroadcastHashJoinExec => true
+ case _ => false
+ }.isEmpty),
+ plan)
+ // Propagates the session's confs, such as the batch size, to the tasks as collect() does.
+ val parts = SQLExecution.withSQLConfPropagated(spark) {
+ query.queryExecution.toRdd.map(_.getInt(0)).glom().collect()
+ }
+ parts.zipWithIndex.foreach { case (keys, i) =>
+ assert(keys.sameElements(keys.sorted), s"partition $i is out of order\n$plan")
+ }
+ plan
+ }
+
+ private val orderedNotInFallback = "null-aware anti-join over an ordered streamed side"
+
+ for (adaptive <- Seq(false, true)) {
+ test(s"NOT IN over a sorted cached side falls back and keeps its order, AQE=$adaptive") {
+ withNotInTables {
+ withSortedCache(spark.table("big")) {
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString) {
+ val plan = checkSortedPartitions(
+ sql("SELECT _1 AS k, _2 AS v FROM streamed WHERE _1 NOT IN (SELECT _1 FROM small)")
+ .sortWithinPartitions("k"))
+ val joins = collect(plan) { case j: BroadcastHashJoinExec => j }
+ assert(joins.length == 1 && joins.head.isNullAwareAntiJoin, plan)
+ assert(collect(plan) { case j: CometBroadcastHashJoinExec => j }.isEmpty, plan)
+ assert(
+ new ExtendedExplainInfo()
+ .getFallbackReasons(plan)
+ .exists(_.contains(orderedNotInFallback)),
+ plan)
+ }
+ }
+ }
+ }
+ }
+
+ // Spark's sort aggregate trusts the same ordering, through an outer join that streams the NOT IN
+ // output, so a NOT IN that loses it splits groups. Comet aggregates are off so that Spark's sort
+ // aggregate runs over the joins, as it does for an aggregate Comet cannot convert.
+ test("NOT IN below an outer hash join keeps its groups and order under a GROUP BY") {
+ withNotInTables {
+ withParquetTable((0 until 50).map(i => (i * 2000, i)), "dim") {
+ withSortedCache(spark.table("big")) {
+ withSQLConf(
+ SQLConf.USE_OBJECT_HASH_AGG.key -> "false",
+ CometConf.COMET_EXEC_AGGREGATE_ENABLED.key -> "false") {
+ val plan = checkSortedPartitions(
+ sql("""SELECT s.k, count(*) AS n, sort_array(collect_list(d._2)) AS w
+ |FROM (SELECT _1 AS k FROM streamed WHERE _1 NOT IN (SELECT _1 FROM small)) s
+ |LEFT JOIN dim d ON s.k = d._1
+ |GROUP BY s.k""".stripMargin).sortWithinPartitions("k"))
+ assert(collect(plan) { case a: SortAggregateExec => a }.nonEmpty, plan)
+ }
+ }
+ }
+ }
+ }
+
+ test("NOT IN over an unordered streamed side runs natively") {
+ withNotInTables {
+ withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10485760") {
+ val (_, plan) = checkSparkAnswerAndOperator(
+ sql("SELECT _1, _2 FROM big WHERE _1 NOT IN (SELECT _1 FROM small)"))
+ val joins = collect(plan) { case j: CometBroadcastHashJoinExec => j }
+ assert(joins.length == 1 && joins.head.isNullAwareAntiJoin, plan)
+ assert(joins.head.outputOrdering.isEmpty, plan)
+ }
+ }
+ }
+
+ // With default settings the streamed side can be native from parquet. A window partitioned by
+ // `_1` sorts its input natively, and `k` carries `_1` through the window, so the NOT IN key is a
+ // window output that the optimizer cannot push below the sort. The ROWS frame keeps the window
+ // on DataFusion's streaming window operator, so the join's build side spans many batches.
+ private val windowedBig =
+ "SELECT _1, _2, min(_1) OVER (PARTITION BY _1 ORDER BY _2 ROWS UNBOUNDED PRECEDING) AS k " +
+ "FROM big"
+
+ for (adaptive <- Seq(false, true)) {
+ test(s"NOT IN over a native window side falls back and keeps its order, AQE=$adaptive") {
+ withNotInTables {
+ withSQLConf(
+ CometConf.COMET_BATCH_SIZE.key -> "64",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString) {
+ val plan = checkSortedPartitions(
+ sql(s"SELECT _1, _2 FROM ($windowedBig) t WHERE k NOT IN (SELECT _1 FROM small)")
+ .sortWithinPartitions("_1"))
+ assert(collect(plan) { case w: CometWindowExec => w }.nonEmpty, plan)
+ val joins = collect(plan) { case j: BroadcastHashJoinExec => j }
+ assert(joins.length == 1 && joins.head.isNullAwareAntiJoin, plan)
+ assert(collect(plan) { case j: CometBroadcastHashJoinExec => j }.isEmpty, plan)
+ // AQE plans the join again once the shuffle below the window completes. By then its
+ // build side is Spark's broadcast stage, so Comet skips the new join without a reason,
+ // and only the initial plan carries the reason the guard recorded.
+ if (!adaptive) {
+ assert(
+ new ExtendedExplainInfo()
+ .getFallbackReasons(plan)
+ .exists(_.contains(orderedNotInFallback)),
+ plan)
+ }
+ }
+ }
+ }
+ }
+
+ // The fallback only covers the null-aware join: a semi join over the same ordered side still
+ // runs natively, since DataFusion probes with that side and keeps its order.
+ test("LEFT SEMI join over the native window side used for NOT IN stays native") {
+ withNotInTables {
+ withSQLConf(CometConf.COMET_BATCH_SIZE.key -> "64") {
+ val plan = checkSortedPartitions(
+ sql(s"SELECT /*+ BROADCAST(small) */ t._1, t._2 FROM ($windowedBig) t " +
+ "LEFT SEMI JOIN small ON t.k = small._1").sortWithinPartitions("_1"))
+ val joins = collect(plan) { case j: CometBroadcastHashJoinExec => j }
+ assert(joins.length == 1 && !joins.head.isNullAwareAntiJoin, plan)
+ assert(collect(plan) { case j: BroadcastHashJoinExec => j }.isEmpty, plan)
+ assert(
+ !new ExtendedExplainInfo()
+ .getFallbackReasons(plan)
+ .exists(_.contains(orderedNotInFallback)),
+ plan)
+ }
+ }
+ }
+
test("BroadcastHashJoin with LeftAnti and NOT IN subquery (null-aware)") {
withSQLConf(
SQLConf.PREFER_SORTMERGEJOIN.key -> "false",
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org