Skip to content

[SPARK-37220][SQL] Do not split input file for Parquet reader with aggregate push down #34498

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
c21 wants to merge 4 commits into from
Closed

[SPARK-37220][SQL] Do not split input file for Parquet reader with aggregate push down #34498

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
4eb0e8d
Do not split input file for Parquet with aggregate push down
c21 Nov 6, 2021
94a1bd4
Fix RowDataSourceScanExec
c21 Nov 6, 2021
3c54820
Removed unused aggString & groupByString in RowDataSourceScanExec
c21 Nov 6, 2021
04c3f20
Backout the change in RowDataSourceScanExec
c21 Nov 6, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 4 additions & 10 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,13 +134,6 @@ case class RowDataSourceScanExec(

def seqToString(seq: Seq[Any]): String = seq.mkString("[", ", ", "]")

val (aggString, groupByString) = if (pushedDownOperators.aggregation.nonEmpty) {
(seqToString(pushedDownOperators.aggregation.get.aggregateExpressions),
seqToString(pushedDownOperators.aggregation.get.groupByColumns))
} else {
("[]", "[]")
}

val markedFilters = if (filters.nonEmpty) {
for (filter <- filters) yield {
if (handledFilters.contains(filter)) s"*$filter" else s"$filter"
Expand All @@ -151,9 +144,10 @@ case class RowDataSourceScanExec(

Map(
"ReadSchema" -> requiredSchema.catalogString,
"PushedFilters" -> seqToString(markedFilters.toSeq),
"PushedAggregates" -> aggString,
"PushedGroupby" -> groupByString) ++
"PushedFilters" -> seqToString(markedFilters.toSeq)) ++
pushedDownOperators.aggregation.fold(Map[String, String]()) { v =>
Comment thread
sunchao marked this conversation as resolved.
Outdated
Map("PushedAggregates" -> seqToString(v.aggregateExpressions),
"PushedGroupby" -> seqToString(v.groupByColumns))} ++
Comment thread
sunchao marked this conversation as resolved.
Outdated
pushedDownOperators.limit.map(value => "PushedLimit" -> s"LIMIT $value") ++
pushedDownOperators.sample.map(v => "PushedSample" ->
s"SAMPLE (${(v.upperBound - v.lowerBound) * 100}) ${v.withReplacement} SEED(${v.seed})"
Expand Down
2 changes: 1 addition & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/orc/OrcScan.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ case class OrcScan(
dataFilters: Seq[Expression] = Seq.empty) extends FileScan {
override def isSplitable(path: Path): Boolean = {
// If aggregate is pushed down, only the file footer will be read once,
// so file should be not split across multiple tasks.
// so file should not be split across multiple tasks.
pushedAggregate.isEmpty
}

Expand Down
5 changes: 0 additions & 5 deletions ...org/apache/spark/sql/execution/datasources/v2/parquet/ParquetPartitionReaderFactory.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,11 +92,6 @@ case class ParquetPartitionReaderFactory(
ParquetFooterReader.readFooter(conf, filePath, SKIP_ROW_GROUPS)
} else {
// For aggregate push down, we will get max/min/count from footer statistics.
// We want to read the footer for the whole file instead of reading multiple
// 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

@c21 c21 Nov 7, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

ParquetFooterReader.readFooter(conf, filePath, NO_FILTER)
}
}
Expand Down
6 changes: 5 additions & 1 deletion ...re/src/main/scala/org/apache/spark/sql/execution/datasources/v2/parquet/ParquetScan.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,11 @@ case class ParquetScan(
pushedAggregate: Option[Aggregation] = None,
partitionFilters: Seq[Expression] = Seq.empty,
dataFilters: Seq[Expression] = Seq.empty) extends FileScan {
override def isSplitable(path: Path): Boolean = true
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

@sunchao sunchao Nov 6, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍

@HyukjinKwon HyukjinKwon Nov 7, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, okay. Got it now.

}

override def readSchema(): StructType = {
// If aggregate is pushed down, schema has already been pruned in `ParquetScanBuilder`
Expand Down