[SPARK-58378][SQL] Fuse compatible approximate percentile sketches

[sunchao]

Why are the changes needed?

This addresses 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:

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:

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:

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)

[sunchao]

cc @viirya @cloud-fan @dongjoon-hyun @peter-toth

[viirya]

Nice optimization, and the correctness reasoning holds up under scrutiny. I traced the sketch semantics and the rewrite end to end.

The core equivalence is sound because the digest is fully decoupled from the requested percentile: ApproximatePercentile is deterministic (no Random/seed, and QuantileSummaries — Greenwald-Khanna — has no randomness), so three scalar percentile_approx(x, {0.5,0.9,0.95}, acc) build three bit-identical digests, and querying one shared digest three times equals querying three identical digests once each. The getPercentiles(array(...)) path preserves input order, so array element index maps exactly to the percentile at expressions(index), and GetArrayItem(combined, index) extracts it. I checked the GK algorithm is order-sensitive in its intermediate state, but that's a non-issue here: fusion doesn't change the input sequence, partitioning, or the partial/final merge topology any buffer sees — it just collapses N identical computations into one — so the fused digest is bit-identical to the originals, not merely within the error bound.

A few specific things I verified:

• Empty/null input: eval returns null for both the scalar and array cases (the whole array is null, not an array of nulls), and GetArrayItem(null, i) short-circuits to null via BinaryExpression.eval, so the fused result matches the original scalar null.
• Result type/nullability: the original scalar type is child.dataType; the fused ArrayType(child.dataType) unwrapped by GetArrayItem gives back child.dataType, unchanged.
• Idempotency: after fusion the inner aggregate is array-valued, so the percentageExpression.dataType == DoubleType guard excludes it on a second pass.
• The child.deterministic / filter.deterministic guard is exactly the right scope. The sketch itself is deterministic, so the only way fusion could change results is a non-deterministic input expression (e.g. percentile_approx(x + rand(), ...)), where collapsing N aggregates into one would evaluate rand() once instead of N times. Excluding those is the precise condition, and structural (not semantic) comparison of inputs/filters keeps FP-associativity and ANSI-overflow behavior intact — it can only miss fusions, never make an unsafe one.

Excluding streaming to preserve checkpoint value schemas is the right call, and the optimizer test suite is thorough — it targets each of these risk areas directly (duplicate/non-monotonic order, nested expressions, structural input/filter equivalence, differently-associated FP, distinct, empty/null, existing array percentiles, streaming).

One non-blocking thought: the rule is a standalone Once batch, so it won't re-fire if a later fixed-point batch ever exposes newly-fusible percentiles. That's almost certainly fine (percentiles rarely materialize late in optimization) and cheaper — just confirming Once is the intended choice rather than folding it into an existing fixed-point batch.

LGTM.