sunchao commented on code in PR #6180:
URL: https://github.com/apache/datafusion-comet/pull/6180#discussion_r4315619757
##########
spark/src/main/scala/org/apache/comet/serde/predicates.scala:
##########
@@ -491,3 +491,35 @@ object ComparisonUtils {
}
}
}
+
+object CometAtLeastNNonNulls extends CometExpressionSerde[AtLeastNNonNulls] {
+ override def convert(
+ expr: AtLeastNNonNulls,
+ inputs: Seq[Attribute],
+ binding: Boolean): Option[Expr] = {
+ // Native evaluation, like Spark's, stops counting a row once it reaches
n, which takes at
+ // least the first n children. So it may skip the children after those,
which a shared
+ // subexpression that Spark eliminates would evaluate for every row.
+ val skippable = if (expr.n <= 0) expr.children else
expr.children.drop(expr.n)
+ if (!canSkipEvaluation(skippable)) {
Review Comment:
[P2] [P2] Preserve shared-expression matching across decimal promotion
With ANSI mode and subexpression elimination enabled, project
`AtLeastNNonNulls(2, Seq(Literal(1), marker, r, r))`, where `r = Cast(Divide(a,
b), IntegerType)` and `a` and `b` are decimal columns.
`withSharedSubexpressions` records the original shared cast, but `exprToProto`
runs `DecimalPrecision.promote` before this check. The skippable child becomes
`Cast(CheckOverflow(Divide(...)))`, so it no longer matches the original shared
node and this fallback is bypassed. For `(marker=1, a=1, b=0)`, Spark evaluates
the shared expression and raises `DIVIDE_BY_ZERO`, while Comet reaches the
threshold before the division and returns `true`. Please compare both sides in
the same representation, or perform the guard against the original expression
tree, and add a repeated-decimal regression.
Evidence: Reproduced on the pinned head with Spark 4.1.3 and Temurin JDK 21,
using native-lib-linux from exact-head CI run 39165588736. A disposable
CometTestBase suite wrote Parquet rows (Some(1), BigDecimal(1), BigDecimal(0))
and (None, BigDecimal(7), BigDecimal(1)), then projected the expression above.
The plan contained CometProject over CometNativeScan. Output:
spark_error=Some([DIVIDE_BY_ZERO] ...), comet_error=None,
native_rows=ArraySeq([true], [true]). Setting
spark.comet.expression.AtLeastNNonNulls.enabled=false produced DIVIDE_BY_ZERO
in both engines. Setting spark.sql.subexpressionElimination.enabled=false
produced matching native results. The regression failed its error-parity
assertion. Saved reproduction:
/tmp/comet-6180-dd8ce9-review-source/Review6180DecimalCSESuite.scala. Log:
/tmp/comet-6180-dd8ce9-repro-controls.log.
--
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]
