[SPARK-37220][SQL] Do not split input file for Parquet reader with aggregate push down #34498
Conversation
|
@huaxingao, @sunchao and @viirya - could you help take a look when you have time? Thanks. |
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
Outdated
Show resolved
Hide resolved
| override def isSplitable(path: Path): Boolean = { | ||
| // If aggregate is pushed down, only the file footer will be read once, | ||
| // so file should not be split across multiple tasks. | ||
| pushedAggregate.isEmpty |
|
Kubernetes integration test starting |
|
Kubernetes integration test starting |
|
Kubernetes integration test status failure |
|
Kubernetes integration test status failure |
|
Test build #144942 has finished for PR 34498 at commit
|
|
Test build #144944 has finished for PR 34498 at commit
|
sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
Outdated
Show resolved
Hide resolved
|
Test build #144945 has finished for PR 34498 at commit
|
|
Kubernetes integration test starting |
|
Kubernetes integration test status failure |
|
Test build #144948 has finished for PR 34498 at commit
|
|
Thank you @sunchao, @huaxingao and @viirya for review! |
|
Quick question on:
How did the existing tests pass before this PR? |
| override def isSplitable(path: Path): Boolean = { | ||
| // If aggregate is pushed down, only the file footer will be read once, | ||
| // so file should not be split across multiple tasks. | ||
| pushedAggregate.isEmpty |
| // footers for every split of the file. Basically if the start (the beginning of) | ||
| // the offset in PartitionedFile is 0, we will read the footer. Otherwise, it means | ||
| // that we have already read footer for that file, so we will skip reading again. | ||
| if (file.start != 0) return null |
There was a problem hiding this comment.
Quick question on: Existing unit test in FileSourceAggregatePushDownSuite.scala. How did the existing tests pass before this PR?
@HyukjinKwon - I think we are in the same page based on your latest comment, but just to be noisy here in case anything is missing. Before this PR, when a single file is split into multiple splits across multiple tasks, we have the logic here to only process the split of file if file.start == 0, so only the first split of file will be processed, and every file is processed only once. So here is the trick. Before this PR, the logic for Parquet aggregate push down was still correct. We want to avoid unnecessary file splitting so update logic in this PR here.
6 participants
What changes were proposed in this pull request?
As a followup of https://github.com/apache/spark/pull/34298/files#r734795801, Similar to ORC aggregate push down, we can disallow split input files for Parquet reader as well. See original comment for more details of motivation. Also fix the string of
RowDataSourceScanExecto only print outPushedAggregatesandPushedGroupby, to be aligned withPushedLimitandPushedSample, as there's not so many queries can benefit from aggregate push down, so we don't need to print those unnecessary information.Why are the changes needed?
Avoid unnecessary file splits in multiple tasks for Parquet reader with aggregate push down.
Does this PR introduce any user-facing change?
No.
How was this patch tested?
Existing unit test in
FileSourceAggregatePushDownSuite.scala.