Repository: NetEase/arctic
Pull request #340 (2022-09-07 07:25:59)

[ARCTIC-291] [Improvement]: support access Tables tab only for display table info only
<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://arctic.netease.com/ch/contribute/
  2. If the PR is related to an issue in https://github.com/NetEase/arctic/issues, add '[ARCTIC-XXXX]' in your PR title, e.g., '[ARCTIC-XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][ARCTIC #XXXX] Your PR title ...'.
-->

## Why are the changes needed?
fix https://github.com/NetEase/arctic/issues/291
people can use  ams table tab page only to display arctic table detail;


## Brief change log

First, i will support an api to generate a token with catalog,db,table.

second , all tabls tab request url will be /ams/v1/tables/catalogs/{catalog}/dbs/{db}/tables/{table}, i will extract catalog,db,table from url, and check the token.

if token mismatch , return signatureExecption.

if match, return the result of query.


## How was this patch tested?
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] Run test locally before making a pull request

## Documentation

  - Does this pull request introduces a new feature? ( no)
  - If yes, how is the feature documented? (not documented)


hameizi: LGTM

Patch:
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/AmsRestServer.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/AmsRestServer.java
index 6fb2d431c..951eecb10 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/AmsRestServer.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/AmsRestServer.java
@@ -30,9 +30,12 @@ import com.netease.arctic.ams.server.exception.ForbiddenException;
 import com.netease.arctic.ams.server.exception.SignatureCheckException;
 import com.netease.arctic.ams.server.service.impl.ApiTokenService;
 import com.netease.arctic.ams.server.utils.ParamSignatureCalculator;
+import com.netease.arctic.ams.server.utils.Utils;
 import io.javalin.Javalin;
+import io.javalin.http.Context;
 import io.javalin.http.HttpCode;
 import io.javalin.http.staticfiles.Location;
+import org.apache.arrow.util.VisibleForTesting;
 import org.apache.commons.lang3.StringUtils;
 import org.eclipse.jetty.server.session.SessionHandler;
 import org.slf4j.Logger;
@@ -89,14 +92,20 @@ public class AmsRestServer {
     // before
     app.before(ctx -> {
       String uriPath = ctx.path();
-      if (needApiKeyCheck(uriPath)) {
-        checkApiToken(ctx.method(), ctx.url(), ctx.queryParam("apiKey"),
-                ctx.queryParam("signature"), ctx.queryParamMap());
-      } else if (needLoginCheck(uriPath)) {
-        if (null == ctx.sessionAttribute("user")) {
-          LOG.info("session info: {}", ctx.sessionAttributeMap() == null ? null : JSONObject.toJSONString(
-              ctx.sessionAttributeMap()));
-          throw new ForbiddenException();
+      String token = ctx.queryParam("token");
+      // if token of api request is not empty, so we check the query by token first
+      if (uriPath.startsWith("/api/") && StringUtils.isNotEmpty(token)) {
+        Utils.checkSinglePageToken(ctx);
+      } else {
+        if (needApiKeyCheck(uriPath)) {
+          checkApiToken(ctx.method(), ctx.url(), ctx.queryParam("apiKey"),
+                  ctx.queryParam("signature"), ctx.queryParamMap());
+        } else if (needLoginCheck(uriPath)) {
+          if (null == ctx.sessionAttribute("user")) {
+            LOG.info("session info: {}", ctx.sessionAttributeMap() == null ? null : JSONObject.toJSONString(
+                    ctx.sessionAttributeMap()));
+            throw new ForbiddenException();
+          }
         }
       }
     });
@@ -172,6 +181,7 @@ public class AmsRestServer {
         get("/tables/catalogs/{catalog}/dbs/{db}/tables/{table}/partitions", TableController::getTablePartitions);
         get("/tables/catalogs/{catalog}/dbs/{db}/tables/{table}/partitions/{partition}/files",
                 TableController::getPartitionFileListInfo);
+        get("/tables/catalogs/{catalog}/dbs/{db}/tables/{table}/signature", TableController::getTableDetailTabToken);
         get("/catalogs/{catalog}/databases/{db}/tables", TableController::getTableList);
         get("/catalogs/{catalog}/databases", TableController::getDatabaseList);
         get("/catalogs", TableController::getCatalogs);
@@ -227,7 +237,7 @@ public class AmsRestServer {
       ctx.json(new ErrorResponse(HttpCode.INTERNAL_SERVER_ERROR, "internal error!", ""));
     });
   }
-  
+
   public static void stopRestServer() {
     if (app != null) {
       app.stop();
@@ -270,7 +280,7 @@ public class AmsRestServer {
   }
 
   private static void checkApiToken(String requestMethod, String requestUrl, String apiKey, String signature,
-                             Map<String, List<String>> params) {
+                                    Map<String, List<String>> params) {
     String plainText;
     String encryptString;
     String signCal;
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/TableController.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/TableController.java
index 625bf2a47..64748b268 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/TableController.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/TableController.java
@@ -29,6 +29,7 @@ import com.netease.arctic.ams.server.config.ServerTableProperties;
 import com.netease.arctic.ams.server.controller.response.ErrorResponse;
 import com.netease.arctic.ams.server.controller.response.OkResponse;
 import com.netease.arctic.ams.server.controller.response.PageResult;
+import com.netease.arctic.ams.server.exception.SignatureCheckException;
 import com.netease.arctic.ams.server.model.AMSColumnInfo;
 import com.netease.arctic.ams.server.model.AMSDataFileInfo;
 import com.netease.arctic.ams.server.model.AMSTransactionsOfTable;
@@ -56,6 +57,8 @@ import com.netease.arctic.ams.server.service.impl.DDLTracerService;
 import com.netease.arctic.ams.server.service.impl.FileInfoCacheService;
 import com.netease.arctic.ams.server.utils.AmsUtils;
 import com.netease.arctic.ams.server.utils.CatalogUtil;
+import com.netease.arctic.ams.server.utils.ParamSignatureCalculator;
+import com.netease.arctic.ams.server.utils.Utils;
 import com.netease.arctic.catalog.ArcticCatalog;
 import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.catalog.ArcticHiveCatalog;
@@ -65,6 +68,7 @@ import com.netease.arctic.table.TableIdentifier;
 import com.netease.arctic.table.TableProperties;
 import io.javalin.http.Context;
 import io.javalin.http.HttpCode;
+import org.apache.arrow.util.VisibleForTesting;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hive.metastore.api.Table;
 import org.slf4j.Logger;
@@ -494,4 +498,18 @@ public class TableController extends RestBaseController {
             new CatalogMeta(t.getCatalogName(), t.getCatalogType())).collect(Collectors.toList());
     ctx.json(OkResponse.of(catalogs));
   }
+
+  /**
+   * get single page query token
+   * @param ctx
+   */
+  public static void getTableDetailTabToken(Context ctx) {
+    String catalog =  ctx.pathParam("catalog");
+    String db =  ctx.pathParam("db");
+    String table =  ctx.pathParam("table");
+
+    String signCal = Utils.generateTablePageToken(catalog, db, table);
+    ctx.json(OkResponse.of(signCal));
+  }
+
 }
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/TableOptimizeItem.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/TableOptimizeItem.java
index 87b643db3..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
@@ -56,6 +56,7 @@ 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;
@@ -602,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()));
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 1938e39d9..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
@@ -39,6 +39,7 @@ 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;
@@ -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/ParamSignatureCalculator.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/ParamSignatureCalculator.java
index 68fba65b7..b0810400e 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/ParamSignatureCalculator.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/ParamSignatureCalculator.java
@@ -18,6 +18,7 @@
 
 package com.netease.arctic.ams.server.utils;
 
+import com.netease.arctic.ams.server.controller.response.OkResponse;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -27,6 +28,7 @@ import java.net.URLDecoder;
 import java.text.SimpleDateFormat;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -105,4 +107,36 @@ public class ParamSignatureCalculator {
     return sb.toString();
   }
 
+  /**
+   * Gets an ascending KeyValue concatenation string based on the request argument pair.
+   * Example：
+   * params: name=&value=111&age=11&sex=1&high=180&nick=
+   * Remove null and arrange in ascending order： age11high180sex1value111
+   *
+   * @param map
+   * @return
+   */
+  public static String generateParamStringWithValue(Map<String, String> map) {
+    Set<String> set = map.keySet();
+    String[] keyArray = (String[]) set.toArray(new String[set.size()]);
+    StringBuffer sb = new StringBuffer("");
+    Arrays.sort(keyArray);
+    String firstValue = "";
+    for (int i = 0; i < keyArray.length; i++) {
+      String value = map.get(keyArray[i]);
+      if (!StringUtils.isBlank(value)) {
+        try {
+          firstValue = URLDecoder.decode(value, "utf-8");
+        } catch (UnsupportedEncodingException e) {
+          LOG.error("Failed to caculate signature", e);
+          return null;
+        }
+        sb.append(keyArray[i]).append(firstValue);
+      }
+    }
+    return sb.toString();
+  }
+
+
+
 }
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/Utils.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/Utils.java
index 98ff71e8d..8e2c8b8e2 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/Utils.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/Utils.java
@@ -18,11 +18,17 @@
 
 package com.netease.arctic.ams.server.utils;
 
+import com.netease.arctic.ams.server.exception.SignatureCheckException;
+import io.javalin.http.Context;
+import org.apache.arrow.util.VisibleForTesting;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.net.telnet.TelnetClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.net.InetAddress;
+import java.util.HashMap;
+import java.util.Map;
 
 public class Utils {
   private static final Logger LOG = LoggerFactory.getLogger(Utils.class);
@@ -85,4 +91,54 @@ public class Utils {
       return false;
     }
   }
+
+  /**
+   * check single page access token
+   *
+   * @param ctx
+   * @return
+   */
+  public static void checkSinglePageToken(Context ctx) {
+    // check if query parameters contain  token key
+    String token = ctx.queryParam("token");
+
+    if (StringUtils.isNotEmpty(token)) {
+      // regrex extract  catalog, db, table
+      String url = ctx.req.getRequestURI();
+      String catalog = null;
+      String db = null;
+      String table = null;
+      String[] splitResult = url.split("/");
+      for (int i = 0; i < splitResult.length; i++) {
+        if (splitResult[i].equals("catalogs")) {
+          catalog = splitResult[i + 1];
+        } else if (splitResult[i].equals("dbs")) {
+          db = splitResult[i + 1];
+        } else if (splitResult[i].equals("tables")) {
+          table = splitResult[i + 1];
+        }
+      }
+      if (StringUtils.isEmpty(catalog) ||
+              StringUtils.isEmpty(db) ||
+              StringUtils.isEmpty(table) ||
+              !token.equals(generateTablePageToken(catalog, db, table))) {
+        throw new SignatureCheckException();
+      }
+    }
+  }
+
+  /**
+   * generate the token for single table page access.
+   * @return
+   */
+  public static String generateTablePageToken(String catalog, String db, String table) {
+    Map<String, String> params = new HashMap<>();
+    params.put("catalog", catalog);
+    params.put("db",  db);
+    params.put("table", table);
+
+    String paramString = ParamSignatureCalculator.generateParamStringWithValue(params);
+    String plainText = String.format("%s%s%s", paramString, paramString, paramString);
+    return ParamSignatureCalculator.getMD5(plainText);
+  }
 }
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.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.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/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();
     }
   }
 }


Test patch:
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TableControllerTest.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TableControllerTest.java
index 9ccf8a974..0c3ae3c03 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TableControllerTest.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/controller/TableControllerTest.java
@@ -26,10 +26,13 @@ import com.netease.arctic.ams.api.MockArcticMetastoreServer;
 import com.netease.arctic.ams.api.OptimizeRangeType;
 import com.netease.arctic.ams.api.TableMeta;
 import com.netease.arctic.ams.api.properties.CatalogMetaProperties;
+import com.netease.arctic.ams.server.AmsRestServer;
 import com.netease.arctic.ams.server.ArcticMetaStore;
 import com.netease.arctic.ams.server.config.ArcticMetaStoreConf;
 import com.netease.arctic.ams.server.controller.response.OkResponse;
 import com.netease.arctic.ams.server.controller.response.PageResult;
+import com.netease.arctic.ams.server.exception.ForbiddenException;
+import com.netease.arctic.ams.server.exception.SignatureCheckException;
 import com.netease.arctic.ams.server.handler.impl.OptimizeManagerHandler;
 import com.netease.arctic.ams.server.model.AMSColumnInfo;
 import com.netease.arctic.ams.server.model.AMSDataFileInfo;
@@ -55,14 +58,17 @@ import com.netease.arctic.ams.server.util.DerbyTestUtil;
 import com.netease.arctic.ams.server.utils.AmsUtils;
 import com.netease.arctic.ams.server.utils.CatalogUtil;
 import com.netease.arctic.ams.server.utils.JDBCSqlSessionFactoryProvider;
+import com.netease.arctic.ams.server.utils.Utils;
 import com.netease.arctic.catalog.ArcticCatalog;
 import com.netease.arctic.hive.catalog.ArcticHiveCatalog;
 import com.netease.arctic.hive.utils.HiveTableUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.PrimaryKeySpec;
 import com.netease.arctic.table.TableIdentifier;
+import io.javalin.apibuilder.ApiBuilder;
 import io.javalin.testtools.JavalinTest;
 import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
 import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
 import org.apache.hadoop.hive.metastore.api.Table;
@@ -194,6 +200,21 @@ public class TableControllerTest {
     });
   }
 
+  @Test
+  public void testSinglePageToken() throws Exception {
+    mockService(catalogName, database, table);
+    JavalinTest.test((app, client) -> {
+      app.get("/tables/catalogs/{catalog}/dbs/{db}/tables/{table}/signature",
+              TableController::getTableDetailTabToken);
+      String url = String.format("/tables/catalogs/%s/dbs/%s/tables/%s/signature", catalogName, database,table);
+      final okhttp3.Response resp = client.get(String.format(url), x -> {});
+      OkResponse result = JSONObject.parseObject(resp.body().string(), OkResponse.class);
+      LOG.info("xxx: {}", JSONObject.toJSONString(result));
+      assert result.getCode() == 200;
+      assert Utils.generateTablePageToken(catalogName,database,table).equals(result.getResult());
+    });
+  }
+
   @Test
   public void testGetDatabaseList() {
     JavalinTest.test((app, client) -> {
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/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/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) {