[kernel-spark] Refactor DeltaSourceMetadataEvolutionSupport to be reuseable in v2

[PorridgeSwim]

🥞 Stacked PR

Use this link to review incremental changes.

• stack/SparkMetadataAdapter [Files changed] [MERGED]
  • stack/RefactorMetadataTrackingLog [Files changed] [MERGED]
    • stack/RefactorDeltaSourceMetadataEvolutionSupport [Files changed]
      • stack/MetadataEvolutionHandler2 [Files changed]
        • stack/NonAdditiveSchemaEvolution2 [Files changed]
          • stack/NonAdditiveSchemaEvolution3 [Files changed]
            • stack/consecutiveSchemaChangesMerger [Files changed]

---

Which Delta project/connector is this regarding?

• Spark
• Standalone
• Flink
• Kernel
• Other (fill in here)

Description

PR 3/7 in the non-additive schema evolution for V2 streaming connector stack.

Refactor DeltaSourceMetadataEvolutionSupport and DeltaColumnMapping so the schema change detection logic can be called from V2 without depending on V1 instance state.

DeltaSourceMetadataEvolutionSupport:

• Extract instance methods (validateAndResolveMetadataEvolution, checkColumnMappingSchemaChangesDuringStreaming, resolveMetadataEvolutionForCommitRange, etc.) to companion object statics that accept explicit parameters instead of accessing V1 DeltaSource via this
• V1 trait methods now delegate to the companion object statics

DeltaColumnMapping:

• Widen hasNoColumnMappingSchemaChanges from V1 Metadata to AbstractMetadata so V2 can call it via the adapter layer
• Extract assignColumnIdAndPhysicalNameToSchema(StructType, Map) from assignColumnIdAndPhysicalName(Metadata, Metadata, ...) — needed for simulating column mapping upgrades during NoMapping-to-NameMapping transitions

All changes are structural refactors with no behavioral change.

How was this patch tested?

Existing tests in DeltaSourceSchemaEvolutionSuite continue to pass. No behavioral changes.

Does this PR introduce any user-facing changes?

No.

[PorridgeSwim]

Range-diff: stack/RefactorMetadataTrackingLog (de14314 -> 6d9822e)

spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSourceMetadataEvolutionSupport.scala

@@ -160,7 +160,7 @@
 +      metadataChangeOpt.exists { newMetadata =>
 +        hasSchemaChangeComparedToStreamMetadata(
 +          newMetadata.schema, readSchemaAtSourceInit, spark) ||
-+          newMetadata.partitionColumns != readPartitionSchemaAtSourceInit ||
++          newMetadata.partitionSchema != readPartitionSchemaAtSourceInit ||
 +          newMetadata.configuration.filterKeys(_.startsWith("delta.")).toMap !=
 +            readConfigurationsAtSourceInit.filterKeys(_.startsWith("delta.")).toMap
 +      }

_{Reproduce locally: git range-diff cfcfa95..de14314 cfcfa95..6d9822e | Disable: git config gitstack.push-range-diff false}

[PorridgeSwim]

Range-diff: stack/RefactorMetadataTrackingLog (2490e84 -> 9491023)

spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSourceMetadataEvolutionSupport.scala

@@ -96,8 +96,10 @@
        DeltaSQLConf.DELTA_TYPE_WIDENING_BYPASS_STREAMING_TYPE_CHANGE_CHECK)
  
 +  /**
-+   * Whether this source should use schema tracking for metadata evolution.
-+   * Shared between v1 and v2 connectors.
++   * Whether this DeltaSource is utilizing a schema log entry as its read schema.
++   *
++   * If user explicitly turn on the flag to fall back to using latest schema to read (i.e. the
++   * legacy mode), we will ignore the schema log.
 +   */
 +  def shouldTrackMetadataChange(
 +      schemaReadOptions: DeltaStreamUtils.SchemaReadOptions,
@@ -107,8 +109,9 @@
 +  }
 +
 +  /**
-+   * Whether the tracking log should be initialized eagerly (log is provided but empty).
-+   * Shared between v1 and v2 connectors.
++   * Whether a schema tracking log is provided (and is empty), so we could initialize eagerly.
++   * This should only be used for the first write to the schema log, after then, schema tracking
++   * should not rely on this state any more.
 +   */
 +  def shouldInitializeMetadataTrackingEagerly(
 +      schemaReadOptions: DeltaStreamUtils.SchemaReadOptions,
@@ -169,7 +172,6 @@
 +
 +  /**
 +   * Check that the given schema is the same as the schema from the initial read snapshot.
-+   * This is shared between v1 and v2 connectors.
 +   */
 +  def hasSchemaChangeComparedToStreamMetadata(
 +      newSchema: StructType,

_{Reproduce locally: git range-diff 20f7851..2490e84 20f7851..9491023 | Disable: git config gitstack.push-range-diff false}

[PorridgeSwim]

Range-diff: stack/RefactorMetadataTrackingLog (9491023 -> 44b86f0)

spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala

@@ -0,0 +1,11 @@
+diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala b/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala
+--- a/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala
++++ b/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala
+  * Note: Please initialize this class using the companion object's `apply` method, which will
+  * assign correct values (`Set()` vs `None`) to [[readerFeatures]] and [[writerFeatures]].
+  */
+-case class Protocol private (
++case class Protocol (
+     minReaderVersion: Int,
+     minWriterVersion: Int,
+     @JsonInclude(Include.NON_ABSENT) // write to JSON only when the field is not `None`
\ No newline at end of file

_{Reproduce locally: git range-diff 20f7851..9491023 37e60f2..44b86f0 | Disable: git config gitstack.push-range-diff false}

[PorridgeSwim]

Range-diff: stack/RefactorMetadataTrackingLog (685d49e -> 38b283e)

spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala

@@ -1,11 +0,0 @@
-diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala b/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala
---- a/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala
-+++ b/spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala
-  * Note: Please initialize this class using the companion object's `apply` method, which will
-  * assign correct values (`Set()` vs `None`) to [[readerFeatures]] and [[writerFeatures]].
-  */
--case class Protocol private (
-+case class Protocol (
-     minReaderVersion: Int,
-     minWriterVersion: Int,
-     @JsonInclude(Include.NON_ABSENT) // write to JSON only when the field is not `None`
\ No newline at end of file

_{Reproduce locally: git range-diff 0951054..685d49e b1fa08b..38b283e | Disable: git config gitstack.push-range-diff false}

[TimothyW553]

a few comments.

for testing, existing tests pass through the v1 wrapper, so they would still pass even if the new static has a bug. Can we add one test that calls the static with an anonymous AbstractMetadata / AbstractProtocol directly? That is the test that proves the refactor is real.

[TimothyW553]

if Protocol ever gets a 5th field, this check will silently miss it. how about an equalsByFields on AbstractProtocol?

[PorridgeSwim]

sounds great to me

[TimothyW553]

9 params is a lot. can we bundle the *AtSourceInit ones into a small case class before v2 calls in?

[PorridgeSwim]

replaced it with readMetadataAtSourceInit

[TimothyW553]

9 lines of anonymous trait in the middle of a method. v2 will copy this pattern. can we extract a small AbstractMetadataAdapter case class?

[PorridgeSwim]

There is already KernelMetadataAdapter in v2 so we won't copy this pattern. Because oldMetadata is of trait AbstractMetadata, we cannot call copy() on it, current code is the idiomatic way to "copy with overrides" on a trait

[TimothyW553]

nit: old path wrote COLUMN_MAPPING_MAX_ID into config, this one forgets it. inert today, but breaks the 'no behavior change' claim - can we add it back?

[PorridgeSwim]

Added back, thank you for pointing it out

[TimothyW553]

two StructType then two Map right next to each other - easy to swap new and old by mistake. group them into one case class?

[PorridgeSwim]

changes to use abstractMetadata

[TimothyW553]

nit: this rebuilds on every call, and the streaming path hits it once per metadata action. v1 Metadata already has a precomputed partitionSchema -- can the v1 adapter just override this?

[PorridgeSwim]

The override is already added, you can check the file change in: spark/src/main/scala/org/apache/spark/sql/delta/actions/actions.scala

[PorridgeSwim]

Range-diff: stack/RefactorMetadataTrackingLog (a80dfae -> e3eb104)

spark-unified/src/test/scala/org/apache/spark/sql/delta/test/DeltaV2SourceSchemaEvolutionSuite.scala

@@ -0,0 +1,10 @@
+diff --git a/spark-unified/src/test/scala/org/apache/spark/sql/delta/test/DeltaV2SourceSchemaEvolutionSuite.scala b/spark-unified/src/test/scala/org/apache/spark/sql/delta/test/DeltaV2SourceSchemaEvolutionSuite.scala
+--- a/spark-unified/src/test/scala/org/apache/spark/sql/delta/test/DeltaV2SourceSchemaEvolutionSuite.scala
++++ b/spark-unified/src/test/scala/org/apache/spark/sql/delta/test/DeltaV2SourceSchemaEvolutionSuite.scala
+     "schema / checkpoint location unit tests - special characters in schema location",
+ 
+     // ========== Schema log core ==========
++    "detects metadata/protocol changes through the AbstractMetadata/AbstractProtocol surface",
+     "multiple delta source sharing same schema log is blocked",
+     "schema log is applied",
+     "concurrent schema log modification should be detected",
\ No newline at end of file

spark/src/main/scala/org/apache/spark/sql/delta/DeltaColumnMapping.scala

@@ -19,8 +19,9 @@
 -   *                                      mapping mode on a existing table
 -   * @return new metadata with Ids and physical names assigned
 +   * Core logic for assigning column IDs and physical names to a schema.
-+   * Takes raw schema and configuration inputs (no v1 Metadata dependency) so it can be
-+   * reused by both v1 and v2 connectors.
++   * Takes [[AbstractMetadata]] (no v1 Metadata dependency) so it can be reused by both v1
++   * and v2 connectors. Bundling schema + configuration on each side avoids the swap footgun
++   * of having two `StructType` and two `Map` parameters next to each other.
 +   *
 +   * @return (upgradedSchema, maxColumnId) - the schema with IDs/physical names assigned,
 +   *         and the final max column ID.
@@ -29,10 +30,8 @@
 -      newMetadata: Metadata,
 -      oldMetadata: Metadata,
 +  private[delta] def assignColumnIdAndPhysicalNameToSchema(
-+      newSchema: StructType,
-+      oldSchema: StructType,
-+      newConfiguration: Map[String, String],
-+      oldConfiguration: Map[String, String],
++      newMetadata: AbstractMetadata,
++      oldMetadata: AbstractMetadata,
        isChangingModeOnExistingTable: Boolean,
 -      isOverwritingSchema: Boolean): Metadata = {
 -    val rawSchema = newMetadata.schema
@@ -43,6 +42,10 @@
 -    val newSchema =
 -      SchemaMergingUtils.transformColumns(rawSchema)((path, field, _) => {
 +      isOverwritingSchema: Boolean): (StructType, Long) = {
++    val newSchema = newMetadata.schema
++    val oldSchema = oldMetadata.schema
++    val newConfiguration = newMetadata.configuration
++    val oldConfiguration = oldMetadata.configuration
 +    var maxId = DeltaConfigs.COLUMN_MAPPING_MAX_ID.fromMap(newConfiguration) max
 +      DeltaConfigs.COLUMN_MAPPING_MAX_ID.fromMap(oldConfiguration) max
 +      findMaxColumnId(newSchema)
@@ -102,9 +105,7 @@
 +      isChangingModeOnExistingTable: Boolean,
 +      isOverwritingSchema: Boolean): Metadata = {
 +    val (finalSchema, newMaxId) = assignColumnIdAndPhysicalNameToSchema(
-+      newMetadata.schema, oldMetadata.schema,
-+      newMetadata.configuration, oldMetadata.configuration,
-+      isChangingModeOnExistingTable, isOverwritingSchema)
++      newMetadata, oldMetadata, isChangingModeOnExistingTable, isOverwritingSchema)
  
      newMetadata.copy(
        schemaString = finalSchema.json,
@@ -135,9 +136,8 @@
 -          Map(DeltaConfigs.COLUMN_MAPPING_MODE.key -> newMetadata.columnMappingMode.name)
 -      )
 -      // use the same check
-+      val (upgradedSchema, _) = assignColumnIdAndPhysicalNameToSchema(
-+        oldMetadata.schema, oldMetadata.schema,
-+        oldMetadata.configuration, oldMetadata.configuration,
++      val (upgradedSchema, upgradedMaxId) = assignColumnIdAndPhysicalNameToSchema(
++        newMetadata = oldMetadata, oldMetadata = oldMetadata,
 +        isChangingModeOnExistingTable = true, isOverwritingSchema = false)
 +      // Construct an AbstractMetadata with the upgraded schema and the new column mapping mode
 +      // so the comparison utils below can recognize column mapping metadata.
@@ -148,7 +148,8 @@
 +        val schema: StructType = upgradedSchema
 +        val partitionColumns: Seq[String] = oldMetadata.partitionColumns
 +        val configuration: Map[String, String] = oldMetadata.configuration +
-+          (DeltaConfigs.COLUMN_MAPPING_MODE.key -> newMetadata.columnMappingMode.name)
++          (DeltaConfigs.COLUMN_MAPPING_MODE.key -> newMetadata.columnMappingMode.name,
++            DeltaConfigs.COLUMN_MAPPING_MAX_ID.key -> upgradedMaxId.toString)
 +        val columnMappingMode: DeltaColumnMappingMode = newMetadata.columnMappingMode
 +      }
        !hasColMappingOrPartitionSchemaChangeByMetadata(newMetadata, upgradedMetadata)

spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSource.scala

@@ -0,0 +1,17 @@
+diff --git a/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSource.scala b/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSource.scala
+--- a/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSource.scala
++++ b/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSource.scala
+    */
+   protected val readSchemaAtSourceInit: StructType = readSnapshotDescriptor.metadata.schema
+ 
+-  protected val readPartitionSchemaAtSourceInit: StructType =
+-    readSnapshotDescriptor.metadata.partitionSchema
+-
+   protected val readProtocolAtSourceInit: Protocol = readSnapshotDescriptor.protocol
+ 
+-  protected val readConfigurationsAtSourceInit: Map[String, String] =
+-    readSnapshotDescriptor.metadata.configuration
+-
+   /**
+    * Create a snapshot descriptor, customizing its metadata using metadata tracking if necessary
+    */
\ No newline at end of file

spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSourceMetadataEvolutionSupport.scala

@@ -72,9 +72,7 @@
 +      newSchemaVersion,
 +      persistedMetadataAtSourceInit,
 +      readProtocolAtSourceInit,
-+      readSchemaAtSourceInit,
-+      readPartitionSchemaAtSourceInit,
-+      readConfigurationsAtSourceInit,
++      readSnapshotDescriptor.metadata,
 +      spark)
    }
  
@@ -137,9 +135,9 @@
 +   * @param newSchemaVersion The version of the incoming change.
 +   * @param persistedMetadataAtSourceInit The persisted metadata at source init, if any.
 +   * @param readProtocolAtSourceInit The protocol at source init.
-+   * @param readSchemaAtSourceInit The schema at source init.
-+   * @param readPartitionSchemaAtSourceInit The partition schema at source init.
-+   * @param readConfigurationsAtSourceInit The table configurations at source init.
++   * @param readMetadataAtSourceInit The metadata at source init (schema, partition schema, and
++   *                                 configuration). Bundled to avoid the swap footgun of three
++   *                                 adjacent params.
 +   * @param spark The SparkSession (used for SQL conf checks).
 +   */
 +  def hasMetadataOrProtocolChangeComparedToStreamMetadata(
@@ -148,24 +146,18 @@
 +      newSchemaVersion: Long,
 +      persistedMetadataAtSourceInit: Option[PersistedMetadata],
 +      readProtocolAtSourceInit: AbstractProtocol,
-+      readSchemaAtSourceInit: StructType,
-+      readPartitionSchemaAtSourceInit: StructType,
-+      readConfigurationsAtSourceInit: Map[String, String],
++      readMetadataAtSourceInit: AbstractMetadata,
 +      spark: SparkSession): Boolean = {
 +    if (persistedMetadataAtSourceInit.exists(_.deltaCommitVersion >= newSchemaVersion)) {
 +      false
 +    } else {
-+      protocolChangeOpt.exists(p =>
-+        p.minReaderVersion != readProtocolAtSourceInit.minReaderVersion ||
-+          p.minWriterVersion != readProtocolAtSourceInit.minWriterVersion ||
-+          p.readerFeatures != readProtocolAtSourceInit.readerFeatures ||
-+          p.writerFeatures != readProtocolAtSourceInit.writerFeatures) ||
++      protocolChangeOpt.exists(p => !p.equalsByFields(readProtocolAtSourceInit)) ||
 +      metadataChangeOpt.exists { newMetadata =>
 +        hasSchemaChangeComparedToStreamMetadata(
-+          newMetadata.schema, readSchemaAtSourceInit, spark) ||
-+          newMetadata.partitionSchema != readPartitionSchemaAtSourceInit ||
++          newMetadata.schema, readMetadataAtSourceInit.schema, spark) ||
++          newMetadata.partitionSchema != readMetadataAtSourceInit.partitionSchema ||
 +          newMetadata.configuration.filterKeys(_.startsWith("delta.")).toMap !=
-+            readConfigurationsAtSourceInit.filterKeys(_.startsWith("delta.")).toMap
++            readMetadataAtSourceInit.configuration.filterKeys(_.startsWith("delta.")).toMap
 +      }
 +    }
 +  }

spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceSchemaEvolutionSuite.scala

@@ -0,0 +1,99 @@
+diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceSchemaEvolutionSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceSchemaEvolutionSuite.scala
+--- a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceSchemaEvolutionSuite.scala
++++ b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceSchemaEvolutionSuite.scala
+ import org.apache.spark.sql.delta.actions.{Metadata, Protocol}
+ import org.apache.spark.sql.delta.sources._
+ import org.apache.spark.sql.delta.test.{DeltaColumnMappingSelectedTestMixin, DeltaSQLCommandTest}
++import org.apache.spark.sql.delta.v2.interop.{AbstractMetadata, AbstractProtocol}
+ import org.apache.spark.sql.delta.util.JsonUtils
+ import org.apache.commons.io.FileUtils
+ import org.apache.commons.lang3.exception.ExceptionUtils
+   with DeltaSourceSuiteBase with DeltaColumnMappingSelectedTestMixin with DeltaSQLCommandTest {
+ 
+   override protected def runOnlyTests: Seq[String] = Seq(
++    "detects metadata/protocol changes through the AbstractMetadata/AbstractProtocol surface",
+     "schema log initialization with additive schema changes",
+     "detect incompatible schema change while streaming",
+     "trigger.Once with deferred commit should work",
+     ))
+   }
+ 
++  test("detects metadata/protocol changes through the AbstractMetadata/AbstractProtocol " +
++      "surface") {
++    // Anonymous trait impls (not V1 Metadata/Protocol) prove the static actually relies on
++    // the abstract surface. The trait wrapper used in production always passes V1 types,
++    // so this is the only path that would catch a regression specific to non-V1 impls.
++    val baseSchema = new StructType().add("a", StringType, nullable = true)
++
++    def mkMetadata(
++        sch: StructType = baseSchema,
++        partCols: Seq[String] = Seq.empty,
++        conf: Map[String, String] = Map.empty): AbstractMetadata = new AbstractMetadata {
++      override def id: String = "tid"
++      override def name: String = ""
++      override def description: String = ""
++      override def schema: StructType = sch
++      override def partitionColumns: Seq[String] = partCols
++      override def configuration: Map[String, String] = conf
++      override def columnMappingMode: DeltaColumnMappingMode = NoMapping
++    }
++
++    def mkProtocol(
++        readerV: Int = 1,
++        writerV: Int = 2,
++        readerFs: Option[Set[String]] = None,
++        writerFs: Option[Set[String]] = None): AbstractProtocol = new AbstractProtocol {
++      override def minReaderVersion: Int = readerV
++      override def minWriterVersion: Int = writerV
++      override def readerFeatures: Option[Set[String]] = readerFs
++      override def writerFeatures: Option[Set[String]] = writerFs
++    }
++
++    val readMetadata = mkMetadata()
++    val readProtocol = mkProtocol()
++
++    def call(
++        metadataChange: Option[AbstractMetadata] = None,
++        protocolChange: Option[AbstractProtocol] = None,
++        newVer: Long = 1L,
++        persisted: Option[PersistedMetadata] = None): Boolean =
++      DeltaSourceMetadataEvolutionSupport.hasMetadataOrProtocolChangeComparedToStreamMetadata(
++        metadataChange, protocolChange, newVer, persisted, readProtocol, readMetadata, spark)
++
++    // No change: both sides identical anonymous impls -> false.
++    assert(!call(metadataChange = Some(mkMetadata()), protocolChange = Some(mkProtocol())))
++
++    // Schema differs.
++    assert(call(metadataChange =
++      Some(mkMetadata(sch = baseSchema.add("b", StringType, nullable = true)))))
++
++    // Partition schema differs (same data schema, different partition columns).
++    assert(call(metadataChange = Some(mkMetadata(partCols = Seq("a")))))
++
++    // delta.* configuration differs.
++    assert(call(metadataChange = Some(mkMetadata(conf = Map("delta.foo" -> "bar")))))
++
++    // Non-delta.* configuration differs -> filtered out, no change.
++    assert(!call(metadataChange = Some(mkMetadata(conf = Map("foo" -> "bar")))))
++
++    // Protocol differs by a single field (uses equalsByFields under the hood).
++    assert(call(protocolChange = Some(mkProtocol(readerV = 2))))
++
++    // Persisted metadata is at or beyond newSchemaVersion -> short-circuits to false even if
++    // every other input would otherwise indicate a change.
++    val persisted = PersistedMetadata(
++      tableId = "tid",
++      deltaCommitVersion = 5L,
++      dataSchemaJson = baseSchema.json,
++      partitionSchemaJson = new StructType().json,
++      sourceMetadataPath = "")
++    assert(!call(
++      metadataChange = Some(mkMetadata(sch = baseSchema.add("b", StringType, nullable = true))),
++      protocolChange = Some(mkProtocol(readerV = 99)),
++      newVer = 3L,
++      persisted = Some(persisted)))
++  }
++
+   test("forward-compat: older version can read back newer JSON") {
+     val newSchema = PersistedMetadata(
+       tableId = "test",
\ No newline at end of file

_{Reproduce locally: git range-diff dd2dc63..a80dfae ba9972a..e3eb104 | Disable: git config gitstack.push-range-diff false}

[TimothyW553]

LGTM. but a note before complete approval: please add a test to validate the hasNoColumnMappingSchemaChanges in DeltaColumnMapping.

[TimothyW553]

Can we add one test that calls hasNoColumnMappingSchemaChanges with non-Metadata AbstractMetadata inputs, especially the NoMapping -> NameMapping upgrade path, so we prove this refactor actually works for the V2 caller?

[TimothyW553]

rough idea: add a small DeltaColumnMappingSuite test with anonymous AbstractMetadata wrappers around the same schemas/configs, then call DeltaColumnMapping.hasNoColumnMappingSchemaChanges(newAbstractMetadata, oldAbstractMetadata) and assert the expected true/false result.

[zikangh]

This is a unit test, perhaps it should live in DeltaSourceMetadataEvolutionSupportSuite.scala.

[zikangh]

This should be private.

[PorridgeSwim]

Range-diff: stack/RefactorMetadataTrackingLog (28bb702 -> 83962e6)

spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSourceMetadataEvolutionSupport.scala

@@ -165,7 +165,7 @@
 +  /**
 +   * Check that the given schema is the same as the schema from the initial read snapshot.
 +   */
-+  def hasSchemaChangeComparedToStreamMetadata(
++  private def hasSchemaChangeComparedToStreamMetadata(
 +      newSchema: StructType,
 +      readSchemaAtSourceInit: StructType,
 +      spark: SparkSession): Boolean = {

spark/src/test/scala/org/apache/spark/sql/delta/sources/DeltaSourceMetadataEvolutionSupportSuite.scala

@@ -0,0 +1,98 @@
+diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/sources/DeltaSourceMetadataEvolutionSupportSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/sources/DeltaSourceMetadataEvolutionSupportSuite.scala
+--- a/spark/src/test/scala/org/apache/spark/sql/delta/sources/DeltaSourceMetadataEvolutionSupportSuite.scala
++++ b/spark/src/test/scala/org/apache/spark/sql/delta/sources/DeltaSourceMetadataEvolutionSupportSuite.scala
+ 
+ package org.apache.spark.sql.delta.sources
+ 
+-import org.apache.spark.sql.delta.{DeltaColumnMapping, DeltaOptions, DeltaTestUtilsBase, DeltaThrowable}
++import org.apache.spark.sql.delta.{DeltaColumnMapping, DeltaColumnMappingMode, DeltaOptions}
++import org.apache.spark.sql.delta.{DeltaTestUtilsBase, DeltaThrowable, NoMapping}
++import org.apache.spark.sql.delta.v2.interop.{AbstractMetadata, AbstractProtocol}
+ 
+ import org.apache.spark.{SparkConf, SparkFunSuite}
+ import org.apache.spark.sql.test.SharedSparkSession
+-import org.apache.spark.sql.types.StructType
++import org.apache.spark.sql.types.{StringType, StructType}
+ 
+ /**
+  * Unit tests covering `DeltaSourceMetadataEvolutionSupport`, which detects non-additive schema
+       previousSchema = persistedMetadata("a byte", Map.empty)
+     )
+   }
++
++  test("detects metadata/protocol changes through the AbstractMetadata/AbstractProtocol " +
++      "surface") {
++    // Anonymous trait impls (not V1 Metadata/Protocol) prove the static actually relies on
++    // the abstract surface. The trait wrapper used in production always passes V1 types,
++    // so this is the only path that would catch a regression specific to non-V1 impls.
++    val baseSchema = new StructType().add("a", StringType, nullable = true)
++
++    def mkMetadata(
++        sch: StructType = baseSchema,
++        partCols: Seq[String] = Seq.empty,
++        conf: Map[String, String] = Map.empty): AbstractMetadata = new AbstractMetadata {
++      override def id: String = "tid"
++      override def name: String = ""
++      override def description: String = ""
++      override def schema: StructType = sch
++      override def partitionColumns: Seq[String] = partCols
++      override def configuration: Map[String, String] = conf
++      override def columnMappingMode: DeltaColumnMappingMode = NoMapping
++    }
++
++    def mkProtocol(
++        readerV: Int = 1,
++        writerV: Int = 2,
++        readerFs: Option[Set[String]] = None,
++        writerFs: Option[Set[String]] = None): AbstractProtocol = new AbstractProtocol {
++      override def minReaderVersion: Int = readerV
++      override def minWriterVersion: Int = writerV
++      override def readerFeatures: Option[Set[String]] = readerFs
++      override def writerFeatures: Option[Set[String]] = writerFs
++    }
++
++    val readMetadata = mkMetadata()
++    val readProtocol = mkProtocol()
++
++    def call(
++        metadataChange: Option[AbstractMetadata] = None,
++        protocolChange: Option[AbstractProtocol] = None,
++        newVer: Long = 1L,
++        persisted: Option[PersistedMetadata] = None): Boolean =
++      DeltaSourceMetadataEvolutionSupport.hasMetadataOrProtocolChangeComparedToStreamMetadata(
++        metadataChange, protocolChange, newVer, persisted, readProtocol, readMetadata, spark)
++
++    // No change: both sides identical anonymous impls -> false.
++    assert(!call(metadataChange = Some(mkMetadata()), protocolChange = Some(mkProtocol())))
++
++    // Schema differs.
++    assert(call(metadataChange =
++      Some(mkMetadata(sch = baseSchema.add("b", StringType, nullable = true)))))
++
++    // Partition schema differs (same data schema, different partition columns).
++    assert(call(metadataChange = Some(mkMetadata(partCols = Seq("a")))))
++
++    // delta.* configuration differs.
++    assert(call(metadataChange = Some(mkMetadata(conf = Map("delta.foo" -> "bar")))))
++
++    // Non-delta.* configuration differs -> filtered out, no change.
++    assert(!call(metadataChange = Some(mkMetadata(conf = Map("foo" -> "bar")))))
++
++    // Protocol differs by a single field (uses equalsByFields under the hood).
++    assert(call(protocolChange = Some(mkProtocol(readerV = 2))))
++
++    // Persisted metadata is at or beyond newSchemaVersion -> short-circuits to false even if
++    // every other input would otherwise indicate a change.
++    val persisted = PersistedMetadata(
++      tableId = "tid",
++      deltaCommitVersion = 5L,
++      dataSchemaJson = baseSchema.json,
++      partitionSchemaJson = new StructType().json,
++      sourceMetadataPath = "")
++    assert(!call(
++      metadataChange = Some(mkMetadata(sch = baseSchema.add("b", StringType, nullable = true))),
++      protocolChange = Some(mkProtocol(readerV = 99)),
++      newVer = 3L,
++      persisted = Some(persisted)))
++  }
+ }
\ No newline at end of file

spark-unified/src/test/scala/org/apache/spark/sql/delta/test/DeltaV2SourceSchemaEvolutionSuite.scala

@@ -1,10 +0,0 @@
-diff --git a/spark-unified/src/test/scala/org/apache/spark/sql/delta/test/DeltaV2SourceSchemaEvolutionSuite.scala b/spark-unified/src/test/scala/org/apache/spark/sql/delta/test/DeltaV2SourceSchemaEvolutionSuite.scala
---- a/spark-unified/src/test/scala/org/apache/spark/sql/delta/test/DeltaV2SourceSchemaEvolutionSuite.scala
-+++ b/spark-unified/src/test/scala/org/apache/spark/sql/delta/test/DeltaV2SourceSchemaEvolutionSuite.scala
-     "schema / checkpoint location unit tests - special characters in schema location",
- 
-     // ========== Schema log core ==========
-+    "detects metadata/protocol changes through the AbstractMetadata/AbstractProtocol surface",
-     "multiple delta source sharing same schema log is blocked",
-     "schema log is applied",
-     "concurrent schema log modification should be detected",
\ No newline at end of file

spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceSchemaEvolutionSuite.scala

@@ -1,99 +0,0 @@
-diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceSchemaEvolutionSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceSchemaEvolutionSuite.scala
---- a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceSchemaEvolutionSuite.scala
-+++ b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaSourceSchemaEvolutionSuite.scala
- import org.apache.spark.sql.delta.actions.{Metadata, Protocol}
- import org.apache.spark.sql.delta.sources._
- import org.apache.spark.sql.delta.test.{DeltaColumnMappingSelectedTestMixin, DeltaSQLCommandTest}
-+import org.apache.spark.sql.delta.v2.interop.{AbstractMetadata, AbstractProtocol}
- import org.apache.spark.sql.delta.util.JsonUtils
- import org.apache.commons.io.FileUtils
- import org.apache.commons.lang3.exception.ExceptionUtils
-   with DeltaSourceSuiteBase with DeltaColumnMappingSelectedTestMixin with DeltaSQLCommandTest {
- 
-   override protected def runOnlyTests: Seq[String] = Seq(
-+    "detects metadata/protocol changes through the AbstractMetadata/AbstractProtocol surface",
-     "schema log initialization with additive schema changes",
-     "detect incompatible schema change while streaming",
-     "trigger.Once with deferred commit should work",
-     ))
-   }
- 
-+  test("detects metadata/protocol changes through the AbstractMetadata/AbstractProtocol " +
-+      "surface") {
-+    // Anonymous trait impls (not V1 Metadata/Protocol) prove the static actually relies on
-+    // the abstract surface. The trait wrapper used in production always passes V1 types,
-+    // so this is the only path that would catch a regression specific to non-V1 impls.
-+    val baseSchema = new StructType().add("a", StringType, nullable = true)
-+
-+    def mkMetadata(
-+        sch: StructType = baseSchema,
-+        partCols: Seq[String] = Seq.empty,
-+        conf: Map[String, String] = Map.empty): AbstractMetadata = new AbstractMetadata {
-+      override def id: String = "tid"
-+      override def name: String = ""
-+      override def description: String = ""
-+      override def schema: StructType = sch
-+      override def partitionColumns: Seq[String] = partCols
-+      override def configuration: Map[String, String] = conf
-+      override def columnMappingMode: DeltaColumnMappingMode = NoMapping
-+    }
-+
-+    def mkProtocol(
-+        readerV: Int = 1,
-+        writerV: Int = 2,
-+        readerFs: Option[Set[String]] = None,
-+        writerFs: Option[Set[String]] = None): AbstractProtocol = new AbstractProtocol {
-+      override def minReaderVersion: Int = readerV
-+      override def minWriterVersion: Int = writerV
-+      override def readerFeatures: Option[Set[String]] = readerFs
-+      override def writerFeatures: Option[Set[String]] = writerFs
-+    }
-+
-+    val readMetadata = mkMetadata()
-+    val readProtocol = mkProtocol()
-+
-+    def call(
-+        metadataChange: Option[AbstractMetadata] = None,
-+        protocolChange: Option[AbstractProtocol] = None,
-+        newVer: Long = 1L,
-+        persisted: Option[PersistedMetadata] = None): Boolean =
-+      DeltaSourceMetadataEvolutionSupport.hasMetadataOrProtocolChangeComparedToStreamMetadata(
-+        metadataChange, protocolChange, newVer, persisted, readProtocol, readMetadata, spark)
-+
-+    // No change: both sides identical anonymous impls -> false.
-+    assert(!call(metadataChange = Some(mkMetadata()), protocolChange = Some(mkProtocol())))
-+
-+    // Schema differs.
-+    assert(call(metadataChange =
-+      Some(mkMetadata(sch = baseSchema.add("b", StringType, nullable = true)))))
-+
-+    // Partition schema differs (same data schema, different partition columns).
-+    assert(call(metadataChange = Some(mkMetadata(partCols = Seq("a")))))
-+
-+    // delta.* configuration differs.
-+    assert(call(metadataChange = Some(mkMetadata(conf = Map("delta.foo" -> "bar")))))
-+
-+    // Non-delta.* configuration differs -> filtered out, no change.
-+    assert(!call(metadataChange = Some(mkMetadata(conf = Map("foo" -> "bar")))))
-+
-+    // Protocol differs by a single field (uses equalsByFields under the hood).
-+    assert(call(protocolChange = Some(mkProtocol(readerV = 2))))
-+
-+    // Persisted metadata is at or beyond newSchemaVersion -> short-circuits to false even if
-+    // every other input would otherwise indicate a change.
-+    val persisted = PersistedMetadata(
-+      tableId = "tid",
-+      deltaCommitVersion = 5L,
-+      dataSchemaJson = baseSchema.json,
-+      partitionSchemaJson = new StructType().json,
-+      sourceMetadataPath = "")
-+    assert(!call(
-+      metadataChange = Some(mkMetadata(sch = baseSchema.add("b", StringType, nullable = true))),
-+      protocolChange = Some(mkProtocol(readerV = 99)),
-+      newVer = 3L,
-+      persisted = Some(persisted)))
-+  }
-+
-   test("forward-compat: older version can read back newer JSON") {
-     val newSchema = PersistedMetadata(
-       tableId = "test",
\ No newline at end of file

_{Reproduce locally: git range-diff 8378d33..28bb702 8378d33..83962e6 | Disable: git config gitstack.push-range-diff false}

[PorridgeSwim]

Range-diff: stack/RefactorMetadataTrackingLog (83962e6 -> 9036543)

spark/src/test/scala/org/apache/spark/sql/delta/DeltaColumnMappingSuite.scala

@@ -0,0 +1,67 @@
+diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaColumnMappingSuite.scala b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaColumnMappingSuite.scala
+--- a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaColumnMappingSuite.scala
++++ b/spark/src/test/scala/org/apache/spark/sql/delta/DeltaColumnMappingSuite.scala
+ import org.apache.spark.sql.delta.sources.DeltaSQLConf
+ import org.apache.spark.sql.delta.test.{DeltaSQLCommandTest, DeltaSQLTestUtils}
+ import org.apache.spark.sql.delta.test.DeltaTestImplicits._
++import org.apache.spark.sql.delta.v2.interop.AbstractMetadata
+ import org.apache.hadoop.fs.Path
+ import org.apache.parquet.format.converter.ParquetMetadataConverter
+ import org.apache.parquet.hadoop.ParquetFileReader
+     }
+   }
+ 
++  test("hasNoColumnMappingSchemaChanges accepts non-Metadata AbstractMetadata inputs") {
++    // Anonymous AbstractMetadata impls (not the V1 Metadata action) prove the API actually
++    // relies on the abstract surface. Especially exercises the NoMapping -> NameMapping upgrade
++    // branch, which synthesizes a fresh AbstractMetadata internally and reads back from it.
++    def mkAbstractMetadata(
++        sch: StructType,
++        mode: DeltaColumnMappingMode,
++        conf: Map[String, String] = Map.empty): AbstractMetadata = new AbstractMetadata {
++      override def id: String = "tid"
++      override def name: String = ""
++      override def description: String = ""
++      override def schema: StructType = sch
++      override def partitionColumns: Seq[String] = Seq.empty
++      override def configuration: Map[String, String] = conf
++      override def columnMappingMode: DeltaColumnMappingMode = mode
++    }
++
++    val baseSchema = new StructType().add("a", IntegerType).add("b", IntegerType)
++    val oldNoMapping = mkAbstractMetadata(baseSchema, NoMapping)
++
++    // Upgrade with no other change: synthesized post-upgrade schema uses logical names as
++    // physical names, matching what we pass for `new` -> read-compatible.
++    val newNameMappingSameSchema = mkAbstractMetadata(
++      DeltaColumnMapping.setPhysicalNames(
++        baseSchema, Map(Seq("a") -> "a", Seq("b") -> "b")),
++      NameMapping)
++    assert(DeltaColumnMapping.hasNoColumnMappingSchemaChanges(
++      newNameMappingSameSchema, oldNoMapping))
++
++    // Upgrade + drop: detected as a non-additive change.
++    val newNameMappingDropped = mkAbstractMetadata(
++      DeltaColumnMapping.setPhysicalNames(
++        new StructType().add("a", IntegerType), Map(Seq("a") -> "a")),
++      NameMapping)
++    assert(!DeltaColumnMapping.hasNoColumnMappingSchemaChanges(
++      newNameMappingDropped, oldNoMapping))
++
++    // Upgrade + rename (renamed column's physical name diverges from its logical name).
++    val newNameMappingRenamed = mkAbstractMetadata(
++      DeltaColumnMapping.setPhysicalNames(
++        new StructType().add("c", IntegerType).add("b", IntegerType),
++        Map(Seq("c") -> "a", Seq("b") -> "b")),
++      NameMapping)
++    assert(!DeltaColumnMapping.hasNoColumnMappingSchemaChanges(
++      newNameMappingRenamed, oldNoMapping))
++
++    // Downgrade NameMapping -> NoMapping is prohibited.
++    assert(!DeltaColumnMapping.hasNoColumnMappingSchemaChanges(
++      oldNoMapping, newNameMappingSameSchema))
++  }
++
+   testColumnMapping("create table through raw schema API should " +
+     "auto bump the version and retain input metadata") { mode =>
+ 
\ No newline at end of file

_{Reproduce locally: git range-diff 8378d33..83962e6 8378d33..9036543 | Disable: git config gitstack.push-range-diff false}