This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 8261b7884102 [SPARK-58149][SQL][TEST] Refactor some DSv2 tests to use
SessionQueryTest
8261b7884102 is described below
commit 8261b7884102cc2fb1084bb4d5c553d51eb5fedc
Author: Matthis Gördel <[email protected]>
AuthorDate: Mon Jul 20 22:15:33 2026 +0800
[SPARK-58149][SQL][TEST] Refactor some DSv2 tests to use SessionQueryTest
### What changes were proposed in this pull request?
This PR builds upon https://github.com/apache/spark/pull/56190 and
refactors the existing classic/connect-targeting DSv2 tests to use
`SessionQueryTest`.
NOTE: this PR is stacked upon https://github.com/apache/spark/pull/56190
and WIP until that PR is merged and this PR is rebased. The actual changes are
to the files in `sql/core/src/test/scala/org/apache/spark/sql/connector/` and
to
`sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2DataFrameConnectSuite.scala`
(currently ea62c553484b9e293677a7516d8ed0ed3e2d2ba4)
### Why are the changes needed?
This PR reduces the amount of custom test helpers.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
This patch is test-only
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #56708 from fwc/refactor-some-tests-to-sessionquerytest.
Lead-authored-by: Matthis Gördel <[email protected]>
Co-authored-by: Matthis Gördel <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit b50454d6c48b88dbfb4bd1e482126e2e767b636e)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../DataSourceV2DataFrameConnectSuite.scala | 61 +-
.../sql/connector/DSv2CacheTableReadTests.scala | 322 +++++------
.../connector/DSv2ExternalMutationTestBase.scala | 21 +-
.../DSv2IncrementallyConstructedQueryTests.scala | 641 ++++++++++-----------
.../connector/DSv2RepeatedTableAccessTests.scala | 252 ++++----
.../DSv2TempViewWithStoredPlanTests.scala | 412 ++++++-------
.../sql/connector/DataSourceV2DataFrameSuite.scala | 19 +-
7 files changed, 819 insertions(+), 909 deletions(-)
diff --git
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2DataFrameConnectSuite.scala
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2DataFrameConnectSuite.scala
index 1a31e5f8ac1a..bbff687a16b1 100644
---
a/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2DataFrameConnectSuite.scala
+++
b/sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2DataFrameConnectSuite.scala
@@ -20,7 +20,8 @@ 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.{classic, connect, SparkSession}
+import org.apache.spark.sql.connect.service.{SessionKey, SparkConnectService}
import org.apache.spark.sql.connector.{DSv2CacheTableReadTests,
DSv2IncrementallyConstructedQueryTests, DSv2RepeatedTableAccessTests,
DSv2TempViewWithStoredPlanTests}
import org.apache.spark.sql.connector.catalog.{CachingInMemoryTableCatalog,
InMemoryTableCatalog, NullTableIdAndNullColumnIdInMemoryTableCatalog,
NullTableIdInMemoryTableCatalog, TableCatalog}
@@ -34,7 +35,7 @@ import
org.apache.spark.sql.connector.catalog.{CachingInMemoryTableCatalog, InMe
* this class only provides the Connect-specific session, catalog access, and
result comparison.
*/
class DataSourceV2DataFrameConnectSuite
- extends SparkConnectServerTest
+ extends SessionQueryTest
with DSv2TempViewWithStoredPlanTests
with DSv2RepeatedTableAccessTests
with DSv2IncrementallyConstructedQueryTests
@@ -53,45 +54,43 @@ class DataSourceV2DataFrameConnectSuite
.set("spark.sql.catalog.nullbothidscat.copyOnLoad", "true")
override protected def testPrefix: String = "[connect] "
- override protected def isConnect: Boolean = true
- override protected def withTestSession(fn: SparkSession => Unit): Unit =
- withSession(fn)
+ // Unlike the classic suite, the connect suite shares a single server-side
session across all
+ // tests in the suite (created once by SparkSessionBinder), so catalog state
is not reset
+ // between tests. Like the classic suite's `after` block, clear the caching
connector's cache
+ // so each test starts from a clean catalog state. Without this,
+ // CachingInMemoryTableCatalog retains stale table entries (it does not
invalidate on drop), so a
+ // later test's CREATE TABLE fails with TABLE_OR_VIEW_ALREADY_EXISTS.
+ override protected def afterEach(): Unit = {
+ val serverSession = getServerSession(spark)
+ if
(serverSession.sessionState.catalogManager.isCatalogRegistered("cachingcat")) {
+ serverSession.sessionState.catalogManager
+ .catalog("cachingcat")
+ .asInstanceOf[CachingInMemoryTableCatalog]
+ .clearCache()
+ }
+ super.afterEach()
+ }
- // 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))
+ protected def getServerSession(clientSession: SparkSession):
classic.SparkSession = {
+ val connectSession = clientSession.asInstanceOf[connect.SparkSession]
+ val userId = connectSession.client.userId
+ val sessionId = connectSession.sessionId
+ val key = SessionKey(userId, sessionId)
+ SparkConnectService.sessionManager
+ .getIsolatedSessionIfPresent(key)
+ .get
+ .session
+ }
override protected def getTableCatalog[C <: TableCatalog: ClassTag](
session: SparkSession,
catalogName: String): C = {
- val serverSession = getServerSession(session)
- val catalog =
serverSession.sessionState.catalogManager.catalog(catalogName)
+ val catalog =
getServerSession(session).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()
- }
- }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2CacheTableReadTests.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2CacheTableReadTests.scala
index ac6ffcc6ecc0..cd9673ae7283 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2CacheTableReadTests.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2CacheTableReadTests.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.connector
-import org.apache.spark.sql.{Row, SparkSession}
+import org.apache.spark.sql.Row
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.connector.catalog.{CachingInMemoryTableCatalog,
Column, InMemoryTableCatalog, TableChange, TableInfo}
import org.apache.spark.sql.types.IntegerType
@@ -49,223 +49,209 @@ import org.apache.spark.sql.types.IntegerType
* (via the CacheManager), making a session drop+recreate scenario trivially
different from
* the external variant.
*
- * NOTE: All `session.sql(...)` calls append `.collect()` because Connect
client DataFrames
+ * NOTE: All `spark.sql(...)` calls append `.collect()` because Connect client
DataFrames
* are lazy and require an action to trigger execution. In classic mode
`.collect()` on
* DDL / DML is a no-op (these execute eagerly), so this is harmless.
*/
trait DSv2CacheTableReadTests extends DSv2ExternalMutationTestBase {
- private def assertTableCached(session: SparkSession, tableName: String):
Unit =
- assert(session.catalog.isCached(tableName))
+ private def assertTableCached(tableName: String): Unit =
+ assert(spark.catalog.isCached(tableName))
test(s"${testPrefix}SPARK-54022: cached table pinned against external data
write") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
- session.table(testTable).cache()
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100)))
+ spark.table(testTable).cache()
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100)))
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100)))
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100)))
- session.sql(s"REFRESH TABLE $testTable").collect()
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100), Row(2, 200)))
- }
+ spark.sql(s"REFRESH TABLE $testTable").collect()
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100), Row(2, 200)))
}
}
test(s"${testPrefix}SPARK-54022: connector w/ cache: cached table pinned, " +
"REFRESH clears both layers") {
- withTestSession { session =>
- withTestTableAndViews(session, cachingTestTable) {
- session.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT)
USING foo").collect()
- session.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100)").collect()
-
- session.table(cachingTestTable).cache()
- assertTableCached(session, cachingTestTable)
- checkRows(session.table(cachingTestTable), Seq(Row(1, 100)))
-
- val catalog =
- getTableCatalog[CachingInMemoryTableCatalog](session, "cachingcat")
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
-
- // Both CacheManager and connector cache are stale: external write
invisible
- assertTableCached(session, cachingTestTable)
- checkRows(session.table(cachingTestTable), Seq(Row(1, 100)))
-
- // REFRESH TABLE calls invalidateTable (clears connector cache) and
rebuilds
- // the CacheManager entry, so the external write becomes visible.
- session.sql(s"REFRESH TABLE $cachingTestTable").collect()
- assertTableCached(session, cachingTestTable)
- checkRows(session.table(cachingTestTable), Seq(Row(1, 100), Row(2,
200)))
- }
+ withTable(cachingTestTable) {
+ spark.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100)").collect()
+
+ spark.table(cachingTestTable).cache()
+ assertTableCached(cachingTestTable)
+ checkAnswer(spark.table(cachingTestTable), Seq(Row(1, 100)))
+
+ val catalog =
+ getTableCatalog[CachingInMemoryTableCatalog](spark, "cachingcat")
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
+
+ // Both CacheManager and connector cache are stale: external write
invisible
+ assertTableCached(cachingTestTable)
+ checkAnswer(spark.table(cachingTestTable), Seq(Row(1, 100)))
+
+ // REFRESH TABLE calls invalidateTable (clears connector cache) and
rebuilds
+ // the CacheManager entry, so the external write becomes visible.
+ spark.sql(s"REFRESH TABLE $cachingTestTable").collect()
+ assertTableCached(cachingTestTable)
+ checkAnswer(spark.table(cachingTestTable), Seq(Row(1, 100), Row(2, 200)))
}
}
test(s"${testPrefix}SPARK-54022: session write invalidates cache, " +
"then external write invisible") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
- session.table(testTable).cache()
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100)))
+ spark.table(testTable).cache()
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100)))
- session.sql(s"INSERT INTO $testTable VALUES (2, 200)").collect()
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100), Row(2, 200)))
+ spark.sql(s"INSERT INTO $testTable VALUES (2, 200)").collect()
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100), Row(2, 200)))
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(3, 300))
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(3, 300))
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100), Row(2, 200)))
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100), Row(2, 200)))
- session.sql(s"REFRESH TABLE $testTable").collect()
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100), Row(2, 200),
Row(3, 300)))
- }
+ spark.sql(s"REFRESH TABLE $testTable").collect()
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100), Row(2, 200), Row(3,
300)))
}
}
test(s"${testPrefix}SPARK-54022: cached table pinned against external schema
change") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
-
- session.table(testTable).cache()
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100)))
-
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
- val addCol = TableChange.addColumn(Array("new_column"), IntegerType,
true)
- catalog.alterTable(testIdent, addCol)
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200, -1))
-
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100)))
-
- session.sql(s"REFRESH TABLE $testTable").collect()
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100, null), Row(2, 200,
-1)))
- }
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+
+ spark.table(testTable).cache()
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100)))
+
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
+ val addCol = TableChange.addColumn(Array("new_column"), IntegerType,
true)
+ catalog.alterTable(testIdent, addCol)
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200, -1))
+
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100)))
+
+ spark.sql(s"REFRESH TABLE $testTable").collect()
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100, null), Row(2, 200,
-1)))
}
}
test(s"${testPrefix}SPARK-54022: session schema change invalidates cache, " +
"external write invisible") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
- session.table(testTable).cache()
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100)))
+ spark.table(testTable).cache()
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100)))
- session.sql(s"ALTER TABLE $testTable ADD COLUMN new_column
INT").collect()
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100, null)))
+ spark.sql(s"ALTER TABLE $testTable ADD COLUMN new_column INT").collect()
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100, null)))
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200, -1))
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200, -1))
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100, null)))
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100, null)))
- session.sql(s"REFRESH TABLE $testTable").collect()
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100, null), Row(2, 200,
-1)))
- }
+ spark.sql(s"REFRESH TABLE $testTable").collect()
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100, null), Row(2, 200,
-1)))
}
}
test(s"${testPrefix}SPARK-54022: cached table after external drop and " +
"recreate sees empty table") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
-
- session.table(testTable).cache()
- assertTableCached(session, testTable)
- checkRows(session.table(testTable), Seq(Row(1, 100)))
-
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
- val originalTableId = catalog.loadTable(testIdent).id
-
- catalog.dropTable(testIdent)
- catalog.createTable(
- testIdent,
- new TableInfo.Builder()
- .withColumns(Array(
- Column.create("id", IntegerType),
- Column.create("salary", IntegerType)))
- .build())
-
- val newTableId = catalog.loadTable(testIdent).id
- assert(originalTableId != newTableId)
-
- val result = session.table(testTable)
- assert(result.schema.fieldNames.toSeq == Seq("id", "salary"))
- checkRows(result, Seq.empty)
-
- // External drop+recreate produces a new table identity, so the prior
cache entry
- // is unreachable via name lookup (unlike external write/schema change
where the
- // cache stays pinned).
- assert(!session.catalog.isCached(testTable))
-
- session.sql(s"REFRESH TABLE $testTable").collect()
- checkRows(session.table(testTable), Seq.empty)
- }
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+
+ spark.table(testTable).cache()
+ assertTableCached(testTable)
+ checkAnswer(spark.table(testTable), Seq(Row(1, 100)))
+
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
+ val originalTableId = catalog.loadTable(testIdent).id
+
+ catalog.dropTable(testIdent)
+ catalog.createTable(
+ testIdent,
+ new TableInfo.Builder()
+ .withColumns(Array(
+ Column.create("id", IntegerType),
+ Column.create("salary", IntegerType)))
+ .build())
+
+ val newTableId = catalog.loadTable(testIdent).id
+ assert(originalTableId != newTableId)
+
+ val result = spark.table(testTable)
+ assert(result.schema.fieldNames.toSeq == Seq("id", "salary"))
+ checkAnswer(result, Seq.empty)
+
+ // External drop+recreate produces a new table identity, so the prior
cache entry
+ // is unreachable via name lookup (unlike external write/schema change
where the
+ // cache stays pinned).
+ assert(!spark.catalog.isCached(testTable))
+
+ spark.sql(s"REFRESH TABLE $testTable").collect()
+ checkAnswer(spark.table(testTable), Seq.empty)
}
}
test(s"${testPrefix}SPARK-54022: connector w/ cache: cached table stale
after " +
"external drop and recreate") {
- withTestSession { session =>
- withTestTableAndViews(session, cachingTestTable) {
- session.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT)
USING foo").collect()
- session.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100)").collect()
-
- session.table(cachingTestTable).cache()
- assertTableCached(session, cachingTestTable)
- checkRows(session.table(cachingTestTable), Seq(Row(1, 100)))
-
- val catalog =
- getTableCatalog[CachingInMemoryTableCatalog](session, "cachingcat")
- val originalTableId = catalog.loadTable(testIdent).id
-
- catalog.dropTable(testIdent)
- catalog.createTable(
- testIdent,
- new TableInfo.Builder()
- .withColumns(Array(
- Column.create("id", IntegerType),
- Column.create("salary", IntegerType)))
- .build())
-
- // CachingInMemoryTableCatalog does not invalidate on drop/create, so
loadTable
- // still returns the old cached table object. CacheManager still
matches and
- // serves the stale cached data.
- assertTableCached(session, cachingTestTable)
- checkRows(session.table(cachingTestTable), Seq(Row(1, 100)))
-
- // REFRESH TABLE calls invalidateTable (clears connector cache) and
rebuilds
- // the CacheManager entry, so the new empty table becomes visible.
- session.sql(s"REFRESH TABLE $cachingTestTable").collect()
- checkRows(session.table(cachingTestTable), Seq.empty)
- }
+ withTable(cachingTestTable) {
+ spark.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100)").collect()
+
+ spark.table(cachingTestTable).cache()
+ assertTableCached(cachingTestTable)
+ checkAnswer(spark.table(cachingTestTable), Seq(Row(1, 100)))
+
+ val catalog =
+ getTableCatalog[CachingInMemoryTableCatalog](spark, "cachingcat")
+ val originalTableId = catalog.loadTable(testIdent).id
+
+ catalog.dropTable(testIdent)
+ catalog.createTable(
+ testIdent,
+ new TableInfo.Builder()
+ .withColumns(Array(
+ Column.create("id", IntegerType),
+ Column.create("salary", IntegerType)))
+ .build())
+
+ // CachingInMemoryTableCatalog does not invalidate on drop/create, so
loadTable
+ // still returns the old cached table object. CacheManager still matches
and
+ // serves the stale cached data.
+ assertTableCached(cachingTestTable)
+ checkAnswer(spark.table(cachingTestTable), Seq(Row(1, 100)))
+
+ // REFRESH TABLE calls invalidateTable (clears connector cache) and
rebuilds
+ // the CacheManager entry, so the new empty table becomes visible.
+ spark.sql(s"REFRESH TABLE $cachingTestTable").collect()
+ checkAnswer(spark.table(cachingTestTable), Seq.empty)
}
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2ExternalMutationTestBase.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2ExternalMutationTestBase.scala
index 0b2a50534447..73c69f8a9de4 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2ExternalMutationTestBase.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2ExternalMutationTestBase.scala
@@ -21,7 +21,7 @@ import java.util
import scala.reflect.ClassTag
-import org.apache.spark.sql.{DataFrame, QueryTest, Row, SparkSession}
+import org.apache.spark.sql.{SessionQueryTestBase, SparkSession}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.connector.catalog.{BufferedRows, CatalogV2Util,
Identifier, InMemoryBaseTable, TableCatalog, TableWritePrivilege}
@@ -37,7 +37,7 @@ import org.apache.spark.sql.connector.catalog.{BufferedRows,
CatalogV2Util, Iden
* [[DSv2TempViewWithStoredPlanTests]], [[DSv2RepeatedTableAccessTests]],
* [[DSv2IncrementallyConstructedQueryTests]], or [[DSv2CacheTableReadTests]].
*/
-trait DSv2ExternalMutationTestBase extends QueryTest {
+trait DSv2ExternalMutationTestBase extends SessionQueryTestBase {
/** Fully qualified table name under the non-caching test catalog. */
protected val testTable: String = "testcat.ns1.ns2.tbl"
@@ -51,17 +51,6 @@ trait DSv2ExternalMutationTestBase extends QueryTest {
/** Prefix for test names, e.g. "" or "[connect] ". */
protected def testPrefix: String
- /** Whether this suite runs under Spark Connect. */
- protected def isConnect: Boolean
-
- /** 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.
*/
@@ -69,12 +58,6 @@ trait DSv2ExternalMutationTestBase extends QueryTest {
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,
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2IncrementallyConstructedQueryTests.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2IncrementallyConstructedQueryTests.scala
index 9753f7db2097..aa8c63ff602d 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2IncrementallyConstructedQueryTests.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2IncrementallyConstructedQueryTests.scala
@@ -32,7 +32,7 @@ import org.apache.spark.unsafe.types.UTF8String
* mode, resolution is deferred until execution, so both sides of a join
always see the
* latest table state.
*
- * NOTE: All `session.sql(...)` calls append `.collect()` because Connect
client DataFrames
+ * NOTE: All `spark.sql(...)` calls append `.collect()` because Connect client
DataFrames
* are lazy and require an action to trigger execution. In classic mode
`.collect()` on
* eager statements (DDL, INSERT) is a no-op, so this is harmless.
*/
@@ -45,42 +45,38 @@ trait DSv2IncrementallyConstructedQueryTests extends
DSv2ExternalMutationTestBas
test(s"${testPrefix}SPARK-54157: join refreshes both sides after external
insert" +
" (table with both table and column ID support)") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
- val df1 = session.table(testTable)
+ val df1 = spark.table(testTable)
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
- val df2 = session.table(testTable)
+ val df2 = spark.table(testTable)
- checkRows(
- df1.join(df2, df1("id") === df2("id")),
- Seq(Row(1, 100, 1, 100), Row(2, 200, 2, 200)))
- }
+ checkAnswer(
+ df1.join(df2, df1("id") === df2("id")),
+ Seq(Row(1, 100, 1, 100), Row(2, 200, 2, 200)))
}
}
test(s"${testPrefix}SPARK-54157: join refreshes both sides after
same-session insert" +
" (table with both table and column ID support)") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
- val df1 = session.table(testTable)
+ val df1 = spark.table(testTable)
- session.sql(s"INSERT INTO $testTable VALUES (2, 200)").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (2, 200)").collect()
- val df2 = session.table(testTable)
+ val df2 = spark.table(testTable)
- checkRows(
- df1.join(df2, df1("id") === df2("id")),
- Seq(Row(1, 100, 1, 100), Row(2, 200, 2, 200)))
- }
+ checkAnswer(
+ df1.join(df2, df1("id") === df2("id")),
+ Seq(Row(1, 100, 1, 100), Row(2, 200, 2, 200)))
}
}
@@ -92,67 +88,63 @@ trait DSv2IncrementallyConstructedQueryTests extends
DSv2ExternalMutationTestBas
test(s"${testPrefix}SPARK-54157: join after external ADD COLUMN" +
" (table with both table and column ID support)") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
-
- val df1 = session.table(testTable)
-
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
- catalog.alterTable(
- testIdent, TableChange.addColumn(Array("new_column"), IntegerType,
true))
- externalAppend(
- catalog = catalog, ident = testIdent, row = InternalRow(2, 200, -1))
-
- val df2 = session.table(testTable)
- val selfJoin = df1.join(df2, df1("id") === df2("id"))
-
- if (isConnect) {
- // Connect re-resolves df1 with the new 3-column schema (id, salary,
new_column).
- assert(selfJoin.columns.length == 6,
- s"Expected 6 columns (3 + 3) but got:
${selfJoin.columns.mkString(", ")}")
- checkRows(selfJoin,
- Seq(Row(1, 100, null, 1, 100, null), Row(2, 200, -1, 2, 200, -1)))
- } else {
- // Classic: df1 keeps its original 2-column schema (id, salary).
- assert(selfJoin.columns.length == 5,
- s"Expected 5 columns (2 + 3) but got:
${selfJoin.columns.mkString(", ")}")
- checkRows(selfJoin,
- Seq(Row(1, 100, 1, 100, null), Row(2, 200, 2, 200, -1)))
- }
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+
+ val df1 = spark.table(testTable)
+
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
+ catalog.alterTable(
+ testIdent, TableChange.addColumn(Array("new_column"), IntegerType,
true))
+ externalAppend(
+ catalog = catalog, ident = testIdent, row = InternalRow(2, 200, -1))
+
+ val df2 = spark.table(testTable)
+ val selfJoin = df1.join(df2, df1("id") === df2("id"))
+
+ if (isConnect) {
+ // Connect re-resolves df1 with the new 3-column schema (id, salary,
new_column).
+ assert(selfJoin.columns.length == 6,
+ s"Expected 6 columns (3 + 3) but got: ${selfJoin.columns.mkString(",
")}")
+ checkAnswer(selfJoin,
+ Seq(Row(1, 100, null, 1, 100, null), Row(2, 200, -1, 2, 200, -1)))
+ } else {
+ // Classic: df1 keeps its original 2-column schema (id, salary).
+ assert(selfJoin.columns.length == 5,
+ s"Expected 5 columns (2 + 3) but got: ${selfJoin.columns.mkString(",
")}")
+ checkAnswer(selfJoin,
+ Seq(Row(1, 100, 1, 100, null), Row(2, 200, 2, 200, -1)))
}
}
}
test(s"${testPrefix}SPARK-54157: join after same-session ADD COLUMN" +
" (table with both table and column ID support)") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
-
- val df1 = session.table(testTable)
-
- session.sql(s"ALTER TABLE $testTable ADD COLUMN new_column
INT").collect()
- session.sql(s"INSERT INTO $testTable VALUES (2, 200, -1)").collect()
-
- val df2 = session.table(testTable)
- val selfJoin = df1.join(df2, df1("id") === df2("id"))
-
- if (isConnect) {
- // Connect re-resolves df1 with the new 3-column schema (id, salary,
new_column).
- assert(selfJoin.columns.length == 6,
- s"Expected 6 columns (3 + 3) but got:
${selfJoin.columns.mkString(", ")}")
- checkRows(selfJoin,
- Seq(Row(1, 100, null, 1, 100, null), Row(2, 200, -1, 2, 200, -1)))
- } else {
- // Classic: df1 keeps its original 2-column schema (id, salary).
- assert(selfJoin.columns.length == 5,
- s"Expected 5 columns (2 + 3) but got:
${selfJoin.columns.mkString(", ")}")
- checkRows(selfJoin,
- Seq(Row(1, 100, 1, 100, null), Row(2, 200, 2, 200, -1)))
- }
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+
+ val df1 = spark.table(testTable)
+
+ spark.sql(s"ALTER TABLE $testTable ADD COLUMN new_column INT").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (2, 200, -1)").collect()
+
+ val df2 = spark.table(testTable)
+ val selfJoin = df1.join(df2, df1("id") === df2("id"))
+
+ if (isConnect) {
+ // Connect re-resolves df1 with the new 3-column schema (id, salary,
new_column).
+ assert(selfJoin.columns.length == 6,
+ s"Expected 6 columns (3 + 3) but got: ${selfJoin.columns.mkString(",
")}")
+ checkAnswer(selfJoin,
+ Seq(Row(1, 100, null, 1, 100, null), Row(2, 200, -1, 2, 200, -1)))
+ } else {
+ // Classic: df1 keeps its original 2-column schema (id, salary).
+ assert(selfJoin.columns.length == 5,
+ s"Expected 5 columns (2 + 3) but got: ${selfJoin.columns.mkString(",
")}")
+ checkAnswer(selfJoin,
+ Seq(Row(1, 100, 1, 100, null), Row(2, 200, 2, 200, -1)))
}
}
}
@@ -165,68 +157,64 @@ trait DSv2IncrementallyConstructedQueryTests extends
DSv2ExternalMutationTestBas
test(s"${testPrefix}SPARK-54157: join after external DROP COLUMN" +
" (table with both table and column ID support)") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
-
- val df1 = session.table(testTable)
-
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
- catalog.alterTable(
- testIdent, TableChange.deleteColumn(Array("salary"), false))
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2))
-
- val df2 = session.table(testTable)
-
- if (isConnect) {
- // Connect re-resolves df1 without the dropped column.
- checkRows(
- df1.join(df2, df1("id") === df2("id")),
- Seq(Row(1, 1), Row(2, 2)))
- } else {
- // Classic: df1 references the dropped column.
- checkError(
- exception = intercept[AnalysisException] {
- df1.join(df2, df1("id") === df2("id")).collect()
- },
- condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH",
- matchPVals = true,
- parameters = Map("tableName" -> ".*", "errors" -> "(?s).*"))
- }
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+
+ val df1 = spark.table(testTable)
+
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
+ catalog.alterTable(
+ testIdent, TableChange.deleteColumn(Array("salary"), false))
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2))
+
+ val df2 = spark.table(testTable)
+
+ if (isConnect) {
+ // Connect re-resolves df1 without the dropped column.
+ checkAnswer(
+ df1.join(df2, df1("id") === df2("id")),
+ Seq(Row(1, 1), Row(2, 2)))
+ } else {
+ // Classic: df1 references the dropped column.
+ checkError(
+ exception = intercept[AnalysisException] {
+ df1.join(df2, df1("id") === df2("id")).collect()
+ },
+ condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH",
+ matchPVals = true,
+ parameters = Map("tableName" -> ".*", "errors" -> "(?s).*"))
}
}
}
test(s"${testPrefix}SPARK-54157: join after same-session DROP COLUMN" +
" (table with both table and column ID support)") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
-
- val df1 = session.table(testTable)
-
- session.sql(s"ALTER TABLE $testTable DROP COLUMN salary").collect()
- session.sql(s"INSERT INTO $testTable VALUES (2)").collect()
-
- val df2 = session.table(testTable)
-
- if (isConnect) {
- // Connect re-resolves df1 without the dropped column.
- checkRows(
- df1.join(df2, df1("id") === df2("id")),
- Seq(Row(1, 1), Row(2, 2)))
- } else {
- // Classic: df1 references the dropped column.
- checkError(
- exception = intercept[AnalysisException] {
- df1.join(df2, df1("id") === df2("id")).collect()
- },
- condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH",
- matchPVals = true,
- parameters = Map("tableName" -> ".*", "errors" -> "(?s).*"))
- }
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+
+ val df1 = spark.table(testTable)
+
+ spark.sql(s"ALTER TABLE $testTable DROP COLUMN salary").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (2)").collect()
+
+ val df2 = spark.table(testTable)
+
+ if (isConnect) {
+ // Connect re-resolves df1 without the dropped column.
+ checkAnswer(
+ df1.join(df2, df1("id") === df2("id")),
+ Seq(Row(1, 1), Row(2, 2)))
+ } else {
+ // Classic: df1 references the dropped column.
+ checkError(
+ exception = intercept[AnalysisException] {
+ df1.join(df2, df1("id") === df2("id")).collect()
+ },
+ condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH",
+ matchPVals = true,
+ parameters = Map("tableName" -> ".*", "errors" -> "(?s).*"))
}
}
}
@@ -240,47 +228,45 @@ trait DSv2IncrementallyConstructedQueryTests extends
DSv2ExternalMutationTestBas
test(s"${testPrefix}SPARK-54157: join after external table drop and
recreate" +
" (table with both table and column ID support)") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
-
- val df1 = session.table(testTable)
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
- val originTableId = catalog.loadTable(testIdent).id
-
- catalog.dropTable(testIdent)
- catalog.createTable(
- testIdent,
- new TableInfo.Builder()
- .withColumns(Array(
- Column.create("id", IntegerType),
- Column.create("salary", IntegerType)))
- .build())
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
-
- val df2 = session.table(testTable)
- val newTableId = catalog.loadTable(testIdent).id
- assert(originTableId != newTableId)
-
- if (isConnect) {
- // Connect re-resolves both sides to the recreated table.
- checkRows(
- df1.join(df2, df1("id") === df2("id")),
- Seq(Row(2, 200, 2, 200)))
- } else {
- // Classic: table ID changed.
- checkError(
- exception = intercept[AnalysisException] {
- df1.join(df2, df1("id") === df2("id")).collect()
- },
- condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.TABLE_ID_MISMATCH",
- matchPVals = true,
- parameters = Map(
- "tableName" -> ".*",
- "capturedTableId" -> ".*",
- "currentTableId" -> ".*"))
- }
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+
+ val df1 = spark.table(testTable)
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
+ val originTableId = catalog.loadTable(testIdent).id
+
+ catalog.dropTable(testIdent)
+ catalog.createTable(
+ testIdent,
+ new TableInfo.Builder()
+ .withColumns(Array(
+ Column.create("id", IntegerType),
+ Column.create("salary", IntegerType)))
+ .build())
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
+
+ val df2 = spark.table(testTable)
+ val newTableId = catalog.loadTable(testIdent).id
+ assert(originTableId != newTableId)
+
+ if (isConnect) {
+ // Connect re-resolves both sides to the recreated table.
+ checkAnswer(
+ df1.join(df2, df1("id") === df2("id")),
+ Seq(Row(2, 200, 2, 200)))
+ } else {
+ // Classic: table ID changed.
+ checkError(
+ exception = intercept[AnalysisException] {
+ df1.join(df2, df1("id") === df2("id")).collect()
+ },
+ condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.TABLE_ID_MISMATCH",
+ matchPVals = true,
+ parameters = Map(
+ "tableName" -> ".*",
+ "capturedTableId" -> ".*",
+ "currentTableId" -> ".*"))
}
}
}
@@ -288,43 +274,41 @@ trait DSv2IncrementallyConstructedQueryTests extends
DSv2ExternalMutationTestBas
test(s"${testPrefix}SPARK-54157: join after external drop/recreate" +
" (table without table ID support, but with column ID support)") {
val nullIdT = "nullidcat.ns1.ns2.tbl"
- withTestSession { session =>
- withTestTableAndViews(session, nullIdT) {
- session.sql(s"CREATE TABLE $nullIdT (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $nullIdT VALUES (1, 100)").collect()
-
- val df1 = session.table(nullIdT)
- val catalog = getTableCatalog[TableCatalog](session, "nullidcat")
- assert(catalog.loadTable(testIdent).id == null,
- "NullTableIdInMemoryTableCatalog should produce null table IDs")
-
- catalog.dropTable(testIdent)
- catalog.createTable(
- testIdent,
- new TableInfo.Builder()
- .withColumns(Array(
- Column.create("id", IntegerType),
- Column.create("salary", IntegerType)))
- .build())
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
-
- val df2 = session.table(nullIdT)
-
- if (isConnect) {
- // Connect re-resolves both sides to the recreated table.
- checkRows(
- df1.join(df2, df1("id") === df2("id")),
- Seq(Row(2, 200, 2, 200)))
- } else {
- // Classic: column IDs changed.
- checkError(
- exception = intercept[AnalysisException] {
- df1.join(df2, df1("id") === df2("id")).collect()
- },
- condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH",
- matchPVals = true,
- parameters = Map("tableName" -> ".*", "errors" -> "(?s).*"))
- }
+ withTable(nullIdT) {
+ spark.sql(s"CREATE TABLE $nullIdT (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $nullIdT VALUES (1, 100)").collect()
+
+ val df1 = spark.table(nullIdT)
+ val catalog = getTableCatalog[TableCatalog](spark, "nullidcat")
+ assert(catalog.loadTable(testIdent).id == null,
+ "NullTableIdInMemoryTableCatalog should produce null table IDs")
+
+ catalog.dropTable(testIdent)
+ catalog.createTable(
+ testIdent,
+ new TableInfo.Builder()
+ .withColumns(Array(
+ Column.create("id", IntegerType),
+ Column.create("salary", IntegerType)))
+ .build())
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
+
+ val df2 = spark.table(nullIdT)
+
+ if (isConnect) {
+ // Connect re-resolves both sides to the recreated table.
+ checkAnswer(
+ df1.join(df2, df1("id") === df2("id")),
+ Seq(Row(2, 200, 2, 200)))
+ } else {
+ // Classic: column IDs changed.
+ checkError(
+ exception = intercept[AnalysisException] {
+ df1.join(df2, df1("id") === df2("id")).collect()
+ },
+ condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH",
+ matchPVals = true,
+ parameters = Map("tableName" -> ".*", "errors" -> "(?s).*"))
}
}
}
@@ -332,46 +316,44 @@ trait DSv2IncrementallyConstructedQueryTests extends
DSv2ExternalMutationTestBas
test(s"${testPrefix}SPARK-54157: join does not detect external table drop
and recreate" +
" (table without table ID support and without column ID support)") {
val nullBothT = "nullbothidscat.ns1.ns2.tbl"
- withTestSession { session =>
- withTestTableAndViews(session, nullBothT) {
- session.sql(s"CREATE TABLE $nullBothT (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $nullBothT VALUES (1, 100)").collect()
-
- val df1 = session.table(nullBothT)
- val catalog = getTableCatalog[TableCatalog](
- session, "nullbothidscat")
- assert(catalog.loadTable(testIdent).id == null,
- "NullTableIdAndNullColumnIdInMemoryTableCatalog should produce null
table IDs")
- assert(catalog.loadTable(testIdent).columns().forall(_.id() == null),
- "NullTableIdAndNullColumnIdInMemoryTableCatalog should produce null
column IDs")
-
- catalog.dropTable(testIdent)
- catalog.createTable(
- testIdent,
- new TableInfo.Builder()
- .withColumns(Array(
- Column.create("id", IntegerType),
- Column.create("salary", IntegerType)))
- .build())
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
-
- val df2 = session.table(nullBothT)
-
- if (isConnect) {
- // Connect re-resolves both sides to the recreated table, so the join
- // sees the row appended after recreate.
- checkRows(
- df1.join(df2, df1("id") === df2("id")),
- Seq(Row(2, 200, 2, 200)))
- } else {
- // Classic: neither TABLE_ID_MISMATCH nor COLUMNS_MISMATCH fires, so
the
- // drop and recreate goes undetected. df1 keeps its pre-drop snapshot
- // (1, 100) while df2 reads the recreated table (2, 200), so the
join finds
- // no matching ids and returns no rows.
- checkRows(
- df1.join(df2, df1("id") === df2("id")),
- Seq.empty)
- }
+ withTable(nullBothT) {
+ spark.sql(s"CREATE TABLE $nullBothT (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $nullBothT VALUES (1, 100)").collect()
+
+ val df1 = spark.table(nullBothT)
+ val catalog = getTableCatalog[TableCatalog](
+ spark, "nullbothidscat")
+ assert(catalog.loadTable(testIdent).id == null,
+ "NullTableIdAndNullColumnIdInMemoryTableCatalog should produce null
table IDs")
+ assert(catalog.loadTable(testIdent).columns().forall(_.id() == null),
+ "NullTableIdAndNullColumnIdInMemoryTableCatalog should produce null
column IDs")
+
+ catalog.dropTable(testIdent)
+ catalog.createTable(
+ testIdent,
+ new TableInfo.Builder()
+ .withColumns(Array(
+ Column.create("id", IntegerType),
+ Column.create("salary", IntegerType)))
+ .build())
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
+
+ val df2 = spark.table(nullBothT)
+
+ if (isConnect) {
+ // Connect re-resolves both sides to the recreated table, so the join
+ // sees the row appended after recreate.
+ checkAnswer(
+ df1.join(df2, df1("id") === df2("id")),
+ Seq(Row(2, 200, 2, 200)))
+ } else {
+ // Classic: neither TABLE_ID_MISMATCH nor COLUMNS_MISMATCH fires, so
the
+ // drop and recreate goes undetected. df1 keeps its pre-drop snapshot
+ // (1, 100) while df2 reads the recreated table (2, 200), so the join
finds
+ // no matching ids and returns no rows.
+ checkAnswer(
+ df1.join(df2, df1("id") === df2("id")),
+ Seq.empty)
}
}
}
@@ -385,36 +367,34 @@ trait DSv2IncrementallyConstructedQueryTests extends
DSv2ExternalMutationTestBas
test(s"${testPrefix}SPARK-54157: join after external drop+re-add column" +
" (table without table ID support, but with column ID support)") {
val nullIdT = "nullidcat.ns1.ns2.tbl"
- withTestSession { session =>
- withTestTableAndViews(session, nullIdT) {
- session.sql(s"CREATE TABLE $nullIdT (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $nullIdT VALUES (1, 100)").collect()
-
- val df1 = session.table(nullIdT)
-
- val catalog = getTableCatalog[TableCatalog](session, "nullidcat")
- catalog.alterTable(
- testIdent, TableChange.deleteColumn(Array("salary"), false))
- catalog.alterTable(
- testIdent, TableChange.addColumn(Array("salary"), IntegerType, true))
-
- val df2 = session.table(nullIdT)
-
- if (isConnect) {
- // Connect re-resolves both sides with the new column ID.
- checkRows(
- df1.join(df2, df1("id") === df2("id")),
- Seq(Row(1, null, 1, null)))
- } else {
- // Classic: column ID changed.
- checkError(
- exception = intercept[AnalysisException] {
- df1.join(df2, df1("id") === df2("id")).collect()
- },
- condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH",
- matchPVals = true,
- parameters = Map("tableName" -> ".*", "errors" -> "(?s).*"))
- }
+ withTable(nullIdT) {
+ spark.sql(s"CREATE TABLE $nullIdT (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $nullIdT VALUES (1, 100)").collect()
+
+ val df1 = spark.table(nullIdT)
+
+ val catalog = getTableCatalog[TableCatalog](spark, "nullidcat")
+ catalog.alterTable(
+ testIdent, TableChange.deleteColumn(Array("salary"), false))
+ catalog.alterTable(
+ testIdent, TableChange.addColumn(Array("salary"), IntegerType, true))
+
+ val df2 = spark.table(nullIdT)
+
+ if (isConnect) {
+ // Connect re-resolves both sides with the new column ID.
+ checkAnswer(
+ df1.join(df2, df1("id") === df2("id")),
+ Seq(Row(1, null, 1, null)))
+ } else {
+ // Classic: column ID changed.
+ checkError(
+ exception = intercept[AnalysisException] {
+ df1.join(df2, df1("id") === df2("id")).collect()
+ },
+ condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH",
+ matchPVals = true,
+ parameters = Map("tableName" -> ".*", "errors" -> "(?s).*"))
}
}
}
@@ -422,28 +402,25 @@ trait DSv2IncrementallyConstructedQueryTests extends
DSv2ExternalMutationTestBas
test(s"${testPrefix}SPARK-54157: join does not detect external drop+re-add
column" +
" (table without table ID support and without column ID support)") {
val nullBothT = "nullbothidscat.ns1.ns2.tbl"
- withTestSession { session =>
- withTestTableAndViews(session, nullBothT) {
- session.sql(s"CREATE TABLE $nullBothT (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $nullBothT VALUES (1, 100)").collect()
+ withTable(nullBothT) {
+ spark.sql(s"CREATE TABLE $nullBothT (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $nullBothT VALUES (1, 100)").collect()
- val df1 = session.table(nullBothT)
+ val df1 = spark.table(nullBothT)
- val catalog = getTableCatalog[TableCatalog](
- session, "nullbothidscat")
- catalog.alterTable(
- testIdent, TableChange.deleteColumn(Array("salary"), false))
- catalog.alterTable(
- testIdent, TableChange.addColumn(Array("salary"), IntegerType, true))
+ val catalog = getTableCatalog[TableCatalog](spark, "nullbothidscat")
+ catalog.alterTable(
+ testIdent, TableChange.deleteColumn(Array("salary"), false))
+ catalog.alterTable(
+ testIdent, TableChange.addColumn(Array("salary"), IntegerType, true))
- val df2 = session.table(nullBothT)
+ val df2 = spark.table(nullBothT)
- // Neither TABLE_ID_MISMATCH nor COLUMNS_MISMATCH fires.
- // The change goes undetected and the join succeeds.
- checkRows(
- df1.join(df2, df1("id") === df2("id")),
- Seq(Row(1, null, 1, null)))
- }
+ // Neither TABLE_ID_MISMATCH nor COLUMNS_MISMATCH fires.
+ // The change goes undetected and the join succeeds.
+ checkAnswer(
+ df1.join(df2, df1("id") === df2("id")),
+ Seq(Row(1, null, 1, null)))
}
}
@@ -456,38 +433,36 @@ trait DSv2IncrementallyConstructedQueryTests extends
DSv2ExternalMutationTestBas
test(s"${testPrefix}SPARK-54157: join after external drop+re-add
different-type column" +
" (table with both table and column ID support)") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
-
- val df1 = session.table(testTable)
-
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
- catalog.alterTable(
- testIdent, TableChange.deleteColumn(Array("salary"), false))
- catalog.alterTable(
- testIdent, TableChange.addColumn(Array("salary"), StringType, true))
- externalAppend(catalog = catalog, ident = testIdent,
- row = InternalRow(2, UTF8String.fromString("high")))
-
- val df2 = session.table(testTable)
-
- if (isConnect) {
- // Connect re-resolves both sides with the new column type.
- checkRows(
- df1.join(df2, df1("id") === df2("id")),
- Seq(Row(1, null, 1, null), Row(2, "high", 2, "high")))
- } else {
- // Classic: column ID changed.
- checkError(
- exception = intercept[AnalysisException] {
- df1.join(df2, df1("id") === df2("id")).collect()
- },
- condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH",
- matchPVals = true,
- parameters = Map("tableName" -> ".*", "errors" -> "(?s).*"))
- }
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+
+ val df1 = spark.table(testTable)
+
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
+ catalog.alterTable(
+ testIdent, TableChange.deleteColumn(Array("salary"), false))
+ catalog.alterTable(
+ testIdent, TableChange.addColumn(Array("salary"), StringType, true))
+ externalAppend(catalog = catalog, ident = testIdent,
+ row = InternalRow(2, UTF8String.fromString("high")))
+
+ val df2 = spark.table(testTable)
+
+ if (isConnect) {
+ // Connect re-resolves both sides with the new column type.
+ checkAnswer(
+ df1.join(df2, df1("id") === df2("id")),
+ Seq(Row(1, null, 1, null), Row(2, "high", 2, "high")))
+ } else {
+ // Classic: column ID changed.
+ checkError(
+ exception = intercept[AnalysisException] {
+ df1.join(df2, df1("id") === df2("id")).collect()
+ },
+ condition =
"INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH",
+ matchPVals = true,
+ parameters = Map("tableName" -> ".*", "errors" -> "(?s).*"))
}
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2RepeatedTableAccessTests.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2RepeatedTableAccessTests.scala
index 533d10a94979..fb22a8bb7ab7 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2RepeatedTableAccessTests.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2RepeatedTableAccessTests.scala
@@ -34,7 +34,7 @@ import org.apache.spark.sql.types.IntegerType
* Each scenario includes a session mutation baseline, an external mutation
test, and a
* caching-connector variant showing stale results until `REFRESH TABLE`.
*
- * NOTE: All `session.sql(...)` calls append `.collect()` because Connect
client DataFrames
+ * NOTE: All `spark.sql(...)` calls append `.collect()` because Connect client
DataFrames
* are lazy and require an action to trigger execution. In classic mode
`.collect()` on
* DDL / DML is a no-op (these execute eagerly), so this is harmless.
*/
@@ -45,178 +45,160 @@ trait DSv2RepeatedTableAccessTests extends
DSv2ExternalMutationTestBase {
// Scenario 1: data changes via writes
test(s"${testPrefix}repeated sql() reflects session write") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
- checkRows(session.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100)))
-
- session.sql(s"INSERT INTO $testTable VALUES (2, 200)").collect()
- checkRows(session.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100),
Row(2, 200)))
- }
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+ checkAnswer(spark.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100)))
+
+ spark.sql(s"INSERT INTO $testTable VALUES (2, 200)").collect()
+ checkAnswer(spark.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100),
Row(2, 200)))
}
}
test(s"${testPrefix}repeated sql() reflects external write") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
- checkRows(session.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100)))
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+ checkAnswer(spark.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
- checkRows(session.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100),
Row(2, 200)))
- }
+ checkAnswer(spark.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100),
Row(2, 200)))
}
}
test(s"${testPrefix}connector w/ cache: repeated sql() stale after external
write") {
- withTestSession { session =>
- withTestTableAndViews(session, cachingTestTable) {
- session.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT)
USING foo").collect()
- session.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100)").collect()
- checkRows(session.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100)))
-
- val catalog = getTableCatalog[CachingInMemoryTableCatalog](session,
"cachingcat")
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
-
- // Caching connector returns stale table: external write invisible
- checkRows(session.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100)))
-
- // REFRESH TABLE invalidates the connector cache, external write
becomes visible
- session.sql(s"REFRESH TABLE $cachingTestTable").collect()
- checkRows(session.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100), Row(2, 200)))
- }
+ withTable(cachingTestTable) {
+ spark.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100)").collect()
+ checkAnswer(spark.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100)))
+
+ val catalog = getTableCatalog[CachingInMemoryTableCatalog](spark,
"cachingcat")
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
+
+ // Caching connector returns stale table: external write invisible
+ checkAnswer(spark.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100)))
+
+ // REFRESH TABLE invalidates the connector cache, external write becomes
visible
+ spark.sql(s"REFRESH TABLE $cachingTestTable").collect()
+ checkAnswer(spark.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100), Row(2, 200)))
}
}
// Scenario 2: schema changes
test(s"${testPrefix}repeated sql() reflects session schema change") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
- checkRows(session.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100)))
-
- session.sql(s"ALTER TABLE $testTable ADD COLUMN new_col INT").collect()
- session.sql(s"INSERT INTO $testTable VALUES (2, 200, -1)").collect()
- checkRows(
- session.sql(s"SELECT * FROM $testTable"),
- Seq(Row(1, 100, null), Row(2, 200, -1)))
- }
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+ checkAnswer(spark.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100)))
+
+ spark.sql(s"ALTER TABLE $testTable ADD COLUMN new_col INT").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (2, 200, -1)").collect()
+ checkAnswer(
+ spark.sql(s"SELECT * FROM $testTable"),
+ Seq(Row(1, 100, null), Row(2, 200, -1)))
}
}
test(s"${testPrefix}repeated sql() reflects external schema change") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
- checkRows(session.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100)))
-
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
- val addCol = TableChange.addColumn(Array("new_col"), IntegerType, true)
- catalog.alterTable(testIdent, addCol)
-
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200, -1))
-
- checkRows(
- session.sql(s"SELECT * FROM $testTable"),
- Seq(Row(1, 100, null), Row(2, 200, -1)))
- }
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+ checkAnswer(spark.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100)))
+
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
+ val addCol = TableChange.addColumn(Array("new_col"), IntegerType, true)
+ catalog.alterTable(testIdent, addCol)
+
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200, -1))
+
+ checkAnswer(
+ spark.sql(s"SELECT * FROM $testTable"),
+ Seq(Row(1, 100, null), Row(2, 200, -1)))
}
}
test(s"${testPrefix}connector w/ cache: repeated sql() stale after external
schema change") {
- withTestSession { session =>
- withTestTableAndViews(session, cachingTestTable) {
- session.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT)
USING foo").collect()
- session.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100)").collect()
- checkRows(session.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100)))
-
- val catalog = getTableCatalog[CachingInMemoryTableCatalog](session,
"cachingcat")
- val addCol = TableChange.addColumn(Array("new_col"), IntegerType, true)
- catalog.alterTable(testIdent, addCol)
-
- externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200, -1))
-
- // Caching connector returns stale table: external changes invisible
- checkRows(session.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100)))
-
- // REFRESH TABLE invalidates the connector cache, schema change + data
visible
- session.sql(s"REFRESH TABLE $cachingTestTable").collect()
- checkRows(
- session.sql(s"SELECT * FROM $cachingTestTable"),
- Seq(Row(1, 100, null), Row(2, 200, -1)))
- }
+ withTable(cachingTestTable) {
+ spark.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100)").collect()
+ checkAnswer(spark.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100)))
+
+ val catalog = getTableCatalog[CachingInMemoryTableCatalog](spark,
"cachingcat")
+ val addCol = TableChange.addColumn(Array("new_col"), IntegerType, true)
+ catalog.alterTable(testIdent, addCol)
+
+ externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200, -1))
+
+ // Caching connector returns stale table: external changes invisible
+ checkAnswer(spark.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100)))
+
+ // REFRESH TABLE invalidates the connector cache, schema change + data
visible
+ spark.sql(s"REFRESH TABLE $cachingTestTable").collect()
+ checkAnswer(
+ spark.sql(s"SELECT * FROM $cachingTestTable"),
+ Seq(Row(1, 100, null), Row(2, 200, -1)))
}
}
// Scenario 3: drop and recreate table
test(s"${testPrefix}repeated sql() reflects session drop/recreate") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
- checkRows(session.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100)))
-
- session.sql(s"DROP TABLE $testTable").collect()
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- checkRows(session.sql(s"SELECT * FROM $testTable"), Seq.empty)
- }
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+ checkAnswer(spark.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100)))
+
+ spark.sql(s"DROP TABLE $testTable").collect()
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ checkAnswer(spark.sql(s"SELECT * FROM $testTable"), Seq.empty)
}
}
test(s"${testPrefix}repeated sql() reflects external drop/recreate") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
- checkRows(session.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100)))
-
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
- catalog.dropTable(testIdent)
- catalog.createTable(
- testIdent,
- new TableInfo.Builder()
- .withColumns(Array(
- Column.create("id", IntegerType),
- Column.create("salary", IntegerType)))
- .build())
-
- checkRows(session.sql(s"SELECT * FROM $testTable"), Seq.empty)
- }
+ withTable(testTable) {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100)").collect()
+ checkAnswer(spark.sql(s"SELECT * FROM $testTable"), Seq(Row(1, 100)))
+
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
+ catalog.dropTable(testIdent)
+ catalog.createTable(
+ testIdent,
+ new TableInfo.Builder()
+ .withColumns(Array(
+ Column.create("id", IntegerType),
+ Column.create("salary", IntegerType)))
+ .build())
+
+ checkAnswer(spark.sql(s"SELECT * FROM $testTable"), Seq.empty)
}
}
test(s"${testPrefix}connector w/ cache: repeated sql() stale after external
drop/recreate") {
- withTestSession { session =>
- withTestTableAndViews(session, cachingTestTable) {
- session.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT)
USING foo").collect()
- session.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100)").collect()
- checkRows(session.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100)))
-
- val catalog = getTableCatalog[CachingInMemoryTableCatalog](session,
"cachingcat")
- catalog.dropTable(testIdent)
- catalog.createTable(
- testIdent,
- new TableInfo.Builder()
- .withColumns(Array(
- Column.create("id", IntegerType),
- Column.create("salary", IntegerType)))
- .build())
-
- // Caching connector returns stale table: drop/recreate invisible
- checkRows(session.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100)))
-
- // REFRESH TABLE invalidates the connector cache, new empty table
visible
- session.sql(s"REFRESH TABLE $cachingTestTable").collect()
- checkRows(session.sql(s"SELECT * FROM $cachingTestTable"), Seq.empty)
- }
+ withTable(cachingTestTable) {
+ spark.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100)").collect()
+ checkAnswer(spark.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100)))
+
+ val catalog = getTableCatalog[CachingInMemoryTableCatalog](spark,
"cachingcat")
+ catalog.dropTable(testIdent)
+ catalog.createTable(
+ testIdent,
+ new TableInfo.Builder()
+ .withColumns(Array(
+ Column.create("id", IntegerType),
+ Column.create("salary", IntegerType)))
+ .build())
+
+ // Caching connector returns stale table: drop/recreate invisible
+ checkAnswer(spark.sql(s"SELECT * FROM $cachingTestTable"), Seq(Row(1,
100)))
+
+ // REFRESH TABLE invalidates the connector cache, new empty table visible
+ spark.sql(s"REFRESH TABLE $cachingTestTable").collect()
+ checkAnswer(spark.sql(s"SELECT * FROM $cachingTestTable"), Seq.empty)
}
}
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2TempViewWithStoredPlanTests.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2TempViewWithStoredPlanTests.scala
index 9f8a93e30550..e473968794c3 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2TempViewWithStoredPlanTests.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2TempViewWithStoredPlanTests.scala
@@ -27,7 +27,7 @@ import org.apache.spark.sql.types.{IntegerType, LongType,
StringType}
* backed by DSv2 tables correctly handle data changes, schema changes, and
table recreation,
* both via session SQL and external catalog mutations.
*
- * NOTE: All `session.sql(...)` calls append `.collect()` because Connect
client DataFrames
+ * NOTE: All `spark.sql(...)` calls append `.collect()` because Connect client
DataFrames
* are lazy and require an action to trigger execution. In classic mode
`.collect()` on DDL
* is a no-op (DDL executes eagerly), so this is harmless.
*/
@@ -35,143 +35,143 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
// Scenario 1.1 (session write)
test(s"${testPrefix}temp view with stored plan reflects session write") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(testTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- session.sql(s"INSERT INTO $testTable VALUES (2, 200)").collect()
- checkRows(session.table("v"), Seq(Row(1, 100), Row(2, 200)))
+ spark.sql(s"INSERT INTO $testTable VALUES (2, 200)").collect()
+ checkAnswer(spark.table("v"), Seq(Row(1, 100), Row(2, 200)))
}
}
}
// Scenario 1.2 (external write)
test(s"${testPrefix}temp view with stored plan reflects external write") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(testTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
- checkRows(session.table("v"), Seq(Row(1, 100), Row(2, 200)))
+ checkAnswer(spark.table("v"), Seq(Row(1, 100), Row(2, 200)))
}
}
}
// Scenario 1.2 connector w/ cache (external write, caching connector)
test(s"${testPrefix}connector w/ cache: temp view stale after external
write") {
- withTestSession { session =>
- withTestTableAndViews(session, cachingTestTable, Seq("v")) {
- session.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT)
USING foo").collect()
- session.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(cachingTestTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
- session.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[CachingInMemoryTableCatalog](session,
"cachingcat")
+ val catalog = getTableCatalog[CachingInMemoryTableCatalog](spark,
"cachingcat")
externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200))
// Caching connector returns stale table: external write invisible
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
// REFRESH TABLE invalidates the connector cache, external write
becomes visible
- session.sql(s"REFRESH TABLE $cachingTestTable").collect()
- checkRows(session.table("v"), Seq(Row(1, 100), Row(2, 200)))
+ spark.sql(s"REFRESH TABLE $cachingTestTable").collect()
+ checkAnswer(spark.table("v"), Seq(Row(1, 100), Row(2, 200)))
}
}
}
// Scenario 2.1 (session ADD COLUMN)
test(s"${testPrefix}temp view with stored plan preserves schema after
session ADD COLUMN") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(testTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- session.sql(s"ALTER TABLE $testTable ADD COLUMN new_column
INT").collect()
- session.sql(s"INSERT INTO $testTable VALUES (2, 200, -1)").collect()
+ spark.sql(s"ALTER TABLE $testTable ADD COLUMN new_column
INT").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (2, 200, -1)").collect()
// view preserves original 2-column schema, filter still applied
- checkRows(session.table("v"), Seq(Row(1, 100), Row(2, 200)))
+ checkAnswer(spark.table("v"), Seq(Row(1, 100), Row(2, 200)))
}
}
}
// Scenario 2.2 (external ADD COLUMN)
test(s"${testPrefix}temp view with stored plan preserves schema after
external ADD COLUMN") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(testTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
// external schema change via catalog API
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
val addCol = TableChange.addColumn(Array("new_column"), IntegerType,
true)
catalog.alterTable(testIdent, addCol)
externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200, -1))
// view preserves original 2-column schema, filter still applied
- checkRows(session.table("v"), Seq(Row(1, 100), Row(2, 200)))
+ checkAnswer(spark.table("v"), Seq(Row(1, 100), Row(2, 200)))
}
}
}
// Scenario 2.2 connector w/ cache (external ADD COLUMN, caching connector)
test(s"${testPrefix}connector w/ cache: temp view stale after external ADD
COLUMN") {
- withTestSession { session =>
- withTestTableAndViews(session, cachingTestTable, Seq("v")) {
- session.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT)
USING foo").collect()
- session.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(cachingTestTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
- session.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[CachingInMemoryTableCatalog](session,
"cachingcat")
+ val catalog = getTableCatalog[CachingInMemoryTableCatalog](spark,
"cachingcat")
val addCol = TableChange.addColumn(Array("new_column"), IntegerType,
true)
catalog.alterTable(testIdent, addCol)
externalAppend(catalog = catalog, ident = testIdent, row =
InternalRow(2, 200, -1))
// Caching connector returns stale table: external changes invisible
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
// REFRESH TABLE invalidates the connector cache, view preserves
original 2-column schema
- session.sql(s"REFRESH TABLE $cachingTestTable").collect()
- checkRows(session.table("v"), Seq(Row(1, 100), Row(2, 200)))
+ spark.sql(s"REFRESH TABLE $cachingTestTable").collect()
+ checkAnswer(spark.table("v"), Seq(Row(1, 100), Row(2, 200)))
}
}
}
// Scenario 3.1 (session column removal)
test(s"${testPrefix}temp view with stored plan detects session column
removal") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(testTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- session.sql(s"ALTER TABLE $testTable DROP COLUMN salary").collect()
+ spark.sql(s"ALTER TABLE $testTable DROP COLUMN salary").collect()
checkError(
- exception = intercept[AnalysisException] {
session.table("v").collect() },
+ exception = intercept[AnalysisException] {
spark.table("v").collect() },
condition =
"INCOMPATIBLE_COLUMN_CHANGES_AFTER_VIEW_WITH_PLAN_CREATION",
parameters = Map(
"viewName" -> "`v`",
@@ -184,20 +184,20 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
// Scenario 3.2 (external column removal)
test(s"${testPrefix}temp view with stored plan detects external column
removal") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(testTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
val dropCol = TableChange.deleteColumn(Array("salary"), false)
catalog.alterTable(testIdent, dropCol)
checkError(
- exception = intercept[AnalysisException] {
session.table("v").collect() },
+ exception = intercept[AnalysisException] {
spark.table("v").collect() },
condition =
"INCOMPATIBLE_COLUMN_CHANGES_AFTER_VIEW_WITH_PLAN_CREATION",
parameters = Map(
"viewName" -> "`v`",
@@ -210,25 +210,25 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
// Scenario 3.2 connector w/ cache (external column removal, caching
connector)
test(s"${testPrefix}connector w/ cache: temp view stale after external
column removal") {
- withTestSession { session =>
- withTestTableAndViews(session, cachingTestTable, Seq("v")) {
- session.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT)
USING foo").collect()
- session.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(cachingTestTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
- session.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[CachingInMemoryTableCatalog](session,
"cachingcat")
+ val catalog = getTableCatalog[CachingInMemoryTableCatalog](spark,
"cachingcat")
val dropCol = TableChange.deleteColumn(Array("salary"), false)
catalog.alterTable(testIdent, dropCol)
// Caching connector returns stale table: column removal invisible, no
error
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
// REFRESH TABLE invalidates the connector cache, column removal
detected
- session.sql(s"REFRESH TABLE $cachingTestTable").collect()
+ spark.sql(s"REFRESH TABLE $cachingTestTable").collect()
checkError(
- exception = intercept[AnalysisException] {
session.table("v").collect() },
+ exception = intercept[AnalysisException] {
spark.table("v").collect() },
condition =
"INCOMPATIBLE_COLUMN_CHANGES_AFTER_VIEW_WITH_PLAN_CREATION",
parameters = Map(
"viewName" -> "`v`",
@@ -241,43 +241,43 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
// Scenario 4.1 (session drop and recreate table)
test(s"${testPrefix}temp view with stored plan resolves to session-recreated
table") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(testTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
val originalTableId = catalog.loadTable(testIdent).id
- session.sql(s"DROP TABLE $testTable").collect()
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"DROP TABLE $testTable").collect()
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
val newTableId = catalog.loadTable(testIdent).id
assert(originalTableId != newTableId)
// view resolves to the new empty table
- checkRows(session.table("v"), Seq.empty)
+ checkAnswer(spark.table("v"), Seq.empty)
- session.sql(s"INSERT INTO $testTable VALUES (2, 200)").collect()
- checkRows(session.table("v"), Seq(Row(2, 200)))
+ spark.sql(s"INSERT INTO $testTable VALUES (2, 200)").collect()
+ checkAnswer(spark.table("v"), Seq(Row(2, 200)))
}
}
}
// Scenario 4.2 (external drop and recreate table)
test(s"${testPrefix}temp view with stored plan resolves to externally
recreated table") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(testTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
val originalTableId = catalog.loadTable(testIdent).id
catalog.dropTable(testIdent)
@@ -293,25 +293,25 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
assert(originalTableId != newTableId)
// view resolves to the new empty table
- checkRows(session.table("v"), Seq.empty)
+ checkAnswer(spark.table("v"), Seq.empty)
- session.sql(s"INSERT INTO $testTable VALUES (2, 200)").collect()
- checkRows(session.table("v"), Seq(Row(2, 200)))
+ spark.sql(s"INSERT INTO $testTable VALUES (2, 200)").collect()
+ checkAnswer(spark.table("v"), Seq(Row(2, 200)))
}
}
}
// Scenario 4.2 connector w/ cache (external drop/recreate, caching
connector)
test(s"${testPrefix}connector w/ cache: temp view stale after external
drop/recreate") {
- withTestSession { session =>
- withTestTableAndViews(session, cachingTestTable, Seq("v")) {
- session.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT)
USING foo").collect()
- session.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(cachingTestTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
- session.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[CachingInMemoryTableCatalog](session,
"cachingcat")
+ val catalog = getTableCatalog[CachingInMemoryTableCatalog](spark,
"cachingcat")
catalog.dropTable(testIdent)
catalog.createTable(
testIdent,
@@ -322,11 +322,11 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
.build())
// Caching connector returns stale table: drop/recreate invisible
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
// REFRESH TABLE invalidates the connector cache, view resolves to new
empty table
- session.sql(s"REFRESH TABLE $cachingTestTable").collect()
- checkRows(session.table("v"), Seq.empty)
+ spark.sql(s"REFRESH TABLE $cachingTestTable").collect()
+ checkAnswer(spark.table("v"), Seq.empty)
}
}
}
@@ -334,29 +334,29 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
// Scenario 5.1 (session drop and re-add column with same type, multiple
views)
test(s"${testPrefix}temp view with stored plan after session drop and re-add
column same type" +
" with unfiltered view") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v", "v_no_filter",
"v_filter_is_null")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
-
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- session.table(testTable).createOrReplaceTempView("v_no_filter")
- session.table(testTable).filter("salary IS NULL")
+ withTable(testTable) {
+ withView("v", "v_no_filter", "v_filter_is_null") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ spark.table(testTable).createOrReplaceTempView("v_no_filter")
+ spark.table(testTable).filter("salary IS NULL")
.createOrReplaceTempView("v_filter_is_null")
- checkRows(session.table("v"), Seq(Row(1, 100)))
- checkRows(session.table("v_no_filter"), Seq(Row(1, 100), Row(10,
1000)))
- checkRows(session.table("v_filter_is_null"), Seq.empty)
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
+ checkAnswer(spark.table("v_no_filter"), Seq(Row(1, 100), Row(10,
1000)))
+ checkAnswer(spark.table("v_filter_is_null"), Seq.empty)
// drop and re-add column with same name and type
- session.sql(s"ALTER TABLE $testTable DROP COLUMN salary").collect()
- session.sql(s"ALTER TABLE $testTable ADD COLUMN salary INT").collect()
+ spark.sql(s"ALTER TABLE $testTable DROP COLUMN salary").collect()
+ spark.sql(s"ALTER TABLE $testTable ADD COLUMN salary INT").collect()
// salary values are now null, so the filtered view returns nothing
- checkRows(session.table("v"), Seq.empty)
+ checkAnswer(spark.table("v"), Seq.empty)
// unfiltered view returns rows with null salary
- checkRows(session.table("v_no_filter"), Seq(Row(1, null), Row(10,
null)))
+ checkAnswer(spark.table("v_no_filter"), Seq(Row(1, null), Row(10,
null)))
// IS NULL filter now matches all rows
- checkRows(session.table("v_filter_is_null"), Seq(Row(1, null), Row(10,
null)))
+ checkAnswer(spark.table("v_filter_is_null"), Seq(Row(1, null), Row(10,
null)))
}
}
}
@@ -364,31 +364,31 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
// Scenario 5.2 (external drop and re-add column with same type)
test(s"${testPrefix}temp view with stored plan after external drop and
re-add column " +
"same type") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v", "v_no_filter",
"v_filter_is_null")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
-
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- session.table(testTable).createOrReplaceTempView("v_no_filter")
- session.table(testTable).filter("salary IS NULL")
+ withTable(testTable) {
+ withView("v", "v_no_filter", "v_filter_is_null") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ spark.table(testTable).createOrReplaceTempView("v_no_filter")
+ spark.table(testTable).filter("salary IS NULL")
.createOrReplaceTempView("v_filter_is_null")
- checkRows(session.table("v"), Seq(Row(1, 100)))
- checkRows(session.table("v_no_filter"), Seq(Row(1, 100), Row(10,
1000)))
- checkRows(session.table("v_filter_is_null"), Seq.empty)
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
+ checkAnswer(spark.table("v_no_filter"), Seq(Row(1, 100), Row(10,
1000)))
+ checkAnswer(spark.table("v_filter_is_null"), Seq.empty)
// external drop and re-add column via catalog API
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
val dropCol = TableChange.deleteColumn(Array("salary"), false)
val addCol = TableChange.addColumn(Array("salary"), IntegerType, true)
catalog.alterTable(testIdent, dropCol, addCol)
// salary values are now null, so the filtered view returns nothing
- checkRows(session.table("v"), Seq.empty)
+ checkAnswer(spark.table("v"), Seq.empty)
// unfiltered view returns rows with null salary
- checkRows(session.table("v_no_filter"), Seq(Row(1, null), Row(10,
null)))
+ checkAnswer(spark.table("v_no_filter"), Seq(Row(1, null), Row(10,
null)))
// IS NULL filter now matches all rows
- checkRows(session.table("v_filter_is_null"), Seq(Row(1, null), Row(10,
null)))
+ checkAnswer(spark.table("v_filter_is_null"), Seq(Row(1, null), Row(10,
null)))
}
}
}
@@ -396,44 +396,44 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
// Scenario 5.2 connector w/ cache (external drop/re-add column, caching
connector)
test(s"${testPrefix}connector w/ cache: temp view stale after external
drop/re-add column " +
"same type") {
- withTestSession { session =>
- withTestTableAndViews(session, cachingTestTable, Seq("v")) {
- session.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT)
USING foo").collect()
- session.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(cachingTestTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
- session.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[CachingInMemoryTableCatalog](session,
"cachingcat")
+ val catalog = getTableCatalog[CachingInMemoryTableCatalog](spark,
"cachingcat")
val dropCol = TableChange.deleteColumn(Array("salary"), false)
val addCol = TableChange.addColumn(Array("salary"), IntegerType, true)
catalog.alterTable(testIdent, dropCol, addCol)
// Caching connector returns stale table: column drop/re-add invisible
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
// REFRESH TABLE invalidates the connector cache, salary values are
null
- session.sql(s"REFRESH TABLE $cachingTestTable").collect()
- checkRows(session.table("v"), Seq.empty)
+ spark.sql(s"REFRESH TABLE $cachingTestTable").collect()
+ checkAnswer(spark.table("v"), Seq.empty)
}
}
}
// Scenario 6.1 (session drop and re-add column with different type)
test(s"${testPrefix}temp view with stored plan detects session column type
change") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(testTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- session.sql(s"ALTER TABLE $testTable DROP COLUMN salary").collect()
- session.sql(s"ALTER TABLE $testTable ADD COLUMN salary
STRING").collect()
+ spark.sql(s"ALTER TABLE $testTable DROP COLUMN salary").collect()
+ spark.sql(s"ALTER TABLE $testTable ADD COLUMN salary STRING").collect()
checkError(
- exception = intercept[AnalysisException] {
session.table("v").collect() },
+ exception = intercept[AnalysisException] {
spark.table("v").collect() },
condition =
"INCOMPATIBLE_COLUMN_CHANGES_AFTER_VIEW_WITH_PLAN_CREATION",
parameters = Map(
"viewName" -> "`v`",
@@ -446,21 +446,21 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
// Scenario 6.2 (external drop and re-add column with different type)
test(s"${testPrefix}temp view with stored plan detects external column type
change") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(testTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
val dropCol = TableChange.deleteColumn(Array("salary"), false)
val addCol = TableChange.addColumn(Array("salary"), StringType, true)
catalog.alterTable(testIdent, dropCol, addCol)
checkError(
- exception = intercept[AnalysisException] {
session.table("v").collect() },
+ exception = intercept[AnalysisException] {
spark.table("v").collect() },
condition =
"INCOMPATIBLE_COLUMN_CHANGES_AFTER_VIEW_WITH_PLAN_CREATION",
parameters = Map(
"viewName" -> "`v`",
@@ -473,26 +473,26 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
// Scenario 6.2 connector w/ cache (external column type change, caching
connector)
test(s"${testPrefix}connector w/ cache: temp view stale after external
column type change") {
- withTestSession { session =>
- withTestTableAndViews(session, cachingTestTable, Seq("v")) {
- session.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT)
USING foo").collect()
- session.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(cachingTestTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
- session.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[CachingInMemoryTableCatalog](session,
"cachingcat")
+ val catalog = getTableCatalog[CachingInMemoryTableCatalog](spark,
"cachingcat")
val dropCol = TableChange.deleteColumn(Array("salary"), false)
val addCol = TableChange.addColumn(Array("salary"), StringType, true)
catalog.alterTable(testIdent, dropCol, addCol)
// Caching connector returns stale table: type change invisible, no
error
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
// REFRESH TABLE invalidates the connector cache, type change detected
- session.sql(s"REFRESH TABLE $cachingTestTable").collect()
+ spark.sql(s"REFRESH TABLE $cachingTestTable").collect()
checkError(
- exception = intercept[AnalysisException] {
session.table("v").collect() },
+ exception = intercept[AnalysisException] {
spark.table("v").collect() },
condition =
"INCOMPATIBLE_COLUMN_CHANGES_AFTER_VIEW_WITH_PLAN_CREATION",
parameters = Map(
"viewName" -> "`v`",
@@ -505,18 +505,18 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
// Scenario 7.1 (session type widening from INT to BIGINT)
test(s"${testPrefix}temp view with stored plan detects session type
widening") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(testTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- session.sql(s"ALTER TABLE $testTable ALTER COLUMN salary TYPE
LONG").collect()
+ spark.sql(s"ALTER TABLE $testTable ALTER COLUMN salary TYPE
LONG").collect()
checkError(
- exception = intercept[AnalysisException] {
session.table("v").collect() },
+ exception = intercept[AnalysisException] {
spark.table("v").collect() },
condition =
"INCOMPATIBLE_COLUMN_CHANGES_AFTER_VIEW_WITH_PLAN_CREATION",
parameters = Map(
"viewName" -> "`v`",
@@ -529,20 +529,20 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
// Scenario 7.2 (external type widening from INT to BIGINT)
test(s"${testPrefix}temp view with stored plan detects external type
widening") {
- withTestSession { session =>
- withTestTableAndViews(session, testTable, Seq("v")) {
- session.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
- session.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(testTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $testTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $testTable VALUES (1, 100), (10,
1000)").collect()
- session.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(testTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat")
+ val catalog = getTableCatalog[InMemoryTableCatalog](spark, "testcat")
val updateType = TableChange.updateColumnType(Array("salary"),
LongType)
catalog.alterTable(testIdent, updateType)
checkError(
- exception = intercept[AnalysisException] {
session.table("v").collect() },
+ exception = intercept[AnalysisException] {
spark.table("v").collect() },
condition =
"INCOMPATIBLE_COLUMN_CHANGES_AFTER_VIEW_WITH_PLAN_CREATION",
parameters = Map(
"viewName" -> "`v`",
@@ -555,25 +555,25 @@ trait DSv2TempViewWithStoredPlanTests extends
DSv2ExternalMutationTestBase {
// Scenario 7.2 connector w/ cache (external type widening, caching
connector)
test(s"${testPrefix}connector w/ cache: temp view stale after external type
widening") {
- withTestSession { session =>
- withTestTableAndViews(session, cachingTestTable, Seq("v")) {
- session.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT)
USING foo").collect()
- session.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
+ withTable(cachingTestTable) {
+ withView("v") {
+ spark.sql(s"CREATE TABLE $cachingTestTable (id INT, salary INT) USING
foo").collect()
+ spark.sql(s"INSERT INTO $cachingTestTable VALUES (1, 100), (10,
1000)").collect()
- session.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ spark.table(cachingTestTable).filter("salary <
999").createOrReplaceTempView("v")
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
- val catalog = getTableCatalog[CachingInMemoryTableCatalog](session,
"cachingcat")
+ val catalog = getTableCatalog[CachingInMemoryTableCatalog](spark,
"cachingcat")
val updateType = TableChange.updateColumnType(Array("salary"),
LongType)
catalog.alterTable(testIdent, updateType)
// Caching connector returns stale table: type change invisible, no
error
- checkRows(session.table("v"), Seq(Row(1, 100)))
+ checkAnswer(spark.table("v"), Seq(Row(1, 100)))
// REFRESH TABLE invalidates the connector cache, type change detected
- session.sql(s"REFRESH TABLE $cachingTestTable").collect()
+ spark.sql(s"REFRESH TABLE $cachingTestTable").collect()
checkError(
- exception = intercept[AnalysisException] {
session.table("v").collect() },
+ exception = intercept[AnalysisException] {
spark.table("v").collect() },
condition =
"INCOMPATIBLE_COLUMN_CHANGES_AFTER_VIEW_WITH_PLAN_CREATION",
parameters = Map(
"viewName" -> "`v`",
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala
index 2e2ad7361f5f..c18b8ef09e96 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2DataFrameSuite.scala
@@ -24,7 +24,7 @@ import scala.jdk.CollectionConverters._
import scala.reflect.ClassTag
import org.apache.spark.{SparkConf, SparkException}
-import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SaveMode,
SparkSession}
+import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SaveMode,
SessionQueryTest, SparkSession}
import org.apache.spark.sql.QueryTest.withQueryExecutionsCaptured
import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
import org.apache.spark.sql.catalyst.plans.logical.{AppendData,
CreateTableAsSelect, LogicalPlan, ReplaceTableAsSelect}
@@ -47,6 +47,7 @@ import org.apache.spark.unsafe.types.UTF8String
class DataSourceV2DataFrameSuite
extends InsertIntoTests(supportsDynamicOverwrite = true, includeSQLOnlyTests
= false)
+ with SessionQueryTest
with DSv2TempViewWithStoredPlanTests
with DSv2RepeatedTableAccessTests
with DSv2IncrementallyConstructedQueryTests
@@ -94,12 +95,6 @@ class DataSourceV2DataFrameSuite
// DSv2ExternalMutationTestBase implementations for classic mode
override protected def testPrefix: String = ""
- override protected def isConnect: Boolean = false
-
- 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,
@@ -112,16 +107,6 @@ class DataSourceV2DataFrameSuite
c.asInstanceOf[C]
}
- override protected def withTestTableAndViews(
- session: SparkSession,
- table: String,
- views: Seq[String] = Seq.empty)(fn: => Unit): Unit = {
- withTable(table) {
- try { fn }
- finally { views.foreach(v => session.sql(s"DROP VIEW IF EXISTS $v")) }
- }
- }
-
override def verifyTable(tableName: String, expected: DataFrame): Unit = {
checkAnswer(spark.table(tableName), expected)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
