Repository: NetEase/arctic
Pull request #247 (2022-08-29 15:10:19)

[Arctic-43][Hive]Implement hive synchronizer
<!--
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?
Impelement #43. 

## Brief change log

  - *Impelement Hive Meta Synchronizer*
  - *Add table properties `base.hive.auto-sync-schema-change` to config if synchronizing hive schema change to arctic automatically*
  - *Add table properties `base.hive.auto-sync-data-write` to config if synchronizing hive data write to arctic automatically*

## How was this patch tested?
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible

## Documentation

  - Does this pull request introduces a new feature? (yes)
  - If yes, how is the feature documented? (docs )



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/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/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/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/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/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.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.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 =