[SPARK-57625][SDP] Manage Auxiliary Table Lifecycle in DatasetManager 1/2

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

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: _*)
+  }
 }