comphead commented on PR #5421: URL: https://github.com/apache/datafusion-comet/pull/5421#issuecomment-5592903348
> **Note on this review:** generated with LLM assistance (Claude Code) while I worked through the PR. I have read the findings and agree with them, but flagging the tooling. Push back on anything that is wrong or already handled. Four things at the Major level. I checked the two load-bearing claims first and both hold up, so none of this is about the diagnosis: - The AVG `(null, 0)` claim is correct. `native/spark-expr/src/agg_funcs/avg.rs:143-152` has `sum: Option<f64>` and `state()` returns `ScalarValue::Float64(self.sum)`, while Spark's `Average.mergeExpressions` is the null-intolerant `sum.left + sum.right` over an initial `0.0`. - The COUNT direction argument is correct. Both rules the old docstring cites key on the **Final** being a `BaseAggregateExec`: `LogicalQueryStage.scala:58` (`computeStats` bumps a 0 row count to 1 for a global aggregate) and `AQEPropagateEmptyRelation.scala:58`. Keeping the Final in Spark preserves both, and the Partial's engine does not participate in either match. ### 1. The direction split defaults one direction to the other `supportsNativePartialToSparkFinal(fn) = supportsSparkPartialToNativeFinal(fn)` in `CometAggregateExpressionSerde.scala`. The premise of the split is that the two directions are independent, and AVG is the proof. With this default, non-decimal SUM, MIN, MAX, the three bitwise aggregates, HLL++ and `bloom_filter_agg` all claim forward-direction safety on the strength of an analysis that was only ever done for the reverse direction. The AVG defect is a per-function property, namely the state a never-updated accumulator emits, and nothing forces that question to be answered when a new aggregate opts in. I checked the least obvious one and it happens to be fine: `spark_bloom_filter.rs:282` makes `state_as_bytes()` an alias for `spark_serialization()`, and `bloom_filter_agg.rs:171` always emits `Binary(Some(..))`, so an empty partition does not produce the null that Spark's `BloomFilterAggregate.deserialize` would NPE on (it calls `BloomFilter.readFrom` with no null check). That reads as luck rather than design. Suggestion: default `supportsNativePartialToSparkFinal` to `false` and add explicit `= true` overrides for the functions that were actually checked, recording the empty-partition state in each comment. That is a handful of one-line overrides, and it makes the next contributor answer the right question instead of inheriting an answer to a different one. ### 2. `revertChain` bails out silently, leaving the plan it exists to prevent The trailing `case _ => None` in `revertChain` feeds `getOrElse(agg)`, so an unrecognised node between the Spark Final and the native Partial ships the unsafe boundary with no signal at all. I do not think the earlier reply covers the proposal that was made. An unconditional warning would indeed fire on the benign q10/q35 shape where the Partial is already Spark, but gating the signal on `findCometPartialAgg(agg.child)` returning a Comet Partial that `revertChain` failed to reach is false on exactly those cases. The two traversals accept different node sets: `findCometPartialAgg` handles `AQEShuffleReadExec` and `ShuffleQueryStageExec`, `revertChain` handles `CometSinkPlaceHolder` and `CometShuffleExchangeExec`. Any future divergence between them is a live crash, and today it is undetectable. ### 3. Fourth overlapping mechanism for one invariant The Comet-Partial-to-Spark-Final boundary is now guarded in four places: `tagUnsafePartialAggregates`, `preserveSparkAggregateBuffers`, the new `revertUnsafePartialAggregates`, and `RevertNativeForTransitionHeavyStages.hasUnsafeMixedAggregateAtStageBoundary`. All of them are still needed as written, since the Celeborn one also covers a Comet Final over a fallen-back Comet exchange, which the new pass does not match (it only matches a Spark `BaseAggregateExec` Final). But the first three duplicate the chain walk and the reason string, and they descend through different node sets, which is the same drift risk as finding 2. At minimum `preserveSparkAggregateBuffers` should call the shared `revertChain` rather than keeping its own recursive `restore`. ### 4. The reverted Partial gets the tag but no fallback reason `revertChain` calls `setTagValue(COMET_UNSAFE_PARTIAL, ...)` without the matching `withFallbackReason`, unlike the sibling at `CometExecRule.scala:228-229`. The reason is only recovered when the re-entrant `transform` reaches `doConvert`, which requires the Partial's children to all be `CometNativeExec`. That holds in every new test, since there is always a `CometFilterExec` below, so the gap is untested. Where the Partial's child is not native, `explainFallback` and `EXTENDED` output silently lose the explanation for the one node this PR changes. One line next to the `setTagValue`. -- 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]
