You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2025/11/30 18:49:52 UTC
(spark) branch master updated: [SPARK-54424][SQL] Failures during recaching must not fail operations
This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 7833e2fccf9f [SPARK-54424][SQL] Failures during recaching must not fail operations
7833e2fccf9f is described below
commit 7833e2fccf9faf24175f60db6e6187edd2e1be9f
Author: Anton Okolnychyi <ao...@apache.org>
AuthorDate: Sun Nov 30 10:49:40 2025 -0800
[SPARK-54424][SQL] Failures during recaching must not fail operations
### What changes were proposed in this pull request?
This PR prevents failures during recaching failing write/refresh operations.
### Why are the changes needed?
After recent changes in SPARK-54387, we may now mark write operations as failed even though they successfully committed to the table but the cache refresh was unsuccessful.
### Does this PR introduce _any_ user-facing change?
Yes, `recacheByXXX` will no longer throw an exception if recaching fails.
### How was this patch tested?
This PR comes with tests.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #53143 from aokolnychyi/spark-54424.
Authored-by: Anton Okolnychyi <ao...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../apache/spark/sql/execution/CacheManager.scala | 73 +++++++++--
.../spark/sql/execution/QueryExecution.scala | 23 +++-
.../datasources/v2/V2TableRefreshUtil.scala | 22 ----
.../datasources/v2/WriteToDataSourceV2Exec.scala | 14 +-
.../sql/connector/DataSourceV2DataFrameSuite.scala | 146 +++++++++++++++++++++
5 files changed, 233 insertions(+), 45 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
index 5a38751b61e1..34e47084f656 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/CacheManager.scala
@@ -17,6 +17,8 @@
package org.apache.spark.sql.execution
+import scala.util.control.NonFatal
+
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.internal.{Logging, MessageWithContext}
@@ -374,25 +376,68 @@ class CacheManager extends Logging with AdaptiveSparkPlanHelper {
}
needToRecache.foreach { cd =>
cd.cachedRepresentation.cacheBuilder.clearCache()
- val sessionWithConfigsOff = getOrCloneSessionWithConfigsOff(spark)
- val (newKey, newCache) = sessionWithConfigsOff.withActive {
- val refreshedPlan = V2TableRefreshUtil.refresh(sessionWithConfigsOff, cd.plan)
- val qe = sessionWithConfigsOff.sessionState.executePlan(refreshedPlan)
- qe.normalized -> InMemoryRelation(cd.cachedRepresentation.cacheBuilder, qe)
- }
- val recomputedPlan = cd.copy(plan = newKey, cachedRepresentation = newCache)
- this.synchronized {
- if (lookupCachedDataInternal(recomputedPlan.plan).nonEmpty) {
- logWarning("While recaching, data was already added to cache.")
- } else {
- cachedData = recomputedPlan +: cachedData
- CacheManager.logCacheOperation(log"Re-cached Dataframe cache entry:" +
- log"${MDC(DATAFRAME_CACHE_ENTRY, recomputedPlan)}")
+ tryRebuildCacheEntry(spark, cd).foreach { entry =>
+ this.synchronized {
+ if (lookupCachedDataInternal(entry.plan).nonEmpty) {
+ logWarning("While recaching, data was already added to cache.")
+ } else {
+ cachedData = entry +: cachedData
+ CacheManager.logCacheOperation(log"Re-cached Dataframe cache entry:" +
+ log"${MDC(DATAFRAME_CACHE_ENTRY, entry)}")
+ }
}
}
}
}
+ private def tryRebuildCacheEntry(spark: SparkSession, cd: CachedData): Option[CachedData] = {
+ val sessionWithConfigsOff = getOrCloneSessionWithConfigsOff(spark)
+ sessionWithConfigsOff.withActive {
+ tryRefreshPlan(sessionWithConfigsOff, cd.plan).map { refreshedPlan =>
+ val qe = QueryExecution.create(
+ sessionWithConfigsOff,
+ refreshedPlan,
+ refreshPhaseEnabled = false)
+ val newKey = qe.normalized
+ val newCache = InMemoryRelation(cd.cachedRepresentation.cacheBuilder, qe)
+ cd.copy(plan = newKey, cachedRepresentation = newCache)
+ }
+ }
+ }
+
+ /**
+ * Attempts to refresh table metadata loaded through the catalog.
+ *
+ * If the table state is cached (e.g., via `CACHE TABLE t`), the relation is replaced with
+ * updated metadata as long as the table ID still matches, ensuring that all schema changes
+ * are reflected. Otherwise, a new plan is produced using refreshed table metadata but
+ * retaining the original schema, provided the schema changes are still compatible with the
+ * query (e.g., adding new columns should be acceptable).
+ *
+ * Note this logic applies only to V2 tables at the moment.
+ *
+ * @return the refreshed plan if refresh succeeds, None otherwise
+ */
+ private def tryRefreshPlan(spark: SparkSession, plan: LogicalPlan): Option[LogicalPlan] = {
+ try {
+ EliminateSubqueryAliases(plan) match {
+ case r @ ExtractV2CatalogAndIdentifier(catalog, ident) if r.timeTravelSpec.isEmpty =>
+ val table = catalog.loadTable(ident)
+ if (r.table.id == table.id) {
+ Some(DataSourceV2Relation.create(table, Some(catalog), Some(ident)))
+ } else {
+ None
+ }
+ case _ =>
+ Some(V2TableRefreshUtil.refresh(spark, plan))
+ }
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to refresh plan while attempting to recache", e)
+ None
+ }
+ }
+
private[sql] def lookupCachedTable(
name: Seq[String],
resolver: Resolver): Option[LogicalPlan] = {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
index 26d2078791aa..3e0aef962e71 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala
@@ -66,7 +66,8 @@ class QueryExecution(
val logical: LogicalPlan,
val tracker: QueryPlanningTracker = new QueryPlanningTracker,
val mode: CommandExecutionMode.Value = CommandExecutionMode.ALL,
- val shuffleCleanupMode: ShuffleCleanupMode = DoNotCleanup) extends Logging {
+ val shuffleCleanupMode: ShuffleCleanupMode = DoNotCleanup,
+ val refreshPhaseEnabled: Boolean = true) extends Logging {
val id: Long = QueryExecution.nextExecutionId
@@ -178,7 +179,7 @@ class QueryExecution(
// for eagerly executed commands we mark this place as beginning of execution.
tracker.setReadyForExecution()
val qe = new QueryExecution(sparkSession, p, mode = mode,
- shuffleCleanupMode = shuffleCleanupMode)
+ shuffleCleanupMode = shuffleCleanupMode, refreshPhaseEnabled = refreshPhaseEnabled)
val result = QueryExecution.withInternalError(s"Eagerly executed $name failed.") {
SQLExecution.withNewExecutionId(qe, Some(name)) {
qe.executedPlan.executeCollect()
@@ -207,7 +208,11 @@ class QueryExecution(
// there may be delay between analysis and subsequent phases
// therefore, refresh captured table versions to reflect latest data
private val lazyTableVersionsRefreshed = LazyTry {
- V2TableRefreshUtil.refresh(sparkSession, commandExecuted, versionedOnly = true)
+ if (refreshPhaseEnabled) {
+ V2TableRefreshUtil.refresh(sparkSession, commandExecuted, versionedOnly = true)
+ } else {
+ commandExecuted
+ }
}
private[sql] def tableVersionsRefreshed: LogicalPlan = lazyTableVersionsRefreshed.get
@@ -569,6 +574,18 @@ object QueryExecution {
private def nextExecutionId: Long = _nextExecutionId.getAndIncrement
+ private[execution] def create(
+ sparkSession: SparkSession,
+ logical: LogicalPlan,
+ refreshPhaseEnabled: Boolean = true): QueryExecution = {
+ new QueryExecution(
+ sparkSession,
+ logical,
+ mode = CommandExecutionMode.ALL,
+ shuffleCleanupMode = determineShuffleCleanupMode(sparkSession.sessionState.conf),
+ refreshPhaseEnabled = refreshPhaseEnabled)
+ }
+
/**
* Construct a sequence of rules that are used to prepare a planned [[SparkPlan]] for execution.
* These rules will make sure subqueries are planned, make sure the data partitioning and ordering
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala
index 945ab122d54e..151329de9e6f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2TableRefreshUtil.scala
@@ -21,7 +21,6 @@ import scala.collection.mutable
import org.apache.spark.internal.Logging
import org.apache.spark.sql.catalyst.SQLConfHelper
-import org.apache.spark.sql.catalyst.analysis.AsOfVersion
import org.apache.spark.sql.catalyst.plans.logical.{Command, LogicalPlan}
import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.connector.catalog.{Identifier, Table, TableCatalog, V2TableUtil}
@@ -32,27 +31,6 @@ import org.apache.spark.sql.util.SchemaValidationMode.ALLOW_NEW_FIELDS
import org.apache.spark.sql.util.SchemaValidationMode.PROHIBIT_CHANGES
private[sql] object V2TableRefreshUtil extends SQLConfHelper with Logging {
- /**
- * Pins table versions for all versioned tables in the plan.
- *
- * This method captures the current version of each versioned table by adding time travel
- * specifications. Tables that already have time travel specifications or are not versioned
- * are left unchanged.
- *
- * @param plan the logical plan to pin versions for
- * @return plan with pinned table versions
- */
- def pinVersions(plan: LogicalPlan): LogicalPlan = {
- plan transform {
- case r @ ExtractV2CatalogAndIdentifier(catalog, ident)
- if r.isVersioned && r.timeTravelSpec.isEmpty =>
- val tableName = V2TableUtil.toQualifiedName(catalog, ident)
- val version = r.table.version
- logDebug(s"Pinning table version for $tableName to $version")
- r.copy(timeTravelSpec = Some(AsOfVersion(version)))
- }
- }
-
/**
* Refreshes table metadata for tables in the plan.
*
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
index 6c574be91ebf..75915d97ba4b 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala
@@ -33,7 +33,7 @@ import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.connector.metric.CustomMetric
import org.apache.spark.sql.connector.write.{BatchWrite, DataWriter, DataWriterFactory, DeltaWrite, DeltaWriter, MergeSummaryImpl, PhysicalWriteInfoImpl, RowLevelOperationTable, Write, WriterCommitMessage, WriteSummary}
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryExecutionErrors}
-import org.apache.spark.sql.execution.{SparkPlan, SQLExecution, UnaryExecNode}
+import org.apache.spark.sql.execution.{QueryExecution, SparkPlan, SQLExecution, UnaryExecNode}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.execution.joins.BaseJoinExec
import org.apache.spark.sql.execution.metric.{CustomMetrics, SQLMetric, SQLMetrics}
@@ -177,7 +177,6 @@ case class ReplaceTableAsSelectExec(
query,
versionedOnly = true,
schemaValidationMode = PROHIBIT_CHANGES)
- val pinnedQuery = V2TableRefreshUtil.pinVersions(refreshedQuery)
if (catalog.tableExists(ident)) {
invalidateCache(catalog, ident)
catalog.dropTable(ident)
@@ -185,13 +184,15 @@ case class ReplaceTableAsSelectExec(
throw QueryCompilationErrors.cannotReplaceMissingTableError(ident)
}
val tableInfo = new TableInfo.Builder()
- .withColumns(getV2Columns(pinnedQuery.schema, catalog.useNullableQuerySchema))
+ .withColumns(getV2Columns(refreshedQuery.schema, catalog.useNullableQuerySchema))
.withPartitions(partitioning.toArray)
.withProperties(properties.asJava)
.build()
val table = Option(catalog.createTable(ident, tableInfo))
.getOrElse(catalog.loadTable(ident, Set(TableWritePrivilege.INSERT).asJava))
- writeToTable(catalog, table, writeOptions, ident, pinnedQuery, overwrite = true)
+ writeToTable(
+ catalog, table, writeOptions, ident, refreshedQuery,
+ overwrite = true, refreshPhaseEnabled = false)
}
}
@@ -764,7 +765,8 @@ private[v2] trait V2CreateTableAsSelectBaseExec extends LeafV2CommandExec {
writeOptions: Map[String, String],
ident: Identifier,
query: LogicalPlan,
- overwrite: Boolean): Seq[InternalRow] = {
+ overwrite: Boolean,
+ refreshPhaseEnabled: Boolean = true): Seq[InternalRow] = {
Utils.tryWithSafeFinallyAndFailureCallbacks({
val relation = DataSourceV2Relation.create(table, Some(catalog), Some(ident))
val writeCommand = if (overwrite) {
@@ -772,7 +774,7 @@ private[v2] trait V2CreateTableAsSelectBaseExec extends LeafV2CommandExec {
} else {
AppendData.byPosition(relation, query, writeOptions)
}
- val qe = session.sessionState.executePlan(writeCommand)
+ val qe = QueryExecution.create(session, writeCommand, refreshPhaseEnabled)
qe.assertCommandExecuted()
DataSourceV2Utils.commitStagedChanges(sparkContext, table, metrics)
Nil
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 5f4aee277831..cdb498c8f2bf 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
@@ -30,6 +30,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{AppendData, CreateTableAsSel
import org.apache.spark.sql.connector.catalog.{Column, ColumnDefaultValue, DefaultValue, Identifier, InMemoryTableCatalog, SupportsV1OverwriteWithSaveAsTable, TableInfo}
import org.apache.spark.sql.connector.catalog.BasicInMemoryTableCatalog
import org.apache.spark.sql.connector.catalog.TableChange.{AddColumn, UpdateColumnDefaultValue}
+import org.apache.spark.sql.connector.catalog.TableChange
import org.apache.spark.sql.connector.catalog.TableWritePrivilege
import org.apache.spark.sql.connector.catalog.TruncatableTable
import org.apache.spark.sql.connector.expressions.{ApplyTransform, GeneralScalarExpression, LiteralValue, Transform}
@@ -1894,6 +1895,151 @@ class DataSourceV2DataFrameSuite
}
}
+ test("SPARK-54424: refresh table cache on schema changes (column removed)") {
+ val t = "testcat.ns1.ns2.tbl"
+ val ident = Identifier.of(Array("ns1", "ns2"), "tbl")
+ withTable(t) {
+ sql(s"CREATE TABLE $t (id INT, value INT, category STRING) USING foo")
+ sql(s"INSERT INTO $t VALUES (1, 10, 'A'), (2, 20, 'B'), (3, 30, 'A')")
+
+ // cache table
+ spark.table(t).cache()
+
+ // verify caching works as expected
+ assertCached(spark.table(t))
+ checkAnswer(
+ spark.table(t),
+ Seq(Row(1, 10, "A"), Row(2, 20, "B"), Row(3, 30, "A")))
+
+ // evolve table directly to mimic external changes
+ // these external changes make cached plan invalid (column is no longer there)
+ val change = TableChange.deleteColumn(Array("category"), false)
+ catalog("testcat").alterTable(ident, change)
+
+ // refresh table is supposed to trigger recaching
+ spark.sql(s"REFRESH TABLE $t")
+
+ // recaching is expected to succeed
+ assert(spark.sharedState.cacheManager.numCachedEntries == 1)
+
+ // verify cache reflects latest schema and data
+ assertCached(spark.table(t))
+ checkAnswer(spark.table(t), Seq(Row(1, 10), Row(2, 20), Row(3, 30)))
+ }
+ }
+
+ test("SPARK-54424: refresh table cache on schema changes (column added)") {
+ val t = "testcat.ns1.ns2.tbl"
+ val ident = Identifier.of(Array("ns1", "ns2"), "tbl")
+ withTable(t) {
+ sql(s"CREATE TABLE $t (id INT, value INT) USING foo")
+ sql(s"INSERT INTO $t VALUES (1, 10), (2, 20), (3, 30)")
+
+ // cache table
+ spark.table(t).cache()
+
+ // verify caching works as expected
+ assertCached(spark.table(t))
+ checkAnswer(
+ spark.table(t),
+ Seq(Row(1, 10), Row(2, 20), Row(3, 30)))
+
+ // evolve table directly to mimic external changes
+ // these external changes make cached plan invalid (table state has changed)
+ val change = TableChange.addColumn(Array("category"), StringType, true)
+ catalog("testcat").alterTable(ident, change)
+
+ // refresh table is supposed to trigger recaching
+ spark.sql(s"REFRESH TABLE $t")
+
+ // recaching is expected to succeed
+ assert(spark.sharedState.cacheManager.numCachedEntries == 1)
+
+ // verify cache reflects latest schema and data
+ assertCached(spark.table(t))
+ checkAnswer(spark.table(t), Seq(Row(1, 10, null), Row(2, 20, null), Row(3, 30, null)))
+ }
+ }
+
+ test("SPARK-54424: successfully refresh cache with compatible schema changes") {
+ val t = "testcat.ns1.ns2.tbl"
+ val ident = Identifier.of(Array("ns1", "ns2"), "tbl")
+ withTable(t) {
+ sql(s"CREATE TABLE $t (id INT, value INT) USING foo")
+ sql(s"INSERT INTO $t VALUES (1, 10), (2, 20), (3, 30)")
+
+ // cache query
+ val df = spark.table(t).filter("id < 100")
+ df.cache()
+
+ // verify caching works as expected
+ assertCached(spark.table(t).filter("id < 100"))
+ checkAnswer(
+ spark.table(t).filter("id < 100"),
+ Seq(Row(1, 10), Row(2, 20), Row(3, 30)))
+
+ // evolve table directly to mimic external changes
+ // adding columns should be OK
+ val change = TableChange.addColumn(Array("category"), StringType, true)
+ catalog("testcat").alterTable(ident, change)
+
+ // refresh table is supposed to trigger recaching
+ spark.sql(s"REFRESH TABLE $t")
+
+ // recaching is expected to succeed
+ assert(spark.sharedState.cacheManager.numCachedEntries == 1)
+
+ // verify derived queries still benefit from refreshed cache
+ assertCached(df.filter("id > 0"))
+ checkAnswer(df.filter("id > 0"), Seq(Row(1, 10), Row(2, 20), Row(3, 30)))
+
+ // add more data
+ sql(s"INSERT INTO $t VALUES (4, 40, '40')")
+
+ // verify derived queries still benefit from refreshed cache
+ assertCached(df.filter("id > 0"))
+ checkAnswer(df.filter("id > 0"), Seq(Row(1, 10), Row(2, 20), Row(3, 30), Row(4, 40)))
+
+ // verify latest schema is propagated (new column has NULL values for existing rows)
+ checkAnswer(
+ spark.table(t),
+ Seq(Row(1, 10, null), Row(2, 20, null), Row(3, 30, null), Row(4, 40, "40")))
+ }
+ }
+
+ test("SPARK-54424: inability to refresh cache shouldn't fail operations") {
+ val t = "testcat.ns1.ns2.tbl"
+ val ident = Identifier.of(Array("ns1", "ns2"), "tbl")
+ withTable(t) {
+ sql(s"CREATE TABLE $t (id INT, value INT) USING foo")
+ sql(s"INSERT INTO $t VALUES (1, 10), (2, 20), (3, 30)")
+
+ // cache query
+ val df = spark.table(t).filter("id < 100")
+ df.cache()
+
+ // verify caching works as expected
+ assertCached(spark.table(t).filter("id < 100"))
+ checkAnswer(
+ spark.table(t).filter("id < 100"),
+ Seq(Row(1, 10), Row(2, 20), Row(3, 30)))
+
+ // evolve table directly to mimic external changes
+ // removing columns should be make cached plan invalid
+ val change = TableChange.deleteColumn(Array("value"), false)
+ catalog("testcat").alterTable(ident, change)
+
+ // refresh table is supposed to trigger recaching
+ spark.sql(s"REFRESH TABLE $t")
+
+ // recaching is expected to fail
+ assert(spark.sharedState.cacheManager.isEmpty)
+
+ // verify latest schema is propagated
+ checkAnswer(spark.table(t), Seq(Row(1), Row(2), Row(3)))
+ }
+ }
+
private def pinTable(catalogName: String, ident: Identifier, version: String): Unit = {
catalog(catalogName) match {
case inMemory: BasicInMemoryTableCatalog => inMemory.pinTable(ident, version)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org