Josh Rosen created HIVE-29771:
---------------------------------
Summary: OptimizeExpand inserts a de-duplicating pre-aggregate
under a non-distinct COUNT, returning the distinct-row count
Key: HIVE-29771
URL: https://issues.apache.org/jira/browse/HIVE-29771
Project: Hive
Issue Type: Bug
Affects Versions: 4.2.0
Reporter: Josh Rosen
This is a report of a correctness bug in the {{OptimizeExpand}} rule introduced
by SPARK-56315 (released in 4.2.0). The rule is gated by the internal conf
{{spark.sql.optimizer.optimizeExpandRatio}} (default {{-1}} = disabled), so
hitting it requires explicit opt-in.
----
Background:
* queries with multiple {{COUNT(DISTINCT)}} on different columns are rewritten
by {{RewriteDistinctAggregates}} into an {{Expand}} that duplicates each input
row once per distinct group.
* {{OptimizeExpand}} reduces that amplification by inserting a de-duplicating
{{Aggregate}} on (grouping keys + distinct columns) beneath the {{{}Expand{}}}.
* The de-duplication is sound only for pure distinct aggregates (collapsing
duplicate rows changes the value of any duplicate-sensitive aggregate computed
alongside) and the [conf's own
doc|https://github.com/apache/spark/blob/5ce51a624421acceec378e0170da24a8fc2a96eb/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L3003-L3020]
states exactly that precondition: _"Only applies to pure distinct aggregates
without non-distinct aggregates or FILTER clauses."_
The bug: the guard implementing that precondition tests {_}attributes{_}, not
_aggregate functions_
([OptimizeExpand.scala|https://github.com/apache/spark/blob/5ce51a624421acceec378e0170da24a8fc2a96eb/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeExpand.scala#L106]):
{code:java}
val innerGroupByAttrs =
AttributeSet(innerAgg.groupingExpressions.flatMap(_.references))
if (!expand.producedAttributes.subsetOf(innerGroupByAttrs)) return false {code}
The reasoning is that a non-distinct aggregate would force an Expand output
column outside the inner GROUP BY. That only detects aggregates to which
{{RewriteDistinctAggregates}} assigns a dedicated Expand slot.
Any aggregate that reaches the inner Aggregate as {{count(1)}} —
{{{}COUNT(1){}}}, {{{}COUNT(*){}}}, and {{COUNT(col)}} of a non-nullable column
(which the rewrite normalizes to {{{}count(1){}}}) — references no attribute at
all and always passes the guard:
||SQL shape||after RewriteDistinctAggregates||dedicated Expand
slot?||guard||result||
|{{COUNT(1)}} / {{COUNT(*)}}|{{count(1)}}|no|passes|*wrong*|
|{{{}COUNT(a){}}}, {{a}} non-nullable|normalized to
{{count(1)}}|no|passes|*wrong*|
|{{{}COUNT(a){}}}, {{a}} nullable|{{count(a)}} over its own
slot|yes|rejects|correct|
|{{SUM(a)}}|{{sum(a)}} over its own slot|yes|rejects|correct|
{*}Bug repro{*}:
{code:java}
CREATE TABLE oe USING parquet AS SELECT * FROM VALUES
(1,5,7),(1,5,7),(1,5,7),(1,6,8),(2,9,9) AS t(k,a,b);
SET spark.sql.optimizer.optimizeExpandRatio=2;
SELECT k, COUNT(DISTINCT a), COUNT(DISTINCT b), COUNT(1) FROM oe GROUP BY k
ORDER BY k;
-- outputs: [1,2,2,2], [2,1,1,1] WRONG (expected [1,2,2,4]: k=1 has four
rows) {code}
{{}}
{{}}
{{k=1}} has four rows (three copies of {{(1,5,7)}} plus {{{}(1,6,8){}}}), so
{{COUNT(1)}} must be 4; the returned 2 is the number of distinct {{(k,a,b)}}
rows.
{{COUNT(*)}} behaves identically. With the conf unset (or the rule excluded)
the answer is correct.
The optimized plan shows the inserted pre-aggregate (last {{{}Aggregate [k, a,
b]{}}}) feeding the Expand whose downstream {{count(1)}} was supposed to count
base rows:
{code:java}
Aggregate [k], [k, count(a) FILTER (gid=1), count(b) FILTER (gid=2),
coalesce(first(count(1)) FILTER (gid=0), 0)]
+- Aggregate [k, a, b, gid], [k, a, b, gid, count(1)]
+- Expand [[k,null,null,0], [k,a,null,1], [k,null,b,2]], [k, a, b, gid]
+- Aggregate [k, a, b], [k, a, b] <- inserted by OptimizeExpand
+- Relation oe[k,a,b] parquet {code}
{{}}
{{}}
With the rule excluded, the Expand's child is the relation itself and
{{count(1)}} is 4.
To fix this: check the inner aggregate's aggregate expressions rather than its
grouping references — reject any {{AggregateExpression}} that is non-distinct
and not duplicate-agnostic (cf. {{{}EliminateDistinct.isDuplicateAgnostic{}}},
the check {{RemoveRedundantAggregates}} uses for the same soundness question).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
