Yibo Dong created SPARK-59609:
---------------------------------
Summary: Parquet aggregate pushdown reads statistics from the
wrong column after schema merging
Key: SPARK-59609
URL: https://issues.apache.org/jira/browse/SPARK-59609
Project: Spark
Issue Type: Bug
Components: Optimizer, 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
With Parquet schema merging, aggregate pushdown can read statistics from the
wrong physical column when one file is missing the aggregated column.
This causes silently incorrect aggregate results.
h2. How to reproduce
Run the following SQL:
{code:sql}
CREATE TABLE with_value
USING PARQUET
LOCATION '${spark.sql.warehouse.dir}/with_value'
AS
SELECT id, value, other
FROM VALUES
(1, 10, 100),
(2, 20, 200)
AS t(id, value, other);
CREATE TABLE without_value
USING PARQUET
LOCATION '${spark.sql.warehouse.dir}/without_value'
AS
SELECT id, other
FROM VALUES
(3, 300),
(4, 400)
AS t(id, other);
CREATE OR REPLACE TEMP VIEW t
USING PARQUET
OPTIONS (
path '${spark.sql.warehouse.dir}',
mergeSchema 'true',
recursiveFileLookup 'true'
);
SELECT id, value, other
FROM t
ORDER BY id;
SELECT count(*) AS rows_all FROM t;
SELECT count(value) AS rows_value FROM t;
SELECT min(value) AS lo FROM t;
SELECT max(value) AS hi FROM t;
EXPLAIN EXTENDED
SELECT count(value), min(value), max(value)
FROM t;
{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
After schema merging, Spark reads the rows as:
|| id || value || other ||
| 1 | 10 | 100 |
| 2 | 20 | 200 |
| 3 | NULL | 300 |
| 4 | NULL | 400 |
The file created by {{without_value}} does not contain the {{value}} column, so
the merged schema should supply {{NULL}} for {{value}} in those rows.
Both configurations should therefore produce:
|| Aggregate || Result ||
| COUNT(*) | 4 |
| COUNT(value) | 2 |
| MIN(value) | 10 |
| MAX(value) | 20 |
h2. Actual result
With {{spark.sql.parquet.aggregatePushdown=true}}:
|| Aggregate || Result ||
| COUNT(*) | 4 |
| COUNT(value) | *4* |
| MIN(value) | 10 |
| MAX(value) | *400* |
With {{spark.sql.parquet.aggregatePushdown=false}}:
|| Aggregate || Result ||
| COUNT(*) | 4 |
| COUNT(value) | 2 |
| MIN(value) | 10 |
| MAX(value) | 20 |
With aggregate pushdown enabled, the physical plan reports:
{code}
PushedAggregation: [COUNT(value), MIN(value), MAX(value)]
{code}
With aggregate pushdown disabled, it reports:
{code}
PushedAggregation: []
{code}
The pushed-down result counts rows from the file where {{value}} is absent and
returns {{400}} as {{MAX(value)}}. The value {{400}} belongs to the {{other}}
column in that file.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
