Yaoxuan Wu created SPARK-60487:
----------------------------------
Summary: Parquet/ORC aggregate pushdown returns NULL instead of 0
for COUNT when no file is read
Key: SPARK-60487
URL: https://issues.apache.org/jira/browse/SPARK-60487
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 4.1.1, 4.2.0, 3.5.8, 4.0.0
Reporter: Yaoxuan Wu
{code:python}
spark.conf.set("spark.sql.sources.useV1SourceList", "") # DSv2 file source
spark.conf.set("spark.sql.parquet.aggregatePushdown", "true")
spark.createDataFrame([(1, 1.0), (2, 2.0)], "p INT, v DOUBLE") \
.write.mode("overwrite").partitionBy("p").parquet("/tmp/agg_pd")
spark.read.parquet("/tmp/agg_pd").where("p >
100").selectExpr("count(*)").collect()
# expected: [Row(count(1)=0)]
# actual: [Row(count(1)=None)]
{code}
COUNT is never NULL, so this is a wrong result: e.g. {{WHERE (SELECT count(*)
...) < x}} drops every row. The same for {{count(v)}}, for ORC
({{spark.sql.orc.aggregatePushdown}}), and for an empty Parquet table without
any filter. Correct with aggregate pushdown off or with the DSv1 file source.
Cause: the file sources take the pushed COUNT only partially (one count per
file), so {{V2ScanRelationPushDown}} keeps a final aggregate over the scan and
rewrites {{Count}} into {{Sum}} of the per-file counts. When every partition is
pruned (or the table has no file) the scan returns no rows, and a global
{{Sum}} over no rows is NULL. Wrapping it as {{coalesce(sum(...), 0)}} for
COUNT would fix it.
{code}
HashAggregate(keys=[], functions=[sum(agg_func_0)])
+- BatchScan parquet ... PartitionFilters: [isnotnull(p), (p > 100)],
PushedAggregation: [COUNT(*)]
{code}
Reproduces on 3.5.8, 4.0.0, 4.1.1 and 4.2.0.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
