[SPARK-15678][SQL] Not use cache on appends and overwrites

[sameeragarwal]

What changes were proposed in this pull request?

Spark currently incorrectly continues to use cached data even if the underlying data is overwritten.

Current behavior:

val dir = "/tmp/test"
sqlContext.range(1000).write.mode("overwrite").parquet(dir)
val df = sqlContext.read.parquet(dir).cache()
df.count() // outputs 1000
sqlContext.range(10).write.mode("overwrite").parquet(dir)
sqlContext.read.parquet(dir).count() // outputs 1000 <---- We are still using the cached dataset

Expected behavior:

val dir = "/tmp/test"
sqlContext.range(1000).write.mode("overwrite").parquet(dir)
val df = sqlContext.read.parquet(dir).cache()
df.count() // outputs 1000
sqlContext.range(10).write.mode("overwrite").parquet(dir)
df.count() // outputs 1000
sqlContext.read.parquet(dir).count() // outputs 10 <---- We are not using the cached dataset

This patch fixes this bug by modifying the ListingFileCatalog logic that used to only compare the directory name (as opposed to individual files) while comparing 2 plans. Note that in theory, this could lead to a slight regression (for large number of files) but I didn't notice any regression for micro-benchmarks with 1000s of files.

How was this patch tested?

Unit tests for overwrites and appends in ParquetQuerySuite.

[sameeragarwal]

@yhuai @mengxr what are your thoughts on this approach?

[dongjoon-hyun]

sqlContext -> spark

[SparkQA]

Test build #59668 has finished for PR 13419 at commit ee631d2.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[dongjoon-hyun]

Hi, @sameeragarwal .
Is there any reason to use SQLContext instead of SparkSession in this PR?

[sameeragarwal]

@dongjoon-hyun no reason; old habits. I'll fix this. Thanks! :)

[mengxr]

I will prefer refreshing the dataset every time a dataset is reloaded but keeping existing ones unchanged.

val df1 = sqlContext.read.parquet(dir).cache()
df1.count() // outputs 1000
sqlContext.range(10).write.mode("overwrite").parquet(dir)
val df2 = sqlContext.read.parquet(dir).count() // outputs 10
df2.count() // outputs 10
df1.count() // still outputs 1000 because it was cached

Neither approach is perfectly safe. So I don't have strong preference on either.

[sameeragarwal]

@mengxr it seems like overwriting generates new files so we can achieve the same semantics without introducing an additional timestamp. The current solution should respect the contract for old dataframes while making sure that the new ones don't use the cached value. Let me know what you think.

[sameeragarwal]

Also cc'ing @davies

[SparkQA]

Test build #59706 has finished for PR 13419 at commit a21013a.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[tejasapatil]

I guess that the caching is done over multiple nodes. If the data for a dataset is updated physically and some of the nodes where the data was cached go down, would the existing cached dataset be invalidated and refreshed ? If not, then old dataframes can give inconsistent or incomplete data.

[sameeragarwal]

@tejasapatil if the nodes where the data was cached go down, the CacheManager should still consider that data as cached. In that case, the next time the data is accessed, the underlying RDD will be recomputed and cached again.

[sameeragarwal]

I ended up creating a small design doc describing the problem and presenting 2 possible solutions at https://docs.google.com/document/d/1h5SzfC5UsvIrRpeLNDKSMKrKJvohkkccFlXo-GBAwQQ/edit?ts=574f717f#. Based on this, we decided in favor of option 2 (#13566) as it is a less intrusive change to the default behavior. I'm going to close this PR for now, but we may revisit this approach (i.e., option 1) for 2.1.