[UniForm] Refactor IcebergCompat (delta-io#4356)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-io/delta/blob/master/CONTRIBUTING.md
2. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  3. Be sure to keep the PR description updated to reflect all changes.
  4. Please write your PR title to summarize what this PR proposes.
5. If possible, provide a concise example to reproduce the issue for a
faster review.
6. If applicable, include the corresponding issue number in the PR title
and link it in the body.
-->

#### Which Delta project/connector is this regarding?
<!--
Please add the component selected below to the beginning of the pull
request title
For example: [Spark] Title of my pull request
-->

- [x] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description
This PR refactors existing IcebergCompat code and test cases.
<!--
- Describe what this PR changes.
- Describe why we need the change.
 
If this PR resolves an issue be sure to include "Resolves #XXX" to
correctly link and close the issue upon merge.
-->

## How was this patch tested?
UT/Integration
<!--
If tests were added, say they were added here. Please make sure to test
the changes thoroughly including negative and positive cases if
possible.
If the changes were tested in any way other than unit tests, please
clarify how you tested step by step (ideally copy and paste-able, so
that other reviewers can test and check, and descendants can verify in
the future).
If the changes were not tested, please explain why.
-->

## Does this PR introduce _any_ user-facing changes?
No
<!--
If yes, please clarify the previous behavior and the change this PR
proposes - provide the console output, description and/or an example to
show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change
compared to the released Delta Lake versions or within the unreleased
branches such as master.
If no, write 'No'.
-->

Author: harperjiang

iceberg/src/test/scala/org/apache/spark/sql/delta/uniform/UniversalFormatSuite.scala

@@ -56,114 +56,14 @@ trait UniversalFormatSuiteUtilsBase
     .filterNot { case (r, w) => w < 7 && r >= 3 }
 }
 
-/** Contains shared tests for both IcebergCompatV1 and IcebergCompatV2 suites. */
-trait UniversalFormatSuiteTestBase extends UniversalFormatSuiteUtilsBase {
-  private def loadIcebergTable(
-      spark: SparkSession,
-      id: TableIdentifier): Option[shadedForDelta.org.apache.iceberg.Table] = {
-    val deltaLog = DeltaLog.forTable(spark, id)
-    val catalogTable = spark.sessionState.catalog.getTableMetadata(id)
-    val hiveCatalog = IcebergTransactionUtils
-      .createHiveCatalog(deltaLog.newDeltaHadoopConf())
-    val icebergTableId = IcebergTransactionUtils
-      .convertSparkTableIdentifierToIcebergHive(catalogTable.identifier)
-    if (hiveCatalog.tableExists(icebergTableId)) {
-      Some(hiveCatalog.loadTable(icebergTableId))
-    } else {
-      None
-    }
-  }
-
-  protected def getCompatVersionOtherThan(version: Int): Int
-
-  protected def getCompatVersionsOtherThan(version: Int): Seq[Int]
-
-  test("CTAS new UniForm (Iceberg) table without manually enabling column mapping") {
-    // These are the versions with column mapping enabled
-    allReaderWriterVersions.foreach { case (r, w) =>
-      withTempTableAndDir { case (from_id, from_loc) =>
-        withTempTableAndDir { case (to_id1, to_loc1) =>
-          withTempTableAndDir { case (to_id2, to_loc2) =>
-            executeSql(
-              s"""
-                 |CREATE TABLE $from_id (PID INT, PCODE INT) USING DELTA LOCATION $from_loc
-                 | TBLPROPERTIES (
-                 |  'delta.minReaderVersion' = $r,
-                 |  'delta.minWriterVersion' = $w
-                 |)""".stripMargin)
-            executeSql(
-              s"""
-                 |INSERT INTO TABLE $from_id (PID, PCODE)
-                 | VALUES (1,2),(2,3),(3,4),(4,5),(5,6),(6,7),(7,8)""".stripMargin)
-            executeSql(
-              s"""
-                 |CREATE TABLE $to_id1 USING DELTA LOCATION $to_loc1 TBLPROPERTIES (
-                 |  'delta.universalFormat.enabledFormats' = 'iceberg',
-                 |  'delta.enableIcebergCompatV$compatVersion' = 'true',
-                 |  'delta.minReaderVersion' = $r,
-                 |  'delta.minWriterVersion' = $w
-                 |) AS SELECT * FROM $from_id""".stripMargin)
-            executeSql(
-              s"""
-                 |CREATE TABLE $to_id2 USING DELTA LOCATION $to_loc2 TBLPROPERTIES (
-                 |  'delta.universalFormat.enabledFormats' = 'iceberg',
-                 |  'delta.columnMapping.mode' = 'name',
-                 |  'delta.enableIcebergCompatV$compatVersion' = 'true',
-                 |  'delta.minReaderVersion' = $r,
-                 |  'delta.minWriterVersion' = $w
-                 |) AS SELECT * FROM $from_id""".stripMargin)
-
-            val icebergTable1 = loadIcebergTable(spark, new TableIdentifier(to_id1))
-            if (icebergTable1.isDefined) {
-              assert(icebergTable1.get.schema().asStruct().fields().toArray.length == 2)
-            }
-
-            val expected1: Array[Row] = (1 to 7).map { i => Row(i, i + 1) }.toArray
-            readAndVerify(to_id1, "PID, PCODE", "PID", expected1)
-
-            val icebergTable2 = loadIcebergTable(spark, new TableIdentifier(to_id2))
-            if (icebergTable2.isDefined) {
-              assert(icebergTable2.get.schema().asStruct().fields().toArray.length == 2)
-            }
-
-            val expected2: Array[Row] = (1 to 7).map { i => Row(i, i + 1) }.toArray
-            readAndVerify(to_id2, "PID, PCODE", "PID", expected2)
-          }
-        }
-      }
-    }
-  }
-
-  test("REORG TABLE: command does not support where clause") {
-    withTempTableAndDir { case (id, loc) =>
-      val anotherCompatVersion = getCompatVersionOtherThan(compatVersion)
-      executeSql(s"""
-                    | CREATE TABLE $id (ID INT) USING DELTA LOCATION $loc TBLPROPERTIES (
-                    |  'delta.universalFormat.enabledFormats' = 'iceberg',
-                    |  'delta.enableIcebergCompatV$anotherCompatVersion' = 'true'
-                    |)""".stripMargin)
-      val e = intercept[ParseException] {
-        spark.sessionState.sqlParser.parsePlan(
-          s"""
-             | REORG TABLE $id
-             | WHERE ID > 0
-             | APPLY (UPGRADE UNIFORM (ICEBERGCOMPATVERSION = $compatVersion))
-         """.stripMargin).asInstanceOf[DeltaReorgTableCommand]
-      }
-      assert(e.getErrorClass === "PARSE_SYNTAX_ERROR")
-      assert(e.getMessage.contains("Syntax error at or near 'REORG'"))
-    }
-  }
-}
-
 class UniversalFormatSuite
     extends UniversalFormatMiscSuiteBase
     with UniversalFormatSuiteUtilsBase
 
 class UniFormWithIcebergCompatV1Suite
-    extends UniversalFormatSuiteTestBase
+    extends UniversalFormatSuiteUtilsBase
     with UniFormWithIcebergCompatV1SuiteBase
 
 class UniFormWithIcebergCompatV2Suite
-    extends UniversalFormatSuiteTestBase
+    extends UniversalFormatSuiteUtilsBase
     with UniFormWithIcebergCompatV2SuiteBase

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

@@ -491,7 +491,8 @@ trait DeltaColumnMappingBase extends DeltaLogging {
         field.copy(metadata = builder.build())
       })
 
-    val (finalSchema, newMaxId) = if (IcebergCompatV2.isEnabled(newMetadata)) {
+    // Starting from IcebergCompatV2, we require writing field-id for List/Map nested fields
+    val (finalSchema, newMaxId) = if (IcebergCompat.isGeqEnabled(newMetadata, 2)) {
       rewriteFieldIdsForIceberg(newSchema, maxId)
     } else {
       (newSchema, maxId)

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

@@ -269,13 +269,13 @@ case class DeltaParquetFileFormat(
        dataSchema: StructType): OutputWriterFactory = {
     val factory = super.prepareWrite(sparkSession, job, options, dataSchema)
     val conf = ContextUtil.getConfiguration(job)
-    // Always write timestamp as TIMESTAMP_MICROS for Iceberg compat based on Iceberg spec
-    if (IcebergCompatV1.isEnabled(metadata) || IcebergCompatV2.isEnabled(metadata)) {
+    // Always write timestamp as TIMESTAMP_MICROS for IcebergCompat based on Iceberg spec
+    if (IcebergCompat.isAnyEnabled(metadata)) {
       conf.set(SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key,
         SQLConf.ParquetOutputTimestampType.TIMESTAMP_MICROS.toString)
     }
-    if (IcebergCompatV2.isEnabled(metadata)) {
-      // For Uniform with IcebergCompatV2, we need to write nested field IDs for list and map
+    if (IcebergCompat.isGeqEnabled(metadata, 2)) {
+      // Starting from IcebergCompatV2, we need to write nested field IDs for list and map
       // types to the parquet schema. Spark currently does not support it so we hook in our
       // own write support class.
       ParquetOutputFormat.setWriteSupportClass(job, classOf[DeltaParquetWriteSupport])

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

@@ -244,21 +244,18 @@ case class IcebergCompatVersionBase(knownVersions: Set[IcebergCompatBase]) {
       .map{ _.version }
 
   /**
-   * Get the DeltaConfig for the given IcebergCompat version. If version is not valid,
+   * Get the IcebergCompat by version. If version is not valid,
    * throw an exception.
-   * @return the DeltaConfig for the given version. E.g.,
-   *         [[DeltaConfigs.ICEBERG_COMPAT_V1_ENABLED]] for version 1.
+   * @return the IcebergCompatVx object
    */
-  def getConfigForVersion(version: Int): DeltaConfig[Option[Boolean]] = {
+  def getForVersion(version: Int): IcebergCompatBase =
     knownVersions
       .find(_.version == version)
-      .map(_.config)
       .getOrElse(
         throw DeltaErrors.icebergCompatVersionNotSupportedException(
           version, knownVersions.size
         )
     )
-  }
 
   /**
    * @return any enabled IcebergCompat in the conf
@@ -268,17 +265,22 @@ case class IcebergCompatVersionBase(knownVersions: Set[IcebergCompatBase]) {
       conf.getOrElse[String](compat.config.key, "false").toBoolean
     }
 
+  def anyEnabled(metadata: Metadata): Option[IcebergCompatBase] =
+    knownVersions.find { _.config.fromMetaData(metadata).getOrElse(false) }
+
   /**
    * @return true if any version of IcebergCompat is enabled
    */
   def isAnyEnabled(conf: Map[String, String]): Boolean = anyEnabled(conf).nonEmpty
 
-  /**
-   * @return true if any version of IcebergCompat is enabled
-   */
   def isAnyEnabled(metadata: Metadata): Boolean =
     knownVersions.exists { _.config.fromMetaData(metadata).getOrElse(false) }
 
+  /**
+   * @return true if a CompatVx greater or eq to the required version is enabled
+   */
+  def isGeqEnabled(metadata: Metadata, requiredVersion: Int): Boolean =
+    anyEnabled(metadata).exists(_.version >= requiredVersion)
   /**
    * @return true if any version of IcebergCompat is enabled, and is incompatible
    *         with the given table feature
@@ -291,8 +293,9 @@ case class IcebergCompatVersionBase(knownVersions: Set[IcebergCompatBase]) {
     }
 }
 
-object IcebergCompat
-  extends IcebergCompatVersionBase(Set(IcebergCompatV1, IcebergCompatV2)) with DeltaLogging
+object IcebergCompat extends IcebergCompatVersionBase(
+    Set(IcebergCompatV1, IcebergCompatV2)
+  ) with DeltaLogging
 
 
 

spark/src/main/scala/org/apache/spark/sql/delta/commands/ReorgTableForUpgradeUniformHelper.scala

@@ -19,8 +19,8 @@ package org.apache.spark.sql.delta.commands
 // scalastyle:off import.ordering.noEmptyLine
 import scala.util.control.NonFatal
 
-import org.apache.spark.sql.delta.{DeltaConfig, DeltaConfigs, DeltaErrors, DeltaOperations, Snapshot}
-import org.apache.spark.sql.delta.IcebergCompat.{getConfigForVersion, getEnabledVersion}
+import org.apache.spark.sql.delta.{DeletionVectorsTableFeature, DeltaConfigs, DeltaErrors, DeltaOperations, IcebergCompatBase, Snapshot}
+import org.apache.spark.sql.delta.IcebergCompat.{getEnabledVersion, getForVersion}
 import org.apache.spark.sql.delta.UniversalFormat.{icebergEnabled, ICEBERG_FORMAT}
 import org.apache.spark.sql.delta.actions.{AddFile, Protocol}
 import org.apache.spark.sql.delta.catalog.DeltaTableV2
@@ -37,15 +37,15 @@ import org.apache.spark.sql.functions.col
  */
 trait ReorgTableForUpgradeUniformHelper extends DeltaLogging {
 
-  private val versionChangesRequireRewrite: Map[Int, Set[Int]] =
-    Map(0 -> Set(2), 1 -> Set(2), 2 -> Set(2))
+  private val rewriteCheckTable: Map[Int, Set[Int]] =
+    Map(0 -> Set(2, 3), 1 -> Set(2, 3), 2 -> Set(2, 3), 3 -> Set(2, 3))
 
   /**
-   * Helper function to check if the table data may need to be rewritten to be iceberg compatible.
-   * Only if not all addFiles has the tag, Rewriting would be performed.
+   * Check if the given pair of (old_version, new_version) should trigger a rewrite check.
+   * NOTE: Actual rewrite only happens when not all addFiles has tags with newVersion.
    */
-  private def reorgMayNeedRewrite(oldVersion: Int, newVersion: Int): Boolean = {
-    versionChangesRequireRewrite.getOrElse(oldVersion, Set.empty[Int]).contains(newVersion)
+  private def shallCheckRewrite(oldVersion: Int, newVersion: Int): Boolean = {
+    rewriteCheckTable.getOrElse(oldVersion, Set.empty[Int]).contains(newVersion)
   }
 
   /**
@@ -54,42 +54,38 @@ trait ReorgTableForUpgradeUniformHelper extends DeltaLogging {
   def optimizeByReorg(sparkSession: SparkSession): Seq[Row]
 
   /**
-   * Helper function to update the table icebergCompat properties.
-   * We can not use AlterTableSetPropertiesDeltaCommand here because we don't allow customer to
-   * change icebergCompatVersion by using Alter Table command.
+   * Enable the new IcebergCompat on the table by updating table conf.
    */
   private def enableIcebergCompat(
-      target: DeltaTableV2,
-      currIcebergCompatVersionOpt: Option[Int],
-      targetVersionDeltaConfig: DeltaConfig[Option[Boolean]]): Unit = {
-    var enableIcebergCompatConf = Map(
-      targetVersionDeltaConfig.key -> "true",
-      DeltaConfigs.ENABLE_DELETION_VECTORS_CREATION.key -> "false",
-      DeltaConfigs.COLUMN_MAPPING_MODE.key -> "name"
-    )
-    if (currIcebergCompatVersionOpt.nonEmpty) {
-      val currIcebergCompatVersionDeltaConfig = getConfigForVersion(
-        currIcebergCompatVersionOpt.get)
-      enableIcebergCompatConf ++= Map(currIcebergCompatVersionDeltaConfig.key -> "false")
+      table: DeltaTableV2,
+      currentCompatVersion: Option[Int],
+      compatToEnable: IcebergCompatBase): Unit = {
+    var newConf: Map[String, String] = Map(
+      compatToEnable.config.key -> "true",
+      DeltaConfigs.COLUMN_MAPPING_MODE.key -> "name") ++
+      currentCompatVersion.map(getForVersion(_).config.key -> "false") // Disable old IcebergCompat
+
+    if (compatToEnable.incompatibleTableFeatures.contains(DeletionVectorsTableFeature)) {
+      newConf += DeltaConfigs.ENABLE_DELETION_VECTORS_CREATION.key -> "false"
     }
 
-    val alterConfTxn = target.startTransaction()
+    val alterConfTxn = table.startTransaction()
 
     if (alterConfTxn.protocol.minWriterVersion < 7) {
-      enableIcebergCompatConf += Protocol.MIN_WRITER_VERSION_PROP -> "7"
+      newConf += Protocol.MIN_WRITER_VERSION_PROP -> "7"
     }
     if (alterConfTxn.protocol.minReaderVersion < 3) {
-      enableIcebergCompatConf += Protocol.MIN_READER_VERSION_PROP -> "3"
+      newConf += Protocol.MIN_READER_VERSION_PROP -> "3"
     }
 
     val metadata = alterConfTxn.metadata
     val newMetadata = metadata.copy(
       description = metadata.description,
-      configuration = metadata.configuration ++ enableIcebergCompatConf)
+      configuration = metadata.configuration ++ newConf)
     alterConfTxn.updateMetadata(newMetadata)
     alterConfTxn.commit(
       Nil,
-      DeltaOperations.UpgradeUniformProperties(enableIcebergCompatConf)
+      DeltaOperations.UpgradeUniformProperties(newConf)
     )
   }
 
@@ -131,13 +127,13 @@ trait ReorgTableForUpgradeUniformHelper extends DeltaLogging {
    *
    * * There are six possible write combinations:
    * | CurrentIcebergCompatVersion | TargetIcebergCompatVersion | Required steps|
-   * | --------------- | --------------- | --------------- |
-   * |      None       |         1       |   1, 3          |
-   * |      None       |         2       |   1, 2, 3       |
-   * |      1          |         1       |   3             |
-   * |      1          |         2       |   1, 2, 3       |
-   * |      2          |         1       |   1, 3          |
-   * |      2          |         2       |   2, 3          |
+   * | --------------------------- | -------------------------- | ------------- |
+   * |      None                   |         1                  |   1, 3        |
+   * |      None                   |         2+                 |   1, 2, 3     |
+   * |      1                      |         1                  |   3           |
+   * |      1                      |         2+                 |   1, 2, 3     |
+   * |      2+                     |         1                  |   1, 3        |
+   * |      2+                     |         2+                 |   2, 3        |
    */
   private def doRewrite(
       target: DeltaTableV2,
@@ -146,15 +142,14 @@ trait ReorgTableForUpgradeUniformHelper extends DeltaLogging {
 
     val snapshot = target.update()
     val currIcebergCompatVersionOpt = getEnabledVersion(snapshot.metadata)
-    val targetVersionDeltaConfig = getConfigForVersion(
-      targetIcebergCompatVersion)
-    val versionChangeMayNeedRewrite = reorgMayNeedRewrite(
+    val targetIcebergCompatObject = getForVersion(targetIcebergCompatVersion)
+    val mayNeedRewrite = shallCheckRewrite(
       currIcebergCompatVersionOpt.getOrElse(0), targetIcebergCompatVersion)
 
     // Step 1: Update the table properties to enable the target iceberg compat version
     val didUpdateIcebergCompatVersion =
       if (!currIcebergCompatVersionOpt.contains(targetIcebergCompatVersion)) {
-        enableIcebergCompat(target, currIcebergCompatVersionOpt, targetVersionDeltaConfig)
+        enableIcebergCompat(target, currIcebergCompatVersionOpt, targetIcebergCompatObject)
         logInfo(log"Update table ${MDC(DeltaLogKeys.TABLE_NAME, target.tableIdentifier)} " +
           log"to iceberg compat version = " +
           log"${MDC(DeltaLogKeys.VERSION, targetIcebergCompatVersion)} successfully.")
@@ -170,7 +165,7 @@ trait ReorgTableForUpgradeUniformHelper extends DeltaLogging {
     // The table needs to be rewritten if:
     //   1. The target iceberg compat version requires rewrite.
     //   2. Not all addFile have ICEBERG_COMPAT_VERSION=targetVersion tag
-    val (metricsOpt, didRewrite) = if (versionChangeMayNeedRewrite && !allAddFilesHaveTag) {
+    val (metricsOpt, didRewrite) = if (mayNeedRewrite && !allAddFilesHaveTag) {
       logInfo(log"Reorg Table ${MDC(DeltaLogKeys.TABLE_NAME, target.tableIdentifier)} to " +
         log"iceberg compat version = ${MDC(DeltaLogKeys.VERSION, targetIcebergCompatVersion)} " +
         log"need rewrite data files.")
@@ -191,7 +186,7 @@ trait ReorgTableForUpgradeUniformHelper extends DeltaLogging {
     val updatedSnapshot = target.deltaLog.update()
     val (numOfAddFiles, numOfAddFilesWithIcebergCompatTag) = getNumOfAddFiles(
       targetIcebergCompatVersion, target, updatedSnapshot)
-    if (versionChangeMayNeedRewrite && numOfAddFilesWithIcebergCompatTag != numOfAddFiles) {
+    if (mayNeedRewrite && numOfAddFilesWithIcebergCompatTag != numOfAddFiles) {
       throw DeltaErrors.icebergCompatReorgAddFileTagsMissingException(
         updatedSnapshot.version,
         targetIcebergCompatVersion,
@@ -215,7 +210,7 @@ trait ReorgTableForUpgradeUniformHelper extends DeltaLogging {
       "targetIcebergCompatVersion" -> targetIcebergCompatVersion.toString,
       "metrics" -> metricsOpt.toString,
       "didUpdateIcebergCompatVersion" -> didUpdateIcebergCompatVersion.toString,
-      "needRewrite" -> versionChangeMayNeedRewrite.toString,
+      "needRewrite" -> mayNeedRewrite.toString,
       "didRewrite" -> didRewrite.toString,
       "numOfAddFilesBefore" -> numOfAddFilesBefore.toString,
       "numOfAddFilesWithIcebergCompatTagBefore" -> numOfAddFilesWithTagBefore.toString,

spark/src/main/scala/org/apache/spark/sql/delta/files/TransactionalWrite.scala

@@ -518,10 +518,13 @@ trait TransactionalWrite extends DeltaLogging { self: OptimisticTransactionImpl
     }
 
     // add [[AddFile.Tags.ICEBERG_COMPAT_VERSION.name]] tags to addFiles
-    if (IcebergCompatV2.isEnabled(metadata)) {
+    // starting from IcebergCompatV2
+    val enabledCompat = IcebergCompat.anyEnabled(metadata)
+    if (enabledCompat.exists(_.version >= 2)) {
       resultFiles = resultFiles.map { addFile =>
-        val tags = if (addFile.tags != null) addFile.tags else Map.empty[String, String]
-        addFile.copy(tags = tags + (AddFile.Tags.ICEBERG_COMPAT_VERSION.name -> "2"))
+        addFile.copy(tags = Option(addFile.tags).getOrElse(Map.empty[String, String]) +
+          (AddFile.Tags.ICEBERG_COMPAT_VERSION.name -> enabledCompat.get.version.toString)
+        )
       }
     }