comphead commented on PR #5421:
URL:
https://github.com/apache/datafusion-comet/pull/5421#issuecomment-5690517929
> **Note on this review:** generated with LLM assistance (Claude Code) while
I worked through the second round. I have read the findings and agree with
them, but flagging the tooling. Push back on anything that is wrong or already
handled.
Second pass, against `fe66f74643`. My four points from the last round are
addressed in the code and the direction split reads well now. The per-function
opt-in comments each state the empty-partition state, which is the right
question to be making a contributor answer, and the two I spot-checked hold up.
Sharing `restoreSparkPartial` with the Celeborn path while keeping the two
walks separate is a good resolution to the duplication point, and having the
new execution tests check both the initial and the materialized AQE plan is
exactly where this class of bug hides.
Three things at the Major level. Nothing I found makes results wrong, so
this is about scope, diagnostics, and coverage rather than the fix itself.
Two of the three concern `fe66f74643`. That commit has an empty body and no
comment on the thread, and it removes several things that were added in
`6c14796f5` and `cb22f68c4` in response to the last round. I am raising the
underlying gaps below rather than the commit, but if the removals were
deliberate I would like to understand the reasoning, and if the rebase ate them
then two of these resolve themselves.
### 1. The fallback is wider than the evidence, and nothing tracks the
remainder
Outside the nine opt-ins, every aggregate now loses its native Partial
whenever the Final stays in Spark. The sharpest case is decimal `SUM`.
`CometSum.supportsNativePartialToSparkFinal` (`aggregates.scala:290`) and
`supportsSparkPartialToNativeFinal` (`aggregates.scala:294`) are now
character-for-character the same expression with different comments, and the
new one excludes decimal without saying why. The reverse-direction comment
blames overflow detection not surviving the split, which @andygrove already
noted describes a Comet Final rather than this direction.
Looking at the native side, `sum_decimal.rs:112-120` declares `(sum:
nullable decimal, is_empty: Boolean non-null)` matching Spark's `(sum,
isEmpty)`, and `state()` at line 294 always emits both, so an empty partition
produces `(null, true)`. Spark's decimal merge is `coalesce(coalesce(sum.left,
zero) + sum.right, sum.left)` with `isEmpty.left && isEmpty.right`, which
absorbs that cleanly. The cost is visible in this PR's own diff: `SUM(_8)` and
`SUM(_9)` in `CometAggregateSuite` move from `expectedNumOfCometAggregates` to
`expectedNumOfDecimalAggregates`, so 1 native aggregate becomes 0 when
`nativeShuffleEnabled` is false, and that test was already asserting results as
well as plan shape. Is there a failure mode in this direction I am missing, or
can decimal SUM opt in?
The same question applies to the tail. `FIRST`, `LAST`, and the whole
stddev/variance/covariance/correlation family were measured as matching Spark
on main in this configuration in @andygrove's earlier sweep, and
`variance.rs:100-117` emits `(count, mean, m2)` as three `Float64` values
against Spark's `(n, avg, m2)` three doubles. I am not claiming those are safe,
only that nobody has checked and the conservative default now costs a native
Partial for all of them, plus `MinBy`/`MaxBy`, `Mode`, and `Percentile`. #5420
restores AVG but does not touch any of these. Could you file an issue to audit
the remaining functions' emitted state for this direction and link it here, so
that the default is a staging point rather than the end state? It would also
help for the description to say plainly that this is a measurable
native-aggregation regression for `spark.comet.exec.shuffle.enabled=false`,
since that is the configuration the original report came from.
### 2. `revertChain` is back to bailing out silently
`CometExecRule.scala:1176` is `case _ => None` and line 1187 is
`.getOrElse(agg)`, so an unrecognised node between the Spark Final and the
native Partial ships the unsafe boundary with no signal. The conditional
diagnostic added in `6c14796f5` was removed in `fe66f74643`.
The traversal sets still diverge, and I can name a shape that reaches this.
`CometExchangeSink.createExec` builds `CometSinkPlaceHolder(nativeOp, op, op)`
where `op` is the `ShuffleQueryStageExec` (`CometSink.scala:164-165`), so
```
HashAggregateExec(Final) // Spark, function outside
the allowlist
CometHashAggregateExec(PartialMerge) // native, matches
revertChain's second arm
CometSinkPlaceHolder(_, _, ShuffleQueryStageExec(...))
```
matches the `Partial`/`PartialMerge` arm, recurses into the placeholder,
falls to `case _ => None`, and leaves the native `PartialMerge` feeding the
Spark Final with nothing logged and nothing on `EXPLAIN`. Refusing to rewrite a
stage there is right. I could not construct a query that reaches it and I do
not think you could either, which reads to me as the argument for the
diagnostic rather than against it. Any future divergence between
`findPartialAggInPlan` and `revertChain` is a live crash and is currently
undetectable.
### 3. Coverage the removed tests were carrying
Three gaps, all opened by `fe66f74643`:
- Nothing covers `restoreSparkPartial` recording its reason when the
restored Partial's child is not native. Every remaining test has a
`CometFilterExec` or `CometProjectExec` below, so the explanation is always
recovered by the re-entrant `transform` reaching `doConvert`. The
`withFallbackReason` line that makes it independent of that is the one thing
the last round asked for here, and it is now untested.
- Nothing pins the direction-split invariant. Opting a function into
`supportsSparkPartialToNativeFinal` grants nothing in the other direction
today, but no test fails if the two are re-collapsed.
- `unsafe aggregate buffers fall back when native shuffle is ineligible`
lost its `"prior shuffle fallback"` case, so it now varies only
`COMET_SHUFFLE_NATIVE_HASH_PARTITIONING_ENABLED` and no longer exercises a
sticky prior refusal.
Related, `revertUnsafePartialAggregates` went from `private[rules]` back to
`private`, so the pass can no longer be driven directly from a test. Could that
go back along with the cases above?
---
CI is fully green across all five Spark profiles, Iceberg, the Spark SQL
suites, and TPC-DS/TPC-H verification, so none of this is about the build.
--
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]
