You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "longvu-db (via GitHub)" <gi...@apache.org> on 2026/04/27 21:21:42 UTC
[PR] [SPARK-56643][CONNECT][TESTS] Add DSv2 temp view with stored plan tests using Spark Connect mode [spark]
longvu-db opened a new pull request, #55571:
URL: https://github.com/apache/spark/pull/55571
### What changes were proposed in this pull request?
Add 11 new tests in `DataSourceV2TempViewConnectSuite` that verify temporary view behavior with stored plans when the underlying DSv2 table changes, using Spark Connect. The tests cover all 7 scenarios from the DSv2 table refresh design doc:
1. **Session and external writes** (scenarios 1.1, 1.2): Temp view with filter reflects new data after session or external writes.
2. **Adding columns with data** (scenarios 2.1, 2.2): Temp view preserves original schema after ADD COLUMN, both session and external.
3. **Removing columns** (scenario 3.2): Temp view detects external column removal via catalog API.
4. **Drop and recreate table** (scenario 4.2): Temp view resolves to externally recreated table.
5. **Drop and re-add column with same type** (scenarios 5.1, 5.2): Schema validation passes, view continues working.
6. **Drop and re-add column with different type** (scenarios 6.1, 6.2): Temp view detects type change.
7. **Type widening** (scenario 7): Temp view detects INT to BIGINT type change.
Each test creates a DSv2 table via a Connect session, inserts initial data, builds a temp view with a filter (`salary < 999`) to demonstrate stored plan behavior, and verifies the view after table modifications.
External writes use the direct catalog API (`loadTable` + `withData`) to simulate writes from outside the session.
This is the Spark Connect counterpart of #55540.
### Why are the changes needed?
The existing SPARK-53924 tests in `DataSourceV2DataFrameSuite` cover basic schema change detection for temp views on DSv2 tables but are missing:
- Filter-based stored plan tests (the design doc uses a filter to demonstrate non-trivial stored plans)
- External write variants (using catalog API to simulate writes from outside the session)
- Scenarios 5, 6, 7 (drop+re-add column same/different type, type widening)
These tests provide comprehensive coverage for the DSv2 table refresh design doc's Section 1 (Temp views with stored plans), exercised through Spark Connect.
### Does this PR introduce _any_ user-facing change?
No. This PR only adds tests.
### How was this patch tested?
All 11 new tests pass:
```
build/sbt 'connect/testOnly *DataSourceV2TempViewConnectSuite'
```
### Was this patch authored or co-authored using generative AI tooling?
Yes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56643][CONNECT][TESTS] Add DSv2 temp view with stored plan tests using Spark Connect mode [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55571:
URL: https://github.com/apache/spark/pull/55571#discussion_r3280856607
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2TempViewConnectSuite.scala:
##########
@@ -0,0 +1,748 @@
+/*
+ * 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 java.util
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.classic
+import org.apache.spark.sql.connector.catalog.{BufferedRows, CachingInMemoryTableCatalog, Column, Identifier, InMemoryBaseTable, InMemoryTableCatalog, TableCatalog, TableChange, TableInfo, TableWritePrivilege}
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType}
+
+/**
+ * DSv2 temp view with stored plan tests for Spark Connect, mirroring the classic
+ * DataSourceV2DataFrameSuite temp view scenarios.
+ *
+ * Uses an in-process Connect server ([[SparkConnectServerTest]]) so that the test can access the
+ * server's catalog directly for external changes. All data reads go through the Connect client
+ * session to simulate the real client experience.
+ */
+class DataSourceV2TempViewConnectSuite extends SparkConnectServerTest {
+
+ 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")
+
+ private val T = "testcat.ns1.ns2.tbl"
+ private val CT = "cachingcat.ns1.ns2.tbl"
+ private val ident = Identifier.of(Array("ns1", "ns2"), "tbl")
+
+ /**
+ * Assert that rows collected through the Connect client match expected rows (order-agnostic).
+ */
+ private def assertRows(actual: Array[Row], expected: Seq[Row]): Unit = {
+ assert(
+ actual.toSeq.sortBy(_.toString()) == expected.sortBy(_.toString()),
+ s"Expected ${expected.mkString(", ")} but got ${actual.mkString(", ")}")
+ }
+
+ /** Get a catalog from the server-side session by name. */
+ private def serverCatalog[T <: TableCatalog](
+ serverSession: classic.SparkSession,
+ name: String): T =
+ serverSession.sessionState.catalogManager.catalog(name).asInstanceOf[T]
Review Comment:
@andreaschat-db's earlier suggestion to add `: ClassTag` on the type parameter wasn't picked up in the follow-up commits — worth keeping it, since `asInstanceOf[T]` at this boundary is a no-op under erasure (the cast really happens at the use site, producing a confusing classcast on a downstream method call instead of failing here at the helper boundary).
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2TempViewConnectSuite.scala:
##########
@@ -0,0 +1,748 @@
+/*
+ * 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 java.util
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.classic
+import org.apache.spark.sql.connector.catalog.{BufferedRows, CachingInMemoryTableCatalog, Column, Identifier, InMemoryBaseTable, InMemoryTableCatalog, TableCatalog, TableChange, TableInfo, TableWritePrivilege}
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType}
+
+/**
+ * DSv2 temp view with stored plan tests for Spark Connect, mirroring the classic
+ * DataSourceV2DataFrameSuite temp view scenarios.
+ *
+ * Uses an in-process Connect server ([[SparkConnectServerTest]]) so that the test can access the
+ * server's catalog directly for external changes. All data reads go through the Connect client
+ * session to simulate the real client experience.
+ */
+class DataSourceV2TempViewConnectSuite extends SparkConnectServerTest {
+
+ 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")
+
+ private val T = "testcat.ns1.ns2.tbl"
+ private val CT = "cachingcat.ns1.ns2.tbl"
+ private val ident = Identifier.of(Array("ns1", "ns2"), "tbl")
+
+ /**
+ * Assert that rows collected through the Connect client match expected rows (order-agnostic).
+ */
+ private def assertRows(actual: Array[Row], expected: Seq[Row]): Unit = {
+ assert(
+ actual.toSeq.sortBy(_.toString()) == expected.sortBy(_.toString()),
Review Comment:
Sorting and comparing by `_.toString()` is type-blind — two rows that stringify identically would compare equal regardless of underlying value types. `QueryTest.checkAnswer` is usable from connect server tests (it's already called statically in `sql/connect/server/src/test/scala/org/apache/spark/sql/connect/pipelines/PipelineRefreshFunctionalSuite.scala:351`) and would give value-based, order-agnostic comparison matching what the analogue suite uses. Worth wiring it in here rather than this custom helper.
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2TempViewConnectSuite.scala:
##########
@@ -0,0 +1,748 @@
+/*
+ * 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 java.util
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.classic
+import org.apache.spark.sql.connector.catalog.{BufferedRows, CachingInMemoryTableCatalog, Column, Identifier, InMemoryBaseTable, InMemoryTableCatalog, TableCatalog, TableChange, TableInfo, TableWritePrivilege}
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType}
+
+/**
+ * DSv2 temp view with stored plan tests for Spark Connect, mirroring the classic
+ * DataSourceV2DataFrameSuite temp view scenarios.
+ *
+ * Uses an in-process Connect server ([[SparkConnectServerTest]]) so that the test can access the
+ * server's catalog directly for external changes. All data reads go through the Connect client
+ * session to simulate the real client experience.
+ */
+class DataSourceV2TempViewConnectSuite extends SparkConnectServerTest {
+
+ 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")
+
+ private val T = "testcat.ns1.ns2.tbl"
+ private val CT = "cachingcat.ns1.ns2.tbl"
+ private val ident = Identifier.of(Array("ns1", "ns2"), "tbl")
+
+ /**
+ * Assert that rows collected through the Connect client match expected rows (order-agnostic).
+ */
+ private def assertRows(actual: Array[Row], expected: Seq[Row]): Unit = {
+ assert(
+ actual.toSeq.sortBy(_.toString()) == expected.sortBy(_.toString()),
+ s"Expected ${expected.mkString(", ")} but got ${actual.mkString(", ")}")
+ }
+
+ /** Get a catalog from the server-side session by name. */
+ private def serverCatalog[T <: TableCatalog](
+ serverSession: classic.SparkSession,
+ name: String): T =
+ serverSession.sessionState.catalogManager.catalog(name).asInstanceOf[T]
+
+ /** Appends a row to a DSv2 table via the catalog API, bypassing the session. */
+ private def externalAppend(
+ cat: TableCatalog,
+ ident: Identifier,
+ schema: StructType,
+ row: InternalRow): Unit = {
+ val extTable = cat
+ .loadTable(ident, util.Set.of(TableWritePrivilege.INSERT))
+ .asInstanceOf[InMemoryBaseTable]
+ extTable.withData(Array(new BufferedRows(Seq.empty, schema).withRow(row)))
+ }
+
+ /** Ensure views and table are dropped even if the test body throws. */
+ private def withTableAndViews(
+ session: SparkSession,
+ table: String,
+ views: Seq[String],
+ clearCachingCatalog: Boolean = false)(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()
+ if (clearCachingCatalog) {
+ val serverSession = getServerSession(session)
+ serverCatalog[CachingInMemoryTableCatalog](serverSession, "cachingcat").clearCache()
+ }
Review Comment:
Each `withSession` server-side gets a freshly isolated `SparkSession` (`SparkConnectSessionManager.newIsolatedSession` → `baseSession.newSession()`, which creates an independent `SessionState` and `CatalogManager`), and `afterEach` invalidates all sessions. The `CachingInMemoryTableCatalog` instance is therefore per-test, so `clearCache()` here is clearing state that wouldn't have survived anyway — the auto-review's compile-error fix on `0b76175` paid the complexity cost of a parameter + server-session lookup + generic cast for no observable behavior change. Two options: (a) drop the `clearCachingCatalog` parameter and the conditional entirely, or (b) keep it but add a one-line comment explaining "defensive — even though sessions are isolated, ..." so the next reader doesn't have to re-derive the analysis. Worth confirming this isn't relying on some intra-test reuse I haven't spotted.
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2TempViewConnectSuite.scala:
##########
@@ -0,0 +1,748 @@
+/*
+ * 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 java.util
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.classic
+import org.apache.spark.sql.connector.catalog.{BufferedRows, CachingInMemoryTableCatalog, Column, Identifier, InMemoryBaseTable, InMemoryTableCatalog, TableCatalog, TableChange, TableInfo, TableWritePrivilege}
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType}
+
+/**
+ * DSv2 temp view with stored plan tests for Spark Connect, mirroring the classic
+ * DataSourceV2DataFrameSuite temp view scenarios.
+ *
+ * Uses an in-process Connect server ([[SparkConnectServerTest]]) so that the test can access the
+ * server's catalog directly for external changes. All data reads go through the Connect client
+ * session to simulate the real client experience.
+ */
+class DataSourceV2TempViewConnectSuite extends SparkConnectServerTest {
+
+ 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")
+
+ private val T = "testcat.ns1.ns2.tbl"
+ private val CT = "cachingcat.ns1.ns2.tbl"
+ private val ident = Identifier.of(Array("ns1", "ns2"), "tbl")
+
+ /**
+ * Assert that rows collected through the Connect client match expected rows (order-agnostic).
+ */
+ private def assertRows(actual: Array[Row], expected: Seq[Row]): Unit = {
+ assert(
+ actual.toSeq.sortBy(_.toString()) == expected.sortBy(_.toString()),
+ s"Expected ${expected.mkString(", ")} but got ${actual.mkString(", ")}")
+ }
+
+ /** Get a catalog from the server-side session by name. */
+ private def serverCatalog[T <: TableCatalog](
+ serverSession: classic.SparkSession,
+ name: String): T =
+ serverSession.sessionState.catalogManager.catalog(name).asInstanceOf[T]
+
+ /** Appends a row to a DSv2 table via the catalog API, bypassing the session. */
+ private def externalAppend(
+ cat: TableCatalog,
+ ident: Identifier,
+ schema: StructType,
+ row: InternalRow): Unit = {
+ val extTable = cat
+ .loadTable(ident, util.Set.of(TableWritePrivilege.INSERT))
+ .asInstanceOf[InMemoryBaseTable]
+ extTable.withData(Array(new BufferedRows(Seq.empty, schema).withRow(row)))
+ }
Review Comment:
The analogue's `externalAppend` (`DataSourceV2DataFrameSuite.scala:2997`) derives the schema from the loaded table via `CatalogV2Util.v2ColumnsToStructType(extTable.columns())`. Here every call site duplicates the schema as a DDL string (`StructType.fromDDL("id INT, salary INT[, new_column INT]")` at lines 132, 159, 211, 242), opening the door to a caller's DDL silently diverging from the table's actual schema if a future scenario reshapes the table. Suggest dropping the `schema` parameter and computing it from `extTable.columns()` here, matching the analogue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56643][CONNECT][TESTS] Add DSv2 temp view with stored plan tests using Spark Connect mode [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #55571: [SPARK-56643][CONNECT][TESTS] Add DSv2 temp view with stored plan tests using Spark Connect mode
URL: https://github.com/apache/spark/pull/55571
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56643][CONNECT][TESTS] Add DSv2 temp view with stored plan tests using Spark Connect mode [spark]
Posted by "longvu-db (via GitHub)" <gi...@apache.org>.
longvu-db commented on PR #55571:
URL: https://github.com/apache/spark/pull/55571#issuecomment-4531238514
CI failed with unrelated with unrelated flaky `RocksDBStateStoreIntegrationSuiteWithRowChecksum` failure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56643][CONNECT][TESTS] Add DSv2 temp view with stored plan tests using Spark Connect mode [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #55571:
URL: https://github.com/apache/spark/pull/55571#discussion_r3298222821
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala:
##########
@@ -87,6 +88,30 @@ class DataSourceV2DataFrameSuite
catalog.asInstanceOf[InMemoryTableCatalog]
}
+ // DSv2ExternalMutationTestBase implementations for classic mode
+ override protected def testPrefix: String = "[classic] "
+
+ override protected def withTestSession(fn: SparkSession => Unit): Unit = fn(spark)
+
+ override protected def checkRows(df: => DataFrame, expected: Seq[Row]): Unit =
+ checkAnswer(df, expected)
+
+ override protected def getTableCatalog[C <: TableCatalog: ClassTag](
+ session: SparkSession,
+ catalogName: String): C = {
+ catalog(catalogName).asInstanceOf[C]
Review Comment:
The `ClassTag` context bound is declared but unused here — `catalog(catalogName).asInstanceOf[C]` is a no-op at the boundary under erasure, so a type mismatch surfaces later as a confusing `ClassCastException` on a downstream method call. The Connect impl (`DataSourceV2TempViewConnectSuite.scala:62-66`) already does the right thing with `require(ct.runtimeClass.isInstance(catalog), ...)`; mirroring it here keeps the failure mode symmetric.
```suggestion
val c = catalog(catalogName)
val ct = implicitly[ClassTag[C]]
require(
ct.runtimeClass.isInstance(c),
s"Expected ${ct.runtimeClass.getName} but got ${c.getClass.getName}")
c.asInstanceOf[C]
```
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala:
##########
@@ -87,6 +88,30 @@ class DataSourceV2DataFrameSuite
catalog.asInstanceOf[InMemoryTableCatalog]
}
+ // DSv2ExternalMutationTestBase implementations for classic mode
+ override protected def testPrefix: String = "[classic] "
Review Comment:
`${testPrefix}` renames the 21 classic tests from e.g. `"temp view with stored plan reflects session write"` to `"[classic] temp view with stored plan reflects session write"`. CI test-result trackers and flaky-test dashboards key on test name, so this looks like 21 deletions + 21 additions and the existing history is lost.
The existing Spark precedents for sharing tests between classic and Connect (`StaticProcedureSuiteBase`, `sql.SparkSessionBuilderImplementationBindingSuite`) use plain test names and rely on the suite class name to disambiguate — that would preserve history here and align with the established convention. If you want the `[connect] ` tag for readability on the Connect side specifically, you can leave Connect's prefix as-is and set the classic prefix to `""` so the existing names are preserved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56643][CONNECT][TESTS] Add DSv2 temp view with stored plan tests using Spark Connect mode [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #55571:
URL: https://github.com/apache/spark/pull/55571#issuecomment-4540685695
thanks, merging to master/4.x/4.2 (test only)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56643][CONNECT][TESTS] Add DSv2 temp view with stored plan tests using Spark Connect mode [spark]
Posted by "andreaschat-db (via GitHub)" <gi...@apache.org>.
andreaschat-db commented on code in PR #55571:
URL: https://github.com/apache/spark/pull/55571#discussion_r3208143037
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2TempViewConnectSuite.scala:
##########
@@ -0,0 +1,661 @@
+/*
+ * 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 java.util
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.classic
+import org.apache.spark.sql.connector.catalog.{BufferedRows, CachingInMemoryTableCatalog, Column, Identifier, InMemoryBaseTable, InMemoryTableCatalog, TableChange, TableInfo, TableWritePrivilege}
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType}
+
+/**
+ * DSv2 temp view with stored plan tests for Spark Connect, mirroring the classic
+ * DataSourceV2DataFrameSuite temp view scenarios.
+ *
+ * Uses an in-process Connect server ([[SparkConnectServerTest]]) so that the test can access the
+ * server's catalog directly for external changes. All data reads go through the Connect client
+ * session to simulate the real client experience.
+ */
+class DataSourceV2TempViewConnectSuite extends SparkConnectServerTest {
+
+ 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")
+
+ private val T = "testcat.ns1.ns2.tbl"
+ private val CT = "cachingcat.ns1.ns2.tbl"
+ private val ident = Identifier.of(Array("ns1", "ns2"), "tbl")
+
+ /**
+ * Assert that rows collected through the Connect client match expected rows (order-agnostic).
+ */
+ private def assertRows(actual: Array[Row], expected: Seq[Row]): Unit = {
+ assert(
+ actual.map(_.toString()).toSet == expected.map(_.toString()).toSet,
+ s"Expected ${expected.mkString(", ")} but got ${actual.mkString(", ")}")
+ }
+
+ /** Get the catalog from the server-side session. */
+ private def serverCatalog(serverSession: classic.SparkSession): InMemoryTableCatalog =
+ serverSession.sessionState.catalogManager
+ .catalog("testcat")
+ .asInstanceOf[InMemoryTableCatalog]
+
+ /** Get the caching catalog from the server-side session. */
+ private def serverCachingCatalog(
+ serverSession: classic.SparkSession): CachingInMemoryTableCatalog =
+ serverSession.sessionState.catalogManager
+ .catalog("cachingcat")
+ .asInstanceOf[CachingInMemoryTableCatalog]
+
+ // Temp views with stored plans: scenarios from the DSv2 table refresh tests.
+ // Each test creates a DSv2 table with initial data, builds a temp view with a filter
+ // (to demonstrate that the stored plan is non-trivial), and then verifies the view
+ // behavior after various table modifications (session or external).
+
+ // Scenario 1.1 (session write)
+ test("[connect] temp view with stored plan reflects session write") {
+ withSession { session =>
+ session.sql(s"CREATE TABLE $T (id INT, salary INT) USING foo").collect()
+ session.sql(s"INSERT INTO $T VALUES (1, 100), (10, 1000)").collect()
+
+ session.table(T).filter("salary < 999").createOrReplaceTempView("v")
+ assertRows(session.table("v").collect(), Seq(Row(1, 100)))
+
+ session.sql(s"INSERT INTO $T VALUES (2, 200)").collect()
+
+ assertRows(session.table("v").collect(), Seq(Row(1, 100), Row(2, 200)))
+
+ session.sql("DROP VIEW IF EXISTS v").collect()
Review Comment:
Can we create a helper here so we do not have to DROP VIEW/TABLE at each test? More importantly, the drops need to be in a `try {} finally {}`. That ensure that if a tests fails everything is still cleaned up.
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2TempViewConnectSuite.scala:
##########
@@ -0,0 +1,661 @@
+/*
+ * 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 java.util
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.classic
+import org.apache.spark.sql.connector.catalog.{BufferedRows, CachingInMemoryTableCatalog, Column, Identifier, InMemoryBaseTable, InMemoryTableCatalog, TableChange, TableInfo, TableWritePrivilege}
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType}
+
+/**
+ * DSv2 temp view with stored plan tests for Spark Connect, mirroring the classic
+ * DataSourceV2DataFrameSuite temp view scenarios.
+ *
+ * Uses an in-process Connect server ([[SparkConnectServerTest]]) so that the test can access the
+ * server's catalog directly for external changes. All data reads go through the Connect client
+ * session to simulate the real client experience.
+ */
+class DataSourceV2TempViewConnectSuite extends SparkConnectServerTest {
+
+ 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")
+
+ private val T = "testcat.ns1.ns2.tbl"
+ private val CT = "cachingcat.ns1.ns2.tbl"
+ private val ident = Identifier.of(Array("ns1", "ns2"), "tbl")
+
+ /**
+ * Assert that rows collected through the Connect client match expected rows (order-agnostic).
+ */
+ private def assertRows(actual: Array[Row], expected: Seq[Row]): Unit = {
+ assert(
+ actual.map(_.toString()).toSet == expected.map(_.toString()).toSet,
Review Comment:
We could instead do: `actual.toSeq.sortBy(_.toString) == expected.sortBy(_.toString)` so we preserve the original type and duplicates.
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2TempViewConnectSuite.scala:
##########
@@ -0,0 +1,661 @@
+/*
+ * 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 java.util
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.classic
+import org.apache.spark.sql.connector.catalog.{BufferedRows, CachingInMemoryTableCatalog, Column, Identifier, InMemoryBaseTable, InMemoryTableCatalog, TableChange, TableInfo, TableWritePrivilege}
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType}
+
+/**
+ * DSv2 temp view with stored plan tests for Spark Connect, mirroring the classic
+ * DataSourceV2DataFrameSuite temp view scenarios.
+ *
+ * Uses an in-process Connect server ([[SparkConnectServerTest]]) so that the test can access the
+ * server's catalog directly for external changes. All data reads go through the Connect client
+ * session to simulate the real client experience.
+ */
+class DataSourceV2TempViewConnectSuite extends SparkConnectServerTest {
+
+ 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")
+
+ private val T = "testcat.ns1.ns2.tbl"
+ private val CT = "cachingcat.ns1.ns2.tbl"
+ private val ident = Identifier.of(Array("ns1", "ns2"), "tbl")
+
+ /**
+ * Assert that rows collected through the Connect client match expected rows (order-agnostic).
+ */
+ private def assertRows(actual: Array[Row], expected: Seq[Row]): Unit = {
+ assert(
+ actual.map(_.toString()).toSet == expected.map(_.toString()).toSet,
+ s"Expected ${expected.mkString(", ")} but got ${actual.mkString(", ")}")
+ }
+
+ /** Get the catalog from the server-side session. */
+ private def serverCatalog(serverSession: classic.SparkSession): InMemoryTableCatalog =
Review Comment:
```suggestion
private def serverCatalog[T <: TableCatalog: ClassTag](
serverSession: classic.SparkSession, name: String): T =
serverSession.sessionState.catalogManager.catalog(name).asInstanceOf[T]
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56643][CONNECT][TESTS] Add DSv2 temp view with stored plan tests using Spark Connect mode [spark]
Posted by "andreaschat-db (via GitHub)" <gi...@apache.org>.
andreaschat-db commented on code in PR #55571:
URL: https://github.com/apache/spark/pull/55571#discussion_r3209131688
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2TempViewConnectSuite.scala:
##########
@@ -0,0 +1,781 @@
+/*
+ * 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 java.util
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{AnalysisException, Row, SparkSession}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.classic
+import org.apache.spark.sql.connector.catalog.{BufferedRows, CachingInMemoryTableCatalog, Column, Identifier, InMemoryBaseTable, InMemoryTableCatalog, TableCatalog, TableChange, TableInfo, TableWritePrivilege}
+import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType}
+
+/**
+ * DSv2 temp view with stored plan tests for Spark Connect, mirroring the classic
+ * DataSourceV2DataFrameSuite temp view scenarios.
+ *
+ * Uses an in-process Connect server ([[SparkConnectServerTest]]) so that the test can access the
+ * server's catalog directly for external changes. All data reads go through the Connect client
+ * session to simulate the real client experience.
+ */
+class DataSourceV2TempViewConnectSuite extends SparkConnectServerTest {
+
+ 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")
+
+ private val T = "testcat.ns1.ns2.tbl"
+ private val CT = "cachingcat.ns1.ns2.tbl"
+ private val ident = Identifier.of(Array("ns1", "ns2"), "tbl")
+
+ /**
+ * Assert that rows collected through the Connect client match expected rows (order-agnostic).
+ */
+ private def assertRows(actual: Array[Row], expected: Seq[Row]): Unit = {
+ assert(
+ actual.toSeq.sortBy(_.toString()) == expected.sortBy(_.toString()),
+ s"Expected ${expected.mkString(", ")} but got ${actual.mkString(", ")}")
+ }
+
+ /** Get a catalog from the server-side session by name. */
+ private def serverCatalog[T <: TableCatalog](
+ serverSession: classic.SparkSession, name: String): T =
+ serverSession.sessionState.catalogManager.catalog(name).asInstanceOf[T]
+
+ /** Appends a row to a DSv2 table via the catalog API, bypassing the session. */
+ private def externalAppend(
+ cat: TableCatalog,
+ ident: Identifier,
+ schema: StructType,
+ row: InternalRow): Unit = {
+ val extTable = cat
+ .loadTable(ident, util.Set.of(TableWritePrivilege.INSERT))
+ .asInstanceOf[InMemoryBaseTable]
+ extTable.withData(Array(new BufferedRows(Seq.empty, schema).withRow(row)))
+ }
+
+ /** Ensure views and table are dropped even if the test body throws. */
+ private def withTableAndView(
Review Comment:
nit: `withTableAndViews`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56643][CONNECT][TESTS] Add DSv2 temp view with stored plan tests using Spark Connect mode [spark]
Posted by "longvu-db (via GitHub)" <gi...@apache.org>.
longvu-db commented on PR #55571:
URL: https://github.com/apache/spark/pull/55571#issuecomment-4499515171
## Auto-review results
**1 critical issue found and fixed:**
`CachingInMemoryTableCatalog.clearCache()` in `withTableAndViews` (line 87) was called as a companion-object static method, but no companion object exists — only an instance method. This would not compile.
**Fix** (commit 0b76175): Get the server-side catalog instance and call `clearCache()` on it, matching the pattern used by the classic `DataSourceV2DataFrameSuite`:
```scala
if (clearCachingCatalog) {
val serverSession = getServerSession(session)
serverCatalog[CachingInMemoryTableCatalog](serverSession, "cachingcat").clearCache()
}
```
Also removed `CachingInMemoryTableCatalog.scala` and `InMemoryTableCatalog.scala` from this PR (both already in master).
No other issues found. Test logic, assertions, error conditions, cleanup patterns, imports, and style all look good. ✅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org