sunchao commented on code in PR #7155:
URL: https://github.com/apache/datafusion-comet/pull/7155#discussion_r4344266566
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -1176,11 +1174,46 @@ case class CometExecRule(session: SparkSession,
queryStagePrep: Boolean = false)
serde.convert(op, builder)
}
}
- return nativeOp.map { nativeOp =>
- val exec = serde.createExec(nativeOp, op)
- rollUpInfoMessages(op, exec)
- exec
- }
+ return nativeOp
+ .filter { proto =>
+ // A serde may emit several native nodes for one Spark operator.
Inspect those nodes,
+ // but stop at real Spark children, whose consumption context was
tagged independently.
+ def containsOwnAssertion(node: Operator): Boolean =
+
CometAssertNotNull.containsNativeAssertion(node.toBuilder.clearChildren().build())
||
+ node.getChildrenList.asScala.exists { child =>
+ child.getPlanId == proto.getPlanId &&
containsOwnAssertion(child)
+ }
+
+ def hasNativeAssertion: Boolean = op match {
+ // TopK's sink proto omits its runtime projection over the
selected rows.
+ case topK: TakeOrderedAndProjectExec =>
+ QueryPlanSerde.withSharedSubexpressions(topK.projectList) {
+ topK.projectList.exists { expression =>
+ QueryPlanSerde
+ .exprToProto(expression, topK.child.output)
+ .exists(CometAssertNotNull.containsNativeAssertion)
+ }
+ }
+ case _: HashAggregateExec | _: ObjectHashAggregateExec =>
+ // Input/grouping expressions consume all rows; only the result
projection is lazy.
+ proto.hasProjection &&
CometAssertNotNull.containsNativeAssertion(
+ proto.getProjection)
+ case _ => containsOwnAssertion(proto)
+ }
+ op.getTagValue(CometExecRule.PARTIAL_READER) match {
Review Comment:
[P2] Preserve early-exit semantics in semi/anti join conditions. This guard
only checks assertions when an external consumer supplied `PARTIAL_READER`, but
a broadcast semi/anti join also stops evaluating candidates after its first
match. For a `BuildRight` join with left IDs `[0]`, right IDs `[0, 1]`, and
Catalyst condition `IF(l.id = r.id, true, AssertNotNull(NULL::BOOLEAN))`, Spark
returns `[0]` for LEFT SEMI and no rows for LEFT ANTI without evaluating the
assertion. The native join evaluates the second candidate and throws
`NOT_NULL_ASSERT_VIOLATION`. `If` is admitted by the new expression guard, and
the join receives no partial-reader tag, so the previously unsupported
assertion now enables this failing native path. Please decline native
assertions in these early-exit join conditions until their candidate-level
short-circuit behavior is preserved.
Evidence: Ran the four exact-head native assertion unit tests successfully.
A disposable Rust test used the exact-head `assert_not_null.rs`, locked
DataFusion 55.2.0, batches `[0]` and `[0,1]`, a lazy CASE equivalent to the IF
condition, and `NestedLoopJoinExec::try_new(...).swap_inputs()` matching
Comet's BuildRight planner. Both LeftSemi and LeftAnti returned
NOT_NULL_ASSERT_VIOLATION. A Spark 4.1.3 JVM probe tested interpreted and
generated predicates with `buildRows.exists`, the consumer used by
`BroadcastNestedLoopJoinExec.leftExistenceJoin`: both stopped after one
candidate and succeeded. Evaluating the skipped candidate separately raised the
assertion. SimplifyConditionals, NullPropagation, ConstantFolding, and
PushPredicateThroughJoin retained the conditional assertion. Sources: Spark
v4.1.3 BroadcastNestedLoopJoinExec.scala:221–224, Comet
predicates.scala:505–508, and
native/core/src/execution/planner/join.rs:309–318. Reproduction files and logs
are under `/tmp/come
t7155-review/probe/`. A full Comet JVM query was not executed locally.
--
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]
