Skip to content

[SPARK-24521][SQL][TEST] Fix ineffective test in CachedTableSuite #21531

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

[SPARK-24521][SQL][TEST] Fix ineffective test in CachedTableSuite #21531

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0d855e5
Fix cache test
icexelloss Jun 11, 2018
377e049
Add clean up code
icexelloss Jun 11, 2018
e01e603
Revert unneeded changes
icexelloss Jun 11, 2018
658539a
Fix unneeded changes
icexelloss Jun 11, 2018
c9db68d
Address comments
icexelloss Jun 13, 2018
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: 0 additions & 19 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 @@ -83,25 +83,6 @@ class CachedTableSuite extends QueryTest with SQLTestUtils with SharedSQLContext
}.sum
}

test("withColumn doesn't invalidate cached dataframe") {
var evalCount = 0
val myUDF = udf((x: String) => { evalCount += 1; "result" })
val df = Seq(("test", 1)).toDF("s", "i").select(myUDF($"s"))
df.cache()
Copy link
Contributor Author

@icexelloss icexelloss Jun 11, 2018 •
edited
Loading

Choose a reason for hiding this comment

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

Here is when evalCount is incremented (because the optimizer will execute the UDF locally on driver), the rest of the tests have no effect on the value of evalCount.

evalCount is copied (by value) to the closure and therefore it's not possible to change the value evalCount defined in the driver by incrementing it in the UDF (because each closure has a different copy of evalCount)


df.collect()
assert(evalCount === 1)

df.collect()
assert(evalCount === 1)

val df2 = df.withColumn("newColumn", lit(1))
df2.collect()

// We should not reevaluate the cached dataframe
assert(evalCount === 1)
}

test("cache temp table") {
withTempView("tempTable") {
testData.select('key).createOrReplaceTempView("tempTable")
Expand Down
38 changes: 37 additions & 1 deletion sql/core/src/test/scala/org/apache/spark/sql/DatasetCacheSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,15 @@

package org.apache.spark.sql

import org.scalatest.concurrent.TimeLimits
import org.scalatest.time.SpanSugar._

import org.apache.spark.sql.functions._
import org.apache.spark.sql.test.SharedSQLContext
import org.apache.spark.storage.StorageLevel


class DatasetCacheSuite extends QueryTest with SharedSQLContext {
class DatasetCacheSuite extends QueryTest with SharedSQLContext with TimeLimits {
import testImplicits._

test("get storage level") {
Expand Down Expand Up @@ -96,4 +99,37 @@ class DatasetCacheSuite extends QueryTest with SharedSQLContext {
agged.unpersist()
assert(agged.storageLevel == StorageLevel.NONE, "The Dataset agged should not be cached.")
}

test("persist and then withColumn") {
val df = Seq(("test", 1)).toDF("s", "i")
val df2 = df.withColumn("newColumn", lit(1))

df.cache()
assertCached(df)
assertCached(df2)

df.count()
assertCached(df2)

df.unpersist()
assert(df.storageLevel == StorageLevel.NONE)
}

test("cache UDF result correctly") {
val expensiveUDF = udf({x: Int => Thread.sleep(10000); x})
val df = spark.range(0, 10).toDF("a").withColumn("b", expensiveUDF($"a"))
val df2 = df.agg(sum(df("b")))

df.cache()
df.count()
assertCached(df2)

// udf has been evaluated during caching, and thus should not be re-evaluated here
failAfter(5 seconds) {
df2.collect()
}

df.unpersist()
assert(df.storageLevel == StorageLevel.NONE)
}
}