[SPARK-30494][SQL] Fix cached data leakage during replacing an existing view #27185

LantaoJin · 2020-01-13T06:55:07Z

What changes were proposed in this pull request?

The cached RDD for plan "select 1" stays in memory forever until the session close. This cached data cannot be used since the view temp1 has been replaced by another plan. It's a memory leak.

We can reproduce by below commands:

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.0.0-SNAPSHOT
      /_/

Using Scala version 2.12.10 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_201)
Type in expressions to have them evaluated.
Type :help for more information.

scala> spark.sql("create or replace temporary view temp1 as select 1")
scala> spark.sql("cache table temp1")
scala> spark.sql("create or replace temporary view temp1 as select 1, 2")
scala> spark.sql("cache table temp1")
scala> assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1, 2")).isDefined)
scala> assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1")).isDefined)

Why are the changes needed?

Fix the memory leak, specially for long running mode.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Add an unit test.

… an existing view

LantaoJin · 2020-01-13T06:59:11Z

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

@@ -125,6 +132,9 @@ case class CreateViewCommand(
        val viewIdent = tableMetadata.identifier
        checkCyclicViewReference(analyzedPlan, Seq(viewIdent), viewIdent)

+        // uncache the cached data before replacing an exists view
+        sparkSession.catalog.uncacheTable(viewIdent.quotedString)


FYI. There is a condition } else if (replace) { in line 130.

SparkQA · 2020-01-13T08:05:01Z

Test build #116607 has finished for PR 27185 at commit 48c8b37.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

LantaoJin · 2020-01-13T09:55:48Z

retest it please

LantaoJin · 2020-01-13T09:58:55Z

Retest this please

SparkQA · 2020-01-13T14:20:25Z

Test build #116629 has finished for PR 27185 at commit 48c8b37.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

LantaoJin · 2020-01-14T00:15:47Z

Gentle ping @cloud-fan @gengliangwang @maropu @dongjoon-hyun @kiszk @wangyum

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala

maropu · 2020-01-14T00:23:23Z

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

@@ -108,9 +108,16 @@ case class CreateViewCommand(
    verifyTemporaryObjectsNotExists(catalog)

    if (viewType == LocalTempView) {
+      if (replace && catalog.getTempView(name.table).isDefined) {
+        sparkSession.catalog.uncacheTable(name.quotedString)


How about showing an INFO log for this case instead of implicitly uncaching?

I prefer to adding INFO log, but not removing the uncacheTable here, because after replacing, there is no way to drop the old cached data.

I didn't mean so and its ok to just add logging.

Yes. Added.

SparkQA · 2020-01-14T05:13:14Z

Test build #116668 has finished for PR 27185 at commit 69ab94f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

LantaoJin · 2020-01-31T04:43:25Z

Gentle ping @cloud-fan @HyukjinKwon @maropu @dongjoon-hyun @kiszk @srowen

srowen · 2020-01-31T05:11:14Z

Seems reasonable to me?

cloud-fan · 2020-01-31T09:01:45Z

I think it makes sense, but we should follow how similar things are done in DROP TABLE.

LantaoJin · 2020-02-01T13:28:48Z

I think it makes sense, but we should follow how similar things are done in DROP TABLE.

I think they are similar with DROP TABLE
In DropTableCommand:

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

Line 239 in da32d1e

sparkSession.sharedState.cacheManager.uncacheQuery(

      try {
        sparkSession.sharedState.cacheManager.uncacheQuery(
          sparkSession.table(tableName), cascade = !isTempView)
      } catch {
        case NonFatal(e) => log.warn(e.toString, e)
      }

I added the sparkSession.catalog.uncacheTable() in views.scala, uncacheTable() has the similar logic:

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

Line 114 in 69ab94f

sparkSession.catalog.uncacheTable(name.quotedString)

  override def uncacheTable(tableName: String): Unit = {
    val tableIdent = sparkSession.sessionState.sqlParser.parseTableIdentifier(tableName)
    val cascade = !sessionCatalog.isTemporaryTable(tableIdent)
    sparkSession.sharedState.cacheManager.uncacheQuery(sparkSession.table(tableName), cascade)
  }

Can I just use try..cache to wrap the uncacheTable()?

SparkQA · 2020-02-01T17:40:48Z

Test build #117717 has finished for PR 27185 at commit 8a2a9f4.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2020-02-03T07:25:58Z

shall we further generalize it? Currently we un-cache tables in several commands like DROP TABLE, TRUNCATE TABLE, etc. and now we find more missing places like CREATE VIEW.

Instead of un-caching tables in the commands, I feel it's better to do it in low-level basic operations like SessionCatalog.dropTable, createTempView, etc.

LantaoJin · 2020-02-03T08:19:24Z

To do that, SessionCatalog needs a reference of SparkSession. We need pass SparkSession to the construct arguments of SessionCatalog. Is that okey?

LantaoJin · 2020-02-03T08:30:19Z

To do that, SessionCatalog needs a reference of SparkSession. We need pass SparkSession to the construct arguments of SessionCatalog. Is that okey?

Oh, checked code. Catalog and CacheManager are all in sql-core package, but SessionCatalog is in sql-catalyst package. So we cannot reference them here.

cloud-fan · 2020-02-03T11:47:19Z

ah I see. CacheManager is in sql/core not catalyst. Then can we create a util function to make sure we uncache table in a consistent way?

LantaoJin · 2020-02-03T12:58:26Z

I think def uncacheTable(tableName: String): Unit in Catalog.scala is already the general function to uncache table. But it still can not prevent to use other ways in invoke side.

spark/sql/core/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala

Line 380 in 69ab94f

def uncacheTable(tableName: String): Unit

Can we move sql/core/src/main/scala/org/apache/spark/sql/catalog/Catalog.scala to catalyst package?

cloud-fan · 2020-02-03T13:10:04Z

The entry point of Catalog is SparkSession#catalog. It doesn't help if we only move Catalog to catalyst.

LantaoJin · 2020-02-03T13:52:17Z

The entry point of Catalog is SparkSession#catalog. It doesn't help if we only move Catalog to catalyst.

Yes. Another way is adding an uncacheTable method in interface ExternalCatalog. And pass cacheManager to HiveExternalCatalog construct arguments. But InMemoryCatalog in catalyst package can not use cacheManager. So the unit test only works when use HiveExternalCatalog.

But I am not sure whether or not worth to touch the interface ExternalCatalog

LantaoJin · 2020-03-13T11:01:26Z

retest please

maropu · 2020-03-14T07:08:03Z

retest this please

SparkQA · 2020-03-14T11:17:17Z

Test build #119788 has finished for PR 27185 at commit 8a2a9f4.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

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

SparkQA · 2020-03-21T13:38:56Z

Test build #120127 has finished for PR 27185 at commit 57c3c33.

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

SparkQA · 2020-03-21T16:36:25Z

Test build #120128 has finished for PR 27185 at commit 8839bdb.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun · 2020-03-21T22:01:57Z

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

@@ -21,7 +21,6 @@ import java.util.Locale

 import org.apache.spark.sql.{Dataset, Row, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.analysis.NoSuchTableException


Shall we revert this change in this file from this PR?
This kind of irrelevant change makes the backport difficult.

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

dongjoon-hyun · 2020-03-22T00:06:15Z

Also, @LantaoJin . Please fix the PR description. For example, the following doesn't work.

beeline> create or replace temporary view temp1 as select 1
beeline> cache table tempView
beeline> create or replace temporary view temp1 as select 1, 2
beeline> cache table tempView

dongjoon-hyun · 2020-03-22T00:12:14Z

sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala

+      assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1")).isEmpty)
+      sql("cache table tempView")
+      assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1, 2")).isDefined)
+      assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1")).isEmpty)


This seems to be redundant due to line 1133.

dongjoon-hyun · 2020-03-22T00:12:35Z

sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala

+      assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1")).isEmpty)
+      sql("cache table global_temp.tempGlobalTempView")
+      assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1, 2")).isDefined)
+      assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1")).isEmpty)


This seems to be redundant due to 1145.

dongjoon-hyun · 2020-03-22T00:14:13Z

sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala

+      //        |
+      //        + Project[1 AS 1]
+      spark.sharedState.cacheManager.uncacheQuery(spark.table("view1"), cascade = false)
+      assert(spark.sharedState.cacheManager.isEmpty)


This looks like an invalid test scope because this is testing uncacheQuery at line 1164 instead of create or replace view. Shall we revise this whole test case?

1164 and 1165 is to make sure our patch is working, no cached data leak for persisted view. Without this patch, the assert 1165 will fails. So I don't think it's an invalid test.

dongjoon-hyun · 2020-03-22T00:15:16Z

sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala

@@ -1122,4 +1122,47 @@ class CachedTableSuite extends QueryTest with SQLTestUtils
      assert(!spark.catalog.isCached("t1"))
    }
  }
+
+  test("SPARK-30494 avoid duplicated cached RDD when replace an existing view") {


This can be misleading because the cached RDD is not duplicated. Two cached RDD are independent.

Ok. how about "Fix the leak of cached data when replace an existing view"

dongjoon-hyun

Hi, @LantaoJin . Could you revise the PR once more? Thanks!

LantaoJin · 2020-03-22T03:44:07Z

Ah, you meant the beeline>. Sorry, I tested via thriftserver, I will change to spark

Also, @LantaoJin . Please fix the PR description. For example, the following doesn't work.
beeline> create or replace temporary view temp1 as select 1
beeline> cache table tempView
beeline> create or replace temporary view temp1 as select 1, 2
beeline> cache table tempView

Ah I get it.

SparkQA · 2020-03-22T11:42:48Z

Test build #120160 has finished for PR 27185 at commit e59fc98.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dongjoon-hyun

+1, LGTM. Thank you, @LantaoJin and all!
Merged to master/3.0.

…ng view ### What changes were proposed in this pull request? The cached RDD for plan "select 1" stays in memory forever until the session close. This cached data cannot be used since the view temp1 has been replaced by another plan. It's a memory leak. We can reproduce by below commands: ``` Welcome to ____ __ / __/__ ___ _____/ /__ _\ \/ _ \/ _ `/ __/ '_/ /___/ .__/\_,_/_/ /_/\_\ version 3.0.0-SNAPSHOT /_/ Using Scala version 2.12.10 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_201) Type in expressions to have them evaluated. Type :help for more information. scala> spark.sql("create or replace temporary view temp1 as select 1") scala> spark.sql("cache table temp1") scala> spark.sql("create or replace temporary view temp1 as select 1, 2") scala> spark.sql("cache table temp1") scala> assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1, 2")).isDefined) scala> assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1")).isDefined) ``` ### Why are the changes needed? Fix the memory leak, specially for long running mode. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? Add an unit test. Closes #27185 from LantaoJin/SPARK-30494. Authored-by: LantaoJin <jinlantao@gmail.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org> (cherry picked from commit 929b794) Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

dongjoon-hyun · 2020-03-23T05:23:19Z

Hi, @LantaoJin . Could you make a backport PR against branch-2.4?

…ng view ### What changes were proposed in this pull request? The cached RDD for plan "select 1" stays in memory forever until the session close. This cached data cannot be used since the view temp1 has been replaced by another plan. It's a memory leak. We can reproduce by below commands: ``` Welcome to ____ __ / __/__ ___ _____/ /__ _\ \/ _ \/ _ `/ __/ '_/ /___/ .__/\_,_/_/ /_/\_\ version 3.0.0-SNAPSHOT /_/ Using Scala version 2.12.10 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_201) Type in expressions to have them evaluated. Type :help for more information. scala> spark.sql("create or replace temporary view temp1 as select 1") scala> spark.sql("cache table temp1") scala> spark.sql("create or replace temporary view temp1 as select 1, 2") scala> spark.sql("cache table temp1") scala> assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1, 2")).isDefined) scala> assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1")).isDefined) ``` ### Why are the changes needed? Fix the memory leak, specially for long running mode. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? Add an unit test. Closes apache#27185 from LantaoJin/SPARK-30494. Authored-by: LantaoJin <jinlantao@gmail.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org> (cherry picked from commit 929b794)

…xisting view ### What changes were proposed in this pull request? This is backport of #27185 to branch-2.4. The cached RDD for plan "select 1" stays in memory forever until the session close. This cached data cannot be used since the view temp1 has been replaced by another plan. It's a memory leak. We can reproduce by below commands: ``` Welcome to ____ __ / __/__ ___ _____/ /__ _\ \/ _ \/ _ `/ __/ '_/ /___/ .__/\_,_/_/ /_/\_\ version 3.0.0-SNAPSHOT /_/ Using Scala version 2.12.10 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_201) Type in expressions to have them evaluated. Type :help for more information. scala> spark.sql("create or replace temporary view temp1 as select 1") scala> spark.sql("cache table temp1") scala> spark.sql("create or replace temporary view temp1 as select 1, 2") scala> spark.sql("cache table temp1") scala> assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1, 2")).isDefined) scala> assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1")).isDefined) ``` ### Why are the changes needed? Fix the memory leak, specially for long running mode. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? Add a unit test. Closes #28000 from LantaoJin/SPARK-30494_2.4. Lead-authored-by: lajin <lajin@ebay.com> Co-authored-by: LantaoJin <jinlantao@gmail.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

…ng view ### What changes were proposed in this pull request? The cached RDD for plan "select 1" stays in memory forever until the session close. This cached data cannot be used since the view temp1 has been replaced by another plan. It's a memory leak. We can reproduce by below commands: ``` Welcome to ____ __ / __/__ ___ _____/ /__ _\ \/ _ \/ _ `/ __/ '_/ /___/ .__/\_,_/_/ /_/\_\ version 3.0.0-SNAPSHOT /_/ Using Scala version 2.12.10 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_201) Type in expressions to have them evaluated. Type :help for more information. scala> spark.sql("create or replace temporary view temp1 as select 1") scala> spark.sql("cache table temp1") scala> spark.sql("create or replace temporary view temp1 as select 1, 2") scala> spark.sql("cache table temp1") scala> assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1, 2")).isDefined) scala> assert(spark.sharedState.cacheManager.lookupCachedData(sql("select 1")).isDefined) ``` ### Why are the changes needed? Fix the memory leak, specially for long running mode. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? Add an unit test. Closes apache#27185 from LantaoJin/SPARK-30494. Authored-by: LantaoJin <jinlantao@gmail.com> Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

[SPARK-30494][SQL] Avoid duplicated cached RDD when create or replace…

48c8b37

… an existing view

LantaoJin commented Jan 13, 2020

View reviewed changes

maropu reviewed Jan 14, 2020

View reviewed changes

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/catalog/SessionCatalog.scala Outdated Show resolved Hide resolved

maropu reviewed Jan 14, 2020

View reviewed changes

LantaoJin changed the title ~~[SPARK-30494][SQL] Avoid duplicated cached RDD when create or replace an existing view~~ [SPARK-30494][SQL] Avoid duplicated cached RDD when replace an existing view Jan 14, 2020

address comment

69ab94f

add try catch block

8a2a9f4

dongjoon-hyun added the SQL label Feb 5, 2020

maropu reviewed Mar 19, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala Outdated Show resolved Hide resolved

LantaoJin added 2 commits March 21, 2020 20:15

remove useless imports

57c3c33

fix ut

8839bdb

dongjoon-hyun reviewed Mar 21, 2020

View reviewed changes

dongjoon-hyun reviewed Mar 22, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala Outdated Show resolved Hide resolved

dongjoon-hyun reviewed Mar 22, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala Outdated Show resolved Hide resolved

dongjoon-hyun reviewed Mar 22, 2020

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/command/views.scala Outdated Show resolved Hide resolved

dongjoon-hyun reviewed Mar 22, 2020

View reviewed changes

dongjoon-hyun requested changes Mar 22, 2020

View reviewed changes

address comments

e59fc98

LantaoJin changed the title ~~[SPARK-30494][SQL] Avoid duplicated cached RDD when replace an existing view~~ [SPARK-30494][SQL] Fix the leak of cached data when replace an existing view Mar 22, 2020

LantaoJin requested a review from dongjoon-hyun March 22, 2020 12:23

dongjoon-hyun changed the title ~~[SPARK-30494][SQL] Fix the leak of cached data when replace an existing view~~ [SPARK-30494][SQL] Fix the leak of cached data when replace an existing temp view Mar 23, 2020

dongjoon-hyun changed the title ~~[SPARK-30494][SQL] Fix the leak of cached data when replace an existing temp view~~ [SPARK-30494][SQL] Fix the leak of cached data when replace an existing view Mar 23, 2020

dongjoon-hyun changed the title ~~[SPARK-30494][SQL] Fix the leak of cached data when replace an existing view~~ [SPARK-30494][SQL] Fix cached data leakage during replacing an existing view Mar 23, 2020

dongjoon-hyun approved these changes Mar 23, 2020

View reviewed changes

dongjoon-hyun closed this in 929b794 Mar 23, 2020

LantaoJin mentioned this pull request Mar 24, 2020

[SPARK-30494][SQL][2.4] Fix cached data leakage during replacing an existing view #28000

Closed

[SPARK-30494][SQL] Fix cached data leakage during replacing an existing view #27185

[SPARK-30494][SQL] Fix cached data leakage during replacing an existing view #27185

Conversation

LantaoJin commented Jan 13, 2020 • edited

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?

LantaoJin Jan 13, 2020 • edited

Choose a reason for hiding this comment

SparkQA commented Jan 13, 2020

LantaoJin commented Jan 13, 2020

LantaoJin commented Jan 13, 2020

SparkQA commented Jan 13, 2020

LantaoJin commented Jan 14, 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

SparkQA commented Jan 14, 2020

LantaoJin commented Jan 31, 2020

srowen commented Jan 31, 2020

cloud-fan commented Jan 31, 2020

LantaoJin commented Feb 1, 2020 • edited

SparkQA commented Feb 1, 2020

cloud-fan commented Feb 3, 2020 • edited

LantaoJin commented Feb 3, 2020 • edited

LantaoJin commented Feb 3, 2020 • edited

cloud-fan commented Feb 3, 2020

LantaoJin commented Feb 3, 2020

cloud-fan commented Feb 3, 2020

LantaoJin commented Feb 3, 2020 • edited

LantaoJin commented Mar 13, 2020

maropu commented Mar 14, 2020

SparkQA commented Mar 14, 2020

SparkQA commented Mar 21, 2020

SparkQA commented Mar 21, 2020

dongjoon-hyun Mar 21, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dongjoon-hyun commented Mar 22, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dongjoon-hyun Mar 22, 2020 • edited

Choose a reason for hiding this comment

LantaoJin Mar 22, 2020 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dongjoon-hyun left a comment

Choose a reason for hiding this comment

LantaoJin commented Mar 22, 2020

SparkQA commented Mar 22, 2020

dongjoon-hyun left a comment • edited

Choose a reason for hiding this comment

dongjoon-hyun commented Mar 23, 2020

LantaoJin commented Jan 13, 2020 •

edited

LantaoJin Jan 13, 2020 •

edited

LantaoJin commented Feb 1, 2020 •

edited

cloud-fan commented Feb 3, 2020 •

edited

LantaoJin commented Feb 3, 2020 •

edited

LantaoJin commented Feb 3, 2020 •

edited

LantaoJin commented Feb 3, 2020 •

edited

dongjoon-hyun Mar 21, 2020 •

edited

dongjoon-hyun Mar 22, 2020 •

edited

LantaoJin Mar 22, 2020 •

edited

dongjoon-hyun left a comment •

edited