From ebf9d772f29c98aa0a5116391b93d99b2b0f4c38 Mon Sep 17 00:00:00 2001
From: Fred Storage Liu <fred.liu@databricks.com>
Date: Tue, 27 Aug 2024 21:01:14 -0700
Subject: [PATCH] impl

---
 .../delta/icebergShaded/IcebergConverter.scala   | 16 ++++++++++++----
 1 file changed, 12 insertions(+), 4 deletions(-)

diff --git a/iceberg/src/main/scala/org/apache/spark/sql/delta/icebergShaded/IcebergConverter.scala b/iceberg/src/main/scala/org/apache/spark/sql/delta/icebergShaded/IcebergConverter.scala
index f5ab98a2f32..64e3925c810 100644
--- a/iceberg/src/main/scala/org/apache/spark/sql/delta/icebergShaded/IcebergConverter.scala
+++ b/iceberg/src/main/scala/org/apache/spark/sql/delta/icebergShaded/IcebergConverter.scala
@@ -361,6 +361,7 @@ class IcebergConverter(spark: SparkSession)
         var hasRemoves = false
         var hasDataChange = false
         var hasCommitInfo = false
+        var commitInfo: Option[CommitInfo] = None
         breakable {
           for (action <- actionsToCommit) {
             action match {
@@ -370,7 +371,9 @@ class IcebergConverter(spark: SparkSession)
               case r: RemoveFile =>
                 hasRemoves = true
                 if (r.dataChange) hasDataChange = true
-              case _: CommitInfo => hasCommitInfo = true
+              case ci: CommitInfo =>
+                commitInfo = Some(ci)
+                hasCommitInfo = true
               case _ => // Do nothing
             }
             if (hasAdds && hasRemoves && hasDataChange && hasCommitInfo) break // Short-circuit
@@ -404,9 +407,14 @@ class IcebergConverter(spark: SparkSession)
           }
           overwriteHelper.commit()
         } else if (hasAdds) {
-          val appendHelper = icebergTxn.getAppendOnlyHelper()
-          addsAndRemoves.foreach(action => appendHelper.add(action.add))
-          appendHelper.commit()
+          if (!hasRemoves && !hasDataChange && allDeltaActionsCaptured) {
+            logInfo(s"Skip Iceberg conversion for commit that only has AddFiles " +
+              s"without any RemoveFiles or data change. CommitInfo: $commitInfo")
+          } else {
+            val appendHelper = icebergTxn.getAppendOnlyHelper()
+              addsAndRemoves.foreach(action => appendHelper.add(action.add))
+              appendHelper.commit()
+          }
         } else if (hasRemoves) {
           val removeHelper = icebergTxn.getRemoveOnlyHelper()
           addsAndRemoves.foreach(action => removeHelper.remove(action.remove))