Skip to content

[SPARK-19736][SQL] refreshByPath should clear all cached plans with the specified path #17064

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 2 commits into from
Closed

[SPARK-19736][SQL] refreshByPath should clear all cached plans with the specified path #17064

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
dd6d8ca
refreshByPath should clear all cached plans with the specified path.
viirya Feb 25, 2017
a9fe0be
Address comments.
viirya Mar 1, 2017
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
19 changes: 10 additions & 9 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,15 +168,16 @@ class CacheManager extends Logging {
(fs, path.makeQualified(fs.getUri, fs.getWorkingDirectory))
}

cachedData.foreach {
case data if data.plan.find(lookupAndRefresh(_, fs, qualifiedPath)).isDefined =>
val dataIndex = cachedData.indexWhere(cd => data.plan.sameResult(cd.plan))
if (dataIndex >= 0) {
data.cachedRepresentation.cachedColumnBuffers.unpersist(blocking = true)
cachedData.remove(dataIndex)
}
sparkSession.sharedState.cacheManager.cacheQuery(Dataset.ofRows(sparkSession, data.plan))
case _ => // Do Nothing
cachedData.filter {
Copy link
Contributor

@cloud-fan cloud-fan Feb 27, 2017

Choose a reason for hiding this comment

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

why the previous one doesn't work?

Copy link
Member Author

@viirya viirya Feb 28, 2017

Choose a reason for hiding this comment

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

This kind of collection can't be modified during iterating. Some elements are not iterated over if we delete/add elements.

Copy link
Contributor

@cloud-fan cloud-fan Feb 28, 2017

Choose a reason for hiding this comment

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

but we are still modifying it during iteration, after the filter. can you be more specific about what the problem is?

Copy link
Contributor

@cloud-fan cloud-fan Feb 28, 2017

Choose a reason for hiding this comment

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

can we use a java collection so that we can remove elements while iterating?

Copy link
Member Author

@viirya viirya Feb 28, 2017

Choose a reason for hiding this comment

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

After filter, we iterate on a different collection than cachedData, so it is no problem to add/delete elements to cachedData.

Copy link
Member Author

@viirya viirya Feb 28, 2017 •
edited
Loading

Choose a reason for hiding this comment

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

The problem can be shown clearly with an example code snippet:

val t = new scala.collection.mutable.ArrayBuffer[Int]

t += 1
t += 2

t.foreach { 
  case i if i > 0 =>
    println(s"i = $i")
    val index = t.indexWhere(_ == i)
    if (index >= 0) {
      t.remove(index)
    }
    println(s"t: $t")
    t += (i + 2)
    println(s"t: $t")
}

Output:

i = 1    // The first iteration, we get the first element "1"
t: ArrayBuffer(2)   // "1" has been removed from the array
t: ArrayBuffer(2, 3) // New element "3" has been inserted
i = 3   // In next iteration, element "2" is wrongly skipped
t: ArrayBuffer(2)     // "3" has been removed from the array
t: ArrayBuffer(2, 5)

The element "2" is never iterated over.

case data if data.plan.find(lookupAndRefresh(_, fs, qualifiedPath)).isDefined => true
case _ => false
}.foreach { data =>
val dataIndex = cachedData.indexWhere(cd => data.plan.sameResult(cd.plan))
if (dataIndex >= 0) {
data.cachedRepresentation.cachedColumnBuffers.unpersist(blocking = true)
cachedData.remove(dataIndex)
}
sparkSession.sharedState.cacheManager.cacheQuery(Dataset.ofRows(sparkSession, data.plan))
}
}

Expand Down
24 changes: 24 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -634,4 +634,28 @@ class CachedTableSuite extends QueryTest with SQLTestUtils with SharedSQLContext
assert(getNumInMemoryRelations(cachedPlan2) == 4)
}
}

test("refreshByPath should refresh all cached plans with the specified path") {
def f(path: String, spark: SparkSession, dataCount: Int): DataFrame = {
spark.catalog.refreshByPath(path)
Copy link
Contributor

@cloud-fan cloud-fan Feb 28, 2017

Choose a reason for hiding this comment

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

we can put spark.range(dataCount).write.mode("overwrite").parquet(path) at the beginning of this method and name it testRefreshByPath instead of f

val data = spark.read.parquet(path)
assert(data.count == dataCount)
val df = data.filter("id > 10")
df.cache
assert(df.count == dataCount - 11)
val df1 = df.filter("id > 11")
df1.cache
assert(df1.count == dataCount - 12)
df1
Copy link
Contributor

@cloud-fan cloud-fan Feb 28, 2017

Choose a reason for hiding this comment

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

I don't get it, so we call refreshByPath before caching the query? Shouldn't we test the opposite order?

Copy link
Member Author

@viirya viirya Mar 1, 2017

Choose a reason for hiding this comment

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

The function is called twice. So actually it is meant to refresh the cache in first call. Since I will change the test to what you suggested #17064 (comment), we can get rid of this confusing.

}

withTempDir { dir =>
val path = dir.getPath()
Copy link
Contributor

@cloud-fan cloud-fan Feb 28, 2017

Choose a reason for hiding this comment

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

we usually call dir.getCanonicalPath

spark.range(100).write.mode("overwrite").parquet(path)
assert(f(path, spark, 100).count == 88)

spark.range(1000).write.mode("overwrite").parquet(path)
assert(f(path, spark, 1000).count == 988)
Copy link
Contributor

@cloud-fan cloud-fan Feb 28, 2017

Choose a reason for hiding this comment

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

we can make this test more explicit

      spark.range(10).write.mode("overwrite").parquet(path)
      spark.read.parquet(path).cache()
      spark.read.parquet(path).filter($"id" > 4).cache()
      assert(spark.read.parquet(path).filter($"id" > 4).count() == 5)

      spark.range(20).write.mode("overwrite").parquet(path)
      spark.catalog.refreshByPath(path)
      assert(spark.read.parquet(path).filter($"id" > 4).count() == 15)

Copy link
Member Author

@viirya viirya Mar 1, 2017

Choose a reason for hiding this comment

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

Ok. Looks simpler and more explicit.

}
}
}