sunchao opened a new pull request, #57576: URL: https://github.com/apache/spark/pull/57576
### Why are the changes needed? This addresses [SPARK-58378](https://issues.apache.org/jira/browse/SPARK-58378). An approximate percentile sketch summarizes a distribution. Once the sketch exists, Spark can read p50, p90, p95, or any other requested percentile from the same state. Nevertheless, a query that writes those percentiles as separate scalar aggregates currently constructs and maintains a separate sketch for each one. For example: ```sql SELECT service, percentile_approx(latency_ms, 0.50, 10000) AS p50, percentile_approx(latency_ms, 0.90, 10000) AS p90, percentile_approx(latency_ms, 0.95, 10000) AS p95 FROM request_metrics GROUP BY service; ``` All three aggregates receive the same values and use the same accuracy. The percentile only affects how the completed sketch is queried. Despite that, Spark currently allocates three aggregate buffers per group, inserts every qualifying input into all three, serializes three partial sketches, and merges three sketches during final aggregation. Callers can avoid that work by rewriting the query to use `percentile_approx(latency_ms, array(0.50, 0.90, 0.95), 10000)` and extracting individual array elements. However, requiring that rewrite is awkward for existing SQL, generated reports, named scalar output columns, and expressions that consume individual percentile values. ### What changes were proposed in this pull request? Teach Catalyst to recognize scalar approximate percentile aggregates that describe the same distribution, calculate that distribution once, and extract each original scalar result from the shared array-valued aggregate. For the example above, the resulting aggregation is equivalent to: ```sql percentile_approx(latency_ms, array(0.50, 0.90, 0.95), 10000) ``` Catalyst returns elements `0`, `1`, and `2` under the original `p50`, `p90`, and `p95` expressions. Callers do not change their query. Both partial and final aggregation maintain one percentile sketch instead of three. Fusion is intentionally limited to deterministic scalar aggregates with the same input expression, evaluated accuracy, aggregation mode, distinctness, and filter. Inputs and filters are compared structurally so that floating-point evaluation and ANSI overflow behavior do not change. Accuracy is compared after evaluation, which remains correct when `ConstantFolding` is excluded. Existing array-valued aggregates, incompatible distributions, and Structured Streaming aggregates are left unchanged; excluding streaming preserves existing checkpoint state schemas. ### Does this PR introduce _any_ user-facing change? No SQL syntax, public API, output schema, or percentile result changes. Eligible batch queries use fewer percentile aggregation buffers and can perform less sketch update, serialization, shuffle, and merge work. ### How was this patch tested? Added Catalyst optimizer tests, end-to-end physical-plan and result tests, a Spark 5 `TIME` regression, and streaming checkpoint recovery tests. The regression coverage includes distinct and filtered aggregates, differently associated floating-point expressions, ANSI overflow, null and empty inputs, grouping sets, decimal and temporal types, aliases, repeated percentile values, pre-existing array-valued aggregates, and differently evaluated accuracy expressions. All 54 focused tests and all three Scala style checks pass: ```text build/sbt \ 'catalyst/testOnly org.apache.spark.sql.catalyst.optimizer.CombineApproximatePercentilesSuite' \ 'sql/testOnly org.apache.spark.sql.ApproximatePercentileQuerySuite' \ 'sql/testOnly org.apache.spark.sql.streaming.StreamingAggregationSuite -- -z "approximate percentiles preserve existing streaming checkpoints"' \ 'catalyst/scalastyle' \ 'sql/scalastyle' \ 'sql/Test/scalastyle' ``` ### Was this patch authored or co-authored using generative AI tooling? Generated-by: OpenAI Codex (GPT-5) -- 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]
