[jira] [Created] (SPARK-60501) [SQL] PullupCorrelatedPredicates returns 0 instead of 1 for COUNT(*) over a correlated scalar aggregate

["Cheng Wang (Jira)" <ji...@apache.org>]

Cheng Wang created SPARK-60501:
----------------------------------

             Summary:  [SQL] PullupCorrelatedPredicates returns 0 instead of 1 for COUNT(*) over a correlated scalar aggregate
                 Key: SPARK-60501
                 URL: https://issues.apache.org/jira/browse/SPARK-60501
             Project: Spark
          Issue Type: Bug
          Components: SQL
    Affects Versions: 4.0.0
         Environment:  macOS / Linux, PySpark / Spark 4.0.0 and master branch
            Reporter: Cheng Wang


h3. What happened?

PullupCorrelatedPredicates.rewriteSubQueries uses mayHaveCountBugAgg (sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:488-513) to determine if a correlated scalar aggregate needs count-bug domain join handling:

{code:scala}
def mayHaveCountBugAgg(a: Aggregate): Boolean = {
  a.groupingExpressions.isEmpty && a.aggregateExpressions.exists(_.exists {
    case a: AggregateExpression => a.aggregateFunction.defaultResult.isDefined
    case _ => false
  })
}
{code}

defaultResult.isDefined is true only for COUNT-like functions. However, ANY ungrouped aggregate (including MAX, MIN, SUM, AVG) always produces exactly one row on empty input (evaluating to NULL).

When mayHaveCountBugAgg returns false, the inner aggregate is decorrelated without the LeftOuter domain join, so non-matching outer keys produce zero rows instead of one row (NULL). An enclosing COUNT(*) then counts 0 instead of 1.

h3. To Reproduce (PySpark 4.0.0 / master)

{code:sql}
CREATE OR REPLACE TEMP VIEW a AS SELECT * FROM VALUES (1, 10), (2, 20) AS a(k, x);
CREATE OR REPLACE TEMP VIEW b AS SELECT * FROM VALUES (1, 100), (1, 200) AS b(k, v);

SELECT a.k, (
    SELECT COUNT(*)
    FROM (SELECT MAX(b.v) AS m FROM b WHERE b.k = a.k) sub
) AS cnt
FROM a
ORDER BY a.k;
{code}

h3. Expected vs Actual

For k = 2, table b has no matching rows. The scalar subquery SELECT MAX(b.v) FROM b WHERE b.k = 2 produces 1 row: (NULL). The outer COUNT(*) over this 1 row must return 1.

* Expected:
{noformat}
+---+---+
|  k|cnt|
+---+---+
|  1|  1|
|  2|  1|
+---+---+
{noformat}

* Actual in Spark 4.0.0 / master:
{noformat}
+---+---+
|  k|cnt|
+---+---+
|  1|  1|
|  2|  0|  <-- Silent Wrong Result
+---+---+
{noformat}

h3. Root Cause

mayHaveCountBugAgg checks aggregateFunction.defaultResult.isDefined instead of checking whether the aggregate is ungrouped. An ungrouped aggregate always evaluates to 1 row regardless of whether the function returns NULL or a default value.

h3. Suggested Fix

Treat every ungrouped subquery aggregate as vulnerable, or check whether the parent expression is sensitive to row existence (such as COUNT(*)).



--
This message was sent by Atlassian Jira
(v10.3.26#10030026)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org