James Xu created SPARK-60318:
--------------------------------
Summary: Allow Expand pre-aggregation for filtered distinct
aggregates
Key: SPARK-60318
URL: https://issues.apache.org/jira/browse/SPARK-60318
Project: Spark
Issue Type: Improvement
Components: SQL
Affects Versions: 5.0.0
Reporter: James Xu
Problem:
Queries with multiple distinct aggregates use an Expand operator that
duplicates each input row once per distinct group. Spark can optionally insert
a de-duplication aggregate before Expand when the projected attributes are all
grouping keys of the following aggregate. This is useful for duplicate-heavy
inputs because it reduces the number of rows exposed to Expand amplification.
SPARK-56898 rewrites multiple conditional counts over the same base column into
FILTER form so they share one distinct group. For example, an LLM serving
rollup may count distinct requests under several priority or tier predicates:
SELECT
user_id,
COUNT(DISTINCT IF(priority = 1, request_id, NULL)) AS high_priority_requests,
COUNT(DISTINCT IF(tier = 2, request_id, NULL)) AS premium_requests,
COUNT(DISTINCT model) AS models_used,
COUNT(DISTINCT region) AS regions_used,
COUNT(DISTINCT endpoint) AS endpoints_used
FROM serving_log
GROUP BY user_id
Before the conditional-count rewrite, this query has five distinct groups and
can use pre-aggregation before a 5x Expand. After the rewrite, the first two
counts share one request_id group, reducing the Expand factor to 4x. However,
that branch also carries two FILTER flag columns, which makes the whole Expand
ineligible for pre-aggregation. A duplicate-heavy input therefore changes from
pre-aggregation followed by 5x Expand to a 4x Expand over the raw input.
Depending on the duplicate ratio, this can be a net regression.
Root Cause:
RewriteDistinctAggregates projects each distinct FILTER condition as a flag
column and computes max(flag) in the inner aggregate. The Expand
pre-aggregation rule currently requires every Expand-produced attribute to be
part of the inner aggregate grouping keys. Filter flags fail that structural
check even though they are deterministic functions of the pre-aggregation keys
and max(flag) is insensitive to duplicate multiplicity.
The rule also compares the number of raw child attributes referenced by Expand
with the number of inner grouping slots. After conditional-count normalization,
predicate columns such as priority and tier remain necessary to compute the
flags but are no longer distinct grouping keys. This size heuristic must
account for the safe filter-flag slots, or allowing the flags alone would still
leave the example ineligible.
Solution:
Allow Expand pre-aggregation when the only non-grouping Expand outputs are
Boolean FILTER flags consumed by duplicate-insensitive max aggregates.
Pre-aggregation groups by every raw attribute referenced by Expand, so
identical input tuples have identical flag values. Removing duplicate tuples
therefore preserves both max(flag) and the surrounding distinct aggregate
results.
Keep the optimization conservative:
- continue rejecting aggregates whose results depend on input multiplicity,
such as non-distinct SUM or COUNT;
- account for accepted filter-flag slots in the pre-aggregation size heuristic;
- preserve the existing opt-in behavior for Expand pre-aggregation and do not
change the default.
Expected Impact:
For the example above, the optimized plan can retain pre-aggregation and also
benefit from the conditional-count collapse: pre-aggregate the raw input once,
then apply a 4x Expand. Compared with the pre-normalization plan, this reduces
the expanded input from 5x the de-duplicated row count to 4x that count.
Compared with the current normalized plan, it avoids applying the 4x Expand to
every raw input row when duplicates are common.
The benefit is data-dependent and largest when the number of distinct tuples
referenced by Expand is much smaller than the raw input row count. There is no
expected change when Expand pre-aggregation is disabled.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
