[Spark] Fix DomainMetadata handling for REPLACE TABLE concurrent with a transaction that adds a new domain (delta-io#4712)

#### Which Delta project/connector is this regarding?
- [x] Spark
- [ ] Standalone
- [ ] Flink
- [ ] Kernel
- [ ] Other (fill in here)

## Description
If the winning transaction adds a new DomainMetadata for the first time
in a table's history concurrent to a REPLACE TABLE operation, the
REPLACE TABLE will incorrectly not mark this DomainMetadata as removed.
This PR fixes this issue by marking any DomainMetadata added by the
winning transaction as removed if they appear in the list of
DomainMetadata to be removed and the domain does not appear in the
DomainMetadata added by the REPLACE TABLE.

## How was this patch tested?
N/A - no DomainMetadata currently exist that fall into this category.

Author: chirag-s-db

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

@@ -677,6 +677,11 @@ private[delta] class ConflictChecker(
     }
   }
 
+  private lazy val currentTransactionIsReplaceTable: Boolean = currentTransactionInfo.op match {
+    case _: DeltaOperations.ReplaceTable => true
+    case _ => false
+  }
+
   /**
    * Checks [[DomainMetadata]] to capture whether the current transaction conflicts with the
    * winning transaction at any domain.
@@ -721,9 +726,26 @@ private[delta] class ConflictChecker(
       case other => other
     }
 
+
+    // For the REPLACE TABLE command, if domain metadata of a given domain is added for the first
+    // time by the winning transaction, it may need to be marked as removed.
+    val replaceTableRemoveNewDomainMetadataEnabled = spark.conf.get(
+      DeltaSQLConf.DELTA_CONFLICT_DETECTION_ALLOW_REPLACE_TABLE_TO_REMOVE_NEW_DOMAIN_METADATA)
+    val (finalUpdatedActions, finalMergedDomainMetadata) =
+      if (replaceTableRemoveNewDomainMetadataEnabled && currentTransactionIsReplaceTable) {
+        val (domainMetadataActions, nonDomainMetadataActions) =
+          currentTransactionInfo.actions.partition(_.isInstanceOf[DomainMetadata])
+        val updatedDomainMetadataActions = DomainMetadataUtils.handleDomainMetadataForReplaceTable(
+          winningDomainMetadataMap.values.toSeq,
+          domainMetadataActions.map(_.asInstanceOf[DomainMetadata]))
+        ((nonDomainMetadataActions ++ updatedDomainMetadataActions), updatedDomainMetadataActions)
+      } else {
+        (updatedActions, mergedDomainMetadata)
+      }
+
     currentTransactionInfo = currentTransactionInfo.copy(
-      domainMetadata = mergedDomainMetadata.toSeq,
-      actions = updatedActions)
+      domainMetadata = finalMergedDomainMetadata.toSeq,
+      actions = finalUpdatedActions)
   }
 
   /**

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

@@ -1686,6 +1686,16 @@ trait DeltaSQLConfBase {
       .checkValues(NonDeterministicPredicateWidening.list)
       .createWithDefault(NonDeterministicPredicateWidening.ON)
 
+  val DELTA_CONFLICT_DETECTION_ALLOW_REPLACE_TABLE_TO_REMOVE_NEW_DOMAIN_METADATA =
+    buildConf("conflictDetection.allowReplaceTableToRemoveNewDomainMetadata")
+      .doc("Whether to allow removing new domain metadatas from concurrent transactions during " +
+        "conflict resolution for a REPLACE TABLE operation. Note that this flag applies only " +
+        "to metadata domains where the table snapshot read by the REPLACE TABLE command did " +
+        "not contain a domain metadata of the same domain.")
+      .internal()
+      .booleanConf
+      .createWithDefault(true)
+
   val DELTA_UNIFORM_ICEBERG_SYNC_CONVERT_ENABLED =
     buildConf("uniform.iceberg.sync.convert.enabled")
       .doc("If enabled, iceberg conversion will be done synchronously. " +