sunchao commented on code in PR #4818:
URL: https://github.com/apache/datafusion-comet/pull/4818#discussion_r4104816189
##########
spark/src/main/scala/org/apache/comet/serde/aggregates.scala:
##########
@@ -1116,6 +1116,44 @@ object CometApproxCountDistinct extends
CometAggregateExpressionSerde[HyperLogLo
}
}
+object CometKurtosis extends CometAggregateExpressionSerde[Kurtosis] {
+
+ // Not marked safe for mixed partial/final: follows the same policy as
`Variance` / `Stddev`,
+ // whose complex `[n, avg, m2, ...]` buffer is not certified compatible
across engines. The
+ // native accumulator does mirror Spark's `[n, avg, m2, m3, m4]` wire
format, so lifting this
+ // to `true` should be considered together with the other `CentralMomentAgg`
serdes.
+
+ override def convert(
+ aggExpr: AggregateExpression,
+ kurtosis: Kurtosis,
+ inputs: Seq[Attribute],
+ binding: Boolean,
+ conf: SQLConf): Option[ExprOuterClass.AggExpr] = {
+ val child = kurtosis.child
+ val childExpr = exprToProto(child, inputs, binding)
+
+ if (childExpr.isDefined) {
+ val builder = ExprOuterClass.Kurtosis.newBuilder()
+ builder.setChild(childExpr.get)
+ builder.setNullOnDivideByZero(kurtosis.nullOnDivideByZero)
+ // Spark's evaluate expression divides by `m2 * m2`, and that `Divide`
picks up its eval
+ // mode from the session. `m2` can be non-zero while `m2 * m2`
underflows to zero, which
+ // the `m2 === 0` guard above it does not catch, so the native side
needs to know whether
+ // that divisor should raise or return null.
+ builder.setAnsiEnabled(conf.ansiEnabled)
Review Comment:
[P2] Preserve the aggregate’s captured division mode. Create a SQL temporary
view containing `kurtosis(v)` over Parquet doubles `1e-100` and `2e-100` with
ANSI disabled, then enable ANSI and query the view. Spark retains the `LEGACY`
mode captured by the `Divide` in `Kurtosis.evaluateExpression` and returns
`NULL`. This line instead serializes the current `conf.ansiEnabled=true`,
making native evaluation raise `DIVIDE_BY_ZERO`. Reversing the settings
silently returns `NULL` where Spark raises. This changes saved-view semantics
depending on whether Comet executes the aggregate. Derive the flag from the
captured `Divide` mode and add a regression covering both transitions.
Evidence: Reproduced with Spark 4.1.3, `local[1]`, AQE disabled, and a
one-partition Parquet input containing doubles `1e-100` and `2e-100`. Register
it as `k_input`, then run: `SET spark.sql.ansi.enabled=false; CREATE OR REPLACE
TEMP VIEW k_saved AS SELECT kurtosis(v) k FROM k_input; SET
spark.sql.ansi.enabled=true; SELECT * FROM k_saved`. The result was
`Row(k=None)`. Both physical HashAggregate nodes reported `SESSION_ANSI True`
and captured division mode `LEGACY`. Reversing the settings reported
`SESSION_ANSI False`, captured mode `ANSI`, and raised `DIVIDE_BY_ZERO`.
`operators.scala:1798` passes that aggregate configuration to this serializer,
and `planner.rs:3135` forwards the resulting flag to the UDAF. The exact-head
Rust test `divisor_underflow_follows_spark_division_semantics` passed locally
and verifies that these same inputs return NULL for a false flag and raise for
a true flag. Full Comet JVM execution was not run 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]
