sunchao commented on code in PR #6835:
URL: https://github.com/apache/datafusion-comet/pull/6835#discussion_r4251452268
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -2876,12 +2921,44 @@ trait CometHashJoin {
else OperatorOuterClass.BuildSide.BuildRight)
.setNullAwareAntiJoin(isNullAwareAntiJoin)
.setDynamicFilterEnabled(CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ENABLED.get(join.conf))
+ .addAllStreamedSortOrders(streamedSortOrders.asJava)
condition.foreach(joinBuilder.setCondition)
Some(builder.setHashJoin(joinBuilder).build())
} else {
None
}
}
+
+ // Whether a Comet join that may move its streamed rows, while reporting
Spark's order for
+ // them, feeds the order that `plan` reports with no sort in between.
Whether an outer join
+ // declared the order on its probe input is up to the native planner, so
every one built that
+ // way counts. A cache reports the order of the plan that wrote it, so the
walk goes on into
+ // that plan. While CometExecRule converts a plan, a native plan's JVM
inputs sit behind a
+ // CometScanWrapper, a leaf, so the walk looks through that too.
+ private def mayReportFalseOrder(plan: SparkPlan): Boolean = plan match {
+ case _ if plan.outputOrdering.isEmpty => false
+ case _: SortExec | _: CometSortExec => false
+ case join: CometBroadcastHashJoinExec
+ if join.isNullAwareAntiJoin || runsAsRightOuterJoin(join.joinType,
join.buildSide) =>
+ true
+ case join: CometHashJoinExec if runsAsRightOuterJoin(join.joinType,
join.buildSide) => true
+ case _: CometBroadcastNestedLoopJoinExec => true
+ case wrapper: CometScanWrapper => mayReportFalseOrder(wrapper.originalPlan)
+ case scan: CometInMemoryTableScanExec =>
+ mayReportFalseOrder(scan.originalPlan.relation.cachedPlan)
Review Comment:
[P1] Inspect the finalized adaptive cache producer before accepting its
order. With AQE enabled, materialize a native `NOT IN` over a sorted cache into
another cache, then broadcast-left-join that result and `GROUP BY k` without an
intervening exchange. Spark’s `InMemoryTableScanExec` derives ordering from the
finalized adaptive `executedPlan`, but this traversal passes the raw
`AdaptiveSparkPlanExec` to `mayReportFalseOrder`. That wrapper reports no
ordering, so line 2939 returns `false` before reaching the native anti join.
The outer join consequently declares an order over unordered rows and enables
sorted aggregation, splitting keys into multiple groups. The bounded
resulting-plan reproduction returned 49 groups for 37 keys, versus 37 correct
groups without the declaration. Unwrap finalized adaptive plans before the
empty-order shortcut, mirroring Spark’s effective cached-plan selection, and
add an AQE-enabled cached-producer regression.
Evidence: Fresh Spark 3.5.9 probe
`/tmp/review6835-final-o8rmfzw9/metadata.py` confirmed: materialized cache scan
ordering `k ASC`; raw cached plan `AdaptiveSparkPlan`, empty ordering, zero
children, `isFinalPlan=true`; finalized producer ordering `k ASC`. The
consuming outer join retained that order and `SinglePartition`, with two
aggregates and no exchange above it. Recompiled the exact-head wrapper into
`/tmp/review6835-final-o8rmfzw9/cache_boundary`: 100,000 sorted integer rows,
`k=(i%40)*1000000`, batches of 8,192, excluding 1000000, 10000000, 20000000 and
absent 100000000 through a null-aware anti join, then materializing,
outer-joining even keys and grouping. No declaration returned 37 groups of
2,500 rows. Declaration returned 49 groups for 37 keys, including key 3000000
split into counts 692 and 1808. An actual-sort control returned 37 correct
groups. Relevant cached DataFusion 55.1.0 join/aggregate sources were verified
byte-identical to locked 55.2.0. This was separate
Spark metadata and native execution validation, not an end-to-end Comet JVM
run.
--
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]
