Yaoxuan Wu created SPARK-60486:
----------------------------------
Summary: DSv2 file source silently drops a Python UDF filter on a
partition column
Key: SPARK-60486
URL: https://issues.apache.org/jira/browse/SPARK-60486
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}
from pyspark.sql import functions as F
from pyspark.sql.types import BooleanType
spark.conf.set("spark.sql.sources.useV1SourceList", "") # DSv2 file source
spark.range(5).selectExpr("id AS v", "CAST(id AS INT) AS p") \
.write.mode("overwrite").partitionBy("p").parquet("/tmp/udf_part")
gt2 = F.udf(lambda x: x is not None and x > 2, BooleanType())
spark.read.parquet("/tmp/udf_part").filter(gt2("p")).collect()
# expected: v in [3, 4]
# actual: v in [0, 1, 2, 3, 4]
{code}
The filter is in the analyzed plan but gone from the optimized plan, which is
just the {{RelationV2}} scan ({{PartitionFilters: []}}, no
{{BatchEvalPython}}). The same with ORC and with Arrow UDFs. Correct with the
DSv1 file source, with a built-in predicate {{p > 2}}, and with a UDF on a data
column.
Cause: {{FileScanBuilder.pushFilters}} treats a filter that references only
partition columns as handled by partition pruning and does not return it as a
post-scan filter. Since SPARK-48666 (3.5.2), a partition filter containing a
{{PythonUDF}} is kept out of {{partitionFilters}} (the file index cannot
evaluate it), but it is not returned as a post-scan filter either, so
{{V2ScanRelationPushDown}} removes the Filter and nothing evaluates it.
Returning those filters as post-scan filters would fix it.
Reproduces on 3.5.8, 4.0.0, 4.1.1 and 4.2.0 (3.5.1 and earlier fail with
INTERNAL_ERROR instead, which SPARK-48666 fixed).
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
