Skip to content

[SPARK-8756][SQL] Keep cached information and avoid re-calculating footers in ParquetRelation2 #7154

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
wants to merge 11 commits into from
Closed

[SPARK-8756][SQL] Keep cached information and avoid re-calculating footers in ParquetRelation2 #7154

Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
0ef8caf
Keep cached information and avoid re-calculating footers.
viirya Jul 1, 2015
186429d
Merge remote-tracking branch 'upstream/master' into cached_footer_par…
viirya Jul 20, 2015
12a0ed9
Add check of FileStatus's modification time.
viirya Jul 20, 2015
21bbdec
Merge remote-tracking branch 'upstream/master' into cached_footer_par…
viirya Jul 21, 2015
6ae0911
Merge remote-tracking branch 'upstream/master' into cached_footer_par…
viirya Jul 22, 2015
fa5458f
Use Map to cache FileStatus and do merging previously loaded schema a…
viirya Jul 22, 2015
c2a2420
Merge remote-tracking branch 'upstream/master' into cached_footer_par…
viirya Jul 23, 2015
a52b6d1
For comments.
viirya Jul 23, 2015
c8fdfb7
Fix it.
viirya Jul 23, 2015
ae0ec64
Fix wrong assignment.
viirya Jul 23, 2015
92e9347
Fix indentation.
viirya Jul 24, 2015
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
38 changes: 24 additions & 14 deletions sql/core/src/main/scala/org/apache/spark/sql/parquet/newParquet.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -345,24 +345,34 @@ private[sql] class ParquetRelation2(
// Schema of the whole table, including partition columns.
var schema: StructType = _

// Cached leaves
var cachedLeaves: Set[FileStatus] = null

/**
* Refreshes `FileStatus`es, footers, partition spec, and table schema.
*/
def refresh(): Unit = {
// Lists `FileStatus`es of all leaf nodes (files) under all base directories.
val leaves = cachedLeafStatuses().filter { f =>
isSummaryFile(f.getPath) ||
!(f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith("."))
}.toArray

dataStatuses = leaves.filterNot(f => isSummaryFile(f.getPath))
metadataStatuses = leaves.filter(_.getPath.getName == ParquetFileWriter.PARQUET_METADATA_FILE)
commonMetadataStatuses =
leaves.filter(_.getPath.getName == ParquetFileWriter.PARQUET_COMMON_METADATA_FILE)

// If we already get the schema, don't need to re-compute it since the schema merging is
// time-consuming.
if (dataSchema == null) {
val currentLeafStatuses = cachedLeafStatuses()

// Check if cachedLeafStatuses is changed or not
val leafStatusesChanged = (cachedLeaves == null) ||
!cachedLeaves.equals(currentLeafStatuses)

if (leafStatusesChanged) {
cachedLeaves = currentLeafStatuses.toIterator.toSet

// Lists `FileStatus`es of all leaf nodes (files) under all base directories.
val leaves = currentLeafStatuses.filter { f =>
Copy link
Contributor

@liancheng liancheng Jul 22, 2015

Choose a reason for hiding this comment

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

This is irrelevant to this PR, but I'd like to point out that this check is unnecessary now. IIRC, at the time PR #6012 was merged, ParquetRelation2 still needed to be serialized to executor side, thus avoiding schema merging on executor side makes sense. However, after migrating to HadoopFsRelation, ParquetRelation2 won't be serialized anymore. You may see that ParquetRelation2 is no longer a case class and doesn't even extend Serializable now. I made this change intentionally to make sure ParquetRelation2 is never serialized.

isSummaryFile(f.getPath) ||
!(f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith("."))
}.toArray

dataStatuses = leaves.filterNot(f => isSummaryFile(f.getPath))
metadataStatuses =
leaves.filter(_.getPath.getName == ParquetFileWriter.PARQUET_METADATA_FILE)
commonMetadataStatuses =
leaves.filter(_.getPath.getName == ParquetFileWriter.PARQUET_COMMON_METADATA_FILE)

dataSchema = {
val dataSchema0 = maybeDataSchema
.orElse(readSchema())
Expand Down