Repository: NetEase/arctic
Pull request #244 (2022-08-29 12:19:57)

[ARCTIC-238][DIST]rename CompactJob.jar to OptimizeJob.jar and mv OptimizeJob.jar to di…
…rectory plugin

<!--
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 issue https://github.com/NetEase/arctic/issues/238

## Brief change log

  - *rename CompactJob.jar to OptimizeJob.jar and mv OptimizeJob.jar to directory plugin*

## How was this patch tested?
- [ ] mvn package module `dist` and OptimizeJob.jar should exist in dir /plugin

## Documentation

  - Does this pull request introduces a new feature? (no)
  - If yes, how is the feature documented? ( not documented)


hameizi: @hhippodnsla please take a look

Patch:
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/DDLRecordMapper.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/DDLRecordMapper.java
index 86b87420a..7bbbd25db 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/DDLRecordMapper.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/DDLRecordMapper.java
@@ -6,9 +6,9 @@
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
- *  *
+ *
  *     http://www.apache.org/licenses/LICENSE-2.0
- *  *
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/BaseOptimizeTaskRuntime.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/BaseOptimizeTaskRuntime.java
index d16e5ae8c..fe17c0d7b 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/BaseOptimizeTaskRuntime.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/BaseOptimizeTaskRuntime.java
@@ -26,7 +26,7 @@ import com.netease.arctic.ams.api.OptimizeTaskId;
 import java.nio.ByteBuffer;
 import java.util.List;
 
-public class BaseOptimizeTaskRuntime {
+public class BaseOptimizeTaskRuntime implements Cloneable {
   public static final long INVALID_TIME = 0;
   private OptimizeTaskId optimizeTaskId;
   private OptimizeStatus status = OptimizeStatus.Init;
@@ -202,4 +202,13 @@ public class BaseOptimizeTaskRuntime {
         ", targetFiles=" + targetFiles +
         '}';
   }
+
+  @Override
+  public BaseOptimizeTaskRuntime clone() {
+    try {
+      return (BaseOptimizeTaskRuntime) super.clone();
+    } catch (CloneNotSupportedException e) {
+      throw new RuntimeException(e);
+    }
+  }
 }
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/CacheFileInfo.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/CacheFileInfo.java
index b441b699c..cd4ba9612 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/CacheFileInfo.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/CacheFileInfo.java
@@ -59,7 +59,8 @@ public class CacheFileInfo {
 
   }
 
-  public CacheFileInfo(String primaryKeyMd5, TableIdentifier tableIdentifier, Long addSnapshotId,
+  public CacheFileInfo(
+      String primaryKeyMd5, TableIdentifier tableIdentifier, Long addSnapshotId,
       Long parentSnapshotId, Long deleteSnapshotId, String innerTable,
       String filePath, String fileType, Long fileSize, Long fileMask, Long fileIndex, Long specId,
       String partitionName, Long commitTime, Long recordCount, String action, Long watermark) {
@@ -93,7 +94,8 @@ public class CacheFileInfo {
     return dataFileInfo;
   }
 
-  public static CacheFileInfo convert(Table table, DataFile amsFile, TableIdentifier identifier, String tableType,
+  public static CacheFileInfo convert(
+      Table table, DataFile amsFile, TableIdentifier identifier, String tableType,
       Snapshot snapshot) {
     long watermark = 0L;
     boolean isDataFile = Objects.equals(amsFile.fileType, DataFileType.INSERT_FILE.name()) ||
@@ -105,14 +107,15 @@ public class CacheFileInfo {
               .get(table.properties().get(TableProperties.TABLE_EVENT_TIME_FIELD))
               .getLong();
     }
-    String primaryKey = TableMetadataUtil.getTableAllIdentifyName(identifier) + tableType + amsFile.getPath();
+    String partitionName = StringUtils.isEmpty(partitionToPath(amsFile.getPartition())) ?
+        "" :
+        partitionToPath(amsFile.getPartition());
+    String primaryKey =
+        TableMetadataUtil.getTableAllIdentifyName(identifier) + tableType + amsFile.getPath() + partitionName;
     String primaryKeyMd5 = Hashing.md5()
         .hashBytes(primaryKey.getBytes(StandardCharsets.UTF_8))
         .toString();
     Long parentId = snapshot.parentId() == null ? -1 : snapshot.parentId();
-    String partitionName = StringUtils.isEmpty(partitionToPath(amsFile.getPartition())) ?
-        null :
-        partitionToPath(amsFile.getPartition());
     return new CacheFileInfo(primaryKeyMd5, identifier, snapshot.snapshotId(),
         parentId, null,
         tableType, amsFile.getPath(), amsFile.getFileType(), amsFile.getFileSize(), amsFile.getMask(),
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/DDLInfo.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/DDLInfo.java
index 4904e64f8..1701a4e5c 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/DDLInfo.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/DDLInfo.java
@@ -6,9 +6,9 @@
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
- *  *
+ *
  *     http://www.apache.org/licenses/LICENSE-2.0
- *  *
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/TableOperation.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/TableOperation.java
index a5c0dca47..52efd3dc9 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/TableOperation.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/TableOperation.java
@@ -1,3 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package com.netease.arctic.ams.server.model;
 
 /**
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/OptimizeTaskItem.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/OptimizeTaskItem.java
index 1dd6607b5..0b7154b07 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/OptimizeTaskItem.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/OptimizeTaskItem.java
@@ -52,7 +52,7 @@ public class OptimizeTaskItem extends IJDBCService {
   private static final long RETRY_INTERVAL = 60000; // 60s
 
   private final BaseOptimizeTask optimizeTask;
-  private final BaseOptimizeTaskRuntime optimizeRuntime;
+  private volatile BaseOptimizeTaskRuntime optimizeRuntime;
   private final ReentrantLock lock = new ReentrantLock();
 
   public OptimizeTaskItem(BaseOptimizeTask optimizeTask,
@@ -80,12 +80,14 @@ public class OptimizeTaskItem extends IJDBCService {
   public void onPending() {
     lock.lock();
     try {
-      if (optimizeRuntime.getStatus() == OptimizeStatus.Failed) {
-        optimizeRuntime.setRetry(optimizeRuntime.getRetry() + 1);
+      BaseOptimizeTaskRuntime newRuntime = optimizeRuntime.clone();
+      if (newRuntime.getStatus() == OptimizeStatus.Failed) {
+        newRuntime.setRetry(newRuntime.getRetry() + 1);
       }
-      optimizeRuntime.setPendingTime(System.currentTimeMillis());
-      optimizeRuntime.setStatus(OptimizeStatus.Pending);
-      persistTaskRuntime(false);
+      newRuntime.setPendingTime(System.currentTimeMillis());
+      newRuntime.setStatus(OptimizeStatus.Pending);
+      persistTaskRuntime(newRuntime, false);
+      optimizeRuntime = newRuntime;
     } finally {
       lock.unlock();
     }
@@ -94,12 +96,17 @@ public class OptimizeTaskItem extends IJDBCService {
   public TableTaskHistory onExecuting(JobId jobId, String attemptId) {
     lock.lock();
     try {
+      BaseOptimizeTaskRuntime newRuntime = optimizeRuntime.clone();
       long currentTime = System.currentTimeMillis();
-      optimizeRuntime.setAttemptId(attemptId);
-      optimizeRuntime.setExecuteTime(currentTime);
-      optimizeRuntime.setJobId(jobId);
-      optimizeRuntime.setStatus(OptimizeStatus.Executing);
-      persistTaskRuntime(false);
+      newRuntime.setAttemptId(attemptId);
+      newRuntime.setExecuteTime(currentTime);
+      newRuntime.setJobId(jobId);
+      newRuntime.setStatus(OptimizeStatus.Executing);
+      newRuntime.setPreparedTime(BaseOptimizeTaskRuntime.INVALID_TIME);
+      newRuntime.setCostTime(0);
+      newRuntime.setErrorMessage(null);
+      persistTaskRuntime(newRuntime, false);
+      optimizeRuntime = newRuntime;
       return constructTableTaskHistory(currentTime);
     } catch (Throwable t) {
       onFailed(new ErrorMessage(System.currentTimeMillis(),
@@ -113,10 +120,12 @@ public class OptimizeTaskItem extends IJDBCService {
   public void onCommitted(long commitTime) {
     lock.lock();
     try {
-      optimizeRuntime.setCommitTime(commitTime);
-      optimizeRuntime.setStatus(OptimizeStatus.Committed);
-      // after commit if will be delete, there is no need to update
-      persistTaskRuntime(false);
+      BaseOptimizeTaskRuntime newRuntime = optimizeRuntime.clone();
+      newRuntime.setCommitTime(commitTime);
+      newRuntime.setStatus(OptimizeStatus.Committed);
+      // after commit, task will be deleted, there is no need to update
+      persistTaskRuntime(newRuntime, false);
+      optimizeRuntime = newRuntime;
     } finally {
       lock.unlock();
     }
@@ -126,12 +135,15 @@ public class OptimizeTaskItem extends IJDBCService {
     long reportTime = System.currentTimeMillis();
     lock.lock();
     try {
-      optimizeRuntime.setErrorMessage(errorMessage);
-      optimizeRuntime.setStatus(OptimizeStatus.Failed);
-      optimizeRuntime.setReportTime(reportTime);
-      optimizeRuntime.setAttemptId(null);
-      optimizeRuntime.setCostTime(costTime);
-      persistTaskRuntime(false);
+      BaseOptimizeTaskRuntime newRuntime = optimizeRuntime.clone();
+      newRuntime.setErrorMessage(errorMessage);
+      newRuntime.setStatus(OptimizeStatus.Failed);
+      newRuntime.setPreparedTime(BaseOptimizeTaskRuntime.INVALID_TIME);
+      newRuntime.setReportTime(reportTime);
+      newRuntime.setAttemptId(null);
+      newRuntime.setCostTime(costTime);
+      persistTaskRuntime(newRuntime, false);
+      optimizeRuntime = newRuntime;
     } finally {
       lock.unlock();
     }
@@ -141,17 +153,19 @@ public class OptimizeTaskItem extends IJDBCService {
     long reportTime = System.currentTimeMillis();
     lock.lock();
     try {
-      if (optimizeRuntime.getExecuteTime() == BaseOptimizeTaskRuntime.INVALID_TIME) {
-        optimizeRuntime.setExecuteTime(preparedTime);
+      BaseOptimizeTaskRuntime newRuntime = optimizeRuntime.clone();
+      if (newRuntime.getExecuteTime() == BaseOptimizeTaskRuntime.INVALID_TIME) {
+        newRuntime.setExecuteTime(preparedTime);
       }
-      optimizeRuntime.setPreparedTime(preparedTime);
-      optimizeRuntime.setStatus(OptimizeStatus.Prepared);
-      optimizeRuntime.setReportTime(reportTime);
-      optimizeRuntime.setNewFileCnt(targetFiles == null ? 0 : targetFiles.size());
-      optimizeRuntime.setNewFileSize(newFileSize);
-      optimizeRuntime.setTargetFiles(targetFiles);
-      optimizeRuntime.setCostTime(costTime);
-      persistTaskRuntime(true);
+      newRuntime.setPreparedTime(preparedTime);
+      newRuntime.setStatus(OptimizeStatus.Prepared);
+      newRuntime.setReportTime(reportTime);
+      newRuntime.setNewFileCnt(targetFiles == null ? 0 : targetFiles.size());
+      newRuntime.setNewFileSize(newFileSize);
+      newRuntime.setTargetFiles(targetFiles);
+      newRuntime.setCostTime(costTime);
+      persistTaskRuntime(newRuntime, true);
+      optimizeRuntime = newRuntime;
     } finally {
       lock.unlock();
     }
@@ -228,18 +242,18 @@ public class OptimizeTaskItem extends IJDBCService {
     }
   }
 
-  private void persistTaskRuntime(boolean updateTargetFiles) {
+  private void persistTaskRuntime(BaseOptimizeTaskRuntime newRuntime, boolean updateTargetFiles) {
     try (SqlSession sqlSession = getSqlSession(false)) {
       OptimizeTaskRuntimesMapper optimizeTaskRuntimesMapper =
           getMapper(sqlSession, OptimizeTaskRuntimesMapper.class);
       InternalTableFilesMapper internalTableFilesMapper =
           getMapper(sqlSession, InternalTableFilesMapper.class);
 
-      optimizeTaskRuntimesMapper.updateOptimizeTaskRuntime(optimizeRuntime);
+      optimizeTaskRuntimesMapper.updateOptimizeTaskRuntime(newRuntime);
       if (updateTargetFiles) {
         try {
           internalTableFilesMapper.deleteOptimizeTaskTargetFile(optimizeTask.getTaskId());
-          optimizeRuntime.getTargetFiles().forEach(file -> {
+          newRuntime.getTargetFiles().forEach(file -> {
             ContentFile<?> contentFile = SerializationUtil.toInternalTableFile(file);
             if (contentFile instanceof DataFile) {
               internalTableFilesMapper.insertOptimizeTaskFile(optimizeTask.getTaskId(),
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/IJDBCService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/IJDBCService.java
index 623b52802..e0e37598b 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/IJDBCService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/IJDBCService.java
@@ -24,6 +24,7 @@ import com.netease.arctic.ams.server.utils.JDBCSqlSessionFactoryProvider;
 import org.apache.ibatis.io.ResolverUtil;
 import org.apache.ibatis.session.SqlSession;
 import org.apache.ibatis.session.SqlSessionFactory;
+import org.apache.ibatis.session.TransactionIsolationLevel;
 
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -42,7 +43,12 @@ public abstract class IJDBCService {
   }
 
   public SqlSession getSqlSession(boolean autoCommit) {
-    return getSqlSessionFactory().openSession(autoCommit);
+    if (autoCommit) {
+      return this.getSqlSessionFactory().openSession(true);
+    } else {
+      // when openSession with TransactionIsolationLevel, autoCommit is always set to false
+      return this.getSqlSessionFactory().openSession(TransactionIsolationLevel.READ_COMMITTED);
+    }
   }
 
   public <T> T getMapper(SqlSession sqlSession, Class<T> type) {
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/DDLTracerService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/DDLTracerService.java
index dfd600933..e28e9181d 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/DDLTracerService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/DDLTracerService.java
@@ -1,5 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package com.netease.arctic.ams.server.service.impl;
 
+import com.google.common.collect.Maps;
 import com.netease.arctic.ams.api.SchemaUpdateMeta;
 import com.netease.arctic.ams.api.TableIdentifier;
 import com.netease.arctic.ams.api.UpdateColumn;
@@ -14,7 +33,6 @@ import com.netease.arctic.catalog.ArcticCatalog;
 import com.netease.arctic.catalog.CatalogLoader;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.TableProperties;
-import com.netease.arctic.trace.AmsTableTracer;
 import com.netease.arctic.trace.TableTracer;
 import org.apache.ibatis.session.SqlSession;
 import org.apache.iceberg.Schema;
@@ -23,15 +41,12 @@ import org.apache.iceberg.types.Types;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
-import static com.netease.arctic.table.TableProperties.BASE_TABLE_MAX_TRANSACTION_ID;
-
 public class DDLTracerService extends IJDBCService {
 
   private static final Logger LOG = LoggerFactory.getLogger(DDLTracerService.class);
@@ -292,7 +307,7 @@ public class DDLTracerService extends IJDBCService {
       ServiceContainer.getDdlTracerService()
           .commitProperties(arcticTable.id().buildTableIdentifier(), tableMetadata.getProperties(),
               table.properties());
-      ServiceContainer.getMetaService().updateTableProperties(arcticTable.id(), table.properties());
+      ServiceContainer.getMetaService().updateTableProperties(arcticTable.id(), Maps.newHashMap(table.properties()));
     }
 
     public String compareSchema(String tableName, Schema before, Schema after) {
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/FileInfoCacheService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/FileInfoCacheService.java
index d44665509..e89a081c5 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/FileInfoCacheService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/FileInfoCacheService.java
@@ -127,6 +127,9 @@ public class FileInfoCacheService extends IJDBCService {
   }
 
   public Boolean snapshotIsCached(TableIdentifier identifier, String innerTable, Long snapshotId) {
+    if (snapshotId == -1) {
+      return true;
+    }
     try (SqlSession sqlSession = getSqlSession(true)) {
       SnapInfoCacheMapper snapInfoCacheMapper = getMapper(sqlSession, SnapInfoCacheMapper.class);
       return snapInfoCacheMapper.snapshotIsCached(identifier, innerTable, snapshotId);
@@ -380,8 +383,10 @@ public class FileInfoCacheService extends IJDBCService {
             cacheFileInfo.setInnerTable(tableChange.getInnerTable());
             cacheFileInfo.setFilePath(datafile.getPath());
             cacheFileInfo.setFileType(datafile.getFileType());
+            String partitionName = partitionToPath(datafile.getPartition());
+            cacheFileInfo.setPartitionName(StringUtils.isEmpty(partitionName) ? "" : partitionName);
             String primaryKey = TableMetadataUtil.getTableAllIdentifyName(tableCommitMeta.getTableIdentifier()) +
-                tableChange.getInnerTable() + datafile.getPath();
+                tableChange.getInnerTable() + datafile.getPath() + partitionName;
             String primaryKeyMd5 = Hashing.md5()
                 .hashBytes(primaryKey.getBytes(StandardCharsets.UTF_8))
                 .toString();
@@ -402,8 +407,6 @@ public class FileInfoCacheService extends IJDBCService {
               cacheFileInfo.setWatermark(0L);
             }
             cacheFileInfo.setAction(tableCommitMeta.getAction());
-            String partitionName = partitionToPath(datafile.getPartition());
-            cacheFileInfo.setPartitionName(StringUtils.isEmpty(partitionName) ? null : partitionName);
             cacheFileInfo.setCommitTime(tableCommitMeta.getCommitTime());
             rs.add(cacheFileInfo);
           });
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/RuntimeDataExpireService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/RuntimeDataExpireService.java
index 9353157c9..e099a98e1 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/RuntimeDataExpireService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/RuntimeDataExpireService.java
@@ -1,3 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package com.netease.arctic.ams.server.service.impl;
 
 import com.netease.arctic.ams.server.model.TableMetadata;
diff --git a/core/src/main/java/com/netease/arctic/io/ArcticHadoopFileIO.java b/core/src/main/java/com/netease/arctic/io/ArcticHadoopFileIO.java
index 21c94d98f..c3b78c13a 100644
--- a/core/src/main/java/com/netease/arctic/io/ArcticHadoopFileIO.java
+++ b/core/src/main/java/com/netease/arctic/io/ArcticHadoopFileIO.java
@@ -43,7 +43,6 @@ import java.util.concurrent.Callable;
  * Implementation of {@link ArcticFileIO} for hadoop file system with authentication.
  */
 public class ArcticHadoopFileIO extends HadoopFileIO implements ArcticFileIO {
-  private static final Logger LOG = LoggerFactory.getLogger(ArcticHadoopFileIO.class);
   private final TableMetaStore tableMetaStore;
 
   public ArcticHadoopFileIO(TableMetaStore tableMetaStore) {
@@ -180,6 +179,10 @@ public class ArcticHadoopFileIO extends HadoopFileIO implements ArcticFileIO {
     });
   }
 
+  public TableMetaStore getTableMetaStore() {
+    return tableMetaStore;
+  }
+
   private FileSystem getFs(Path path) {
     return Util.getFs(path, conf());
   }
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 b2d960f02..c43508ff4 100644
--- a/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java
+++ b/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java
@@ -34,18 +34,22 @@ 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
  */
 public class OverwriteBaseFiles extends PartitionTransactionOperation {
 
+  public static final String PROPERTIES_TRANSACTION_ID = "txId";
+
   private final List<DataFile> deleteFiles;
   private final List<DataFile> addFiles;
   private final List<DeleteFile> deleteDeleteFiles;
@@ -118,14 +122,14 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
       partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
       partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
     }
-    // 为什么？
+
     for (DataFile d : this.deleteFiles) {
       overwriteFiles.deleteFile(d);
       partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
       partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
     }
     if (transactionId != null && transactionId > 0) {
-      overwriteFiles.set("txId", transactionId + "");
+      overwriteFiles.set(PROPERTIES_TRANSACTION_ID, transactionId + "");
     }
 
     if (MapUtils.isNotEmpty(properties)) {
@@ -145,7 +149,7 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
           partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
           partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
         }
-        // 为什么？
+
         for (DataFile d : this.deleteFiles) {
           partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
           partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
@@ -165,7 +169,7 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
           partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
           partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
         }
-        // 为什么？
+
         for (DataFile d : this.deleteFiles) {
           partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
           partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
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 46e566f67..4149d0620 100644
--- a/core/src/main/java/com/netease/arctic/op/PartitionPropertiesUpdate.java
+++ b/core/src/main/java/com/netease/arctic/op/PartitionPropertiesUpdate.java
@@ -84,7 +84,6 @@ public class PartitionPropertiesUpdate implements UpdatePartitionProperties {
 
   @Override
   public StructLikeMap<Map<String, String>> apply() {
-    table.refresh();
     StructLikeMap<Map<String, String>> partitionProperties = table.partitionProperty();
     setProperties.forEach((partitionData, properties) -> {
       Map<String, String> oldProperties = partitionProperties.get(partitionData);
diff --git a/core/src/main/java/com/netease/arctic/table/ArcticTable.java b/core/src/main/java/com/netease/arctic/table/ArcticTable.java
index 9e0ea7654..fa22196df 100644
--- a/core/src/main/java/com/netease/arctic/table/ArcticTable.java
+++ b/core/src/main/java/com/netease/arctic/table/ArcticTable.java
@@ -21,6 +21,8 @@ package com.netease.arctic.table;
 import com.netease.arctic.io.ArcticFileIO;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.UpdateSchema;
 
 import java.io.Serializable;
 import java.util.Map;
@@ -70,6 +72,20 @@ public interface ArcticTable extends Serializable {
    */
   void refresh();
 
+  /**
+   * Create a new {@link UpdateSchema} to alter the columns of this table and commit the change.
+   *
+   * @return a new {@link UpdateSchema}
+   */
+  UpdateSchema updateSchema();
+
+  /**
+   * Create a new {@link UpdateProperties} to update table properties and commit the changes.
+   *
+   * @return a new {@link UpdateProperties}
+   */
+  UpdateProperties updateProperties();
+
   /**
    * Returns true if this table is an {@link UnkeyedTable}
    */
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 e37e744a6..40cacde93 100644
--- a/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
@@ -109,10 +109,7 @@ public class BaseKeyedTable implements KeyedTable {
 
   @Override
   public Map<String, String> properties() {
-    return baseTable.properties().entrySet()
-        .stream()
-        .filter(e -> !TableProperties.PROTECTED_PROPERTIES.contains(e.getKey()))
-        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    return baseTable.properties();
   }
 
   @Override
@@ -224,10 +221,6 @@ public class BaseKeyedTable implements KeyedTable {
       super(tableIdentifier, baseIcebergTable, arcticFileIO, client);
     }
 
-    @Override
-    public Map<String, String> properties() {
-      return Maps.newHashMap(icebergTable.properties());
-    }
   }
 
   public static class ChangeInternalTable extends BaseUnkeyedTable implements ChangeTable {
diff --git a/core/src/main/java/com/netease/arctic/table/BaseUnkeyedTable.java b/core/src/main/java/com/netease/arctic/table/BaseUnkeyedTable.java
index c6c71541a..aef3599cc 100644
--- a/core/src/main/java/com/netease/arctic/table/BaseUnkeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/BaseUnkeyedTable.java
@@ -141,10 +141,7 @@ public class BaseUnkeyedTable implements UnkeyedTable, HasTableOperations {
 
   @Override
   public Map<String, String> properties() {
-    return icebergTable.properties().entrySet()
-        .stream()
-        .filter(e -> !TableProperties.PROTECTED_PROPERTIES.contains(e.getKey()))
-        .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+    return icebergTable.properties();
   }
 
   @Override
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 c5872be2c..eed9f0479 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,8 @@ 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.UpdateProperties;
-import org.apache.iceberg.UpdateSchema;
 import org.apache.iceberg.util.StructLikeMap;
+
 import java.util.Map;
 
 /**
@@ -67,20 +66,6 @@ public interface KeyedTable extends ArcticTable {
    */
   KeyedTableScan newScan();
 
-  /**
-   * Create a new {@link UpdateSchema} to alter the columns of this table and commit the change.
-   *
-   * @return a new {@link UpdateSchema}
-   */
-  UpdateSchema updateSchema();
-
-  /**
-   * Create a new {@link UpdateProperties} to update table properties and commit the changes.
-   *
-   * @return a new {@link UpdateProperties}
-   */
-  UpdateProperties updateProperties();
-
   /**
    * Allocate a new transaction id from this table
    *
diff --git a/core/src/main/java/com/netease/arctic/table/TableProperties.java b/core/src/main/java/com/netease/arctic/table/TableProperties.java
index 8eacfe861..327fc9678 100644
--- a/core/src/main/java/com/netease/arctic/table/TableProperties.java
+++ b/core/src/main/java/com/netease/arctic/table/TableProperties.java
@@ -43,12 +43,8 @@ public class TableProperties {
 
   public static final String TABLE_PARTITION_PROPERTIES = "table.partition-properties";
 
-  public static final String PARTITION_PROPERTIES_KEY_HIVE_LOCATION = "hive-location";
-
   public static final String BASE_TABLE_MAX_TRANSACTION_ID = "base.table.max-transaction-id";
 
-  public static final String BASE_HIVE_LOCATION_ROOT = "base.hive.location-root";
-
   public static final String LOCATION = "location";
 
   public static final String TABLE_CREATE_TIME = "table.create-timestamp";
diff --git a/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java b/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java
index d2beb7bd6..a992933d6 100644
--- a/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java
+++ b/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java
@@ -161,7 +161,6 @@ public class AmsTableTracer implements TableTracer {
       update = true;
     }
     if (updateColumns.size() > 0 && Constants.INNER_TABLE_BASE.equals(innerTable)) {
-      table.refresh();
       int schemaId = table.schema().schemaId();
       SchemaUpdateMeta ddlCommitMeta = new SchemaUpdateMeta();
       ddlCommitMeta.setSchemaId(schemaId);
diff --git a/dist/src/main/arctic-bin/bin/ams.sh b/dist/src/main/arctic-bin/bin/ams.sh
index da64af46d..139da4728 100644
--- a/dist/src/main/arctic-bin/bin/ams.sh
+++ b/dist/src/main/arctic-bin/bin/ams.sh
@@ -67,7 +67,7 @@ CMDS="$JAVA_RUN -Dlog.home=${LOG_DIR} -Dlog.dir=${LOG_DIR} -Duser.dir=${ARCTIC_H
 #0:pid bad and proc OK;   1:pid ok and proc bad;    2:pid bad
 function status(){
     test -e ${PID} || return 2
-    test -d /proc/$(cat ${PID}) && return 0 || return 1
+    test -n "$(ps -p $(cat ${PID}) -o pid=)" && return 0 || return 1
 }
 
 function start() {
diff --git a/dist/src/main/assemblies/bin.xml b/dist/src/main/assemblies/bin.xml
index 555a86757..786f21a20 100644
--- a/dist/src/main/assemblies/bin.xml
+++ b/dist/src/main/assemblies/bin.xml
@@ -30,8 +30,8 @@
     <files>
         <file>
             <source>../optimizer/target/arctic-optimizer-${project.version}-jar-with-dependencies.jar</source>
-            <outputDirectory>lib/</outputDirectory>
-            <destName>CompactJob.jar</destName>
+            <outputDirectory>plugin/optimize</outputDirectory>
+            <destName>OptimizeJob.jar</destName>
             <fileMode>0644</fileMode>
         </file>
         <file>
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/descriptors/ArcticValidator.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/descriptors/ArcticValidator.java
index b0607aaf5..fd1450236 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/descriptors/ArcticValidator.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/descriptors/ArcticValidator.java
@@ -105,6 +105,15 @@ public class ArcticValidator extends ConnectorDescriptorValidator {
           " the current snapshot, \"latest\": read all incremental data in the change table starting from the" +
           " current snapshot (the current snapshot will be excluded).");
 
+  public static final ConfigOption<Boolean> SUBMIT_EMPTY_SNAPSHOTS = ConfigOptions
+      .key("submit.empty.snapshots")
+      .booleanType()
+      .defaultValue(true)
+      .withDescription("Optional submit empty snapshots to the arctic table, false means that writers will not emit" +
+          " empty WriteResults to the committer operator, and reduce the number of snapshots in File Cache; true" +
+          " means this job will submit empty snapshots to the table, it is suitable with some valid reasons, e.g." +
+          " advance watermark metadata stored in the table(https://github.com/apache/iceberg/pull/5561).");
+
   @Override
   public void validate(DescriptorProperties properties) {
     String emitMode = properties.getString(ARCTIC_EMIT_MODE);
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/ArcticFileWriter.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/ArcticFileWriter.java
index 3e7a8b7de..b0057c90c 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/ArcticFileWriter.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/ArcticFileWriter.java
@@ -24,6 +24,7 @@ import com.netease.arctic.flink.shuffle.ShuffleRulePolicy;
 import com.netease.arctic.flink.table.ArcticTableLoader;
 import com.netease.arctic.flink.util.ArcticUtils;
 import com.netease.arctic.table.ArcticTable;
+import org.apache.commons.lang.ArrayUtils;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
 import org.apache.flink.streaming.api.operators.BoundedOneInput;
@@ -59,6 +60,7 @@ public class ArcticFileWriter extends AbstractStreamOperator<WriteResult>
   private final int minFileSplitCount;
   private final ArcticTableLoader tableLoader;
   private final boolean upsert;
+  private final boolean submitEmptySnapshot;
 
   private transient org.apache.iceberg.io.TaskWriter<RowData> writer;
   private transient int subTaskId;
@@ -77,12 +79,16 @@ public class ArcticFileWriter extends AbstractStreamOperator<WriteResult>
       TaskWriterFactory<RowData> taskWriterFactory,
       int minFileSplitCount,
       ArcticTableLoader tableLoader,
-      boolean upsert) {
+      boolean upsert,
+      boolean submitEmptySnapshot) {
     this.shuffleRule = shuffleRule;
     this.taskWriterFactory = taskWriterFactory;
     this.minFileSplitCount = minFileSplitCount;
     this.tableLoader = tableLoader;
     this.upsert = upsert;
+    this.submitEmptySnapshot = submitEmptySnapshot;
+    LOG.info("ArcticFileWriter is created with minFileSplitCount: {}, upsert: {}, submitEmptySnapshot: {}",
+        minFileSplitCount, upsert, submitEmptySnapshot);
   }
 
   @Override
@@ -198,7 +204,24 @@ public class ArcticFileWriter extends AbstractStreamOperator<WriteResult>
   }
 
   private void emit(WriteResult writeResult) {
-    output.collect(new StreamRecord<>(writeResult));
+    if (shouldEmit(writeResult)) {
+      // Only emit a non-empty WriteResult to committer operator, thus avoiding submitting too much empty snapshots.
+      output.collect(new StreamRecord<>(writeResult));
+    }
+  }
+
+  /**
+   * Whether to emit the WriteResult.
+   *
+   * @param writeResult the WriteResult to emit
+   * @return true if the WriteResult should be emitted, or the WriteResult isn't empty,
+   *         false only if the WriteResult is empty and the submitEmptySnapshot is false.
+   */
+  private boolean shouldEmit(WriteResult writeResult) {
+    return submitEmptySnapshot || (writeResult != null &&
+        (!ArrayUtils.isEmpty(writeResult.dataFiles()) ||
+            !ArrayUtils.isEmpty(writeResult.deleteFiles()) ||
+            !ArrayUtils.isEmpty(writeResult.referencedDataFiles())));
   }
 
   @VisibleForTesting
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
index d1ba0d3c2..163891f42 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
@@ -67,6 +67,7 @@ import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_
 import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_THROUGHPUT_METRIC_ENABLE_DEFAULT;
 import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_WRITE_MAX_OPEN_FILE_SIZE;
 import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_WRITE_MAX_OPEN_FILE_SIZE_DEFAULT;
+import static com.netease.arctic.flink.table.descriptors.ArcticValidator.SUBMIT_EMPTY_SNAPSHOTS;
 import static com.netease.arctic.table.TableProperties.DEFAULT_FILE_FORMAT;
 import static com.netease.arctic.table.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
 import static com.netease.arctic.table.TableProperties.WRITE_DISTRIBUTION_HASH_DEFAULT;
@@ -148,10 +149,12 @@ public class FlinkSink {
         ArcticFileWriter fileWriter,
         OneInputStreamOperator<WriteResult, Void> committer,
         int writeOperatorParallelism,
-        MetricsGenerator metricsGenerator) {
+        MetricsGenerator metricsGenerator,
+        String arcticEmitMode) {
       SingleOutputStreamOperator writerStream = input
           .transform(ArcticWriter.class.getName(), TypeExtractor.createTypeInfo(WriteResult.class),
               new ArcticWriter<>(logWriter, fileWriter, metricsGenerator))
+          .name(String.format("ArcticWriter %s(%s)", table.name(), arcticEmitMode))
           .setParallelism(writeOperatorParallelism);
 
       if (committer != null) {
@@ -216,7 +219,8 @@ public class FlinkSink {
           fileWriter,
           createFileCommitter(table, tableLoader, overwrite, arcticEmitMode),
           writeOperatorParallelism,
-          metricsGenerator);
+          metricsGenerator,
+          arcticEmitMode);
     }
 
     private void initTableIfNeeded() {
@@ -325,11 +329,15 @@ public class FlinkSink {
 
     boolean upsert = arcticTable.isKeyedTable() && PropertyUtil.propertyAsBoolean(arcticTable.properties(),
         TableProperties.UPSERT_ENABLED, TableProperties.UPSERT_ENABLED_DEFAULT);
+    boolean submitEmptySnapshot = PropertyUtil.propertyAsBoolean(
+        arcticTable.properties(), SUBMIT_EMPTY_SNAPSHOTS.key(), SUBMIT_EMPTY_SNAPSHOTS.defaultValue());
+
     return new ArcticFileWriter(shufflePolicy,
         createTaskWriterFactory(arcticTable, overwrite, flinkSchema, equalityColumns),
         minFileSplitCount,
         tableLoader,
-        upsert);
+        upsert,
+        submitEmptySnapshot);
   }
 
   private static TaskWriterFactory<RowData> createTaskWriterFactory(ArcticTable arcticTable,
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/descriptors/ArcticValidator.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/descriptors/ArcticValidator.java
index b0607aaf5..fd1450236 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/descriptors/ArcticValidator.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/descriptors/ArcticValidator.java
@@ -105,6 +105,15 @@ public class ArcticValidator extends ConnectorDescriptorValidator {
           " the current snapshot, \"latest\": read all incremental data in the change table starting from the" +
           " current snapshot (the current snapshot will be excluded).");
 
+  public static final ConfigOption<Boolean> SUBMIT_EMPTY_SNAPSHOTS = ConfigOptions
+      .key("submit.empty.snapshots")
+      .booleanType()
+      .defaultValue(true)
+      .withDescription("Optional submit empty snapshots to the arctic table, false means that writers will not emit" +
+          " empty WriteResults to the committer operator, and reduce the number of snapshots in File Cache; true" +
+          " means this job will submit empty snapshots to the table, it is suitable with some valid reasons, e.g." +
+          " advance watermark metadata stored in the table(https://github.com/apache/iceberg/pull/5561).");
+
   @Override
   public void validate(DescriptorProperties properties) {
     String emitMode = properties.getString(ARCTIC_EMIT_MODE);
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/ArcticFileWriter.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/ArcticFileWriter.java
index 19730bdd9..fc4a156cc 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/ArcticFileWriter.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/ArcticFileWriter.java
@@ -24,7 +24,7 @@ import com.netease.arctic.flink.shuffle.ShuffleRulePolicy;
 import com.netease.arctic.flink.table.ArcticTableLoader;
 import com.netease.arctic.flink.util.ArcticUtils;
 import com.netease.arctic.table.ArcticTable;
-import com.netease.arctic.table.KeyedTable;
+import org.apache.commons.lang.ArrayUtils;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
 import org.apache.flink.streaming.api.operators.BoundedOneInput;
@@ -60,6 +60,7 @@ public class ArcticFileWriter extends AbstractStreamOperator<WriteResult>
   private final int minFileSplitCount;
   private final ArcticTableLoader tableLoader;
   private final boolean upsert;
+  private final boolean submitEmptySnapshot;
 
   private transient org.apache.iceberg.io.TaskWriter<RowData> writer;
   private transient int subTaskId;
@@ -78,14 +79,19 @@ public class ArcticFileWriter extends AbstractStreamOperator<WriteResult>
       TaskWriterFactory<RowData> taskWriterFactory,
       int minFileSplitCount,
       ArcticTableLoader tableLoader,
-      boolean upsert) {
+      boolean upsert,
+      boolean submitEmptySnapshot) {
     this.shuffleRule = shuffleRule;
     this.taskWriterFactory = taskWriterFactory;
     this.minFileSplitCount = minFileSplitCount;
     this.tableLoader = tableLoader;
     this.upsert = upsert;
+    this.submitEmptySnapshot = submitEmptySnapshot;
+    LOG.info("ArcticFileWriter is created with minFileSplitCount: {}, upsert: {}, submitEmptySnapshot: {}",
+        minFileSplitCount, upsert, submitEmptySnapshot);
   }
 
+
   @Override
   public void open() {
     this.subTaskId = getRuntimeContext().getIndexOfThisSubtask();
@@ -199,7 +205,24 @@ public class ArcticFileWriter extends AbstractStreamOperator<WriteResult>
   }
 
   private void emit(WriteResult writeResult) {
-    output.collect(new StreamRecord<>(writeResult));
+    if (shouldEmit(writeResult)) {
+      // Only emit a non-empty WriteResult to committer operator, thus avoiding submitting too much empty snapshots.
+      output.collect(new StreamRecord<>(writeResult));
+    }
+  }
+
+  /**
+   * Whether to emit the WriteResult.
+   *
+   * @param writeResult the WriteResult to emit
+   * @return true if the WriteResult should be emitted, or the WriteResult isn't empty,
+   *         false only if the WriteResult is empty and the submitEmptySnapshot is false.
+   */
+  private boolean shouldEmit(WriteResult writeResult) {
+    return submitEmptySnapshot || (writeResult != null &&
+        (!ArrayUtils.isEmpty(writeResult.dataFiles()) ||
+            !ArrayUtils.isEmpty(writeResult.deleteFiles()) ||
+            !ArrayUtils.isEmpty(writeResult.referencedDataFiles())));
   }
 
   @VisibleForTesting
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
index 54ef540dd..1b49f5e12 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
@@ -67,6 +67,7 @@ import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_
 import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_THROUGHPUT_METRIC_ENABLE_DEFAULT;
 import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_WRITE_MAX_OPEN_FILE_SIZE;
 import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_WRITE_MAX_OPEN_FILE_SIZE_DEFAULT;
+import static com.netease.arctic.flink.table.descriptors.ArcticValidator.SUBMIT_EMPTY_SNAPSHOTS;
 import static com.netease.arctic.table.TableProperties.DEFAULT_FILE_FORMAT;
 import static com.netease.arctic.table.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
 import static com.netease.arctic.table.TableProperties.WRITE_DISTRIBUTION_HASH_DEFAULT;
@@ -148,10 +149,12 @@ public class FlinkSink {
         ArcticFileWriter fileWriter,
         OneInputStreamOperator<WriteResult, Void> committer,
         int writeOperatorParallelism,
-        MetricsGenerator metricsGenerator) {
+        MetricsGenerator metricsGenerator,
+        String arcticEmitMode) {
       SingleOutputStreamOperator writerStream = input
           .transform(ArcticWriter.class.getName(), TypeExtractor.createTypeInfo(WriteResult.class),
               new ArcticWriter<>(logWriter, fileWriter, metricsGenerator))
+          .name(String.format("ArcticWriter %s(%s)", table.name(), arcticEmitMode))
           .setParallelism(writeOperatorParallelism);
 
       if (committer != null) {
@@ -217,7 +220,8 @@ public class FlinkSink {
           fileWriter,
           createFileCommitter(table, tableLoader, overwrite, arcticEmitMode),
           writeOperatorParallelism,
-          metricsGenerator);
+          metricsGenerator,
+          arcticEmitMode);
     }
 
     private void initTableIfNeeded() {
@@ -332,12 +336,16 @@ public class FlinkSink {
 
     boolean upsert = arcticTable.isKeyedTable() && PropertyUtil.propertyAsBoolean(arcticTable.properties(),
         TableProperties.UPSERT_ENABLED, TableProperties.UPSERT_ENABLED_DEFAULT);
+    boolean submitEmptySnapshot = PropertyUtil.propertyAsBoolean(
+        arcticTable.properties(), SUBMIT_EMPTY_SNAPSHOTS.key(), SUBMIT_EMPTY_SNAPSHOTS.defaultValue());
+
     return new ArcticFileWriter(
         shufflePolicy,
         createTaskWriterFactory(arcticTable, overwrite, flinkSchema, equalityColumns),
         minFileSplitCount,
         tableLoader,
-        upsert);
+        upsert,
+        submitEmptySnapshot);
   }
 
   private static TaskWriterFactory<RowData> createTaskWriterFactory(
diff --git a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/table/descriptors/ArcticValidator.java b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/table/descriptors/ArcticValidator.java
index b0607aaf5..fd1450236 100644
--- a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/table/descriptors/ArcticValidator.java
+++ b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/table/descriptors/ArcticValidator.java
@@ -105,6 +105,15 @@ public class ArcticValidator extends ConnectorDescriptorValidator {
           " the current snapshot, \"latest\": read all incremental data in the change table starting from the" +
           " current snapshot (the current snapshot will be excluded).");
 
+  public static final ConfigOption<Boolean> SUBMIT_EMPTY_SNAPSHOTS = ConfigOptions
+      .key("submit.empty.snapshots")
+      .booleanType()
+      .defaultValue(true)
+      .withDescription("Optional submit empty snapshots to the arctic table, false means that writers will not emit" +
+          " empty WriteResults to the committer operator, and reduce the number of snapshots in File Cache; true" +
+          " means this job will submit empty snapshots to the table, it is suitable with some valid reasons, e.g." +
+          " advance watermark metadata stored in the table(https://github.com/apache/iceberg/pull/5561).");
+
   @Override
   public void validate(DescriptorProperties properties) {
     String emitMode = properties.getString(ARCTIC_EMIT_MODE);
diff --git a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/write/ArcticFileWriter.java b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/write/ArcticFileWriter.java
index 4fb61a5a9..8a27ea9dc 100644
--- a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/write/ArcticFileWriter.java
+++ b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/write/ArcticFileWriter.java
@@ -24,6 +24,7 @@ import com.netease.arctic.flink.shuffle.ShuffleRulePolicy;
 import com.netease.arctic.flink.table.ArcticTableLoader;
 import com.netease.arctic.flink.util.ArcticUtils;
 import com.netease.arctic.table.ArcticTable;
+import org.apache.commons.lang.ArrayUtils;
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
 import org.apache.flink.streaming.api.operators.BoundedOneInput;
@@ -59,6 +60,7 @@ public class ArcticFileWriter extends AbstractStreamOperator<WriteResult>
   private final int minFileSplitCount;
   private final ArcticTableLoader tableLoader;
   private final boolean upsert;
+  private final boolean submitEmptySnapshot;
 
   private transient org.apache.iceberg.io.TaskWriter<RowData> writer;
   private transient int subTaskId;
@@ -77,12 +79,16 @@ public class ArcticFileWriter extends AbstractStreamOperator<WriteResult>
       TaskWriterFactory<RowData> taskWriterFactory,
       int minFileSplitCount,
       ArcticTableLoader tableLoader,
-      boolean upsert) {
+      boolean upsert,
+      boolean submitEmptySnapshot) {
     this.shuffleRule = shuffleRule;
     this.taskWriterFactory = taskWriterFactory;
     this.minFileSplitCount = minFileSplitCount;
     this.tableLoader = tableLoader;
     this.upsert = upsert;
+    this.submitEmptySnapshot = submitEmptySnapshot;
+    LOG.info("ArcticFileWriter is created with minFileSplitCount: {}, upsert: {}, submitEmptySnapshot: {}",
+        minFileSplitCount, upsert, submitEmptySnapshot);
   }
 
   @Override
@@ -198,7 +204,24 @@ public class ArcticFileWriter extends AbstractStreamOperator<WriteResult>
   }
 
   private void emit(WriteResult writeResult) {
-    output.collect(new StreamRecord<>(writeResult));
+    if (shouldEmit(writeResult)) {
+      // Only emit a non-empty WriteResult to committer operator, thus avoiding submitting too much empty snapshots.
+      output.collect(new StreamRecord<>(writeResult));
+    }
+  }
+
+  /**
+   * Whether to emit the WriteResult.
+   *
+   * @param writeResult the WriteResult to emit
+   * @return true if the WriteResult should be emitted, or the WriteResult isn't empty,
+   *         false only if the WriteResult is empty and the submitEmptySnapshot is false.
+   */
+  private boolean shouldEmit(WriteResult writeResult) {
+    return submitEmptySnapshot || (writeResult != null &&
+        (!ArrayUtils.isEmpty(writeResult.dataFiles()) ||
+            !ArrayUtils.isEmpty(writeResult.deleteFiles()) ||
+            !ArrayUtils.isEmpty(writeResult.referencedDataFiles())));
   }
 
   @VisibleForTesting
diff --git a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
index 0976ca0ca..16c57754a 100644
--- a/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
+++ b/flink/v1.15/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
@@ -68,6 +68,7 @@ import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_
 import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_THROUGHPUT_METRIC_ENABLE_DEFAULT;
 import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_WRITE_MAX_OPEN_FILE_SIZE;
 import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_WRITE_MAX_OPEN_FILE_SIZE_DEFAULT;
+import static com.netease.arctic.flink.table.descriptors.ArcticValidator.SUBMIT_EMPTY_SNAPSHOTS;
 import static com.netease.arctic.table.TableProperties.DEFAULT_FILE_FORMAT;
 import static com.netease.arctic.table.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
 import static com.netease.arctic.table.TableProperties.WRITE_DISTRIBUTION_HASH_DEFAULT;
@@ -155,10 +156,12 @@ public class FlinkSink {
         ArcticFileWriter fileWriter,
         OneInputStreamOperator<WriteResult, Void> committer,
         int writeOperatorParallelism,
-        MetricsGenerator metricsGenerator) {
+        MetricsGenerator metricsGenerator,
+        String arcticEmitMode) {
       SingleOutputStreamOperator writerStream = input
           .transform(ArcticWriter.class.getName(), TypeExtractor.createTypeInfo(WriteResult.class),
               new ArcticWriter<>(logWriter, fileWriter, metricsGenerator))
+          .name(String.format("ArcticWriter %s(%s)", table.name(), arcticEmitMode))
           .setParallelism(writeOperatorParallelism);
       context.generateUid("arctic-writer").ifPresent(writerStream::uid);
 
@@ -226,7 +229,8 @@ public class FlinkSink {
           fileWriter,
           createFileCommitter(table, tableLoader, overwrite, arcticEmitMode),
           writeOperatorParallelism,
-          metricsGenerator);
+          metricsGenerator,
+          arcticEmitMode);
     }
 
     private void initTableIfNeeded() {
@@ -341,12 +345,16 @@ public class FlinkSink {
 
     boolean upsert = arcticTable.isKeyedTable() && PropertyUtil.propertyAsBoolean(arcticTable.properties(),
         TableProperties.UPSERT_ENABLED, TableProperties.UPSERT_ENABLED_DEFAULT);
+    boolean submitEmptySnapshot = PropertyUtil.propertyAsBoolean(
+        arcticTable.properties(), SUBMIT_EMPTY_SNAPSHOTS.key(), SUBMIT_EMPTY_SNAPSHOTS.defaultValue());
+
     return new ArcticFileWriter(
         shufflePolicy,
         createTaskWriterFactory(arcticTable, overwrite, flinkSchema, equalityColumns),
         minFileSplitCount,
         tableLoader,
-        upsert);
+        upsert,
+        submitEmptySnapshot);
   }
 
   private static TaskWriterFactory<RowData> createTaskWriterFactory(
diff --git a/hive/src/main/java/com/netease/arctic/hive/ArcticHiveClientPool.java b/hive/src/main/java/com/netease/arctic/hive/ArcticHiveClientPool.java
index cf66f88f8..f093a72dd 100644
--- a/hive/src/main/java/com/netease/arctic/hive/ArcticHiveClientPool.java
+++ b/hive/src/main/java/com/netease/arctic/hive/ArcticHiveClientPool.java
@@ -1,7 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package com.netease.arctic.hive;
 
 import com.netease.arctic.table.TableMetaStore;
-import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
 import org.apache.iceberg.hive.HiveClientPool;
 import org.slf4j.Logger;
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 6c472b4b8..3fddfeb64 100644
--- a/hive/src/main/java/com/netease/arctic/hive/HiveTableProperties.java
+++ b/hive/src/main/java/com/netease/arctic/hive/HiveTableProperties.java
@@ -1,3 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package com.netease.arctic.hive;
 
 
@@ -7,5 +25,16 @@ public class HiveTableProperties {
 
   public static final String ARCTIC_TABLE_PRIMARY_KEYS = "arctic.table.primary-keys";
 
+  public static final String PARTITION_PROPERTIES_KEY_HIVE_LOCATION = "hive-location";
+
+  public static final String PARTITION_PROPERTIES_KEY_TRANSIENT_TIME = "transient-time";
+
+  public static final String BASE_HIVE_LOCATION_ROOT = "base.hive.location-root";
+
+  public static final String AUTO_SYNC_HIVE_SCHEMA_CHANGE = "base.hive.auto-sync-schema-change";
+  public static final boolean AUTO_SYNC_HIVE_SCHEMA_CHANGE_DEFAULT = true;
+
+  public static final String AUTO_SYNC_HIVE_DATA_WRITE = "base.hive.auto-sync-data-write";
+  public static final boolean AUTO_SYNC_HIVE_DATA_WRITE_DEFAULT = false;
 }
 
diff --git a/hive/src/main/java/com/netease/arctic/hive/catalog/ArcticHiveCatalog.java b/hive/src/main/java/com/netease/arctic/hive/catalog/ArcticHiveCatalog.java
index dba0db00b..546b6b45d 100644
--- a/hive/src/main/java/com/netease/arctic/hive/catalog/ArcticHiveCatalog.java
+++ b/hive/src/main/java/com/netease/arctic/hive/catalog/ArcticHiveCatalog.java
@@ -28,10 +28,10 @@ import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.table.KeyedHiveTable;
 import com.netease.arctic.hive.table.UnkeyedHiveTable;
 import com.netease.arctic.hive.utils.HiveSchemaUtil;
+import com.netease.arctic.hive.utils.HiveTableUtil;
 import com.netease.arctic.io.ArcticFileIO;
 import com.netease.arctic.io.ArcticHadoopFileIO;
 import com.netease.arctic.table.BaseKeyedTable;
-import com.netease.arctic.table.BaseTable;
 import com.netease.arctic.table.ChangeTable;
 import com.netease.arctic.table.TableBuilder;
 import com.netease.arctic.table.TableIdentifier;
@@ -45,6 +45,8 @@ import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.PartitionField;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.mapping.MappingUtil;
+import org.apache.iceberg.mapping.NameMappingParser;
 import org.apache.iceberg.util.PropertyUtil;
 import org.apache.thrift.TException;
 import org.slf4j.Logger;
@@ -142,7 +144,7 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
     Table baseIcebergTable = tableMetaStore.doAs(() -> tables.load(baseLocation));
     UnkeyedHiveTable baseTable = new UnkeyedHiveTable(tableIdentifier,
         useArcticTableOperations(baseIcebergTable, baseLocation, fileIO, tableMetaStore.getConfiguration()),
-        fileIO, tableLocation, client, hiveClientPool);
+        fileIO, tableLocation, client, hiveClientPool, false);
 
     Table changeIcebergTable = tableMetaStore.doAs(() -> tables.load(changeLocation));
     ChangeTable changeTable = new BaseKeyedTable.ChangeInternalTable(tableIdentifier,
@@ -201,11 +203,13 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
       String tableLocation = checkLocation(meta, MetaTableProperties.LOCATION_KEY_TABLE);
       String baseLocation = checkLocation(meta, MetaTableProperties.LOCATION_KEY_BASE);
       String changeLocation = checkLocation(meta, MetaTableProperties.LOCATION_KEY_CHANGE);
-      String hiveLocation = tableLocation + "/hive";
+      String hiveLocation = HiveTableUtil.hiveRootLocation(tableLocation);
 
       meta.putToProperties(TableProperties.TABLE_CREATE_TIME, String.valueOf(System.currentTimeMillis()));
       meta.putToProperties(org.apache.iceberg.TableProperties.FORMAT_VERSION, "2");
-      meta.putToProperties(TableProperties.BASE_HIVE_LOCATION_ROOT, hiveLocation);
+      meta.putToProperties(HiveTableProperties.BASE_HIVE_LOCATION_ROOT, hiveLocation);
+      meta.putToProperties(org.apache.iceberg.TableProperties.DEFAULT_NAME_MAPPING,
+          NameMappingParser.toJson(MappingUtil.create(schema)));
 
       ArcticFileIO fileIO = new ArcticHadoopFileIO(tableMetaStore);
       Table baseIcebergTable = tableMetaStore.doAs(() -> {
@@ -217,7 +221,7 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
       });
       UnkeyedHiveTable baseTable = new UnkeyedHiveTable(tableIdentifier,
           useArcticTableOperations(baseIcebergTable, baseLocation, fileIO, tableMetaStore.getConfiguration()),
-          fileIO, tableLocation, client, hiveClientPool);
+          fileIO, tableLocation, client, hiveClientPool, false);
 
       Table changeIcebergTable = tableMetaStore.doAs(() -> {
         try {
@@ -233,8 +237,7 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
       try {
         hiveClientPool.run(client -> {
           org.apache.hadoop.hive.metastore.api.Table hiveTable = newHiveTable(meta);
-          hiveTable.setSd(storageDescriptor(
-              hiveLocation,
+          hiveTable.setSd(HiveTableUtil.storageDescriptor(schema, partitionSpec, hiveLocation,
               FileFormat.valueOf(PropertyUtil.propertyAsString(properties, TableProperties.DEFAULT_FILE_FORMAT,
                   TableProperties.DEFAULT_FILE_FORMAT_DEFAULT).toUpperCase(Locale.ENGLISH))));
           setProToHive(hiveTable);
@@ -253,11 +256,13 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
       TableIdentifier tableIdentifier = TableIdentifier.of(meta.getTableIdentifier());
       String baseLocation = checkLocation(meta, MetaTableProperties.LOCATION_KEY_BASE);
       String tableLocation = checkLocation(meta, MetaTableProperties.LOCATION_KEY_TABLE);
-      String hiveLocation = tableLocation + "/hive";
+      String hiveLocation = HiveTableUtil.hiveRootLocation(tableLocation);
 
       meta.putToProperties(TableProperties.TABLE_CREATE_TIME, String.valueOf(System.currentTimeMillis()));
-      meta.putToProperties(TableProperties.BASE_HIVE_LOCATION_ROOT, hiveLocation);
+      meta.putToProperties(HiveTableProperties.BASE_HIVE_LOCATION_ROOT, hiveLocation);
       meta.putToProperties(org.apache.iceberg.TableProperties.FORMAT_VERSION, "2");
+      meta.putToProperties(org.apache.iceberg.TableProperties.DEFAULT_NAME_MAPPING,
+          NameMappingParser.toJson(MappingUtil.create(schema)));
 
       Table table = tableMetaStore.doAs(() -> {
         try {
@@ -269,8 +274,7 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
       try {
         hiveClientPool.run(client -> {
           org.apache.hadoop.hive.metastore.api.Table hiveTable = newHiveTable(meta);
-          hiveTable.setSd(storageDescriptor(
-              hiveLocation,
+          hiveTable.setSd(HiveTableUtil.storageDescriptor(schema, partitionSpec, hiveLocation,
               FileFormat.valueOf(PropertyUtil.propertyAsString(properties, TableProperties.BASE_FILE_FORMAT,
                   TableProperties.BASE_FILE_FORMAT_DEFAULT).toUpperCase(Locale.ENGLISH))));
           client.createTable(hiveTable);
@@ -305,25 +309,6 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
       return newTable;
     }
 
-    private StorageDescriptor storageDescriptor(String location, FileFormat format) {
-      final StorageDescriptor storageDescriptor = new StorageDescriptor();
-      storageDescriptor.setCols(HiveSchemaUtil.hiveTableFields(schema, partitionSpec));
-      storageDescriptor.setLocation(location);
-      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");
-          break;
-        default:
-          throw new IllegalArgumentException("Unsupported hive table file format:" + format);
-      }
-
-      storageDescriptor.setSerdeInfo(serDeInfo);
-      return storageDescriptor;
-    }
-
     @Override
     protected String getDatabaseLocation() {
       try {
diff --git a/hive/src/main/java/com/netease/arctic/hive/op/HiveSchemaUpdate.java b/hive/src/main/java/com/netease/arctic/hive/op/HiveSchemaUpdate.java
index cadb93cd3..8f42211a9 100644
--- a/hive/src/main/java/com/netease/arctic/hive/op/HiveSchemaUpdate.java
+++ b/hive/src/main/java/com/netease/arctic/hive/op/HiveSchemaUpdate.java
@@ -6,9 +6,9 @@
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
- *  *
+ *
  *     http://www.apache.org/licenses/LICENSE-2.0
- *  *
+ *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -23,31 +23,28 @@ import com.netease.arctic.hive.utils.HiveSchemaUtil;
 import com.netease.arctic.hive.utils.HiveTableUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.KeyedTable;
+import com.netease.arctic.table.TableProperties;
+import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.PartitionField;
 import org.apache.iceberg.Schema;
-import org.apache.iceberg.Table;
 import org.apache.iceberg.UpdateSchema;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.PropertyUtil;
 
 import java.util.Collection;
+import java.util.Locale;
 
 /**
  * Schema evolution API implementation for {@link KeyedTable}.
  */
 public class HiveSchemaUpdate implements UpdateSchema {
-  private ArcticTable arcticTable;
-  private Table baseTable;
-  private HMSClient hiveClient;
-  private UpdateSchema updateSchema;
+  private final ArcticTable arcticTable;
+  private final HMSClient hiveClient;
+  private final UpdateSchema updateSchema;
 
   public HiveSchemaUpdate(ArcticTable arcticTable, HMSClient hiveClient, UpdateSchema updateSchema) {
     this.arcticTable = arcticTable;
-    if (arcticTable.isKeyedTable()) {
-      this.baseTable = arcticTable.asKeyedTable().baseTable();
-    } else {
-      this.baseTable = arcticTable.asUnkeyedTable();
-    }
     this.hiveClient = hiveClient;
     this.updateSchema = updateSchema;
   }
@@ -63,7 +60,6 @@ public class HiveSchemaUpdate implements UpdateSchema {
     if (HiveTableUtil.loadHmsTable(hiveClient, arcticTable) == null) {
       throw new RuntimeException(String.format("there is no such hive table named %s", arcticTable.id().toString()));
     }
-    baseTable.refresh();
     syncSchemaToHive();
   }
 
@@ -72,7 +68,9 @@ public class HiveSchemaUpdate implements UpdateSchema {
     if (tbl == null) {
       throw new RuntimeException(String.format("there is no such hive table named %s", arcticTable.id().toString()));
     }
-    tbl.setSd(HiveSchemaUtil.storageDescriptor(baseTable.schema(), baseTable.spec()));
+    tbl.setSd(HiveTableUtil.storageDescriptor(arcticTable.schema(), arcticTable.spec(), tbl.getSd().getLocation(),
+        FileFormat.valueOf(PropertyUtil.propertyAsString(arcticTable.properties(), TableProperties.DEFAULT_FILE_FORMAT,
+            TableProperties.DEFAULT_FILE_FORMAT_DEFAULT).toUpperCase(Locale.ENGLISH))));
     HiveTableUtil.persistTable(hiveClient, tbl);
   }
 
@@ -85,13 +83,13 @@ public class HiveSchemaUpdate implements UpdateSchema {
   public UpdateSchema addColumn(String name, Type type, String doc) {
     this.updateSchema.addColumn(name, type, doc);
     //It is strictly required that all non-partitioned columns precede partitioned columns in the schema.
-    if (!baseTable.spec().isUnpartitioned()) {
+    if (!arcticTable.spec().isUnpartitioned()) {
       int parFieldMinIndex = Integer.MAX_VALUE;
       Types.NestedField firstParField = null;
-      for (PartitionField partitionField : baseTable.spec().fields()) {
-        Types.NestedField sourceField = baseTable.schema().findField(partitionField.sourceId());
-        if (baseTable.schema().columns().indexOf(sourceField) < parFieldMinIndex) {
-          parFieldMinIndex = baseTable.schema().columns().indexOf(sourceField);
+      for (PartitionField partitionField : arcticTable.spec().fields()) {
+        Types.NestedField sourceField = arcticTable.schema().findField(partitionField.sourceId());
+        if (arcticTable.schema().columns().indexOf(sourceField) < parFieldMinIndex) {
+          parFieldMinIndex = arcticTable.schema().columns().indexOf(sourceField);
           firstParField = sourceField;
         }
       }
diff --git a/hive/src/main/java/com/netease/arctic/hive/op/OverwriteHiveFiles.java b/hive/src/main/java/com/netease/arctic/hive/op/OverwriteHiveFiles.java
index 7ab45cd35..16bfefeaa 100644
--- a/hive/src/main/java/com/netease/arctic/hive/op/OverwriteHiveFiles.java
+++ b/hive/src/main/java/com/netease/arctic/hive/op/OverwriteHiveFiles.java
@@ -1,11 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package com.netease.arctic.hive.op;
 
 import com.netease.arctic.hive.HMSClient;
+import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.exceptions.CannotAlterHiveLocationException;
 import com.netease.arctic.hive.table.UnkeyedHiveTable;
 import com.netease.arctic.hive.utils.HivePartitionUtil;
+import com.netease.arctic.hive.utils.HiveTableUtil;
 import com.netease.arctic.op.UpdatePartitionProperties;
-import com.netease.arctic.table.TableProperties;
 import com.netease.arctic.utils.FileUtil;
 import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.hadoop.fs.FileStatus;
@@ -39,10 +58,14 @@ import java.util.Set;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
+import static com.netease.arctic.op.OverwriteBaseFiles.PROPERTIES_TRANSACTION_ID;
+
 public class OverwriteHiveFiles implements OverwriteFiles {
 
   private static final Logger LOG = LoggerFactory.getLogger(OverwriteHiveFiles.class);
 
+  public static final String PROPERTIES_VALIDATE_LOCATION = "validate-location";
+
   private final Transaction transaction;
   private final boolean insideTransaction;
   private final OverwriteFiles delegate;
@@ -60,8 +83,11 @@ public class OverwriteHiveFiles implements OverwriteFiles {
 
   private Map<StructLike, Partition> partitionToDelete = Maps.newHashMap();
   private Map<StructLike, Partition> partitionToCreate = Maps.newHashMap();
+  private final Map<StructLike, Partition> partitionToAlter = Maps.newHashMap();
   private String unpartitionTableLocation;
   private long txId = -1;
+  private boolean validateLocation = true;
+  private int commitTimestamp; // in seconds
 
   public OverwriteHiveFiles(Transaction transaction, boolean insideTransaction, UnkeyedHiveTable table,
       HMSClient hmsClient, HMSClient transactionClient) {
@@ -144,10 +170,14 @@ public class OverwriteHiveFiles implements OverwriteFiles {
 
   @Override
   public OverwriteFiles set(String property, String value) {
-    if ("txId".equals(property)) {
+    if (PROPERTIES_TRANSACTION_ID.equals(property)) {
       this.txId = Long.parseLong(value);
     }
 
+    if (PROPERTIES_VALIDATE_LOCATION.equals(property)) {
+      this.validateLocation = Boolean.parseBoolean(value);
+    }
+
     delegate.set(property, value);
     return this;
   }
@@ -171,6 +201,7 @@ public class OverwriteHiveFiles implements OverwriteFiles {
 
   @Override
   public void commit() {
+    commitTimestamp = (int) (System.currentTimeMillis() / 1000);
     if (table.spec().isUnpartitioned()) {
       generateUnpartitionTableLocation();
     } else {
@@ -179,7 +210,7 @@ public class OverwriteHiveFiles implements OverwriteFiles {
     }
 
     delegate.commit();
-    setHiveLocations();
+    commitPartitionProperties();
     if (!insideTransaction) {
       transaction.commitTransaction();
     }
@@ -191,20 +222,31 @@ public class OverwriteHiveFiles implements OverwriteFiles {
     }
   }
 
-  private void setHiveLocations() {
+  private void commitPartitionProperties() {
     UpdatePartitionProperties updatePartitionProperties = table.updatePartitionProperties(transaction);
     if (table.spec().isUnpartitioned()) {
       updatePartitionProperties.set(TablePropertyUtil.EMPTY_STRUCT,
-          TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION, unpartitionTableLocation);
+          HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION, unpartitionTableLocation);
+      updatePartitionProperties.set(TablePropertyUtil.EMPTY_STRUCT,
+          HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME, commitTimestamp + "");
     } else {
       partitionToDelete.forEach((partitionData, partition) -> {
         if (!partitionToCreate.containsKey(partitionData)) {
-          updatePartitionProperties.remove(partitionData, TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION);
+          updatePartitionProperties.remove(partitionData, HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION);
+          updatePartitionProperties.remove(partitionData, HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME);
         }
       });
       partitionToCreate.forEach((partitionData, partition) -> {
-        updatePartitionProperties.set(partitionData, TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION,
+        updatePartitionProperties.set(partitionData, HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION,
+            partition.getSd().getLocation());
+        updatePartitionProperties.set(partitionData, HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME,
+            commitTimestamp + "");
+      });
+      partitionToAlter.forEach((partitionData, partition) -> {
+        updatePartitionProperties.set(partitionData, HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION,
             partition.getSd().getLocation());
+        updatePartitionProperties.set(partitionData, HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME,
+            commitTimestamp + "");
       });
     }
     updatePartitionProperties.commit();
@@ -244,7 +286,7 @@ public class OverwriteHiveFiles implements OverwriteFiles {
       List<DataFile> dataFiles = partitionDataFileMap.get(val);
 
       checkCreatePartitionDataFiles(dataFiles, location);
-      Partition p = HivePartitionUtil.newPartition(hiveTable, values, location, dataFiles);
+      Partition p = HivePartitionUtil.newPartition(hiveTable, values, location, dataFiles, commitTimestamp);
       createPartitions.put(dataFiles.get(0).partition(), p);
     }
 
@@ -286,7 +328,9 @@ public class OverwriteHiveFiles implements OverwriteFiles {
       checkedPartitionValues.add(pathValue);
     }
 
-    deletePartitions.values().forEach(p -> checkPartitionDelete(deleteFileLocations, p));
+    if (validateLocation) {
+      deletePartitions.values().forEach(p -> checkPartitionDelete(deleteFileLocations, p));
+    }
     return deletePartitions;
   }
 
@@ -346,7 +390,7 @@ public class OverwriteHiveFiles implements OverwriteFiles {
         String deleteLocation = toDelete.getSd().getLocation();
         // if exists partition to delete with same value
         // make sure location is different
-        if (isPathEquals(location, deleteLocation)) {
+        if (isPathEquals(location, deleteLocation) && validateLocation) {
           throw new CannotAlterHiveLocationException("can't create new partition: " +
               partitionToString(entry.getValue()) + ", this " +
               "partition will be " +
@@ -361,7 +405,7 @@ public class OverwriteHiveFiles implements OverwriteFiles {
         Partition partitionInHive = hmsClient.run(c -> c.getPartition(db, tableName, entry.getValue().getValues()));
         String locationInHive = partitionInHive.getSd().getLocation();
         if (isPathEquals(location, locationInHive)) {
-          // exists same location, skip create operation
+          partitionToAlter.put(entry.getKey(), entry.getValue());
           continue;
         }
         throw new CannotAlterHiveLocationException("can't create new partition: " +
@@ -399,7 +443,18 @@ public class OverwriteHiveFiles implements OverwriteFiles {
 
     if (!partitionToCreate.isEmpty()) {
       try {
-        transactionClient.run(c -> c.add_partitions(partitionToCreate.values().stream().collect(Collectors.toList())));
+        transactionClient.run(c -> c.add_partitions(Lists.newArrayList(partitionToCreate.values())));
+      } catch (TException | InterruptedException e) {
+        throw new RuntimeException(e);
+      }
+    }
+
+    if (!partitionToAlter.isEmpty()) {
+      try {
+        transactionClient.run(c ->  {
+          c.alter_partitions(db, tableName, Lists.newArrayList(partitionToAlter.values()), null);
+          return null;
+        });
       } catch (TException | InterruptedException e) {
         throw new RuntimeException(e);
       }
@@ -421,6 +476,8 @@ public class OverwriteHiveFiles implements OverwriteFiles {
       transactionClient.run(c -> {
         Table hiveTable = c.getTable(db, tableName);
         hiveTable.getSd().setLocation(finalLocation);
+        HiveTableUtil.generateTableProperties(commitTimestamp, addFiles)
+            .forEach((key, value) -> hiveTable.getParameters().put(key, value));
         c.alter_table(db, tableName, hiveTable);
         return 0;
       });
@@ -431,7 +488,7 @@ public class OverwriteHiveFiles implements OverwriteFiles {
 
   private String createEmptyLocationForHive() {
     // create a new empty location for hive
-    String newLocation = null;
+    String newLocation;
     if (txId > 0) {
       newLocation = table.hiveLocation() + "/txId=" + txId;
     } else {
diff --git a/hive/src/main/java/com/netease/arctic/hive/op/ReplaceHivePartitions.java b/hive/src/main/java/com/netease/arctic/hive/op/ReplaceHivePartitions.java
index da0b49a05..fa1de9374 100644
--- a/hive/src/main/java/com/netease/arctic/hive/op/ReplaceHivePartitions.java
+++ b/hive/src/main/java/com/netease/arctic/hive/op/ReplaceHivePartitions.java
@@ -1,18 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package com.netease.arctic.hive.op;
 
 import com.netease.arctic.hive.HMSClient;
+import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.exceptions.CannotAlterHiveLocationException;
 import com.netease.arctic.hive.table.UnkeyedHiveTable;
 import com.netease.arctic.hive.utils.HivePartitionUtil;
+import com.netease.arctic.hive.utils.HiveTableUtil;
 import com.netease.arctic.op.UpdatePartitionProperties;
-import com.netease.arctic.table.TableProperties;
 import com.netease.arctic.utils.FileUtil;
 import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
 import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.Table;
-import org.apache.iceberg.ContentFile;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.ReplacePartitions;
 import org.apache.iceberg.Snapshot;
@@ -27,7 +45,6 @@ import org.apache.thrift.TException;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
-import java.util.stream.Collectors;
 
 public class ReplaceHivePartitions implements ReplacePartitions {
 
@@ -47,6 +64,7 @@ public class ReplaceHivePartitions implements ReplacePartitions {
   private final Map<StructLike, Partition> rewritePartitions = Maps.newHashMap();
   private final Map<StructLike, Partition> newPartitions = Maps.newHashMap();
   private String unpartitionTableLocation;
+  private int commitTimestamp; // in seconds
 
   public ReplaceHivePartitions(
       Transaction transaction,
@@ -113,6 +131,7 @@ public class ReplaceHivePartitions implements ReplacePartitions {
   @Override
   public void commit() {
     if (!addFiles.isEmpty()) {
+      commitTimestamp = (int) (System.currentTimeMillis() / 1000);
       if (table.spec().isUnpartitioned()) {
         generateUnpartitionTableLocation();
       } else {
@@ -120,7 +139,7 @@ public class ReplaceHivePartitions implements ReplacePartitions {
       }
 
       delegate.commit();
-      setHiveLocations();
+      commitPartitionProperties();
       if (!insideTransaction) {
         transaction.commitTransaction();
       }
@@ -133,20 +152,25 @@ public class ReplaceHivePartitions implements ReplacePartitions {
     }
   }
 
-  private void setHiveLocations() {
+  private void commitPartitionProperties() {
     UpdatePartitionProperties updatePartitionProperties = table.updatePartitionProperties(transaction);
     if (table.spec().isUnpartitioned() && unpartitionTableLocation != null) {
-      updatePartitionProperties.set(
-          TablePropertyUtil.EMPTY_STRUCT,
-          TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION, unpartitionTableLocation);
+      updatePartitionProperties.set(TablePropertyUtil.EMPTY_STRUCT,
+          HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION, unpartitionTableLocation);
+      updatePartitionProperties.set(TablePropertyUtil.EMPTY_STRUCT,
+          HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME, commitTimestamp + "");
     } else {
       rewritePartitions.forEach((partitionData, partition) -> {
-        updatePartitionProperties.set(partitionData, TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION,
+        updatePartitionProperties.set(partitionData, HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION,
             partition.getSd().getLocation());
+        updatePartitionProperties.set(partitionData, HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME,
+            commitTimestamp + "");
       });
       newPartitions.forEach((partitionData, partition) -> {
-        updatePartitionProperties.set(partitionData, TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION,
+        updatePartitionProperties.set(partitionData, HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION,
             partition.getSd().getLocation());
+        updatePartitionProperties.set(partitionData, HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME,
+            commitTimestamp + "");
       });
     }
     updatePartitionProperties.commit();
@@ -185,10 +209,10 @@ public class ReplaceHivePartitions implements ReplacePartitions {
 
       try {
         Partition partition = hmsClient.run(c -> c.getPartition(db, tableName, values));
-        rewriteHivePartitions(partition, location, dataFiles);
+        HivePartitionUtil.rewriteHivePartitions(partition, location, dataFiles, commitTimestamp);
         rewritePartitions.put(dataFiles.get(0).partition(), partition);
       } catch (NoSuchObjectException e) {
-        Partition p = HivePartitionUtil.newPartition(hiveTable, values, location, dataFiles);
+        Partition p = HivePartitionUtil.newPartition(hiveTable, values, location, dataFiles, commitTimestamp);
         newPartitions.put(dataFiles.get(0).partition(), p);
       } catch (TException | InterruptedException e) {
         throw new RuntimeException(e);
@@ -203,6 +227,8 @@ public class ReplaceHivePartitions implements ReplacePartitions {
         transactionalHMSClient.run(c -> {
           Table tbl = c.getTable(db, tableName);
           tbl.getSd().setLocation(newDataLocation);
+          HiveTableUtil.generateTableProperties(commitTimestamp, addFiles)
+              .forEach((key, value) -> hiveTable.getParameters().put(key, value));
           c.alter_table(db, tableName, tbl);
           return 0;
         });
@@ -216,10 +242,10 @@ public class ReplaceHivePartitions implements ReplacePartitions {
     try {
       transactionalHMSClient.run(c -> {
         if (!rewritePartitions.isEmpty()) {
-          c.alter_partitions(db, tableName, rewritePartitions.values().stream().collect(Collectors.toList()), null);
+          c.alter_partitions(db, tableName, Lists.newArrayList(rewritePartitions.values()), null);
         }
         if (!newPartitions.isEmpty()) {
-          c.add_partitions(newPartitions.values().stream().collect(Collectors.toList()));
+          c.add_partitions(Lists.newArrayList(newPartitions.values()));
         }
         return 0;
       });
@@ -245,15 +271,4 @@ public class ReplaceHivePartitions implements ReplacePartitions {
   private void generateUnpartitionTableLocation() {
     unpartitionTableLocation = FileUtil.getFileDir(this.addFiles.get(0).path().toString());
   }
-
-  private static void rewriteHivePartitions(Partition partition, String location, List<DataFile> dataFiles) {
-    partition.getSd().setLocation(location);
-    int lastAccessTime = (int) (System.currentTimeMillis() / 1000);
-    partition.setLastAccessTime(lastAccessTime);
-    int files = dataFiles.size();
-    long totalSize = dataFiles.stream().map(ContentFile::fileSizeInBytes).reduce(0L, Long::sum);
-    partition.putToParameters("transient_lastDdlTime", lastAccessTime + "");
-    partition.putToParameters("totalSize", totalSize + "");
-    partition.putToParameters("numFiles", files + "");
-  }
 }
diff --git a/hive/src/main/java/com/netease/arctic/hive/table/KeyedHiveTable.java b/hive/src/main/java/com/netease/arctic/hive/table/KeyedHiveTable.java
index f5f0ad03a..80d8bf5d7 100644
--- a/hive/src/main/java/com/netease/arctic/hive/table/KeyedHiveTable.java
+++ b/hive/src/main/java/com/netease/arctic/hive/table/KeyedHiveTable.java
@@ -21,18 +21,19 @@ package com.netease.arctic.hive.table;
 import com.netease.arctic.AmsClient;
 import com.netease.arctic.ams.api.TableMeta;
 import com.netease.arctic.hive.HMSClient;
-import com.netease.arctic.hive.op.HiveSchemaUpdate;
+import com.netease.arctic.hive.HiveTableProperties;
+import com.netease.arctic.hive.utils.HiveMetaSynchronizer;
 import com.netease.arctic.table.BaseKeyedTable;
 import com.netease.arctic.table.ChangeTable;
 import com.netease.arctic.table.PrimaryKeySpec;
-import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.util.PropertyUtil;
 
 /**
  * Implementation of {@link com.netease.arctic.table.KeyedTable} with Hive table as base store.
  */
 public class KeyedHiveTable extends BaseKeyedTable implements SupportHive {
 
-  private HMSClient hiveClient;
+  private final HMSClient hiveClient;
 
   public KeyedHiveTable(
       TableMeta tableMeta,
@@ -44,10 +45,38 @@ public class KeyedHiveTable extends BaseKeyedTable implements SupportHive {
       ChangeTable changeTable) {
     super(tableMeta, tableLocation, primaryKeySpec, client, baseTable, changeTable);
     this.hiveClient = hiveClient;
+    syncHiveSchemaToArctic();
+    syncHiveDataToArctic();
   }
 
+  @Override
+  public void refresh() {
+    super.refresh();
+    syncHiveSchemaToArctic();
+    syncHiveDataToArctic();
+  }
+
+
   @Override
   public String hiveLocation() {
     return ((SupportHive)baseTable()).hiveLocation();
   }
+
+  private void syncHiveSchemaToArctic() {
+    if (PropertyUtil.propertyAsBoolean(
+        properties(),
+        HiveTableProperties.AUTO_SYNC_HIVE_SCHEMA_CHANGE,
+        HiveTableProperties.AUTO_SYNC_HIVE_SCHEMA_CHANGE_DEFAULT)) {
+      HiveMetaSynchronizer.syncHiveSchemaToArctic(this, hiveClient);
+    }
+  }
+
+  private void syncHiveDataToArctic() {
+    if (PropertyUtil.propertyAsBoolean(
+        properties(),
+        HiveTableProperties.AUTO_SYNC_HIVE_DATA_WRITE,
+        HiveTableProperties.AUTO_SYNC_HIVE_DATA_WRITE_DEFAULT)) {
+      HiveMetaSynchronizer.syncHiveDataToArctic(this, hiveClient);
+    }
+  }
 }
diff --git a/hive/src/main/java/com/netease/arctic/hive/table/UnkeyedHiveTable.java b/hive/src/main/java/com/netease/arctic/hive/table/UnkeyedHiveTable.java
index 22dd8b7d9..e79c18a92 100644
--- a/hive/src/main/java/com/netease/arctic/hive/table/UnkeyedHiveTable.java
+++ b/hive/src/main/java/com/netease/arctic/hive/table/UnkeyedHiveTable.java
@@ -20,29 +20,35 @@ package com.netease.arctic.hive.table;
 
 import com.netease.arctic.AmsClient;
 import com.netease.arctic.hive.HMSClient;
+import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.op.HiveOperationTransaction;
 import com.netease.arctic.hive.op.HiveSchemaUpdate;
 import com.netease.arctic.hive.op.OverwriteHiveFiles;
 import com.netease.arctic.hive.op.ReplaceHivePartitions;
+import com.netease.arctic.hive.utils.HiveMetaSynchronizer;
+import com.netease.arctic.hive.utils.HiveTableUtil;
 import com.netease.arctic.io.ArcticFileIO;
 import com.netease.arctic.table.BaseTable;
 import com.netease.arctic.table.BaseUnkeyedTable;
 import com.netease.arctic.table.TableIdentifier;
-import org.apache.iceberg.OverwriteFiles;
 import org.apache.iceberg.ReplacePartitions;
+import org.apache.iceberg.Schema;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.util.PropertyUtil;
 
-import static com.netease.arctic.table.TableProperties.BASE_HIVE_LOCATION_ROOT;
+import static com.netease.arctic.hive.HiveTableProperties.BASE_HIVE_LOCATION_ROOT;
 
 /**
  * Implementation of {@link com.netease.arctic.table.UnkeyedTable} with Hive table as base store.
  */
 public class UnkeyedHiveTable extends BaseUnkeyedTable implements BaseTable, SupportHive {
 
-  HMSClient hiveClient;
-  String tableLocation;
+  private final HMSClient hiveClient;
+  private final String tableLocation;
+
+  private boolean syncHiveChange = true;
 
   public UnkeyedHiveTable(
       TableIdentifier tableIdentifier,
@@ -51,14 +57,40 @@ public class UnkeyedHiveTable extends BaseUnkeyedTable implements BaseTable, Sup
       String tableLocation,
       AmsClient client,
       HMSClient hiveClient) {
+    this(tableIdentifier, icebergTable, arcticFileIO, tableLocation, client, hiveClient, true);
+  }
+
+  public UnkeyedHiveTable(
+      TableIdentifier tableIdentifier,
+      Table icebergTable,
+      ArcticFileIO arcticFileIO,
+      String tableLocation,
+      AmsClient client,
+      HMSClient hiveClient,
+      boolean syncHiveChange) {
     super(tableIdentifier, icebergTable, arcticFileIO, client);
     this.hiveClient = hiveClient;
     this.tableLocation = tableLocation;
+    this.syncHiveChange = syncHiveChange;
+    syncHiveSchemaToArctic();
+    syncHiveDataToArctic();
+  }
+
+  @Override
+  public void refresh() {
+    super.refresh();
+    syncHiveSchemaToArctic();
+    syncHiveDataToArctic();
+  }
+
+  @Override
+  public Schema schema() {
+    return super.schema();
   }
 
   @Override
   public ReplacePartitions newReplacePartitions() {
-    return new ReplaceHivePartitions(icebergTable.newTransaction(),
+    return new ReplaceHivePartitions(super.newTransaction(),
         false, this, hiveClient, hiveClient);
   }
 
@@ -71,12 +103,12 @@ public class UnkeyedHiveTable extends BaseUnkeyedTable implements BaseTable, Sup
   public String hiveLocation() {
     return properties().containsKey(BASE_HIVE_LOCATION_ROOT) ?
         properties().get(BASE_HIVE_LOCATION_ROOT) :
-        tableLocation + "/hive";
+        HiveTableUtil.hiveRootLocation(tableLocation);
   }
 
   @Override
   public OverwriteHiveFiles newOverwrite() {
-    return new OverwriteHiveFiles(super.newTransaction(), false,this, hiveClient, hiveClient);
+    return new OverwriteHiveFiles(super.newTransaction(), false, this, hiveClient, hiveClient);
   }
 
   @Override
@@ -89,4 +121,22 @@ public class UnkeyedHiveTable extends BaseUnkeyedTable implements BaseTable, Sup
   public UpdateSchema updateSchema() {
     return new HiveSchemaUpdate(this, hiveClient, super.updateSchema());
   }
+
+  private void syncHiveSchemaToArctic() {
+    if (syncHiveChange && PropertyUtil.propertyAsBoolean(
+        properties(),
+        HiveTableProperties.AUTO_SYNC_HIVE_SCHEMA_CHANGE,
+        HiveTableProperties.AUTO_SYNC_HIVE_SCHEMA_CHANGE_DEFAULT)) {
+      HiveMetaSynchronizer.syncHiveSchemaToArctic(this, hiveClient);
+    }
+  }
+
+  private void syncHiveDataToArctic() {
+    if (syncHiveChange && PropertyUtil.propertyAsBoolean(
+        properties(),
+        HiveTableProperties.AUTO_SYNC_HIVE_DATA_WRITE,
+        HiveTableProperties.AUTO_SYNC_HIVE_DATA_WRITE_DEFAULT)) {
+      HiveMetaSynchronizer.syncHiveDataToArctic(this, hiveClient);
+    }
+  }
 }
diff --git a/hive/src/main/java/com/netease/arctic/hive/utils/HiveMetaSynchronizer.java b/hive/src/main/java/com/netease/arctic/hive/utils/HiveMetaSynchronizer.java
new file mode 100644
index 000000000..018d6a46d
--- /dev/null
+++ b/hive/src/main/java/com/netease/arctic/hive/utils/HiveMetaSynchronizer.java
@@ -0,0 +1,221 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.hive.utils;
+
+import com.netease.arctic.hive.HMSClient;
+import com.netease.arctic.hive.HiveTableProperties;
+import com.netease.arctic.hive.op.OverwriteHiveFiles;
+import com.netease.arctic.io.ArcticHadoopFileIO;
+import com.netease.arctic.op.OverwriteBaseFiles;
+import com.netease.arctic.table.ArcticTable;
+import com.netease.arctic.table.TableProperties;
+import com.netease.arctic.table.UnkeyedTable;
+import com.netease.arctic.utils.TablePropertyUtil;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.data.TableMigrationUtil;
+import org.apache.iceberg.mapping.NameMappingParser;
+import org.apache.iceberg.relocated.com.google.common.collect.ListMultimap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Multimaps;
+import org.apache.iceberg.types.Types;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.xml.crypto.Data;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * Synchronize the metadata of the hive table to arctic table
+ */
+public class HiveMetaSynchronizer {
+
+  private static final Logger LOG = LoggerFactory.getLogger(HiveMetaSynchronizer.class);
+
+  /**
+   * Synchronize the schema change of the hive table to arctic table
+   * @param table arctic table to accept the schema change
+   * @param hiveClient hive client
+   */
+  public static void syncHiveSchemaToArctic(ArcticTable table, HMSClient hiveClient) {
+    try {
+      Table hiveTable = hiveClient.run(client -> client.getTable(table.id().getDatabase(), table.id().getTableName()));
+      List<FieldSchema> fieldSchemas =  hiveTable.getSd().getCols();
+      Schema hiveSchema = org.apache.iceberg.hive.HiveSchemaUtil.convert(fieldSchemas);
+      UpdateSchema updateSchema = table.updateSchema();
+      boolean update = false;
+      for (int i = 0; i < hiveSchema.columns().size(); i++) {
+        Types.NestedField hiveField = hiveSchema.columns().get(i);
+        Types.NestedField icebergField = table.schema().findField(hiveField.name());
+        if (icebergField == null) {
+          updateSchema.addColumn(hiveField.name(), hiveField.type(), hiveField.doc());
+          update = true;
+          LOG.info("Table {} sync new hive column {} to arctic", table.id(), hiveField);
+        } else if (!icebergField.type().equals(hiveField.type()) ||
+            !Objects.equals(icebergField.doc(), (hiveField.doc()))) {
+          updateSchema.updateColumn(icebergField.name(), icebergField.type().asPrimitiveType(), hiveField.doc());
+          update = true;
+          LOG.info("Table {} sync updated hive column {} to arctic", table.id(), icebergField);
+        }
+      }
+      if (update) {
+        updateSchema.commit();
+      }
+    } catch (TException | InterruptedException e) {
+      throw new RuntimeException("Failed to get hive table:" + table.id(), e);
+    }
+  }
+
+  /**
+   * Synchronize the data change of the hive table to arctic table
+   * @param table arctic table to accept the data change
+   * @param hiveClient hive client
+   */
+  public static void syncHiveDataToArctic(ArcticTable table, HMSClient hiveClient) {
+    UnkeyedTable baseStore;
+    if (table.isKeyedTable()) {
+      baseStore = table.asKeyedTable().baseTable();
+    } else {
+      baseStore = table.asUnkeyedTable();
+    }
+    try {
+      if (table.spec().isUnpartitioned()) {
+        Table hiveTable =
+            hiveClient.run(client -> client.getTable(table.id().getDatabase(), table.id().getTableName()));
+        String hiveTransientTime =  hiveTable.getParameters().get("transient_lastDdlTime");
+        String arcticTransientTime = baseStore.partitionProperty().get(TablePropertyUtil.EMPTY_STRUCT)
+            .get(HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME);
+        if (arcticTransientTime == null || !arcticTransientTime.equals(hiveTransientTime)) {
+          List<DataFile> hiveDataFiles = TableMigrationUtil.listPartition(Maps.newHashMap(),
+              hiveTable.getSd().getLocation(),
+              table.properties().getOrDefault(TableProperties.DEFAULT_FILE_FORMAT,
+                  TableProperties.DEFAULT_FILE_FORMAT_DEFAULT),
+              table.spec(), ((ArcticHadoopFileIO)table.io()).getTableMetaStore().getConfiguration(),
+              MetricsConfig.fromProperties(table.properties()),
+              NameMappingParser.fromJson(
+                  table.properties().get(org.apache.iceberg.TableProperties.DEFAULT_NAME_MAPPING)));
+          List<DataFile> deleteFiles = Lists.newArrayList();
+          baseStore.newScan().planFiles().forEach(fileScanTask -> deleteFiles.add(fileScanTask.file()));
+          overwriteTable(table, deleteFiles, hiveDataFiles);
+        }
+      } else {
+        // list all hive partitions.
+        List<Partition> hivePartitions = hiveClient.run(client -> client.listPartitions(table.id().getDatabase(),
+            table.id().getTableName(), Short.MAX_VALUE));
+        // group arctic files by partition.
+        ListMultimap<StructLike, DataFile> filesGroupedByPartition
+            = Multimaps.newListMultimap(Maps.newHashMap(), Lists::newArrayList);
+        TableScan tableScan = baseStore.newScan();
+        for (org.apache.iceberg.FileScanTask fileScanTask : tableScan.planFiles()) {
+          filesGroupedByPartition.put(fileScanTask.file().partition(),fileScanTask.file());
+        }
+        Map<StructLike, Collection<DataFile>> filesMap = filesGroupedByPartition.asMap();
+        List<DataFile> filesToDelete = Lists.newArrayList();
+        List<DataFile> filesToAdd = Lists.newArrayList();
+        List<StructLike> icebergPartitions = Lists.newArrayList(filesMap.keySet());
+        for (Partition hivePartition : hivePartitions) {
+          StructLike partitionData = HivePartitionUtil.buildPartitionData(hivePartition.getValues(), table.spec());
+          icebergPartitions.remove(partitionData);
+          String hiveTransientTime =  hivePartition.getParameters().get("transient_lastDdlTime");
+          String arcticTransientTime = baseStore.partitionProperty().containsKey(partitionData) ?
+              baseStore.partitionProperty().get(partitionData)
+                  .get(HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME) : null;
+          // compare hive partition parameter transient_lastDdlTime with arctic partition properties to
+          // find out if the partition is changed.
+          if (arcticTransientTime == null || !arcticTransientTime.equals(hiveTransientTime)) {
+            List<DataFile> hiveDataFiles = TableMigrationUtil.listPartition(
+                buildPartitionValueMap(hivePartition.getValues(), table.spec()),
+                hivePartition.getSd().getLocation(),
+                table.properties().getOrDefault(TableProperties.DEFAULT_FILE_FORMAT,
+                    TableProperties.DEFAULT_FILE_FORMAT_DEFAULT),
+                table.spec(), ((ArcticHadoopFileIO)table.io()).getTableMetaStore().getConfiguration(),
+                MetricsConfig.fromProperties(table.properties()), NameMappingParser.fromJson(
+                    table.properties().get(org.apache.iceberg.TableProperties.DEFAULT_NAME_MAPPING)));
+            if (filesMap.get(partitionData) != null) {
+              filesToDelete.addAll(filesMap.get(partitionData));
+              filesToAdd.addAll(hiveDataFiles);
+              // make sure new partition is not created by arctic
+            } else if (hivePartition.getParameters().get(HiveTableProperties.ARCTIC_TABLE_FLAG) == null) {
+              filesToAdd.addAll(hiveDataFiles);
+            }
+          }
+        }
+
+        icebergPartitions.forEach(partition -> {
+          List<DataFile> dataFiles = Lists.newArrayList(filesMap.get(partition));
+          if (dataFiles.size() > 0) {
+            // make sure dropped partition with no files
+            if (!table.io().exists(dataFiles.get(0).path().toString())) {
+              filesToDelete.addAll(filesMap.get(partition));
+            }
+          }
+        });
+        overwriteTable(table, filesToDelete, filesToAdd);
+      }
+    } catch (TException | InterruptedException e) {
+      throw new RuntimeException("Failed to get hive table:" + table.id(), e);
+    }
+  }
+
+  private static Map<String, String> buildPartitionValueMap(List<String> partitionValues, PartitionSpec spec) {
+    Map<String, String> partitionValueMap = Maps.newHashMap();
+    for (int i = 0; i < partitionValues.size(); i++) {
+      partitionValueMap.put(spec.fields().get(i).name(), partitionValues.get(i));
+    }
+    return partitionValueMap;
+  }
+
+  private static void overwriteTable(ArcticTable table, List<DataFile> filesToDelete, List<DataFile> filesToAdd) {
+    if (filesToDelete.size() > 0 || filesToAdd.size() > 0) {
+      LOG.info("Table {} sync hive data change to arctic, delete files: {}, add files {}", table.id(),
+          filesToDelete.stream().map(DataFile::path).collect(Collectors.toList()),
+          filesToAdd.stream().map(DataFile::path).collect(Collectors.toList()));
+      if (table.isKeyedTable()) {
+        long txId = table.asKeyedTable().beginTransaction(null);
+        OverwriteBaseFiles overwriteBaseFiles = table.asKeyedTable().newOverwriteBaseFiles();
+        overwriteBaseFiles.set(OverwriteHiveFiles.PROPERTIES_VALIDATE_LOCATION, "false");
+        filesToDelete.forEach(overwriteBaseFiles::deleteFile);
+        filesToAdd.forEach(overwriteBaseFiles::addFile);
+        overwriteBaseFiles.withTransactionId(txId);
+        overwriteBaseFiles.commit();
+      } else {
+        OverwriteFiles overwriteFiles = table.asUnkeyedTable().newOverwrite();
+        overwriteFiles.set(OverwriteHiveFiles.PROPERTIES_VALIDATE_LOCATION, "false");
+        filesToDelete.forEach(overwriteFiles::deleteFile);
+        filesToAdd.forEach(overwriteFiles::addFile);
+        overwriteFiles.commit();
+      }
+    }
+  }
+}
diff --git a/hive/src/main/java/com/netease/arctic/hive/utils/HivePartitionUtil.java b/hive/src/main/java/com/netease/arctic/hive/utils/HivePartitionUtil.java
index 84fa2eac6..391745812 100644
--- a/hive/src/main/java/com/netease/arctic/hive/utils/HivePartitionUtil.java
+++ b/hive/src/main/java/com/netease/arctic/hive/utils/HivePartitionUtil.java
@@ -22,8 +22,9 @@ import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.PrincipalPrivilegeSet;
 import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
 import org.apache.hadoop.hive.metastore.api.Table;
-import org.apache.iceberg.ContentFile;
 import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.types.Type;
@@ -44,32 +45,51 @@ public class HivePartitionUtil {
     return values;
   }
 
+  public static StructLike buildPartitionData(List<String> partitionValues, PartitionSpec spec) {
+    StringBuilder pathBuilder = new StringBuilder();
+    for (int i = 0; i < spec.partitionType().fields().size(); i++) {
+      Types.NestedField field = spec.partitionType().fields().get(i);
+      pathBuilder.append(field.name()).append("=").append(partitionValues.get(i));
+      if (i > 0) {
+        pathBuilder.append("/");
+      }
+    }
+    return DataFiles.data(spec, pathBuilder.toString());
+  }
+
   public static Partition newPartition(
       Table hiveTable,
       List<String> values,
       String location,
-      List<DataFile> dataFiles) {
+      List<DataFile> dataFiles,
+      int createTimeInSeconds) {
     StorageDescriptor tableSd = hiveTable.getSd();
     PrincipalPrivilegeSet privilegeSet = hiveTable.getPrivileges();
-    int lastAccessTime = (int) (System.currentTimeMillis() / 1000);
     Partition p = new Partition();
     p.setValues(values);
     p.setDbName(hiveTable.getDbName());
     p.setTableName(hiveTable.getTableName());
-    p.setCreateTime(lastAccessTime);
-    p.setLastAccessTime(lastAccessTime);
+    p.setCreateTime(createTimeInSeconds);
+    p.setLastAccessTime(createTimeInSeconds);
     StorageDescriptor sd = tableSd.deepCopy();
     sd.setLocation(location);
     p.setSd(sd);
 
-    int files = dataFiles.size();
-    long totalSize = dataFiles.stream().map(ContentFile::fileSizeInBytes).reduce(0L, Long::sum);
-    p.putToParameters("transient_lastDdlTime", lastAccessTime + "");
-    p.putToParameters("totalSize", totalSize + "");
-    p.putToParameters("numFiles", files + "");
+    HiveTableUtil.generateTableProperties(createTimeInSeconds, dataFiles)
+        .forEach((key, value) -> p.putToParameters(key, value));
+
     if (privilegeSet != null) {
       p.setPrivileges(privilegeSet.deepCopy());
     }
     return p;
   }
+
+  public static void rewriteHivePartitions(Partition partition, String location, List<DataFile> dataFiles,
+      int accessTimestamp) {
+    partition.getSd().setLocation(location);
+    partition.setLastAccessTime(accessTimestamp);
+    HiveTableUtil.generateTableProperties(accessTimestamp, dataFiles)
+        .forEach(partition::putToParameters);
+  }
+
 }
diff --git a/hive/src/main/java/com/netease/arctic/hive/utils/HiveSchemaUtil.java b/hive/src/main/java/com/netease/arctic/hive/utils/HiveSchemaUtil.java
index 51798ba40..3eb6758b9 100644
--- a/hive/src/main/java/com/netease/arctic/hive/utils/HiveSchemaUtil.java
+++ b/hive/src/main/java/com/netease/arctic/hive/utils/HiveSchemaUtil.java
@@ -21,6 +21,7 @@ package com.netease.arctic.hive.utils;
 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.iceberg.FileFormat;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.types.TypeUtil;
@@ -70,12 +71,4 @@ public class HiveSchemaUtil {
   public static List<FieldSchema> hivePartitionFields(Schema schema, PartitionSpec spec) {
     return org.apache.iceberg.hive.HiveSchemaUtil.convert(TypeUtil.select(schema, spec.identitySourceIds()));
   }
-
-  public static StorageDescriptor storageDescriptor(Schema schema, PartitionSpec partitionSpec) {
-    final StorageDescriptor storageDescriptor = new StorageDescriptor();
-    storageDescriptor.setCols(HiveSchemaUtil.hiveTableFields(schema, partitionSpec));
-    SerDeInfo serDeInfo = new SerDeInfo();
-    storageDescriptor.setSerdeInfo(serDeInfo);
-    return storageDescriptor;
-  }
 }
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 1c1cb5550..e71251aee 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
@@ -19,13 +19,24 @@
 package com.netease.arctic.hive.utils;
 
 import com.netease.arctic.hive.HMSClient;
+import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.table.SupportHive;
 import com.netease.arctic.table.ArcticTable;
 import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.thrift.TException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.List;
+import java.util.Map;
+
 public class HiveTableUtil {
 
   private static final Logger LOG = LoggerFactory.getLogger(HiveTableUtil.class);
@@ -61,4 +72,41 @@ public class HiveTableUtil {
       throw new RuntimeException(e);
     }
   }
+
+  public static Map<String, String> generateTableProperties(int accessTimeInSeconds, List<DataFile> files) {
+    Map<String, String> properties = Maps.newHashMap();
+    long totalSize = files.stream().map(DataFile::fileSizeInBytes).reduce(0L, Long::sum);
+    long numRows = files.stream().map(DataFile::recordCount).reduce(0L, Long::sum);
+    properties.put("transient_lastDdlTime", accessTimeInSeconds + "");
+    properties.put("totalSize", totalSize + "");
+    properties.put("numRows", numRows + "");
+    properties.put("numFiles", files.size() + "");
+    properties.put(HiveTableProperties.ARCTIC_TABLE_FLAG, "true");
+    return properties;
+  }
+
+  public static String hiveRootLocation(String tableLocation) {
+    return tableLocation + "/hive";
+  }
+
+  public static StorageDescriptor storageDescriptor(
+      Schema schema, PartitionSpec partitionSpec, String location,
+      FileFormat format) {
+    final StorageDescriptor storageDescriptor = new StorageDescriptor();
+    storageDescriptor.setCols(HiveSchemaUtil.hiveTableFields(schema, partitionSpec));
+    storageDescriptor.setLocation(location);
+    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");
+        break;
+      default:
+        throw new IllegalArgumentException("Unsupported hive table file format:" + format);
+    }
+
+    storageDescriptor.setSerdeInfo(serDeInfo);
+    return storageDescriptor;
+  }
 }
diff --git a/hive/src/main/java/org/apache/hadoop/hive/ql/optimizer/ppr/PartitionExpressionForMetastore.java b/hive/src/main/java/org/apache/hadoop/hive/ql/optimizer/ppr/PartitionExpressionForMetastore.java
index 61441d08d..0aa6cbee2 100644
--- a/hive/src/main/java/org/apache/hadoop/hive/ql/optimizer/ppr/PartitionExpressionForMetastore.java
+++ b/hive/src/main/java/org/apache/hadoop/hive/ql/optimizer/ppr/PartitionExpressionForMetastore.java
@@ -7,7 +7,7 @@
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
- *    http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
diff --git a/hive/src/main/java/org/apache/hadoop/hive/ql/security/authorization/AuthorizationPreEventListener.java b/hive/src/main/java/org/apache/hadoop/hive/ql/security/authorization/AuthorizationPreEventListener.java
index ef64db78c..61f783df2 100644
--- a/hive/src/main/java/org/apache/hadoop/hive/ql/security/authorization/AuthorizationPreEventListener.java
+++ b/hive/src/main/java/org/apache/hadoop/hive/ql/security/authorization/AuthorizationPreEventListener.java
@@ -7,7 +7,7 @@
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
- *    http://www.apache.org/licenses/LICENSE-2.0
+ *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
diff --git a/hive/src/main/java/org/apache/iceberg/data/AdaptHiveGenericAppenderFactory.java b/hive/src/main/java/org/apache/iceberg/data/AdaptHiveGenericAppenderFactory.java
index 632fa244a..f34e426ac 100644
--- a/hive/src/main/java/org/apache/iceberg/data/AdaptHiveGenericAppenderFactory.java
+++ b/hive/src/main/java/org/apache/iceberg/data/AdaptHiveGenericAppenderFactory.java
@@ -35,7 +35,6 @@ import org.apache.iceberg.io.FileAppenderFactory;
 import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.orc.ORC;
 import org.apache.iceberg.parquet.AdaptHiveParquet;
-import org.apache.iceberg.parquet.Parquet;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 
diff --git a/optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkOptimizer.java b/optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkOptimizer.java
index 1e729bb82..feee56c68 100644
--- a/optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkOptimizer.java
+++ b/optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkOptimizer.java
@@ -85,7 +85,7 @@ public class FlinkOptimizer implements StatefulOptimizer {
 
     //add compact execute config
     String arcticHome = systemInfo.getString(OptimizerProperties.ARCTIC_HOME);
-    String jarPath = " " + arcticHome + "/lib/CompactJob.jar ";
+    String jarPath = " " + arcticHome + "/plugin/optimize/OptimizeJob.jar ";
     String entryClass = this.getClass().getName();
     cmd += " -c " + entryClass + jarPath;
     String amsUrl =
diff --git a/site/docs/ch/meta-service/table-properties.md b/site/docs/ch/meta-service/table-properties.md
index 2f8684fea..6e808226e 100644
--- a/site/docs/ch/meta-service/table-properties.md
+++ b/site/docs/ch/meta-service/table-properties.md
@@ -53,7 +53,7 @@
 | clean-orphan-file.enable                    | false     | 是否开启游离文件自动清理                       |
 | clean-orphan-file.min-existing-time-minutes | 2880（2天）  | 清理游离文件时为了防止错误清理正在写入的文件，判断文件最低的存在时间 |
 
-### Logstore相关配置
+### Logstore 相关配置
 
 | 配置名称                            | 默认值             | 描述                                     |
 | ---------------------------------- | ---------------- | ----------------------------------       |
@@ -64,3 +64,10 @@
 | log-store.data-format              | json             | Logstore 中的消息格式，当前仅支持 json         |
 | log-store.data-version             | v1               | Logstore 中消息的版本，当前仅支持 v1           |
 | log.consistency.guarantee.enable   | false            | 标记是否开启一致性保证                       |
+
+### Hive format 相关配置
+
+| 配置名称                            | 默认值             | 描述                                     |
+| ---------------------------------- | ---------------- | ----------------------------------       |
+| base.hive.auto-sync-schema-change  | true             | 是否自动同步 Hive 的原生的结构变更             |
+| base.hive.auto-sync-data-write     | false            | 是否自动同步 Hive 的原生的数据写入             |
\ No newline at end of file


Test patch:
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java
index 2cb152e0c..77f1e038b 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java
@@ -37,6 +37,8 @@ import com.netease.arctic.ams.server.optimize.TestOrphanFileClean;
 import com.netease.arctic.ams.server.service.MetaService;
 import com.netease.arctic.ams.server.service.ServiceContainer;
 import com.netease.arctic.ams.server.service.TestDDLTracerService;
+import com.netease.arctic.ams.server.service.TestFileInfoCacheService;
+import com.netease.arctic.ams.server.service.impl.ArcticTransactionService;
 import com.netease.arctic.ams.server.service.impl.CatalogMetadataService;
 import com.netease.arctic.ams.server.service.impl.DDLTracerService;
 import com.netease.arctic.ams.server.service.impl.FileInfoCacheService;
@@ -74,10 +76,19 @@ import static org.powermock.api.mockito.PowerMockito.when;
 
 @RunWith(PowerMockRunner.class)
 @PowerMockRunnerDelegate(Suite.class)
-@Suite.SuiteClasses({OptimizerControllerTest.class, TableControllerTest.class, TerminalControllerTest.class,
-                     TestDDLTracerService.class,LoginControllerTest.class, TestExpiredFileClean.class,
-                     TestMajorOptimizeCommit.class, TestMajorOptimizePlan.class, TestMinorOptimizeCommit.class,
-                     TestMinorOptimizePlan.class, TestOrphanFileClean.class})
+@Suite.SuiteClasses({
+    OptimizerControllerTest.class,
+    TableControllerTest.class,
+    TerminalControllerTest.class,
+    TestDDLTracerService.class,
+    LoginControllerTest.class,
+    TestExpiredFileClean.class,
+    TestMajorOptimizeCommit.class,
+    TestMajorOptimizePlan.class,
+    TestMinorOptimizeCommit.class,
+    TestMinorOptimizePlan.class,
+    TestOrphanFileClean.class,
+    TestFileInfoCacheService.class})
 @PrepareForTest({
     JDBCSqlSessionFactoryProvider.class,
     ArcticMetaStore.class,
@@ -124,14 +135,16 @@ public class AmsTestBase {
     mockStatic(CatalogMetadataService.class);
 
     //mock service
+    FileInfoCacheService fileInfoCacheService = new FileInfoCacheService();
+    when(ServiceContainer.getFileInfoCacheService()).thenReturn(fileInfoCacheService);
+    ArcticTransactionService arcticTransactionService = new ArcticTransactionService();
+    when(ServiceContainer.getArcticTransactionService()).thenReturn(arcticTransactionService);
+    DDLTracerService ddlTracerService = new DDLTracerService();
+    when(ServiceContainer.getDdlTracerService()).thenReturn(ddlTracerService);
     CatalogMetadataService catalogMetadataService = new CatalogMetadataService();
     when(ServiceContainer.getCatalogMetadataService()).thenReturn(catalogMetadataService);
     JDBCMetaService metaService = new JDBCMetaService();
     when(ServiceContainer.getMetaService()).thenReturn(metaService);
-    DDLTracerService ddlTracerService = new DDLTracerService();
-    when(ServiceContainer.getDdlTracerService()).thenReturn(ddlTracerService);
-    FileInfoCacheService fileInfoCacheService = new FileInfoCacheService();
-    when(ServiceContainer.getFileInfoCacheService()).thenReturn(fileInfoCacheService);
 
     //mock handler
     amsHandler = new ArcticTableMetastoreHandler(ServiceContainer.getMetaService());
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestDDLTracerService.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestDDLTracerService.java
index 4feb1a572..b603d7ab2 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestDDLTracerService.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestDDLTracerService.java
@@ -33,18 +33,18 @@ import org.apache.iceberg.types.Types;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;
-import org.powermock.core.classloader.annotations.PowerMockIgnore;
-import org.powermock.core.classloader.annotations.PrepareForTest;
 
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+
 import static com.netease.arctic.ams.server.AmsTestBase.AMS_TEST_CATALOG_NAME;
 import static com.netease.arctic.ams.server.AmsTestBase.AMS_TEST_DB_NAME;
 
-// @RunWith(PowerMockRunner.class)
 @PowerMockIgnore({"javax.management.*"})
 @PrepareForTest({
     ServiceContainer.class,
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestFileInfoCacheService.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestFileInfoCacheService.java
new file mode 100644
index 000000000..8f095983c
--- /dev/null
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestFileInfoCacheService.java
@@ -0,0 +1,267 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *  *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *  *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.ams.server.service;
+
+import com.netease.arctic.TableTestBase;
+import com.netease.arctic.ams.api.CommitMetaProducer;
+import com.netease.arctic.ams.api.DataFile;
+import com.netease.arctic.ams.api.DataFileInfo;
+import com.netease.arctic.ams.api.MetaException;
+import com.netease.arctic.ams.api.TableChange;
+import com.netease.arctic.ams.api.TableCommitMeta;
+import com.netease.arctic.ams.api.TableIdentifier;
+import com.netease.arctic.ams.server.model.TransactionsOfTable;
+import com.netease.arctic.table.KeyedTable;
+import com.netease.arctic.table.TableProperties;
+import com.netease.arctic.table.UnkeyedTable;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static com.netease.arctic.ams.server.AmsTestBase.AMS_TEST_CATALOG_NAME;
+import static com.netease.arctic.ams.server.AmsTestBase.AMS_TEST_DB_NAME;
+import static com.netease.arctic.ams.server.AmsTestBase.catalog;
+
+public class TestFileInfoCacheService extends TableTestBase {
+
+  TableIdentifier tableIdentifier = new TableIdentifier("test", "test", "test");
+  Random random = new Random();
+
+  @Test
+  public void testAppendCommit() throws MetaException {
+    TableCommitMeta meta = new TableCommitMeta();
+    meta.setAction("append");
+    meta.setCommitTime(System.currentTimeMillis());
+    meta.setCommitMetaProducer(CommitMetaProducer.INGESTION);
+    meta.setTableIdentifier(tableIdentifier);
+    List<TableChange> changes = new ArrayList<>();
+    TableChange change = new TableChange();
+    change.setParentSnapshotId(-1);
+    change.setInnerTable("base");
+    List<DataFile> dataFiles = new ArrayList<>();
+    dataFiles.add(genDatafile());
+    change.setAddFiles(dataFiles);
+    long snapshotId = 1L;
+    change.setSnapshotId(snapshotId);
+    TableChange change1 = new TableChange();
+    change1.setParentSnapshotId(snapshotId);
+    change1.setInnerTable("base");
+    List<DataFile> dataFiles1 = new ArrayList<>();
+    dataFiles1.add(genDatafile());
+    change1.setAddFiles(dataFiles1);
+    long snapshotId1 = 2L;
+    change1.setSnapshotId(snapshotId1);
+
+    changes.add(change);
+    changes.add(change1);
+    meta.setChanges(changes);
+    Map<String, String> properties = new HashMap<>();
+    properties.put(TableProperties.TABLE_EVENT_TIME_FIELD, "eventTime");
+    meta.setProperties(properties);
+    ServiceContainer.getFileInfoCacheService().commitCacheFileInfo(meta);
+
+    List<TransactionsOfTable> transactionsOfTables =
+        ServiceContainer.getFileInfoCacheService().getTransactions(tableIdentifier);
+    Assert.assertEquals(2, transactionsOfTables.size());
+    Assert.assertEquals(snapshotId1, transactionsOfTables.get(0).getTransactionId());
+    Assert.assertEquals(snapshotId, transactionsOfTables.get(1).getTransactionId());
+  }
+
+  @Test
+  public void testNeedFixCacheWhenParentNotCached() throws MetaException {
+    TableCommitMeta meta = new TableCommitMeta();
+    meta.setAction("append");
+    meta.setCommitTime(System.currentTimeMillis());
+    meta.setCommitMetaProducer(CommitMetaProducer.INGESTION);
+    meta.setTableIdentifier(tableIdentifier);
+    List<TableChange> changes = new ArrayList<>();
+    TableChange change = new TableChange();
+    change.setParentSnapshotId(1);
+    change.setInnerTable("base");
+    List<DataFile> dataFiles = new ArrayList<>();
+    dataFiles.add(genDatafile());
+    change.setAddFiles(dataFiles);
+    long snapshotId = 2L;
+    change.setSnapshotId(snapshotId);
+    ServiceContainer.getFileInfoCacheService().commitCacheFileInfo(meta);
+    meta.setChanges(changes);
+
+    AtomicBoolean isCached = new AtomicBoolean(false);
+    ServiceContainer.getFileInfoCacheService().getTransactions(tableIdentifier).forEach(transactionsOfTable -> {
+      if (transactionsOfTable.getTransactionId() == snapshotId) {
+        isCached.set(true);
+      }
+    });
+    Assert.assertFalse(isCached.get());
+  }
+
+  @Test
+  public void testUpdateDeleteCommit() throws MetaException {
+    TableCommitMeta meta = new TableCommitMeta();
+    meta.setAction("append");
+    meta.setCommitTime(System.currentTimeMillis());
+    meta.setCommitMetaProducer(CommitMetaProducer.INGESTION);
+    meta.setTableIdentifier(tableIdentifier);
+    List<TableChange> changes = new ArrayList<>();
+    TableChange change = new TableChange();
+    change.setParentSnapshotId(-1);
+    change.setInnerTable("base");
+    List<DataFile> dataFiles = new ArrayList<>();
+    dataFiles.add(genDatafile());
+    change.setAddFiles(dataFiles);
+    long snapshotId = 1L;
+    change.setSnapshotId(snapshotId);
+
+    TableChange change1 = new TableChange();
+    change1.setParentSnapshotId(snapshotId);
+    change1.setInnerTable("base");
+    List<DataFile> dataFiles1 = new ArrayList<>();
+    dataFiles1.add(genDatafile());
+    change1.setDeleteFiles(dataFiles);
+    change1.setAddFiles(dataFiles1);
+    long snapshotId1 = 2L;
+    change1.setSnapshotId(snapshotId1);
+
+    changes.add(change);
+    changes.add(change1);
+    meta.setChanges(changes);
+    Map<String, String> properties = new HashMap<>();
+    properties.put(TableProperties.TABLE_EVENT_TIME_FIELD, "eventTime");
+    meta.setProperties(properties);
+    ServiceContainer.getFileInfoCacheService().commitCacheFileInfo(meta);
+
+    List<TransactionsOfTable> transactionsOfTables =
+        ServiceContainer.getFileInfoCacheService().getTransactions(tableIdentifier);
+    Assert.assertEquals(2, transactionsOfTables.size());
+    Assert.assertEquals(snapshotId1, transactionsOfTables.get(0).getTransactionId());
+    Assert.assertEquals(snapshotId, transactionsOfTables.get(1).getTransactionId());
+
+    List<DataFileInfo> dataFileInfos = ServiceContainer.getFileInfoCacheService().getOptimizeDatafiles(
+        tableIdentifier,
+        "base");
+    Assert.assertEquals(dataFiles1.size(), dataFileInfos.size());
+    Assert.assertEquals(dataFiles1.get(0).getPath(), dataFileInfos.get(0).getPath());
+  }
+
+  @Test
+  public void testUnkeyedTableSyncFileCache() {
+    com.netease.arctic.table.TableIdentifier tableId =
+        com.netease.arctic.table.TableIdentifier.of(AMS_TEST_CATALOG_NAME, AMS_TEST_DB_NAME,
+            "file_sync_test_unkeyed_table");
+    UnkeyedTable fileSyncUnkeyedTable = catalog
+        .newTableBuilder(
+            tableId,
+            TABLE_SCHEMA).withPartitionSpec(SPEC).create().asUnkeyedTable();
+    testSyncFileCache(fileSyncUnkeyedTable, tableId);
+  }
+
+  @Test
+  public void testKeyedTableSyncFileCache() {
+    com.netease.arctic.table.TableIdentifier tableId =
+        com.netease.arctic.table.TableIdentifier.of(AMS_TEST_CATALOG_NAME, AMS_TEST_DB_NAME,
+            "file_sync_test_keyed_table");
+    KeyedTable fileSyncKeyedTable = catalog
+        .newTableBuilder(
+            tableId,
+            TABLE_SCHEMA).withPrimaryKeySpec(PRIMARY_KEY_SPEC).withPartitionSpec(SPEC).create().asKeyedTable();
+    testSyncFileCache(fileSyncKeyedTable.baseTable(), tableId);
+  }
+
+  public void testSyncFileCache(UnkeyedTable fileSyncUnkeyedTable, com.netease.arctic.table.TableIdentifier tableId) {
+    fileSyncUnkeyedTable.newFastAppend()
+        .appendFile(FILE_A)
+        .appendFile(FILE_B)
+        .commit();
+    List<DataFileInfo> commitDataFileInfos = ServiceContainer.getFileInfoCacheService().getOptimizeDatafiles(
+        fileSyncUnkeyedTable.id().buildTableIdentifier(),
+        "base");
+    List<TransactionsOfTable> commitSnapInfos = ServiceContainer.getFileInfoCacheService().getTransactions(
+        fileSyncUnkeyedTable.id().buildTableIdentifier());
+    ServiceContainer.getFileInfoCacheService().deleteTableCache(tableId);
+    List<DataFileInfo> cacheDataFileInfos = ServiceContainer.getFileInfoCacheService().getOptimizeDatafiles(
+        fileSyncUnkeyedTable.id().buildTableIdentifier(),
+        "base");
+    Assert.assertEquals(0, cacheDataFileInfos.size());
+    ServiceContainer.getFileInfoCacheService()
+        .syncTableFileInfo(fileSyncUnkeyedTable.id().buildTableIdentifier(), "base");
+    List<DataFileInfo> syncDataFileInfos = ServiceContainer.getFileInfoCacheService().getOptimizeDatafiles(
+        fileSyncUnkeyedTable.id().buildTableIdentifier(),
+        "base");
+    List<TransactionsOfTable> syncSnapInfos = ServiceContainer.getFileInfoCacheService().getTransactions(
+        fileSyncUnkeyedTable.id().buildTableIdentifier());
+    Assert.assertEquals(commitDataFileInfos.size(), syncDataFileInfos.size());
+    for (DataFileInfo commitDataFileInfo : commitDataFileInfos) {
+      boolean isCached = false;
+      for (DataFileInfo syncDataFileInfo : syncDataFileInfos) {
+        if (commitDataFileInfo.getPath().equals(syncDataFileInfo.getPath())) {
+          isCached = true;
+          break;
+        }
+      }
+      Assert.assertTrue(isCached);
+    }
+    Assert.assertEquals(commitSnapInfos.size(), syncSnapInfos.size());
+    for (int i = 0; i < commitSnapInfos.size(); i++) {
+      Assert.assertEquals(commitSnapInfos.get(i).getTransactionId(), syncSnapInfos.get(i).getTransactionId());
+    }
+
+    fileSyncUnkeyedTable.newOverwrite()
+        .deleteFile(FILE_A)
+        .deleteFile(FILE_B)
+        .addFile(FILE_C)
+        .commit();
+    List<TransactionsOfTable> overwriteCommitSnapInfos = ServiceContainer.getFileInfoCacheService().getTransactions(
+        fileSyncUnkeyedTable.id().buildTableIdentifier());
+    ServiceContainer.getFileInfoCacheService().deleteTableCache(tableId);
+    ServiceContainer.getFileInfoCacheService()
+        .syncTableFileInfo(fileSyncUnkeyedTable.id().buildTableIdentifier(), "base");
+    List<DataFileInfo> overwriteSyncDataFileInfos = ServiceContainer.getFileInfoCacheService().getOptimizeDatafiles(
+        fileSyncUnkeyedTable.id().buildTableIdentifier(),
+        "base");
+    List<TransactionsOfTable> overwriteSnapInfos = ServiceContainer.getFileInfoCacheService().getTransactions(
+        fileSyncUnkeyedTable.id().buildTableIdentifier());
+    Assert.assertEquals(1, overwriteSyncDataFileInfos.size());
+    Assert.assertEquals(FILE_C.path(), overwriteSyncDataFileInfos.get(0).getPath());
+    Assert.assertEquals(1, overwriteSnapInfos.size());
+    Assert.assertEquals(overwriteCommitSnapInfos.get(0).getTransactionId(),
+        overwriteSnapInfos.get(0).getTransactionId());
+  }
+
+  private DataFile genDatafile() {
+    DataFile dataFile = new DataFile();
+    dataFile.setFileSize(1);
+    dataFile.setFileType("INSERT_FILE");
+    dataFile.setIndex(0);
+    dataFile.setMask(0);
+    dataFile.setPath("/tmp/test" + random.nextInt() + ".file");
+    Map<String, ByteBuffer> upperBounds = new HashMap<>();
+    byte[] bytes = ByteBuffer.allocate(Long.SIZE / Byte.SIZE).putLong(1000L).array();
+    upperBounds.put("eventTime", ByteBuffer.wrap(bytes));
+    dataFile.setUpperBounds(upperBounds);
+    dataFile.setPartition(new ArrayList<>());
+    return dataFile;
+  }
+}
diff --git a/core/src/test/java/com/netease/arctic/op/UpdatePartitionPropertiesTest.java b/core/src/test/java/com/netease/arctic/op/UpdatePartitionPropertiesTest.java
index 946e0a13b..7770c04c2 100644
--- a/core/src/test/java/com/netease/arctic/op/UpdatePartitionPropertiesTest.java
+++ b/core/src/test/java/com/netease/arctic/op/UpdatePartitionPropertiesTest.java
@@ -42,6 +42,7 @@ public class UpdatePartitionPropertiesTest extends TableTestBase {
     Assert.assertEquals("value", partitionProperties.get(p0).get("key"));
   }
 
+  @Test
   public void testUpdatePartitionPropertiesInTx() {
     StructLikeMap<Map<String, String>> partitionProperties = testTable.partitionProperty();
     Transaction transaction = testTable.newTransaction();
diff --git a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java
index 6b39dac7f..5fe596da3 100644
--- a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java
+++ b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java
@@ -199,10 +199,10 @@ public class AdaptHiveWriterTest extends HiveTableTestBase {
   private List<RowData> geneRowData(){
     RowData rowData = GenericRowData.of(
         1,
-        StringData.fromString("jack"),
         TimestampData.fromLocalDateTime(LocalDateTime.of(2022, 1, 1 , 10, 0, 0)),
         TimestampData.fromInstant(LocalDateTime.of(2022, 1, 1 , 10, 0, 0).toInstant(ZoneOffset.ofHours(8))),
-        DecimalData.fromBigDecimal(new BigDecimal("100"), 3, 0)
+        DecimalData.fromBigDecimal(new BigDecimal("100"), 3, 0),
+        StringData.fromString("jack")
     );
     return Lists.newArrayList(rowData);
   }
diff --git a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/ArcticFileWriterTest.java b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/ArcticFileWriterTest.java
index 1b8829bf8..1d5f6fc31 100644
--- a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/ArcticFileWriterTest.java
+++ b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/ArcticFileWriterTest.java
@@ -21,7 +21,6 @@ package com.netease.arctic.flink.write;
 import com.netease.arctic.flink.FlinkTestBase;
 import com.netease.arctic.flink.table.ArcticTableLoader;
 import com.netease.arctic.table.ArcticTable;
-import java.util.List;
 import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.logical.RowType;
@@ -35,16 +34,42 @@ import org.apache.iceberg.io.TaskWriter;
 import org.apache.iceberg.io.WriteResult;
 import org.junit.Assert;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.util.List;
+
+import static com.netease.arctic.flink.table.descriptors.ArcticValidator.SUBMIT_EMPTY_SNAPSHOTS;
 
+@RunWith(Parameterized.class)
 public class ArcticFileWriterTest extends FlinkTestBase {
 
   public static final long TARGET_FILE_SIZE = 128 * 1024 * 1024;
   public ArcticTableLoader tableLoader;
+  private final boolean submitEmptySnapshots;
+
+  @Parameterized.Parameters(name = "submitEmptySnapshots = {0}")
+  public static Object[][] parameters() {
+    return new Object[][] {
+        {false},
+        {true}
+    };
+  }
+
+  public ArcticFileWriterTest(boolean submitEmptySnapshots) {
+    this.submitEmptySnapshots = submitEmptySnapshots;
+  }
 
   public static OneInputStreamOperatorTestHarness<RowData, WriteResult> createArcticStreamWriter(
       ArcticTableLoader tableLoader) throws Exception {
+    return createArcticStreamWriter(tableLoader, true);
+  }
+
+  public static OneInputStreamOperatorTestHarness<RowData, WriteResult> createArcticStreamWriter(
+      ArcticTableLoader tableLoader, boolean submitEmptySnapshots) throws Exception {
     tableLoader.open();
     ArcticTable arcticTable = tableLoader.loadArcticTable();
+    arcticTable.properties().put(SUBMIT_EMPTY_SNAPSHOTS.key(), String.valueOf(submitEmptySnapshots));
 
     ArcticFileWriter streamWriter = FlinkSink.createFileWriter(arcticTable,
         null,
@@ -234,4 +259,25 @@ public class ArcticFileWriterTest extends FlinkTestBase {
     }
   }
 
+  @Test
+  public void testEmitEmptyResults() throws Exception {
+
+    tableLoader = ArcticTableLoader.of(PK_TABLE_ID, catalogBuilder);
+    long checkpointId = 1L;
+    long excepted = submitEmptySnapshots ? 1 : 0;
+    try (
+        OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness = createArcticStreamWriter(
+            tableLoader, submitEmptySnapshots)) {
+      // The first checkpoint
+      testHarness.prepareSnapshotPreBarrier(checkpointId);
+      Assert.assertEquals(excepted, testHarness.extractOutputValues().size());
+
+      checkpointId = checkpointId + 1;
+
+      // The second checkpoint
+      testHarness.prepareSnapshotPreBarrier(checkpointId);
+      Assert.assertEquals(excepted, testHarness.extractOutputValues().size());
+    }
+  }
+
 }
\ No newline at end of file
diff --git a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java
index 0afaf1366..347cb80c8 100644
--- a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java
+++ b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java
@@ -193,10 +193,10 @@ public class AdaptHiveWriterTest extends HiveTableTestBase {
   private List<RowData> geneRowData(){
     RowData rowData = GenericRowData.of(
         1,
-        StringData.fromString("jack"),
         TimestampData.fromLocalDateTime(LocalDateTime.of(2022, 1, 1 , 10, 0, 0)),
         TimestampData.fromInstant(LocalDateTime.of(2022, 1, 1 , 10, 0, 0).toInstant(ZoneOffset.ofHours(8))),
-        DecimalData.fromBigDecimal(new BigDecimal("100"), 3, 0)
+        DecimalData.fromBigDecimal(new BigDecimal("100"), 3, 0),
+        StringData.fromString("jack")
     );
     return Lists.newArrayList(rowData);
   }
diff --git a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/ArcticFileWriterTest.java b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/ArcticFileWriterTest.java
index 35e1ea02a..1c3f98b4a 100644
--- a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/ArcticFileWriterTest.java
+++ b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/ArcticFileWriterTest.java
@@ -34,18 +34,42 @@ import org.apache.iceberg.io.TaskWriter;
 import org.apache.iceberg.io.WriteResult;
 import org.junit.Assert;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
 import java.util.List;
 
+import static com.netease.arctic.flink.table.descriptors.ArcticValidator.SUBMIT_EMPTY_SNAPSHOTS;
+
+@RunWith(Parameterized.class)
 public class ArcticFileWriterTest extends FlinkTestBase {
 
   public static final long TARGET_FILE_SIZE = 128 * 1024 * 1024;
   public ArcticTableLoader tableLoader;
+  private final boolean submitEmptySnapshots;
+
+  @Parameterized.Parameters(name = "submitEmptySnapshots = {0}")
+  public static Object[][] parameters() {
+    return new Object[][] {
+        {false},
+        {true}
+    };
+  }
+
+  public ArcticFileWriterTest(boolean submitEmptySnapshots) {
+    this.submitEmptySnapshots = submitEmptySnapshots;
+  }
 
   public static OneInputStreamOperatorTestHarness<RowData, WriteResult> createArcticStreamWriter(
       ArcticTableLoader tableLoader) throws Exception {
+    return createArcticStreamWriter(tableLoader, true);
+  }
+
+  public static OneInputStreamOperatorTestHarness<RowData, WriteResult> createArcticStreamWriter(
+      ArcticTableLoader tableLoader, boolean submitEmptySnapshots) throws Exception {
     tableLoader.open();
     ArcticTable arcticTable = tableLoader.loadArcticTable();
+    arcticTable.properties().put(SUBMIT_EMPTY_SNAPSHOTS.key(), String.valueOf(submitEmptySnapshots));
 
     ArcticFileWriter streamWriter = FlinkSink.createFileWriter(arcticTable,
         null,
@@ -235,4 +259,26 @@ public class ArcticFileWriterTest extends FlinkTestBase {
     }
   }
 
+  @Test
+  public void testEmitEmptyResults() throws Exception {
+
+    tableLoader = ArcticTableLoader.of(PK_TABLE_ID, catalogBuilder);
+    long checkpointId = 1L;
+    long excepted = submitEmptySnapshots ? 1 : 0;
+    try (
+        OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness = createArcticStreamWriter(
+            tableLoader, submitEmptySnapshots)) {
+      // The first checkpoint
+
+      testHarness.prepareSnapshotPreBarrier(checkpointId);
+      Assert.assertEquals(excepted, testHarness.extractOutputValues().size());
+
+      checkpointId = checkpointId + 1;
+
+      // The second checkpoint
+      testHarness.prepareSnapshotPreBarrier(checkpointId);
+      Assert.assertEquals(excepted, testHarness.extractOutputValues().size());
+    }
+  }
+
 }
\ No newline at end of file
diff --git a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java
index a2d30ba44..7de3b7bc6 100644
--- a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java
+++ b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java
@@ -193,10 +193,10 @@ public class AdaptHiveWriterTest extends HiveTableTestBase {
   private List<RowData> geneRowData(){
     RowData rowData = GenericRowData.of(
         1,
-        StringData.fromString("jack"),
         TimestampData.fromLocalDateTime(LocalDateTime.of(2022, 1, 1 , 10, 0, 0)),
         TimestampData.fromInstant(LocalDateTime.of(2022, 1, 1 , 10, 0, 0).toInstant(ZoneOffset.ofHours(8))),
-        DecimalData.fromBigDecimal(new BigDecimal("100"), 3, 0)
+        DecimalData.fromBigDecimal(new BigDecimal("100"), 3, 0),
+        StringData.fromString("jack")
     );
     return Lists.newArrayList(rowData);
   }
diff --git a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/ArcticFileWriterTest.java b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/ArcticFileWriterTest.java
index 35e1ea02a..1c3f98b4a 100644
--- a/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/ArcticFileWriterTest.java
+++ b/flink/v1.15/flink/src/test/java/com/netease/arctic/flink/write/ArcticFileWriterTest.java
@@ -34,18 +34,42 @@ import org.apache.iceberg.io.TaskWriter;
 import org.apache.iceberg.io.WriteResult;
 import org.junit.Assert;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
 import java.util.List;
 
+import static com.netease.arctic.flink.table.descriptors.ArcticValidator.SUBMIT_EMPTY_SNAPSHOTS;
+
+@RunWith(Parameterized.class)
 public class ArcticFileWriterTest extends FlinkTestBase {
 
   public static final long TARGET_FILE_SIZE = 128 * 1024 * 1024;
   public ArcticTableLoader tableLoader;
+  private final boolean submitEmptySnapshots;
+
+  @Parameterized.Parameters(name = "submitEmptySnapshots = {0}")
+  public static Object[][] parameters() {
+    return new Object[][] {
+        {false},
+        {true}
+    };
+  }
+
+  public ArcticFileWriterTest(boolean submitEmptySnapshots) {
+    this.submitEmptySnapshots = submitEmptySnapshots;
+  }
 
   public static OneInputStreamOperatorTestHarness<RowData, WriteResult> createArcticStreamWriter(
       ArcticTableLoader tableLoader) throws Exception {
+    return createArcticStreamWriter(tableLoader, true);
+  }
+
+  public static OneInputStreamOperatorTestHarness<RowData, WriteResult> createArcticStreamWriter(
+      ArcticTableLoader tableLoader, boolean submitEmptySnapshots) throws Exception {
     tableLoader.open();
     ArcticTable arcticTable = tableLoader.loadArcticTable();
+    arcticTable.properties().put(SUBMIT_EMPTY_SNAPSHOTS.key(), String.valueOf(submitEmptySnapshots));
 
     ArcticFileWriter streamWriter = FlinkSink.createFileWriter(arcticTable,
         null,
@@ -235,4 +259,26 @@ public class ArcticFileWriterTest extends FlinkTestBase {
     }
   }
 
+  @Test
+  public void testEmitEmptyResults() throws Exception {
+
+    tableLoader = ArcticTableLoader.of(PK_TABLE_ID, catalogBuilder);
+    long checkpointId = 1L;
+    long excepted = submitEmptySnapshots ? 1 : 0;
+    try (
+        OneInputStreamOperatorTestHarness<RowData, WriteResult> testHarness = createArcticStreamWriter(
+            tableLoader, submitEmptySnapshots)) {
+      // The first checkpoint
+
+      testHarness.prepareSnapshotPreBarrier(checkpointId);
+      Assert.assertEquals(excepted, testHarness.extractOutputValues().size());
+
+      checkpointId = checkpointId + 1;
+
+      // The second checkpoint
+      testHarness.prepareSnapshotPreBarrier(checkpointId);
+      Assert.assertEquals(excepted, testHarness.extractOutputValues().size());
+    }
+  }
+
 }
\ No newline at end of file
diff --git a/hive/src/test/java/com/netease/arctic/hive/ArcticHiveTestMain.java b/hive/src/test/java/com/netease/arctic/hive/ArcticHiveTestMain.java
index 0b690c4ee..33f49857f 100644
--- a/hive/src/test/java/com/netease/arctic/hive/ArcticHiveTestMain.java
+++ b/hive/src/test/java/com/netease/arctic/hive/ArcticHiveTestMain.java
@@ -19,9 +19,12 @@
 package com.netease.arctic.hive;
 
 import com.netease.arctic.hive.catalog.TestArcticHiveCatalog;
+import com.netease.arctic.hive.io.AdaptHiveWriterTest;
+import com.netease.arctic.hive.op.AutoSyncHiveTest;
 import com.netease.arctic.hive.op.TestHiveSchemaUpdate;
 import com.netease.arctic.hive.op.TestOverwriteFiles;
 import com.netease.arctic.hive.op.TestRewritePartitions;
+import com.netease.arctic.hive.utils.HiveMetaSynchronizerTest;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.runner.RunWith;
@@ -32,7 +35,10 @@ import org.junit.runners.Suite;
     TestArcticHiveCatalog.class,
     TestRewritePartitions.class,
     TestOverwriteFiles.class,
-    TestHiveSchemaUpdate.class
+    TestHiveSchemaUpdate.class,
+    HiveMetaSynchronizerTest.class,
+    AdaptHiveWriterTest.class,
+    AutoSyncHiveTest.class
 })
 public class ArcticHiveTestMain {
 
diff --git a/hive/src/test/java/com/netease/arctic/hive/HMSMockServer.java b/hive/src/test/java/com/netease/arctic/hive/HMSMockServer.java
index 33551a8d0..066a5c83e 100644
--- a/hive/src/test/java/com/netease/arctic/hive/HMSMockServer.java
+++ b/hive/src/test/java/com/netease/arctic/hive/HMSMockServer.java
@@ -18,10 +18,6 @@
 
 package com.netease.arctic.hive;
 
-import java.io.File;
-import java.util.Random;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
@@ -45,6 +41,11 @@ import org.apache.thrift.transport.TTransportFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.File;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 public class HMSMockServer {
 
   public static final Logger LOG = LoggerFactory.getLogger(HMSMockServer.class);
diff --git a/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java b/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java
index 0b8512646..e98ece153 100644
--- a/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java
+++ b/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java
@@ -29,32 +29,29 @@ import com.netease.arctic.hive.table.KeyedHiveTable;
 import com.netease.arctic.hive.table.UnkeyedHiveTable;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.TableIdentifier;
-import com.netease.arctic.table.TableProperties;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicInteger;
-
-import com.netease.arctic.table.TableProperties;
 import com.netease.arctic.table.UnkeyedTable;
 import org.apache.hadoop.hive.metastore.api.Database;
 import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.StructLikeMap;
 import org.apache.thrift.TException;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.types.Types;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.BeforeClass;
+import org.junit.rules.TemporaryFolder;
 
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
-import org.junit.rules.TemporaryFolder;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_HIVE;
 
@@ -65,7 +62,7 @@ public class HiveTableTestBase extends TableTestBase {
 
   protected static final TemporaryFolder tempFolder = new TemporaryFolder();
 
-  protected static HMSMockServer hms  ;
+  protected static HMSMockServer hms;
 
   protected static final TableIdentifier HIVE_TABLE_ID =
       TableIdentifier.of(HIVE_CATALOG_NAME, HIVE_DB_NAME, "test_hive_table");
@@ -79,10 +76,10 @@ public class HiveTableTestBase extends TableTestBase {
 
   public static final Schema HIVE_TABLE_SCHEMA = new Schema(
       Types.NestedField.required(1, "id", Types.IntegerType.get()),
-      Types.NestedField.required(2, "name", Types.StringType.get()),
-      Types.NestedField.required(3, "op_time", Types.TimestampType.withoutZone()),
-      Types.NestedField.required(4, "op_time_with_zone", Types.TimestampType.withZone()),
-      Types.NestedField.required(5, "d", Types.DecimalType.of(3, 0))
+      Types.NestedField.required(2, "op_time", Types.TimestampType.withoutZone()),
+      Types.NestedField.required(3, "op_time_with_zone", Types.TimestampType.withZone()),
+      Types.NestedField.required(4, "d", Types.DecimalType.of(3, 0)),
+      Types.NestedField.required(5, "name", Types.StringType.get())
   );
 
   protected static final PartitionSpec HIVE_SPEC =
@@ -209,14 +206,6 @@ public class HiveTableTestBase extends TableTestBase {
         tableName,
         (short) -1);
 
-    System.out.println("> assert hive partition location as expected");
-    System.out.printf("HiveTable[%s.%s] partition count: %d \n", database, tableName, partitions.size());
-    for (Partition p: partitions){
-      System.out.printf(
-          "HiveTablePartition[%s.%s  %s] location:%s \n", database, tableName,
-          Joiner.on("/").join(p.getValues()), p.getSd().getLocation());
-    }
-
     Assert.assertEquals("expect " + partitionLocations.size() + " partition after first rewrite partition",
         partitionLocations.size(), partitions.size());
 
@@ -238,8 +227,11 @@ public class HiveTableTestBase extends TableTestBase {
           "partition location is not expected, expect " + actualLocation + " end-with " + locationExpect,
           actualLocation.contains(locationExpect));
       Map<String, String> properties = partitionProperties.get(getPartitionData(valuePath, table.spec()));
-      Assert.assertEquals("partition properties is not expected", actualLocation,
-          properties.get(TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION));
+      Assert.assertEquals("partition property hive location is not expected", actualLocation,
+          properties.get(HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION));
+      Assert.assertEquals("partition property transient_lastDdlTime is not expected",
+          p.getParameters().get("transient_lastDdlTime"),
+          properties.get(HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME));
     }
   }
 }
diff --git a/hive/src/test/java/com/netease/arctic/hive/MockDataFileBuilder.java b/hive/src/test/java/com/netease/arctic/hive/MockDataFileBuilder.java
index 667ee4152..fd4e0242a 100644
--- a/hive/src/test/java/com/netease/arctic/hive/MockDataFileBuilder.java
+++ b/hive/src/test/java/com/netease/arctic/hive/MockDataFileBuilder.java
@@ -3,10 +3,6 @@ package com.netease.arctic.hive;
 import com.netease.arctic.hive.table.SupportHive;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.TableIdentifier;
-import java.io.IOException;
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
 import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
 import org.apache.hadoop.hive.metastore.api.Table;
@@ -14,6 +10,11 @@ import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
 import org.apache.thrift.TException;
 
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
 public class MockDataFileBuilder {
   final TableIdentifier identifier;
   final Table hiveTable;
diff --git a/hive/src/test/java/com/netease/arctic/hive/catalog/TestArcticHiveCatalog.java b/hive/src/test/java/com/netease/arctic/hive/catalog/TestArcticHiveCatalog.java
index 2187fb91d..815bf619e 100644
--- a/hive/src/test/java/com/netease/arctic/hive/catalog/TestArcticHiveCatalog.java
+++ b/hive/src/test/java/com/netease/arctic/hive/catalog/TestArcticHiveCatalog.java
@@ -21,10 +21,8 @@ package com.netease.arctic.hive.catalog;
 import com.netease.arctic.hive.HiveTableTestBase;
 import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.UnkeyedTable;
-import org.apache.iceberg.Schema;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
-import org.apache.iceberg.types.Types;
 import org.junit.Assert;
 import org.junit.Test;
 
diff --git a/hive/src/test/java/com/netease/arctic/hive/io/AdaptHiveWriterTest.java b/hive/src/test/java/com/netease/arctic/hive/io/AdaptHiveWriterTest.java
index 3b550aa1d..1cdf19ef7 100644
--- a/hive/src/test/java/com/netease/arctic/hive/io/AdaptHiveWriterTest.java
+++ b/hive/src/test/java/com/netease/arctic/hive/io/AdaptHiveWriterTest.java
@@ -19,8 +19,8 @@
 package com.netease.arctic.hive.io;
 
 import com.netease.arctic.hive.HiveTableTestBase;
-import com.netease.arctic.hive.table.HiveLocationKind;
 import com.netease.arctic.hive.io.writer.AdaptHiveGenericTaskWriterBuilder;
+import com.netease.arctic.hive.table.HiveLocationKind;
 import com.netease.arctic.io.writer.GenericBaseTaskWriter;
 import com.netease.arctic.io.writer.GenericChangeTaskWriter;
 import com.netease.arctic.table.ArcticTable;
@@ -28,14 +28,6 @@ import com.netease.arctic.table.BaseLocationKind;
 import com.netease.arctic.table.ChangeLocationKind;
 import com.netease.arctic.table.LocationKind;
 import com.netease.arctic.table.WriteOperationKind;
-import java.io.File;
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import java.util.stream.Collectors;
 import org.apache.iceberg.Files;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.data.Record;
@@ -48,6 +40,14 @@ import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
 public class AdaptHiveWriterTest extends HiveTableTestBase {
 
   @Test
@@ -172,7 +172,7 @@ public class AdaptHiveWriterTest extends HiveTableTestBase {
 
   private CloseableIterable<Record> readParquet(Schema schema, String path){
     AdaptHiveParquet.ReadBuilder builder = AdaptHiveParquet.read(
-            Files.localInput(path))
+        Files.localInput(path))
         .project(schema)
         .createReaderFunc(fileSchema -> AdaptHiveGenericParquetReaders.buildReader(schema, fileSchema, new HashMap<>()))
         .caseSensitive(false);
diff --git a/hive/src/test/java/com/netease/arctic/hive/io/HiveTestRecords.java b/hive/src/test/java/com/netease/arctic/hive/io/HiveTestRecords.java
index 2512ba1ef..37d257554 100644
--- a/hive/src/test/java/com/netease/arctic/hive/io/HiveTestRecords.java
+++ b/hive/src/test/java/com/netease/arctic/hive/io/HiveTestRecords.java
@@ -18,15 +18,16 @@
 
 package com.netease.arctic.hive.io;
 
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+
 import java.math.BigDecimal;
 import java.time.LocalDateTime;
 import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
 import java.util.List;
-import org.apache.iceberg.data.GenericRecord;
-import org.apache.iceberg.data.Record;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 
 import static com.netease.arctic.hive.HiveTableTestBase.HIVE_TABLE_SCHEMA;
 
@@ -36,16 +37,16 @@ public class HiveTestRecords {
     GenericRecord record = GenericRecord.create(HIVE_TABLE_SCHEMA);
 
     ImmutableList.Builder<Record> builder = ImmutableList.builder();
-    builder.add(record.copy(ImmutableMap.of("id", 3, "name", "jake",
+    builder.add(record.copy(ImmutableMap.of("id", 3,
         "op_time", LocalDateTime.of(2022, 1, 3, 12, 0, 0),
         "op_time_with_zone", OffsetDateTime.of(
             LocalDateTime.of(2022, 1, 3, 12, 0, 0), ZoneOffset.UTC),
-        "d", new BigDecimal("102"))));
-    builder.add(record.copy(ImmutableMap.of("id", 4, "name", "sam",
+        "d", new BigDecimal("102"), "name", "jake")));
+    builder.add(record.copy(ImmutableMap.of("id", 4,
         "op_time", LocalDateTime.of(2022, 1, 4, 12, 0, 0),
         "op_time_with_zone", OffsetDateTime.of(
             LocalDateTime.of(2022, 1, 4, 12, 0, 0), ZoneOffset.UTC),
-        "d", new BigDecimal("103"))));
+        "d", new BigDecimal("103"), "name", "sam")));
 
     return builder.build();
   }
@@ -54,16 +55,16 @@ public class HiveTestRecords {
     GenericRecord record = GenericRecord.create(HIVE_TABLE_SCHEMA);
 
     ImmutableList.Builder<Record> builder = ImmutableList.builder();
-    builder.add(record.copy(ImmutableMap.of("id", 1, "name", "john",
+    builder.add(record.copy(ImmutableMap.of("id", 1,
         "op_time", LocalDateTime.of(2022, 1, 1, 12, 0, 0),
         "op_time_with_zone", OffsetDateTime.of(
             LocalDateTime.of(2022, 1, 1, 12, 0, 0), ZoneOffset.UTC),
-        "d", new BigDecimal("100"))));
-    builder.add(record.copy(ImmutableMap.of("id", 2, "name", "lily",
+        "d", new BigDecimal("100"), "name", "john")));
+    builder.add(record.copy(ImmutableMap.of("id", 2,
         "op_time", LocalDateTime.of(2022, 1, 2, 12, 0, 0),
         "op_time_with_zone", OffsetDateTime.of(
             LocalDateTime.of(2022, 1, 2, 12, 0, 0), ZoneOffset.UTC),
-        "d", new BigDecimal("101"))));
+        "d", new BigDecimal("101"), "name", "lily")));
     return builder.build();
   }
 
@@ -71,37 +72,37 @@ public class HiveTestRecords {
     GenericRecord record = GenericRecord.create(HIVE_TABLE_SCHEMA);
 
     ImmutableList.Builder<Record> builder = ImmutableList.builder();
-    builder.add(record.copy(ImmutableMap.of("id", 5, "name", "mary",
+    builder.add(record.copy(ImmutableMap.of("id", 5,
         "op_time", LocalDateTime.of(2022, 1, 1, 12, 0, 0),
         "op_time_with_zone", OffsetDateTime.of(
             LocalDateTime.of(2022, 1, 1, 12, 0, 0), ZoneOffset.UTC),
-        "d", new BigDecimal("104"))));
-    builder.add(record.copy(ImmutableMap.of("id", 6, "name", "mack",
+        "d", new BigDecimal("104"), "name", "mary")));
+    builder.add(record.copy(ImmutableMap.of("id", 6,
         "op_time", LocalDateTime.of(2022, 1, 1, 12, 0, 0),
         "op_time_with_zone", OffsetDateTime.of(
             LocalDateTime.of(2022, 1, 1, 12, 0, 0), ZoneOffset.UTC),
-        "d", new BigDecimal("105"))));
+        "d", new BigDecimal("105"), "name", "mack")));
     return builder.build();
   }
 
   public static List<Record> changeDeleteRecords() {
     GenericRecord record = GenericRecord.create(HIVE_TABLE_SCHEMA);
     ImmutableList.Builder<Record> builder = ImmutableList.builder();
-    builder.add(record.copy(ImmutableMap.of("id", 5, "name", "mary",
+    builder.add(record.copy(ImmutableMap.of("id", 5,
         "op_time", LocalDateTime.of(2022, 1, 1, 12, 0, 0),
         "op_time_with_zone", OffsetDateTime.of(
             LocalDateTime.of(2022, 1, 1, 12, 0, 0), ZoneOffset.UTC),
-        "d", new BigDecimal("104"))));
-    builder.add(record.copy(ImmutableMap.of("id", 1, "name", "john",
+        "d", new BigDecimal("104"), "name", "mary")));
+    builder.add(record.copy(ImmutableMap.of("id", 1,
         "op_time", LocalDateTime.of(2022, 1, 1, 12, 0, 0),
         "op_time_with_zone", OffsetDateTime.of(
             LocalDateTime.of(2022, 1, 1, 12, 0, 0), ZoneOffset.UTC),
-        "d", new BigDecimal("100"))));
-    builder.add(record.copy(ImmutableMap.of("id", 3, "name", "jake",
+        "d", new BigDecimal("100"), "name", "john")));
+    builder.add(record.copy(ImmutableMap.of("id", 3,
         "op_time", LocalDateTime.of(2022, 1, 3, 12, 0, 0),
         "op_time_with_zone", OffsetDateTime.of(
             LocalDateTime.of(2022, 1, 3, 12, 0, 0), ZoneOffset.UTC),
-        "d", new BigDecimal("102"))));
+        "d", new BigDecimal("102"), "name", "jake")));
     return builder.build();
   }
 }
diff --git a/hive/src/test/java/com/netease/arctic/hive/op/AutoSyncHiveTest.java b/hive/src/test/java/com/netease/arctic/hive/op/AutoSyncHiveTest.java
new file mode 100644
index 000000000..a0405821d
--- /dev/null
+++ b/hive/src/test/java/com/netease/arctic/hive/op/AutoSyncHiveTest.java
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.hive.op;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import com.netease.arctic.hive.HiveTableProperties;
+import com.netease.arctic.hive.HiveTableTestBase;
+import com.netease.arctic.hive.io.writer.AdaptHiveGenericTaskWriterBuilder;
+import com.netease.arctic.hive.table.HiveLocationKind;
+import com.netease.arctic.hive.utils.HiveMetaSynchronizer;
+import com.netease.arctic.hive.utils.HiveMetaSynchronizerTest;
+import com.netease.arctic.hive.utils.HivePartitionUtil;
+import com.netease.arctic.hive.utils.HiveSchemaUtil;
+import com.netease.arctic.table.ArcticTable;
+import com.netease.arctic.table.LocationKind;
+import com.netease.arctic.table.TableProperties;
+import com.netease.arctic.table.UnkeyedTable;
+import com.netease.arctic.utils.FileUtil;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.Util;
+import org.apache.iceberg.io.TaskWriter;
+import org.apache.iceberg.io.WriteResult;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.thrift.TException;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class AutoSyncHiveTest extends HiveTableTestBase {
+
+
+  @Test
+  public void testAutoSyncHiveSchemaChange() throws TException {
+    testAutoSyncHiveSchemaChange(testHiveTable);
+    testAutoSyncHiveSchemaChange(testKeyedHiveTable);
+    testAutoSyncHiveSchemaChange(testUnPartitionHiveTable);
+    testAutoSyncHiveSchemaChange(testUnPartitionKeyedHiveTable);
+  }
+
+  @Test
+  public void testAutoSyncPartitionedTableHiveDataWrite() throws TException, IOException {
+    testAutoSyncPartitionedTableHiveDataWrite(testHiveTable);
+    testAutoSyncPartitionedTableHiveDataWrite(testKeyedHiveTable);
+  }
+
+  @Test
+  public void testAutoSyncUnpartitionedTableHiveDataWrite() throws TException, IOException {
+    testAutoSyncUnpartitionedTableHiveDataWrite(testUnPartitionHiveTable);
+    testAutoSyncUnpartitionedTableHiveDataWrite(testUnPartitionKeyedHiveTable);
+  }
+
+  private void testAutoSyncUnpartitionedTableHiveDataWrite(ArcticTable testTable) throws TException, IOException {
+    testTable.updateProperties().set(HiveTableProperties.AUTO_SYNC_HIVE_DATA_WRITE, "true").commit();
+    List<DataFile> dataFiles = writeDataFiles(testTable, HiveLocationKind.INSTANT,
+        writeRecords("p1"));
+    UnkeyedTable tableStore;
+    if (testTable.isKeyedTable()) {
+      tableStore = testTable.asKeyedTable().baseTable();
+    } else {
+      tableStore = testTable.asUnkeyedTable();
+    }
+    OverwriteFiles overwriteFiles = tableStore.newOverwrite();
+    dataFiles.forEach(overwriteFiles::addFile);
+    overwriteFiles.commit();
+
+    Table hiveTable = hms.getClient().getTable(testTable.id().getDatabase(), testTable.id().getTableName());
+
+    //test rewrite table
+    Assert.assertEquals(1, dataFiles.size());
+    String dataFilePath = dataFiles.get(0).path().toString();
+    FileSystem fs = Util.getFs(new Path(dataFilePath), new Configuration());
+    fs.rename(new Path(dataFilePath), new Path(dataFilePath + ".bak"));
+    hiveTable.putToParameters("transient_lastDdlTime", "0");
+    hms.getClient().alter_table(testTable.id().getDatabase(), testTable.id().getTableName(), hiveTable);
+    testTable.refresh();
+    Assert.assertEquals(
+        Sets.newHashSet(dataFilePath + ".bak"),
+        listTableFiles(tableStore).stream().map(DataFile::path).collect(Collectors.toSet()));
+  }
+
+  private void testAutoSyncPartitionedTableHiveDataWrite(ArcticTable testTable) throws TException, IOException {
+    testTable.updateProperties().set(HiveTableProperties.AUTO_SYNC_HIVE_DATA_WRITE, "true").commit();
+    Table hiveTable = hms.getClient().getTable(testTable.id().getDatabase(), testTable.id().getTableName());
+    List<DataFile> dataFiles = writeDataFiles(testTable, HiveLocationKind.INSTANT,
+        writeRecords("p1", "p2"));
+    UnkeyedTable tableStore;
+    if (testTable.isKeyedTable()) {
+      tableStore = testTable.asKeyedTable().baseTable();
+    } else {
+      tableStore = testTable.asUnkeyedTable();
+    }
+    OverwriteFiles overwriteFiles = tableStore.newOverwrite();
+    dataFiles.forEach(overwriteFiles::addFile);
+    overwriteFiles.commit();
+
+    //test add new hive partition
+    List<DataFile> newFiles = writeDataFiles(testTable, HiveLocationKind.INSTANT,
+        writeRecords("p3"));
+    dataFiles.addAll(newFiles);
+    Partition newPartition = HivePartitionUtil.newPartition(hiveTable, Lists.newArrayList("p3"),
+        FileUtil.getFileDir(newFiles.get(0).path().toString()), newFiles,
+        (int) (System.currentTimeMillis() / 1000));
+    newPartition.getParameters().remove(HiveTableProperties.ARCTIC_TABLE_FLAG);
+    hms.getClient().add_partition(newPartition);
+    testTable.refresh();
+    Assert.assertEquals(
+        dataFiles.stream().map(DataFile::path).collect(Collectors.toSet()),
+        listTableFiles(tableStore).stream().map(DataFile::path).collect(Collectors.toSet()));
+  }
+
+  private List<DataFile> listTableFiles(UnkeyedTable table) {
+    List<DataFile> dataFiles = Lists.newArrayList();
+    table.newScan().planFiles().forEach(fileScanTask -> dataFiles.add(fileScanTask.file()));
+    return dataFiles;
+  }
+
+  private List<Record> writeRecords(String... partitionValues) {
+    GenericRecord record = GenericRecord.create(HIVE_TABLE_SCHEMA);
+
+    ImmutableList.Builder<Record> builder = ImmutableList.builder();
+    for (String partitionValue : partitionValues) {
+      builder.add(record.copy(ImmutableMap.of("id", 1, "name", partitionValue,
+          "op_time", LocalDateTime.of(2022, 1, 1, 12, 0, 0),
+          "op_time_with_zone", OffsetDateTime.of(
+              LocalDateTime.of(2022, 1, 1, 12, 0, 0), ZoneOffset.UTC),
+          "d", new BigDecimal("100"))));
+    }
+    return builder.build();
+  }
+
+  private List<DataFile> writeDataFiles(ArcticTable table, LocationKind locationKind,
+      List<Record> records) throws IOException {
+    AdaptHiveGenericTaskWriterBuilder builder = AdaptHiveGenericTaskWriterBuilder
+        .builderFor(table)
+        .withTransactionId(1);
+
+    TaskWriter<Record> writer = builder.buildWriter(locationKind);
+    for (Record record: records) {
+      writer.write(record);
+    }
+    WriteResult complete = writer.complete();
+    return Lists.newArrayList(complete.dataFiles());
+  }
+
+  private void testAutoSyncHiveSchemaChange(ArcticTable testTable) throws TException {
+    Table hiveTable = hms.getClient().getTable(testTable.id().getDatabase(), testTable.id().getTableName());
+    List<FieldSchema> hiveFields = Lists.newArrayList();
+    hiveTable.getSd().getCols().add(new FieldSchema("add_column", "bigint", "add column"));
+    hiveFields.addAll(hiveTable.getSd().getCols());
+    hms.getClient().alter_table(testTable.id().getDatabase(), testTable.id().getTableName(), hiveTable);
+    hiveTable = hms.getClient().getTable(testTable.id().getDatabase(), testTable.id().getTableName());
+    testTable.refresh();
+    Assert.assertEquals(hiveTable.getSd().getCols(), HiveSchemaUtil.hiveTableFields(
+        testTable.schema(), testTable.spec()));
+  }
+}
diff --git a/hive/src/test/java/com/netease/arctic/hive/op/TestOverwriteFiles.java b/hive/src/test/java/com/netease/arctic/hive/op/TestOverwriteFiles.java
index 0371ebb13..5fc70e900 100644
--- a/hive/src/test/java/com/netease/arctic/hive/op/TestOverwriteFiles.java
+++ b/hive/src/test/java/com/netease/arctic/hive/op/TestOverwriteFiles.java
@@ -1,5 +1,6 @@
 package com.netease.arctic.hive.op;
 
+import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.HiveTableTestBase;
 import com.netease.arctic.hive.MockDataFileBuilder;
 import com.netease.arctic.hive.exceptions.CannotAlterHiveLocationException;
@@ -7,15 +8,8 @@ import com.netease.arctic.hive.table.UnkeyedHiveTable;
 import com.netease.arctic.op.OverwriteBaseFiles;
 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 java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.function.Predicate;
-import java.util.stream.Collectors;
-
 import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.hadoop.hive.metastore.api.Table;
 import org.apache.iceberg.DataFile;
@@ -29,6 +23,12 @@ import org.apache.thrift.TException;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
 public class TestOverwriteFiles extends HiveTableTestBase {
 
   @Test
@@ -170,10 +170,14 @@ public class TestOverwriteFiles extends HiveTableTestBase {
     Assert.assertTrue(
         "table location to new path",
         hiveTable.getSd().getLocation().endsWith("/test_path/hive_data_location"));
-    Assert.assertEquals("table partition properties is error",
+    Assert.assertEquals("table partition properties hive location is error",
         hiveTable.getSd().getLocation(),
         table.partitionProperty().get(TablePropertyUtil.EMPTY_STRUCT)
-            .get(TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION));
+            .get(HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION));
+    Assert.assertEquals("table partition properties transient_lastDdlTime is error",
+        hiveTable.getParameters().get("transient_lastDdlTime"),
+        table.partitionProperty().get(TablePropertyUtil.EMPTY_STRUCT)
+            .get(HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME));
 
     // =================== test delete all files and add no file to un-partitioned table ===================
     overwriteFiles = table.newOverwrite();
@@ -190,11 +194,14 @@ public class TestOverwriteFiles extends HiveTableTestBase {
     Assert.assertTrue(
         "table location change to hive location",
         hiveLocation.startsWith(table.hiveLocation()));
-    Assert.assertEquals("table partition properties is error",
+    Assert.assertEquals("table partition property hive location is error",
         hiveTable.getSd().getLocation(),
         table.partitionProperty().get(TablePropertyUtil.EMPTY_STRUCT)
-            .get(TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION));
-    System.out.println(hiveLocation);
+            .get(HiveTableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION));
+    Assert.assertEquals("table partition property transient_lastDdlTime is error",
+        hiveTable.getParameters().get("transient_lastDdlTime"),
+        table.partitionProperty().get(TablePropertyUtil.EMPTY_STRUCT)
+            .get(HiveTableProperties.PARTITION_PROPERTIES_KEY_TRANSIENT_TIME));
 
     hiveCatalog.dropTable(identifier, true);
     AMS.handler().getTableCommitMetas().remove(identifier.buildTableIdentifier());
diff --git a/hive/src/test/java/com/netease/arctic/hive/op/TestRewritePartitions.java b/hive/src/test/java/com/netease/arctic/hive/op/TestRewritePartitions.java
index 1a8f127cb..972e02c0b 100644
--- a/hive/src/test/java/com/netease/arctic/hive/op/TestRewritePartitions.java
+++ b/hive/src/test/java/com/netease/arctic/hive/op/TestRewritePartitions.java
@@ -25,8 +25,6 @@ import com.netease.arctic.op.RewritePartitions;
 import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.UnkeyedTable;
 import com.netease.arctic.utils.FileUtil;
-import java.util.List;
-import java.util.Map;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.ReplacePartitions;
 import org.apache.iceberg.Transaction;
@@ -37,6 +35,9 @@ import org.apache.thrift.TException;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.List;
+import java.util.Map;
+
 public class TestRewritePartitions extends HiveTableTestBase {
 
   @Test
diff --git a/hive/src/test/java/com/netease/arctic/hive/utils/HiveMetaSynchronizerTest.java b/hive/src/test/java/com/netease/arctic/hive/utils/HiveMetaSynchronizerTest.java
new file mode 100644
index 000000000..c2abf8b4c
--- /dev/null
+++ b/hive/src/test/java/com/netease/arctic/hive/utils/HiveMetaSynchronizerTest.java
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.hive.utils;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import com.netease.arctic.hive.HMSClient;
+import com.netease.arctic.hive.HiveTableProperties;
+import com.netease.arctic.hive.HiveTableTestBase;
+import com.netease.arctic.hive.io.writer.AdaptHiveGenericTaskWriterBuilder;
+import com.netease.arctic.hive.table.HiveLocationKind;
+import com.netease.arctic.table.ArcticTable;
+import com.netease.arctic.table.LocationKind;
+import com.netease.arctic.table.UnkeyedTable;
+import com.netease.arctic.utils.FileUtil;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.Util;
+import org.apache.iceberg.io.TaskWriter;
+import org.apache.iceberg.io.WriteResult;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.thrift.TException;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class HiveMetaSynchronizerTest extends HiveTableTestBase {
+
+  @Test
+  public void testSyncMetaToHive() throws TException {
+    Table hiveTable = hms.getClient().getTable(HIVE_TABLE_ID.getDatabase(), HIVE_TABLE_ID.getTableName());
+    Assert.assertEquals(HiveSchemaUtil.hiveTableFields(testHiveTable.schema(), testHiveTable.spec()),
+        hiveTable.getSd().getCols());
+    List<FieldSchema> hiveFields = Lists.newArrayList();
+    hiveTable.getSd().getCols().add(new FieldSchema("add_column", "bigint", "add column"));
+    hiveFields.addAll(hiveTable.getSd().getCols());
+    hms.getClient().alter_table(HIVE_TABLE_ID.getDatabase(), HIVE_TABLE_ID.getTableName(), hiveTable);
+    HiveMetaSynchronizer.syncHiveSchemaToArctic(testHiveTable, new TestHMSClient());
+    Assert.assertEquals(hiveFields, hiveTable.getSd().getCols());
+    Assert.assertEquals(hiveFields, HiveSchemaUtil.hiveTableFields(testHiveTable.schema(), testHiveTable.spec()));
+  }
+
+  @Test
+  public void testSyncDataToHive() throws IOException, TException {
+    Table hiveTable = hms.getClient().getTable(HIVE_TABLE_ID.getDatabase(), HIVE_TABLE_ID.getTableName());
+    Assert.assertEquals(0, Iterables.size(testHiveTable.snapshots()));
+    List<DataFile> dataFiles = writeDataFiles(testHiveTable, HiveLocationKind.INSTANT,
+        writeRecords("p1", "p2"));
+    String partition1FilePath = null;
+    String partition2FilePath = null;
+    String partition3FilePath = null;
+    for (DataFile dataFile : dataFiles) {
+      if (dataFile.partition().get(0, String.class).equals("p1")) {
+        partition1FilePath = dataFile.path().toString();
+      } else if (dataFile.partition().get(0, String.class).equals("p2")) {
+        partition2FilePath = dataFile.path().toString();
+      }
+    }
+    Assert.assertNotNull(partition1FilePath);
+    Assert.assertNotNull(partition2FilePath);
+
+    OverwriteFiles overwriteFiles = testHiveTable.newOverwrite();
+    dataFiles.forEach(overwriteFiles::addFile);
+    overwriteFiles.commit();
+    Assert.assertEquals(1, Iterables.size(testHiveTable.snapshots()));
+    List<Partition> partitions = hms.getClient()
+        .listPartitions(HIVE_TABLE_ID.getDatabase(), HIVE_TABLE_ID.getTableName(), Short.MAX_VALUE);
+    Assert.assertEquals(2, partitions.size());
+
+    //test add new hive partition
+    List<DataFile> newFiles = writeDataFiles(testHiveTable, HiveLocationKind.INSTANT,
+        writeRecords("p3"));
+    Assert.assertEquals(1, newFiles.size());
+    partition3FilePath = newFiles.get(0).path().toString();
+    Partition newPartition = HivePartitionUtil.newPartition(hiveTable, Lists.newArrayList("p3"),
+        FileUtil.getFileDir(newFiles.get(0).path().toString()), newFiles,
+        (int) (System.currentTimeMillis() / 1000));
+    newPartition.getParameters().remove(HiveTableProperties.ARCTIC_TABLE_FLAG);
+    hms.getClient().add_partition(newPartition);
+    HiveMetaSynchronizer.syncHiveDataToArctic(testHiveTable, new TestHMSClient());
+    Assert.assertEquals(2, Iterables.size(testHiveTable.snapshots()));
+    partitions = hms.getClient()
+        .listPartitions(HIVE_TABLE_ID.getDatabase(), HIVE_TABLE_ID.getTableName(), Short.MAX_VALUE);
+    Assert.assertEquals(3, partitions.size());
+    Assert.assertEquals(Sets.newHashSet(partition1FilePath, partition2FilePath, partition3FilePath),
+        listTableFiles(testHiveTable).stream().map(DataFile::path).collect(Collectors.toSet()));
+
+    //test drop hive partition
+    hms.getClient().dropPartition(HIVE_TABLE_ID.getDatabase(), HIVE_TABLE_ID.getTableName(),
+        Lists.newArrayList("p1"));
+    testHiveTable.io().deleteFile(partition1FilePath);
+    HiveMetaSynchronizer.syncHiveDataToArctic(testHiveTable, new TestHMSClient());
+    Assert.assertEquals(3, Iterables.size(testHiveTable.snapshots()));
+    partitions = hms.getClient()
+        .listPartitions(HIVE_TABLE_ID.getDatabase(), HIVE_TABLE_ID.getTableName(), Short.MAX_VALUE);
+    Assert.assertEquals(2, partitions.size());
+    dataFiles.remove(0); // remove p1 file
+    Assert.assertEquals(Sets.newHashSet(partition2FilePath, partition3FilePath),
+        listTableFiles(testHiveTable).stream().map(DataFile::path).collect(Collectors.toSet()));
+
+    //test rewrite hive partition
+    FileSystem fs = Util.getFs(new Path(partition2FilePath), new Configuration());
+    fs.rename(new Path(partition2FilePath), new Path(partition2FilePath + ".bak"));
+    Partition hivePartition2 = hms.getClient().getPartition(HIVE_TABLE_ID.getDatabase(), HIVE_TABLE_ID.getTableName(),
+        Lists.newArrayList("p2"));
+    hivePartition2.putToParameters("transient_lastDdlTime", String.valueOf(1000));
+    hms.getClient().alter_partition(HIVE_TABLE_ID.getDatabase(), HIVE_TABLE_ID.getTableName(), hivePartition2);
+    HiveMetaSynchronizer.syncHiveDataToArctic(testHiveTable, new TestHMSClient());
+    Assert.assertEquals(4, Iterables.size(testHiveTable.snapshots()));
+    partitions = hms.getClient()
+        .listPartitions(HIVE_TABLE_ID.getDatabase(), HIVE_TABLE_ID.getTableName(), Short.MAX_VALUE);
+    Assert.assertEquals(2, partitions.size());
+    partition2FilePath = partition2FilePath + ".bak";
+    Assert.assertEquals(Sets.newHashSet(partition2FilePath, partition3FilePath),
+        listTableFiles(testHiveTable).stream().map(DataFile::path).collect(Collectors.toSet()));
+  }
+
+  private List<DataFile> listTableFiles(UnkeyedTable table) {
+    List<DataFile> dataFiles = Lists.newArrayList();
+    table.newScan().planFiles().forEach(fileScanTask -> dataFiles.add(fileScanTask.file()));
+    return dataFiles;
+  }
+
+  private List<Record> writeRecords(String... partitionValues) {
+    GenericRecord record = GenericRecord.create(HIVE_TABLE_SCHEMA);
+
+    ImmutableList.Builder<Record> builder = ImmutableList.builder();
+    for (String partitionValue : partitionValues) {
+      builder.add(record.copy(ImmutableMap.of("id", 1, "name", partitionValue,
+          "op_time", LocalDateTime.of(2022, 1, 1, 12, 0, 0),
+          "op_time_with_zone", OffsetDateTime.of(
+              LocalDateTime.of(2022, 1, 1, 12, 0, 0), ZoneOffset.UTC),
+          "d", new BigDecimal("100"))));
+    }
+    return builder.build();
+  }
+
+  private List<DataFile> writeDataFiles(ArcticTable table, LocationKind locationKind,
+      List<Record> records) throws IOException {
+    AdaptHiveGenericTaskWriterBuilder builder = AdaptHiveGenericTaskWriterBuilder
+        .builderFor(table)
+        .withTransactionId(1);
+
+    TaskWriter<Record> writer = builder.buildWriter(locationKind);
+    for (Record record: records) {
+      writer.write(record);
+    }
+    WriteResult complete = writer.complete();
+    return Lists.newArrayList(complete.dataFiles());
+  }
+
+  private static class TestHMSClient implements HMSClient {
+
+    @Override
+    public <R> R run(Action<R, HiveMetaStoreClient, TException> action) throws TException, InterruptedException {
+      return action.run(hms.getClient());
+    }
+  }
+}
diff --git a/trino/src/test/java/com/netease/arctic/trino/arctic/TestHiveTableBaseForTrino.java b/trino/src/test/java/com/netease/arctic/trino/arctic/TestHiveTableBaseForTrino.java
index 8cff3cb20..a5dcd0aa1 100644
--- a/trino/src/test/java/com/netease/arctic/trino/arctic/TestHiveTableBaseForTrino.java
+++ b/trino/src/test/java/com/netease/arctic/trino/arctic/TestHiveTableBaseForTrino.java
@@ -66,7 +66,7 @@ public abstract class TestHiveTableBaseForTrino extends TableTestBaseForTrino {
 
   protected static final TemporaryFolder tempFolder = new TemporaryFolder();
 
-  protected static HMSMockServer hms  ;
+  protected static HMSMockServer hms;
 
   protected static final TableIdentifier HIVE_TABLE_ID =
       TableIdentifier.of(HIVE_CATALOG_NAME, HIVE_DB_NAME, "test_hive_table");
@@ -80,10 +80,10 @@ public abstract class TestHiveTableBaseForTrino extends TableTestBaseForTrino {
 
   public static final Schema HIVE_TABLE_SCHEMA = new Schema(
       Types.NestedField.required(1, "id", Types.IntegerType.get()),
-      Types.NestedField.required(2, "name", Types.StringType.get()),
-      Types.NestedField.required(3, "op_time", Types.TimestampType.withoutZone()),
-      Types.NestedField.required(4, "op_time_with_zone", Types.TimestampType.withZone()),
-      Types.NestedField.required(5, "d", Types.DecimalType.of(3, 0))
+      Types.NestedField.required(2, "op_time", Types.TimestampType.withoutZone()),
+      Types.NestedField.required(3, "op_time_with_zone", Types.TimestampType.withZone()),
+      Types.NestedField.required(4, "d", Types.DecimalType.of(3, 0)),
+      Types.NestedField.required(5, "name", Types.StringType.get())
   );
 
   protected static final PartitionSpec HIVE_SPEC =