cloud-fan commented on code in PR #56708:
URL: https://github.com/apache/spark/pull/56708#discussion_r3579949890
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2IncrementallyConstructedQueryTests.scala:
##########
@@ -45,44 +45,40 @@ 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) {
Review Comment:
Nit: all 12 tests in this file are over-indented by 2 spaces. Collapsing
`withTestSession { withTestTableAndViews { ... } }` (two levels) into
`withTable { ... }` (one level) removed a nesting level, but the bodies weren't
dedented to match: `withTable` sits at 6 spaces here (should be 4, one level
under `test(...) {`), bodies at 8, and each test's closing brace at 4 instead
of 2. The sibling files `DSv2RepeatedTableAccessTests` and
`DSv2TempViewWithStoredPlanTests` were dedented correctly. Note scalafmt only
runs on `sql/api` and `sql/connect` (`dev/lint-scala`), not `sql/core`, so CI
won't catch this.
##########
sql/connect/server/src/test/scala/org/apache/spark/sql/connect/DataSourceV2DataFrameConnectSuite.scala:
##########
@@ -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. Mirror the classic suite's `after` block: clear the
caching connector's cache
+ // and reset the catalog manager so each test starts from a clean catalog
state. Without this,
Review Comment:
Claim accuracy: the comment says it clears the cache "and reset the catalog
manager" and "Mirror[s] the classic suite's `after` block", but the code below
only calls `clearCache()` — it never calls `catalogManager.reset()`, whereas
the classic `after` block it references calls both. The reset looks genuinely
unnecessary here (tests use fully-qualified names, no current-namespace
mutation), so I'd fix the comment rather than add a `reset()` call:
```suggestion
// 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,
```
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2CacheTableReadTests.scala:
##########
@@ -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")
Review Comment:
Nit: broken continuation indentation in this last test. The
`getTableCatalog[...]` RHS here, and the `createTable(` args below
(`testIdent`, `new TableInfo.Builder()`), sit at the same column as the line
they continue instead of being indented. The rest of the file dedented cleanly
— looks like an isolated slip during the refactor. (Same
scalafmt-doesn't-cover-`sql/core` caveat as the other file.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
