Skip to content

[SPARK-54216][SQL] Add regression tests for V2 table cache refresh with immutable Table instances #55416

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
yadavay-amzn wants to merge 1 commit into from
Closed

[SPARK-54216][SQL] Add regression tests for V2 table cache refresh with immutable Table instances #55416

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
47 changes: 47 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 @@ -2681,6 +2681,53 @@ class CachedTableSuite extends QueryTest
}
}

test("SPARK-54216: refreshTable should return fresh data for V2 tables with immutable Table") {
val t = "testcat.tbl"
withTable(t) {
sql(s"CREATE TABLE $t (id int, data string) USING foo")
sql(s"INSERT INTO $t VALUES (1, 'a'), (2, 'b')")

// Cache the table and verify initial data
sql(s"CACHE TABLE $t")
assertCached(sql(s"SELECT * FROM $t"))
checkAnswer(sql(s"SELECT * FROM $t"), Seq(Row(1, "a"), Row(2, "b")))

// Insert new data
sql(s"INSERT INTO $t VALUES (3, 'c')")

// Refresh the table cache
spark.catalog.refreshTable(t)

// After refresh, the cache should return fresh data including the new row
assertCached(sql(s"SELECT * FROM $t"))
checkAnswer(sql(s"SELECT * FROM $t"), Seq(Row(1, "a"), Row(2, "b"), Row(3, "c")))
}
}

test("SPARK-54216: recacheByPlan should return fresh data for V2 tables") {
val t = "testcat.tbl"
withTable(t) {
sql(s"CREATE TABLE $t (id int, data string) USING foo")
sql(s"INSERT INTO $t VALUES (1, 'a'), (2, 'b')")

// Cache via DataFrame API
spark.table(t).cache().count()
assertCached(sql(s"SELECT * FROM $t"))
checkAnswer(sql(s"SELECT * FROM $t"), Seq(Row(1, "a"), Row(2, "b")))

// Insert new data
sql(s"INSERT INTO $t VALUES (3, 'c')")

// Recache by plan
val plan = spark.table(t).queryExecution.analyzed
cacheManager.recacheByPlan(spark, plan)

// After recache, should return fresh data
assertCached(sql(s"SELECT * FROM $t"))
checkAnswer(sql(s"SELECT * FROM $t"), Seq(Row(1, "a"), Row(2, "b"), Row(3, "c")))
}
}

private def cacheManager = spark.sharedState.cacheManager

private def pinTable(
Expand Down