Merge branch 'master' into timothy/delta-column-default-capability

Author: TimothyW553

iceberg/src/main/scala/org/apache/spark/sql/delta/icebergShaded/IcebergConverter.scala

@@ -28,11 +28,13 @@ import org.apache.spark.sql.delta.{CommittedTransaction, CurrentTransactionInfo,
 import org.apache.spark.sql.delta.DeltaOperations.OPTIMIZE_OPERATION_NAME
 import org.apache.spark.sql.delta.RowId.RowTrackingMetadataDomain
 import org.apache.spark.sql.delta.actions.{Action, AddFile, CommitInfo, DomainMetadata, FileAction, InMemoryLogReplay, Metadata, Protocol, RemoveFile}
+import org.apache.spark.sql.delta.catalog.DeltaTableV2
 import org.apache.spark.sql.delta.hooks.IcebergConverterHook
 import org.apache.spark.sql.delta.logging.DeltaLogKeys
 import org.apache.spark.sql.delta.metering.DeltaLogging
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.util.TransactionHelper
+import io.delta.storage.commit.{CommitFailedException => DeltaCommitFailedException}
 import org.apache.commons.lang3.exception.ExceptionUtils
 import org.apache.hadoop.fs.Path
 import shadedForDelta.org.apache.iceberg.{Table => IcebergTable, TableProperties}
@@ -44,6 +46,7 @@ import shadedForDelta.org.apache.iceberg.util.LocationUtil
 import org.apache.spark.internal.MDC
 import org.apache.spark.sql.{Dataset, SparkSession}
 import org.apache.spark.sql.catalyst.catalog.CatalogTable
+import org.apache.spark.sql.connector.catalog.{CatalogV2Util, Identifier, V1Table}
 
 object IcebergConverter {
 
@@ -217,13 +220,83 @@ class IcebergConverter
       (newMetadataPath, lastConvertedInfo.deltaVersionConverted)
     }
 
-  private def refreshCatalogTableIfNeeded(
+  protected def refreshCatalogTableIfNeeded(
       txnInfo: CurrentTransactionInfo,
       deltaAttemptVersion: Long,
       catalogTable: CatalogTable): CatalogTable = {
-    catalogTable
+    val lastDeltaVersionConvertedOpt =
+      catalogTable.storage.properties
+        .get(IcebergConstants.CATALOG_TABLE_ICEBERG_CONVERTED_DELTA_VERSION_PROP)
+        .map(_.toLong)
+    val needsRefreshCatalogTable =
+      deltaAttemptVersion > txnInfo.readSnapshot.version + 1 &&
+        UniversalFormat.icebergEnabled(txnInfo.metadata) &&
+        lastDeltaVersionConvertedOpt.nonEmpty
+    if (!needsRefreshCatalogTable) {
+      return catalogTable
+    }
+    val refreshedTable = reloadCatalogTable(catalogTable)
+    if (refreshedTable.storage.locationUri != catalogTable.storage.locationUri) {
+      throw new DeltaCommitFailedException(
+        false, // retryable
+        true, // conflict
+        s"Table location changed when refreshing catalogTable. " +
+          s"catalogTable location: ${catalogTable.storage.locationUri} " +
+          s"refreshedTable location: ${refreshedTable.storage.locationUri}")
+    }
+    val refreshedLastDeltaVersionConvertedOpt =
+      refreshedTable.storage.properties
+        .get(IcebergConstants.CATALOG_TABLE_ICEBERG_CONVERTED_DELTA_VERSION_PROP)
+        .map(_.toLong)
+    val oldBaseMetadataLocation =
+      catalogTable.storage.properties
+        .get(IcebergConstants.CATALOG_TABLE_ICEBERG_METADATA_LOCATION_PROP)
+    val newBaseMetadataLocation =
+      refreshedTable.storage.properties
+        .get(IcebergConstants.CATALOG_TABLE_ICEBERG_METADATA_LOCATION_PROP)
+    logInfo(s"refresh CatalogTable for UniForm for table ${catalogTable.identifier}: " +
+      s"oldLastDeltaVersionConvertedOpt=$lastDeltaVersionConvertedOpt " +
+      s"refreshedLastDeltaVersionConvertedOpt=$refreshedLastDeltaVersionConvertedOpt " +
+      s"oldBaseMetadataLocation=$oldBaseMetadataLocation " +
+      s"newBaseMetadataLocation=$newBaseMetadataLocation")
+    if (
+      refreshedLastDeltaVersionConvertedOpt.exists(_ >= deltaAttemptVersion)
+    ) {
+      throw new DeltaCommitFailedException(
+        true, // retryable
+        true, // conflict
+        s"Attempts to commit Delta version $deltaAttemptVersion while the last converted delta " +
+          s"version is already $refreshedLastDeltaVersionConvertedOpt")
+    }
+    refreshedTable
   }
 
+  /**
+   * Reloads a CatalogTable via the V2 catalog API
+   * The V1 SessionCatalog only reaches HMS and
+   * fails for UC tables whose schema doesn't exist there
+   */
+  protected def reloadCatalogTable(catalogTable: CatalogTable): CatalogTable = {
+    val catalog = catalogTable.identifier.catalog
+      .map(spark.sessionState.catalogManager.catalog)
+      .getOrElse(spark.sessionState.catalogManager.v2SessionCatalog)
+    val ident = Identifier.of(
+      catalogTable.identifier.database.toArray,
+      catalogTable.identifier.table)
+    CatalogV2Util.loadTable(catalog, ident)
+      .flatMap {
+        case dt: DeltaTableV2 => dt.catalogTable
+        case other => throw new DeltaCommitFailedException(
+          false, // retryable
+          false, // conflict
+          s"Expected DeltaTableV2 when reloading catalogTable for ${catalogTable.identifier}, " +
+            s"got ${other.getClass.getName}")
+      }
+      .getOrElse(throw new DeltaCommitFailedException(
+        false, // retryable
+        false, // conflict
+        s"Table ${catalogTable.identifier} not found when reloading catalogTable through $catalog"))
+  }
 
   /**
    * Reads the last converted Iceberg state from the catalogTable storage properties.

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

@@ -34,6 +34,16 @@ import org.apache.spark.sql.delta.util.JsonUtils
 import org.apache.spark.sql.test.SharedSparkSession
 
 class IcebergConverterForTest extends IcebergConverter {
+  /**
+   * Disable catalogTable refresh for this test as this test is not using
+   * UC so UniForm metadata won't be stored in catalogTable. This test
+   * needs to manually inject UniForm metadata into catalogTable
+   */
+  override protected def refreshCatalogTableIfNeeded(
+      txnInfo: CurrentTransactionInfo,
+      deltaAttemptVersion: Long,
+      catalogTable: CatalogTable): CatalogTable = catalogTable
+
   def convertSnapshotAndReturnMetadataPath(
       snapshotToConvert: Snapshot,
       catalogTable: CatalogTable): String = {

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

@@ -20,6 +20,7 @@ import java.util.{Collections, Optional, UUID}
 
 import scala.collection.JavaConverters._
 
+import com.databricks.spark.util.Log4jUsageLogger
 import io.delta.storage.commit.{CommitCoordinatorClient => JCommitCoordinatorClient}
 import io.delta.storage.commit.{TableIdentifier => UCTableIdentifier}
 import io.delta.storage.commit.actions.{AbstractMetadata, AbstractProtocol}
@@ -29,9 +30,9 @@ import org.apache.hadoop.fs.Path
 import org.apache.spark.{SparkConf, SparkSessionSwitch}
 import org.apache.spark.sql.{Row, SparkSession}
 import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.connector.catalog.{Identifier, Table}
-import org.apache.spark.sql.delta.{DeltaLog, IcebergConstants}
+import org.apache.spark.sql.catalyst.analysis.{FunctionRegistry, TableFunctionRegistry}
+import org.apache.spark.sql.catalyst.catalog.{CatalogTable, SessionCatalog}
+import org.apache.spark.sql.delta.{DeltaLog, DeltaOperations, DeltaTestUtils, IcebergConstants}
 import org.apache.spark.sql.delta.DeltaConfigs.{
   COORDINATED_COMMITS_COORDINATOR_CONF,
   COORDINATED_COMMITS_COORDINATOR_NAME
@@ -45,7 +46,6 @@ import org.apache.spark.sql.delta.coordinatedcommits.{
   InMemoryUCCommitCoordinator,
   UCCommitCoordinatorBuilder
 }
-import org.apache.spark.sql.delta.icebergShaded.IcebergConverter
 import org.apache.spark.sql.delta.sources.DeltaSQLConf
 import org.apache.spark.sql.delta.test.DeltaSQLCommandTest
 import org.apache.spark.sql.delta.uniform.hms.HMSTest
@@ -95,41 +95,54 @@ trait WriteDeltaHMSReadIceberg extends UniFormE2ETest
 }
 
 /**
- * A [[DeltaCatalog]] subclass that enriches [[CatalogTable]] with the last converted Iceberg
- * metadata from [[InMemoryUCCommitCoordinator]] before loading the table. This simulates what
- * the real UC REST catalog does: injecting
- * [[IcebergConstants.CATALOG_TABLE_ICEBERG_METADATA_LOCATION_PROP]]
- * and [[IcebergConstants.CATALOG_TABLE_ICEBERG_CONVERTED_DELTA_VERSION_PROP]] into
- * catalog storage properties, matching the UC REST path in [[UCDeltaCatalogClientImpl]].
+ * A [[SessionCatalog]] wrapper that overrides [[getTableMetadata]] to inject fresh Iceberg
+ * metadata from [[UCEnrichedSessionCatalog.currentCoordinator]] on every call. This mirrors what
+ * [[UCDeltaCatalogClientImpl.toV1Table]] does in production. Installed via reflection into
+ * both [[org.apache.spark.sql.internal.SessionState.catalog]] (for the V1 path used by
+ * [[org.apache.spark.sql.delta.icebergShaded.IcebergConverter.refreshCatalogTableIfNeeded]])
+ * and the [[org.apache.spark.sql.execution.datasources.v2.V2SessionCatalog]] delegate (for
+ * the V2 write path that goes through [[DeltaCatalog.loadTable]]). Both injections happen in
+ * [[WriteDeltaUCCCReadIceberg.beforeEach]].
  */
-class UCBackedDeltaCatalog extends DeltaCatalog {
-  override def loadCatalogTable(ident: Identifier, catalogTable: CatalogTable): Table = {
-    val forbiddenKeys = catalogTable.properties.keys.filter(_.startsWith("deltaUniformIceberg."))
+private class UCEnrichedSessionCatalog(
+    spark: SparkSession,
+    delegate: SessionCatalog,
+    fr: FunctionRegistry,
+    tfr: TableFunctionRegistry)
+  extends SessionCatalog(delegate.externalCatalog, fr, tfr) {
+
+  setCurrentDatabase(delegate.getCurrentDatabase)
+
+  override def getTableMetadata(name: TableIdentifier): CatalogTable = {
+    val base = delegate.getTableMetadata(name)
+    val forbiddenKeys = base.properties.keys.filter(_.startsWith("deltaUniformIceberg."))
     assert(forbiddenKeys.isEmpty,
       s"deltaUniformIceberg.* keys must only appear in storage.properties, " +
         s"never in catalogTable.properties. Found: ${forbiddenKeys.mkString(", ")}")
-    val enriched = UCBackedDeltaCatalog.currentCoordinator.flatMap { coordinator =>
-      val deltaLog = DeltaLog.forTable(spark, new Path(catalogTable.location))
-      val tableConf = deltaLog.update().metadata.coordinatedCommitsTableConf
-      tableConf.get(UCCommitCoordinatorClient.UC_TABLE_ID_KEY).flatMap { tableId =>
-        coordinator.getUniformMetadata(tableId)
-          .filter(_.getIcebergMetadata.isPresent)
-          .map { meta =>
-            val icebergMeta = meta.getIcebergMetadata.get
-            catalogTable.copy(storage = catalogTable.storage.copy(
-              properties = catalogTable.storage.properties +
-                (IcebergConstants.CATALOG_TABLE_ICEBERG_METADATA_LOCATION_PROP ->
-                  icebergMeta.getMetadataLocation) +
-                (IcebergConstants.CATALOG_TABLE_ICEBERG_CONVERTED_DELTA_VERSION_PROP ->
-                  icebergMeta.getConvertedDeltaVersion.toString)))
-          }
-      }
-    }.getOrElse(catalogTable)
-    super.loadCatalogTable(ident, enriched)
+    UCEnrichedSessionCatalog.currentCoordinator.flatMap { coordinator =>
+      scala.util.Try {
+        val deltaLog = DeltaLog.forTable(spark, new Path(base.location))
+        val tableConf = deltaLog.unsafeVolatileSnapshot.metadata.coordinatedCommitsTableConf
+        tableConf.get(UCCommitCoordinatorClient.UC_TABLE_ID_KEY).flatMap { tableId =>
+          coordinator.getUniformMetadata(tableId)
+            .filter(_.getIcebergMetadata.isPresent)
+            .map { meta =>
+              val icebergMeta = meta.getIcebergMetadata.get
+              base.copy(storage = base.storage.copy(
+                properties = base.storage.properties +
+                  (IcebergConstants.CATALOG_TABLE_ICEBERG_METADATA_LOCATION_PROP ->
+                    icebergMeta.getMetadataLocation) +
+                  (IcebergConstants.CATALOG_TABLE_ICEBERG_CONVERTED_DELTA_VERSION_PROP ->
+                    icebergMeta.getConvertedDeltaVersion.toString)
+              ))
+            }
+        }
+      }.toOption.flatten
+    }.getOrElse(base)
   }
 }
 
-object UCBackedDeltaCatalog {
+private object UCEnrichedSessionCatalog {
   @volatile var currentCoordinator: Option[InMemoryUCCommitCoordinator] = None
 }
 
@@ -154,7 +167,7 @@ trait WriteDeltaUCCCReadIceberg extends UniFormE2ETest
   override protected def sparkConf: SparkConf =
     super.sparkConf.set(
       SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION.key,
-      classOf[UCBackedDeltaCatalog].getName)
+      classOf[DeltaCatalog].getName)
 
   /**
    * A [[UCCommitCoordinatorClient]] subclass that overrides [[registerTable]] to auto-assign
@@ -187,13 +200,15 @@ trait WriteDeltaUCCCReadIceberg extends UniFormE2ETest
 
   protected var ucCommitCoordinator: InMemoryUCCommitCoordinator = _
   private var testCoordinator: TestUCBackedCommitCoordinator = _
+  private var origSessionCatalog: SessionCatalog = _
+  private var origV2Catalog: SessionCatalog = _
 
   abstract override def beforeEach(): Unit = {
     super.beforeEach()
     DeltaLog.clearCache()
     CommitCoordinatorProvider.clearAllBuilders()
     ucCommitCoordinator = new InMemoryUCCommitCoordinator()
-    UCBackedDeltaCatalog.currentCoordinator = Some(ucCommitCoordinator)
+    UCEnrichedSessionCatalog.currentCoordinator = Some(ucCommitCoordinator)
     val ucClient = new InMemoryUCClient("test-metastore", ucCommitCoordinator)
     testCoordinator = new TestUCBackedCommitCoordinator(ucClient)
     CommitCoordinatorProvider.registerBuilder(new CatalogOwnedCommitCoordinatorBuilder {
@@ -205,15 +220,63 @@ trait WriteDeltaUCCCReadIceberg extends UniFormE2ETest
           spark: SparkSession, catalogName: String): JCommitCoordinatorClient =
         testCoordinator
     })
+
+    // Inject UCEnrichedSessionCatalog into two places so all catalog reads:
+    // whether from the
+    // V1 path (refreshCatalogTableIfNeeded -> spark.sessionState.catalog.getTableMetadata) or
+    // V2 write path (DeltaCatalog.loadTable -> V2SessionCatalog.catalog.getTableMetadata)
+    // see up-to-date coordinator Iceberg metadata,
+    // mirroring what toV1Table does in production for UCDeltaCatalogClientImpl
+    origSessionCatalog = spark.sessionState.catalog  // force lazy init of the field
+    val fr = reflectField(origSessionCatalog, "functionRegistry")
+      .get(origSessionCatalog).asInstanceOf[FunctionRegistry]
+    val tfr = reflectField(origSessionCatalog, "tableFunctionRegistry")
+      .get(origSessionCatalog).asInstanceOf[TableFunctionRegistry]
+    val enriched = new UCEnrichedSessionCatalog(spark, origSessionCatalog, fr, tfr)
+
+    // V1 path: spark.sessionState.catalog (used by refreshCatalogTableIfNeeded)
+    reflectField(spark.sessionState, "catalog").set(spark.sessionState, enriched)
+
+    // V2 path: V2SessionCatalog holds a direct reference to SessionCatalog captured at
+    // session construction; it is unaffected by the above swap, so replace it separately.
+    val v2Cat = spark.sessionState.catalogManager.catalog("spark_catalog")
+    val v2Delegate = reflectField(v2Cat, "delegate").get(v2Cat).asInstanceOf[AnyRef]
+    val v2CatalogField = reflectField(v2Delegate, "catalog")
+    origV2Catalog = v2CatalogField.get(v2Delegate).asInstanceOf[SessionCatalog]
+    v2CatalogField.set(v2Delegate, enriched)
   }
 
   abstract override def afterEach(): Unit = {
-    UCBackedDeltaCatalog.currentCoordinator = None
+    if (origV2Catalog != null) {
+      val v2Cat = spark.sessionState.catalogManager.catalog("spark_catalog")
+      val v2Delegate = reflectField(v2Cat, "delegate").get(v2Cat).asInstanceOf[AnyRef]
+      reflectField(v2Delegate, "catalog").set(v2Delegate, origV2Catalog)
+      origV2Catalog = null
+    }
+    if (origSessionCatalog != null) {
+      reflectField(spark.sessionState, "catalog").set(spark.sessionState, origSessionCatalog)
+      origSessionCatalog = null
+    }
+    UCEnrichedSessionCatalog.currentCoordinator = None
     CommitCoordinatorProvider.clearAllBuilders()
     DeltaLog.clearCache()
     super.afterEach()
   }
 
+  private def reflectField(obj: AnyRef, name: String): java.lang.reflect.Field = {
+    var cls: Class[_] = obj.getClass
+    while (cls != null) {
+      try {
+        val f = cls.getDeclaredField(name)
+        f.setAccessible(true)
+        return f
+      } catch {
+        case _: NoSuchFieldException => cls = cls.getSuperclass
+      }
+    }
+    throw new NoSuchFieldException(s"Field '$name' not found in ${obj.getClass.getName}")
+  }
+
   /**
    * Returns the TBLPROPERTIES SQL fragment required to enable the UC commit coordinator.
    * Concrete suites should append this to their [[extraTableProperties]] override.
@@ -244,8 +307,67 @@ trait WriteDeltaUCCCReadIceberg extends UniFormE2ETest
  * by an in-memory UC commit coordinator, reading results via the native Iceberg reader.
  */
 class UniFormE2EIcebergUCSuite extends UniFormE2EIcebergSuiteBase
-    with WriteDeltaUCCCReadIceberg {
-  // No test should go here. Please add tests in [[UniFormE2EIcebergSuiteBase]]
+  with WriteDeltaUCCCReadIceberg {
+  // Tests that don't require CC infrastructure should be added in [[UniFormE2EIcebergSuiteBase]].
+  // Tests that specifically exercise the UC commit coordinator path may be added here.
   override def extraTableProperties(compatVersion: Int): String =
     super.extraTableProperties(compatVersion) + requiredTableProperties
+
+  test("conflict resolution refreshes catalogTable with fresh uniform metadata " +
+    "for incremental Iceberg conversion") {
+    val tableName = "test_cc_conflict_iceberg_refresh"
+    withTable(tableName) {
+      // Create a CC + UniForm table.
+      write(
+        s"""CREATE TABLE $tableName (id INT) USING DELTA
+           |TBLPROPERTIES (
+           |  'delta.columnMapping.mode' = 'name',
+           |  'delta.enableIcebergCompatV2' = 'true',
+           |  'delta.universalFormat.enabledFormats' = 'iceberg'
+           |  $requiredTableProperties
+           |)""".stripMargin)
+
+      // v1: insert row 1 - triggers atomic Iceberg conversion at v1.
+      write(s"INSERT INTO $tableName VALUES (1)")
+      val deltaLog = DeltaLog.forTable(spark, TableIdentifier(tableName))
+      val v1Snapshot = deltaLog.update()
+      val v1CatalogTable =
+        spark.sessionState.catalog.getTableMetadata(TableIdentifier(tableName))
+
+      // Start a transaction with v1 snapshot and v1 Iceberg catalog.
+      val txn = deltaLog.startTransaction(Some(v1CatalogTable), Some(v1Snapshot))
+
+      // v2: insert row 2 - the "winning" commit that the stale txn will conflict with.
+      // Now v1 snapshot and v1 Iceberg catalog becomes stale so txn's commit would hit
+      // conflict.
+      write(s"INSERT INTO $tableName VALUES (2)")
+
+      // Try to commit the transaction as v2, hit a conflict, then retry as v3.
+      // During conflict resolution, getCommits returns v2's UniformMetadata, which
+      // refreshes catalogTable to convertedDeltaVersion=2. The retry commit then converts
+      // only v3 incrementally (fromVersion=3, toVersion=3) rather than from stale v1.
+      val events = Log4jUsageLogger.track {
+        txn.commit(Seq.empty, DeltaOperations.ManualUpdate)
+      }
+
+      // Latest version is now 3. There are two deltaCommitRange events: one from the failed
+      // first attempt (v2, using stale v1 base) and one from the successful retry (v3).
+      // The retry must use the refreshed catalog (convertedDeltaVersion=2 from v2's
+      // UniformMetadata), so conversion covers only v3 (fromVersion=3, toVersion=3).
+      // Without the fix, the retry would still use the stale v1 base and fromVersion would be 2.
+      val latestVersion = deltaLog.update().version
+      val rangeEvents = DeltaTestUtils.filterUsageRecords(
+        events, "delta.iceberg.conversion.deltaCommitRange")
+      assert(rangeEvents.size == 2,
+        s"Expected 2 deltaCommitRange events (failed attempt + retry), got ${rangeEvents.size}")
+      val retryEventData = JsonUtils.fromJson[Map[String, Any]](rangeEvents.last.blob)
+      // Jackson deserializes small JSON integers as Int, but latestVersion is Long;
+      // use Number.longValue for a type-safe comparison.
+      assert(retryEventData("fromVersion").asInstanceOf[Number].longValue === latestVersion,
+        s"Expected fromVersion=$latestVersion (fresh v2 base), " +
+          s"got ${retryEventData("fromVersion")} (stale v1 base would give ${latestVersion - 1})")
+      assert(retryEventData("toVersion").asInstanceOf[Number].longValue === latestVersion,
+        s"Expected toVersion=$latestVersion")
+    }
+  }
 }