ctagard commented on PR #59403: URL: https://github.com/apache/spark/pull/59403#issuecomment-6254757864
Thanks @HyukjinKwon. The pandas finding, the three nits and the three decision challenges are addressed, one commit each: - **Pandas UDF skip decided before ColumnPruning** (f4d62f0): as suggested, `pushBelowAggregate` now makes the whole split (the reference check moved into it), and below an Aggregate with any `PythonUDAF` it pushes only the conjuncts that normalizing leaves unchanged; the rest stay above. The CTE mirror uses the same split. Tests cover both shapes you described (mixed aggregate with the SQL aggregate pruned returns the -0.0 group; all-pandas aggregate with the UDF pruned returns only matching rows), plus plan and CTE tests. Plans for other Aggregates are unchanged. - **LIMIT 1** (ecd3565): the Aggregate is kept and `LocalLimit(1)` is still pushed below it. - **RemoveRedundantAggregates** (3391cc2): the lower Aggregate is kept only if the upper one would read its float key raw (through a lower output computed from the key, inside an aggregate function, or in a computed grouping expression). Your `SELECT d, max(a) FROM (SELECT DISTINCT d, a FROM t) GROUP BY d` removes it again; the end-to-end test checks the rows and that one Aggregate remains. Case 2 stays conservative, with its test comment corrected. - **Streaming** (1a7627c): I checked it. With the guard, your query planned two stateful aggregations, and restarting from a checkpoint written with the previous plan failed with `STREAMING_STATEFUL_OPERATOR_NOT_MATCH_IN_STATE_METADATA`. Keeping a sensitive filter above a streaming aggregate has the same effect, because the Filter stops `RemoveRedundantAggregates` from merging the two aggregates. So streaming Aggregates now behave exactly as before: the removal rules are unchanged, and filters are pushed below them unnormalized (a normalized one would disagree with a streaming aggregate those rules still replace with a raw-key Project). A restart from a previous-plan checkpoint works, and `StreamingAggregationSuite` pins one stateful operator for both shapes. Streaming keeps the existing normalization behavior (including SPARK-60093) as a possible follow-up; I can file it. - **Nits:** the array/map comment is corrected, with an array-key test (44ec4a4); the RRA test is restored with an integer key, with the double-key case as its own test (7483323); blank lines fixed (2517918). - Also, `InSet` and IN subquery operands are now exempt like the other comparisons, so they are pushed unchanged (1f941a0); `PredicateSuite` checks `InSet` for raw and normalized values under interpreted and generated evaluation. Locally, `catalyst/test`, plan stability (no golden changes), `SQLQueryTestSuite`, the streaming aggregation, deduplication and multi-stateful-operator suites, and the related `sql/core` suites pass (Python tests with Python 3.11, pandas and pyarrow). The PR description is updated. -- 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]
