[
https://issues.apache.org/jira/browse/SPARK-59607?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18118036#comment-18118036
]
Peifeng Li commented on SPARK-59607:
------------------------------------
I'd like to work on this issue. I've reproduced it in both Parquet and ORC.
> 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
> Priority: Major
>
> 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]
