[SPARK-55631][SQL] ALTER TABLE must invalidate cache for DSv2 tables

### What changes were proposed in this pull request?

This PR makes ALTER TABLE command invalidate cache.

### Why are the changes needed?

These changes are needed to reflect changes made in the session correctly.

### Does this PR introduce _any_ user-facing change?

ALTER TABLE commands will now invalidate cache.

### How was this patch tested?

This PR comes with test that would previously fail.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes #54427 from aokolnychyi/spark-55631.

Authored-by: Anton Okolnychyi <aokolnychyi@apache.org>
Signed-off-by: Gengliang Wang <gengliang@apache.org>

Author: aokolnychyi

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AlterTableExec.scala

@@ -31,7 +31,8 @@ import org.apache.spark.sql.execution.{SparkPlan, UnaryExecNode}
 case class AlterTableExec(
     catalog: TableCatalog,
     ident: Identifier,
-    changes: Seq[TableChange]) extends LeafV2CommandExec {
+    changes: Seq[TableChange],
+    refreshCache: () => Unit) extends LeafV2CommandExec {
 
   override def output: Seq[Attribute] = Seq.empty
 
@@ -42,7 +43,7 @@ case class AlterTableExec(
       case e: IllegalArgumentException if !e.isInstanceOf[SparkThrowable] =>
         throw QueryExecutionErrors.unsupportedTableChangeError(e)
     }
-
+    refreshCache()
     Seq.empty
   }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala

@@ -579,7 +579,11 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
       val table = a.table.asInstanceOf[ResolvedTable]
       ResolveTableConstraints.validateCatalogForTableChange(
         a.changes, table.catalog, table.identifier)
-      AlterTableExec(table.catalog, table.identifier, a.changes) :: Nil
+      AlterTableExec(
+        table.catalog,
+        table.identifier,
+        a.changes,
+        recacheTable(table, includeTimeTravel = false)) :: Nil
 
     case CreateIndex(ResolvedTable(_, _, table, _),
         indexName, indexType, ifNotExists, columns, properties) =>

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

@@ -2663,6 +2663,24 @@ class CachedTableSuite extends QueryTest
     }
   }
 
+  test("ALTER TABLE invalidates cached 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')")
+
+      sql(s"CACHE TABLE $t")
+      assertCached(sql(s"SELECT * FROM $t"))
+      checkAnswer(sql(s"SELECT * FROM $t"), Seq(Row(1, "a"), Row(2, "b")))
+
+      sql(s"ALTER TABLE $t ADD COLUMN new_col int")
+
+      val result = sql(s"SELECT * FROM $t ORDER BY id")
+      assertCached(result)
+      checkAnswer(result, Seq(Row(1, "a", null), Row(2, "b", null)))
+    }
+  }
+
   private def cacheManager = spark.sharedState.cacheManager
 
   private def pinTable(