Repository: NetEase/arctic
Pull request #338 (2022-09-06 12:40:46)

move base.table.max-transaction-id from table property to partitionProperties
## Why are the changes needed?
  1. make partition properties to unified saved.

## Brief change log
  - *move base.table.max-transaction-id from table property to partitionProperties*


Patch:
diff --git a/ams/ams-server/pom.xml b/ams/ams-server/pom.xml
index 7e2f639b6..a73b8ed54 100644
--- a/ams/ams-server/pom.xml
+++ b/ams/ams-server/pom.xml
@@ -214,7 +214,7 @@
         <dependency>
             <groupId>mysql</groupId>
             <artifactId>mysql-connector-java</artifactId>
-            <version>5.1.36</version>
+            <version>8.0.30</version>
         </dependency>
 
         <dependency>
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/OptimizeTasksMapper.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/OptimizeTasksMapper.java
index 1eb84cb6a..d17b6b226 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/OptimizeTasksMapper.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/OptimizeTasksMapper.java
@@ -36,7 +36,7 @@ public interface OptimizeTasksMapper {
   String TABLE_NAME = "optimize_task";
 
   @Select("select trace_id, optimize_type, catalog_name, db_name, table_name, `partition`," +
-      " task_group, task_history_id, max_change_transaction_id, is_delete_pos_delete," +
+      " task_group, task_history_id, max_change_transaction_id," +
       " source_nodes, create_time, properties, queue_id," +
       " insert_file_size, delete_file_size, base_file_size, pos_delete_file_size," +
       " insert_files, delete_files, base_files, pos_delete_files" +
@@ -51,7 +51,6 @@ public interface OptimizeTasksMapper {
       @Result(property = "taskGroup", column = "task_group"),
       @Result(property = "taskHistoryId", column = "task_history_id"),
       @Result(property = "maxChangeTransactionId", column = "max_change_transaction_id"),
-      @Result(property = "isDeletePosDelete", column = "is_delete_pos_delete"),
       @Result(property = "queueId", column = "queue_id"),
       @Result(property = "insertFileSize", column = "insert_file_size"),
       @Result(property = "deleteFileSize", column = "delete_file_size"),
@@ -72,7 +71,7 @@ public interface OptimizeTasksMapper {
 
   @Insert("insert into " + TABLE_NAME + " (" +
       " trace_id, optimize_type, catalog_name, db_name, table_name, `partition`," +
-      " task_group, task_history_id, max_change_transaction_id, is_delete_pos_delete," +
+      " task_group, task_history_id, max_change_transaction_id," +
       " source_nodes, create_time, properties, queue_id," +
       " insert_file_size, delete_file_size, base_file_size, pos_delete_file_size," +
       " insert_files, delete_files, base_files, pos_delete_files," +
@@ -89,7 +88,6 @@ public interface OptimizeTasksMapper {
       " #{optimizeTask.taskGroup}," +
       " #{optimizeTask.taskHistoryId}," +
       " #{optimizeTask.maxChangeTransactionId}," +
-      " #{optimizeTask.isDeletePosDelete}," +
       " #{optimizeTask.sourceNodes, " +
       "typeHandler=com.netease.arctic.ams.server.mybatis.ListOfTreeNode2StringConverter}," +
       " #{optimizeTask.createTime, " +
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/derby/DerbyOptimizeTasksMapper.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/derby/DerbyOptimizeTasksMapper.java
index b3005827d..7316af8f4 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/derby/DerbyOptimizeTasksMapper.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/derby/DerbyOptimizeTasksMapper.java
@@ -36,7 +36,7 @@ public interface DerbyOptimizeTasksMapper extends OptimizeTasksMapper {
   String TABLE_NAME = "optimize_task";
 
   @Select("select trace_id, optimize_type, catalog_name, db_name, table_name, partition," +
-      " task_group, task_history_id, max_change_transaction_id, is_delete_pos_delete," +
+      " task_group, task_history_id, max_change_transaction_id," +
       " source_nodes, create_time, properties, queue_id," +
       " insert_file_size, delete_file_size, base_file_size, pos_delete_file_size" +
       " insert_files, delete_files, base_files, pos_delete_files from " + TABLE_NAME)
@@ -50,7 +50,6 @@ public interface DerbyOptimizeTasksMapper extends OptimizeTasksMapper {
       @Result(property = "taskGroup", column = "task_group"),
       @Result(property = "taskHistoryId", column = "task_history_id"),
       @Result(property = "maxChangeTransactionId", column = "max_change_transaction_id"),
-      @Result(property = "isDeletePosDelete", column = "is_delete_pos_delete"),
       @Result(property = "queueId", column = "queue_id"),
       @Result(property = "insertFileSize", column = "insert_file_size"),
       @Result(property = "deleteFileSize", column = "delete_file_size"),
@@ -71,7 +70,7 @@ public interface DerbyOptimizeTasksMapper extends OptimizeTasksMapper {
 
   @Insert("insert into " + TABLE_NAME + " (" +
           " trace_id, optimize_type, catalog_name, db_name, table_name, partition," +
-          " task_group, task_history_id, max_change_transaction_id, is_delete_pos_delete," +
+          " task_group, task_history_id, max_change_transaction_id," +
           " source_nodes, create_time, properties, queue_id," +
           " insert_file_size, delete_file_size, base_file_size, pos_delete_file_size," +
           " insert_files, delete_files, base_files, pos_delete_files," +
@@ -88,7 +87,6 @@ public interface DerbyOptimizeTasksMapper extends OptimizeTasksMapper {
           " #{optimizeTask.taskGroup, jdbcType=VARCHAR}," +
           " #{optimizeTask.taskHistoryId, jdbcType=VARCHAR}," +
           " #{optimizeTask.maxChangeTransactionId}," +
-          " #{optimizeTask.isDeletePosDelete}," +
           " #{optimizeTask.sourceNodes, " +
           "typeHandler=com.netease.arctic.ams.server.mybatis.ListOfTreeNode2StringConverter}," +
           " #{optimizeTask.createTime, " +
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/BaseOptimizeTask.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/BaseOptimizeTask.java
index 6fd74dfcc..bb420861b 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/BaseOptimizeTask.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/BaseOptimizeTask.java
@@ -41,8 +41,6 @@ public class BaseOptimizeTask extends OptimizeTask {
   protected long createTime;
 
   private long maxChangeTransactionId = INVALID_TRANSACTION_ID;
-  @Deprecated
-  private int isDeletePosDelete;
 
   public BaseOptimizeTask() {
   }
@@ -143,16 +141,6 @@ public class BaseOptimizeTask extends OptimizeTask {
     this.deleteFileCnt = deleteFileCnt;
   }
 
-  @Deprecated
-  public int getIsDeletePosDelete() {
-    return isDeletePosDelete;
-  }
-
-  @Deprecated
-  public void setIsDeletePosDelete(int isDeletePosDelete) {
-    this.isDeletePosDelete = isDeletePosDelete;
-  }
-
   public long getPosDeleteFileSize() {
     return posDeleteFileSize;
   }
@@ -186,7 +174,6 @@ public class BaseOptimizeTask extends OptimizeTask {
         ", queueId=" + queueId +
         ", createTime=" + createTime +
         ", maxChangeTransactionId=" + maxChangeTransactionId +
-        ", isDeletePosDelete=" + isDeletePosDelete +
         "} " + superToString();
   }
 
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/BaseOptimizeCommit.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/BaseOptimizeCommit.java
index 6c0a40808..18c2f9cfd 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/BaseOptimizeCommit.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/BaseOptimizeCommit.java
@@ -30,6 +30,7 @@ import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.UnkeyedTable;
 import com.netease.arctic.trace.SnapshotSummary;
 import com.netease.arctic.utils.SerializationUtil;
+import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.iceberg.ContentFile;
 import org.apache.iceberg.DataFile;
@@ -103,7 +104,7 @@ public class BaseOptimizeCommit {
             long maxTransactionId = task.getOptimizeTask().getMaxChangeTransactionId();
             if (maxTransactionId != BaseOptimizeTask.INVALID_TRANSACTION_ID) {
               if (arcticTable.asKeyedTable().baseTable().spec().isUnpartitioned()) {
-                maxTransactionIds.put(null, maxTransactionId);
+                maxTransactionIds.put(TablePropertyUtil.EMPTY_STRUCT, maxTransactionId);
               } else {
                 maxTransactionIds.putIfAbsent(
                     DataFiles.data(spec, entry.getKey()), maxTransactionId);
@@ -172,7 +173,8 @@ public class BaseOptimizeCommit {
         });
 
         if (spec.isUnpartitioned()) {
-          overwriteBaseFiles.withMaxTransactionId(null, maxTransactionIds.get(null));
+          overwriteBaseFiles.withMaxTransactionId(TablePropertyUtil.EMPTY_STRUCT,
+              maxTransactionIds.get(TablePropertyUtil.EMPTY_STRUCT));
         } else {
           maxTransactionIds.forEach(overwriteBaseFiles::withMaxTransactionId);
         }
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/BaseOptimizePlan.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/BaseOptimizePlan.java
index 4aa3e8ff6..1c98b8820 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/BaseOptimizePlan.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/BaseOptimizePlan.java
@@ -106,7 +106,7 @@ public abstract class BaseOptimizePlan {
   protected abstract boolean partitionNeedPlan(String partitionToPath);
 
   protected abstract void addOptimizeFilesTree();
-  
+
   protected abstract OptimizeType getOptimizeType();
 
   protected abstract List<BaseOptimizeTask> collectTask(String partition);
@@ -173,11 +173,11 @@ public abstract class BaseOptimizePlan {
   }
 
   protected BaseOptimizeTask buildOptimizeTask(@Nullable List<DataTreeNode> sourceNodes,
-                                            List<DataFile> insertFiles,
-                                            List<DataFile> deleteFiles,
-                                            List<DataFile> baseFiles,
-                                            List<DeleteFile> posDeleteFiles,
-                                            TaskConfig taskConfig) {
+                                               List<DataFile> insertFiles,
+                                               List<DataFile> deleteFiles,
+                                               List<DataFile> baseFiles,
+                                               List<DeleteFile> posDeleteFiles,
+                                               TaskConfig taskConfig) {
     // build task
     BaseOptimizeTask optimizeTask = new BaseOptimizeTask();
     optimizeTask.setTaskGroup(taskConfig.getGroup());
@@ -250,9 +250,6 @@ public abstract class BaseOptimizePlan {
     if (taskConfig.getMaxTransactionId() != null) {
       optimizeTask.setMaxChangeTransactionId(taskConfig.getMaxTransactionId());
     }
-    if (taskConfig.getOptimizeType() == OptimizeType.FullMajor) {
-      optimizeTask.setIsDeletePosDelete(1);
-    }
 
     // table ams url
     Map<String, String> properties = new HashMap<>();
@@ -272,7 +269,7 @@ public abstract class BaseOptimizePlan {
       snapshot = arcticTable.asKeyedTable().baseTable().currentSnapshot();
       if (snapshot != null && !snapshotIsCached.test(snapshot.snapshotId())) {
         LOG.debug("File cache don't have cache snapshotId:{}," +
-                "wait file cache sync latest file info", snapshot.snapshotId());
+            "wait file cache sync latest file info", snapshot.snapshotId());
         return false;
       }
     }
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/MinorOptimizePlan.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/MinorOptimizePlan.java
index dda0e1640..01b4673b2 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/MinorOptimizePlan.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/MinorOptimizePlan.java
@@ -32,13 +32,16 @@ import com.netease.arctic.data.DefaultKeyedFile;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.TableProperties;
+import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.iceberg.ContentFile;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileContent;
 import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.StructLikeMap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -61,7 +64,7 @@ public class MinorOptimizePlan extends BaseOptimizePlan {
   // partition -> delete file
   protected final Map<String, List<DataFile>> partitionDeleteFiles = new LinkedHashMap<>();
   // partition -> maxBaseTableTransactionId
-  protected Map<String, Long> baseTableMaxTransactionId = null;
+  protected StructLikeMap<Long> baseTableMaxTransactionId = null;
   // partition -> maxChangeTableTransactionId
   protected final Map<String, Long> changeTableMaxTransactionId = new HashMap<>();
 
@@ -158,7 +161,7 @@ public class MinorOptimizePlan extends BaseOptimizePlan {
       ContentFile<?> dataFile = ContentFileUtil.buildContentFile(dataFileInfo, partitionSpec);
       currentPartitions.add(partition);
       allPartitions.add(partition);
-      if (isOptimized(dataFile, partition)) {
+      if (isOptimized(dataFile)) {
         return null;
       } else {
         if (!anyTaskRunning(partition)) {
@@ -232,6 +235,7 @@ public class MinorOptimizePlan extends BaseOptimizePlan {
     List<BaseOptimizeTask> collector = new ArrayList<>();
     String group = UUID.randomUUID().toString();
     long createTime = System.currentTimeMillis();
+
     TaskConfig taskPartitionConfig = new TaskConfig(partition, changeTableMaxTransactionId.get(partition),
         group, historyId, OptimizeType.Minor, createTime);
     treeRoot.completeTree(false);
@@ -268,28 +272,29 @@ public class MinorOptimizePlan extends BaseOptimizePlan {
     return collector;
   }
 
-  private boolean isOptimized(ContentFile<?> dataFile, String partition) {
+  private boolean isOptimized(ContentFile<?> dataFile) {
     // if Pos-Delete files, ignore
     if (dataFile.content() == FileContent.POSITION_DELETES) {
       return false;
     }
 
-    Long currentValue = changeTableMaxTransactionId.get(partition);
+    StructLike partition = dataFile.partition();
+    String partitionToPath = arcticTable.spec().partitionToPath(partition);
+    Long currentValue = changeTableMaxTransactionId.get(partitionToPath);
     long transactionId = DefaultKeyedFile.parseMetaFromFileName(dataFile.path().toString()).transactionId();
     if (currentValue == null) {
-      changeTableMaxTransactionId.put(partition, transactionId);
+      changeTableMaxTransactionId.put(partitionToPath, transactionId);
     } else {
       if (currentValue < transactionId) {
-        changeTableMaxTransactionId.put(partition, transactionId);
+        changeTableMaxTransactionId.put(partitionToPath, transactionId);
       }
     }
     return transactionId <= getBaseMaxTransactionId(partition);
   }
 
-  private long getBaseMaxTransactionId(String partition) {
+  private long getBaseMaxTransactionId(StructLike partition) {
     if (baseTableMaxTransactionId == null) {
-      baseTableMaxTransactionId = new HashMap<>();
-      baseTableMaxTransactionId.putAll(arcticTable.asKeyedTable().maxTransactionId());
+      baseTableMaxTransactionId = TablePropertyUtil.getPartitionMaxTransactionId(arcticTable.asKeyedTable());
       LOG.debug("{} ==== get base table max transaction id: {}", tableId(), baseTableMaxTransactionId);
     }
     Long maxTransactionId = baseTableMaxTransactionId.get(partition);
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/SupportHiveCommit.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/SupportHiveCommit.java
index 161eccce1..301efd865 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/SupportHiveCommit.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/SupportHiveCommit.java
@@ -10,6 +10,7 @@ import com.netease.arctic.hive.HMSClient;
 import com.netease.arctic.hive.table.SupportHive;
 import com.netease.arctic.hive.utils.HivePartitionUtil;
 import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.utils.FileUtil;
 import com.netease.arctic.utils.SerializationUtil;
@@ -39,7 +40,7 @@ public class SupportHiveCommit extends BaseOptimizeCommit {
                            Map<String, List<OptimizeTaskItem>> optimizeTasksToCommit,
                            Consumer<OptimizeTaskItem> updateTargetFiles) {
     super(arcticTable, optimizeTasksToCommit);
-    Preconditions.checkArgument(HiveTableUtil.isHive(arcticTable), "The table not support hive");
+    Preconditions.checkArgument(TableTypeUtil.isHive(arcticTable), "The table not support hive");
     this.updateTargetFiles = updateTargetFiles;
   }
 
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/SupportHiveFullOptimizePlan.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/SupportHiveFullOptimizePlan.java
index 53b2f5a6b..d03018c53 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/SupportHiveFullOptimizePlan.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/SupportHiveFullOptimizePlan.java
@@ -22,7 +22,7 @@ import com.google.common.base.Preconditions;
 import com.netease.arctic.ams.api.DataFileInfo;
 import com.netease.arctic.ams.server.model.TableOptimizeRuntime;
 import com.netease.arctic.hive.table.SupportHive;
-import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.TableProperties;
 import org.apache.commons.collections.CollectionUtils;
@@ -53,7 +53,7 @@ public class SupportHiveFullOptimizePlan extends FullOptimizePlan {
     super(arcticTable, tableOptimizeRuntime, baseTableFileList, posDeleteFileList,
         partitionTaskRunning, queueId, currentTime, snapshotIsCached);
 
-    Preconditions.checkArgument(HiveTableUtil.isHive(arcticTable), "The table not support hive");
+    Preconditions.checkArgument(TableTypeUtil.isHive(arcticTable), "The table not support hive");
     hiveLocation = ((SupportHive) arcticTable).hiveLocation();
     excludeLocations.add(hiveLocation);
   }
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/SupportHiveMajorOptimizePlan.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/SupportHiveMajorOptimizePlan.java
index b42fc92ad..6c6978a1c 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/SupportHiveMajorOptimizePlan.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/SupportHiveMajorOptimizePlan.java
@@ -5,7 +5,7 @@ import com.netease.arctic.ams.api.DataFileInfo;
 import com.netease.arctic.ams.api.OptimizeType;
 import com.netease.arctic.ams.server.model.TableOptimizeRuntime;
 import com.netease.arctic.hive.table.SupportHive;
-import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.TableProperties;
 import org.apache.commons.collections.CollectionUtils;
@@ -39,7 +39,7 @@ public class SupportHiveMajorOptimizePlan extends MajorOptimizePlan {
     super(arcticTable, tableOptimizeRuntime, baseTableFileList, posDeleteFileList,
         partitionTaskRunning, queueId, currentTime, snapshotIsCached);
 
-    Preconditions.checkArgument(HiveTableUtil.isHive(arcticTable), "The table not support hive");
+    Preconditions.checkArgument(TableTypeUtil.isHive(arcticTable), "The table not support hive");
     hiveLocation = ((SupportHive) arcticTable).hiveLocation();
     excludeLocations.add(hiveLocation);
   }
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/TableOptimizeItem.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/TableOptimizeItem.java
index 8d3546059..eb81057e1 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/TableOptimizeItem.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/TableOptimizeItem.java
@@ -51,11 +51,12 @@ import com.netease.arctic.catalog.ArcticCatalog;
 import com.netease.arctic.catalog.CatalogLoader;
 import com.netease.arctic.data.DataFileType;
 import com.netease.arctic.hive.table.SupportHive;
-import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.TableIdentifier;
 import com.netease.arctic.table.TableProperties;
+import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.apache.ibatis.session.SqlSession;
@@ -130,6 +131,7 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Initial optimize tasks.
+   *
    * @param optimizeTasks -
    */
   public void initOptimizeTasks(List<OptimizeTaskItem> optimizeTasks) {
@@ -141,6 +143,7 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Initial TableOptimizeRuntime.
+   *
    * @param runtime -
    * @return this for chain
    */
@@ -189,6 +192,7 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Get table identifier.
+   *
    * @return TableIdentifier
    */
   public TableIdentifier getTableIdentifier() {
@@ -197,7 +201,8 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Get Arctic Table, refresh if expired.
-   * @return  ArcticTable
+   *
+   * @return ArcticTable
    */
   public ArcticTable getArcticTable() {
     if (arcticTable == null) {
@@ -208,6 +213,7 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Get arcticTable, refresh immediately or not.
+   *
    * @param forceRefresh - refresh immediately
    * @return ArcticTable
    */
@@ -218,7 +224,7 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * If arctic table is KeyedTable.
-   * 
+   *
    * @return true/false
    */
   public boolean isKeyedTable() {
@@ -230,6 +236,7 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Get cached quota, cache will be updated when arctic table refresh.
+   *
    * @return quota
    */
   public double getQuotaCache() {
@@ -270,6 +277,7 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Update optimize task result, Failed or Prepared.
+   *
    * @param optimizeTaskStat - optimizeTaskStat
    */
   public void updateOptimizeTaskStat(OptimizeTaskStat optimizeTaskStat) {
@@ -309,6 +317,7 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Build current table optimize info.
+   *
    * @return TableOptimizeInfo
    */
   public TableOptimizeInfo buildTableOptimizeInfo() {
@@ -594,7 +603,7 @@ public class TableOptimizeItem extends IJDBCService {
     if (isKeyedTable()) {
       KeyedTable keyedHiveTable = getArcticTable(true).asKeyedTable();
       record.setSnapshotInfo(TableStatCollector.buildBaseTableSnapshotInfo(keyedHiveTable.baseTable()));
-      record.setBaseTableMaxTransactionId(keyedHiveTable.maxTransactionId().toString());
+      record.setBaseTableMaxTransactionId(TablePropertyUtil.getPartitionMaxTransactionId(keyedHiveTable).toString());
     } else {
       getArcticTable(true);
       record.setSnapshotInfo(TableStatCollector.buildBaseTableSnapshotInfo(getArcticTable(true).asUnkeyedTable()));
@@ -698,6 +707,7 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Commit optimize tasks.
+   *
    * @throws Exception -
    */
   public void commitOptimizeTasks() throws Exception {
@@ -719,7 +729,7 @@ public class TableOptimizeItem extends IJDBCService {
       if (MapUtils.isNotEmpty(tasksToCommit)) {
         LOG.info("{} get {} tasks of {} partitions to commit", tableIdentifier, taskCount, tasksToCommit.size());
         BaseOptimizeCommit optimizeCommit;
-        if (HiveTableUtil.isHive(getArcticTable())) {
+        if (TableTypeUtil.isHive(getArcticTable())) {
           optimizeCommit = new SupportHiveCommit(getArcticTable(true),
               tasksToCommit, OptimizeTaskItem::persistTargetFiles);
         } else {
@@ -743,6 +753,7 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Get all optimize tasks.
+   *
    * @return list of all optimize tasks
    */
   public List<OptimizeTaskItem> getOptimizeTasks() {
@@ -751,7 +762,8 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Get Full Plan.
-   * @param queueId -
+   *
+   * @param queueId     -
    * @param currentTime -
    * @return -
    */
@@ -773,7 +785,8 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Get Major Plan.
-   * @param queueId -
+   *
+   * @param queueId     -
    * @param currentTime -
    * @return -
    */
@@ -795,7 +808,8 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Get Minor Plan.
-   * @param queueId -
+   *
+   * @param queueId     -
    * @param currentTime -
    * @return -
    */
@@ -815,6 +829,7 @@ public class TableOptimizeItem extends IJDBCService {
 
   /**
    * Get optimizeRuntime.
+   *
    * @return -
    */
   public TableOptimizeRuntime getTableOptimizeRuntime() {
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/AdaptHiveService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/AdaptHiveService.java
index c88fd186b..7dae1711c 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/AdaptHiveService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/AdaptHiveService.java
@@ -46,7 +46,7 @@ public class AdaptHiveService {
   private static final ThreadPoolExecutor executor = new ThreadPoolExecutor(CORE_POOL_SIZE, CORE_POOL_SIZE * 2,
       QUEUE_CAPACITY, TimeUnit.SECONDS, new LinkedBlockingDeque<>(5));
 
-  public Object upgradeHiveTable(ArcticHiveCatalog ac, TableIdentifier tableIdentifier,
+  public Object upgradeHiveTable(ArcticHiveCatalog arcticHiveCatalog, TableIdentifier tableIdentifier,
                                  UpgradeHiveMeta upgradeHiveMeta) {
     LOG.info("Start to upgrade hive table to arctic" + tableIdentifier.toString());
     executor.submit(() -> {
@@ -54,7 +54,8 @@ public class AdaptHiveService {
       try {
         List<String> pkList = upgradeHiveMeta.getPkList().stream()
             .map(UpgradeHiveMeta.PrimaryKeyField::getFieldName).collect(Collectors.toList());
-        UpgradeHiveTableUtil.upgradeHiveTable(ac, tableIdentifier, pkList, upgradeHiveMeta.getProperties());
+        UpgradeHiveTableUtil.upgradeHiveTable(arcticHiveCatalog, tableIdentifier,
+            pkList, upgradeHiveMeta.getProperties());
         runningInfoCache.get(tableIdentifier).setStatus(UpgradeStatus.SUCCESS.toString());
       } catch (Throwable t) {
         LOG.error("Failed to upgrade hive table to arctic ", t);
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/SupportHiveSyncService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/SupportHiveSyncService.java
index 63ba6d92b..7d0fd717b 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/SupportHiveSyncService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/SupportHiveSyncService.java
@@ -28,7 +28,7 @@ import com.netease.arctic.catalog.CatalogLoader;
 import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.table.SupportHive;
 import com.netease.arctic.hive.utils.HivePartitionUtil;
-import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.TableIdentifier;
 import com.netease.arctic.table.UnkeyedTable;
@@ -94,7 +94,7 @@ public class SupportHiveSyncService implements ISupportHiveSyncService {
         ArcticCatalog catalog =
             CatalogLoader.load(ServiceContainer.getTableMetastoreHandler(), tableIdentifier.getCatalog());
         ArcticTable arcticTable = catalog.loadTable(tableIdentifier);
-        if (!HiveTableUtil.isHive(arcticTable)) {
+        if (!TableTypeUtil.isHive(arcticTable)) {
           LOG.debug("[{}] {} is not a support hive table", traceId, tableIdentifier);
           return;
         }
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/TableExpireService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/TableExpireService.java
index 3ebe7a431..965012612 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/TableExpireService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/TableExpireService.java
@@ -32,13 +32,14 @@ import com.netease.arctic.catalog.ArcticCatalog;
 import com.netease.arctic.catalog.CatalogLoader;
 import com.netease.arctic.data.DefaultKeyedFile;
 import com.netease.arctic.data.PrimaryKeyedFile;
-import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.TableIdentifier;
 import com.netease.arctic.table.TableProperties;
 import com.netease.arctic.table.UnkeyedTable;
 import com.netease.arctic.utils.FileUtil;
+import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.apache.hadoop.fs.Path;
@@ -115,7 +116,7 @@ public class TableExpireService implements ITableExpireService {
                 TableProperties.CHANGE_SNAPSHOT_KEEP_MINUTES_DEFAULT)) * 60 * 1000;
 
         Set<String> hiveLocation = new HashSet<>();
-        if (HiveTableUtil.isHive(arcticTable)) {
+        if (TableTypeUtil.isHive(arcticTable)) {
           hiveLocation = HiveLocationUtils.getHiveLocation(arcticTable);
         }
 
@@ -168,7 +169,7 @@ public class TableExpireService implements ITableExpireService {
   }
 
   public static void deleteChangeFile(KeyedTable keyedTable, List<DataFileInfo> changeDataFiles) {
-    StructLikeMap<Long> baseMaxTransactionId = keyedTable.partitionMaxTransactionId();
+    StructLikeMap<Long> baseMaxTransactionId = TablePropertyUtil.getPartitionMaxTransactionId(keyedTable);
     if (MapUtils.isEmpty(baseMaxTransactionId)) {
       LOG.info("table {} not contains max transaction id", keyedTable.id());
       return;
@@ -186,7 +187,7 @@ public class TableExpireService implements ITableExpireService {
       List<DataFileInfo> partitionDataFiles =
           partitionDataFileMap.get(null);
 
-      Long maxTransactionId = baseMaxTransactionId.get(null);
+      Long maxTransactionId = baseMaxTransactionId.get(TablePropertyUtil.EMPTY_STRUCT);
       if (CollectionUtils.isNotEmpty(partitionDataFiles)) {
         deleteFiles.addAll(partitionDataFiles.stream()
             .filter(dataFileInfo ->
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/HiveLocationUtils.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/HiveLocationUtils.java
index c89fb305c..99b397a31 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/HiveLocationUtils.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/HiveLocationUtils.java
@@ -19,7 +19,7 @@
 package com.netease.arctic.ams.server.utils;
 
 import com.netease.arctic.hive.table.SupportHive;
-import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.table.ArcticTable;
 import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.Table;
@@ -35,7 +35,7 @@ public class HiveLocationUtils {
 
   public static Set<String> getHiveLocation(ArcticTable table) {
     Set<String> hiveLocations = new HashSet<>();
-    if (HiveTableUtil.isHive(table)) {
+    if (TableTypeUtil.isHive(table)) {
       if (table.spec().isUnpartitioned()) {
         try {
           Table hiveTable = ((SupportHive) table).getHMSClient().run(client ->
diff --git a/ams/ams-server/src/main/resources/sql/mysql/ams-init.sql b/ams/ams-server/src/main/resources/sql/mysql/ams-init.sql
index 2c01b1815..51eda63b0 100644
--- a/ams/ams-server/src/main/resources/sql/mysql/ams-init.sql
+++ b/ams/ams-server/src/main/resources/sql/mysql/ams-init.sql
@@ -156,7 +156,6 @@ CREATE TABLE `optimize_task`
     `base_file_size`            bigint(20) DEFAULT NULL COMMENT 'Base file size in bytes',
     `pos_delete_file_size`      bigint(20) DEFAULT NULL COMMENT 'Pos-Delete file size in bytes',
     `source_nodes`              varchar(2048) DEFAULT NULL COMMENT 'Source nodes of task',
-    `is_delete_pos_delete`      tinyint(4) DEFAULT NULL COMMENT 'Delete pos delete files or not',
     `task_history_id`           varchar(40)   DEFAULT NULL COMMENT 'Task history id',
     `status`        varchar(16)   DEFAULT NULL  COMMENT 'Optimize Status: Init, Pending, Executing, Failed, Prepared, Committed',
     `pending_time`  datetime(3) DEFAULT NULL COMMENT 'Time when task start waiting to execute',
diff --git a/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java b/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java
index 35b0012f0..a467d6919 100644
--- a/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java
+++ b/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java
@@ -21,6 +21,7 @@ package com.netease.arctic.op;
 import com.netease.arctic.scan.CombinedScanTask;
 import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.UnkeyedTable;
+import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.apache.iceberg.DataFile;
@@ -34,14 +35,12 @@ import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.util.StructLikeMap;
 
 import java.io.IOException;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 
 /**
  * Overwrite {@link com.netease.arctic.table.BaseTable} and change max transaction id map
@@ -109,7 +108,7 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
   protected StructLikeMap<Long> apply(Transaction transaction, StructLikeMap<Long> partitionMaxTxId) {
     applyDeleteExpression();
 
-    StructLike partitionData = null;
+    StructLike partitionData = TablePropertyUtil.EMPTY_STRUCT;
     UnkeyedTable baseTable = keyedTable.baseTable();
 
     // step1: overwrite data files
@@ -120,13 +119,13 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
       }
       for (DataFile d : this.addFiles) {
         overwriteFiles.addFile(d);
-        partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
+        partitionData = keyedTable.spec().isUnpartitioned() ? TablePropertyUtil.EMPTY_STRUCT : d.partition();
         partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
       }
 
       for (DataFile d : this.deleteFiles) {
         overwriteFiles.deleteFile(d);
-        partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
+        partitionData = keyedTable.spec().isUnpartitioned() ? TablePropertyUtil.EMPTY_STRUCT : d.partition();
         partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
       }
       if (transactionId != null && transactionId > 0) {
@@ -148,12 +147,12 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
         }
 
         for (DeleteFile d : this.addDeleteFiles) {
-          partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
+          partitionData = keyedTable.spec().isUnpartitioned() ? TablePropertyUtil.EMPTY_STRUCT : d.partition();
           partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
         }
 
         for (DataFile d : this.deleteFiles) {
-          partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
+          partitionData = keyedTable.spec().isUnpartitioned() ? TablePropertyUtil.EMPTY_STRUCT : d.partition();
           partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
         }
         addDeleteFiles.forEach(rowDelta::addDeletes);
@@ -168,12 +167,12 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
         }
 
         for (DeleteFile d : this.addDeleteFiles) {
-          partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
+          partitionData = keyedTable.spec().isUnpartitioned() ? TablePropertyUtil.EMPTY_STRUCT : d.partition();
           partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
         }
 
         for (DataFile d : this.deleteFiles) {
-          partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
+          partitionData = keyedTable.spec().isUnpartitioned() ? TablePropertyUtil.EMPTY_STRUCT : d.partition();
           partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
         }
         rewriteFiles.rewriteFiles(Collections.emptySet(), new HashSet<>(deleteDeleteFiles),
diff --git a/core/src/main/java/com/netease/arctic/op/PartitionPropertiesUpdate.java b/core/src/main/java/com/netease/arctic/op/PartitionPropertiesUpdate.java
index 4149d0620..830cd886c 100644
--- a/core/src/main/java/com/netease/arctic/op/PartitionPropertiesUpdate.java
+++ b/core/src/main/java/com/netease/arctic/op/PartitionPropertiesUpdate.java
@@ -22,6 +22,7 @@ import com.netease.arctic.table.TableProperties;
 import com.netease.arctic.table.UnkeyedTable;
 import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.UpdateProperties;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -34,6 +35,7 @@ import java.util.Set;
 public class PartitionPropertiesUpdate implements UpdatePartitionProperties {
 
   private final UnkeyedTable table;
+  private final Table transactionTable;
   private final StructLikeMap<Map<String, String>> setProperties;
   private final StructLikeMap<Set<String>> removeProperties;
   private final Transaction transaction;
@@ -43,6 +45,11 @@ public class PartitionPropertiesUpdate implements UpdatePartitionProperties {
     this.setProperties = StructLikeMap.create(table.spec().partitionType());
     this.removeProperties = StructLikeMap.create(table.spec().partitionType());
     this.transaction = transaction;
+    if (transaction != null) {
+      transactionTable = transaction.table();
+    } else {
+      transactionTable = null;
+    }
   }
 
   @Override
@@ -55,11 +62,7 @@ public class PartitionPropertiesUpdate implements UpdatePartitionProperties {
     Preconditions.checkArgument(removePropertiesForPartition == null ||
             !removePropertiesForPartition.contains(key),
         "Cannot remove and update the same key: %s", key);
-    Map<String, String> properties = setProperties.get(partitionData);
-    if (properties == null) {
-      properties = Maps.newHashMap();
-      setProperties.put(partitionData, properties);
-    }
+    Map<String, String> properties = setProperties.computeIfAbsent(partitionData, k -> Maps.newHashMap());
     properties.put(key, value);
     return this;
   }
@@ -84,13 +87,19 @@ public class PartitionPropertiesUpdate implements UpdatePartitionProperties {
 
   @Override
   public StructLikeMap<Map<String, String>> apply() {
-    StructLikeMap<Map<String, String>> partitionProperties = table.partitionProperty();
-    setProperties.forEach((partitionData, properties) -> {
-      Map<String, String> oldProperties = partitionProperties.get(partitionData);
-      if (oldProperties == null) {
-        oldProperties = Maps.newHashMap();
-        partitionProperties.put(partitionData, oldProperties);
+    StructLikeMap<Map<String, String>> partitionProperties;
+    if (transactionTable != null) {
+      String s = transactionTable.properties().get(TableProperties.TABLE_PARTITION_PROPERTIES);
+      if (s != null) {
+        partitionProperties = TablePropertyUtil.decodePartitionProperties(table.spec(), s);
+      } else {
+        partitionProperties = StructLikeMap.create(table.spec().partitionType());
       }
+    } else {
+      partitionProperties = table.partitionProperty();
+    }
+    setProperties.forEach((partitionData, properties) -> {
+      Map<String, String> oldProperties = partitionProperties.computeIfAbsent(partitionData, k -> Maps.newHashMap());
       oldProperties.putAll(properties);
     });
     removeProperties.forEach((partitionData, keys) -> {
diff --git a/core/src/main/java/com/netease/arctic/op/PartitionTransactionOperation.java b/core/src/main/java/com/netease/arctic/op/PartitionTransactionOperation.java
index 45e547222..d35465753 100644
--- a/core/src/main/java/com/netease/arctic/op/PartitionTransactionOperation.java
+++ b/core/src/main/java/com/netease/arctic/op/PartitionTransactionOperation.java
@@ -24,7 +24,6 @@ import com.netease.arctic.table.TableProperties;
 import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.iceberg.PendingUpdate;
 import org.apache.iceberg.Transaction;
-import org.apache.iceberg.UpdateProperties;
 import org.apache.iceberg.util.StructLikeMap;
 
 import java.util.HashMap;
@@ -62,7 +61,7 @@ public abstract class PartitionTransactionOperation implements PendingUpdate<Str
 
   @Override
   public StructLikeMap<Long> apply() {
-    return apply(tx, keyedTable.partitionMaxTransactionId());
+    return apply(tx, TablePropertyUtil.getPartitionMaxTransactionId(keyedTable));
   }
 
 
@@ -70,11 +69,10 @@ public abstract class PartitionTransactionOperation implements PendingUpdate<Str
     this.tx = keyedTable.baseTable().newTransaction();
 
     StructLikeMap<Long> partitionMaxTxId = apply();
-
-    String propertyValue = TablePropertyUtil.encodePartitionMaxTxId(keyedTable.spec(), partitionMaxTxId);
-    UpdateProperties updateProperties = tx.updateProperties();
-    updateProperties.set(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID, propertyValue);
-    updateProperties.commit();
+    UpdatePartitionProperties updatePartitionProperties = keyedTable.baseTable().updatePartitionProperties(tx);
+    partitionMaxTxId.forEach((partition, txId) ->
+        updatePartitionProperties.set(partition, TableProperties.BASE_TABLE_MAX_TRANSACTION_ID, String.valueOf(txId)));
+    updatePartitionProperties.commit();
 
     tx.commitTransaction();
   }
diff --git a/core/src/main/java/com/netease/arctic/scan/BaseKeyedTableScan.java b/core/src/main/java/com/netease/arctic/scan/BaseKeyedTableScan.java
index a439bfeb2..9663a925f 100644
--- a/core/src/main/java/com/netease/arctic/scan/BaseKeyedTableScan.java
+++ b/core/src/main/java/com/netease/arctic/scan/BaseKeyedTableScan.java
@@ -22,6 +22,7 @@ import com.netease.arctic.data.DataTreeNode;
 import com.netease.arctic.table.BaseKeyedTable;
 import com.netease.arctic.table.TableProperties;
 import com.netease.arctic.table.UnkeyedTable;
+import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.TableScan;
@@ -42,7 +43,6 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -116,9 +116,8 @@ public class BaseKeyedTableScan implements KeyedTableScan {
     split();
     LOG.info("planning table {} split end", table.id());
     // 3.combine node task (FileScanTask List -> CombinedScanTask)
-    CloseableIterable<CombinedScanTask> tasksIterable = combineNode(CloseableIterable.withNoopClose(splitTasks),
+    return combineNode(CloseableIterable.withNoopClose(splitTasks),
         splitSize, lookBack, openFileCost);
-    return tasksIterable;
   }
 
   private CloseableIterable<FileScanTask> planFiles(UnkeyedTable internalTable) {
@@ -131,9 +130,7 @@ public class BaseKeyedTableScan implements KeyedTableScan {
 
   private void split() {
     fileScanTasks.forEach((structLike, fileScanTasks1) -> {
-      Iterator<NodeFileScanTask> it = fileScanTasks1.iterator();
-      while (it.hasNext()) {
-        NodeFileScanTask task = it.next();
+      for (NodeFileScanTask task : fileScanTasks1) {
         if (task.cost() <= splitSize) {
           splitTasks.add(task);
           continue;
@@ -145,7 +142,7 @@ public class BaseKeyedTableScan implements KeyedTableScan {
         CloseableIterable<NodeFileScanTask> tasksIterable = splitNode(CloseableIterable.withNoopClose(task.dataTasks()),
             task.arcticEquityDeletes(), splitSize, lookBack, openFileCost);
         List<NodeFileScanTask> tasks =
-            org.apache.iceberg.relocated.com.google.common.collect.Lists.newArrayList(tasksIterable);
+            Lists.newArrayList(tasksIterable);
         splitTasks.addAll(tasks);
       }
     });
@@ -189,7 +186,7 @@ public class BaseKeyedTableScan implements KeyedTableScan {
     // planfiles() cannot guarantee the uniqueness of the file,
     // so Set<path> here is used to remove duplicate files
     Set<String> pathSets = new HashSet<>();
-    keyedTableTasks.stream().forEach(task -> {
+    keyedTableTasks.forEach(task -> {
       if (!pathSets.contains(task.file().path().toString())) {
         pathSets.add(task.file().path().toString());
         DataTreeNode treeNode = task.file().node();
@@ -230,14 +227,14 @@ public class BaseKeyedTableScan implements KeyedTableScan {
       List<ArcticFileScanTask> baseTasks) {
     ListMultimap<StructLike, ArcticFileScanTask> filesGroupedByPartition
         = Multimaps.newListMultimap(Maps.newHashMap(), Lists::newArrayList);
-    StructLikeMap<Long> partitionMaxTxId = table.partitionMaxTransactionId();
+    StructLikeMap<Long> partitionMaxTxId = TablePropertyUtil.getPartitionMaxTransactionId(table);
 
     // filter change files according to max transaction id
     changeTasks.forEach(task -> {
       StructLike structLike = task.file().partition();
-      Long txId = null;
+      Long txId;
       if (structLike.size() == 0) {
-        txId = partitionMaxTxId.get(null);
+        txId = partitionMaxTxId.get(TablePropertyUtil.EMPTY_STRUCT);
       } else {
         txId = partitionMaxTxId.get(task.file().partition());
       }
@@ -248,9 +245,7 @@ public class BaseKeyedTableScan implements KeyedTableScan {
       }
     });
 
-    baseTasks.forEach(task -> {
-      filesGroupedByPartition.put(task.file().partition(), task);
-    });
+    baseTasks.forEach(task -> filesGroupedByPartition.put(task.file().partition(), task));
     return filesGroupedByPartition.asMap();
   }
 }
\ No newline at end of file
diff --git a/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java b/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
index 40cacde93..615cb346b 100644
--- a/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
@@ -18,8 +18,6 @@
 
 package com.netease.arctic.table;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.netease.arctic.AmsClient;
 import com.netease.arctic.ams.api.TableMeta;
 import com.netease.arctic.io.ArcticFileIO;
@@ -30,23 +28,14 @@ import com.netease.arctic.op.RewritePartitions;
 import com.netease.arctic.op.UpdateKeyedTableProperties;
 import com.netease.arctic.scan.BaseKeyedTableScan;
 import com.netease.arctic.scan.KeyedTableScan;
-import com.netease.arctic.trace.AmsTableTracer;
-import com.netease.arctic.trace.TracedSchemaUpdate;
-import com.netease.arctic.trace.TracedUpdateProperties;
-import com.netease.arctic.trace.TrackerOperations;
-import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.UpdateProperties;
 import org.apache.iceberg.UpdateSchema;
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
-import org.apache.iceberg.util.StructLikeMap;
 import org.apache.thrift.TException;
 
-import java.util.Collections;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 /**
  * Base implementation of {@link KeyedTable}, wrapping a {@link BaseTable} and a {@link ChangeTable}.
@@ -174,35 +163,6 @@ public class BaseKeyedTable implements KeyedTable {
     return name();
   }
 
-  @Override
-  public Map<String, Long> maxTransactionId() {
-    String s = baseTable.properties().get(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID);
-    if (s != null) {
-      try {
-        Map<String, Long> results = Maps.newHashMap();
-        Map map = new ObjectMapper().readValue(s, Map.class);
-        for (Object key : map.keySet()) {
-          results.put(key.toString(), Long.parseLong(map.get(key).toString()));
-        }
-        return results;
-      } catch (JsonProcessingException e) {
-        throw new UnsupportedOperationException("Failed to get " + TableProperties.BASE_TABLE_MAX_TRANSACTION_ID, e);
-      }
-    } else {
-      return Collections.emptyMap();
-    }
-  }
-
-  @Override
-  public StructLikeMap<Long> partitionMaxTransactionId() {
-    String s = baseTable.properties().get(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID);
-    if (s != null) {
-      return TablePropertyUtil.decodePartitionMaxTxId(spec(), s);
-    } else {
-      return StructLikeMap.create(spec().partitionType());
-    }
-  }
-
   @Override
   public OverwriteBaseFiles newOverwriteBaseFiles() {
     return new OverwriteBaseFiles(this);
diff --git a/core/src/main/java/com/netease/arctic/table/KeyedTable.java b/core/src/main/java/com/netease/arctic/table/KeyedTable.java
index eed9f0479..8bf1e71b0 100644
--- a/core/src/main/java/com/netease/arctic/table/KeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/KeyedTable.java
@@ -21,9 +21,6 @@ package com.netease.arctic.table;
 import com.netease.arctic.op.OverwriteBaseFiles;
 import com.netease.arctic.op.RewritePartitions;
 import com.netease.arctic.scan.KeyedTableScan;
-import org.apache.iceberg.util.StructLikeMap;
-
-import java.util.Map;
 
 /**
  * Represents an arctic table with keys supported, consist of one {@link ChangeTable} and one {@link BaseTable}.
@@ -75,24 +72,6 @@ public interface KeyedTable extends ArcticTable {
    */
   long beginTransaction(String signature);
 
-  /**
-   * get max transactionId of each partition. use {@link #partitionMaxTransactionId()} instead.
-   *
-   * @return map of max transactionId of each partition
-   * @deprecated use {@link UnkeyedTable#partitionProperty()} instead.
-   */
-  @Deprecated
-  Map<String, Long> maxTransactionId();
-
-  /**
-   * get max transactionId of each partition
-   *
-   * @return map of max transactionId of each partition
-   * @deprecated use {@link UnkeyedTable#partitionProperty()} instead.
-   */
-  @Deprecated
-  StructLikeMap<Long> partitionMaxTransactionId();
-
   @Override
   default boolean isKeyedTable() {
     return true;
diff --git a/core/src/main/java/com/netease/arctic/utils/TablePropertyUtil.java b/core/src/main/java/com/netease/arctic/utils/TablePropertyUtil.java
index 12b9d6d6a..a92f0c215 100644
--- a/core/src/main/java/com/netease/arctic/utils/TablePropertyUtil.java
+++ b/core/src/main/java/com/netease/arctic/utils/TablePropertyUtil.java
@@ -21,6 +21,8 @@ package com.netease.arctic.utils;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.netease.arctic.table.KeyedTable;
+import com.netease.arctic.table.TableProperties;
 import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
@@ -39,28 +41,6 @@ public class TablePropertyUtil {
 
   public static final StructLike EMPTY_STRUCT = GenericRecord.create(new Schema());
 
-  /**
-   * Encode max transaction id map of each partition to string.
-   *
-   * @param spec table partition spec
-   * @param partitionMaxTransactionId max transaction id map of each partition
-   * @return encoded string
-   */
-  public static String encodePartitionMaxTxId(PartitionSpec spec, StructLikeMap<Long> partitionMaxTransactionId) {
-    Map<String, Long> pathPartitionMaxId = Maps.newHashMap();
-    for (StructLike pd : partitionMaxTransactionId.keySet()) {
-      String pathLike = spec.partitionToPath(pd);
-      pathPartitionMaxId.put(pathLike, partitionMaxTransactionId.get(pd));
-    }
-    String value;
-    try {
-      value = new ObjectMapper().writeValueAsString(pathPartitionMaxId);
-    } catch (JsonProcessingException e) {
-      throw new UncheckedIOException(e);
-    }
-    return value;
-  }
-
   /**
    * Decode string to max transaction id map of each partition.
    *
@@ -122,4 +102,18 @@ public class TablePropertyUtil {
     }
     return value;
   }
+
+  public static StructLikeMap<Long> getPartitionMaxTransactionId(KeyedTable keyedTable) {
+    StructLikeMap<Long> baseTableMaxTransactionId = StructLikeMap.create(keyedTable.spec().partitionType());
+
+    StructLikeMap<Map<String, String>> partitionProperty = keyedTable.asKeyedTable().baseTable().partitionProperty();
+    partitionProperty.forEach((partitionKey, propertyValue) -> {
+      Long maxTxId = (propertyValue == null ||
+          propertyValue.get(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID) == null) ?
+          0 : Long.parseLong(propertyValue.get(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID));
+      baseTableMaxTransactionId.put(partitionKey, maxTxId);
+    });
+
+    return baseTableMaxTransactionId;
+  }
 }
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/reader/RowDataReaderFunction.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/reader/RowDataReaderFunction.java
index c2edccfcc..7a68c75e6 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/reader/RowDataReaderFunction.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/reader/RowDataReaderFunction.java
@@ -49,6 +49,7 @@ public class RowDataReaderFunction extends DataIteratorReaderFunction<RowData> {
   private final boolean caseSensitive;
   private final ArcticFileIO io;
   private final PrimaryKeySpec primaryKeySpec;
+  private final int columnSize;
 
   public RowDataReaderFunction(
       ReadableConfig config, Schema tableSchema, Schema projectedSchema, PrimaryKeySpec primaryKeySpec,
@@ -61,6 +62,8 @@ public class RowDataReaderFunction extends DataIteratorReaderFunction<RowData> {
     this.nameMapping = nameMapping;
     this.caseSensitive = caseSensitive;
     this.io = io;
+    // Add file offset column after readSchema. Refer to this#wrapArcticFileOffsetColumnMeta
+    this.columnSize = readSchema.columns().size();
   }
 
   @Override
@@ -99,12 +102,14 @@ public class RowDataReaderFunction extends DataIteratorReaderFunction<RowData> {
   }
 
   long arcticFileOffset(RowData rowData) {
-    int index = rowData.getArity() - 1;
-    return rowData.getLong(index);
+    return rowData.getLong(columnSize);
   }
 
+  /**
+   * @param rowData It may have more columns than readSchema. Refer to {@link FlinkArcticDataReader}'s annotation.
+   */
   RowData removeArcticMetaColumn(RowData rowData) {
-    GenericRowData newRowData = new GenericRowData(rowData.getRowKind(), rowData.getArity() - 1);
+    GenericRowData newRowData = new GenericRowData(rowData.getRowKind(), columnSize);
     if (rowData instanceof GenericRowData) {
       GenericRowData before = (GenericRowData) rowData;
       for (int i = 0; i < newRowData.getArity(); i++) {
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/source/FlinkArcticDataReader.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/source/FlinkArcticDataReader.java
index c27410695..303281a90 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/source/FlinkArcticDataReader.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/source/FlinkArcticDataReader.java
@@ -44,6 +44,10 @@ import java.util.function.Function;
 
 /**
  * This is an arctic reader accepts a {@link FileScanTask} and produces a {@link CloseableIterator<RowData>}.
+ * The RowData read from this reader may have more columns than the original schema.
+ * The additional columns are added after the original columns,
+ * see {@link com.netease.arctic.iceberg.optimize.DeleteFilter}.
+ * It shall be projected before sent to downstream. This can be processed in {@link DataIterator#next()}
  */
 public class FlinkArcticDataReader extends BaseIcebergDataReader<RowData> implements FileScanTaskReader<RowData> {
   private static final long serialVersionUID = -6773693031945244386L;
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
index 9bde044c6..88bd8ae9a 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
@@ -22,7 +22,7 @@ import com.netease.arctic.hive.io.writer.AdaptHiveOperateToTableRelation;
 import com.netease.arctic.hive.io.writer.AdaptHiveOutputFileFactory;
 import com.netease.arctic.hive.table.HiveLocationKind;
 import com.netease.arctic.hive.table.SupportHive;
-import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.io.writer.CommonOutputFileFactory;
 import com.netease.arctic.io.writer.OutputFileFactory;
 import com.netease.arctic.io.writer.SortedPosDeleteWriter;
@@ -143,7 +143,7 @@ public class FlinkTaskWriterBuilder implements TaskWriterBuilder<RowData> {
             encryptionManager, partitionId, taskId, transactionId) :
         new CommonOutputFileFactory(baseLocation, table.spec(), fileFormat, table.io(),
             encryptionManager, partitionId, taskId, transactionId);
-    FileAppenderFactory<RowData> appenderFactory = HiveTableUtil.isHive(table) ?
+    FileAppenderFactory<RowData> appenderFactory = TableTypeUtil.isHive(table) ?
         new AdaptHiveFlinkAppenderFactory(schema, flinkSchema, table.properties(), table.spec()) :
         new FlinkAppenderFactory(
             schema, flinkSchema, table.properties(), table.spec());
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/reader/RowDataReaderFunction.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/reader/RowDataReaderFunction.java
index c743b8b00..101b29c78 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/reader/RowDataReaderFunction.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/hybrid/reader/RowDataReaderFunction.java
@@ -50,6 +50,7 @@ public class RowDataReaderFunction extends DataIteratorReaderFunction<RowData> {
   private final boolean caseSensitive;
   private final ArcticFileIO io;
   private final PrimaryKeySpec primaryKeySpec;
+  private final int columnSize;
 
   public RowDataReaderFunction(
       ReadableConfig config, Schema tableSchema, Schema projectedSchema, PrimaryKeySpec primaryKeySpec,
@@ -62,6 +63,8 @@ public class RowDataReaderFunction extends DataIteratorReaderFunction<RowData> {
     this.nameMapping = nameMapping;
     this.caseSensitive = caseSensitive;
     this.io = io;
+    // Add file offset column after readSchema. Refer to this#wrapArcticFileOffsetColumnMeta
+    this.columnSize = readSchema.columns().size();
   }
 
   @Override
@@ -100,12 +103,14 @@ public class RowDataReaderFunction extends DataIteratorReaderFunction<RowData> {
   }
 
   long arcticFileOffset(RowData rowData) {
-    int index = rowData.getArity() - 1;
-    return rowData.getLong(index);
+    return rowData.getLong(columnSize);
   }
 
+  /**
+   * @param rowData It may have more columns than readSchema. Refer to {@link FlinkArcticDataReader}'s annotation.
+   */
   RowData removeArcticMetaColumn(RowData rowData) {
-    GenericRowData newRowData = new GenericRowData(rowData.getRowKind(), rowData.getArity() - 1);
+    GenericRowData newRowData = new GenericRowData(rowData.getRowKind(), columnSize);
     if (rowData instanceof GenericRowData) {
       GenericRowData before = (GenericRowData) rowData;
       for (int i = 0; i < newRowData.getArity(); i++) {
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/source/FlinkArcticDataReader.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/source/FlinkArcticDataReader.java
index 6a3f0399e..9621d491f 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/source/FlinkArcticDataReader.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/read/source/FlinkArcticDataReader.java
@@ -45,6 +45,10 @@ import java.util.function.Function;
 
 /**
  * This is an arctic reader accepts a {@link FileScanTask} and produces a {@link CloseableIterator<RowData>}.
+ * The RowData read from this reader may have more columns than the original schema.
+ * The additional columns are added after the original columns,
+ * see {@link com.netease.arctic.iceberg.optimize.DeleteFilter}.
+ * It shall be projected before sent to downstream. This can be processed in {@link DataIterator#next()}
  */
 public class FlinkArcticDataReader extends BaseIcebergDataReader<RowData> implements FileScanTaskReader<RowData> {
   private static final long serialVersionUID = -6773693031945244386L;
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
index 9bde044c6..88bd8ae9a 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
@@ -22,7 +22,7 @@ import com.netease.arctic.hive.io.writer.AdaptHiveOperateToTableRelation;
 import com.netease.arctic.hive.io.writer.AdaptHiveOutputFileFactory;
 import com.netease.arctic.hive.table.HiveLocationKind;
 import com.netease.arctic.hive.table.SupportHive;
-import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.io.writer.CommonOutputFileFactory;
 import com.netease.arctic.io.writer.OutputFileFactory;
 import com.netease.arctic.io.writer.SortedPosDeleteWriter;
@@ -143,7 +143,7 @@ public class FlinkTaskWriterBuilder implements TaskWriterBuilder<RowData> {
             encryptionManager, partitionId, taskId, transactionId) :
         new CommonOutputFileFactory(baseLocation, table.spec(), fileFormat, table.io(),
             encryptionManager, partitionId, taskId, transactionId);
-    FileAppenderFactory<RowData> appenderFactory = HiveTableUtil.isHive(table) ?
+    FileAppenderFactory<RowData> appenderFactory = TableTypeUtil.isHive(table) ?
         new AdaptHiveFlinkAppenderFactory(schema, flinkSchema, table.properties(), table.spec()) :
         new FlinkAppenderFactory(
             schema, flinkSchema, table.properties(), table.spec());
diff --git a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/reader/RowDataReaderFunction.java b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/reader/RowDataReaderFunction.java
index c743b8b00..101b29c78 100644
--- a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/reader/RowDataReaderFunction.java
+++ b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/hybrid/reader/RowDataReaderFunction.java
@@ -50,6 +50,7 @@ public class RowDataReaderFunction extends DataIteratorReaderFunction<RowData> {
   private final boolean caseSensitive;
   private final ArcticFileIO io;
   private final PrimaryKeySpec primaryKeySpec;
+  private final int columnSize;
 
   public RowDataReaderFunction(
       ReadableConfig config, Schema tableSchema, Schema projectedSchema, PrimaryKeySpec primaryKeySpec,
@@ -62,6 +63,8 @@ public class RowDataReaderFunction extends DataIteratorReaderFunction<RowData> {
     this.nameMapping = nameMapping;
     this.caseSensitive = caseSensitive;
     this.io = io;
+    // Add file offset column after readSchema. Refer to this#wrapArcticFileOffsetColumnMeta
+    this.columnSize = readSchema.columns().size();
   }
 
   @Override
@@ -100,12 +103,14 @@ public class RowDataReaderFunction extends DataIteratorReaderFunction<RowData> {
   }
 
   long arcticFileOffset(RowData rowData) {
-    int index = rowData.getArity() - 1;
-    return rowData.getLong(index);
+    return rowData.getLong(columnSize);
   }
 
+  /**
+   * @param rowData It may have more columns than readSchema. Refer to {@link FlinkArcticDataReader}'s annotation.
+   */
   RowData removeArcticMetaColumn(RowData rowData) {
-    GenericRowData newRowData = new GenericRowData(rowData.getRowKind(), rowData.getArity() - 1);
+    GenericRowData newRowData = new GenericRowData(rowData.getRowKind(), columnSize);
     if (rowData instanceof GenericRowData) {
       GenericRowData before = (GenericRowData) rowData;
       for (int i = 0; i < newRowData.getArity(); i++) {
diff --git a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/source/FlinkArcticDataReader.java b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/source/FlinkArcticDataReader.java
index 6a3f0399e..9621d491f 100644
--- a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/source/FlinkArcticDataReader.java
+++ b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/read/source/FlinkArcticDataReader.java
@@ -45,6 +45,10 @@ import java.util.function.Function;
 
 /**
  * This is an arctic reader accepts a {@link FileScanTask} and produces a {@link CloseableIterator<RowData>}.
+ * The RowData read from this reader may have more columns than the original schema.
+ * The additional columns are added after the original columns,
+ * see {@link com.netease.arctic.iceberg.optimize.DeleteFilter}.
+ * It shall be projected before sent to downstream. This can be processed in {@link DataIterator#next()}
  */
 public class FlinkArcticDataReader extends BaseIcebergDataReader<RowData> implements FileScanTaskReader<RowData> {
   private static final long serialVersionUID = -6773693031945244386L;
diff --git a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
index 9bde044c6..88bd8ae9a 100644
--- a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
+++ b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
@@ -22,7 +22,7 @@ import com.netease.arctic.hive.io.writer.AdaptHiveOperateToTableRelation;
 import com.netease.arctic.hive.io.writer.AdaptHiveOutputFileFactory;
 import com.netease.arctic.hive.table.HiveLocationKind;
 import com.netease.arctic.hive.table.SupportHive;
-import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.io.writer.CommonOutputFileFactory;
 import com.netease.arctic.io.writer.OutputFileFactory;
 import com.netease.arctic.io.writer.SortedPosDeleteWriter;
@@ -143,7 +143,7 @@ public class FlinkTaskWriterBuilder implements TaskWriterBuilder<RowData> {
             encryptionManager, partitionId, taskId, transactionId) :
         new CommonOutputFileFactory(baseLocation, table.spec(), fileFormat, table.io(),
             encryptionManager, partitionId, taskId, transactionId);
-    FileAppenderFactory<RowData> appenderFactory = HiveTableUtil.isHive(table) ?
+    FileAppenderFactory<RowData> appenderFactory = TableTypeUtil.isHive(table) ?
         new AdaptHiveFlinkAppenderFactory(schema, flinkSchema, table.properties(), table.spec()) :
         new FlinkAppenderFactory(
             schema, flinkSchema, table.properties(), table.spec());
diff --git a/hive/src/main/java/com/netease/arctic/hive/HiveTableProperties.java b/hive/src/main/java/com/netease/arctic/hive/HiveTableProperties.java
index 6b9bf46d9..cbea554bb 100644
--- a/hive/src/main/java/com/netease/arctic/hive/HiveTableProperties.java
+++ b/hive/src/main/java/com/netease/arctic/hive/HiveTableProperties.java
@@ -47,5 +47,9 @@ public class HiveTableProperties {
   public static final boolean AUTO_SYNC_HIVE_DATA_WRITE_DEFAULT = false;
 
   public static final String ALLOW_HIVE_TABLE_EXISTED = "allow-hive-table-existed";
+
+  public static final String PARQUET_INPUT_FORMAT = "org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat";
+  public static final String PARQUET_OUTPUT_FORMAT = "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat";
+  public static final String PARQUET_ROW_FORMAT_SERDE = "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe";
 }
 
diff --git a/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveGenericTaskWriterBuilder.java b/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveGenericTaskWriterBuilder.java
index d5460e569..3ab8f48ac 100644
--- a/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveGenericTaskWriterBuilder.java
+++ b/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveGenericTaskWriterBuilder.java
@@ -21,7 +21,7 @@ package com.netease.arctic.hive.io.writer;
 import com.netease.arctic.data.ChangeAction;
 import com.netease.arctic.hive.table.HiveLocationKind;
 import com.netease.arctic.hive.table.SupportHive;
-import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.io.writer.CommonOutputFileFactory;
 import com.netease.arctic.io.writer.GenericBaseTaskWriter;
 import com.netease.arctic.io.writer.GenericChangeTaskWriter;
@@ -163,7 +163,7 @@ public class AdaptHiveGenericTaskWriterBuilder implements TaskWriterBuilder<Reco
             encryptionManager, partitionId, taskId, transactionId) :
         new CommonOutputFileFactory(baseLocation, table.spec(), fileFormat, table.io(),
             encryptionManager, partitionId, taskId, transactionId);
-    FileAppenderFactory<Record> appenderFactory = HiveTableUtil.isHive(table) ?
+    FileAppenderFactory<Record> appenderFactory = TableTypeUtil.isHive(table) ?
         new AdaptHiveGenericAppenderFactory(schema, table.spec()) :
         new GenericAppenderFactory(schema, table.spec());
     return new GenericBaseTaskWriter(fileFormat, appenderFactory,
@@ -186,7 +186,7 @@ public class AdaptHiveGenericTaskWriterBuilder implements TaskWriterBuilder<Reco
     long mask = PropertyUtil.propertyAsLong(table.properties(), TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET,
         TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET_DEFAULT) - 1;
     Schema changeWriteSchema = SchemaUtil.changeWriteSchema(table.changeTable().schema());
-    FileAppenderFactory<Record> appenderFactory = HiveTableUtil.isHive(table) ?
+    FileAppenderFactory<Record> appenderFactory = TableTypeUtil.isHive(table) ?
         new AdaptHiveGenericAppenderFactory(changeWriteSchema, table.spec()) :
         new GenericAppenderFactory(changeWriteSchema, table.spec());
     return new GenericChangeTaskWriter(fileFormat,
diff --git a/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOperateToTableRelation.java b/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOperateToTableRelation.java
index a84000051..afd8e014a 100644
--- a/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOperateToTableRelation.java
+++ b/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOperateToTableRelation.java
@@ -19,7 +19,7 @@
 package com.netease.arctic.hive.io.writer;
 
 import com.netease.arctic.hive.table.HiveLocationKind;
-import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.BaseLocationKind;
 import com.netease.arctic.table.ChangeLocationKind;
@@ -34,7 +34,7 @@ public class AdaptHiveOperateToTableRelation implements OperateToTableRelation {
   public LocationKind getLocationKindsFromOperateKind(
       ArcticTable arcticTable, WriteOperationKind writeOperationKind) {
     if (arcticTable.isKeyedTable()) {
-      if (HiveTableUtil.isHive(arcticTable)) {
+      if (TableTypeUtil.isHive(arcticTable)) {
         switch (writeOperationKind) {
           case APPEND:
             return ChangeLocationKind.INSTANT;
@@ -57,7 +57,7 @@ public class AdaptHiveOperateToTableRelation implements OperateToTableRelation {
         }
       }
     } else {
-      if (HiveTableUtil.isHive(arcticTable)) {
+      if (TableTypeUtil.isHive(arcticTable)) {
         switch (writeOperationKind) {
           case APPEND:
           case MAJOR_OPTIMIZE:
diff --git a/hive/src/main/java/com/netease/arctic/hive/op/HiveOperationTransaction.java b/hive/src/main/java/com/netease/arctic/hive/op/HiveOperationTransaction.java
index 8efe38aa0..c04e91f1d 100644
--- a/hive/src/main/java/com/netease/arctic/hive/op/HiveOperationTransaction.java
+++ b/hive/src/main/java/com/netease/arctic/hive/op/HiveOperationTransaction.java
@@ -176,79 +176,85 @@ public class HiveOperationTransaction implements Transaction {
 
   private class TransactionalTable implements Table {
 
+    Table transactionTable;
+
+    public TransactionalTable() {
+      transactionTable = wrapped.table();
+    }
+
     @Override
     public String name() {
-      return unkeyedHiveTable.name();
+      return transactionTable.name();
     }
 
     @Override
     public void refresh() {
-
+      transactionTable.refresh();
     }
 
     @Override
     public TableScan newScan() {
-      throw new UnsupportedOperationException("Transaction tables do not support scans");
+      return transactionTable.newScan();
     }
 
     @Override
     public Schema schema() {
-      return unkeyedHiveTable.schema();
+      return transactionTable.schema();
     }
 
     @Override
     public Map<Integer, Schema> schemas() {
-      return unkeyedHiveTable.schemas();
+      return transactionTable.schemas();
     }
 
     @Override
     public PartitionSpec spec() {
-      return unkeyedHiveTable.spec();
+      return transactionTable.spec();
     }
 
     @Override
     public Map<Integer, PartitionSpec> specs() {
-      return unkeyedHiveTable.specs();
+      return transactionTable.specs();
     }
 
     @Override
     public SortOrder sortOrder() {
-      return unkeyedHiveTable.sortOrder();
+      return transactionTable.sortOrder();
     }
 
     @Override
     public Map<Integer, SortOrder> sortOrders() {
-      return unkeyedHiveTable.sortOrders();
+      return transactionTable.sortOrders();
     }
 
     @Override
     public Map<String, String> properties() {
-      return unkeyedHiveTable.properties();
+      return transactionTable.properties();
     }
 
     @Override
     public String location() {
-      return unkeyedHiveTable.location();
+      return transactionTable.location();
     }
 
     @Override
     public Snapshot currentSnapshot() {
-      return unkeyedHiveTable.currentSnapshot();
+      return transactionTable.currentSnapshot();
     }
 
     @Override
     public Snapshot snapshot(long snapshotId) {
-      return unkeyedHiveTable.currentSnapshot();
+      return transactionTable.snapshot(snapshotId);
     }
 
     @Override
     public Iterable<Snapshot> snapshots() {
-      return unkeyedHiveTable.snapshots();
+      return transactionTable.snapshots();
     }
 
     @Override
     public List<HistoryEntry> history() {
-      return unkeyedHiveTable.history();
+      return transactionTable.history();
     }
 
     @Override
@@ -338,17 +344,17 @@ public class HiveOperationTransaction implements Transaction {
 
     @Override
     public FileIO io() {
-      return unkeyedHiveTable.io();
+      return transactionTable.io();
     }
 
     @Override
     public EncryptionManager encryption() {
-      return unkeyedHiveTable.encryption();
+      return transactionTable.encryption();
     }
 
     @Override
     public LocationProvider locationProvider() {
-      return unkeyedHiveTable.locationProvider();
+      return transactionTable.locationProvider();
     }
   }
 }
diff --git a/hive/src/main/java/com/netease/arctic/hive/utils/HiveTableUtil.java b/hive/src/main/java/com/netease/arctic/hive/utils/HiveTableUtil.java
index 0bb1c4cfd..021657c5f 100644
--- a/hive/src/main/java/com/netease/arctic/hive/utils/HiveTableUtil.java
+++ b/hive/src/main/java/com/netease/arctic/hive/utils/HiveTableUtil.java
@@ -46,10 +46,6 @@ public class HiveTableUtil {
 
   private static final Logger LOG = LoggerFactory.getLogger(HiveTableUtil.class);
 
-  public static boolean isHive(ArcticTable arcticTable) {
-    return arcticTable instanceof SupportHive;
-  }
-
   public static org.apache.hadoop.hive.metastore.api.Table loadHmsTable(
       HMSClient hiveClient, TableIdentifier tableIdentifier) {
     try {
@@ -126,9 +122,9 @@ public class HiveTableUtil {
     SerDeInfo serDeInfo = new SerDeInfo();
     switch (format) {
       case PARQUET:
-        storageDescriptor.setOutputFormat("org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat");
-        storageDescriptor.setInputFormat("org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat");
-        serDeInfo.setSerializationLib("org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe");
+        storageDescriptor.setOutputFormat(HiveTableProperties.PARQUET_OUTPUT_FORMAT);
+        storageDescriptor.setInputFormat(HiveTableProperties.PARQUET_INPUT_FORMAT);
+        serDeInfo.setSerializationLib(HiveTableProperties.PARQUET_ROW_FORMAT_SERDE);
         break;
       default:
         throw new IllegalArgumentException("Unsupported hive table file format:" + format);
diff --git a/hive/src/main/java/com/netease/arctic/hive/utils/TableTypeUtil.java b/hive/src/main/java/com/netease/arctic/hive/utils/TableTypeUtil.java
new file mode 100644
index 000000000..30d9782b8
--- /dev/null
+++ b/hive/src/main/java/com/netease/arctic/hive/utils/TableTypeUtil.java
@@ -0,0 +1,10 @@
+package com.netease.arctic.hive.utils;
+
+import com.netease.arctic.hive.table.SupportHive;
+import com.netease.arctic.table.ArcticTable;
+
+public class TableTypeUtil {
+  public static boolean isHive(ArcticTable arcticTable) {
+    return arcticTable instanceof SupportHive;
+  }
+}
diff --git a/hive/src/main/java/com/netease/arctic/hive/utils/UpgradeHiveTableUtil.java b/hive/src/main/java/com/netease/arctic/hive/utils/UpgradeHiveTableUtil.java
index 0a2bddc3a..215213675 100644
--- a/hive/src/main/java/com/netease/arctic/hive/utils/UpgradeHiveTableUtil.java
+++ b/hive/src/main/java/com/netease/arctic/hive/utils/UpgradeHiveTableUtil.java
@@ -18,6 +18,7 @@
 
 package com.netease.arctic.hive.utils;
 
+import com.netease.arctic.hive.HMSClient;
 import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.catalog.ArcticHiveCatalog;
 import com.netease.arctic.table.ArcticTable;
@@ -25,6 +26,8 @@ import com.netease.arctic.table.PrimaryKeySpec;
 import com.netease.arctic.table.TableIdentifier;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
 import org.apache.hadoop.hive.metastore.api.Table;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
@@ -35,6 +38,7 @@ import java.io.IOException;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class UpgradeHiveTableUtil {
 
@@ -52,6 +56,9 @@ public class UpgradeHiveTableUtil {
    */
   public static void upgradeHiveTable(ArcticHiveCatalog arcticHiveCatalog, TableIdentifier tableIdentifier,
                                       List<String> pkList, Map<String, String> properties) throws Exception {
+    if (!formatCheck(arcticHiveCatalog.getHMSClient(), tableIdentifier)) {
+      throw new IllegalArgumentException("Only support storage format is parquet");
+    }
     boolean upgradeHive = false;
     try {
       Table hiveTable = HiveTableUtil.loadHmsTable(arcticHiveCatalog.getHMSClient(), tableIdentifier);
@@ -126,4 +133,39 @@ public class UpgradeHiveTableUtil {
     }
     HiveMetaSynchronizer.syncHiveDataToArctic(arcticTable, arcticHiveCatalog.getHMSClient());
   }
+
+  /**
+   * Check whether Arctic supports the hive table storage formats.
+   *
+   * @param hiveClient Hive client from ArcticHiveCatalog
+   * @param tableIdentifier A table identifier
+   * @return Support or not
+   */
+  private static boolean formatCheck(HMSClient hiveClient, TableIdentifier tableIdentifier) throws IOException {
+    AtomicBoolean isSupport = new AtomicBoolean(false);
+    try {
+      hiveClient.run(client -> {
+        Table hiveTable = HiveTableUtil.loadHmsTable(hiveClient, tableIdentifier);
+        StorageDescriptor storageDescriptor = hiveTable.getSd();
+        SerDeInfo serDeInfo = storageDescriptor.getSerdeInfo();
+        switch (storageDescriptor.getInputFormat()) {
+          case HiveTableProperties.PARQUET_INPUT_FORMAT:
+            if (storageDescriptor.getOutputFormat().equals(HiveTableProperties.PARQUET_OUTPUT_FORMAT) &&
+                serDeInfo.getSerializationLib().equals(HiveTableProperties.PARQUET_ROW_FORMAT_SERDE)) {
+              isSupport.set(true);
+            } else {
+              throw new IllegalStateException("Please check your hive table storage format is right");
+            }
+            break;
+          default:
+            isSupport.set(false);
+            break;
+        }
+        return null;
+      });
+    } catch (Exception e) {
+      throw new IOException(e);
+    }
+    return isSupport.get();
+  }
 }
diff --git a/site/docs/ch/meta-service/dashboard.md b/site/docs/ch/meta-service/dashboard.md
index cbae01665..e55eb696e 100644
--- a/site/docs/ch/meta-service/dashboard.md
+++ b/site/docs/ch/meta-service/dashboard.md
@@ -34,7 +34,7 @@ AMS安装包中提供了脚本文件`bin/ams.sh`用以处理AMS的日常运维
 AMS完成启动后即可登录 [AMS Dashboard](http://localhost:1630) 来访问AMS的页面，默认的用户名密码为：`admin/admin`。
 
 ## 使用 MySQL 作为系统库
-AMS 默认使用 Derby 作为系统库存储自己的元数据，在生产环境下我们建议换成MySQL以提升系统的高可用。
+AMS 默认使用 Derby 作为系统库存储自己的元数据，在生产环境下我们建议换成MySQL以提升系统的高可用。支持Mysql 5.x到8.0版本。
 
 **1.修改配置文件**
 
@@ -48,9 +48,6 @@ arctic.ams.mybatis.ConnectionPassword: {password}                        #MySQL
 arctic.ams.database.type: mysql                                          #系统库类型
 ```
 
-???+ 注意
-
-    目前只支持 MySQL 5.x 版本，不支持 MySQL 8。
 
 **2.初始化 MySQL 表**
 
diff --git a/trino/src/main/java/com/netease/arctic/trino/unkeyed/IcebergMetadata.java b/trino/src/main/java/com/netease/arctic/trino/unkeyed/IcebergMetadata.java
index cc76419c7..7b660e711 100644
--- a/trino/src/main/java/com/netease/arctic/trino/unkeyed/IcebergMetadata.java
+++ b/trino/src/main/java/com/netease/arctic/trino/unkeyed/IcebergMetadata.java
@@ -27,7 +27,7 @@ import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Sets;
-import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.table.ArcticTable;
 import io.airlift.json.JsonCodec;
 import io.airlift.log.Logger;
@@ -290,7 +290,7 @@ public class IcebergMetadata
 
     Map<String, String> tableProperties = table.properties();
     String nameMappingJson = tableProperties.get(TableProperties.DEFAULT_NAME_MAPPING);
-    if (HiveTableUtil.isHive((ArcticTable) table)) {
+    if (TableTypeUtil.isHive((ArcticTable) table)) {
       return new AdaptHiveIcebergTableHandle(
           tableName.getSchemaName(),
           name.getTableName(),


Test patch:
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestExpiredFileClean.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestExpiredFileClean.java
index 0c95c5469..46ea7ca2e 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestExpiredFileClean.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestExpiredFileClean.java
@@ -18,7 +18,6 @@
 
 package com.netease.arctic.ams.server.optimize;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.netease.arctic.TableTestBase;
 import com.netease.arctic.ams.api.DataFileInfo;
 import com.netease.arctic.ams.server.service.impl.TableExpireService;
@@ -26,10 +25,12 @@ import com.netease.arctic.ams.server.util.DataFileInfoUtils;
 import com.netease.arctic.data.ChangeAction;
 import com.netease.arctic.io.writer.GenericChangeTaskWriter;
 import com.netease.arctic.io.writer.GenericTaskWriters;
+import com.netease.arctic.op.UpdatePartitionProperties;
 import com.netease.arctic.table.TableProperties;
 import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.ContentFile;
 import org.apache.iceberg.DataFile;
-import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.StructLike;
 import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.io.WriteResult;
@@ -43,10 +44,8 @@ import java.io.IOException;
 import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.stream.Collectors;
 
 public class TestExpiredFileClean extends TableTestBase {
@@ -55,14 +54,13 @@ public class TestExpiredFileClean extends TableTestBase {
 
   @Test
   public void testDeleteChangeFiles() throws Exception {
-    insertChangeDataFiles(1);
-
-    Map<String, Long> transactionIdMap = new HashMap<>();
-    transactionIdMap.put("op_time_day=2022-01-01", 2L);
-    transactionIdMap.put("op_time_day=2022-01-02", 0L);
-    UpdateProperties updateProperties = testKeyedTable.baseTable().updateProperties();
-    updateProperties.set(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID,
-        new ObjectMapper().writeValueAsString(transactionIdMap));
+    List<DataFile> s1Files = insertChangeDataFiles(1);
+    List<StructLike> partitions = new ArrayList<>(s1Files.stream().collect(Collectors.groupingBy(ContentFile::partition)).keySet());
+    Assert.assertEquals(2, partitions.size());
+
+    UpdatePartitionProperties updateProperties = testKeyedTable.baseTable().updatePartitionProperties(null);
+    updateProperties.set(partitions.get(0), TableProperties.BASE_TABLE_MAX_TRANSACTION_ID, "3");
+    updateProperties.set(partitions.get(1), TableProperties.BASE_TABLE_MAX_TRANSACTION_ID, "0");
     updateProperties.commit();
     List<DataFile> existedDataFiles = new ArrayList<>();
     testKeyedTable.changeTable().newScan().planFiles()
@@ -79,13 +77,12 @@ public class TestExpiredFileClean extends TableTestBase {
   @Test
   public void testExpireTableFiles() throws Exception {
     List<DataFile> s1Files = insertChangeDataFiles(1);
+    List<StructLike> partitions = new ArrayList<>(s1Files.stream().collect(Collectors.groupingBy(ContentFile::partition)).keySet());
+    Assert.assertEquals(2, partitions.size());
 
-    Map<String, Long> transactionIdMap = new HashMap<>();
-    transactionIdMap.put("op_time_day=2022-01-01", 3L);
-    transactionIdMap.put("op_time_day=2022-01-02", 1L);
-    UpdateProperties updateProperties = testKeyedTable.baseTable().updateProperties();
-    updateProperties.set(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID,
-        new ObjectMapper().writeValueAsString(transactionIdMap));
+    UpdatePartitionProperties updateProperties = testKeyedTable.baseTable().updatePartitionProperties(null);
+    updateProperties.set(partitions.get(0), TableProperties.BASE_TABLE_MAX_TRANSACTION_ID, "3");
+    updateProperties.set(partitions.get(1), TableProperties.BASE_TABLE_MAX_TRANSACTION_ID, "1");
     updateProperties.commit();
     Assert.assertTrue(testKeyedTable.io().exists((String) s1Files.get(0).path()));
     TableExpireService.deleteChangeFile(testKeyedTable, changeTableFilesInfo);
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestExpiredFileCleanSupportHive.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestExpiredFileCleanSupportHive.java
index 2542e0fa8..d8bb6fd46 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestExpiredFileCleanSupportHive.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestExpiredFileCleanSupportHive.java
@@ -19,10 +19,9 @@
 package com.netease.arctic.ams.server.optimize;
 
 import com.netease.arctic.ams.server.service.impl.TableExpireService;
-import com.netease.arctic.ams.server.utils.HiveLocationUtils;
 import com.netease.arctic.hive.io.writer.AdaptHiveGenericTaskWriterBuilder;
 import com.netease.arctic.hive.table.HiveLocationKind;
-import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.hive.utils.TableTypeUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.UnkeyedTable;
 import com.netease.arctic.utils.FileUtil;
@@ -66,7 +65,7 @@ public class TestExpiredFileCleanSupportHive extends TestSupportHiveBase {
     }
 
     Set<String> hiveLocation = new HashSet<>();
-    if (HiveTableUtil.isHive(testUnPartitionKeyedHiveTable)) {
+    if (TableTypeUtil.isHive(testUnPartitionKeyedHiveTable)) {
       hiveLocation.add(FileUtil.getFileDir(hiveFiles.get(0).path().toString()));
     }
     TableExpireService.expireSnapshots(testUnPartitionKeyedHiveTable.baseTable(), System.currentTimeMillis(), hiveLocation);
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestMinorOptimizeCommit.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestMinorOptimizeCommit.java
index 2c59f8113..3413b4773 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestMinorOptimizeCommit.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestMinorOptimizeCommit.java
@@ -28,6 +28,7 @@ import com.netease.arctic.ams.server.utils.JDBCSqlSessionFactoryProvider;
 import com.netease.arctic.data.DataTreeNode;
 import com.netease.arctic.data.DefaultKeyedFile;
 import com.netease.arctic.utils.SerializationUtil;
+import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.iceberg.ContentFile;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
@@ -126,7 +127,7 @@ public class TestMinorOptimizeCommit extends TestMinorOptimizePlan {
           fileScanTask.deletes().forEach(deleteFile -> newDeleteFilesPath.add((String) deleteFile.path()));
         });
 
-    StructLikeMap<Long> maxTxId = testKeyedTable.partitionMaxTransactionId();
+    StructLikeMap<Long> maxTxId = TablePropertyUtil.getPartitionMaxTransactionId(testKeyedTable);
     for (StructLike partitionDatum : partitionData) {
       Assert.assertEquals(4L, (long) maxTxId.get(partitionDatum));
     }
@@ -190,8 +191,8 @@ public class TestMinorOptimizeCommit extends TestMinorOptimizePlan {
           fileScanTask.deletes().forEach(deleteFile -> newDeleteFilesPath.add((String) deleteFile.path()));
         });
 
-    StructLikeMap<Long> maxTxId = testNoPartitionTable.partitionMaxTransactionId();
-    Assert.assertEquals(4L, (long) maxTxId.get(null));
+    StructLikeMap<Long> maxTxId = TablePropertyUtil.getPartitionMaxTransactionId(testNoPartitionTable);
+    Assert.assertEquals(4L, (long) maxTxId.get(TablePropertyUtil.EMPTY_STRUCT));
     Assert.assertNotEquals(oldDataFilesPath, newDataFilesPath);
     Assert.assertNotEquals(oldDeleteFilesPath, newDeleteFilesPath);
   }
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestMinorOptimizePlan.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestMinorOptimizePlan.java
index e04f1ac46..e1bb116d4 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestMinorOptimizePlan.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestMinorOptimizePlan.java
@@ -62,7 +62,6 @@ public class TestMinorOptimizePlan extends TestBaseOptimizeBase {
     Assert.assertEquals(1, tasks.get(0).getPosDeleteFiles().size());
     Assert.assertEquals(10, tasks.get(0).getInsertFileCnt());
     Assert.assertEquals(10, tasks.get(0).getDeleteFileCnt());
-    Assert.assertEquals(0, tasks.get(0).getIsDeletePosDelete());
   }
 
   protected void insertChangeDeleteFiles(ArcticTable arcticTable, long transactionId) throws IOException {
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestAdaptHiveService.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestAdaptHiveService.java
index 142c04440..c0f89c6de 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestAdaptHiveService.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestAdaptHiveService.java
@@ -28,6 +28,7 @@ import com.netease.arctic.ams.server.model.UpgradeStatus;
 import com.netease.arctic.ams.server.service.impl.AdaptHiveService;
 import com.netease.arctic.catalog.CatalogLoader;
 import com.netease.arctic.hive.HMSMockServer;
+import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.catalog.ArcticHiveCatalog;
 import com.netease.arctic.table.TableIdentifier;
 import org.apache.hadoop.hive.metastore.api.Database;
@@ -71,11 +72,15 @@ public class TestAdaptHiveService {
   protected static final TableIdentifier UN_PARTITION_HIVE_PK_TABLE_ID =
       TableIdentifier.of(HIVE_CATALOG_NAME, HIVE_DB_NAME, "un_partition_test_pk_hive_table");
 
+  protected static final TableIdentifier UNSUPPORTED_TABLE =
+      TableIdentifier.of(HIVE_CATALOG_NAME, HIVE_DB_NAME, "unsupported_table");
+
 
   protected static Table PARTITION_TABLE = new Table();
   protected static Table PARTITION_TABLE_2 = new Table();
   protected static Table NO_PARTITION_TABLE = new Table();
   protected static Table NO_PARTITION_TABLE_2 = new Table();
+  protected static Table UNSUPPORTED_ORC_TABLE = new Table();
 
   protected static UpgradeHiveMeta withPk = new UpgradeHiveMeta();
   protected static UpgradeHiveMeta withoutPk = new UpgradeHiveMeta();
@@ -135,9 +140,11 @@ public class TestAdaptHiveService {
     schema.add(new FieldSchema("name", "string", null));
     schema.add(new FieldSchema("age", "int", null));
     StorageDescriptor storageDescriptor = new StorageDescriptor();
+    storageDescriptor.setInputFormat(HiveTableProperties.PARQUET_INPUT_FORMAT);
+    storageDescriptor.setOutputFormat(HiveTableProperties.PARQUET_OUTPUT_FORMAT);
     storageDescriptor.setCols(schema);
     SerDeInfo serDeInfo = new SerDeInfo();
-    serDeInfo.setSerializationLib("org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe");
+    serDeInfo.setSerializationLib(HiveTableProperties.PARQUET_ROW_FORMAT_SERDE);
     storageDescriptor.setSerdeInfo(serDeInfo);
     List<FieldSchema> partitions = new ArrayList<>();
     schema.add(new FieldSchema("dt", "string", null));
@@ -166,6 +173,19 @@ public class TestAdaptHiveService {
     NO_PARTITION_TABLE_2.setPartitionKeys(partitions);
     hms.getClient().createTable(NO_PARTITION_TABLE_2);
 
+    StorageDescriptor wrongSd = new StorageDescriptor();
+    wrongSd.setInputFormat("org.apache.hadoop.hive.ql.io.orc.OrcInputFormat");
+    wrongSd.setOutputFormat("org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat");
+    wrongSd.setCols(schema);
+    SerDeInfo wrongSerDeInfo = new SerDeInfo();
+    wrongSerDeInfo.setSerializationLib("org.apache.hadoop.hive.ql.io.orc.OrcSerde");
+    wrongSd.setSerdeInfo(wrongSerDeInfo);
+    UNSUPPORTED_ORC_TABLE.setDbName(HIVE_DB_NAME);
+    UNSUPPORTED_ORC_TABLE.setTableName(UNSUPPORTED_TABLE.getTableName());
+    UNSUPPORTED_ORC_TABLE.setSd(wrongSd);
+    UNSUPPORTED_ORC_TABLE.setPartitionKeys(partitions);
+    hms.getClient().createTable(UNSUPPORTED_ORC_TABLE);
+
     List<UpgradeHiveMeta.PrimaryKeyField> pkList = new ArrayList<>();
     Map<String, String> properties = new HashMap<>();
     pkList.add(new UpgradeHiveMeta.PrimaryKeyField("id"));
@@ -213,5 +233,14 @@ public class TestAdaptHiveService {
       upgradeRunningInfo = adaptHiveService.getUpgradeRunningInfo(HIVE_PK_TABLE_ID);
     }
     Assert.assertEquals(upgradeRunningInfo.getStatus(), UpgradeStatus.SUCCESS.toString());
+
+    adaptHiveService.upgradeHiveTable(hiveCatalog, UNSUPPORTED_TABLE, withPk);
+    upgradeRunningInfo = adaptHiveService.getUpgradeRunningInfo(UNSUPPORTED_TABLE);
+    while (upgradeRunningInfo.getStatus().equals(UpgradeStatus.NONE.toString())
+        || upgradeRunningInfo.getStatus().equals(UpgradeStatus.UPGRADING.toString())) {
+      Thread.sleep(3000);
+      upgradeRunningInfo = adaptHiveService.getUpgradeRunningInfo(UNSUPPORTED_TABLE);
+    }
+    Assert.assertEquals(upgradeRunningInfo.getStatus(), UpgradeStatus.FAILED.toString());
   }
 }
diff --git a/core/src/test/java/com/netease/arctic/TableTestBase.java b/core/src/test/java/com/netease/arctic/TableTestBase.java
index 822edb481..fa46d9fb2 100644
--- a/core/src/test/java/com/netease/arctic/TableTestBase.java
+++ b/core/src/test/java/com/netease/arctic/TableTestBase.java
@@ -54,6 +54,8 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.File;
 import java.io.IOException;
@@ -67,6 +69,7 @@ import static com.netease.arctic.ams.api.MockArcticMetastoreServer.TEST_DB_NAME;
 
 public class TableTestBase {
 
+  private static final Logger LOG = LoggerFactory.getLogger(TableTestBase.class);
   protected static final MockArcticMetastoreServer AMS = MockArcticMetastoreServer.getInstance();
 
   protected static final TableIdentifier TABLE_ID =
@@ -120,6 +123,7 @@ public class TableTestBase {
 
   @Before
   public void setupTables() throws Exception {
+    LOG.info("setupTables start");
     testCatalog = CatalogLoader.load(AMS.getUrl());
     tableDir = temp.newFolder();
 
@@ -148,6 +152,7 @@ public class TableTestBase {
         .create().asKeyedTable();
 
     this.before();
+    LOG.info("setupTables end");
   }
 
   public void before() throws Exception {
@@ -156,6 +161,7 @@ public class TableTestBase {
 
   @After
   public void clearTable() {
+    LOG.info("clearTable start");
     testCatalog.dropTable(TABLE_ID, true);
     AMS.handler().getTableCommitMetas().remove(TABLE_ID.buildTableIdentifier());
 
@@ -164,6 +170,7 @@ public class TableTestBase {
 
     testCatalog.dropTable(NO_PARTITION_TABLE_ID, true);
     AMS.handler().getTableCommitMetas().remove(NO_PARTITION_TABLE_ID.buildTableIdentifier());
+    LOG.info("clearTable end");
   }
 
   public List<DataFile> writeBase(TableIdentifier identifier, List<Record> records) {
diff --git a/core/src/test/java/com/netease/arctic/op/OverwriteBaseFileTest.java b/core/src/test/java/com/netease/arctic/op/OverwriteBaseFileTest.java
index 9fc4457fc..b809ed44e 100644
--- a/core/src/test/java/com/netease/arctic/op/OverwriteBaseFileTest.java
+++ b/core/src/test/java/com/netease/arctic/op/OverwriteBaseFileTest.java
@@ -22,6 +22,7 @@ import com.netease.arctic.TableTestBase;
 import com.netease.arctic.catalog.CatalogLoader;
 import com.netease.arctic.data.ChangeAction;
 import com.netease.arctic.table.TableProperties;
+import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
@@ -87,7 +88,7 @@ public class OverwriteBaseFileTest extends TableTestBase {
     ));
 
     // init. 3 partition with init txId
-    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.partitionMaxTransactionId();
+    StructLikeMap<Long> partitionMaxTxId = TablePropertyUtil.getPartitionMaxTransactionId(testKeyedTable);
     Assert.assertEquals(initTxId, partitionMaxTxId.get(
         partitionData(TABLE_SCHEMA, SPEC, "2020-1-1")
     ).longValue());
@@ -125,7 +126,7 @@ public class OverwriteBaseFileTest extends TableTestBase {
         .commit();
     // overwrite all partition and add new data file
 
-    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.partitionMaxTransactionId();
+    StructLikeMap<Long> partitionMaxTxId = TablePropertyUtil.getPartitionMaxTransactionId(testKeyedTable);
     // expect result: all partition with new txId
     Assert.assertEquals(txId, partitionMaxTxId.get(
         partitionData(TABLE_SCHEMA, SPEC, "2020-1-1")
@@ -177,7 +178,7 @@ public class OverwriteBaseFileTest extends TableTestBase {
     overwrite.commit();
     // overwrite all partition and add new data file
 
-    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.partitionMaxTransactionId();
+    StructLikeMap<Long> partitionMaxTxId = TablePropertyUtil.getPartitionMaxTransactionId(testKeyedTable);
     // expect result: 1,2 partition with new txId, 3 partition use old txId
     Assert.assertEquals(txId, partitionMaxTxId.get(
         partitionData(TABLE_SCHEMA, SPEC, "2020-1-1")
diff --git a/core/src/test/java/com/netease/arctic/op/RewritePartitionsTest.java b/core/src/test/java/com/netease/arctic/op/RewritePartitionsTest.java
index 86f3ccf9b..fdc8b08f5 100644
--- a/core/src/test/java/com/netease/arctic/op/RewritePartitionsTest.java
+++ b/core/src/test/java/com/netease/arctic/op/RewritePartitionsTest.java
@@ -20,6 +20,7 @@ package com.netease.arctic.op;
 
 import com.netease.arctic.TableTestBase;
 import com.netease.arctic.data.ChangeAction;
+import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -58,7 +59,7 @@ public class RewritePartitionsTest extends TableTestBase {
     ));
 
     // init. 3 partition with init txId
-    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.partitionMaxTransactionId();
+    StructLikeMap<Long> partitionMaxTxId = TablePropertyUtil.getPartitionMaxTransactionId(testKeyedTable);
     Assert.assertEquals(initTxId, partitionMaxTxId.get(
         partitionData(TABLE_SCHEMA, SPEC, quickDate(1))
     ).longValue());
@@ -95,7 +96,7 @@ public class RewritePartitionsTest extends TableTestBase {
     overwrite.commit();
     // overwrite 1 partition by data file
 
-    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.partitionMaxTransactionId();
+    StructLikeMap<Long> partitionMaxTxId = TablePropertyUtil.getPartitionMaxTransactionId(testKeyedTable);
     // expect result: 1 partition with new txId, 2,3 partition use old txId
     Assert.assertEquals(txId, partitionMaxTxId.get(
         partitionData(TABLE_SCHEMA, SPEC, quickDate(1))
diff --git a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/FlinkTestBase.java b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/FlinkTestBase.java
index ec14fe525..23f5ed4ff 100644
--- a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/FlinkTestBase.java
+++ b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/FlinkTestBase.java
@@ -135,6 +135,7 @@ public class FlinkTestBase extends TableTestBase {
 
   @After
   public void clean() {
+    LOG.info("clean start");
     if (IS_LOCAL && testCatalog != null) {
       if (testCatalog.tableExists(PK_TABLE_ID_WITHOUT_PARTITION)) {
         testCatalog.dropTable(PK_TABLE_ID_WITHOUT_PARTITION, true);
@@ -146,6 +147,7 @@ public class FlinkTestBase extends TableTestBase {
         AMS.handler().getTableCommitMetas().remove(PARTITION_TABLE_ID);
       }
     }
+    LOG.info("clean end");
   }
 
   public void config(String catalog) {
diff --git a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/FlinkTestBase.java b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/FlinkTestBase.java
index 85283a25b..aa1b6a0eb 100644
--- a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/FlinkTestBase.java
+++ b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/FlinkTestBase.java
@@ -136,6 +136,7 @@ public class FlinkTestBase extends TableTestBase {
 
   @After
   public void clean() {
+    LOG.info("clean start");
     if (IS_LOCAL && testCatalog != null) {
       if (testCatalog.tableExists(PK_TABLE_ID_WITHOUT_PARTITION)) {
         testCatalog.dropTable(PK_TABLE_ID_WITHOUT_PARTITION, true);
@@ -147,6 +148,7 @@ public class FlinkTestBase extends TableTestBase {
         AMS.handler().getTableCommitMetas().remove(PARTITION_TABLE_ID);
       }
     }
+    LOG.info("clean end");
   }
 
   public void config(String catalog) {
diff --git a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/FlinkTestBase.java b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/FlinkTestBase.java
index 8c0367662..252100e76 100644
--- a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/FlinkTestBase.java
+++ b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/FlinkTestBase.java
@@ -136,6 +136,7 @@ public class FlinkTestBase extends TableTestBase {
 
   @After
   public void clean() {
+    LOG.info("clean start");
     if (IS_LOCAL && testCatalog != null) {
       if (testCatalog.tableExists(PK_TABLE_ID_WITHOUT_PARTITION)) {
         testCatalog.dropTable(PK_TABLE_ID_WITHOUT_PARTITION, true);
@@ -147,6 +148,7 @@ public class FlinkTestBase extends TableTestBase {
         AMS.handler().getTableCommitMetas().remove(PARTITION_TABLE_ID);
       }
     }
+    LOG.info("clean end");
   }
 
   public void config(String catalog) {