cloud-fan commented on code in PR #55536: URL: https://github.com/apache/spark/pull/55536#discussion_r3310685628
########## sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2CacheTableReadTests.scala: ########## @@ -0,0 +1,186 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.connector + +import org.apache.spark.sql.{Row, SparkSession} +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.connector.catalog.{Column, InMemoryTableCatalog, TableChange, TableInfo} +import org.apache.spark.sql.types.IntegerType + +/** + * Shared CACHE TABLE impact on reads tests for DSv2 tables. Write operations ignore the + * cache, so these tests verify how reads behave when a cached table is mutated by session + * SQL or external catalog API calls: + * + * - Scenario 1 (external write): cache pins the read, external write invisible until REFRESH. + * - Scenario 2 (session write and more external changes): session write rebuilds cache, + * subsequent external write invisible until REFRESH. + * - Scenario 3 (external schema changes): cache pinned at original schema until REFRESH. + * - Scenario 4 (session schema changes and more external changes): session ALTER rebuilds + * cache, subsequent external write invisible until REFRESH. + * - Scenario 5 (external drop and recreate table): query sees new empty table. + * + * NOTE: All `session.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)) + + 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() + + session.table(testTable).cache() + assertTableCached(session, testTable) + checkRows(session.table(testTable), Seq(Row(1, 100))) + + val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat") + externalAppend(catalog = catalog, ident = testIdent, row = InternalRow(2, 200)) + + 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), 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() + + session.table(testTable).cache() + assertTableCached(session, testTable) + checkRows(session.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))) + + val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat") + externalAppend(catalog = catalog, ident = testIdent, row = InternalRow(3, 300)) + + assertTableCached(session, testTable) + checkRows(session.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))) + } + } + } + + 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))) + } + } + } + + 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() + + session.table(testTable).cache() + assertTableCached(session, testTable) + checkRows(session.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))) + + val catalog = getTableCatalog[InMemoryTableCatalog](session, "testcat") + externalAppend(catalog = catalog, ident = testIdent, row = InternalRow(2, 200, -1)) + + assertTableCached(session, testTable) + checkRows(session.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))) + } + } + } + + 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) Review Comment: Scenarios 1–4 explicitly assert `assertTableCached(session, testTable)` after the external mutation to lock down "cache still pinned." Scenario 5 only verifies row values are empty — the cache-state outcome is left implicit. Worth adding `assert(!session.catalog.isCached(testTable))` between the empty-rows check and the `REFRESH TABLE` to document the asymmetric behavior: external drop+recreate makes the prior cache unreachable via subsequent name lookups (the new relation has a different table identity, so the cache lookup misses), unlike external write / external schema change where the cache stays pinned. As a side benefit, this also makes the trailing `REFRESH TABLE` a meaningful verification step (otherwise it's a no-op on an already-empty, uncached table). ########## sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2CacheTableReadTests.scala: ########## @@ -0,0 +1,186 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.connector + +import org.apache.spark.sql.{Row, SparkSession} +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.connector.catalog.{Column, InMemoryTableCatalog, TableChange, TableInfo} +import org.apache.spark.sql.types.IntegerType + +/** + * Shared CACHE TABLE impact on reads tests for DSv2 tables. Write operations ignore the + * cache, so these tests verify how reads behave when a cached table is mutated by session + * SQL or external catalog API calls: + * + * - Scenario 1 (external write): cache pins the read, external write invisible until REFRESH. + * - Scenario 2 (session write and more external changes): session write rebuilds cache, Review Comment: Minor wording nit — each test has exactly one external mutation, but scenarios 2, 3, and 4 are described with plurals ("and more external changes" / "external schema changes" / "session schema changes and more external changes"). Suggest singular forms to match the test bodies: ```suggestion * - Scenario 1 (external write): cache pins the read, external write invisible until REFRESH. * - Scenario 2 (session write then external write): session write rebuilds cache, * subsequent external write invisible until REFRESH. * - Scenario 3 (external schema change): cache pinned at original schema until REFRESH. * - Scenario 4 (session schema change then external write): session ALTER rebuilds * cache, subsequent external write invisible until REFRESH. ``` ########## sql/core/src/test/scala/org/apache/spark/sql/connector/DSv2CacheTableReadTests.scala: ########## @@ -0,0 +1,186 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.connector + +import org.apache.spark.sql.{Row, SparkSession} +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.connector.catalog.{Column, InMemoryTableCatalog, TableChange, TableInfo} +import org.apache.spark.sql.types.IntegerType + +/** + * Shared CACHE TABLE impact on reads tests for DSv2 tables. Write operations ignore the + * cache, so these tests verify how reads behave when a cached table is mutated by session + * SQL or external catalog API calls: + * + * - Scenario 1 (external write): cache pins the read, external write invisible until REFRESH. + * - Scenario 2 (session write and more external changes): session write rebuilds cache, + * subsequent external write invisible until REFRESH. + * - Scenario 3 (external schema changes): cache pinned at original schema until REFRESH. + * - Scenario 4 (session schema changes and more external changes): session ALTER rebuilds + * cache, subsequent external write invisible until REFRESH. + * - Scenario 5 (external drop and recreate table): query sees new empty table. Review Comment: Question on coverage scope: the peer traits `DSv2RepeatedTableAccessTests` and `DSv2TempViewWithStoredPlanTests` each include a `cachingcat` variant per scenario (combining the Spark `CacheManager` with the caching connector) and, for drop/recreate, a session counterpart alongside the external one. This trait omits both. If that's a deliberate scope choice — e.g., "CACHE TABLE behavior is orthogonal to connector caching, and session DROP+CREATE is already covered by the temp-view suite" — fine to leave as-is, but a one-line note in this docstring explaining the omission would save future readers from cross-referencing the peer traits. If not deliberate, a `cachingcat` variant of scenario 1 is the highest-value add, since both caches go stale until `REFRESH TABLE` invalidates them at two different layers. -- 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]
