From e31157688d3998c99d37f14c9808b1621dca621a Mon Sep 17 00:00:00 2001
From: Anupam Yadav <anupamya@amazon.com>
Date: Sun, 19 Apr 2026 10:01:19 +0000
Subject: [PATCH] [SPARK-54216][SQL] Add regression tests for V2 table cache
 refresh with immutable Table instances

Add tests verifying that refreshTable() and recacheByPlan() return fresh data
for DataSource V2 tables with immutable Table instances (copyOnLoad=true).

The underlying issue was fixed by SPARK-54387 and SPARK-54424, which introduced
tryRefreshPlan to reload table metadata from the catalog before re-caching.
These tests serve as explicit regression coverage for the scenario described
in SPARK-54216.
---
 .../apache/spark/sql/CachedTableSuite.scala   | 47 +++++++++++++++++++
 1 file changed, 47 insertions(+)

diff --git a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
index a0e16634c63b8..321c307a23082 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/CachedTableSuite.scala
@@ -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(