[SPARK-56643][CONNECT][TESTS] Add DSv2 temp view with stored plan tests using Spark Connect mode

sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2TempViewConnectSuite.scala

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connect
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{DataFrame, QueryTest, Row, SparkSession}
+import org.apache.spark.sql.connector.DSv2TempViewWithStoredPlanTests
+import org.apache.spark.sql.connector.catalog.{CachingInMemoryTableCatalog, InMemoryTableCatalog, TableCatalog}
+
+/**
+ * Connect-mode runner for [[DSv2TempViewWithStoredPlanTests]]. All test logic lives in the shared
+ * trait; this class only provides the Connect-specific session, catalog access, and result
+ * comparison.
+ */
+class DataSourceV2TempViewConnectSuite
+    extends SparkConnectServerTest
+    with DSv2TempViewWithStoredPlanTests {
+
+  override def sparkConf: SparkConf = super.sparkConf
+    .set("spark.sql.catalog.testcat", classOf[InMemoryTableCatalog].getName)
+    .set("spark.sql.catalog.testcat.copyOnLoad", "true")
+    .set("spark.sql.catalog.cachingcat", classOf[CachingInMemoryTableCatalog].getName)
+    .set("spark.sql.catalog.cachingcat.copyOnLoad", "true")
+
+  override protected def testPrefix: String = "[connect] "
+
+  override protected def withTestSession(fn: SparkSession => Unit): Unit =
+    withSession(fn)
+
+  // Cannot use QueryTest.checkAnswer directly because it accesses df.logicalPlan,
+  // df.queryExecution, and df.materializedRdd, which are not available on Connect *client*
+  // DataFrames (they throw ConnectClientUnsupportedErrors). Note: checkAnswer IS usable from
+  // Connect server tests that operate on classic server-side DataFrames, but in this suite
+  // `df` is a Connect client DataFrame returned by session.table() / session.sql().
+  // Instead, collect the rows and delegate to QueryTest.sameRows, which is the same
+  // value-based, order-agnostic comparison that checkAnswer uses internally.
+  override protected def checkRows(df: => DataFrame, expected: Seq[Row]): Unit =
+    QueryTest.sameRows(expected, df.collect().toSeq).foreach(msg => fail(msg))
+
+  override protected def getTableCatalog[C <: TableCatalog: ClassTag](
+      session: SparkSession,
+      catalogName: String): C = {
+    val serverSession = getServerSession(session)
+    val catalog = serverSession.sessionState.catalogManager.catalog(catalogName)
+    val ct = implicitly[ClassTag[C]]
+    require(
+      ct.runtimeClass.isInstance(catalog),
+      s"Expected ${ct.runtimeClass.getName} but got ${catalog.getClass.getName}")
+    catalog.asInstanceOf[C]
+  }
+
+  // No explicit clearCache() for cachingcat is needed here, unlike the classic suite.
+  // Each withSession call creates a freshly isolated SparkSession on the server side
+  // (via SparkConnectSessionManager.newIsolatedSession), and afterEach invalidates all
+  // sessions, so the CachingInMemoryTableCatalog instance is per-test.
+  override protected def withTestTableAndViews(
+      session: SparkSession,
+      table: String,
+      views: Seq[String] = Seq.empty)(fn: => Unit): Unit = {
+    try { fn }
+    finally {
+      views.foreach(v => session.sql(s"DROP VIEW IF EXISTS $v").collect())
+      session.sql(s"DROP TABLE IF EXISTS $table").collect()
+    }
+  }
+}

sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2ExternalMutationTestBase.scala

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.connector
+
+import java.util
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.sql.{DataFrame, QueryTest, Row, SparkSession}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.catalog.{BufferedRows, CatalogV2Util, Identifier, InMemoryBaseTable, TableCatalog, TableWritePrivilege}
+
+/**
+ * Base trait for DSv2 tests that involve external table mutations (writes, schema changes,
+ * drop/recreate) via the catalog API.
+ *
+ * Provides abstract methods so that the same test scenarios can run in both classic mode
+ * (where the test session IS the server session) and Connect mode (where the test session
+ * is a Connect client and catalog access requires the server session).
+ *
+ * Concrete suites override the abstract methods and mix in the test trait
+ * [[DSv2TempViewWithStoredPlanTests]].
+ */
+trait DSv2ExternalMutationTestBase extends QueryTest {
+
+  /** Prefix for test names, e.g. "" or "[connect] ". */
+  protected def testPrefix: String
+
+  /** Execute a test body with a session. */
+  protected def withTestSession(fn: SparkSession => Unit): Unit
+
+  /**
+   * Assert that a DataFrame's rows match the expected rows (order-agnostic).
+   */
+  protected def checkRows(df: => DataFrame, expected: Seq[Row]): Unit
+
+  /**
+   * Get a [[TableCatalog]] by name from the underlying session.
+   */
+  protected def getTableCatalog[C <: TableCatalog: ClassTag](
+      session: SparkSession,
+      catalogName: String): C
+
+  /** Cleanup wrapper: drop views and the table after the test body, even on failure. */
+  protected def withTestTableAndViews(
+      session: SparkSession,
+      table: String,
+      views: Seq[String] = Seq.empty)(fn: => Unit): Unit
+
+  /** Appends a row to a DSv2 table via the catalog API, bypassing the session. */
+  protected def externalAppend(
+      catalog: TableCatalog,
+      ident: Identifier,
+      row: InternalRow): Unit = {
+    val extTable = catalog
+      .loadTable(ident, util.Set.of(TableWritePrivilege.INSERT))
+      .asInstanceOf[InMemoryBaseTable]
+    val schema = CatalogV2Util.v2ColumnsToStructType(extTable.columns())
+    extTable.withData(Array(new BufferedRows(Seq.empty, schema).withRow(row)))
+  }
+}