Xiaoxuan Li created SPARK-58888:
-----------------------------------
Summary: OptimizeExpand turns an empty input into one row, making
COUNT(DISTINCT) return 1 instead of 0
Key: SPARK-58888
URL: https://issues.apache.org/jira/browse/SPARK-58888
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 4.2.0
Reporter: Xiaoxuan Li
This is 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.
Distinct of a constant over an empty input returns 1 instead of 0. No constant
literal has to appear in the query text: it is enough that the optimizer can
prove the distinct columns constant, which it does for the common pattern of
tagging constant dimensions in a subquery, view, or UNION ALL branch.
----
*Repro*
{code:sql}
-- an empty source, e.g. a daily partition with no rows yet
CREATE OR REPLACE TEMP VIEW events AS
SELECT * FROM VALUES (1, 'x', 'y') AS t(id, c1, c2) WHERE 1 = 0;
SET spark.sql.optimizer.optimizeExpandRatio=2;
SELECT COUNT(DISTINCT region), COUNT(DISTINCT channel)
FROM (SELECT 'US' AS region, 'web' AS channel, id FROM events);
-- returns [1,1], expected [0,0]
{code}
Minimal form of the same thing:
{code:sql}
SELECT COUNT(DISTINCT 1), COUNT(DISTINCT 2) FROM events;
-- returns [1,1], expected [0,0]
{code}
Excluding the rule by name gives the correct answer, so the rule is the cause
and not the surrounding rewrite:
{code:sql}
SET
spark.sql.optimizer.excludedRules=org.apache.spark.sql.catalyst.optimizer.OptimizeExpand;
-- returns [0,0]
{code}
----
*Cause*
{\{OptimizeExpand}} collects the pre-aggregate's grouping expressions from the
attributes the \{{Expand}} references (\{{collectPreAggGroupBy}}):
{code:java}
expand.child.output.filter(expand.references.contains)
{code}
With no \{{GROUP BY}} and every distinct aggregate over a constant,
\{{Expand.references}} is empty, so the inserted node is \{{Aggregate(Nil, Nil,
child)}} -- a *global* aggregate, which emits one row even for empty input. The
\{{Expand}} then sees one row where it previously saw none:
{code:java}
Aggregate [count(1) FILTER (gid=2) AS count(DISTINCT region), count(1) FILTER
(gid=1) AS count(DISTINCT channel)]
+- Aggregate ['web', 'US', gid], [gid]
+- Expand [[web, null, 1], [null, US, 2]], ['web', 'US', gid]
+- Aggregate <- inserted by OptimizeExpand: no
grouping, no aggregates
+- LocalRelation <empty>
{code}
Note this is the same empty-table failure that
\{{RewriteDistinctAggregates.mustRewrite}} exists to prevent. Its comment names
\{{SELECT COUNT(DISTINCT 1)}} and "wrong results when working with empty
tables" as the reason the rewrite is forced for this shape; \{{OptimizeExpand}}
then undoes that protection.
----
*Trigger conditions*
The distinct arguments only need to be _foldable_, not literal, so ordinary
column references reach this state once constant projections are folded through:
||shape||result||
|\{{COUNT(DISTINCT c1), COUNT(DISTINCT c2)}} over constant-projected
columns|*wrong*|
|\{{COUNT(DISTINCT 1), COUNT(DISTINCT 2)}}|*wrong*|
|\{{COUNT(DISTINCT 1+1), COUNT(DISTINCT CAST('2020-01-01' AS DATE))}}|*wrong*|
|three or more constant distincts|*wrong*|
|non-empty input|correct (the answer is 1 either way)|
|one distinct argument is a real column|correct (the column enters the
pre-aggregate's GROUP BY)|
|query has a \{{GROUP BY}}|correct|
So all of the following must hold: no \{{GROUP BY}}, at least two distinct
groups (fewer than two leaves the \{{Expand}} below the minimum threshold of
2), every distinct argument foldable, empty input, and the conf explicitly set.
----
*Relationship to SPARK-58387*
Independent. There the inner \{{Aggregate}} computes a duplicate-sensitive
aggregate; here it computes no aggregate at all, so the duplicate-sensitivity
guard proposed in SPARK-58387 does not reject this plan. Reproduced on a build
that already carries that fix.
*Suggested fix*
Reject the rewrite when the pre-aggregate's grouping would be empty, since a
global aggregate is not a row-preserving de-duplication -- e.g. \{{if
(preAggGroupBy.isEmpty) return false}} in \{{canOptimize}}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
