Yibo Dong created SPARK-59607:
---------------------------------
Summary: Parquet aggregate pushdown incorrectly counts NULL
partition values in COUNT(partition_column)
Key: SPARK-59607
URL: https://issues.apache.org/jira/browse/SPARK-59607
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Environment: {code}
Spark: 5.0.0-SNAPSHOT
Commit: 9b4e4f7547c07c5eba3d2b0822a6d3dbb023a150
Java: OpenJDK 17.0.17
OS: macOS 15.7.4
{code}
Reporter: Yibo Dong
h2. What happened
For partitioned Parquet data, {{COUNT(p)}} incorrectly includes a {{NULL}}
partition value when aggregate pushdown is enabled.
With aggregate pushdown disabled, the same query returns the expected result.
h2. How to reproduce
Run the following SQL with {{spark-sql}}:
{code:sql}
INSERT OVERWRITE DIRECTORY '${spark.sql.warehouse.dir}/tp/p=a'
USING PARQUET
SELECT 1 AS v;
INSERT OVERWRITE DIRECTORY
'${spark.sql.warehouse.dir}/tp/p=**HIVE_DEFAULT_PARTITION**'
USING PARQUET
SELECT 2 AS v;
INSERT OVERWRITE DIRECTORY '${spark.sql.warehouse.dir}/tp/p=b'
USING PARQUET
SELECT 3 AS v;
CREATE OR REPLACE TEMP VIEW tp
USING PARQUET
OPTIONS (
path '${spark.sql.warehouse.dir}/tp',
basePath '${spark.sql.warehouse.dir}/tp'
);
SELECT p, v FROM tp ORDER BY v;
SELECT count(*) AS count_star FROM tp;
SELECT count(v) AS count_v FROM tp;
SELECT count(p) AS count_p FROM tp;
SELECT count(DISTINCT p) AS distinct_p FROM tp;
EXPLAIN EXTENDED SELECT count(p) FROM tp;
{code}
Run it once with Parquet aggregate pushdown enabled:
{code:bash}
spark-sql
--master 'local[2]'
--conf spark.ui.enabled=false
--conf spark.sql.adaptive.enabled=false
--conf spark.sql.shuffle.partitions=1
--conf spark.sql.sources.useV1SourceList=
--conf "spark.sql.warehouse.dir=$(mktemp -d)"
--conf spark.sql.parquet.aggregatePushdown=true
-f repro.sql
{code}
Then run it again with a fresh warehouse and aggregate pushdown disabled:
{code:bash}
spark-sql
--master 'local[2]'
--conf spark.ui.enabled=false
--conf spark.sql.adaptive.enabled=false
--conf spark.sql.shuffle.partitions=1
--conf spark.sql.sources.useV1SourceList=
--conf "spark.sql.warehouse.dir=$(mktemp -d)"
--conf spark.sql.parquet.aggregatePushdown=false
-f repro.sql
{code}
h2. Expected result
Spark reads the three rows as:
{code}
(a, 1)
(NULL, 2)
(b, 3)
{code}
The {{p=**HIVE_DEFAULT_PARTITION**}} directory is interpreted as a SQL {{NULL}}
partition value.
Both configurations should therefore produce:
|| Aggregate || Result ||
| COUNT(*) | 3 |
| COUNT(v) | 3 |
| COUNT(p) | 2 |
| COUNT(DISTINCT p) | 2 |
In particular, {{COUNT(p)}} should exclude the {{NULL}} partition value.
h2. Actual result
With {{spark.sql.parquet.aggregatePushdown=true}}:
|| Aggregate || Result ||
| COUNT(*) | 3 |
| COUNT(v) | 3 |
| COUNT(p) | *3* |
| COUNT(DISTINCT p) | 2 |
With {{spark.sql.parquet.aggregatePushdown=false}}:
|| Aggregate || Result ||
| COUNT(*) | 3 |
| COUNT(v) | 3 |
| COUNT(p) | 2 |
| COUNT(DISTINCT p) | 2 |
With aggregate pushdown enabled, the physical plan contains:
{code}
PushedAggregation: [COUNT(p)]
{code}
With aggregate pushdown disabled, it contains:
{code}
PushedAggregation: []
{code}
Thus, enabling Parquet aggregate pushdown changes {{COUNT(p)}} from the
expected value {{2}} to {{3}}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
