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

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

Changes from 4 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
28 changes: 24 additions & 4 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 @@ -342,15 +342,35 @@ private[sql] class ParquetRelation2(
// Schema of the whole table, including partition columns.
var schema: StructType = _

// Cached leaf statuses
var localCachedLeafStatuses: Set[FileStatus] = _

var lastRefreshTime: Long = 0

// Cached leaves
var cachedLeaves: Array[FileStatus] = Array()

/**
* Refreshes `FileStatus`es, footers, partition spec, and table schema.
*/
def refresh(): Unit = {
// Check if cachedLeafStatuses is changed or not
val leafStatusesChanged = localCachedLeafStatuses != cachedLeafStatuses()

// 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
val leaves = if (leafStatusesChanged) {
localCachedLeafStatuses = cachedLeafStatuses()
Copy link
Contributor

@liancheng liancheng Jul 21, 2015

Choose a reason for hiding this comment

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

Should we do a deep copy here? Currently it's OK because cachedLeafStatuses() always returns a new instance of Set[FileStatus], but it's possible that we use a mutable set object in the future. In that case, the != predicate above will always be true.

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.

val updatedLeaves = cachedLeafStatuses().filter { f =>
(isSummaryFile(f.getPath) ||
!(f.getPath.getName.startsWith("_") || f.getPath.getName.startsWith("."))) &&
(f.getModificationTime > lastRefreshTime)
}.toArray
lastRefreshTime = System.currentTimeMillis
Copy link
Contributor

@liancheng liancheng Jul 21, 2015

Choose a reason for hiding this comment

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

Using System.currentTimeMillis is not OK, because clocks on different nodes may differ.

Copy link
Contributor

@liancheng liancheng Jul 21, 2015

Choose a reason for hiding this comment

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

We care about two kinds of files here:

  • New files: files don't exist in the old cache.
  • Files got updated: files exist in the old cache but have different modification time from their old copy.

cachedLeaves = updatedLeaves
Copy link
Contributor

@liancheng liancheng Jul 21, 2015

Choose a reason for hiding this comment

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

We shouldn't drop FileStatuses of non-updated leaves here.

cachedLeaves
} else {
cachedLeaves
}

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