You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Apache Spark (Jira)" <ji...@apache.org> on 2021/07/02 14:08:00 UTC
[jira] [Commented] (SPARK-35985) File source V2 ignores partition
filters when empty readDataSchema
[ https://issues.apache.org/jira/browse/SPARK-35985?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17373557#comment-17373557 ]
Apache Spark commented on SPARK-35985:
--------------------------------------
User 'steven-aerts' has created a pull request for this issue:
https://github.com/apache/spark/pull/33191
> File source V2 ignores partition filters when empty readDataSchema
> ------------------------------------------------------------------
>
> Key: SPARK-35985
> URL: https://issues.apache.org/jira/browse/SPARK-35985
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 3.0.0
> Reporter: Steven Aerts
> Priority: Major
>
> A V2 datasource fails to rely on partition filters when it only wants to know how many entries there are, and is not interested of their context.
> So when the {{readDataSchema}} of the {{FileScan}} is empty, partition filters are not pushed down and all data is scanned.
> Some examples where this happens:
> {code:java}
> scala> spark.sql("SELECT count(*) FROM parq WHERE day=20210702").explain
> == Physical Plan ==
> *(2) HashAggregate(keys=[], functions=[count(1)])
> +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [id=#136]
> +- *(1) HashAggregate(keys=[], functions=[partial_count(1)])
> +- *(1) Project
> +- *(1) Filter (isnotnull(day#68) AND (day#68 = 20210702))
> +- *(1) ColumnarToRow
> +- BatchScan[day#68] ParquetScan DataFilters: [], Format: parquet, Location: InMemoryFileIndex[file:/..., PartitionFilters: [], PushedFilers: [IsNotNull(day), EqualTo(day,20210702)], ReadSchema: struct<>, PushedFilters: [IsNotNull(day), EqualTo(day,20210702)]
> scala> spark.sql("SELECT input_file_name() FROM parq WHERE day=20210702").explain
> == Physical Plan ==
> *(1) Project [input_file_name() AS input_file_name()#131]
> +- *(1) Filter (isnotnull(day#68) AND (day#68 = 20210702))
> +- *(1) ColumnarToRow
> +- BatchScan[day#68] ParquetScan DataFilters: [], Format: parquet, Location: InMemoryFileIndex[file:/..., PartitionFilters: [], PushedFilers: [IsNotNull(day), EqualTo(day,20210702)], ReadSchema: struct<>, PushedFilters: [IsNotNull(day), EqualTo(day,20210702)]
> {code}
>
> Once the {{readDataSchema}} is not empty, it works correctly:
> {code:java}
> scala> spark.sql("SELECT header.tenant FROM parq WHERE day=20210702").explain
> == Physical Plan ==
> *(1) Project [header#51.tenant AS tenant#199]
> +- BatchScan[header#51, day#68] ParquetScan DataFilters: [], Format: parquet, Location: InMemoryFileIndex[file:/..., PartitionFilters: [isnotnull(day#68), (day#68 = 20210702)], PushedFilers: [IsNotNull(day), EqualTo(day,20210702)], ReadSchema: struct<header:struct<tenant:string>>, PushedFilters: [IsNotNull(day), EqualTo(day,20210702)]{code}
>
> In V1 this optimization is available:
> {code:java}
> scala> spark.sql("SELECT count(*) FROM parq WHERE day=20210702").explain
> == Physical Plan ==
> *(2) HashAggregate(keys=[], functions=[count(1)])
> +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [id=#27]
> +- *(1) HashAggregate(keys=[], functions=[partial_count(1)])
> +- *(1) Project
> +- *(1) ColumnarToRow
> +- FileScan parquet [year#15,month#16,day#17,hour#18] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[file:/..., PartitionFilters: [isnotnull(day#17), (day#17 = 20210702)], PushedFilters: [], ReadSchema: struct<>{code}
> The examples use {{ParquetScan}}, but the problem happens for all File based V2 datasources.
> The fix for this issue feels very straight forward. In {{PruneFileSourcePartitions}} queries with an empty {{readDataSchema}} are explicitly excluded from being pushed down:
> {code:java}
> if filters.nonEmpty && scan.readDataSchema.nonEmpty =>{code}
> Removing that condition seems to fix the issue however, this might be too naive.
> I am making a PR with tests where this change can be discussed.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org