[SPARK-33786][SQL] The storage level for a cache should be respected when a table name is altered. #30774

imback82 · 2020-12-15T03:52:55Z

What changes were proposed in this pull request?

This PR proposes to retain the cache's storage level when a table name is altered by ALTER TABLE ... RENAME TO ....

Why are the changes needed?

Currently, when a table name is altered, the table's cache is refreshed (if exists), but the storage level is not retained. For example:

        def getStorageLevel(tableName: String): StorageLevel = {
          val table = spark.table(tableName)
          val cachedData = spark.sharedState.cacheManager.lookupCachedData(table).get
          cachedData.cachedRepresentation.cacheBuilder.storageLevel
        }

        Seq(1 -> "a").toDF("i", "j").write.parquet(path.getCanonicalPath)
        sql(s"CREATE TABLE old USING parquet LOCATION '${path.toURI}'")
        sql("CACHE TABLE old OPTIONS('storageLevel' 'MEMORY_ONLY')")
        val oldStorageLevel = getStorageLevel("old")

        sql("ALTER TABLE old RENAME TO new")
        val newStorageLevel = getStorageLevel("new")

oldStorageLevel will be StorageLevel(memory, deserialized, 1 replicas) whereas newStorageLevel will be StorageLevel(disk, memory, deserialized, 1 replicas), which is the default storage level.

Does this PR introduce any user-facing change?

Yes, now the storage level for the cache will be retained.

How was this patch tested?

Added a unit test.

imback82 · 2020-12-15T03:56:09Z

sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala

@@ -195,16 +195,21 @@ case class AlterTableRenameCommand(
      DDLUtils.verifyAlterTableType(catalog, table, isView)
      // If an exception is thrown here we can just assume the table is uncached;
      // this can happen with Hive tables when the underlying catalog is in-memory.
-      val wasCached = Try(sparkSession.catalog.isCached(oldName.unquotedString)).getOrElse(false)


The existing implementation uses Catalog APIs (isCached), whereas this PR uses CacheManager directly. If this approach is not desired, we can update Catalog API to expose StorageLevel.

imback82 · 2020-12-15T03:57:11Z

I will also update v2 implementation (RenameTableExec) once this PR is done.

SparkQA · 2020-12-15T05:01:43Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37400/

SparkQA · 2020-12-15T05:34:07Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37400/

imback82 · 2020-12-15T05:40:55Z

cc @cloud-fan @viirya

cloud-fan · 2020-12-15T06:01:20Z

sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala

+          sparkSession.table(oldName.unquotedString))
+        optCachedData.map(_.cachedRepresentation.cacheBuilder.storageLevel)
+      }.getOrElse(None)
+      optStorageLevel.foreach { _ =>


nit: if (optStorageLevel.isDefine). It's clearer as a check to see if the table is cached or not.

SparkQA · 2020-12-15T06:38:09Z

Test build #132798 has finished for PR 30774 at commit eb60d1e.

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

viirya · 2020-12-15T06:53:18Z

sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala

-      val wasCached = Try(sparkSession.catalog.isCached(oldName.unquotedString)).getOrElse(false)
-      if (wasCached) {
+      // If `optStorageLevel` is defined, the old table was cached.
+      val optStorageLevel = Try {


Will lookupCachedData throw exception actually? If it's not cached, None should be returned. Isn't?

lookupCachedData doesn't throw an exception. I believe the existing code is wrapped in Try because isCached is calling spark.table (could be related to "this can happen with Hive tables when the underlying catalog is in-memory"). So, I was keeping the same behavior.

Do you think Try can be removed?

Yea I think we can remove it

+1 to remove it. We use lookupCachedData in many places and I think we don't need to add Try here.

viirya · 2020-12-15T06:57:13Z

sql/core/src/main/scala/org/apache/spark/sql/execution/command/tables.scala

-      if (wasCached) {
-        sparkSession.catalog.cacheTable(newName.unquotedString)
+      optStorageLevel.foreach { storageLevel =>
+        sparkSession.catalog.cacheTable(newName.unquotedString, storageLevel)


Does this miss the tableName if there is in the original cache?

Sorry, I didn't get this question. This is creating a new cache with a new table name.

Hm, you can check the change like #30769. Especially how it recaches the table. There is a cacheName parameter. If the table was cached with a cache name, when recaching it, I think we should keep it.

val cache = session.sharedState.cacheManager.lookupCachedData(v2Relation) session.sharedState.cacheManager.uncacheQuery(session, v2Relation, cascade = true) session.sharedState.cacheManager.uncacheQuery(session, v2Relation, cascade = true) if (recacheTable && cache.isDefined) { // save the cache name and cache level for recreation val cacheName = cache.get.cachedRepresentation.cacheBuilder.tableName val cacheLevel = cache.get.cachedRepresentation.cacheBuilder.storageLevel // recache with the same name and cache level. val ds = Dataset.ofRows(session, v2Relation) session.sharedState.cacheManager.cacheQuery(ds, cacheName, cacheLevel) }

The previous code seems also recache with the new name?

No, the refresh table command for v2 doesn't recache the table before #30769.

I mean the previous code in AlterTableRenameCommand. We shouldn't change its behavior regarding cache name in this bug fix PR.

Hmm okay, actually it also sounds like a bug if alter table command changes the cache name. I'm fine to leave it unchanged here.

I believe cache name is used for debug purpose only (for RDD name and InMemoryTableScanExec). So if the cache name - which is tied to the table name - doesn't change when the table is changed, wouldn't it cause a confusion since it will still refer to the old table name? I can do a follow up PR if this seems like a bug.

SparkQA · 2020-12-15T08:42:41Z

Test build #132814 has finished for PR 30774 at commit 11603d5.

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

SparkQA · 2020-12-15T08:49:19Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37415/

SparkQA · 2020-12-15T09:24:31Z

Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/37415/

cloud-fan · 2020-12-16T05:45:43Z

GA passed, merging to master/3.1, thanks!

…when a table name is altered ### What changes were proposed in this pull request? This PR proposes to retain the cache's storage level when a table name is altered by `ALTER TABLE ... RENAME TO ...`. ### Why are the changes needed? Currently, when a table name is altered, the table's cache is refreshed (if exists), but the storage level is not retained. For example: ```scala def getStorageLevel(tableName: String): StorageLevel = { val table = spark.table(tableName) val cachedData = spark.sharedState.cacheManager.lookupCachedData(table).get cachedData.cachedRepresentation.cacheBuilder.storageLevel } Seq(1 -> "a").toDF("i", "j").write.parquet(path.getCanonicalPath) sql(s"CREATE TABLE old USING parquet LOCATION '${path.toURI}'") sql("CACHE TABLE old OPTIONS('storageLevel' 'MEMORY_ONLY')") val oldStorageLevel = getStorageLevel("old") sql("ALTER TABLE old RENAME TO new") val newStorageLevel = getStorageLevel("new") ``` `oldStorageLevel` will be `StorageLevel(memory, deserialized, 1 replicas)` whereas `newStorageLevel` will be `StorageLevel(disk, memory, deserialized, 1 replicas)`, which is the default storage level. ### Does this PR introduce _any_ user-facing change? Yes, now the storage level for the cache will be retained. ### How was this patch tested? Added a unit test. Closes #30774 from imback82/alter_table_rename_cache_fix. Authored-by: Terry Kim <yuminkim@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com> (cherry picked from commit ef7f690) Signed-off-by: Wenchen Fan <wenchen@databricks.com>

cloud-fan · 2020-12-16T05:46:34Z

@imback82 can you open a backport PR for 3.0? thanks!

dongjoon-hyun · 2020-12-16T08:24:20Z

Thank you for the fix, @imback82 and @cloud-fan .

cc @sunchao

…cted when a table name is altered ### What changes were proposed in this pull request? This is a back port of #30774. This PR proposes to retain the cache's storage level when a table name is altered by `ALTER TABLE ... RENAME TO ...`. ### Why are the changes needed? Currently, when a table name is altered, the table's cache is refreshed (if exists), but the storage level is not retained. For example: ```scala def getStorageLevel(tableName: String): StorageLevel = { val table = spark.table(tableName) val cachedData = spark.sharedState.cacheManager.lookupCachedData(table).get cachedData.cachedRepresentation.cacheBuilder.storageLevel } Seq(1 -> "a").toDF("i", "j").write.parquet(path.getCanonicalPath) sql(s"CREATE TABLE old USING parquet LOCATION '${path.toURI}'") sql("CACHE TABLE old OPTIONS('storageLevel' 'MEMORY_ONLY')") val oldStorageLevel = getStorageLevel("old") sql("ALTER TABLE old RENAME TO new") val newStorageLevel = getStorageLevel("new") ``` `oldStorageLevel` will be `StorageLevel(memory, deserialized, 1 replicas)` whereas `newStorageLevel` will be `StorageLevel(disk, memory, deserialized, 1 replicas)`, which is the default storage level. ### Does this PR introduce _any_ user-facing change? Yes, now the storage level for the cache will be retained. ### How was this patch tested? Added a unit test. Closes #30793 from imback82/alter_table_rename_cache_fix_3.0. Authored-by: Terry Kim <yuminkim@gmail.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>

initial commit

eb60d1e

github-actions bot added the SQL label Dec 15, 2020

imback82 commented Dec 15, 2020

View reviewed changes

cloud-fan reviewed Dec 15, 2020

View reviewed changes

viirya reviewed Dec 15, 2020

View reviewed changes

imback82 added 2 commits December 14, 2020 23:33

Address PR comment

5538921

Address comment

11603d5

cloud-fan closed this in ef7f690 Dec 16, 2020

imback82 mentioned this pull request Dec 16, 2020

[SPARK-33786][SQL][3.0] The storage level for a cache should be respected when a table name is altered. #30793

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-33786][SQL] The storage level for a cache should be respected when a table name is altered. #30774

[SPARK-33786][SQL] The storage level for a cache should be respected when a table name is altered. #30774

imback82 commented Dec 15, 2020

imback82 Dec 15, 2020

imback82 commented Dec 15, 2020

SparkQA commented Dec 15, 2020

SparkQA commented Dec 15, 2020

imback82 commented Dec 15, 2020

cloud-fan Dec 15, 2020 •

edited

Loading

viirya Dec 15, 2020

imback82 Dec 15, 2020

SparkQA commented Dec 15, 2020

viirya Dec 15, 2020

imback82 Dec 15, 2020

cloud-fan Dec 15, 2020

viirya Dec 15, 2020

viirya Dec 15, 2020

imback82 Dec 15, 2020

viirya Dec 15, 2020

cloud-fan Dec 15, 2020

viirya Dec 15, 2020

cloud-fan Dec 15, 2020

viirya Dec 15, 2020

imback82 Dec 16, 2020

SparkQA commented Dec 15, 2020

SparkQA commented Dec 15, 2020

SparkQA commented Dec 15, 2020

cloud-fan commented Dec 16, 2020

cloud-fan commented Dec 16, 2020

dongjoon-hyun commented Dec 16, 2020

[SPARK-33786][SQL] The storage level for a cache should be respected when a table name is altered. #30774

[SPARK-33786][SQL] The storage level for a cache should be respected when a table name is altered. #30774

Conversation

imback82 commented Dec 15, 2020

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Choose a reason for hiding this comment

imback82 commented Dec 15, 2020

SparkQA commented Dec 15, 2020

SparkQA commented Dec 15, 2020

imback82 commented Dec 15, 2020

cloud-fan Dec 15, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 15, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Dec 15, 2020

SparkQA commented Dec 15, 2020

SparkQA commented Dec 15, 2020

cloud-fan commented Dec 16, 2020

cloud-fan commented Dec 16, 2020

dongjoon-hyun commented Dec 16, 2020

cloud-fan Dec 15, 2020 •

edited

Loading