andygrove commented on PR #4870: URL: https://github.com/apache/datafusion-comet/pull/4870#issuecomment-5319268733
Up front: I used an LLM to help work through this review, so please push back on anything that looks wrong or off base. Thanks for taking this on. I traced the rank arithmetic in `rank_limit.rs` against Spark's `RankLimitIterator`, `DenseRankLimitIterator`, and `SimpleLimitIterator` and I think it is right. Spark checks `rank < limit` before calling `increaseRank()`, so Spark over-emits by one tie group (RANK over `[10,10,10,20]` with limit 2 emits the 20). Comet computes the exact rank and drops it, which is fine because `InsertWindowGroupLimit` only fires when there is a Filter on the rank above the Window. The degenerate cases all fall out correctly too. The test coverage is also much better than I expected, especially running every query under `windowGroupLimitThreshold=-1,1000` so the pushdown and non-pushdown shapes are both checked, and covering the SPARK-46526 `PARTITION BY t2c ORDER BY t2c` shape. q44 going from 6 transitions down to 2 is great. A few things I would like to see addressed. 1. `equals` and `hashCode` on `CometWindowGroupLimitExec` (lines 140 and 151) leave out the rank-like function. The case class carries `partitionSpec`, `orderSpec`, and `limit` but not the function, and `nativeOp` is not compared either, so two nodes differing only in `RowNumber` versus `Rank` compare equal. `sameResult` and `canonicalized` bottom out in `equals`, and those drive `ReuseSubquery` and the cache manager, so I think two otherwise-identical subqueries using different rank functions could be wrongly reused. `CometWindowExec` includes `windowExpression` in its `equals` for the same reason. Could `rankLikeFunction` go into the case class and both overrides? Spark's `mode` is dropped too, and adding it to `stringArgs` would help, since right now the Partial and Final nodes render identically and that makes the q44 golden file hard to read. 2. A duplicated partition key becomes a hard native error rather than a fallback. `LexOrdering::new` dedups by `PhysicalExpr` keeping the first occurrence, so for `PARTITION BY a, a ORDER BY a` all three collapse to one entry, `partition_prefix_len` is 2 while `expr.len()` is 1, and `try_new` returns `DataFusionError::Internal` (`rank_limit.rs:80`). `tryConvertToComet` has no try/catch, so that fails the query instead of falling back. It is contrived SQL but generated SQL does produce shapes like it. Rather than guarding, would it be cleaner to pass the partition-by and order-by lists to the operator separately and build the two `RowConverter`s from those, using the `LexOrdering` only for `compute_properties` and `required_input_ordering`? That would remove the guard, the failure mode, and the three comment blocks at `planner.rs:2400-2432` that exist purely to reason about the dedup. The `partition_prefix_len` slicing is the only thing coupling correctness to a DataFusion implemen tation detail and I do not think it has to. 3. The `-0.0` versus `0.0` divergence needs more than a test comment. `window_group_limit_rank.sql` says it outright: the FP section keeps the cutoff above the zero values to avoid exercising it. Spark's `SQLOrderingUtil.compareDoubles` is `if (x == y) 0 else Double.compare(x, y)`, so `-0.0 == 0.0` and `RankLimitIterator` ties them, while the Arrow row encoder compares bytes and splits them. With `rank() = 1` over `[0.0, -0.0, 1.0]` in one partition Spark keeps two rows and Comet keeps one. `compatibility/floating-point.md` already has a general positive/negative-zero caveat so this fits an existing family, but nothing a user reads about `WindowGroupLimitExec` points at it. Could we get a tracking issue, a line in `floating-point.md`, and the test the file currently declines to write, marked `ignore(<issue>)` if it fails? Leaving the only record of a known wrong-answer case inside a comment in the test that avoids it means nobody finds it later. 4. `ShimCometWindowGroupLimit.extract` throws `IllegalStateException` on an unrecognized rank-like function, in both the 3.5 and 4.x copies. It is unreachable today since `InsertWindowGroupLimit` only produces the three, but `tryConvertToComet` does not catch, so a future Spark adding a fourth would fail a query that used to work. Could it return `None` and let `convert` record a fallback reason instead? 5. Three hand-edited doc pages still say this is unsupported and need updating here, since `GenerateDocs` does not touch them. `user-guide/latest/operators.md:106` still marks `WindowGroupLimitExec` as 🔜 with "falls back today", `user-guide/latest/compatibility/operators.md:74` still says "not yet supported and falls back to Spark", and `contributor-guide/roadmap.md:33` lists it as pending. 6. `rank_limit.rs` has no unit tests. It is 378 lines with cross-batch state in `prev_partition`, `prev_order`, `rank`, and `count`, and the SQL test tables top out around 16 rows so everything lands in a single batch. The only real multi-batch coverage is indirect via Verify TPC-DS. Could we add tests for a partition boundary landing exactly on a batch boundary, a limit reached mid-batch with a new partition starting in the next one, a batch that filters to zero rows, and a RANK tie run spanning batches? 7. `PartitionedRankLimitExec` never implements `metrics()`, so the `numOutputRows` that Spark's `WindowGroupLimitExec` reports shows up as zero even though `CometWindowGroupLimitExec` inherits `baselineMetrics`. `scan.rs:218` wires up `BaselineMetrics` the same way, so it should be easy to match. 8. There is no early exit once a partition's limit is reached (`rank_limit.rs:324`). `this_rank` is monotonic within a partition, so once `keep` goes false every remaining row in that partition is dropped, yet the loop still encodes the ORDER BY key and does a row comparison for each. Spark's `GroupedLimitIterator.skipRemainingRows` skips that work explicitly. For the skewed-partition small-K case this operator exists to optimize, that is the hot path, and only the partition-key compare is needed once the limit is hit. 9. Two comments describe things that do not exist. `PartitionedTopKExec` in the `planner.rs:2426` comment is not anywhere in the repo, so a reader cannot check the comparison. And the `required_input_ordering` comment at `rank_limit.rs:156` says declaring the requirement lets `EnforceSorting` insert a `SortExec` if the sort got dropped, but we do not run any DataFusion physical optimizer rules on plans built from the serialized Spark plan, so there is no safety net there. Correctness rests entirely on the JVM-side plan, and I would rather the comment say that, since it is exactly what a reviewer needs to trust. 10. The six test files are under `sql-tests/expressions/window/`, which holds window expression tests like `lag_lead.sql`. `sql-tests/windows/` is where operator-level window tests live (`window_functions.sql`), so I think they belong there. 11. The PR description is still all template comments. We generate changelogs from these, so could you fill in the rationale and summary? And since this operator adds per-row Arrow row-encoding on top of what Spark's `UnsafeProjection` grouping already does, some before/after numbers on q44, q67, and q70 would confirm the win the plan-stability diff implies. -- 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]
