diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtils.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtils.scala
new file mode 100644
index 0000000000000..e5015610cca23
--- /dev/null
+++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtils.scala
@@ -0,0 +1,36 @@
+/*
+ * 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.pipelines.autocdc
+
+import org.apache.spark.sql.types.StructType
+
+private[autocdc] object AutoCdcSchemaUtils {
+
+  /**
+   * Enumerates every leaf path in `schema`, in schema order, as its sequence of name parts.
+   * Structs unfold recursively; every other type (including arrays and maps) is an opaque leaf.
+   */
+  def extractLeafPaths(schema: StructType): Seq[Seq[String]] =
+    schema.fields.toSeq.flatMap { field =>
+      field.dataType match {
+        case nested: StructType =>
+          extractLeafPaths(nested).map(field.name +: _)
+        case _ => Seq(Seq(field.name))
+      }
+    }
+}
diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala
index 2910d0664bc22..b8e82343c3297 100644
--- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala
+++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessor.scala
@@ -123,9 +123,10 @@ case class Scd2BatchProcessor(
    * consume.
    *
    * Step ordering is load-bearing: the row-extension steps reference user data columns that
-   * target-column selection is allowed to drop, so selection runs last. Unlike SCD1, no per-key
-   * deduplication step is performed here - SCD2 preserves every event as part of the row's
-   * history, including byte-identical full-event duplicates.
+   * target-column selection is allowed to drop. Selection therefore runs after those extensions,
+   * followed by target-schema alignment and version-map population. Unlike SCD1, no per-key
+   * deduplication step is performed here - SCD2 preserves every event as part of the row's history,
+   * including byte-identical full-event duplicates.
    *
    * Duplicate event elimination (e.g., collapsing two identical events at the same sequence),
    * whether across microbatches or within the same microbatch, is the responsibility of
@@ -133,22 +134,24 @@ case class Scd2BatchProcessor(
    *
    * @param microbatchDf
    *   the incoming CDC microbatch.
+   * @param targetTableDf
+   *   the current persisted target table.
    * @return
    *   a dataframe that retains every input row 1:1 - no rows added, dropped, reordered, or
-   *   merged - with the following schema, in column order:
-   *     1. The user columns of `microbatchDf` that survive [[ChangeArgs.columnSelection]], in
-   *        the order they appeared in the input.
-   *     2. [[startAtColName]], populated with the sequence value of the row.
-   *     3. [[endAtColName]], populated with the sequence value of the row IFF it's a delete
-   *        event, null otherwise.
-   *     4. [[cdcMetadataColName]], conforming to [[cdcMetadataColSchema]].
+   *   merged. Its fields use `targetTableDf` as the authority for order and spelling.
+   *   [[startAtColName]], [[endAtColName]], and [[cdcMetadataColName]] are populated according
+   *   to their documented contracts.
    */
-  private[autocdc] def preprocessMicrobatch(microbatchDf: DataFrame): DataFrame = {
+  private[autocdc] def preprocessMicrobatch(
+      microbatchDf: DataFrame,
+      targetTableDf: DataFrame): DataFrame = {
     microbatchDf
       .transform(extendMicrobatchRowsWithStartAt)
       .transform(extendMicrobatchRowsWithEndAt)
       .transform(extendMicrobatchRowsWithCdcMetadata)
       .transform(projectTargetColumnsOntoMicrobatch)
+      .transform(alignMicrobatchToTargetSchema(_, targetTableDf))
+      .transform(extendMicrobatchRowsWithVersionMap)
   }
 
   /**
@@ -187,21 +190,68 @@ case class Scd2BatchProcessor(
   /**
    * Project the operational CDC metadata column carrying the literal event sequence. Downstream
    * merges rely on it to preserve original event lineage regardless of how rows start/end-at are
-   * coalesced.
+   * coalesced. The version map is initially null; [[extendMicrobatchRowsWithVersionMap]] populates
+   * it after column selection runs.
    */
   private def extendMicrobatchRowsWithCdcMetadata(microbatchDf: DataFrame): DataFrame = {
     microbatchDf.withColumn(
       colName = AutoCdcReservedNames.cdcMetadataColName,
       col = Scd2BatchProcessor.constructCdcMetadataCol(
         recordStartAt = changeArgs.sequencing,
-        // TODO (SPARK-59183): actually populate version map according to ignore-null selection and
-        // actual authorship in microbatch.
         versionMap = F.lit(null),
         sequencingType = resolvedSequencingType
       )
     )
   }
 
+  /**
+   * Populates the version map on each microbatch row, recording which leaves the event authored.
+   * Null leaves in ignore-null columns get a `false` entry (declined); null leaves in other
+   * columns get `true` (authored null); non-null leaves need no entry. No-op when ignore-null
+   * is off.
+   *
+   * Delete-encoded rows (those matching [[ChangeArgs.deleteCondition]]) receive a null version
+   * map: their data-column values are not part of the SCD2 contract, so authorship tracking
+   * is not applicable.
+   *
+   * Must run after [[projectTargetColumnsOntoMicrobatch]] and
+   * [[alignMicrobatchToTargetSchema]], because the eligible schema is computed from the selected,
+   * target-aligned schema.
+   *
+   * TODO(SPARK-59343): decide how to handle the ignore-null selection changing between
+   * partial-retry attempts of the same microbatch.
+   */
+  private def extendMicrobatchRowsWithVersionMap(alignedDf: DataFrame): DataFrame =
+    changeArgs.ignoreNullSelection match {
+      case None => alignedDf
+      case Some(ignoreNullSelection) =>
+        val cdcMetadataCol = F.col(AutoCdcReservedNames.cdcMetadataColName)
+        val resolver = alignedDf.sparkSession.sessionState.conf.resolver
+        val schemaEligibleForNullAuthorshipTracking =
+          Scd2BatchProcessor.computeUserDataSchema(
+            schema = alignedDf.schema,
+            changeArgs = changeArgs,
+            resolver = resolver
+          )
+
+        // Only upsert rows get a populated version map. By convention, delete-encoded
+        // rows always maintain a null version map. We detect deletes via endAt rather
+        // than changeArgs.deleteCondition because column selection may have already
+        // dropped the column the delete condition references.
+        val isUpsertRow = F.col(Scd2BatchProcessor.endAtColName).isNull
+        val versionMap = F.when(isUpsertRow, Scd2VersionMap.buildVersionMap(
+          schema = schemaEligibleForNullAuthorshipTracking,
+          ignoreNullSelection = ignoreNullSelection,
+          resolver = resolver
+        ))
+
+        alignedDf.withColumn(
+          colName = AutoCdcReservedNames.cdcMetadataColName,
+          col = cdcMetadataCol
+            .withField(Scd2BatchProcessor.versionMapFieldName, versionMap)
+        )
+    }
+
   /**
    * Apply the user's target column selection while preserving the SCD2 framework columns; the
    * latter are required by downstream merges and persisted to both the auxiliary and target
@@ -246,6 +296,19 @@ case class Scd2BatchProcessor(
     microbatch.select(finalColumnsToSelect: _*)
   }
 
+  /**
+   * Align the selected incoming rows to the current persisted target schema. The target-shaped
+   * side is empty, so this retains exactly the microbatch's rows while [[DataFrame.unionByName]]
+   * supplies nulls for target columns omitted by the source, including nested struct/array fields.
+   *
+   * Keeping the target as the left schema authority also preserves its column order and exact
+   * case-spelling.
+   */
+  private def alignMicrobatchToTargetSchema(
+      projectedDf: DataFrame,
+      targetTableDf: DataFrame): DataFrame =
+    targetTableDf.limit(0).unionByName(projectedDf, allowMissingColumns = true)
+
   /**
    * For each key in the preprocessed microbatch, compute the earliest [[recordStartAtFieldName]]
    * across the key's events.
@@ -1445,23 +1508,37 @@ object Scd2BatchProcessor {
   private[pipelines] def computeTrackedHistoryColumns(
       schema: StructType,
       changeArgs: ChangeArgs,
-      resolver: Resolver): Seq[String] = {
-    val keyColNames = changeArgs.keys.map(_.name)
-
-    val eligibleSchema = StructType(schema.fields.filterNot { field =>
-      reservedFrameworkColNames.exists(resolver(_, field.name)) ||
-        keyColNames.exists(resolver(_, field.name))
-    })
-
+      resolver: Resolver): Seq[String] =
     ColumnSelection
       .applyToSchema(
         schemaName = "trackHistorySelection",
-        schema = eligibleSchema,
+        schema = computeUserDataSchema(schema, changeArgs, resolver),
         columnSelection = changeArgs.trackHistorySelection,
         resolver = resolver
       )
       .fieldNames
       .toImmutableArraySeq
+
+  /**
+   * The subset of `schema` that is user data a column selection may act on: every field that is
+   * neither a framework reserved column nor one of [[ChangeArgs.keys]]. Field order is preserved.
+   *
+   * Both [[ChangeArgs.trackHistorySelection]] and [[ChangeArgs.ignoreNullSelection]] resolve
+   * against this, so an exclude-list in either cannot pick up a key or a framework column, and
+   * an include-list naming one fails as not found.
+   *
+   * `schema` is expected to have already been narrowed by [[ChangeArgs.columnSelection]] and then
+   * aligned to the persisted target schema. This method does not re-apply the selection.
+   */
+  private[pipelines] def computeUserDataSchema(
+      schema: StructType,
+      changeArgs: ChangeArgs,
+      resolver: Resolver): StructType = {
+    val keyColNames = changeArgs.keys.map(_.name)
+    StructType(schema.fields.filterNot { field =>
+      reservedFrameworkColNames.exists(resolver(_, field.name)) ||
+        keyColNames.exists(resolver(_, field.name))
+    })
   }
 
   /**
diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala
index e4adfcb516be7..124dfba9757ab 100644
--- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala
+++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2ForeachBatchHandler.scala
@@ -68,15 +68,18 @@ case class Scd2ForeachBatchHandler(
       batchId = batchId
     ).validateMicrobatch()
 
-    val preprocessedBatchDf = batchProcessor.preprocessMicrobatch(batchDf)
+    val targetTableDf = batchDf.sparkSession.read.table(targetTableIdentifier.quotedString)
+    val auxTableDf = batchDf.sparkSession.read.table(auxiliaryTableIdentifier.quotedString)
+
+    val preprocessedBatchDf = batchProcessor.preprocessMicrobatch(
+      microbatchDf = batchDf,
+      targetTableDf = targetTableDf
+    )
 
     val perKeyMinimumSequenceInMicrobatchDf = batchProcessor.computeMinimumSequencePerKey(
       preprocessedBatchDf
     )
 
-    val auxTableDf = batchDf.sparkSession.read.table(auxiliaryTableIdentifier.quotedString)
-    val targetTableDf = batchDf.sparkSession.read.table(targetTableIdentifier.quotedString)
-
     val perKeyAffectedSequenceCutoffDf = batchProcessor.computePerKeyAffectedSequenceCutoff(
       rawAuxiliaryTableDf = auxTableDf,
       targetTableDf = targetTableDf,
@@ -95,12 +98,10 @@ case class Scd2ForeachBatchHandler(
       perKeyAffectedSequenceCutoffDf = perKeyAffectedSequenceCutoffDf
     )
 
-    // The three inputs share the canonical SCD2 row schema by name, but not necessarily by column
-    // set: after cross-run schema evolution the target (and the aux table, which mirrors it) can
-    // carry user columns that the current microbatch no longer emits. `allowMissingColumns` pads
-    // such columns with null on the side that lacks them (recursing into structs and arrays; map
-    // types are not supported) instead of failing the union. (findAffectedRowsFromAuxiliaryTable
-    // drops the aux-only deletedByBatchId column.)
+    // Preprocessing has already aligned the microbatch to the target schema. Keep
+    // allowMissingColumns here as a safeguard for nested differences between persisted target and
+    // auxiliary rows; findAffectedRowsFromAuxiliaryTable also drops the aux-only
+    // deletedByBatchId column.
     val microbatchAndAffectedRows = preprocessedBatchDf
       .unionByName(affectedRowsFromAuxiliaryTable, allowMissingColumns = true)
       .unionByName(affectedRowsFromTargetTable, allowMissingColumns = true)
diff --git a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala
index 5926d9e108ad0..d06a85ac7b14f 100644
--- a/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala
+++ b/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMap.scala
@@ -17,7 +17,13 @@
 
 package org.apache.spark.sql.pipelines.autocdc
 
-import org.apache.spark.sql.types.{BooleanType, MapType, StringType}
+import org.json4s.JsonAST.{JArray, JString}
+import org.json4s.jackson.JsonMethods.compact
+
+import org.apache.spark.sql.{functions => F, Column}
+import org.apache.spark.sql.catalyst.analysis.Resolver
+import org.apache.spark.sql.catalyst.util.QuotingUtils
+import org.apache.spark.sql.types.{BooleanType, MapType, StringType, StructType}
 
 /**
  * Per-row column authorship tracker for SCD2 ignore-null semantics.
@@ -77,15 +83,69 @@ private[pipelines] object Scd2VersionMap {
   /**
    * Schema of the version map: `Map(String, Boolean)`.
    *
-   * Keys are dot-delimited paths to *leaf* columns that received a null value in their
-   * corresponding upsert event (e.g. `"address.city"`, `` "`has space`.city" ``). Paths
-   * must be formatted by [[org.apache.spark.sql.catalyst.util.QuotingUtils.quoted]] to
-   * ensure segments that need quoting are back-tick escaped.
+   * Keys are compact JSON arrays of the name parts of *leaf* columns that received a null
+   * value in their upsert event (e.g. `["address","city"]`). Keeping
+   * name parts separate distinguishes a nested path from a column whose name contains dots
+   * and keeps persisted keys independent of SQL identifier quoting rules. Name parts use the
+   * persisted target schema's canonical spelling.
    *
-   * Values indicate authorship. I.e, `true` => authored-null, `false` => unauthored-null.
+   * Values indicate authorship: `true` means authored-null, `false` means unauthored-null.
+   * Null values never appear in the map.
    *
    * Lack of entry in the map for a null-valued leaf column implies the column was
    * schema-evolved with an unauthored-null.
    */
   def mapType: MapType = MapType(StringType, BooleanType, valueContainsNull = false)
+
+  /** Encodes a leaf path as the compact JSON string persisted as its version map key. */
+  private[autocdc] def encodePath(path: Seq[String]): String =
+    compact(JArray(path.map(JString(_)).toList))
+
+  /**
+   * Builds the ingest-time version map column for a microbatch. Each row's map records which
+   * null leaves are authored vs declined, based on the active ignore-null selection.
+   *
+   * @param schema The schema whose leaves the version map covers. Null-authorship is tracked
+   *   for every leaf column in this schema, as per the version map contract.
+   * @param ignoreNullSelection The ignore-null column selection this schema is being ingested
+   *   under.
+   * @param resolver Case-sensitivity resolver for column name matching.
+   * @return A [[Column]] of [[mapType]] schema.
+   */
+  def buildVersionMap(
+      schema: StructType,
+      ignoreNullSelection: ColumnSelection,
+      resolver: Resolver): Column = {
+    val ignoreNullColumns = ColumnSelection.applyToSchema(
+      schemaName = "ignoreNullSelection",
+      schema = schema,
+      columnSelection = Some(ignoreNullSelection),
+      resolver = resolver
+    )
+    val ignoreNullLeafPaths = AutoCdcSchemaUtils.extractLeafPaths(ignoreNullColumns).toSet
+
+    // For each leaf, build a nullable struct (key, value). The struct is non-null only when
+    // the leaf column's runtime value is null (meaning the leaf needs a version map entry).
+    // The value is a non-nullable BooleanType literal indicating authorship: true if the null
+    // is authored, false if declined.
+    val candidateEntries = AutoCdcSchemaUtils.extractLeafPaths(schema).map { path =>
+      val encodedPath = encodePath(path)
+      val isIgnoreNullLeaf = ignoreNullLeafPaths.contains(path)
+      val leafIsNull = F.col(QuotingUtils.quoteNameParts(path)).isNull
+
+      // If the leaf is not null, this candidate entry will simply resolve to null and will not be
+      // added to the version map during construction below.
+      F.when(leafIsNull, F.struct(
+        F.lit(encodedPath).as("key"),
+        F.lit(!isIgnoreNullLeaf).as("value")
+      ))
+    }
+
+    if (candidateEntries.isEmpty) {
+      F.map().cast(mapType)
+    } else {
+      val nonNullEntries = F.filter(F.array(candidateEntries: _*), (e: Column) => e.isNotNull)
+      F.map_from_entries(nonNullEntries)
+    }
+  }
 }
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtilsSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtilsSuite.scala
new file mode 100644
index 0000000000000..d5ce59d60056c
--- /dev/null
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/AutoCdcSchemaUtilsSuite.scala
@@ -0,0 +1,69 @@
+/*
+ * 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.pipelines.autocdc
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.types._
+
+class AutoCdcSchemaUtilsSuite extends SparkFunSuite {
+
+  test("extractLeafPaths returns single-element paths for flat columns") {
+    val schema = new StructType()
+      .add("a", IntegerType)
+      .add("b", StringType)
+      .add("c", DoubleType)
+
+    assert(AutoCdcSchemaUtils.extractLeafPaths(schema) ===
+      Seq(Seq("a"), Seq("b"), Seq("c")))
+  }
+
+  test("extractLeafPaths returns leaves rather than intermediate structs") {
+    val schema = new StructType()
+      .add("x", IntegerType)
+      .add("address", new StructType()
+        .add("city", StringType)
+        .add("zip", IntegerType))
+
+    assert(AutoCdcSchemaUtils.extractLeafPaths(schema) ===
+      Seq(Seq("x"), Seq("address", "city"), Seq("address", "zip")))
+  }
+
+  test("extractLeafPaths returns full paths for deeply nested structs") {
+    val schema = new StructType()
+      .add("top", new StructType()
+        .add("mid", new StructType()
+          .add("leaf", StringType)))
+
+    assert(AutoCdcSchemaUtils.extractLeafPaths(schema) ===
+      Seq(Seq("top", "mid", "leaf")))
+  }
+
+  test("extractLeafPaths treats arrays and maps as opaque leaves") {
+    val schema = new StructType()
+      .add("tags", ArrayType(StringType))
+      .add("props", MapType(StringType, IntegerType))
+      .add("plain", IntegerType)
+
+    assert(AutoCdcSchemaUtils.extractLeafPaths(schema) ===
+      Seq(Seq("tags"), Seq("props"), Seq("plain")))
+  }
+
+  test("extractLeafPaths returns an empty sequence for an empty schema") {
+    assert(AutoCdcSchemaUtils.extractLeafPaths(new StructType()) === Seq.empty)
+  }
+}
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala
index 71d837740321d..9fd1b1dff406d 100644
--- a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2BatchProcessorSuite.scala
@@ -30,6 +30,28 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
   private def microbatchOf(schema: StructType)(rows: Row*): DataFrame =
     spark.createDataFrame(spark.sparkContext.parallelize(rows), schema)
 
+  /**
+   * Preprocess a microbatch against either an explicit persisted user schema or, by default, the
+   * microbatch's user-selected schema. The latter keeps tests unrelated to cross-run evolution
+   * focused on their existing axis.
+   */
+  private def preprocessMicrobatch(
+      processor: Scd2BatchProcessor,
+      microbatch: DataFrame,
+      targetUserSchema: Option[StructType] = None): DataFrame = {
+    val selectedMicrobatchSchema = ColumnSelection.applyToSchema(
+      schemaName = "microbatch",
+      schema = microbatch.schema,
+      columnSelection = processor.changeArgs.columnSelection,
+      resolver = spark.sessionState.conf.resolver
+    )
+    val targetTableDf = targetTableOf(
+      targetUserSchema.getOrElse(selectedMicrobatchSchema),
+      processor.resolvedSequencingType
+    )()
+    processor.preprocessMicrobatch(microbatch, targetTableDf)
+  }
+
   /**
    * Build an mock aux-table [[DataFrame]] from explicit user rows + framework column values.
    */
@@ -270,7 +292,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
       resolvedSequencingType = LongType
     )
 
-    val result = processor.preprocessMicrobatch(batch)
+    val result = preprocessMicrobatch(processor, batch)
 
     assert(result.schema.fieldNames.toSeq == Seq(
       "id", "seq", "value",
@@ -309,7 +331,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
       resolvedSequencingType = LongType
     )
 
-    val result = processor.preprocessMicrobatch(batch)
+    val result = preprocessMicrobatch(processor, batch)
 
     assert(result.collect().isEmpty)
     assert(result.schema.fieldNames.toSeq == Seq(
@@ -353,7 +375,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
     //   - __RECORD_START_AT = sequencing for every row, regardless of delete vs upsert
     //                        (lineage preserved into the merge step)
     checkAnswer(
-      df = processor.preprocessMicrobatch(batch),
+      df = preprocessMicrobatch(processor, batch),
       expectedAnswer = Seq(
         Row(1, 10L, "first-upsert", false, 10L, null, Row(10L, null)),
         Row(1, 20L, "second-upsert", false, 20L, null, Row(20L, null)),
@@ -391,7 +413,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
 
     // Both rows must survive verbatim.
     checkAnswer(
-      df = processor.preprocessMicrobatch(batch),
+      df = preprocessMicrobatch(processor, batch),
       expectedAnswer = Seq(
         Row(1, 10L, "alice", false, 10L, null, Row(10L, null)),
         Row(1, 10L, "alice", false, 10L, null, Row(10L, null))
@@ -421,7 +443,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
     )
 
     checkAnswer(
-      df = processor.preprocessMicrobatch(batch).select(
+      df = preprocessMicrobatch(processor, batch).select(
         F.col(Scd2BatchProcessor.endAtColName)
       ),
       expectedAnswer = Seq(Row(null), Row(null))
@@ -452,7 +474,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
     )
 
     checkAnswer(
-      df = processor.preprocessMicrobatch(batch).select(
+      df = preprocessMicrobatch(processor, batch).select(
         F.col(Scd2BatchProcessor.endAtColName)
       ),
       expectedAnswer = Row(null)
@@ -485,7 +507,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
       resolvedSequencingType = LongType
     )
 
-    val result = processor.preprocessMicrobatch(batch)
+    val result = preprocessMicrobatch(processor, batch)
 
     checkAnswer(
       df = result.select(
@@ -522,7 +544,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
       resolvedSequencingType = LongType
     )
 
-    val result = processor.preprocessMicrobatch(batch)
+    val result = preprocessMicrobatch(processor, batch)
 
     assert(result.schema.fieldNames.toSeq == Seq(
       "id", "name", "age", "seq",
@@ -549,7 +571,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
       resolvedSequencingType = LongType
     )
 
-    val result = processor.preprocessMicrobatch(batch)
+    val result = preprocessMicrobatch(processor, batch)
 
     assert(result.schema.fieldNames.toSeq == Seq(
       "id", "age",
@@ -581,7 +603,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
       resolvedSequencingType = LongType
     )
 
-    val result = processor.preprocessMicrobatch(batch)
+    val result = preprocessMicrobatch(processor, batch)
 
     assert(result.schema.fieldNames.toSeq == Seq(
       "id", "age", "seq",
@@ -614,7 +636,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
       resolvedSequencingType = LongType
     )
 
-    val result = processor.preprocessMicrobatch(batch)
+    val result = preprocessMicrobatch(processor, batch)
 
     // Output column order follows the microbatch schema (id before age), not the user's listing
     // order in IncludeColumns. Framework columns are always appended last.
@@ -646,7 +668,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
         resolvedSequencingType = LongType
       )
 
-      val result = processor.preprocessMicrobatch(batch)
+      val result = preprocessMicrobatch(processor, batch)
 
       // Output column names follow the microbatch schema's casing, not the user's casing.
       assert(result.schema.fieldNames.toSeq == Seq(
@@ -685,7 +707,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
       resolvedSequencingType = LongType
     )
 
-    val result = processor.preprocessMicrobatch(batch)
+    val result = preprocessMicrobatch(processor, batch)
 
     assert(result.schema.fieldNames.toSeq == Seq(
       "id", "user.id",
@@ -729,7 +751,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
 
     // The orchestrator runs row-extension steps before column selection, so the framework
     // columns reference seq / is_delete fully even though the final projection drops them.
-    val result = processor.preprocessMicrobatch(batch)
+    val result = preprocessMicrobatch(processor, batch)
 
     assert(result.schema.fieldNames.toSeq == Seq(
       "id", "value",
@@ -746,6 +768,327 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
     )
   }
 
+  gridTest("preprocessMicrobatch keeps divergent target spelling under case-insensitive analysis")(
+    Seq(
+      ("id", "Value", "ID", "value"),
+      ("ID", "value", "id", "Value")
+    )
+  ) { case (sourceKey, sourceValue, targetKey, targetValue) =>
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+      val batchSchema = new StructType()
+        .add(sourceKey, IntegerType)
+        .add(sourceValue, StringType)
+        .add("seq", LongType)
+      val targetUserSchema = new StructType()
+        .add(targetKey, IntegerType)
+        .add(targetValue, StringType)
+      val batch = microbatchOf(batchSchema)(Row(1, "a", 10L))
+      val processor = Scd2BatchProcessor(
+        changeArgs = ChangeArgs(
+          keys = Seq(UnqualifiedColumnName(sourceKey)),
+          sequencing = F.col("seq"),
+          storedAsScdType = ScdType.Type2,
+          columnSelection = Some(ColumnSelection.ExcludeColumns(
+            Seq(UnqualifiedColumnName("seq"))))
+        ),
+        resolvedSequencingType = LongType
+      )
+
+      val result = preprocessMicrobatch(processor, batch, Some(targetUserSchema))
+      assert(result.schema.fieldNames.take(2).toSeq == Seq(targetKey, targetValue))
+      checkAnswer(result.select(F.col(targetKey), F.col(targetValue)), Row(1, "a"))
+    }
+  }
+
+  test("preprocessMicrobatch keeps distinct case-sensitive columns between target " +
+    "and microbatch") {
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+      val batchSchema = new StructType()
+        .add("id", IntegerType)
+        .add("Value", StringType)
+        .add("seq", LongType)
+      val targetUserSchema = new StructType()
+        .add("id", IntegerType)
+        .add("value", StringType)
+        .add("Value", StringType)
+      val batch = microbatchOf(batchSchema)(Row(1, "a", 10L))
+      val processor = Scd2BatchProcessor(
+        changeArgs = ChangeArgs(
+          keys = Seq(UnqualifiedColumnName("id")),
+          sequencing = F.col("seq"),
+          storedAsScdType = ScdType.Type2,
+          columnSelection = Some(ColumnSelection.ExcludeColumns(
+            Seq(UnqualifiedColumnName("seq"))))
+        ),
+        resolvedSequencingType = LongType
+      )
+
+      val result = preprocessMicrobatch(processor, batch, Some(targetUserSchema))
+      assert(result.schema.fieldNames.take(3).toSeq == Seq("id", "value", "Value"))
+      checkAnswer(result.select(F.col("value"), F.col("Value")), Row(null, "a"))
+    }
+  }
+
+  test("preprocessMicrobatch keeps nested target field spelling recursively under " +
+    "case-insensitive analysis") {
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+      val batchSchema = new StructType()
+        .add("ID", IntegerType)
+        .add("Value", new StructType().add("City", IntegerType))
+        .add("seq", LongType)
+      val targetUserSchema = new StructType()
+        .add("id", IntegerType)
+        .add("value", new StructType()
+          .add("city", IntegerType)
+          .add("removedNested", StringType))
+        .add("removedTopLevel", StringType)
+      val batch = microbatchOf(batchSchema)(Row(1, Row(2), 10L))
+      val processor = Scd2BatchProcessor(
+        changeArgs = ChangeArgs(
+          keys = Seq(UnqualifiedColumnName("id")),
+          sequencing = F.col("seq"),
+          storedAsScdType = ScdType.Type2,
+          columnSelection = Some(ColumnSelection.ExcludeColumns(
+            Seq(UnqualifiedColumnName("seq"))))
+        ),
+        resolvedSequencingType = LongType
+      )
+
+      val result = preprocessMicrobatch(processor, batch, Some(targetUserSchema))
+      assert(result.schema.fieldNames.toSeq == Seq(
+        "id",
+        "value",
+        "removedTopLevel",
+        Scd2BatchProcessor.startAtColName,
+        Scd2BatchProcessor.endAtColName,
+        AutoCdcReservedNames.cdcMetadataColName
+      ))
+      assert(result.schema("value").dataType.asInstanceOf[StructType].fieldNames.toSeq ==
+        Seq("city", "removedNested"))
+      checkAnswer(
+        df = result,
+        expectedAnswer = Row(1, Row(2, null), null, 10L, null, Row(10L, null))
+      )
+    }
+  }
+
+  gridTest("preprocessMicrobatch leaves delete-representing rows with a null version map")(
+    Seq(
+      // Delete condition is specified; all non-matching rows should be treated as upsert.
+      Some(F.col("is_delete")),
+      // Delete condition is unspecified; all microbatch rows should be treated as upsert.
+      None
+    )
+  ) { case (deleteCondition) =>
+    val schema = new StructType()
+      .add("id", IntegerType)
+      .add("value", StringType)
+      .add("seq", LongType)
+      .add("is_delete", BooleanType)
+
+    val batch = microbatchOf(schema)(
+      Row(1, null, 10L, false), // upsert with null value
+      Row(1, "a", 20L, false), // upsert with non-null value
+      Row(1, null, 30L, true) // delete iff deleteCondition is set
+    )
+
+    val processor = Scd2BatchProcessor(
+      changeArgs = ChangeArgs(
+        keys = Seq(UnqualifiedColumnName("id")),
+        sequencing = F.col("seq"),
+        storedAsScdType = ScdType.Type2,
+        deleteCondition = deleteCondition,
+        // Even if we drop `is_delete` from the output schema, delete-row detection should still
+        // work and the row should still receive a null version map.
+        columnSelection = Some(ColumnSelection.ExcludeColumns(
+          Seq(UnqualifiedColumnName("is_delete")))),
+        ignoreNullSelection =
+          Some(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("value"))))
+      ),
+      resolvedSequencingType = LongType
+    )
+
+    val result = preprocessMicrobatch(processor, batch)
+
+    val versionMaps = result.select(
+      F.col("seq"),
+      Scd2BatchProcessor.versionMapOf(
+        F.col(AutoCdcReservedNames.cdcMetadataColName)
+      ).as("vm")
+    )
+
+    // Upsert rows always get a populated version map. The third row is a delete (null
+    // version map) only when deleteCondition is set; otherwise it is an upsert too.
+    val valueVersionMapKey = Scd2VersionMap.encodePath(Seq("value"))
+    val expectedDeleteRowMap: Any =
+      if (deleteCondition.isDefined) null else Map(valueVersionMapKey -> false)
+
+    checkAnswer(
+      df = versionMaps,
+      expectedAnswer = Seq(
+        Row(10L, Map(valueVersionMapKey -> false)),
+        Row(20L, Map.empty[String, Boolean]),
+        Row(30L, expectedDeleteRowMap)
+      )
+    )
+  }
+
+  gridTest("preprocessMicrobatch applies ignore-null selection to reductively removed columns")(
+    Seq(
+      // Include only value: removed is outside ignore-null and its padded null is authored.
+      (ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("value"))), true),
+      // Exclude value: removed is inside ignore-null and its padded null is declined.
+      (ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("value"))), false)
+    )
+  ) { case (ignoreNullSelection, expectedAuthorship) =>
+    val batchSchema = new StructType()
+      .add("id", IntegerType)
+      .add("value", StringType)
+      .add("seq", LongType)
+    val targetUserSchema = new StructType()
+      .add("id", IntegerType)
+      .add("value", StringType)
+      .add("removed", StringType)
+    val batch = microbatchOf(batchSchema)(Row(1, "a", 10L))
+    val processor = Scd2BatchProcessor(
+      changeArgs = ChangeArgs(
+        keys = Seq(UnqualifiedColumnName("id")),
+        sequencing = F.col("seq"),
+        storedAsScdType = ScdType.Type2,
+        columnSelection = Some(ColumnSelection.ExcludeColumns(
+          Seq(UnqualifiedColumnName("seq")))),
+        ignoreNullSelection = Some(ignoreNullSelection)
+      ),
+      resolvedSequencingType = LongType
+    )
+
+    val result = preprocessMicrobatch(processor, batch, Some(targetUserSchema))
+    checkAnswer(
+      df = result.select(
+        F.col("removed"),
+        Scd2BatchProcessor.versionMapOf(
+          F.col(AutoCdcReservedNames.cdcMetadataColName)).as("vm")),
+      expectedAnswer = Row(
+        null,
+        Map(Scd2VersionMap.encodePath(Seq("removed")) -> expectedAuthorship))
+    )
+  }
+
+  test("preprocessMicrobatch applies ignore-null to a reductively removed nested field") {
+    val batchSchema = new StructType()
+      .add("id", IntegerType)
+      .add("value", new StructType().add("a", IntegerType))
+      .add("seq", LongType)
+    val targetUserSchema = new StructType()
+      .add("id", IntegerType)
+      .add("value", new StructType()
+        .add("a", IntegerType)
+        .add("removed", StringType))
+    val batch = microbatchOf(batchSchema)(Row(1, Row(1), 10L))
+    val processor = Scd2BatchProcessor(
+      changeArgs = ChangeArgs(
+        keys = Seq(UnqualifiedColumnName("id")),
+        sequencing = F.col("seq"),
+        storedAsScdType = ScdType.Type2,
+        columnSelection = Some(ColumnSelection.ExcludeColumns(
+          Seq(UnqualifiedColumnName("seq")))),
+        ignoreNullSelection =
+          Some(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("value"))))
+      ),
+      resolvedSequencingType = LongType
+    )
+
+    val result = preprocessMicrobatch(processor, batch, Some(targetUserSchema))
+    checkAnswer(
+      df = result.select(
+        F.col("value"),
+        Scd2BatchProcessor.versionMapOf(
+          F.col(AutoCdcReservedNames.cdcMetadataColName)).as("vm")),
+      expectedAnswer = Row(
+        Row(1, null),
+        Map(Scd2VersionMap.encodePath(Seq("value", "removed")) -> false))
+    )
+  }
+
+  test("version-map key spelling matches the preprocessed microbatch and target") {
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+      val batchSchema = new StructType()
+        .add("id", IntegerType)
+        .add("Value", StringType)
+        .add("seq", LongType)
+      val targetUserSchema = new StructType()
+        .add("id", IntegerType)
+        .add("value", StringType)
+      val batch = microbatchOf(batchSchema)(Row(1, null, 10L))
+      val processor = Scd2BatchProcessor(
+        changeArgs = ChangeArgs(
+          keys = Seq(UnqualifiedColumnName("id")),
+          sequencing = F.col("seq"),
+          storedAsScdType = ScdType.Type2,
+          columnSelection = Some(ColumnSelection.ExcludeColumns(
+            Seq(UnqualifiedColumnName("seq")))),
+          ignoreNullSelection =
+            Some(ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("Value"))))
+        ),
+        resolvedSequencingType = LongType
+      )
+
+      val result = preprocessMicrobatch(processor, batch, Some(targetUserSchema))
+      val sourceValueName = batchSchema.fields(1).name
+      val preprocessedValueName = result.schema.fields(1).name
+      val targetValueName = targetUserSchema.fields(1).name
+      assert(sourceValueName == "Value")
+      assert(preprocessedValueName == targetValueName)
+      assert(preprocessedValueName == "value")
+
+      val expectedVersionMapKey = Scd2VersionMap.encodePath(Seq(preprocessedValueName))
+      checkAnswer(
+        df = result.select(Scd2BatchProcessor.versionMapOf(
+          F.col(AutoCdcReservedNames.cdcMetadataColName)).as("vm")),
+        expectedAnswer = Row(Map(expectedVersionMapKey -> false))
+      )
+    }
+  }
+
+  test("preprocessMicrobatch leaves version map null for all rows when ignore null is off") {
+    val schema = new StructType()
+      .add("id", IntegerType)
+      .add("value", StringType)
+      .add("seq", LongType)
+      .add("is_delete", BooleanType)
+
+    val batch = microbatchOf(schema)(
+      Row(1, null, 10L, false),
+      Row(1, null, 20L, true)
+    )
+
+    val processor = Scd2BatchProcessor(
+      changeArgs = ChangeArgs(
+        keys = Seq(UnqualifiedColumnName("id")),
+        sequencing = F.col("seq"),
+        storedAsScdType = ScdType.Type2,
+        deleteCondition = Some(F.col("is_delete")),
+        // None ignore-null selection should be treated as ignore-null off.
+        ignoreNullSelection = None
+      ),
+      resolvedSequencingType = LongType
+    )
+
+    val result = preprocessMicrobatch(processor, batch)
+
+    val versionMaps = result.select(
+      Scd2BatchProcessor.versionMapOf(
+        F.col(AutoCdcReservedNames.cdcMetadataColName)
+      ).as("vm")
+    )
+
+    // ignoreNullSelection is None -> version map is null on every row.
+    checkAnswer(
+      df = versionMaps,
+      expectedAnswer = Seq(Row(null), Row(null))
+    )
+  }
+
   // =============== computeMinimumSequencePerKey tests ===============
 
   test("computeMinimumSequencePerKey returns one row per distinct key and aggregates across " +
@@ -774,7 +1117,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
       Row(2, 40L, true)    // delete - smallest sequence for key=2
     )
 
-    val preprocessed = processor.preprocessMicrobatch(raw)
+    val preprocessed = preprocessMicrobatch(processor, raw)
     val result = processor.computeMinimumSequencePerKey(preprocessed)
 
     assert(result.schema.fieldNames.toSeq == Seq(
@@ -808,7 +1151,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
       Row("EU", 1, 30L)
     )
 
-    val preprocessed = processor.preprocessMicrobatch(raw)
+    val preprocessed = preprocessMicrobatch(processor, raw)
     val result = processor.computeMinimumSequencePerKey(preprocessed)
 
     assert(result.schema.fieldNames.toSeq == Seq(
@@ -832,7 +1175,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
     val processor = processorWithKeys(keys = Seq("id"))
 
     val raw = microbatchOf(schema)()
-    val preprocessed = processor.preprocessMicrobatch(raw)
+    val preprocessed = preprocessMicrobatch(processor, raw)
     val result = processor.computeMinimumSequencePerKey(preprocessed)
 
     assert(result.collect().isEmpty)
@@ -853,7 +1196,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
       Row(1, 30L),
       Row(1, 10L)
     )
-    val preprocessed = processor.preprocessMicrobatch(raw)
+    val preprocessed = preprocessMicrobatch(processor, raw)
     val result = processor.computeMinimumSequencePerKey(preprocessed)
 
     assert(result.schema.fieldNames.toSeq == Seq(
@@ -2175,7 +2518,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
       Row(1, 10L, "alice", "inactive")
     )
 
-    val result = processor.reconcileStartAndEndAt(processor.preprocessMicrobatch(df))
+    val result = processor.reconcileStartAndEndAt(preprocessMicrobatch(processor, df))
 
     checkAnswer(
       df = result,
@@ -2209,7 +2552,7 @@ class Scd2BatchProcessorSuite extends QueryTest with SharedSparkSession {
     )
 
     val ex = intercept[AnalysisException] {
-      processor.reconcileStartAndEndAt(processor.preprocessMicrobatch(df))
+      processor.reconcileStartAndEndAt(preprocessMicrobatch(processor, df))
     }
     assert(ex.getCondition == "AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA")
   }
diff --git a/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala
new file mode 100644
index 0000000000000..fa4a606e2fa21
--- /dev/null
+++ b/sql/pipelines/src/test/scala/org/apache/spark/sql/pipelines/autocdc/Scd2VersionMapSuite.scala
@@ -0,0 +1,447 @@
+/*
+ * 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.pipelines.autocdc
+
+import org.json4s.JsonAST.{JArray, JString}
+import org.json4s.jackson.JsonMethods.parse
+
+import org.apache.spark.sql.{functions => F, AnalysisException, QueryTest, Row}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+
+class Scd2VersionMapSuite extends QueryTest with SharedSparkSession {
+
+  private def resolver = spark.sessionState.conf.resolver
+
+  // ---- Schema helpers ----
+
+  private val flatSchema = new StructType()
+    .add("a", IntegerType)
+    .add("b", StringType)
+    .add("c", DoubleType)
+
+  private val nestedSchema = new StructType()
+    .add("x", IntegerType)
+    .add("address", new StructType()
+      .add("city", StringType)
+      .add("zip", IntegerType))
+
+  private val deeplyNestedSchema = new StructType()
+    .add("top", new StructType()
+      .add("mid", new StructType()
+        .add("leaf", StringType)))
+
+  private val arrayAndMapSchema = new StructType()
+    .add("tags", ArrayType(StringType))
+    .add("props", MapType(StringType, IntegerType))
+    .add("plain", IntegerType)
+
+  // Schemas with special-char leaves nested inside a normal top-level struct, so that
+  // ColumnSelection (which operates on top-level field names) can still select them.
+  private val specialCharSchema = new StructType()
+    .add("normal", IntegerType)
+    .add("wrapper", new StructType()
+      .add("has space", StringType))
+
+  private val periodInNameSchema = new StructType()
+    .add("wrapper", new StructType()
+      .add("a.b", IntegerType))
+    .add("c", StringType)
+
+  private val hyphenInNameSchema = new StructType()
+    .add("wrapper", new StructType()
+      .add("col-one", IntegerType))
+    .add("col_two", StringType)
+
+  // ---- Row helper ----
+
+  private def singleRow(schema: StructType)(values: Any*) =
+    spark.createDataFrame(
+      spark.sparkContext.parallelize(Seq(Row.fromSeq(values))), schema)
+
+  private def encodedPath(path: String*): String = Scd2VersionMap.encodePath(path)
+
+  // =========================================================================
+  // encodePath
+  // =========================================================================
+
+  test("encodePath - writes name parts as a compact JSON array") {
+    assert(Scd2VersionMap.encodePath(Seq("a")) === """["a"]""")
+    assert(Scd2VersionMap.encodePath(Seq("address", "city")) ===
+      """["address","city"]""")
+  }
+
+  test("encodePath - distinguishes nested paths from names containing periods") {
+    assert(Scd2VersionMap.encodePath(Seq("a", "b")) === """["a","b"]""")
+    assert(Scd2VersionMap.encodePath(Seq("a.b")) === """["a.b"]""")
+  }
+
+  test("encodePath - escapes JSON delimiters and control characters") {
+    val encoded = Scd2VersionMap.encodePath(
+      Seq("quote\"", "back\\slash", "null" + 0.toChar + "byte"))
+    assert(encoded === "[\"quote\\\"\",\"back\\\\slash\",\"null\\" + "u0000byte\"]")
+  }
+
+  test("encodePath - round trips arbitrary name parts") {
+    val path = Seq("", "a.b", "has space", "back`tick", "quote\"", "back\\slash")
+    assert(parse(Scd2VersionMap.encodePath(path)) === JArray(path.map(JString(_)).toList))
+  }
+
+  // =========================================================================
+  // mapType
+  // =========================================================================
+
+  test("mapType is Map(String, Boolean)") {
+    assert(Scd2VersionMap.mapType ===
+      MapType(StringType, BooleanType, valueContainsNull = false))
+  }
+
+  // =========================================================================
+  // buildVersionMap: version map contract cases
+  // =========================================================================
+
+  // Contract case 1: null in event + not part of ignore-null -> (column, true) = authored null
+  test("contract case 1 - null leaf not in ignore-null selection is authored (true)") {
+    val df = singleRow(flatSchema)(null, "hello", 2.0)
+    val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("b")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm"))
+    // a is null + not in ignore-null -> true (authored null)
+    checkAnswer(result, Row(Map(encodedPath("a") -> true)))
+  }
+
+  // Contract case 2: null in event + part of ignore-null -> (column, false) = unauthored null
+  test("contract case 2 - null leaf in ignore-null selection is declined (false)") {
+    val df = singleRow(flatSchema)(null, "hello", 2.0)
+    val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("a")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm"))
+    // a is null + in ignore-null -> false (declined)
+    checkAnswer(result, Row(Map(encodedPath("a") -> false)))
+  }
+
+  // Contract case 3: column not in the event schema (schema evolution later adds it with null).
+  // At ingest time this manifests as no entry for the column. We verify absence by constructing
+  // a narrower schema that omits the column.
+  test("contract case 3 - column absent from schema has no entry (schema evolution)") {
+    // Simulate ingest with a narrower schema that does not yet include column "b".
+    val narrowSchema = new StructType().add("a", IntegerType)
+    val df = singleRow(narrowSchema)(null)
+    val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("a")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(narrowSchema, selection, resolver).as("vm"))
+    // Only "a" appears (declined); a future column "b" added by schema evolution has no entry.
+    checkAnswer(result, Row(Map(encodedPath("a") -> false)))
+  }
+
+  // Non-null values are always considered authored and produce no entry.
+  test("contract - non-null values produce no entry regardless of ignore-null membership") {
+    val df = singleRow(flatSchema)(1, "hello", 2.0)
+    val selection = ColumnSelection.IncludeColumns(
+      Seq(UnqualifiedColumnName("a"), UnqualifiedColumnName("b"), UnqualifiedColumnName("c")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm"))
+    checkAnswer(result, Row(Map.empty[String, Boolean]))
+  }
+
+  // =========================================================================
+  // buildVersionMap: flat schema variations
+  // =========================================================================
+
+  test("flat schema - all null, none in ignore-null -> all authored (true)") {
+    val df = singleRow(flatSchema)(null, null, null)
+    val selection = ColumnSelection.ExcludeColumns(
+      Seq(UnqualifiedColumnName("a"), UnqualifiedColumnName("b"), UnqualifiedColumnName("c")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm"))
+    checkAnswer(result, Row(Map(
+      encodedPath("a") -> true,
+      encodedPath("b") -> true,
+      encodedPath("c") -> true)))
+  }
+
+  test("flat schema - all null, all in ignore-null -> all declined (false)") {
+    val df = singleRow(flatSchema)(null, null, null)
+    val selection = ColumnSelection.IncludeColumns(
+      Seq(UnqualifiedColumnName("a"), UnqualifiedColumnName("b"), UnqualifiedColumnName("c")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm"))
+    checkAnswer(result, Row(Map(
+      encodedPath("a") -> false,
+      encodedPath("b") -> false,
+      encodedPath("c") -> false)))
+  }
+
+  test("flat schema - mixed nulls with partial ignore-null") {
+    // a=null (ignore-null), b="hi" (non-null), c=null (not ignore-null)
+    val df = singleRow(flatSchema)(null, "hi", null)
+    val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("a")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm"))
+    checkAnswer(result, Row(Map(encodedPath("a") -> false, encodedPath("c") -> true)))
+  }
+
+  test("flat schema - multiple rows produce independent per-row maps") {
+    val df = spark.createDataFrame(
+      spark.sparkContext.parallelize(Seq(
+        Row(1, null, 3.0),
+        Row(null, "x", null))),
+      flatSchema)
+    val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("b")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm"))
+    checkAnswer(result, Seq(
+      Row(Map(encodedPath("b") -> false)), // b=null + ignore-null
+      // a,c=null + not ignore-null
+      Row(Map(
+        encodedPath("a") -> true,
+        encodedPath("c") -> true))))
+  }
+
+  // =========================================================================
+  // buildVersionMap: nested schemas
+  // =========================================================================
+
+  test("nested schema - null nested leaf tracked with multipart key") {
+    // x=1, address.city=null, address.zip=100
+    // ColumnSelection operates on top-level fields; include "address" struct.
+    val df = singleRow(nestedSchema)(1, Row(null, 100))
+    val selection = ColumnSelection.IncludeColumns(
+      Seq(UnqualifiedColumnName("address")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(nestedSchema, selection, resolver).as("vm"))
+    // address.city is null + ignore-null leaf -> false
+    // address.zip is null-checked but non-null -> no entry
+    checkAnswer(result, Row(Map(encodedPath("address", "city") -> false)))
+  }
+
+  test("nested schema - entire struct null makes all nested leaves null") {
+    val df = singleRow(nestedSchema)(null, null)
+    // Include "address" in ignore-null; "x" is not included.
+    val selection = ColumnSelection.IncludeColumns(
+      Seq(UnqualifiedColumnName("address")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(nestedSchema, selection, resolver).as("vm"))
+    checkAnswer(result, Row(Map(
+      encodedPath("x") -> true, // null + not ignore-null -> authored
+      encodedPath("address", "city") -> false, // null + ignore-null -> declined
+      encodedPath("address", "zip") -> false))) // null + ignore-null -> declined
+  }
+
+  test("deeply nested schema - three-level path tracked correctly") {
+    val df = singleRow(deeplyNestedSchema)(Row(Row(null)))
+    // Include the top-level "top" struct -> all leaves under it are ignore-null.
+    val selection = ColumnSelection.IncludeColumns(
+      Seq(UnqualifiedColumnName("top")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(deeplyNestedSchema, selection, resolver).as("vm"))
+    checkAnswer(result, Row(Map(encodedPath("top", "mid", "leaf") -> false)))
+  }
+
+  // =========================================================================
+  // buildVersionMap: arrays and maps treated as opaque leaves
+  // =========================================================================
+
+  test("array and map columns are tracked as opaque leaves") {
+    // tags=null, props=null, plain=null
+    val df = singleRow(arrayAndMapSchema)(null, null, null)
+    val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("tags")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(arrayAndMapSchema, selection, resolver).as("vm"))
+    checkAnswer(result, Row(Map(
+      encodedPath("tags") -> false, // null + ignore-null -> declined
+      encodedPath("props") -> true, // null + not ignore-null -> authored
+      encodedPath("plain") -> true))) // null + not ignore-null -> authored
+  }
+
+  // =========================================================================
+  // buildVersionMap: special character column names
+  // =========================================================================
+
+  test("column name with space is encoded in version map key") {
+    // normal=null, wrapper."has space"=null
+    val df = singleRow(specialCharSchema)(null, Row(null))
+    // Include "wrapper" struct -> its leaf "has space" is ignore-null.
+    val selection = ColumnSelection.IncludeColumns(
+      Seq(UnqualifiedColumnName("wrapper")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(
+        specialCharSchema, selection, resolver).as("vm"))
+    checkAnswer(result, Row(Map(
+      encodedPath("normal") -> true,
+      encodedPath("wrapper", "has space") -> false)))
+  }
+
+  test("column name with period is encoded in version map key") {
+    // wrapper."a.b"=null, c=null
+    val df = singleRow(periodInNameSchema)(Row(null), null)
+    val selection = ColumnSelection.IncludeColumns(
+      Seq(UnqualifiedColumnName("wrapper")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(
+        periodInNameSchema, selection, resolver).as("vm"))
+    checkAnswer(result, Row(Map(
+      encodedPath("wrapper", "a.b") -> false,
+      encodedPath("c") -> true)))
+  }
+
+  test("column name with hyphen is encoded in version map key") {
+    // wrapper."col-one"=null, col_two=null
+    val df = singleRow(hyphenInNameSchema)(Row(null), null)
+    val selection = ColumnSelection.IncludeColumns(
+      Seq(UnqualifiedColumnName("wrapper")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(
+        hyphenInNameSchema, selection, resolver).as("vm"))
+    checkAnswer(result, Row(Map(
+      encodedPath("wrapper", "col-one") -> false,
+      encodedPath("col_two") -> true)))
+  }
+
+  test("special-character path parts survive version map encoding") {
+    val specialNames =
+      Seq("a.b", "has space", "back`tick", "quote\"", "back\\slash", "null" + 0.toChar + "byte")
+    val schema = new StructType()
+      .add("wrapper", StructType(specialNames.map(StructField(_, StringType))))
+    val df = singleRow(schema)(Row.fromSeq(Seq.fill[Any](specialNames.size)(null)))
+    val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("wrapper")))
+
+    val encodedKeys = df
+      .select(F.map_keys(Scd2VersionMap.buildVersionMap(schema, selection, resolver)))
+      .head()
+      .getSeq[String](0)
+    val decodedPaths = encodedKeys.map { encodedKey =>
+      parse(encodedKey) match {
+        case JArray(parts) =>
+          parts.map {
+            case JString(part) => part
+            case other => fail(s"Expected a JSON string path part, but found $other")
+          }
+        case other => fail(s"Expected a JSON array version map key, but found $other")
+      }
+    }
+
+    val expectedPaths = specialNames.map(name => Seq("wrapper", name)).toSet
+    assert(decodedPaths.map(_.toSeq).toSet === expectedPaths)
+  }
+
+  // =========================================================================
+  // buildVersionMap: empty schema
+  // =========================================================================
+
+  test("empty schema produces an empty version map") {
+    val emptySchema = new StructType()
+    val df = singleRow(emptySchema)()
+    val selection = ColumnSelection.ExcludeColumns(Seq.empty)
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(emptySchema, selection, resolver).as("vm"))
+    checkAnswer(result, Row(Map.empty[String, Boolean]))
+  }
+
+  // =========================================================================
+  // buildVersionMap: ExcludeColumns variations
+  // =========================================================================
+
+  test("ExcludeColumns - excluded column is NOT ignore-null, others are") {
+    val df = singleRow(flatSchema)(null, null, null)
+    // ExcludeColumns("b") means ignore-null = everything except b.
+    val selection = ColumnSelection.ExcludeColumns(Seq(UnqualifiedColumnName("b")))
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm"))
+    // a,c are in ignore-null -> declined (false); b is NOT -> authored (true).
+    checkAnswer(result, Row(Map(
+      encodedPath("a") -> false,
+      encodedPath("b") -> true,
+      encodedPath("c") -> false)))
+  }
+
+  test("ExcludeColumns - empty exclude list -> ignore-null covers all columns") {
+    val df = singleRow(flatSchema)(null, null, null)
+    val selection = ColumnSelection.ExcludeColumns(Seq.empty)
+    val result = df.select(
+      Scd2VersionMap.buildVersionMap(flatSchema, selection, resolver).as("vm"))
+    // Exclude nothing -> ignore-null applies to all columns -> all nulls are declined.
+    checkAnswer(result, Row(Map(
+      encodedPath("a") -> false,
+      encodedPath("b") -> false,
+      encodedPath("c") -> false)))
+  }
+
+
+  // =========================================================================
+  // buildVersionMap: case sensitivity
+  // =========================================================================
+
+  test("case-insensitive resolver matches ignore-null columns regardless of case") {
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "false") {
+      val df = singleRow(flatSchema)(null, null, null)
+      // Selection uses uppercase "A" but schema has lowercase "a".
+      val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("A")))
+      val caseInsensitiveResolver = spark.sessionState.conf.resolver
+      val result = df.select(
+        Scd2VersionMap.buildVersionMap(
+          flatSchema, selection, caseInsensitiveResolver).as("vm"))
+      checkAnswer(result, Row(Map(
+        encodedPath("a") -> false,
+        encodedPath("b") -> true,
+        encodedPath("c") -> true)))
+    }
+  }
+
+  test("case-sensitive resolver does not match differently-cased column names") {
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+      val caseSensitiveResolver = spark.sessionState.conf.resolver
+      val df = singleRow(flatSchema)(null, null, null)
+      // Selection uses uppercase "A" but schema has lowercase "a" -> "A" is not found.
+      val selection =
+        ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("A")))
+      checkError(
+        exception = intercept[AnalysisException] {
+          df.select(
+            Scd2VersionMap.buildVersionMap(
+              flatSchema, selection, caseSensitiveResolver).as("vm")
+          ).collect()
+        },
+        condition = "AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA",
+        sqlState = "42703",
+        parameters = Map(
+          "caseSensitivity" -> "case-sensitive",
+          "schemaName" -> "ignoreNullSelection",
+          "missingColumns" -> "A",
+          "availableColumns" -> "a, b, c"
+        )
+      )
+    }
+  }
+
+  test("case-sensitive resolver matches exact-case column names") {
+    withSQLConf(SQLConf.CASE_SENSITIVE.key -> "true") {
+      val caseSensitiveResolver = spark.sessionState.conf.resolver
+      val df = singleRow(flatSchema)(null, null, null)
+      val selection = ColumnSelection.IncludeColumns(Seq(UnqualifiedColumnName("a")))
+      val result = df.select(
+        Scd2VersionMap.buildVersionMap(
+          flatSchema, selection, caseSensitiveResolver).as("vm"))
+      checkAnswer(result, Row(Map(
+        encodedPath("a") -> false,
+        encodedPath("b") -> true,
+        encodedPath("c") -> true)))
+    }
+  }
+}