This is an automated email from the ASF dual-hosted git repository.
szehon-ho 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 f8ba02f9f7aa [SPARK-57625][SDP] Manage Auxiliary Table Lifecycle in
DatasetManager 1/2
f8ba02f9f7aa is described below
commit f8ba02f9f7aaf31e3a2bb620ab2f351d3e6dd9b3
Author: Anish Mahto <[email protected]>
AuthorDate: Wed Jul 1 11:58:32 2026 -0700
[SPARK-57625][SDP] Manage Auxiliary Table Lifecycle in DatasetManager 1/2
### What changes were proposed in this pull request?
Step 1/2 for managing auxiliary table lifecycle in `DatasetManager`. Pure
refactor that extracts catalog table creation and evolution logic into their
own helpers so that auxiliary tables can reuse them.
### Why are the changes needed?
Refactor before implementing auxiliary table full refresh, creation, and
schema evolution in `DatasetManager`.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing tests
### Was this patch authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.8.
Closes #56684 from
AnishMahto/SPARK-57625-extract-table-creation-and-evolution.
Authored-by: Anish Mahto <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
(cherry picked from commit bdf6d814ee0930ab7b20c13c84d27d44d8251c4e)
Signed-off-by: Szehon Ho <[email protected]>
---
.../spark/sql/pipelines/graph/DatasetManager.scala | 146 +++++++++++++++----
.../pipelines/graph/MaterializeTablesSuite.scala | 161 ++++++++++++++++++++-
2 files changed, 273 insertions(+), 34 deletions(-)
diff --git
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala
index 456edca8d1e2..3c813dd997de 100644
---
a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala
+++
b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala
@@ -29,16 +29,18 @@ import org.apache.spark.sql.classic.SparkSession
import org.apache.spark.sql.connector.catalog.{
CatalogV2Util,
Identifier,
+ Table => V2Table,
TableCatalog,
TableChange,
TableInfo
}
import
org.apache.spark.sql.connector.catalog.CatalogV2Util.v2ColumnsToStructType
-import org.apache.spark.sql.connector.expressions.{ClusterByTransform,
Expressions}
+import org.apache.spark.sql.connector.expressions.{ClusterByTransform,
Expressions, Transform}
import org.apache.spark.sql.execution.command.CreateViewCommand
import org.apache.spark.sql.pipelines.graph.QueryOrigin.ExceptionHelpers
import org.apache.spark.sql.pipelines.util.SchemaInferenceUtils.diffSchemas
import org.apache.spark.sql.pipelines.util.SchemaMergingUtils
+import org.apache.spark.sql.types.StructType
/**
* `DatasetManager` is responsible for materializing tables in the catalog
based on the given
@@ -278,11 +280,7 @@ object DatasetManager extends Logging {
val allTransforms = partitioning ++ clustering
- val existingTableOpt = if (catalog.tableExists(identifier)) {
- Some(catalog.loadTable(identifier))
- } else {
- None
- }
+ val existingTableOpt = loadTableIfExists(catalog, identifier)
// Error if partitioning/clustering doesn't match
existingTableOpt.foreach { existingTable =>
@@ -298,8 +296,14 @@ object DatasetManager extends Logging {
}
}
+ // A streaming table on a non-full-refresh run is maintained
incrementally: its existing data is
+ // preserved and its schema is merged with (not replaced by) the schema
computed in this run.
+ // Every other case (materialized views, and any full refresh) is
recomputed from scratch:
+ // existing data is wiped and the schema is taken directly from this run's
computed schema.
+ val isTableIncrementallyUpdated = table.isStreamingTable && !isFullRefresh
+
// Wipe the data if we need to
- if ((isFullRefresh || !table.isStreamingTable) &&
existingTableOpt.isDefined) {
+ if (existingTableOpt.isDefined && !isTableIncrementallyUpdated) {
context.spark.sql(s"TRUNCATE TABLE ${table.identifier.quotedString}")
}
@@ -317,31 +321,25 @@ object DatasetManager extends Logging {
context.spark.sql(s"DROP TABLE IF EXISTS
${auxiliaryTableId.quotedString}")
}
- // Alter the table if we need to
- existingTableOpt.foreach { existingTable =>
- val existingSchema = v2ColumnsToStructType(existingTable.columns())
-
- val targetSchema = if (table.isStreamingTable && !isFullRefresh) {
- SchemaMergingUtils.mergeSchemas(existingSchema, outputSchema)
- } else {
- outputSchema
- }
-
- val columnChanges = diffSchemas(existingSchema, targetSchema)
- val setProperties = mergedProperties.map { case (k, v) =>
TableChange.setProperty(k, v) }
- catalog.alterTable(identifier, (columnChanges ++ setProperties).toArray:
_*)
- }
-
- // Create the table if we need to
- if (existingTableOpt.isEmpty) {
- catalog.createTable(
- identifier,
- new TableInfo.Builder()
- .withProperties(mergedProperties.asJava)
- .withColumns(CatalogV2Util.structTypeToV2Columns(outputSchema))
- .withPartitions(allTransforms.toArray)
- .build()
- )
+ // Create the table if absent, otherwise evolve it (schema + properties).
+ existingTableOpt match {
+ case Some(existingTable) =>
+ evolveTable(
+ catalog = catalog,
+ tableIdentifier = identifier,
+ existingTable = existingTable,
+ desiredSchema = outputSchema,
+ properties = mergedProperties,
+ mergeWithExistingSchema = isTableIncrementallyUpdated
+ )
+ case None =>
+ createTable(
+ catalog = catalog,
+ tableIdentifier = identifier,
+ schema = outputSchema,
+ properties = mergedProperties,
+ transforms = allTransforms
+ )
}
table.copy(
@@ -351,6 +349,90 @@ object DatasetManager extends Logging {
)
}
+ /** Loads the table at `identifier` from `catalog`, or `None` if it does not
exist. */
+ private def loadTableIfExists(
+ catalog: TableCatalog,
+ identifier: Identifier): Option[V2Table] = {
+ Option.when(catalog.tableExists(identifier))(catalog.loadTable(identifier))
+ }
+
+ /**
+ * Creates the table at `identifier` with the given schema, properties, and
partition/cluster
+ * transforms. Used when no table yet exists at the identifier.
+ *
+ * @param schema the schema to create the table with.
+ * @param properties the table properties to create the table with.
+ * @param transforms the partition/cluster transforms to create the table
with.
+ */
+ private def createTable(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ schema: StructType,
+ properties: Map[String, String],
+ transforms: Seq[Transform]): Unit = {
+ catalog.createTable(
+ tableIdentifier,
+ new TableInfo.Builder()
+ .withProperties(properties.asJava)
+ .withColumns(CatalogV2Util.structTypeToV2Columns(schema))
+ .withPartitions(transforms.toArray)
+ .build()
+ )
+ }
+
+ /**
+ * Evolves the already-existing `existingTable` at `identifier` in place by
diffing its schema
+ * and properties, skipping the catalog `alterTable` entirely when nothing
actually changes.
+ * Partitioning/clustering cannot change in place, so no transforms are
accepted here.
+ *
+ * @param existingTable the currently materialized table.
+ * @param desiredSchema the schema the table should have as
computed in the current
+ * execution (the user-specified or inferred
schema). This is the
+ * "incoming" side and may differ from
`existingTable`'s recorded
+ * schema due to schema evolution across runs.
+ * @param properties the declared table properties to (re)set
on the table. Note
+ * that properties absent here are NOT
removed from the table (see
+ * the TODO in the body).
+ * @param mergeWithExistingSchema whether the effective schema is the merge
of the existing and
+ * desired schemas (additive evolution)
rather than the desired
+ * schema as-is.
+ */
+ private def evolveTable(
+ catalog: TableCatalog,
+ tableIdentifier: Identifier,
+ existingTable: V2Table,
+ desiredSchema: StructType,
+ properties: Map[String, String],
+ mergeWithExistingSchema: Boolean): Unit = {
+ val currentSchema = v2ColumnsToStructType(existingTable.columns())
+ val targetSchema = if (mergeWithExistingSchema) {
+ SchemaMergingUtils.mergeSchemas(currentSchema, desiredSchema)
+ } else {
+ desiredSchema
+ }
+ val columnChanges = diffSchemas(currentSchema, targetSchema)
+
+ val existingProperties = existingTable.properties()
+
+ // TODO (SPARK-57670): Property removal is intentionally not handled here:
a property dropped
+ // from the table definition between runs is left in place rather than
actually removed from the
+ // corresponding catalog table entry. Removing it reliably is hard because
we cannot distinguish
+ // a user-declared property the user dropped from a catalog/engine-managed
property (e.g. the
+ // non-reserved `clusteringColumns`, or arbitrary catalog-internal keys)
that must never be
+ // removed, and there is no record of which keys the pipeline previously
set.
+ val propertiesToSet = properties.collect {
+ case (k, v) if !Option(existingProperties.get(k)).contains(v) =>
+ TableChange.setProperty(k, v)
+ }
+
+ val allTableChanges = columnChanges ++ propertiesToSet
+
+ // If there are no table changes to evolve with, avoid the no-op
round-trip alter altogether.
+ if (allTableChanges.nonEmpty) {
+ catalog.alterTable(tableIdentifier, allTableChanges.toArray: _*)
+ }
+ }
+
/**
* Some fields on the [[Table]] object are represented as reserved table
properties by the catalog
* APIs. This method creates a table properties map that merges the
user-provided table properties
diff --git
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala
index 29d85e9b4439..46e2d6d9ae63 100644
---
a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala
+++
b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/graph/MaterializeTablesSuite.scala
@@ -17,11 +17,19 @@
package org.apache.spark.sql.pipelines.graph
+import scala.collection.mutable
import scala.jdk.CollectionConverters._
import org.apache.spark.SparkThrowable
-import org.apache.spark.sql.{AnalysisException, SQLContext}
-import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Identifier,
TableCatalog}
+import org.apache.spark.sql.{AnalysisException, Row, SQLContext}
+import org.apache.spark.sql.connector.catalog.{
+ CatalogV2Util,
+ Identifier,
+ InMemoryTableCatalog,
+ Table => V2Table,
+ TableCatalog,
+ TableChange
+}
import org.apache.spark.sql.connector.expressions.{ClusterByTransform,
Expressions, FieldReference}
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
import
org.apache.spark.sql.pipelines.graph.DatasetManager.TableMaterializationException
@@ -1083,4 +1091,153 @@ abstract class MaterializeTablesSuite extends
BaseCoreExecutionTest {
}
assert(ex.cause.isInstanceOf[AnalysisException])
}
+
+ // =============== Table evolution in catalog tests ===============
+
+ private val recordingCatalogName = "recording_cat"
+ private val recordingNamespace = "rec_ns"
+
+ /**
+ * Registers [[RecordingInMemoryTableCatalog]] under `recordingCatalogName`,
creates
+ * `recordingNamespace`, runs `body`, then tears the registration back down.
+ */
+ private def withRecordingCatalog(body: => Unit): Unit = {
+ spark.conf.set(
+ s"spark.sql.catalog.$recordingCatalogName",
+ classOf[RecordingInMemoryTableCatalog].getName
+ )
+ try {
+ spark.sql(s"CREATE NAMESPACE IF NOT EXISTS
$recordingCatalogName.$recordingNamespace")
+ body
+ } finally {
+ spark.sessionState.catalogManager.reset()
+
spark.sessionState.conf.unsetConf(s"spark.sql.catalog.$recordingCatalogName")
+ }
+ }
+
+ /**
+ * Materializes a single streaming table under the recording
catalog/namespace with the given
+ * schema and properties.
+ */
+ private def materializeStreamingTable(
+ name: String,
+ schema: StructType,
+ properties: Map[String, String]): Unit = {
+ // All nulls dummy row, compatible with any schema type
+ val row = Row.fromSeq(Seq.fill(schema.length)(null))
+ val df = spark.createDataFrame(spark.sparkContext.parallelize(Seq(row)),
schema)
+ materializeGraph(
+ new TestGraphRegistrationContext(spark) {
+ registerTable(
+ name,
+ query = Option(dfFlowFunc(df)),
+ specifiedSchema = Option(schema),
+ properties = properties,
+ catalog = Option(recordingCatalogName),
+ database = Option(recordingNamespace)
+ )
+ }.resolveToDataflowGraph(),
+ storageRoot = storageRoot
+ )
+ }
+
+ private def recordingCatalog: RecordingInMemoryTableCatalog =
+ spark.sessionState.catalogManager
+ .catalog(recordingCatalogName)
+ .asInstanceOf[RecordingInMemoryTableCatalog]
+
+ private def loadTableFromRecordingCatalog(name: String): V2Table = {
+ val catalog = spark.sessionState.catalogManager
+ .catalog(recordingCatalogName)
+ .asInstanceOf[TableCatalog]
+ catalog.loadTable(Identifier.of(Array(recordingNamespace), name))
+ }
+
+ test("re-materializing an unchanged table does not issue an alterTable") {
+ withRecordingCatalog {
+ val schema = new StructType().add("id", IntegerType).add("value",
StringType)
+ val props = Map("p.a" -> "1", "p.b" -> "2")
+ // Creating the table issues no alter, and re-materializing the
unchanged table is a no-op,
+ // so no alter is ever recorded.
+ materializeStreamingTable("t", schema, props)
+ assert(recordingCatalog.recordedAlters.isEmpty)
+ materializeStreamingTable("t", schema, props)
+ assert(recordingCatalog.recordedAlters.isEmpty)
+ }
+ }
+
+ test("re-materializing with changed/new properties issues an alterTable that
sets them") {
+ withRecordingCatalog {
+ val schema = new StructType().add("id", IntegerType).add("value",
StringType)
+ // Creating the table issues no alter; re-materializing with
changed/added properties issues
+ // exactly one alter that sets them.
+ materializeStreamingTable("t", schema, Map("p.a" -> "1"))
+ assert(recordingCatalog.recordedAlters.isEmpty)
+ materializeStreamingTable("t", schema, Map("p.a" -> "2", "p.new" -> "n"))
+ assert(recordingCatalog.recordedAlters.size == 1)
+
+ val changes = recordingCatalog.recordedAlters.flatten
+ assert(changes.forall(_.isInstanceOf[TableChange.SetProperty]))
+ val set = changes.collect {
+ case s: TableChange.SetProperty => s.property() -> s.value()
+ }.toMap
+ assert(set == Map("p.a" -> "2", "p.new" -> "n"))
+
+ val table = loadTableFromRecordingCatalog("t")
+ assert(table.properties().get("p.a") == "2")
+ assert(table.properties().get("p.new") == "n")
+ }
+ }
+
+ test("re-materializing with an added column issues an alterTable") {
+ withRecordingCatalog {
+ // Creating the table issues no alter; re-materializing with an added
column issues exactly
+ // one alter that adds it.
+ materializeStreamingTable("t", new StructType().add("id", IntegerType),
Map("p.a" -> "1"))
+ assert(recordingCatalog.recordedAlters.isEmpty)
+ materializeStreamingTable(
+ "t",
+ new StructType().add("id", IntegerType).add("value", StringType),
+ Map("p.a" -> "1")
+ )
+ assert(recordingCatalog.recordedAlters.size == 1)
+
+ val changes = recordingCatalog.recordedAlters.flatten
+ assert(changes.exists(_.isInstanceOf[TableChange.AddColumn]))
+
+ assert(
+ loadTableFromRecordingCatalog("t").columns() sameElements
+ CatalogV2Util.structTypeToV2Columns(
+ new StructType().add("id", IntegerType).add("value", StringType)
+ )
+ )
+ }
+ }
+
+ test("re-materializing with a dropped property neither removes it nor issues
an alterTable") {
+ withRecordingCatalog {
+ val schema = new StructType().add("id", IntegerType)
+ // This test locks in the current buggy behavior where dropped
properties do not materialize
+ // against the catalog table entity. See SPARK-57670.
+ materializeStreamingTable("t", schema, Map("p.keep" -> "v", "p.stale" ->
"old"))
+ assert(recordingCatalog.recordedAlters.isEmpty)
+ materializeStreamingTable("t", schema, Map("p.keep" -> "v"))
+ assert(recordingCatalog.recordedAlters.isEmpty)
+
+ assert(loadTableFromRecordingCatalog("t").properties().get("p.stale") ==
"old")
+ }
+ }
+}
+
+/**
+ * An [[InMemoryTableCatalog]] that records every `alterTable` invocation
while still applying it,
+ * so tests can assert whether materialization issued an alter or skipped it
as a no-op.
+ */
+class RecordingInMemoryTableCatalog extends InMemoryTableCatalog {
+ val recordedAlters: mutable.ArrayBuffer[Seq[TableChange]] =
mutable.ArrayBuffer.empty
+
+ override def alterTable(ident: Identifier, changes: TableChange*): V2Table =
{
+ recordedAlters += changes.toSeq
+ super.alterTable(ident, changes: _*)
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
