From cb5d526638f33ad7aaccd7b8f3ad8c25774d7028 Mon Sep 17 00:00:00 2001
From: lzlfred <lzlfred@gmail.com>
Date: Wed, 13 Nov 2024 21:39:27 -0800
Subject: [PATCH] impl

---
 .../icebergShaded/IcebergConversionTransaction.scala | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

diff --git a/iceberg/src/main/scala/org/apache/spark/sql/delta/icebergShaded/IcebergConversionTransaction.scala b/iceberg/src/main/scala/org/apache/spark/sql/delta/icebergShaded/IcebergConversionTransaction.scala
index 1f05e13ca4..bd69ab3c3f 100644
--- a/iceberg/src/main/scala/org/apache/spark/sql/delta/icebergShaded/IcebergConversionTransaction.scala
+++ b/iceberg/src/main/scala/org/apache/spark/sql/delta/icebergShaded/IcebergConversionTransaction.scala
@@ -350,11 +350,13 @@ class IcebergConversionTransaction(
       .set(IcebergConverter.DELTA_TIMESTAMP_PROPERTY, postCommitSnapshot.timestamp.toString)
       .set(IcebergConstants.ICEBERG_NAME_MAPPING_PROPERTY, nameMapping)
 
-    if (spark.sessionState.conf.getConf(
-        DeltaSQLConf.DELTA_UNIFORM_ICEBERG_INCLUDE_BASE_CONVERTED_VERSION)) {
-      lastConvertedDeltaVersion.foreach { v =>
-        updateTxn = updateTxn.set(IcebergConverter.BASE_DELTA_VERSION_PROPERTY, v.toString)
-      }
+    val includeBaseVersion = spark.sessionState.conf.getConf(
+      DeltaSQLConf.DELTA_UNIFORM_ICEBERG_INCLUDE_BASE_CONVERTED_VERSION)
+    updateTxn = lastConvertedDeltaVersion match {
+      case Some(v) if includeBaseVersion =>
+        updateTxn.set(IcebergConverter.BASE_DELTA_VERSION_PROPERTY, v.toString)
+      case _ =>
+        updateTxn.remove(IcebergConverter.BASE_DELTA_VERSION_PROPERTY)
     }
     updateTxn.commit()