Repository: NetEase/arctic
Pull request #211 (2022-08-23 06:54:42)

[Arctic-43][Hive]Impelement hive meta synchronizer
Implement #43 

To support update schema and write datas in hive way, synchronize meta and data change from hive metastore to iceberg table.

**The brief change logs:**

-  Add partition properties get/set method to unkeyed table
- Set hive location into partition proerties


Patch:
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ServerTableProperties.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ServerTableProperties.java
deleted file mode 100644
index d72c59a36..000000000
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ServerTableProperties.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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.config;
-
-import com.netease.arctic.table.TableProperties;
-
-import java.util.HashSet;
-import java.util.Set;
-
-public class ServerTableProperties {
-
-  /**
-   * Hidden properties which not shown on ams
-   */
-  public static final Set<String> HIDDEN_EXPOSED = new HashSet<>();
-
-  static {
-    HIDDEN_EXPOSED.add(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID);
-    HIDDEN_EXPOSED.add(TableProperties.LOCATION);
-    HIDDEN_EXPOSED.add(TableProperties.TABLE_CREATE_TIME);
-  }
-}
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/MetaService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/MetaService.java
index 29421eca7..97248a59a 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/MetaService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/MetaService.java
@@ -18,7 +18,6 @@
 
 package com.netease.arctic.ams.server.service;
 
-import com.netease.arctic.ams.server.config.ServerTableProperties;
 import com.netease.arctic.ams.server.model.AMSColumnInfo;
 import com.netease.arctic.ams.server.model.AMSPartitionField;
 import com.netease.arctic.ams.server.model.ServerTableMeta;
@@ -58,7 +57,7 @@ public class MetaService {
     serverTableMeta.setCreateTime(PropertyUtil.propertyAsLong(properties, TableProperties.TABLE_CREATE_TIME,
             TableProperties.TABLE_CREATE_TIME_DEFAULT));
 
-    ServerTableProperties.HIDDEN_EXPOSED.forEach(serverTableMeta.getProperties()::remove);
+    TableProperties.PROTECTED_PROPERTIES.forEach(serverTableMeta.getProperties()::remove);
 
     serverTableMeta.setBaseLocation(at.location());
     serverTableMeta.setPartitionColumnList(at
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 ee2d34b6b..dfd600933 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
@@ -13,6 +13,7 @@ import com.netease.arctic.ams.server.utils.TableMetadataUtil;
 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;
@@ -51,11 +52,6 @@ public class DDLTracerService extends IJDBCService {
   private static final String DOC = " COMMENT '%s'";
   private static final String TYPE = " TYPE %s ";
 
-  private static final List<String> skipProperties = new ArrayList<>();
-
-  static {
-    skipProperties.add(BASE_TABLE_MAX_TRANSACTION_ID);
-  }
 
   public void commit(TableIdentifier tableIdentifier, SchemaUpdateMeta commitMeta) {
     Long commitTime = System.currentTimeMillis();
@@ -153,7 +149,7 @@ public class DDLTracerService extends IJDBCService {
     StringBuilder unsetPro = new StringBuilder();
     int c = 0;
     for (String oldPro : before.keySet()) {
-      if (skipProperties.contains(oldPro)) {
+      if (TableProperties.PROTECTED_PROPERTIES.contains(oldPro)) {
         continue;
       }
       if (!after.containsKey(oldPro)) {
@@ -167,7 +163,7 @@ public class DDLTracerService extends IJDBCService {
     StringBuilder setPro = new StringBuilder();
     int c1 = 0;
     for (String newPro : after.keySet()) {
-      if (skipProperties.contains(newPro)) {
+      if (TableProperties.PROTECTED_PROPERTIES.contains(newPro)) {
         continue;
       }
       if (!after.get(newPro).equals(before.get(newPro))) {
diff --git a/core/src/main/java/com/netease/arctic/io/reader/BaseIcebergDataReader.java b/core/src/main/java/com/netease/arctic/io/reader/BaseIcebergDataReader.java
index b9e95ef1b..57eabb4c7 100644
--- a/core/src/main/java/com/netease/arctic/io/reader/BaseIcebergDataReader.java
+++ b/core/src/main/java/com/netease/arctic/io/reader/BaseIcebergDataReader.java
@@ -83,7 +83,7 @@ public abstract class BaseIcebergDataReader<T> {
     }
   }
 
-  public CloseableIterable<T> readData(ArcticFileScanTask task) {
+  public CloseableIterable<T> readData(FileScanTask task) {
 
     Map<Integer, ?> idToConstant = DataReaderCommon.getIdToConstant(task, projectedSchema, convertConstant);
 
diff --git a/core/src/main/java/com/netease/arctic/io/reader/DataReaderCommon.java b/core/src/main/java/com/netease/arctic/io/reader/DataReaderCommon.java
index eac073ffe..ac5539f60 100644
--- a/core/src/main/java/com/netease/arctic/io/reader/DataReaderCommon.java
+++ b/core/src/main/java/com/netease/arctic/io/reader/DataReaderCommon.java
@@ -4,6 +4,7 @@ import com.netease.arctic.data.ChangeAction;
 import com.netease.arctic.data.DataFileType;
 import com.netease.arctic.scan.ArcticFileScanTask;
 import com.netease.arctic.table.MetadataColumns;
+import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.TypeUtil;
@@ -46,4 +47,17 @@ public class DataReaderCommon {
     return idToConstant;
   }
 
+  protected static Map<Integer, ?> getIdToConstant(FileScanTask task, Schema projectedSchema,
+                                                   BiFunction<Type, Object, Object> convertConstant) {
+    Schema partitionSchema = TypeUtil.select(projectedSchema, task.spec().identitySourceIds());
+    Map<Integer, Object> idToConstant = new HashMap<>();
+    if (!partitionSchema.columns().isEmpty()) {
+      idToConstant.putAll(PartitionUtil.constantsMap(task, convertConstant));
+    }
+    idToConstant.put(org.apache.iceberg.MetadataColumns.FILE_PATH.fieldId(),
+        convertConstant.apply(Types.StringType.get(), task.file().path().toString()));
+
+    return idToConstant;
+  }
+
 }
diff --git a/core/src/main/java/com/netease/arctic/io/writer/GenericTaskWriters.java b/core/src/main/java/com/netease/arctic/io/writer/GenericTaskWriters.java
index 3f7669d99..32e31ab31 100644
--- a/core/src/main/java/com/netease/arctic/io/writer/GenericTaskWriters.java
+++ b/core/src/main/java/com/netease/arctic/io/writer/GenericTaskWriters.java
@@ -115,7 +115,7 @@ public class GenericTaskWriters {
       long fileSizeBytes = PropertyUtil.propertyAsLong(table.properties(), TableProperties.WRITE_TARGET_FILE_SIZE_BYTES,
           TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT);
       long mask = PropertyUtil.propertyAsLong(table.properties(), TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET,
-          TableProperties.CHANGE_FILE_INDEX_HASH_MOD_BUCKET) - 1;
+          TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET_DEFAULT) - 1;
       Schema changeWriteSchema = SchemaUtil.changeWriteSchema(table.changeTable().schema());
       return new GenericChangeTaskWriter(fileFormat,
           new GenericAppenderFactory(changeWriteSchema, table.spec()),
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 fcf7b0282..b2d960f02 100644
--- a/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java
+++ b/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java
@@ -182,7 +182,8 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
     // step3: set max transaction id
     if (keyedTable.spec().isUnpartitioned()) {
       long maxTransactionId = partitionMaxTxId.get(partitionData);
-      partitionMaxTxId.put(partitionData, Math.max(maxTransactionId, this.maxTransactionId.get(partitionData)));
+      partitionMaxTxId.put(partitionData, Math.max(maxTransactionId,
+          this.maxTransactionId.getOrDefault(partitionData, 0L)));
     } else {
       this.maxTransactionId.forEach((pd, txId) -> {
         if (partitionMaxTxId.containsKey(pd)) {
@@ -200,7 +201,7 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
       return;
     }
     try (CloseableIterable<CombinedScanTask> combinedScanTasks
-        = keyedTable.newScan().filter(deleteExpression).planTasks()) {
+             = keyedTable.newScan().filter(deleteExpression).planTasks()) {
       combinedScanTasks.forEach(combinedTask -> combinedTask.tasks().forEach(
           t -> {
             t.dataTasks().forEach(ft -> deleteFiles.add(ft.file()));
diff --git a/core/src/main/java/com/netease/arctic/op/PartitionPropertiesUpdate.java b/core/src/main/java/com/netease/arctic/op/PartitionPropertiesUpdate.java
new file mode 100644
index 000000000..46e566f67
--- /dev/null
+++ b/core/src/main/java/com/netease/arctic/op/PartitionPropertiesUpdate.java
@@ -0,0 +1,120 @@
+/*
+ * 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.op;
+
+import com.netease.arctic.table.TableProperties;
+import com.netease.arctic.table.UnkeyedTable;
+import com.netease.arctic.utils.TablePropertyUtil;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateProperties;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.StructLikeMap;
+
+import java.util.Map;
+import java.util.Set;
+
+public class PartitionPropertiesUpdate implements UpdatePartitionProperties {
+
+  private final UnkeyedTable table;
+  private final StructLikeMap<Map<String, String>> setProperties;
+  private final StructLikeMap<Set<String>> removeProperties;
+  private final Transaction transaction;
+
+  public PartitionPropertiesUpdate(UnkeyedTable table, Transaction transaction) {
+    this.table = table;
+    this.setProperties = StructLikeMap.create(table.spec().partitionType());
+    this.removeProperties = StructLikeMap.create(table.spec().partitionType());
+    this.transaction = transaction;
+  }
+
+  @Override
+  public PartitionPropertiesUpdate set(
+      StructLike partitionData, String key, String value) {
+    Preconditions.checkNotNull(partitionData, "partition cannot be null");
+    Preconditions.checkNotNull(key, "Key cannot be null");
+    Preconditions.checkNotNull(key, "Value cannot be null");
+    Set<String> removePropertiesForPartition = removeProperties.get(partitionData);
+    Preconditions.checkArgument(removePropertiesForPartition == null ||
+            !removePropertiesForPartition.contains(key),
+        "Cannot remove and update the same key: %s", key);
+    Map<String, String> properties = setProperties.get(partitionData);
+    if (properties == null) {
+      properties = Maps.newHashMap();
+      setProperties.put(partitionData, properties);
+    }
+    properties.put(key, value);
+    return this;
+  }
+
+  @Override
+  public PartitionPropertiesUpdate remove(StructLike partitionData, String key) {
+    Preconditions.checkNotNull(partitionData, "partition cannot be null");
+    Preconditions.checkNotNull(key, "Key cannot be null");
+    Map<String, String> setPropertiesForPartition = setProperties.get(partitionData);
+    Preconditions.checkArgument(setPropertiesForPartition == null ||
+            !setPropertiesForPartition.containsKey(key),
+        "Cannot remove and update the same key: %s", key);
+    Set<String> properties = removeProperties.get(partitionData);
+    if (properties != null) {
+      properties.remove(key);
+      if (properties.isEmpty()) {
+        removeProperties.remove(partitionData);
+      }
+    }
+    return this;
+  }
+
+  @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);
+      if (oldProperties == null) {
+        oldProperties = Maps.newHashMap();
+        partitionProperties.put(partitionData, oldProperties);
+      }
+      oldProperties.putAll(properties);
+    });
+    removeProperties.forEach((partitionData, keys) -> {
+      Map<String, String> oldProperties = partitionProperties.get(partitionData);
+      if (oldProperties != null) {
+        keys.forEach(oldProperties::remove);
+      }
+    });
+    return partitionProperties;
+  }
+
+  @Override
+  public void commit() {
+    StructLikeMap<Map<String, String>> result = apply();
+    UpdateProperties updateProperties;
+    if (transaction == null) {
+      updateProperties = table.updateProperties();
+    } else {
+      updateProperties = transaction.updateProperties();
+    }
+    updateProperties.set(
+        TableProperties.TABLE_PARTITION_PROPERTIES,
+        TablePropertyUtil.encodePartitionProperties(table.spec(), result));
+    updateProperties.commit();
+  }
+}
diff --git a/core/src/main/java/com/netease/arctic/op/UpdatePartitionProperties.java b/core/src/main/java/com/netease/arctic/op/UpdatePartitionProperties.java
new file mode 100644
index 000000000..05b6b1c11
--- /dev/null
+++ b/core/src/main/java/com/netease/arctic/op/UpdatePartitionProperties.java
@@ -0,0 +1,58 @@
+/*
+ * 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.op;
+
+import org.apache.iceberg.PendingUpdate;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.util.StructLikeMap;
+
+import java.util.Map;
+
+/**
+ * API for updating partition properties.
+ * <p>
+ * Apply returns the updated partition properties as a map for validation.
+ * <p>
+ * When committing, these changes will be applied to the current table metadata. Commit conflicts
+ * will be resolved by applying the pending changes to the new table metadata.
+ */
+public interface UpdatePartitionProperties extends PendingUpdate<StructLikeMap<Map<String, String>>> {
+
+  /**
+   * Add a key/value property to the partition.
+   *
+   * @param partitionData partition to add property
+   * @param key a String key
+   * @param value a String value
+   * @return this for method chaining
+   * @throws NullPointerException If either the key or value is null
+   */
+  UpdatePartitionProperties set(StructLike partitionData, String key, String value);
+
+  /**
+   * Remove the given property key from the partition.
+   *
+   * @param partitionData partition to remove property
+   * @param key a String key
+   * @return this for method chaining
+   * @throws NullPointerException If the key is null
+   */
+  UpdatePartitionProperties remove(StructLike partitionData, String key);
+
+}
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 78d1b5807..cf49ecc46 100644
--- a/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
@@ -111,7 +111,7 @@ public class BaseKeyedTable implements KeyedTable {
   public Map<String, String> properties() {
     return baseTable.properties().entrySet()
         .stream()
-        .filter(e -> !TableConstants.HIDDEN_PROPERTIES.contains(e.getKey()))
+        .filter(e -> !TableProperties.PROTECTED_PROPERTIES.contains(e.getKey()))
         .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
   }
 
@@ -230,6 +230,10 @@ 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 f5a598203..c6c71541a 100644
--- a/core/src/main/java/com/netease/arctic/table/BaseUnkeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/BaseUnkeyedTable.java
@@ -20,6 +20,8 @@ package com.netease.arctic.table;
 
 import com.netease.arctic.AmsClient;
 import com.netease.arctic.io.ArcticFileIO;
+import com.netease.arctic.op.PartitionPropertiesUpdate;
+import com.netease.arctic.op.UpdatePartitionProperties;
 import com.netease.arctic.trace.AmsTableTracer;
 import com.netease.arctic.trace.TableTracer;
 import com.netease.arctic.trace.TracedAppendFiles;
@@ -31,6 +33,7 @@ import com.netease.arctic.trace.TracedSchemaUpdate;
 import com.netease.arctic.trace.TracedTransaction;
 import com.netease.arctic.trace.TracedUpdateProperties;
 import com.netease.arctic.trace.TrackerOperations;
+import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.DeleteFiles;
 import org.apache.iceberg.ExpireSnapshots;
@@ -58,6 +61,7 @@ import org.apache.iceberg.UpdateProperties;
 import org.apache.iceberg.UpdateSchema;
 import org.apache.iceberg.encryption.EncryptionManager;
 import org.apache.iceberg.io.LocationProvider;
+import org.apache.iceberg.util.StructLikeMap;
 
 import java.util.List;
 import java.util.Map;
@@ -139,7 +143,7 @@ public class BaseUnkeyedTable implements UnkeyedTable, HasTableOperations {
   public Map<String, String> properties() {
     return icebergTable.properties().entrySet()
         .stream()
-        .filter(e -> !TableConstants.HIDDEN_PROPERTIES.contains(e.getKey()))
+        .filter(e -> !TableProperties.PROTECTED_PROPERTIES.contains(e.getKey()))
         .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
   }
 
@@ -322,4 +326,19 @@ public class BaseUnkeyedTable implements UnkeyedTable, HasTableOperations {
     }
     throw new UnsupportedOperationException();
   }
+
+  @Override
+  public StructLikeMap<Map<String, String>> partitionProperty() {
+    String s = icebergTable.properties().get(TableProperties.TABLE_PARTITION_PROPERTIES);
+    if (s != null) {
+      return TablePropertyUtil.decodePartitionProperties(spec(), s);
+    } else {
+      return StructLikeMap.create(spec().partitionType());
+    }
+  }
+
+  @Override
+  public UpdatePartitionProperties updatePartitionProperties(Transaction transaction) {
+    return new PartitionPropertiesUpdate(this, transaction);
+  }
 }
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 2b7fa7e53..c5872be2c 100644
--- a/core/src/main/java/com/netease/arctic/table/KeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/KeyedTable.java
@@ -94,6 +94,7 @@ public interface KeyedTable extends ArcticTable {
    * get max transactionId of each partition. use {@link #partitionMaxTransactionId()} instead.
    *
    * @return map of max transactionId of each partition
+   * @deprecated use {@link UnkeyedTable#partitionProperty()} instead.
    */
   @Deprecated
   Map<String, Long> maxTransactionId();
@@ -102,7 +103,9 @@ public interface KeyedTable extends ArcticTable {
    * get max transactionId of each partition
    *
    * @return map of max transactionId of each partition
+   * @deprecated use {@link UnkeyedTable#partitionProperty()} instead.
    */
+  @Deprecated
   StructLikeMap<Long> partitionMaxTransactionId();
 
   @Override
@@ -118,4 +121,5 @@ public interface KeyedTable extends ArcticTable {
   RewritePartitions newRewritePartitions();
 
   OverwriteBaseFiles newOverwriteBaseFiles();
+
 }
diff --git a/core/src/main/java/com/netease/arctic/table/TableConstants.java b/core/src/main/java/com/netease/arctic/table/TableConstants.java
deleted file mode 100644
index 67d456462..000000000
--- a/core/src/main/java/com/netease/arctic/table/TableConstants.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.table;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-public class TableConstants {
-
-  public static final String PROPERTY_KEY_CREATE_TYPE = "_hidden.arctic.table.create_type";
-  public static final String PROPERTY_VALUE_CREATE_TYPE_NEW = "create";
-  public static final String PROPERTY_VALUE_CREATE_TYPE_UPGRADE = "upgrade";
-
-  public static final Set<String> HIDDEN_PROPERTIES = Collections.unmodifiableSet(
-      Arrays.stream(new String[] {
-          PROPERTY_KEY_CREATE_TYPE
-      }).collect(Collectors.toSet())
-  );
-}
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 edd5adc74..86c628e78 100644
--- a/core/src/main/java/com/netease/arctic/table/TableProperties.java
+++ b/core/src/main/java/com/netease/arctic/table/TableProperties.java
@@ -18,6 +18,9 @@
 
 package com.netease.arctic.table;
 
+import java.util.HashSet;
+import java.util.Set;
+
 /**
  * Reserved Arctic table properties list.
  */
@@ -26,6 +29,22 @@ public class TableProperties {
   private TableProperties() {
   }
 
+  /**
+   * Protected properties which should not be exposed to user.
+   */
+  public static final Set<String> PROTECTED_PROPERTIES = new HashSet<>();
+
+  static {
+    PROTECTED_PROPERTIES.add(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID);
+    PROTECTED_PROPERTIES.add(TableProperties.LOCATION);
+    PROTECTED_PROPERTIES.add(TableProperties.TABLE_CREATE_TIME);
+    PROTECTED_PROPERTIES.add(TableProperties.TABLE_PARTITION_PROPERTIES);
+  }
+
+  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";
@@ -115,7 +134,7 @@ public class TableProperties {
   public static final int BASE_FILE_INDEX_HASH_BUCKET_DEFAULT = 4;
 
   public static final String CHANGE_FILE_INDEX_HASH_BUCKET = "change.file-index.hash-bucket";
-  public static final int CHANGE_FILE_INDEX_HASH_MOD_BUCKET = 4;
+  public static final int CHANGE_FILE_INDEX_HASH_BUCKET_DEFAULT = 4;
 
   public static final String WRITE_TARGET_FILE_SIZE_BYTES =
       org.apache.iceberg.TableProperties.WRITE_TARGET_FILE_SIZE_BYTES;
diff --git a/core/src/main/java/com/netease/arctic/table/UnkeyedTable.java b/core/src/main/java/com/netease/arctic/table/UnkeyedTable.java
index 2786ae09a..dab2a02d1 100644
--- a/core/src/main/java/com/netease/arctic/table/UnkeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/UnkeyedTable.java
@@ -18,13 +18,31 @@
 
 package com.netease.arctic.table;
 
+import com.netease.arctic.op.UpdatePartitionProperties;
 import org.apache.iceberg.Table;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.util.StructLikeMap;
+
+import java.util.Map;
 
 /**
  * Represents an arctic table without keys supported, the same as an {@link Table}
  */
 public interface UnkeyedTable extends ArcticTable, Table {
 
+  /**
+   * Returns the partition properties map.
+   */
+  StructLikeMap<Map<String, String>> partitionProperty();
+
+  /**
+   * Create a new {@link UpdatePartitionProperties} to update partition properties and commit the changes.
+   *
+   * @param transaction the transaction to update partition properties
+   * @return a new {@link UpdatePartitionProperties}
+   */
+  UpdatePartitionProperties updatePartitionProperties(Transaction transaction);
+
   @Override
   default String name() {
     return toString();
diff --git a/core/src/main/java/com/netease/arctic/utils/TablePropertyUtil.java b/core/src/main/java/com/netease/arctic/utils/TablePropertyUtil.java
index b6ab5d9b1..12b9d6d6a 100644
--- a/core/src/main/java/com/netease/arctic/utils/TablePropertyUtil.java
+++ b/core/src/main/java/com/netease/arctic/utils/TablePropertyUtil.java
@@ -23,7 +23,9 @@ import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
 import org.apache.iceberg.StructLike;
+import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.util.StructLikeMap;
 
@@ -35,6 +37,8 @@ import java.util.Map;
  */
 public class TablePropertyUtil {
 
+  public static final StructLike EMPTY_STRUCT = GenericRecord.create(new Schema());
+
   /**
    * Encode max transaction id map of each partition to string.
    *
@@ -82,4 +86,40 @@ public class TablePropertyUtil {
       throw new UnsupportedOperationException("Failed to decode partition max txId ", e);
     }
   }
+
+  public static StructLikeMap<Map<String, String>> decodePartitionProperties(PartitionSpec spec, String value) {
+    try {
+      StructLikeMap<Map<String, String>> results = StructLikeMap.create(spec.partitionType());
+      TypeReference<Map<String, Map<String, String>>> typeReference =
+          new TypeReference<Map<String, Map<String, String>>>() {};
+      Map<String, Map<String, String>> map = new ObjectMapper().readValue(value, typeReference);
+      for (String key : map.keySet()) {
+        if (spec.isUnpartitioned()) {
+          results.put(EMPTY_STRUCT, map.get(key));
+        } else {
+          StructLike partitionData = DataFiles.data(spec, key);
+          results.put(partitionData, map.get(key));
+        }
+      }
+      return results;
+    } catch (JsonProcessingException e) {
+      throw new UnsupportedOperationException("Failed to decode partition max txId ", e);
+    }
+  }
+
+  public static String encodePartitionProperties(PartitionSpec spec,
+      StructLikeMap<Map<String, String>> partitionProperties) {
+    Map<String, Map<String, String>> stringKeyMap = Maps.newHashMap();
+    for (StructLike pd : partitionProperties.keySet()) {
+      String pathLike = spec.partitionToPath(pd);
+      stringKeyMap.put(pathLike, partitionProperties.get(pd));
+    }
+    String value;
+    try {
+      value = new ObjectMapper().writeValueAsString(stringKeyMap);
+    } catch (JsonProcessingException e) {
+      throw new UncheckedIOException(e);
+    }
+    return value;
+  }
 }
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
index 7710ccb04..d1ba0d3c2 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
@@ -284,7 +284,7 @@ public class FlinkSink {
         } else {
           writeFileSplit = PropertyUtil.propertyAsInt(table.properties(),
               TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET,
-              TableProperties.CHANGE_FILE_INDEX_HASH_MOD_BUCKET);
+              TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET_DEFAULT);
         }
 
         return new RoundRobinShuffleRulePolicy(helper, writeOperatorParallelism,
@@ -321,7 +321,7 @@ public class FlinkSink {
 
     int minFileSplitCount = PropertyUtil
         .propertyAsInt(arcticTable.properties(), TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET,
-            TableProperties.CHANGE_FILE_INDEX_HASH_MOD_BUCKET);
+            TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET_DEFAULT);
 
     boolean upsert = arcticTable.isKeyedTable() && PropertyUtil.propertyAsBoolean(arcticTable.properties(),
         TableProperties.UPSERT_ENABLED, TableProperties.UPSERT_ENABLED_DEFAULT);
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
index a6b4389ff..54ef540dd 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkSink.java
@@ -30,9 +30,7 @@ import com.netease.arctic.flink.util.IcebergClassUtil;
 import com.netease.arctic.flink.util.ProxyUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.DistributionHashMode;
-import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.TableProperties;
-import com.netease.arctic.table.UnkeyedTable;
 import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
 import org.apache.flink.streaming.api.datastream.DataStream;
@@ -290,7 +288,7 @@ public class FlinkSink {
           writeFileSplit = PropertyUtil.propertyAsInt(
               table.properties(),
               TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET,
-              TableProperties.CHANGE_FILE_INDEX_HASH_MOD_BUCKET);
+              TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET_DEFAULT);
         }
 
         return new RoundRobinShuffleRulePolicy(helper, writeOperatorParallelism,
@@ -330,7 +328,7 @@ public class FlinkSink {
 
     int minFileSplitCount = PropertyUtil
         .propertyAsInt(arcticTable.properties(), TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET,
-            TableProperties.CHANGE_FILE_INDEX_HASH_MOD_BUCKET);
+            TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET_DEFAULT);
 
     boolean upsert = arcticTable.isKeyedTable() && PropertyUtil.propertyAsBoolean(arcticTable.properties(),
         TableProperties.UPSERT_ENABLED, TableProperties.UPSERT_ENABLED_DEFAULT);
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 502ad330a..9dc100cc4 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
@@ -310,9 +310,9 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
       SerDeInfo serDeInfo = new SerDeInfo();
       switch (format) {
         case PARQUET:
-          storageDescriptor.setOutputFormat("org.apache.hadoop.mapred.FileOutputFormat");
-          storageDescriptor.setInputFormat("org.apache.hadoop.mapred.FileInputFormat");
-          serDeInfo.setSerializationLib("org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe");
+          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);
diff --git a/hive/src/main/java/com/netease/arctic/hive/op/HiveOperationTransaction.java b/hive/src/main/java/com/netease/arctic/hive/op/HiveOperationTransaction.java
index 6a43864e6..8efe38aa0 100644
--- a/hive/src/main/java/com/netease/arctic/hive/op/HiveOperationTransaction.java
+++ b/hive/src/main/java/com/netease/arctic/hive/op/HiveOperationTransaction.java
@@ -55,12 +55,12 @@ import java.util.Map;
 
 public class HiveOperationTransaction implements Transaction {
 
-  private UnkeyedHiveTable unkeyedHiveTable;
-  private Transaction wrapped;
-  private HMSClient client;
-  private TransactionalHMSClient transactionalClient;
+  private final UnkeyedHiveTable unkeyedHiveTable;
+  private final Transaction wrapped;
+  private final HMSClient client;
+  private final TransactionalHMSClient transactionalClient;
 
-  private TransactionalTable transactionalTable;
+  private final TransactionalTable transactionalTable;
 
   public HiveOperationTransaction(
       UnkeyedHiveTable unkeyedHiveTable,
@@ -125,8 +125,7 @@ public class HiveOperationTransaction implements Transaction {
 
   @Override
   public OverwriteFiles newOverwrite() {
-    OverwriteFiles overwriteFiles = wrapped.newOverwrite();
-    return new OverwriteHiveFiles(overwriteFiles, unkeyedHiveTable, client, transactionalClient);
+    return new OverwriteHiveFiles(wrapped, true, unkeyedHiveTable, client, transactionalClient);
   }
 
   @Override
@@ -136,8 +135,7 @@ public class HiveOperationTransaction implements Transaction {
 
   @Override
   public ReplacePartitions newReplacePartitions() {
-    ReplacePartitions r = wrapped.newReplacePartitions();
-    return new ReplaceHivePartitions(r, unkeyedHiveTable, client, transactionalClient);
+    return new ReplaceHivePartitions(wrapped, true, unkeyedHiveTable, client, transactionalClient);
   }
 
   @Override
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 88f36be89..7ab45cd35 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,12 +1,13 @@
 package com.netease.arctic.hive.op;
 
-import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
 import com.netease.arctic.hive.HMSClient;
 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.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;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.metastore.PartitionDropOptions;
@@ -17,11 +18,15 @@ import org.apache.iceberg.DataFile;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.OverwriteFiles;
 import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Transaction;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
+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.Sets;
 import org.apache.iceberg.types.Types;
 import org.apache.thrift.TException;
 import org.slf4j.Logger;
@@ -38,26 +43,31 @@ public class OverwriteHiveFiles implements OverwriteFiles {
 
   private static final Logger LOG = LoggerFactory.getLogger(OverwriteHiveFiles.class);
 
-  final OverwriteFiles delegate;
-  final UnkeyedHiveTable table;
-  final HMSClient hmsClient;
-  final HMSClient transactionClient;
-  final String db;
-  final String tableName;
-
-  final Table hiveTable;
-
-  Expression expr;
-  List<DataFile> addFiles = Lists.newArrayList();
-  List<DataFile> deleteFiles = Lists.newArrayList();
-
-  List<Partition> partitionToDelete = Lists.newArrayList();
-  List<Partition> partitionToCreate = Lists.newArrayList();
-  long txId = -1;
-
-  public OverwriteHiveFiles(
-      OverwriteFiles delegate, UnkeyedHiveTable table, HMSClient hmsClient, HMSClient transactionClient) {
-    this.delegate = delegate;
+  private final Transaction transaction;
+  private final boolean insideTransaction;
+  private final OverwriteFiles delegate;
+  private final UnkeyedHiveTable table;
+  private final HMSClient hmsClient;
+  private final HMSClient transactionClient;
+  private final String db;
+  private final String tableName;
+
+  private final Table hiveTable;
+
+  private Expression expr;
+  private final List<DataFile> addFiles = Lists.newArrayList();
+  private final List<DataFile> deleteFiles = Lists.newArrayList();
+
+  private Map<StructLike, Partition> partitionToDelete = Maps.newHashMap();
+  private Map<StructLike, Partition> partitionToCreate = Maps.newHashMap();
+  private String unpartitionTableLocation;
+  private long txId = -1;
+
+  public OverwriteHiveFiles(Transaction transaction, boolean insideTransaction, UnkeyedHiveTable table,
+      HMSClient hmsClient, HMSClient transactionClient) {
+    this.transaction = transaction;
+    this.insideTransaction = insideTransaction;
+    this.delegate = transaction.newOverwrite();
     this.table = table;
     this.hmsClient = hmsClient;
     this.transactionClient = transactionClient;
@@ -161,12 +171,18 @@ public class OverwriteHiveFiles implements OverwriteFiles {
 
   @Override
   public void commit() {
-    if (!table.spec().isUnpartitioned()) {
+    if (table.spec().isUnpartitioned()) {
+      generateUnpartitionTableLocation();
+    } else {
       this.partitionToDelete = getDeletePartition();
       this.partitionToCreate = getCreatePartition(this.partitionToDelete);
     }
 
     delegate.commit();
+    setHiveLocations();
+    if (!insideTransaction) {
+      transaction.commitTransaction();
+    }
 
     if (table.spec().isUnpartitioned()) {
       commitNonPartitionedTable();
@@ -175,14 +191,33 @@ public class OverwriteHiveFiles implements OverwriteFiles {
     }
   }
 
+  private void setHiveLocations() {
+    UpdatePartitionProperties updatePartitionProperties = table.updatePartitionProperties(transaction);
+    if (table.spec().isUnpartitioned()) {
+      updatePartitionProperties.set(TablePropertyUtil.EMPTY_STRUCT,
+          TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION, unpartitionTableLocation);
+    } else {
+      partitionToDelete.forEach((partitionData, partition) -> {
+        if (!partitionToCreate.containsKey(partitionData)) {
+          updatePartitionProperties.remove(partitionData, TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION);
+        }
+      });
+      partitionToCreate.forEach((partitionData, partition) -> {
+        updatePartitionProperties.set(partitionData, TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION,
+            partition.getSd().getLocation());
+      });
+    }
+    updatePartitionProperties.commit();
+  }
+
   @Override
   public Object updateEvent() {
     return delegate.updateEvent();
   }
 
-  protected List<Partition> getCreatePartition(List<Partition> partitionToDelete) {
+  protected Map<StructLike, Partition> getCreatePartition(Map<StructLike, Partition> partitionToDelete) {
     if (this.addFiles.isEmpty()) {
-      return Lists.newArrayList();
+      return Maps.newHashMap();
     }
 
     Map<String, String> partitionLocationMap = Maps.newHashMap();
@@ -196,42 +231,38 @@ public class OverwriteHiveFiles implements OverwriteFiles {
       String location = FileUtil.getFileDir(d.path().toString());
       partitionLocationMap.put(value, location);
       if (!partitionDataFileMap.containsKey(value)) {
-        partitionDataFileMap.put(value, org.apache.iceberg.relocated.com.google.common.collect.Lists.newArrayList());
+        partitionDataFileMap.put(value, Lists.newArrayList());
       }
       partitionDataFileMap.get(value).add(d);
       partitionValueMap.put(value, partitionValues);
     }
 
-    List<Partition> partitions = Lists.newArrayList();
+    Map<StructLike, Partition> createPartitions = Maps.newHashMap();
     for (String val : partitionValueMap.keySet()) {
       List<String> values = partitionValueMap.get(val);
       String location = partitionLocationMap.get(val);
       List<DataFile> dataFiles = partitionDataFileMap.get(val);
 
+      checkCreatePartitionDataFiles(dataFiles, location);
       Partition p = HivePartitionUtil.newPartition(hiveTable, values, location, dataFiles);
-      partitions.add(p);
-    }
-    for (String val : partitionLocationMap.keySet()) {
-      String partitionLocation = partitionLocationMap.get(val);
-      List<DataFile> dataFiles = partitionDataFileMap.get(val);
-      checkCreatePartitionDataFiles(dataFiles, partitionLocation);
+      createPartitions.put(dataFiles.get(0).partition(), p);
     }
 
-    partitions = filterNewPartitionNonExists(partitions, partitionToDelete);
-    return partitions;
+    createPartitions = filterNewPartitionNonExists(createPartitions, partitionToDelete);
+    return createPartitions;
   }
 
-  protected List<Partition> getDeletePartition() {
+  protected Map<StructLike, Partition> getDeletePartition() {
     if (expr != null) {
       List<DataFile> deleteFilesByExpr = applyDeleteExpr();
       this.deleteFiles.addAll(deleteFilesByExpr);
     }
 
+    Map<StructLike, Partition> deletePartitions = Maps.newHashMap();
     if (deleteFiles.isEmpty()) {
-      return Lists.newArrayList();
+      return deletePartitions;
     }
 
-    List<Partition> partitions = Lists.newArrayList();
     Types.StructType partitionSchema = table.spec().partitionType();
 
     Set<String> checkedPartitionValues = Sets.newHashSet();
@@ -246,7 +277,7 @@ public class OverwriteHiveFiles implements OverwriteFiles {
       }
       try {
         Partition partition = hmsClient.run(c -> c.getPartition(db, tableName, values));
-        partitions.add(partition);
+        deletePartitions.put(dataFile.partition(), partition);
       } catch (NoSuchObjectException e) {
         // pass do nothing
       } catch (TException | InterruptedException e) {
@@ -255,8 +286,8 @@ public class OverwriteHiveFiles implements OverwriteFiles {
       checkedPartitionValues.add(pathValue);
     }
 
-    partitions.forEach(p -> checkPartitionDelete(deleteFileLocations, p));
-    return partitions;
+    deletePartitions.values().forEach(p -> checkPartitionDelete(deleteFileLocations, p));
+    return deletePartitions;
   }
 
   private void checkPartitionDelete(Set<Path> deleteFiles, Partition partition) {
@@ -297,45 +328,47 @@ public class OverwriteHiveFiles implements OverwriteFiles {
    * 1. exists but location is same. skip
    * 2. exists but location is not same, throw {@link CannotAlterHiveLocationException}
    */
-  private List<Partition> filterNewPartitionNonExists(
-      List<Partition> partitionToCreate,
-      List<Partition> partitionToDelete) {
-    List<Partition> partitions = Lists.newArrayList();
+  private Map<StructLike, Partition> filterNewPartitionNonExists(
+      Map<StructLike, Partition> partitionToCreate,
+      Map<StructLike, Partition> partitionToDelete) {
+    Map<StructLike, Partition> partitions = Maps.newHashMap();
     Map<String, Partition> deletePartitionValueMap = Maps.newHashMap();
-    for (Partition p : partitionToDelete) {
+    for (Partition p : partitionToDelete.values()) {
       String partValue = Joiner.on("/").join(p.getValues());
       deletePartitionValueMap.put(partValue, p);
     }
 
-    for (Partition p : partitionToCreate) {
-      String partValue = Joiner.on("/").join(p.getValues());
-      String location = p.getSd().getLocation();
+    for (Map.Entry<StructLike, Partition> entry : partitionToCreate.entrySet()) {
+      String partValue = Joiner.on("/").join(entry.getValue().getValues());
+      String location = entry.getValue().getSd().getLocation();
       Partition toDelete = deletePartitionValueMap.get(partValue);
       if (toDelete != null) {
         String deleteLocation = toDelete.getSd().getLocation();
         // if exists partition to delete with same value
         // make sure location is different
         if (isPathEquals(location, deleteLocation)) {
-          throw new CannotAlterHiveLocationException("can't create new partition: " + partitionToString(p) + ", this " +
+          throw new CannotAlterHiveLocationException("can't create new partition: " +
+              partitionToString(entry.getValue()) + ", this " +
               "partition will be " +
               "delete and re-create with same location");
         } else {
-          partitions.add(p);
+          partitions.put(entry.getKey(), entry.getValue());
           continue;
         }
       }
 
       try {
-        Partition partitionInHive = hmsClient.run(c -> c.getPartition(db, tableName, p.getValues()));
+        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
           continue;
         }
-        throw new CannotAlterHiveLocationException("can't create new partition: " + partitionToString(p) +
+        throw new CannotAlterHiveLocationException("can't create new partition: " +
+            partitionToString(entry.getValue()) +
             ", this partition exists in hive with different location: " + locationInHive);
       } catch (NoSuchObjectException e) {
-        partitions.add(p);
+        partitions.put(entry.getKey(), entry.getValue());
       } catch (TException | InterruptedException e) {
         throw new RuntimeException(e);
       }
@@ -345,7 +378,7 @@ public class OverwriteHiveFiles implements OverwriteFiles {
 
   private void commitPartitionedTable() {
     if (!partitionToDelete.isEmpty()) {
-      for (Partition p : partitionToDelete) {
+      for (Partition p : partitionToDelete.values()) {
         try {
           transactionClient.run(c -> {
             PartitionDropOptions options = PartitionDropOptions.instance()
@@ -366,22 +399,24 @@ public class OverwriteHiveFiles implements OverwriteFiles {
 
     if (!partitionToCreate.isEmpty()) {
       try {
-        transactionClient.run(c -> c.add_partitions(partitionToCreate));
+        transactionClient.run(c -> c.add_partitions(partitionToCreate.values().stream().collect(Collectors.toList())));
       } catch (TException | InterruptedException e) {
         throw new RuntimeException(e);
       }
     }
   }
 
-  private void commitNonPartitionedTable() {
-    String newHiveLocation = null;
+  private void generateUnpartitionTableLocation() {
     if (this.addFiles.isEmpty()) {
-      newHiveLocation = createEmptyLocationForHive();
+      unpartitionTableLocation = createEmptyLocationForHive();
     } else {
-      newHiveLocation = FileUtil.getFileDir(this.addFiles.get(0).path().toString());
+      unpartitionTableLocation = FileUtil.getFileDir(this.addFiles.get(0).path().toString());
     }
+  }
+
+  private void commitNonPartitionedTable() {
 
-    final String finalLocation = newHiveLocation;
+    final String finalLocation = unpartitionTableLocation;
     try {
       transactionClient.run(c -> {
         Table hiveTable = c.getTable(db, tableName);
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 4b159dfbd..6be8e1d66 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
@@ -4,7 +4,10 @@ import com.netease.arctic.hive.HMSClient;
 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.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;
@@ -13,6 +16,8 @@ import org.apache.iceberg.ContentFile;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.ReplacePartitions;
 import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Transaction;
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -22,32 +27,36 @@ 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 {
 
-  final ReplacePartitions delegate;
-  final HMSClient hmsClient;
+  private final Transaction transaction;
+  private final boolean insideTransaction;
+  private final ReplacePartitions delegate;
 
-  final HMSClient transactionalHMSClient;
+  private final HMSClient hmsClient;
+  private final HMSClient transactionalHMSClient;
 
-  final UnkeyedHiveTable table;
+  private final UnkeyedHiveTable table;
+  private final List<DataFile> addFiles = Lists.newArrayList();
+  private final String db;
+  private final String tableName;
+  private final Table hiveTable;
 
-  final List<DataFile> addFiles = Lists.newArrayList();
-
-  final String db;
-  final String tableName;
-
-  final Table hiveTable;
-
-  List<Partition> rewritePartitions = Lists.newArrayList();
-  List<Partition> newPartitions = Lists.newArrayList();
+  private final Map<StructLike, Partition> rewritePartitions = Maps.newHashMap();
+  private final Map<StructLike, Partition> newPartitions = Maps.newHashMap();
+  private String unpartitionTableLocation;
 
   public ReplaceHivePartitions(
-      ReplacePartitions delegate,
+      Transaction transaction,
+      boolean insideTransaction,
       UnkeyedHiveTable table,
       HMSClient client,
       HMSClient transactionalClient) {
-    this.delegate = delegate;
+    this.transaction = transaction;
+    this.insideTransaction = insideTransaction;
+    this.delegate = transaction.newReplacePartitions();
     this.hmsClient = client;
     this.transactionalHMSClient = transactionalClient;
     this.table = table;
@@ -103,12 +112,19 @@ public class ReplaceHivePartitions implements ReplacePartitions {
 
   @Override
   public void commit() {
-    if (!addFiles.isEmpty() && !table.spec().isUnpartitioned()) {
-      applyHivePartitions();
-    }
-
-    delegate.commit();
     if (!addFiles.isEmpty()) {
+      if (table.spec().isUnpartitioned()) {
+        generateUnpartitionTableLocation();
+      } else {
+        applyHivePartitions();
+      }
+
+      delegate.commit();
+      setHiveLocations();
+      if (!insideTransaction) {
+        transaction.commitTransaction();
+      }
+
       if (table.spec().isUnpartitioned()) {
         commitUnPartitionedTable();
       } else {
@@ -117,6 +133,25 @@ public class ReplaceHivePartitions implements ReplacePartitions {
     }
   }
 
+  private void setHiveLocations() {
+    UpdatePartitionProperties updatePartitionProperties = table.updatePartitionProperties(transaction);
+    if (table.spec().isUnpartitioned() && unpartitionTableLocation != null) {
+      updatePartitionProperties.set(
+          TablePropertyUtil.EMPTY_STRUCT,
+          TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION, unpartitionTableLocation);
+    } else {
+      rewritePartitions.forEach((partitionData, partition) -> {
+        updatePartitionProperties.set(partitionData, TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION,
+            partition.getSd().getLocation());
+      });
+      newPartitions.forEach((partitionData, partition) -> {
+        updatePartitionProperties.set(partitionData, TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION,
+            partition.getSd().getLocation());
+      });
+    }
+    updatePartitionProperties.commit();
+  }
+
   @Override
   public Object updateEvent() {
     return delegate.updateEvent();
@@ -151,10 +186,10 @@ public class ReplaceHivePartitions implements ReplacePartitions {
       try {
         Partition partition = hmsClient.run(c -> c.getPartition(db, tableName, values));
         rewriteHivePartitions(partition, location, dataFiles);
-        rewritePartitions.add(partition);
+        rewritePartitions.put(dataFiles.get(0).partition(), partition);
       } catch (NoSuchObjectException e) {
         Partition p = HivePartitionUtil.newPartition(hiveTable, values, location, dataFiles);
-        newPartitions.add(p);
+        newPartitions.put(dataFiles.get(0).partition(), p);
       } catch (TException | InterruptedException e) {
         throw new RuntimeException(e);
       }
@@ -181,10 +216,10 @@ public class ReplaceHivePartitions implements ReplacePartitions {
     try {
       transactionalHMSClient.run(c -> {
         if (!rewritePartitions.isEmpty()) {
-          c.alter_partitions(db, tableName, rewritePartitions);
+          c.alter_partitions(db, tableName, rewritePartitions.values().stream().collect(Collectors.toList()));
         }
         if (!newPartitions.isEmpty()) {
-          c.add_partitions(newPartitions);
+          c.add_partitions(newPartitions.values().stream().collect(Collectors.toList()));
         }
         return 0;
       });
@@ -207,6 +242,10 @@ 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);
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 8e3c0c7c6..401e7cb85 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
@@ -56,8 +56,8 @@ public class UnkeyedHiveTable extends BaseUnkeyedTable implements BaseTable, Sup
 
   @Override
   public ReplacePartitions newReplacePartitions() {
-    ReplacePartitions replacePartitions = super.newReplacePartitions();
-    return new ReplaceHivePartitions(replacePartitions, this, hiveClient, hiveClient);
+    return new ReplaceHivePartitions(icebergTable.newTransaction(),
+        false, this, hiveClient, hiveClient);
   }
 
   @Override
@@ -74,8 +74,7 @@ public class UnkeyedHiveTable extends BaseUnkeyedTable implements BaseTable, Sup
 
   @Override
   public OverwriteHiveFiles newOverwrite() {
-    OverwriteFiles overwriteFiles = super.newOverwrite();
-    return new OverwriteHiveFiles(overwriteFiles, this, hiveClient, hiveClient);
+    return new OverwriteHiveFiles(super.newTransaction(), false,this, hiveClient, hiveClient);
   }
 
   @Override
diff --git a/hive/src/main/java/com/netease/arctic/hive/write/AdaptHiveGenericTaskWriterBuilder.java b/hive/src/main/java/com/netease/arctic/hive/write/AdaptHiveGenericTaskWriterBuilder.java
index 09ec28168..cd44b0f0e 100644
--- a/hive/src/main/java/com/netease/arctic/hive/write/AdaptHiveGenericTaskWriterBuilder.java
+++ b/hive/src/main/java/com/netease/arctic/hive/write/AdaptHiveGenericTaskWriterBuilder.java
@@ -185,7 +185,7 @@ public class AdaptHiveGenericTaskWriterBuilder implements TaskWriterBuilder<Reco
     long fileSizeBytes = PropertyUtil.propertyAsLong(table.properties(), TableProperties.WRITE_TARGET_FILE_SIZE_BYTES,
         TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT);
     long mask = PropertyUtil.propertyAsLong(table.properties(), TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET,
-        TableProperties.CHANGE_FILE_INDEX_HASH_MOD_BUCKET) - 1;
+        TableProperties.CHANGE_FILE_INDEX_HASH_BUCKET_DEFAULT) - 1;
     Schema changeWriteSchema = SchemaUtil.changeWriteSchema(table.changeTable().schema());
     FileAppenderFactory<Record> appenderFactory = HiveTableUtil.isHive(table) ?
         new AdaptHiveGenericAppenderFactory(changeWriteSchema, table.spec()) :
diff --git a/site/docs/ch/spark/spark-ddl.md b/site/docs/ch/spark/spark-ddl.md
index b43a7e0c0..336f5d850 100644
--- a/site/docs/ch/spark/spark-ddl.md
+++ b/site/docs/ch/spark/spark-ddl.md
@@ -77,7 +77,7 @@ AS SELECT ...
 ???+note "REPLACE TABLE ... AS SELECT 语法在当前版本只支持无主键表"
 
 ``` 
-CREATE TABLE arctic_catalog.db.sample
+REPLACE TABLE arctic_catalog.db.sample
 USING arctic
 AS SELECT ...
 ```
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkCatalog.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkCatalog.java
index f08dc5c91..76bab0a77 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkCatalog.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkCatalog.java
@@ -21,8 +21,8 @@ package com.netease.arctic.spark;
 import com.netease.arctic.catalog.ArcticCatalog;
 import com.netease.arctic.catalog.CatalogLoader;
 import com.netease.arctic.spark.table.ArcticSparkChangeTable;
+import com.netease.arctic.spark.table.ArcticSparkTable;
 import com.netease.arctic.table.ArcticTable;
-import com.netease.arctic.table.ChangeTable;
 import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.PrimaryKeySpec;
 import com.netease.arctic.table.TableBuilder;
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/command/MigrateToArcticCommand.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/command/MigrateToArcticCommand.java
index be0a4e9aa..707aa6c82 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/command/MigrateToArcticCommand.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/command/MigrateToArcticCommand.java
@@ -19,7 +19,7 @@
 package com.netease.arctic.spark.command;
 
 import com.netease.arctic.spark.ArcticSparkCatalog;
-import com.netease.arctic.spark.table.ArcticUnkeyedSparkTable;
+import com.netease.arctic.spark.table.ArcticIcebergSparkTable;
 import com.netease.arctic.spark.util.ArcticSparkUtils;
 import com.netease.arctic.table.UnkeyedTable;
 import org.apache.hadoop.conf.Configuration;
@@ -28,7 +28,6 @@ import org.apache.iceberg.DataFile;
 import org.apache.iceberg.MetricsConfig;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
-import org.apache.iceberg.StructLike;
 import org.apache.iceberg.data.TableMigrationUtil;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.hadoop.Util;
@@ -206,9 +205,9 @@ public class MigrateToArcticCommand implements ArcticSparkCommand {
       table = targetCatalog.createTable(
           targetIdentifier, schema, partitions, properties
       );
-      if (table instanceof ArcticUnkeyedSparkTable) {
+      if (table instanceof ArcticIcebergSparkTable) {
         threw = false;
-        return ((ArcticUnkeyedSparkTable) table).table();
+        return ((ArcticIcebergSparkTable) table).table();
       }
       throw new IllegalStateException("target table must be un-keyed table");
     } catch (TableAlreadyExistsException e) {
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/AdaptHiveSparkParquetWriters.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/hive/AdaptHiveSparkParquetWriters.java
similarity index 99%
rename from spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/AdaptHiveSparkParquetWriters.java
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/hive/AdaptHiveSparkParquetWriters.java
index ea619fc3b..2df49489b 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/AdaptHiveSparkParquetWriters.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/hive/AdaptHiveSparkParquetWriters.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package com.netease.arctic.spark.writer;
+package com.netease.arctic.spark.hive;
 
 import org.apache.iceberg.parquet.AdaptHivePrimitiveWriter;
 import org.apache.iceberg.parquet.ParquetValueReaders;
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/ArcticSparkBaseTaskWriter.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/io/ArcticSparkBaseTaskWriter.java
similarity index 98%
rename from spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/ArcticSparkBaseTaskWriter.java
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/io/ArcticSparkBaseTaskWriter.java
index 88649ff63..c856f6db5 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/ArcticSparkBaseTaskWriter.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/io/ArcticSparkBaseTaskWriter.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package com.netease.arctic.spark.writer;
+package com.netease.arctic.spark.io;
 
 import com.netease.arctic.io.ArcticFileIO;
 import com.netease.arctic.io.writer.BaseTaskWriter;
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/ArcticSparkInternalRowAppenderFactory.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/io/InternalRowFileAppenderFactory.java
similarity index 91%
rename from spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/ArcticSparkInternalRowAppenderFactory.java
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/io/InternalRowFileAppenderFactory.java
index c49897380..670fc02e6 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/ArcticSparkInternalRowAppenderFactory.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/io/InternalRowFileAppenderFactory.java
@@ -16,8 +16,9 @@
  * limitations under the License.
  */
 
-package com.netease.arctic.spark.writer;
+package com.netease.arctic.spark.io;
 
+import com.netease.arctic.spark.hive.AdaptHiveSparkParquetWriters;
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.MetricsConfig;
 import org.apache.iceberg.PartitionSpec;
@@ -36,6 +37,7 @@ 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.parquet.ParquetValueWriter;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.spark.SparkSchemaUtil;
 import org.apache.iceberg.spark.data.SparkAvroWriter;
@@ -50,11 +52,11 @@ import java.io.UncheckedIOException;
 import java.util.Map;
 
 /**
- * This ArcticSparkInternalRowAppenderFactory file is forked from
+ * This InternalRowFileAppenderFactory file is forked from
  * org.apache.iceberg.spark.source.SparkAppenderFactory
  * for SparkAppenderFactory is package-private class
  */
-public class ArcticSparkInternalRowAppenderFactory implements FileAppenderFactory<InternalRow> {
+public class InternalRowFileAppenderFactory implements FileAppenderFactory<InternalRow> {
 
   private final Map<String, String> properties;
   private final Schema writeSchema;
@@ -66,18 +68,19 @@ public class ArcticSparkInternalRowAppenderFactory implements FileAppenderFactor
 
   private StructType eqDeleteSparkType = null;
   private StructType posDeleteSparkType = null;
-  private boolean writeHive;
+  private final boolean writeHive;
 
-  public static ArcticSparkInternalRowAppenderFactory.Builder builderFor(
+  public static InternalRowFileAppenderFactory.Builder builderFor(
       Table table,
       Schema writeSchema,
       StructType dsSchema) {
-    return new ArcticSparkInternalRowAppenderFactory.Builder(table, writeSchema, dsSchema);
+    return new InternalRowFileAppenderFactory.Builder(table, writeSchema, dsSchema);
   }
 
-  public ArcticSparkInternalRowAppenderFactory(
+  protected InternalRowFileAppenderFactory(
       Map<String, String> properties, Schema writeSchema, StructType dsSchema, PartitionSpec spec,
-      int[] equalityFieldIds, Schema eqDeleteRowSchema, Schema posDeleteRowSchema, boolean writeHive) {
+      int[] equalityFieldIds, Schema eqDeleteRowSchema, Schema posDeleteRowSchema,
+      boolean writeHive) {
     this.properties = properties;
     this.writeSchema = writeSchema;
     this.dsSchema = dsSchema;
@@ -105,32 +108,32 @@ public class ArcticSparkInternalRowAppenderFactory implements FileAppenderFactor
       this.dsSchema = dsSchema;
     }
 
-    public ArcticSparkInternalRowAppenderFactory.Builder spec(PartitionSpec newSpec) {
+    public InternalRowFileAppenderFactory.Builder spec(PartitionSpec newSpec) {
       this.spec = newSpec;
       return this;
     }
 
-    public ArcticSparkInternalRowAppenderFactory.Builder equalityFieldIds(int[] newEqualityFieldIds) {
+    public InternalRowFileAppenderFactory.Builder equalityFieldIds(int[] newEqualityFieldIds) {
       this.equalityFieldIds = newEqualityFieldIds;
       return this;
     }
 
-    public ArcticSparkInternalRowAppenderFactory.Builder eqDeleteRowSchema(Schema newEqDeleteRowSchema) {
+    public InternalRowFileAppenderFactory.Builder eqDeleteRowSchema(Schema newEqDeleteRowSchema) {
       this.eqDeleteRowSchema = newEqDeleteRowSchema;
       return this;
     }
 
-    public ArcticSparkInternalRowAppenderFactory.Builder posDelRowSchema(Schema newPosDelRowSchema) {
+    public InternalRowFileAppenderFactory.Builder posDelRowSchema(Schema newPosDelRowSchema) {
       this.posDeleteRowSchema = newPosDelRowSchema;
       return this;
     }
 
-    public ArcticSparkInternalRowAppenderFactory.Builder writeHive(boolean writeHive) {
+    public InternalRowFileAppenderFactory.Builder writeHive(boolean writeHive) {
       this.writeHive = writeHive;
       return this;
     }
 
-    public ArcticSparkInternalRowAppenderFactory build() {
+    public InternalRowFileAppenderFactory build() {
       Preconditions.checkNotNull(table, "Table must not be null");
       Preconditions.checkNotNull(writeSchema, "Write Schema must not be null");
       Preconditions.checkNotNull(dsSchema, "DS Schema must not be null");
@@ -143,7 +146,8 @@ public class ArcticSparkInternalRowAppenderFactory implements FileAppenderFactor
             " must be set together");
       }
 
-      return new ArcticSparkInternalRowAppenderFactory(table.properties(),
+      return new InternalRowFileAppenderFactory(
+          table.properties(),
           writeSchema,
           dsSchema,
           spec,
@@ -194,6 +198,7 @@ public class ArcticSparkInternalRowAppenderFactory implements FileAppenderFactor
                 .overwrite()
                 .build();
           }
+
         case AVRO:
           return Avro.write(file)
               .createWriterFunc(ignored -> new SparkAvroWriter(dsSchema))
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/io/TaskWriters.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/io/TaskWriters.java
new file mode 100644
index 000000000..0ceaf5115
--- /dev/null
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/io/TaskWriters.java
@@ -0,0 +1,144 @@
+/*
+ * 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.spark.io;
+
+import com.netease.arctic.hive.table.SupportHive;
+import com.netease.arctic.hive.write.AdaptHiveOutputFileFactory;
+import com.netease.arctic.io.writer.CommonOutputFileFactory;
+import com.netease.arctic.io.writer.OutputFileFactory;
+import com.netease.arctic.table.ArcticTable;
+import com.netease.arctic.table.KeyedTable;
+import com.netease.arctic.table.PrimaryKeySpec;
+import com.netease.arctic.table.TableProperties;
+import com.netease.arctic.table.UnkeyedTable;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.TaskWriter;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.StructType;
+
+import java.util.Locale;
+
+
+public class TaskWriters {
+  private ArcticTable table;
+  private Long transactionId;
+  private int partitionId = 0;
+  private long taskId = 0;
+  private StructType dsSchema;
+
+  private boolean isHiveTable;
+  private FileFormat fileFormat;
+  private long fileSize;
+  private long mask;
+
+  protected TaskWriters(ArcticTable table) {
+    this.table = table;
+    this.isHiveTable = table instanceof SupportHive;
+
+    this.fileFormat = FileFormat.valueOf((table.properties().getOrDefault(
+        TableProperties.BASE_FILE_FORMAT,
+        TableProperties.BASE_FILE_FORMAT_DEFAULT).toUpperCase(Locale.ENGLISH)));
+    this.fileSize = PropertyUtil.propertyAsLong(table.properties(), TableProperties.WRITE_TARGET_FILE_SIZE_BYTES,
+        TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT);
+    this.mask = PropertyUtil.propertyAsLong(table.properties(), TableProperties.BASE_FILE_INDEX_HASH_BUCKET,
+        TableProperties.BASE_FILE_INDEX_HASH_BUCKET_DEFAULT) - 1;
+  }
+
+  public static TaskWriters of(ArcticTable table) {
+    return new TaskWriters(table);
+  }
+
+  public TaskWriters withTransactionId(long transactionId) {
+    this.transactionId = transactionId;
+    return this;
+  }
+
+  public TaskWriters withPartitionId(int partitionId) {
+    this.partitionId = partitionId;
+    return this;
+  }
+
+  public TaskWriters withTaskId(long taskId) {
+    this.taskId = taskId;
+    return this;
+  }
+
+  public TaskWriters withDataSourceSchema(StructType dsSchema) {
+    this.dsSchema = dsSchema;
+    return this;
+  }
+
+  public TaskWriter<InternalRow> newBaseWriter(boolean isOverwrite) {
+    preconditions();
+
+    String baseLocation;
+    EncryptionManager encryptionManager;
+    Schema schema;
+    PrimaryKeySpec primaryKeySpec = null;
+    Table icebergTable;
+
+    if (table.isKeyedTable()) {
+      KeyedTable keyedTable = table.asKeyedTable();
+      baseLocation = keyedTable.baseLocation();
+      encryptionManager = keyedTable.baseTable().encryption();
+      schema = keyedTable.baseTable().schema();
+      primaryKeySpec = keyedTable.primaryKeySpec();
+      icebergTable = keyedTable.baseTable();
+    } else {
+      UnkeyedTable table = this.table.asUnkeyedTable();
+      baseLocation = table.location();
+      encryptionManager = table.encryption();
+      schema = table.schema();
+      icebergTable = table;
+    }
+
+    FileAppenderFactory<InternalRow> appenderFactory = InternalRowFileAppenderFactory
+        .builderFor(icebergTable, schema, dsSchema)
+        .writeHive(isHiveTable)
+        .build();
+
+    OutputFileFactory outputFileFactory = null;
+    if (isHiveTable && isOverwrite) {
+      outputFileFactory = new AdaptHiveOutputFileFactory(
+          ((SupportHive) table).hiveLocation(), table.spec(), fileFormat, table.io(),
+          encryptionManager, partitionId, taskId, transactionId);
+    } else {
+      outputFileFactory = new CommonOutputFileFactory(
+          baseLocation, table.spec(), fileFormat, table.io(),
+          encryptionManager, partitionId, taskId, transactionId);
+    }
+
+    return new ArcticSparkBaseTaskWriter(fileFormat, appenderFactory,
+        outputFileFactory,
+        table.io(), fileSize, mask, schema, table.spec(), primaryKeySpec);
+  }
+
+  private void preconditions() {
+    Preconditions.checkState(transactionId != null, "Transaction id is not set");
+    Preconditions.checkState(partitionId >= 0, "Partition id is not set");
+    Preconditions.checkState(taskId >= 0, "Task id is not set");
+    Preconditions.checkState(dsSchema != null, "Data source schema is not set");
+  }
+}
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/ArcticSparkDataReader.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/ArcticSparkKeyedDataReader.java
similarity index 72%
rename from spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/ArcticSparkDataReader.java
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/ArcticSparkKeyedDataReader.java
index 0efd31541..dafac0758 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/ArcticSparkDataReader.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/ArcticSparkKeyedDataReader.java
@@ -21,6 +21,7 @@ package com.netease.arctic.spark.reader;
 import com.netease.arctic.io.ArcticFileIO;
 import com.netease.arctic.io.reader.BaseArcticDataReader;
 import com.netease.arctic.spark.SparkInternalRowWrapper;
+import com.netease.arctic.spark.util.ArcticSparkUtils;
 import com.netease.arctic.table.PrimaryKeySpec;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.util.Utf8;
@@ -42,9 +43,9 @@ import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.function.Function;
 
-public class ArcticSparkDataReader extends BaseArcticDataReader<InternalRow> {
+public class ArcticSparkKeyedDataReader extends BaseArcticDataReader<InternalRow> {
 
-  public ArcticSparkDataReader(
+  public ArcticSparkKeyedDataReader(
       ArcticFileIO fileIO,
       Schema tableSchema,
       Schema projectedSchema,
@@ -52,35 +53,7 @@ public class ArcticSparkDataReader extends BaseArcticDataReader<InternalRow> {
       String nameMapping,
       boolean caseSensitive) {
     super(fileIO, tableSchema, projectedSchema, primaryKeySpec, nameMapping, caseSensitive,
-        ArcticSparkDataReader::convertConstant, false);
-  }
-
-  protected static Object convertConstant(Type type, Object value) {
-    if (value == null) {
-      return null;
-    }
-
-    switch (type.typeId()) {
-      case DECIMAL:
-        return Decimal.apply((BigDecimal) value);
-      case STRING:
-        if (value instanceof Utf8) {
-          Utf8 utf8 = (Utf8) value;
-          return UTF8String.fromBytes(utf8.getBytes(), 0, utf8.getByteLength());
-        }
-        return UTF8String.fromString(value.toString());
-      case FIXED:
-        if (value instanceof byte[]) {
-          return value;
-        } else if (value instanceof GenericData.Fixed) {
-          return ((GenericData.Fixed) value).bytes();
-        }
-        return ByteBuffers.toByteArray((ByteBuffer) value);
-      case BINARY:
-        return ByteBuffers.toByteArray((ByteBuffer) value);
-      default:
-    }
-    return value;
+        ArcticSparkUtils::convertConstant, false);
   }
 
   @Override
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/ArcticSparkUnkeyedDataReader.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/ArcticSparkUnkeyedDataReader.java
new file mode 100644
index 000000000..a1aacbcfe
--- /dev/null
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/ArcticSparkUnkeyedDataReader.java
@@ -0,0 +1,63 @@
+/*
+ * 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.spark.reader;
+
+import com.netease.arctic.io.ArcticFileIO;
+import com.netease.arctic.io.reader.BaseIcebergDataReader;
+import com.netease.arctic.spark.SparkInternalRowWrapper;
+import com.netease.arctic.spark.util.ArcticSparkUtils;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.parquet.ParquetValueReader;
+import org.apache.iceberg.spark.SparkSchemaUtil;
+import org.apache.iceberg.spark.data.SparkParquetReaders;
+import org.apache.parquet.schema.MessageType;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.types.StructType;
+
+import java.util.Map;
+import java.util.function.Function;
+
+public class ArcticSparkUnkeyedDataReader extends BaseIcebergDataReader<InternalRow> {
+
+  public ArcticSparkUnkeyedDataReader(
+      ArcticFileIO fileIO,
+      Schema tableSchema,
+      Schema projectedSchema,
+      String nameMapping,
+      boolean caseSensitive) {
+    super(fileIO, tableSchema, projectedSchema, nameMapping, caseSensitive,
+        ArcticSparkUtils::convertConstant, false);
+  }
+
+  @Override
+  protected Function<MessageType, ParquetValueReader<?>> getNewReaderFunction(
+      Schema projectedSchema,
+      Map<Integer, ?> idToConstant) {
+    return fileSchema -> SparkParquetReaders.buildReader(projectedSchema, fileSchema, idToConstant);
+  }
+
+  @Override
+  protected Function<Schema, Function<InternalRow, StructLike>> toStructLikeFunction() {
+    return schema -> {
+      final StructType structType = SparkSchemaUtil.convert(schema);
+      return row -> new SparkInternalRowWrapper(structType).wrap(row);
+    };
+  }
+}
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/SparkBatchScan.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/KeyedSparkBatchScan.java
similarity index 95%
rename from spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/SparkBatchScan.java
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/KeyedSparkBatchScan.java
index eb01a88ee..3dbd923dd 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/SparkBatchScan.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/KeyedSparkBatchScan.java
@@ -53,8 +53,8 @@ import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
-class SparkBatchScan implements Scan, Batch, SupportsReportStatistics {
-  private static final Logger LOG = LoggerFactory.getLogger(SparkBatchScan.class);
+public class KeyedSparkBatchScan implements Scan, Batch, SupportsReportStatistics {
+  private static final Logger LOG = LoggerFactory.getLogger(KeyedSparkBatchScan.class);
 
   private final KeyedTable table;
   private final boolean caseSensitive;
@@ -67,7 +67,7 @@ class SparkBatchScan implements Scan, Batch, SupportsReportStatistics {
   private StructType readSchema = null;
   private List<CombinedScanTask> tasks = null;
 
-  SparkBatchScan(
+  KeyedSparkBatchScan(
       KeyedTable table, boolean caseSensitive,
       Schema expectedSchema, List<Expression> filters, CaseInsensitiveStringMap options) {
     this.table = table;
@@ -150,8 +150,8 @@ class SparkBatchScan implements Scan, Batch, SupportsReportStatistics {
       return false;
     }
 
-    SparkBatchScan that = (SparkBatchScan) o;
-    return table.name().equals(that.table.name()) &&
+    KeyedSparkBatchScan that = (KeyedSparkBatchScan) o;
+    return table.id().equals(that.table.id()) &&
         readSchema().equals(that.readSchema()) && // compare Spark schemas to ignore field ids
         filterExpressions.toString().equals(that.filterExpressions.toString()) &&
         Objects.equals(snapshotId, that.snapshotId) &&
@@ -163,7 +163,7 @@ class SparkBatchScan implements Scan, Batch, SupportsReportStatistics {
   @Override
   public int hashCode() {
     return Objects.hash(
-        table.name(), readSchema(), filterExpressions.toString(), snapshotId, startSnapshotId, endSnapshotId,
+        table.id(), readSchema(), filterExpressions.toString(), snapshotId, startSnapshotId, endSnapshotId,
         asOfTimestamp);
   }
 
@@ -216,14 +216,14 @@ class SparkBatchScan implements Scan, Batch, SupportsReportStatistics {
 
   private static class RowReader implements PartitionReader<InternalRow> {
 
-    ArcticSparkDataReader reader;
+    ArcticSparkKeyedDataReader reader;
     Iterator<KeyedTableScanTask> scanTasks;
     KeyedTableScanTask currentScanTask;
     CloseableIterator<InternalRow> currentIterator = CloseableIterator.empty();
     InternalRow current;
 
     RowReader(ArcticInputPartition task) {
-      reader = new ArcticSparkDataReader(
+      reader = new ArcticSparkKeyedDataReader(
           task.io, task.tableSchema, task.expectedSchema, task.keySpec,
           null, task.caseSensitive
       );
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/SparkScanBuilder.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/SparkScanBuilder.java
index 38dbb5d2e..9a523ea9d 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/SparkScanBuilder.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/SparkScanBuilder.java
@@ -18,7 +18,7 @@
 
 package com.netease.arctic.spark.reader;
 
-import com.netease.arctic.table.KeyedTable;
+import com.netease.arctic.table.ArcticTable;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.expressions.Binder;
@@ -41,7 +41,7 @@ import java.util.List;
 public class SparkScanBuilder implements ScanBuilder, SupportsPushDownFilters, SupportsPushDownRequiredColumns {
   private static final Filter[] NO_FILTERS = new Filter[0];
 
-  private final KeyedTable table;
+  private final ArcticTable table;
   private final CaseInsensitiveStringMap options;
 
   private Schema schema = null;
@@ -50,8 +50,7 @@ public class SparkScanBuilder implements ScanBuilder, SupportsPushDownFilters, S
   private List<Expression> filterExpressions = null;
   private Filter[] pushedFilters = NO_FILTERS;
 
-
-  public SparkScanBuilder(SparkSession spark, KeyedTable table, CaseInsensitiveStringMap options) {
+  public SparkScanBuilder(SparkSession spark, ArcticTable table, CaseInsensitiveStringMap options) {
     this.table = table;
     this.options = options;
     this.caseSensitive = Boolean.parseBoolean(spark.conf().get("spark.sql.caseSensitive"));
@@ -116,7 +115,13 @@ public class SparkScanBuilder implements ScanBuilder, SupportsPushDownFilters, S
 
   @Override
   public Scan build() {
-    return new SparkBatchScan(table, caseSensitive, lazySchema(), filterExpressions, options);
+    if (table.isKeyedTable()) {
+      return new KeyedSparkBatchScan(table.asKeyedTable(), caseSensitive, lazySchema(), filterExpressions, options);
+    } else if (table.isUnkeyedTable()) {
+      return new UnkeyedSparkBatchScan(table.asUnkeyedTable(), caseSensitive, lazySchema(), filterExpressions, options);
+    } else {
+      throw new IllegalStateException("Unable to build scan for table: " + table.id().toString() + ", unknown table " +
+          "type");
+    }
   }
-
 }
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/UnkeyedSparkBatchScan.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/UnkeyedSparkBatchScan.java
new file mode 100644
index 000000000..5d7e4e724
--- /dev/null
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/reader/UnkeyedSparkBatchScan.java
@@ -0,0 +1,296 @@
+/*
+ * 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.spark.reader;
+
+import com.netease.arctic.io.ArcticFileIO;
+import com.netease.arctic.spark.util.Stats;
+import com.netease.arctic.table.UnkeyedTable;
+import org.apache.iceberg.CombinedScanTask;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SnapshotSummary;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.spark.Spark3Util;
+import org.apache.iceberg.spark.SparkSchemaUtil;
+import org.apache.iceberg.util.PropertyUtil;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.read.Batch;
+import org.apache.spark.sql.connector.read.InputPartition;
+import org.apache.spark.sql.connector.read.PartitionReader;
+import org.apache.spark.sql.connector.read.PartitionReaderFactory;
+import org.apache.spark.sql.connector.read.Scan;
+import org.apache.spark.sql.connector.read.Statistics;
+import org.apache.spark.sql.connector.read.SupportsReportStatistics;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.io.UncheckedIOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+
+public class UnkeyedSparkBatchScan implements Scan, Batch, SupportsReportStatistics {
+  private static final Logger LOG = LoggerFactory.getLogger(UnkeyedSparkBatchScan.class);
+
+  private final UnkeyedTable table;
+  private final boolean caseSensitive;
+  private final Schema expectedSchema;
+  private final List<Expression> filterExpressions;
+  private final Long snapshotId;
+  private final Long startSnapshotId;
+  private final Long endSnapshotId;
+  private final Long asOfTimestamp;
+  private StructType readSchema = null;
+  private List<CombinedScanTask> tasks = null;
+
+
+  UnkeyedSparkBatchScan(
+      UnkeyedTable table, boolean caseSensitive,
+      Schema expectedSchema, List<Expression> filters, CaseInsensitiveStringMap options) {
+    this.table = table;
+    this.caseSensitive = caseSensitive;
+    this.expectedSchema = expectedSchema;
+    this.filterExpressions = filters;
+    this.snapshotId = Spark3Util.propertyAsLong(options, "snapshot-id", null);
+    this.asOfTimestamp = Spark3Util.propertyAsLong(options, "as-of-timestamp", null);
+
+    if (snapshotId != null && asOfTimestamp != null) {
+      throw new IllegalArgumentException(
+          "Cannot scan using both snapshot-id and as-of-timestamp to select the table snapshot");
+    }
+
+    this.startSnapshotId = Spark3Util.propertyAsLong(options, "start-snapshot-id", null);
+    this.endSnapshotId = Spark3Util.propertyAsLong(options, "end-snapshot-id", null);
+    if (snapshotId != null || asOfTimestamp != null) {
+      if (startSnapshotId != null || endSnapshotId != null) {
+        throw new IllegalArgumentException(
+            "Cannot specify start-snapshot-id and end-snapshot-id to do incremental scan when either snapshot-id or " +
+                "as-of-timestamp is specified");
+      }
+    } else if (startSnapshotId == null && endSnapshotId != null) {
+      throw new IllegalArgumentException("Cannot only specify option end-snapshot-id to do incremental scan");
+    }
+  }
+
+  @Override
+  public InputPartition[] planInputPartitions() {
+    List<CombinedScanTask> scanTasks = tasks();
+    ArcticInputPartition[] readTasks = new ArcticInputPartition[scanTasks.size()];
+    for (int i = 0; i < scanTasks.size(); i++) {
+      readTasks[i] = new ArcticInputPartition(scanTasks.get(i), table, expectedSchema,
+          caseSensitive);
+    }
+    return readTasks;
+  }
+
+  @Override
+  public PartitionReaderFactory createReaderFactory() {
+    return new ReaderFactory();
+  }
+
+  @Override
+  public Statistics estimateStatistics() {
+    if (table.currentSnapshot() == null) {
+      return new Stats(0L, 0L);
+    }
+    if (!table.spec().isUnpartitioned() && filterExpressions.isEmpty()) {
+      LOG.debug("using table metadata to estimate table statistics");
+      long totalRecords =
+          PropertyUtil.propertyAsLong(
+              table.currentSnapshot().summary(),
+              SnapshotSummary.TOTAL_RECORDS_PROP,
+              Long.MAX_VALUE);
+      return new Stats(SparkSchemaUtil.estimateSize(readSchema(), totalRecords), totalRecords);
+    }
+
+    long numRows = 0L;
+
+    for (CombinedScanTask task : tasks()) {
+      for (FileScanTask file : task.files()) {
+        double fractionOfFileScanned = ((double) file.length()) / file.file().fileSizeInBytes();
+        numRows += (fractionOfFileScanned * file.file().recordCount());
+      }
+    }
+    long sizeInBytes = SparkSchemaUtil.estimateSize(readSchema(), numRows);
+    return new Stats(sizeInBytes, numRows);
+  }
+
+  @Override
+  public StructType readSchema() {
+    if (readSchema == null) {
+      this.readSchema = SparkSchemaUtil.convert(expectedSchema);
+    }
+    return readSchema;
+  }
+
+  @Override
+  public Batch toBatch() {
+    return this;
+  }
+
+  private List<CombinedScanTask> tasks() {
+    if (tasks == null) {
+      TableScan scan = table.newScan();
+
+      if (filterExpressions != null) {
+        for (Expression filter : filterExpressions) {
+          scan = scan.filter(filter);
+        }
+      }
+      long startTime = System.currentTimeMillis();
+      LOG.info("mor statistics plan task start");
+      try (CloseableIterable<CombinedScanTask> tasksIterable = scan.planTasks()) {
+        this.tasks = Lists.newArrayList(tasksIterable);
+        LOG.info("mor statistics plan task end, cost time {}, tasks num {}",
+            System.currentTimeMillis() - startTime, tasks.size());
+      } catch (IOException e) {
+        throw new UncheckedIOException("Failed to close table scan: %s", e);
+      }
+    }
+    return tasks;
+  }
+
+  private static class ReaderFactory implements PartitionReaderFactory {
+    @Override
+    public PartitionReader<InternalRow> createReader(InputPartition partition) {
+      if (partition instanceof ArcticInputPartition) {
+        return new RowReader((ArcticInputPartition) partition);
+      } else {
+        throw new UnsupportedOperationException("Incorrect input partition type: " + partition);
+      }
+    }
+  }
+
+  private static class RowReader implements PartitionReader<InternalRow> {
+
+    ArcticSparkUnkeyedDataReader reader;
+    Iterator<FileScanTask> scanTasks;
+    FileScanTask currentScanTask;
+    CloseableIterator<InternalRow> currentIterator = CloseableIterator.empty();
+    InternalRow current;
+
+    RowReader(ArcticInputPartition task) {
+      reader = new ArcticSparkUnkeyedDataReader(
+          task.io, task.tableSchema, task.expectedSchema,
+          null, task.caseSensitive
+      );
+      scanTasks = task.combinedScanTask.files().iterator();
+    }
+
+    @Override
+    public boolean next() throws IOException {
+      while (true) {
+        if (currentIterator.hasNext()) {
+          this.current = currentIterator.next();
+          return true;
+        } else if (scanTasks.hasNext()) {
+          this.currentIterator.close();
+          this.currentScanTask = scanTasks.next();
+          this.currentIterator = reader.readData(this.currentScanTask).iterator();
+        } else {
+          this.currentIterator.close();
+          return false;
+        }
+      }
+    }
+
+    @Override
+    public InternalRow get() {
+      return this.current;
+    }
+
+    @Override
+    public void close() throws IOException {
+      this.currentIterator.close();
+      while (scanTasks.hasNext()) {
+        scanTasks.next();
+      }
+    }
+  }
+
+  private static class ArcticInputPartition implements InputPartition, Serializable {
+    final CombinedScanTask combinedScanTask;
+    final ArcticFileIO io;
+    final boolean caseSensitive;
+    final Schema expectedSchema;
+    final Schema tableSchema;
+
+    ArcticInputPartition(
+        CombinedScanTask combinedScanTask,
+        UnkeyedTable table,
+        Schema expectedSchema,
+        boolean caseSensitive) {
+      this.combinedScanTask = combinedScanTask;
+      this.expectedSchema = expectedSchema;
+      this.tableSchema = table.schema();
+      this.caseSensitive = caseSensitive;
+      this.io = table.io();
+    }
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    UnkeyedSparkBatchScan that = (UnkeyedSparkBatchScan) o;
+    return table.id().equals(that.table.id()) &&
+        readSchema().equals(that.readSchema()) &&
+        filterExpressions.toString().equals(that.filterExpressions.toString()) &&
+        Objects.equals(snapshotId, that.snapshotId) &&
+        Objects.equals(startSnapshotId, that.startSnapshotId) &&
+        Objects.equals(endSnapshotId, that.endSnapshotId) &&
+        Objects.equals(asOfTimestamp, that.asOfTimestamp);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(
+        table.id(), readSchema(), filterExpressions.toString(), snapshotId, startSnapshotId, endSnapshotId,
+        asOfTimestamp);
+  }
+
+  @Override
+  public String description() {
+    String filters = filterExpressions.stream().map(Spark3Util::describe).collect(Collectors.joining(", "));
+    return String.format("%s [filters=%s]", table, filters);
+  }
+
+  @Override
+  public String toString() {
+    return String.format(
+        "IcebergScan(table=%s, type=%s, filters=%s, caseSensitive=%s)",
+        table, expectedSchema.asStruct(), filterExpressions, caseSensitive);
+  }
+
+}
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/ArcticUnkeyedSparkTable.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/ArcticIcebergSparkTable.java
similarity index 92%
rename from spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/ArcticUnkeyedSparkTable.java
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/ArcticIcebergSparkTable.java
index fcfce32d0..ce7ebd4ed 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/ArcticUnkeyedSparkTable.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/ArcticIcebergSparkTable.java
@@ -24,10 +24,10 @@ import org.apache.iceberg.spark.source.SparkTable;
 
 import java.util.Map;
 
-public class ArcticUnkeyedSparkTable extends SparkTable {
+public class ArcticIcebergSparkTable extends SparkTable {
   private final UnkeyedTable unkeyedTable;
 
-  public ArcticUnkeyedSparkTable(UnkeyedTable unkeyedTable, boolean refreshEagerly) {
+  public ArcticIcebergSparkTable(UnkeyedTable unkeyedTable, boolean refreshEagerly) {
     super(unkeyedTable, refreshEagerly);
     this.unkeyedTable = unkeyedTable;
   }
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkTable.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/ArcticSparkTable.java
similarity index 85%
rename from spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkTable.java
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/ArcticSparkTable.java
index ef89e95af..b717dae4c 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkTable.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/ArcticSparkTable.java
@@ -16,14 +16,13 @@
  * limitations under the License.
  */
 
-package com.netease.arctic.spark;
-
+package com.netease.arctic.spark.table;
 
+import com.netease.arctic.hive.table.SupportHive;
 import com.netease.arctic.spark.reader.SparkScanBuilder;
-import com.netease.arctic.spark.table.ArcticUnkeyedSparkTable;
-import com.netease.arctic.spark.writer.SparkWriteBuilder;
+import com.netease.arctic.spark.writer.KeyedSparkWriteBuilder;
+import com.netease.arctic.spark.writer.UnkeyedSparkWriteBuilder;
 import com.netease.arctic.table.ArcticTable;
-import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.TableProperties;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -55,7 +54,7 @@ public class ArcticSparkTable implements org.apache.spark.sql.connector.catalog.
       TableCapability.OVERWRITE_BY_FILTER,
       TableCapability.OVERWRITE_DYNAMIC);
 
-  private final KeyedTable arcticTable;
+  private final ArcticTable arcticTable;
   private final StructType requestedSchema;
   private final boolean refreshEagerly;
   private StructType lazyTableSchema = null;
@@ -63,16 +62,18 @@ public class ArcticSparkTable implements org.apache.spark.sql.connector.catalog.
 
   public static Table ofArcticTable(ArcticTable table) {
     if (table.isUnkeyedTable()) {
-      return new ArcticUnkeyedSparkTable(table.asUnkeyedTable(), false);
+      if (!(table instanceof SupportHive)) {
+        return new ArcticIcebergSparkTable(table.asUnkeyedTable(), false);
+      }
     }
-    return new ArcticSparkTable(table.asKeyedTable(), false);
+    return new ArcticSparkTable(table, false);
   }
 
-  public ArcticSparkTable(KeyedTable arcticTable, boolean refreshEagerly) {
+  public ArcticSparkTable(ArcticTable arcticTable, boolean refreshEagerly) {
     this(arcticTable, null, refreshEagerly);
   }
 
-  public ArcticSparkTable(KeyedTable arcticTable, StructType requestedSchema, boolean refreshEagerly) {
+  public ArcticSparkTable(ArcticTable arcticTable, StructType requestedSchema, boolean refreshEagerly) {
     this.arcticTable = arcticTable;
     this.requestedSchema = requestedSchema;
     this.refreshEagerly = refreshEagerly;
@@ -112,7 +113,6 @@ public class ArcticSparkTable implements org.apache.spark.sql.connector.catalog.
       }
     }
 
-
     return lazyTableSchema;
   }
 
@@ -183,7 +183,12 @@ public class ArcticSparkTable implements org.apache.spark.sql.connector.catalog.
 
   @Override
   public WriteBuilder newWriteBuilder(LogicalWriteInfo info) {
-    return new SparkWriteBuilder(arcticTable, info);
+    if (arcticTable.isKeyedTable()) {
+      return new KeyedSparkWriteBuilder(arcticTable.asKeyedTable(), info);
+    } else if (arcticTable.isUnkeyedTable()) {
+      return new UnkeyedSparkWriteBuilder(arcticTable.asUnkeyedTable(), info);
+    } else {
+      throw new IllegalStateException("un-support type of arctic table: " + arcticTable.getClass().getName());
+    }
   }
-
 }
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/util/ArcticSparkUtils.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/util/ArcticSparkUtils.java
index 0d8018686..9063d1c6e 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/util/ArcticSparkUtils.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/util/ArcticSparkUtils.java
@@ -18,14 +18,18 @@
 
 package com.netease.arctic.spark.util;
 
-import com.netease.arctic.spark.ArcticSparkTable;
+import com.netease.arctic.spark.table.ArcticSparkTable;
 import com.netease.arctic.table.DistributionHashMode;
 import com.netease.arctic.table.PrimaryKeySpec;
 import com.netease.arctic.table.TableProperties;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.util.Utf8;
 import org.apache.iceberg.DistributionMode;
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.spark.Spark3Util;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.util.ByteBuffers;
 import org.apache.iceberg.util.PropertyUtil;
 import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.connector.catalog.CatalogPlugin;
@@ -35,9 +39,13 @@ import org.apache.spark.sql.connector.expressions.Expressions;
 import org.apache.spark.sql.connector.expressions.Transform;
 import org.apache.spark.sql.connector.iceberg.distributions.Distribution;
 import org.apache.spark.sql.connector.iceberg.distributions.Distributions;
+import org.apache.spark.sql.types.Decimal;
+import org.apache.spark.unsafe.types.UTF8String;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -132,4 +140,32 @@ public class ArcticSparkUtils {
         TableProperties.BASE_FILE_INDEX_HASH_BUCKET, TableProperties.BASE_FILE_INDEX_HASH_BUCKET_DEFAULT);
     return Expressions.bucket(numBucket, primaryKeySpec.fieldNames().toArray(new String[0]));
   }
+
+  public static Object convertConstant(Type type, Object value) {
+    if (value == null) {
+      return null;
+    }
+
+    switch (type.typeId()) {
+      case DECIMAL:
+        return Decimal.apply((BigDecimal) value);
+      case STRING:
+        if (value instanceof Utf8) {
+          Utf8 utf8 = (Utf8) value;
+          return UTF8String.fromBytes(utf8.getBytes(), 0, utf8.getByteLength());
+        }
+        return UTF8String.fromString(value.toString());
+      case FIXED:
+        if (value instanceof byte[]) {
+          return value;
+        } else if (value instanceof GenericData.Fixed) {
+          return ((GenericData.Fixed) value).bytes();
+        }
+        return ByteBuffers.toByteArray((ByteBuffer) value);
+      case BINARY:
+        return ByteBuffers.toByteArray((ByteBuffer) value);
+      default:
+    }
+    return value;
+  }
 }
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/ArcticSparkTaskWriterBuilder.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/ArcticSparkTaskWriterBuilder.java
deleted file mode 100644
index d6bb316b7..000000000
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/ArcticSparkTaskWriterBuilder.java
+++ /dev/null
@@ -1,164 +0,0 @@
-/*
- * 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.spark.writer;
-
-import com.netease.arctic.hive.table.HiveLocationKind;
-import com.netease.arctic.hive.table.SupportHive;
-import com.netease.arctic.hive.utils.HiveTableUtil;
-import com.netease.arctic.hive.write.AdaptHiveOperateToTableRelation;
-import com.netease.arctic.hive.write.AdaptHiveOutputFileFactory;
-import com.netease.arctic.io.writer.CommonOutputFileFactory;
-import com.netease.arctic.io.writer.OutputFileFactory;
-import com.netease.arctic.io.writer.SortedPosDeleteWriter;
-import com.netease.arctic.io.writer.TaskWriterBuilder;
-import com.netease.arctic.table.ArcticTable;
-import com.netease.arctic.table.BaseLocationKind;
-import com.netease.arctic.table.ChangeLocationKind;
-import com.netease.arctic.table.KeyedTable;
-import com.netease.arctic.table.LocationKind;
-import com.netease.arctic.table.PrimaryKeySpec;
-import com.netease.arctic.table.TableProperties;
-import com.netease.arctic.table.UnkeyedTable;
-import com.netease.arctic.table.WriteOperationKind;
-import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.StructLike;
-import org.apache.iceberg.Table;
-import org.apache.iceberg.data.GenericAppenderFactory;
-import org.apache.iceberg.encryption.EncryptionManager;
-import org.apache.iceberg.io.FileAppenderFactory;
-import org.apache.iceberg.io.TaskWriter;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.util.PropertyUtil;
-import org.apache.spark.sql.catalyst.InternalRow;
-import org.apache.spark.sql.types.StructType;
-
-import java.util.Locale;
-
-public class ArcticSparkTaskWriterBuilder implements TaskWriterBuilder<InternalRow> {
-
-  private final ArcticTable table;
-  private Long transactionId;
-  private int partitionId = 0;
-  private long taskId = 0;
-  private StructType dsSchema;
-
-  private ArcticSparkTaskWriterBuilder(ArcticTable table) {
-    this.table = table;
-  }
-
-  public ArcticSparkTaskWriterBuilder withTransactionId(long transactionId) {
-    this.transactionId = transactionId;
-    return this;
-  }
-
-  public ArcticSparkTaskWriterBuilder withPartitionId(int partitionId) {
-    this.partitionId = partitionId;
-    return this;
-  }
-
-  public ArcticSparkTaskWriterBuilder withTaskId(long taskId) {
-    this.taskId = taskId;
-    return this;
-  }
-
-  public ArcticSparkTaskWriterBuilder withDataSourceSchema(StructType dsSchema) {
-    this.dsSchema = dsSchema;
-    return this;
-  }
-
-  @Override
-  public TaskWriter<InternalRow> buildWriter(WriteOperationKind writeOperationKind) {
-    LocationKind locationKind = AdaptHiveOperateToTableRelation.INSTANT.getLocationKindsFromOperateKind(
-        table,
-        writeOperationKind);
-    return buildWriter(locationKind);
-  }
-
-  @Override
-  public TaskWriter<InternalRow> buildWriter(LocationKind locationKind) {
-    if (locationKind == ChangeLocationKind.INSTANT) {
-      return buildChangeWriter();
-    } else if (locationKind == BaseLocationKind.INSTANT || locationKind == HiveLocationKind.INSTANT) {
-      return buildBaseWriter(locationKind);
-    } else {
-      throw new IllegalArgumentException("Not support Location Kind:" + locationKind);
-    }
-  }
-
-  public SortedPosDeleteWriter<InternalRow> buildBasePosDeleteWriter(long mask, long index, StructLike partitionKey) {
-    throw new UnsupportedOperationException("UnSupport position delete on spark");
-  }
-
-  private ArcticSparkBaseTaskWriter buildBaseWriter(LocationKind locationKind) {
-    Preconditions.checkNotNull(transactionId);
-    FileFormat fileFormat = FileFormat.valueOf((table.properties().getOrDefault(
-        TableProperties.BASE_FILE_FORMAT,
-        TableProperties.BASE_FILE_FORMAT_DEFAULT).toUpperCase(Locale.ENGLISH)));
-    long fileSizeBytes = PropertyUtil.propertyAsLong(table.properties(), TableProperties.WRITE_TARGET_FILE_SIZE_BYTES,
-        TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT);
-    long mask = PropertyUtil.propertyAsLong(table.properties(), TableProperties.BASE_FILE_INDEX_HASH_BUCKET,
-        TableProperties.BASE_FILE_INDEX_HASH_BUCKET_DEFAULT) - 1;
-
-    String baseLocation;
-    EncryptionManager encryptionManager;
-    Schema schema;
-    PrimaryKeySpec primaryKeySpec = null;
-    Table icebergTable;
-    if (table.isKeyedTable()) {
-      KeyedTable keyedTable = table.asKeyedTable();
-      baseLocation = keyedTable.baseLocation();
-      encryptionManager = keyedTable.baseTable().encryption();
-      schema = keyedTable.baseTable().schema();
-      primaryKeySpec = keyedTable.primaryKeySpec();
-      icebergTable = keyedTable.baseTable();
-    } else {
-      UnkeyedTable table = this.table.asUnkeyedTable();
-      baseLocation = table.location();
-      encryptionManager = table.encryption();
-      schema = table.schema();
-      icebergTable = table;
-    }
-
-    OutputFileFactory outputFileFactory = locationKind == HiveLocationKind.INSTANT ?
-        new AdaptHiveOutputFileFactory(((SupportHive) table).hiveLocation(), table.spec(), fileFormat, table.io(),
-            encryptionManager, partitionId, taskId, transactionId) :
-        new CommonOutputFileFactory(baseLocation, table.spec(), fileFormat, table.io(),
-            encryptionManager, partitionId, taskId, transactionId);
-    FileAppenderFactory<InternalRow> appenderFactory = HiveTableUtil.isHive(table) ?
-        ArcticSparkInternalRowAppenderFactory.builderFor(icebergTable, schema, dsSchema)
-            .writeHive(true)
-            .build() :
-        ArcticSparkInternalRowAppenderFactory.builderFor(icebergTable, schema, dsSchema)
-            .writeHive(false)
-            .build();
-    new GenericAppenderFactory(schema, table.spec());
-    return new ArcticSparkBaseTaskWriter(fileFormat, appenderFactory,
-        outputFileFactory,
-        table.io(), fileSizeBytes, mask, schema, table.spec(), primaryKeySpec);
-  }
-
-  private TaskWriter<InternalRow> buildChangeWriter() {
-    throw new UnsupportedOperationException("Spark UnSupport change write");
-  }
-
-  public static ArcticSparkTaskWriterBuilder buildFor(ArcticTable table) {
-    return new ArcticSparkTaskWriterBuilder(table);
-  }
-}
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkInternalRowWriter.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/InternalRowDataWriter.java
similarity index 89%
rename from spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkInternalRowWriter.java
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/InternalRowDataWriter.java
index 5b1912caa..464149525 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkInternalRowWriter.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/InternalRowDataWriter.java
@@ -26,10 +26,10 @@ import org.apache.spark.sql.connector.write.WriterCommitMessage;
 
 import java.io.IOException;
 
-public class SparkInternalRowWriter implements DataWriter<InternalRow> {
+public class InternalRowDataWriter implements DataWriter<InternalRow> {
   final TaskWriter<InternalRow> writer;
 
-  public SparkInternalRowWriter(TaskWriter<InternalRow> writer) {
+  public InternalRowDataWriter(TaskWriter<InternalRow> writer) {
     this.writer = writer;
   }
 
@@ -41,7 +41,7 @@ public class SparkInternalRowWriter implements DataWriter<InternalRow> {
   @Override
   public WriterCommitMessage commit() throws IOException {
     WriteResult result = writer.complete();
-    return new SparkWrite.TaskCommit(result.dataFiles());
+    return new KeyedSparkBatchWrite.TaskCommit(result.dataFiles());
   }
 
   @Override
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkWrite.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/KeyedSparkBatchWrite.java
similarity index 89%
rename from spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkWrite.java
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/KeyedSparkBatchWrite.java
index fdaea03b5..19fb24702 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkWrite.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/KeyedSparkBatchWrite.java
@@ -20,7 +20,7 @@ package com.netease.arctic.spark.writer;
 
 import com.netease.arctic.op.OverwriteBaseFiles;
 import com.netease.arctic.op.RewritePartitions;
-import com.netease.arctic.table.BaseLocationKind;
+import com.netease.arctic.spark.io.TaskWriters;
 import com.netease.arctic.table.KeyedTable;
 import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.DataFile;
@@ -51,15 +51,13 @@ import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES_DEFAULT;
 import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS;
 import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT;
 
-
-
-public class SparkWrite {
+public class KeyedSparkBatchWrite {
   private final KeyedTable table;
   private final StructType dsSchema;
 
   private final long transactionId;
 
-  SparkWrite(KeyedTable table, StructType dsSchema) {
+  KeyedSparkBatchWrite(KeyedTable table, StructType dsSchema) {
     this.table = table;
     this.dsSchema = dsSchema;
     this.transactionId = table.beginTransaction(null);
@@ -78,9 +76,15 @@ public class SparkWrite {
   }
 
   private abstract class BaseBatchWrite implements BatchWrite {
+    private boolean isOverwrite;
+
+    BaseBatchWrite(boolean isOverwrite) {
+      this.isOverwrite = isOverwrite;
+    }
+
     @Override
     public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
-      return new WriterFactory(table, dsSchema, transactionId);
+      return new WriterFactory(table, dsSchema, transactionId, this.isOverwrite);
     }
 
     @Override
@@ -101,6 +105,11 @@ public class SparkWrite {
   }
 
   private class BatchAppend extends BaseBatchWrite {
+
+    BatchAppend() {
+      super(false);
+    }
+
     @Override
     public void commit(WriterCommitMessage[] messages) {
       AppendFiles append = table.baseTable().newAppend();
@@ -114,6 +123,11 @@ public class SparkWrite {
   }
 
   private class DynamicOverwrite extends BaseBatchWrite {
+
+    DynamicOverwrite() {
+      super(true);
+    }
+
     @Override
     public void commit(WriterCommitMessage[] messages) {
       RewritePartitions rewritePartitions = table.newRewritePartitions();
@@ -130,6 +144,7 @@ public class SparkWrite {
     private final Expression overwriteExpr;
 
     private OverwriteByFilter(Expression overwriteExpr) {
+      super(true);
       this.overwriteExpr = overwriteExpr;
     }
 
@@ -152,22 +167,24 @@ public class SparkWrite {
     private final StructType dsSchema;
     private final long transactionId;
 
-    WriterFactory(KeyedTable table, StructType dsSchema, long transactionId) {
+    private boolean isOverwrite;
+
+    WriterFactory(KeyedTable table, StructType dsSchema, long transactionId, boolean isOverwrite) {
       this.table = table;
       this.dsSchema = dsSchema;
       this.transactionId = transactionId;
+      this.isOverwrite = isOverwrite;
     }
 
     @Override
     public DataWriter<InternalRow> createWriter(int partitionId, long taskId) {
-      TaskWriter<InternalRow> writer = ArcticSparkTaskWriterBuilder.buildFor(table)
+      TaskWriter<InternalRow> writer = TaskWriters.of(table)
           .withTransactionId(transactionId)
           .withPartitionId(partitionId)
           .withTaskId(taskId)
           .withDataSourceSchema(dsSchema)
-          .buildWriter(BaseLocationKind.INSTANT);
-
-      return new SparkInternalRowWriter(writer);
+          .newBaseWriter(this.isOverwrite);
+      return new InternalRowDataWriter(writer);
     }
   }
 
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkWriteBuilder.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/KeyedSparkWriteBuilder.java
similarity index 92%
rename from spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkWriteBuilder.java
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/KeyedSparkWriteBuilder.java
index 3b5a60a91..596142912 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkWriteBuilder.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/KeyedSparkWriteBuilder.java
@@ -34,7 +34,7 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap;
 
 import java.util.Locale;
 
-public class SparkWriteBuilder implements WriteBuilder, SupportsDynamicOverwrite, SupportsOverwrite {
+public class KeyedSparkWriteBuilder implements WriteBuilder, SupportsDynamicOverwrite, SupportsOverwrite {
   private final KeyedTable table;
   private final CaseInsensitiveStringMap options;
   private final String overwriteMode;
@@ -44,7 +44,7 @@ public class SparkWriteBuilder implements WriteBuilder, SupportsDynamicOverwrite
 
   private StructType dsSchema = null;
 
-  public SparkWriteBuilder(KeyedTable table, LogicalWriteInfo info) {
+  public KeyedSparkWriteBuilder(KeyedTable table, LogicalWriteInfo info) {
     this.table = table;
     this.options = info.options();
     this.overwriteMode = options.containsKey("overwrite-mode") ?
@@ -75,7 +75,7 @@ public class SparkWriteBuilder implements WriteBuilder, SupportsDynamicOverwrite
 
   @Override
   public BatchWrite buildForBatch() {
-    SparkWrite write = new SparkWrite(table, dsSchema);
+    KeyedSparkBatchWrite write = new KeyedSparkBatchWrite(table, dsSchema);
     if (overwriteByFilter) {
       return write.asOverwriteByFilter(overwriteExpr);
     } else if (overwriteDynamic) {
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/UnkeyedSparkWriteBuilder.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/UnkeyedSparkWriteBuilder.java
new file mode 100644
index 000000000..92a40c4bb
--- /dev/null
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/UnkeyedSparkWriteBuilder.java
@@ -0,0 +1,50 @@
+/*
+ * 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.spark.writer;
+
+import com.netease.arctic.table.UnkeyedTable;
+import org.apache.spark.sql.connector.write.BatchWrite;
+import org.apache.spark.sql.connector.write.LogicalWriteInfo;
+import org.apache.spark.sql.connector.write.SupportsDynamicOverwrite;
+import org.apache.spark.sql.connector.write.SupportsOverwrite;
+import org.apache.spark.sql.connector.write.WriteBuilder;
+import org.apache.spark.sql.sources.Filter;
+
+// TODO: AdaptHiveRead/Writer
+public class UnkeyedSparkWriteBuilder implements WriteBuilder, SupportsDynamicOverwrite, SupportsOverwrite {
+
+  public UnkeyedSparkWriteBuilder(UnkeyedTable table, LogicalWriteInfo info) {
+
+  }
+
+  @Override
+  public WriteBuilder overwriteDynamicPartitions() {
+    return null;
+  }
+
+  @Override
+  public WriteBuilder overwrite(Filter[] filters) {
+    return null;
+  }
+
+  @Override
+  public BatchWrite buildForBatch() {
+    return null;
+  }
+}
diff --git a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/CreateArcticTableLikeExec.scala b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/CreateArcticTableLikeExec.scala
index aa73428f1..3024f2d0b 100644
--- a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/CreateArcticTableLikeExec.scala
+++ b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/CreateArcticTableLikeExec.scala
@@ -1,6 +1,7 @@
 package com.netease.arctic.spark.sql.execution
 
-import com.netease.arctic.spark.{ArcticSparkCatalog, ArcticSparkTable}
+import com.netease.arctic.spark.ArcticSparkCatalog
+import com.netease.arctic.spark.table.ArcticSparkTable
 import com.netease.arctic.table.KeyedTable
 import org.apache.iceberg.spark.Spark3Util
 import org.apache.iceberg.spark.Spark3Util.CatalogAndIdentifier
diff --git a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/DescribeKeyedTableExec.scala b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/DescribeKeyedTableExec.scala
index eae104c11..58aeee9a5 100644
--- a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/DescribeKeyedTableExec.scala
+++ b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/DescribeKeyedTableExec.scala
@@ -18,7 +18,7 @@
 
 package com.netease.arctic.spark.sql.execution
 
-import com.netease.arctic.spark.ArcticSparkTable
+import com.netease.arctic.spark.table.ArcticSparkTable
 import com.netease.arctic.table.KeyedTable
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.encoders.RowEncoder
diff --git a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/optimize/OptimizeWriteRule.scala b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/optimize/OptimizeWriteRule.scala
index 63c535a1f..81a288e1f 100644
--- a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/optimize/OptimizeWriteRule.scala
+++ b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/optimize/OptimizeWriteRule.scala
@@ -18,7 +18,7 @@
 
 package com.netease.arctic.spark.sql.optimize
 
-import com.netease.arctic.spark.ArcticSparkTable
+import com.netease.arctic.spark.table.ArcticSparkTable
 import com.netease.arctic.spark.util.ArcticSparkUtils
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.expressions.{ArcticExpressionUtils, Expression}
diff --git a/trino/src/main/java/com/netease/arctic/trino/ArcticCatalogSupportTableSuffix.java b/trino/src/main/java/com/netease/arctic/trino/ArcticCatalogSupportTableSuffix.java
index a856b9381..63c7a97aa 100644
--- a/trino/src/main/java/com/netease/arctic/trino/ArcticCatalogSupportTableSuffix.java
+++ b/trino/src/main/java/com/netease/arctic/trino/ArcticCatalogSupportTableSuffix.java
@@ -23,6 +23,7 @@ import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.catalog.ArcticCatalog;
 import com.netease.arctic.catalog.BaseArcticCatalog;
 import com.netease.arctic.io.ArcticFileIO;
+import com.netease.arctic.op.UpdatePartitionProperties;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.BaseUnkeyedTable;
 import com.netease.arctic.table.ChangeTable;
@@ -58,6 +59,7 @@ import org.apache.iceberg.UpdateSchema;
 import org.apache.iceberg.encryption.EncryptionManager;
 import org.apache.iceberg.io.LocationProvider;
 import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeMap;
 
 import java.util.List;
 import java.util.Map;
@@ -333,5 +335,15 @@ public class ArcticCatalogSupportTableSuffix implements ArcticCatalog {
     public TableOperations operations() {
       return table.operations();
     }
+
+    @Override
+    public StructLikeMap<Map<String, String>> partitionProperty() {
+      return table.partitionProperty();
+    }
+
+    @Override
+    public UpdatePartitionProperties updatePartitionProperties(Transaction transaction) {
+      return table.updatePartitionProperties(transaction);
+    }
   }
 }


Test patch:
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestDDLTracerService.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestDDLTracerService.java
index 0d7649d55..4feb1a572 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestDDLTracerService.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestDDLTracerService.java
@@ -18,7 +18,6 @@
 
 package com.netease.arctic.ams.server.service;
 
-import com.netease.arctic.ams.api.SchemaUpdateMeta;
 import com.netease.arctic.ams.api.TableIdentifier;
 import com.netease.arctic.ams.server.AmsTestBase;
 import com.netease.arctic.ams.server.ArcticMetaStore;
@@ -32,20 +31,16 @@ import com.netease.arctic.table.PrimaryKeySpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.types.Types;
 import org.junit.Assert;
-import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
 
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-import org.powermock.core.classloader.annotations.PowerMockIgnore;
-import org.powermock.core.classloader.annotations.PrepareForTest;
-import org.powermock.modules.junit4.PowerMockRunner;
-
 import static com.netease.arctic.ams.server.AmsTestBase.AMS_TEST_CATALOG_NAME;
 import static com.netease.arctic.ams.server.AmsTestBase.AMS_TEST_DB_NAME;
 
diff --git a/core/src/test/java/com/netease/arctic/op/UpdatePartitionPropertiesTest.java b/core/src/test/java/com/netease/arctic/op/UpdatePartitionPropertiesTest.java
new file mode 100644
index 000000000..946e0a13b
--- /dev/null
+++ b/core/src/test/java/com/netease/arctic/op/UpdatePartitionPropertiesTest.java
@@ -0,0 +1,59 @@
+/*
+ * 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.op;
+
+import com.netease.arctic.TableTestBase;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.util.StructLikeMap;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Map;
+
+public class UpdatePartitionPropertiesTest extends TableTestBase {
+
+  @Test
+  public void testUpdatePartitionProperties() {
+    StructLikeMap<Map<String, String>> partitionProperties = testTable.partitionProperty();
+    Assert.assertEquals(0, partitionProperties.size());
+    StructLike p0 = GenericRecord.create(SPEC.partitionType());
+    p0.set(0, 1200);
+    testTable.updatePartitionProperties(null).set(p0, "key", "value").commit();
+    partitionProperties = testTable.partitionProperty();
+    Assert.assertEquals(1, partitionProperties.size());
+    Assert.assertEquals("value", partitionProperties.get(p0).get("key"));
+  }
+
+  public void testUpdatePartitionPropertiesInTx() {
+    StructLikeMap<Map<String, String>> partitionProperties = testTable.partitionProperty();
+    Transaction transaction = testTable.newTransaction();
+    Assert.assertEquals(0, partitionProperties.size());
+    StructLike p0 = GenericRecord.create(SPEC.partitionType());
+    p0.set(0, 1200);
+    testTable.updatePartitionProperties(transaction).set(p0, "key", "value").commit();
+    partitionProperties = testTable.partitionProperty();
+    Assert.assertEquals(0, partitionProperties.size());
+    transaction.commitTransaction();
+    partitionProperties = testTable.partitionProperty();
+    Assert.assertEquals(1, partitionProperties.size());
+    Assert.assertEquals("value", partitionProperties.get(p0).get("key"));
+  }
+}
diff --git a/core/src/test/java/com/netease/arctic/utils/TablePropertyUtilTest.java b/core/src/test/java/com/netease/arctic/utils/TablePropertyUtilTest.java
new file mode 100644
index 000000000..0ed2f6fd5
--- /dev/null
+++ b/core/src/test/java/com/netease/arctic/utils/TablePropertyUtilTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.utils;
+
+import com.google.common.collect.Maps;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.StructLikeMap;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Map;
+
+public class TablePropertyUtilTest {
+
+  private static final Schema SCHEMA = new Schema(
+      Types.NestedField.required(1, "id", Types.IntegerType.get()),
+      Types.NestedField.required(2, "name", Types.StringType.get())
+  );
+  private static final PartitionSpec SPEC = PartitionSpec.builderFor(SCHEMA)
+      .identity("name").build();
+
+  private static final String JSON_VALUE = "{\"name=p0\":{\"key1\":\"p0_value1\",\"key0\":\"p0_value0\"}," +
+      "\"name=p1\":{\"key0\":\"p1_value0\"}}";
+
+  private static final StructLikeMap<Map<String, String>> PARTITION_PROPERTIES = buildPartitionProperties();
+
+  @Test
+  public void testEncodePartitionProperties() {
+    Assert.assertEquals(JSON_VALUE, TablePropertyUtil.encodePartitionProperties(SPEC, PARTITION_PROPERTIES));
+  }
+
+  @Test
+  public void testDecodePartitionProperties() {
+    Assert.assertEquals(PARTITION_PROPERTIES, TablePropertyUtil.decodePartitionProperties(SPEC, JSON_VALUE));
+  }
+
+  private static StructLikeMap<Map<String, String>> buildPartitionProperties() {
+    StructLikeMap<Map<String, String>> partitionProperties = StructLikeMap.create(SPEC.partitionType());
+    GenericRecord partition0 = GenericRecord.create(SPEC.partitionType());
+    partition0.set(0, "p0");
+    GenericRecord partition1 = GenericRecord.create(SPEC.partitionType());
+    partition1.set(0, "p1");
+    Map<String, String> partition0Properties = Maps.newHashMap();
+    Map<String, String> partition1Properties = Maps.newHashMap();
+    partition0Properties.put("key0", "p0_value0");
+    partition0Properties.put("key1", "p0_value1");
+    partition1Properties.put("key0", "p1_value0");
+    partitionProperties.put(partition0, partition0Properties);
+    partitionProperties.put(partition1, partition1Properties);
+    return partitionProperties;
+  }
+}
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 80c0dbc40..bfb30df5c 100644
--- a/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java
+++ b/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java
@@ -31,11 +31,17 @@ import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.TableIdentifier;
 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.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.util.StructLikeMap;
 import org.apache.thrift.TException;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.types.Types;
@@ -162,6 +168,10 @@ public class HiveTableTestBase extends TableTestBase {
     return Joiner.on("/").join(nameValues);
   }
 
+  public static StructLike getPartitionData(String partitionPath, PartitionSpec spec) {
+    return DataFiles.data(spec, partitionPath);
+  }
+
   /**
    * assert hive table partition location as expected
    */
@@ -186,6 +196,13 @@ public class HiveTableTestBase extends TableTestBase {
     Assert.assertEquals("expect " + partitionLocations.size() + " partition after first rewrite partition",
         partitionLocations.size(), partitions.size());
 
+    UnkeyedTable unkeyedTable;
+    if (table.isKeyedTable()) {
+      unkeyedTable = table.asKeyedTable().baseTable();
+    } else {
+      unkeyedTable = table.asUnkeyedTable();
+    }
+    StructLikeMap<Map<String, String>> partitionProperties = unkeyedTable.partitionProperty();
     for (Partition p : partitions) {
       String valuePath = getPartitionPath(p.getValues(), table.spec());
       Assert.assertTrue("partition " + valuePath + " is not expected",
@@ -196,6 +213,9 @@ public class HiveTableTestBase extends TableTestBase {
       Assert.assertTrue(
           "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));
     }
   }
 }
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 ea11aada5..0371ebb13 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
@@ -7,6 +7,7 @@ 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;
@@ -14,6 +15,8 @@ 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;
 import org.apache.iceberg.OverwriteFiles;
@@ -167,6 +170,10 @@ 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",
+        hiveTable.getSd().getLocation(),
+        table.partitionProperty().get(TablePropertyUtil.EMPTY_STRUCT)
+            .get(TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION));
 
     // =================== test delete all files and add no file to un-partitioned table ===================
     overwriteFiles = table.newOverwrite();
@@ -183,6 +190,10 @@ public class TestOverwriteFiles extends HiveTableTestBase {
     Assert.assertTrue(
         "table location change to hive location",
         hiveLocation.startsWith(table.hiveLocation()));
+    Assert.assertEquals("table partition properties is error",
+        hiveTable.getSd().getLocation(),
+        table.partitionProperty().get(TablePropertyUtil.EMPTY_STRUCT)
+            .get(TableProperties.PARTITION_PROPERTIES_KEY_HIVE_LOCATION));
     System.out.println(hiveLocation);
 
     hiveCatalog.dropTable(identifier, true);
diff --git a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/ArcticSparkHiveMainTest.java b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/ArcticSparkHiveMainTest.java
index 874d81949..f68e119d0 100644
--- a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/ArcticSparkHiveMainTest.java
+++ b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/ArcticSparkHiveMainTest.java
@@ -15,6 +15,7 @@ import java.io.IOException;
 @Suite.SuiteClasses({
     TestCreateTableDDL.class,
     TestMigrateHiveTable.class,
+    TestHiveTableMergeOnRead.class,
     TestAlterKeyedTable.class,
     TestKeyedTableDDL.class,
     TestKeyedTableDML.class,
diff --git a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
index 2a5c0ea57..5355bad9d 100644
--- a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
+++ b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
@@ -27,19 +27,29 @@ import com.netease.arctic.ams.api.TableMeta;
 import com.netease.arctic.catalog.ArcticCatalog;
 import com.netease.arctic.catalog.CatalogLoader;
 import com.netease.arctic.data.ChangeAction;
-import com.netease.arctic.hive.HMSMockServer;
+import com.netease.arctic.data.DataTreeNode;
+import com.netease.arctic.data.DefaultKeyedFile;
+import com.netease.arctic.hive.write.AdaptHiveGenericTaskWriterBuilder;
 import com.netease.arctic.io.writer.GenericTaskWriters;
+import com.netease.arctic.io.writer.SortedPosDeleteWriter;
+import com.netease.arctic.op.OverwriteBaseFiles;
+import com.netease.arctic.spark.hive.HMSMockServer;
 import com.netease.arctic.spark.hive.HiveCatalogMetaTestUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.KeyedTable;
+import com.netease.arctic.table.LocationKind;
 import com.netease.arctic.table.TableIdentifier;
 import com.netease.arctic.table.UnkeyedTable;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.ContentFile;
 import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.RowDelta;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.data.GenericRecord;
@@ -74,6 +84,7 @@ import java.time.OffsetDateTime;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -108,6 +119,7 @@ public class SparkTestContext extends ExternalResource {
   public static void cleanUpAdditionSparkConfigs() {
     additionSparkConfigs.clear();
   }
+
   @BeforeClass
   public static void startAll() throws IOException {
     int ref = testCount.incrementAndGet();
@@ -130,32 +142,32 @@ public class SparkTestContext extends ExternalResource {
   }
 
   public static void setUpTestDirAndArctic() throws IOException {
-      System.out.println("======================== start AMS  ========================= ");
-      FileUtils.deleteQuietly(testBaseDir);
-      testBaseDir.mkdirs();
+    System.out.println("======================== start AMS  ========================= ");
+    FileUtils.deleteQuietly(testBaseDir);
+    testBaseDir.mkdirs();
 
-      AmsClientPools.cleanAll();
-      if (!ams.isStarted()) {
-        ams.start();
-      }
-      amsUrl = "thrift://127.0.0.1:" + ams.port();
+    AmsClientPools.cleanAll();
+    if (!ams.isStarted()) {
+      ams.start();
+    }
+    amsUrl = "thrift://127.0.0.1:" + ams.port();
 
-      CatalogMeta arctic = CatalogMetaTestUtil.createArcticCatalog(testArcticDir);
-      catalogNameArctic = arctic.getCatalogName();
-      ams.handler().createCatalog(arctic);
+    CatalogMeta arctic = CatalogMetaTestUtil.createArcticCatalog(testArcticDir);
+    catalogNameArctic = arctic.getCatalogName();
+    ams.handler().createCatalog(arctic);
   }
 
   public static void setUpHMS() throws IOException {
     System.out.println("======================== start hive metastore ========================= ");
     hms.start();
-    additionSparkConfigs.put("hive.metastore.uris", "thrift://127.0.0.1:" + hms.getMetastorePort()) ;
+    additionSparkConfigs.put("hive.metastore.uris", "thrift://127.0.0.1:" + hms.getMetastorePort());
     additionSparkConfigs.put("spark.sql.catalogImplementation", "hive");
     additionSparkConfigs.put("spark.sql.hive.metastore.version", "2.3.7");
     //hive.metastore.client.capability.check
     additionSparkConfigs.put("hive.metastore.client.capability.check", "false");
 
     HiveConf entries = hms.hiveConf();
-    CatalogMeta arctic_hive = HiveCatalogMetaTestUtil.createArcticCatalog(testArcticDir,entries);
+    CatalogMeta arctic_hive = HiveCatalogMetaTestUtil.createArcticCatalog(testArcticDir, entries);
     catalogNameHive = arctic_hive.getCatalogName();
     ams.handler().createCatalog(arctic_hive);
   }
@@ -208,7 +220,7 @@ public class SparkTestContext extends ExternalResource {
     hms.stop();
   }
 
-  public static void cleanUpSparkSession(){
+  public static void cleanUpSparkSession() {
     System.out.println("======================== clean up spark session  ========================= ");
     spark.stop();
     spark.close();
@@ -403,7 +415,7 @@ public class SparkTestContext extends ExternalResource {
         primaryKeysBlock = true;
       } else if (StringUtils.startsWith(row[0].toString(), "# ") && primaryKeysBlock) {
         primaryKeysBlock = false;
-      } else if (primaryKeysBlock){
+      } else if (primaryKeysBlock) {
         descPrimaryKeys.add(row[0].toString());
       }
     }
@@ -429,16 +441,16 @@ public class SparkTestContext extends ExternalResource {
         descPrimaryKeys.stream().sorted().distinct().toArray());
   }
 
-  public void assertHiveDesc(List<Object[]> rows, List<String> cols, List<String> partitionKey){
+  public void assertHiveDesc(List<Object[]> rows, List<String> cols, List<String> partitionKey) {
     boolean colsBlock = true;
     boolean partitionBlock = false;
     List<String> descCols = Lists.newArrayList();
     List<String> descPartitionKey = Lists.newArrayList();
-    for (Object[] row : rows){
-      if (StringUtils.startsWith(row[0].toString(), "# Partition Infor")){
+    for (Object[] row : rows) {
+      if (StringUtils.startsWith(row[0].toString(), "# Partition Infor")) {
         partitionBlock = true;
         colsBlock = false;
-      }else if (colsBlock){
+      } else if (colsBlock) {
         descCols.add(row[0].toString());
       } else if (partitionBlock && !StringUtils.startsWith(row[0].toString(), "#")) {
         descPartitionKey.add(row[0].toString());
@@ -450,4 +462,68 @@ public class SparkTestContext extends ExternalResource {
         descPartitionKey.stream().sorted().distinct().toArray());
   }
 
+  public List<DataFile> writeHive(ArcticTable table, LocationKind locationKind, List<Record> records) throws IOException {
+    AdaptHiveGenericTaskWriterBuilder builder = AdaptHiveGenericTaskWriterBuilder
+        .builderFor(table)
+        .withTransactionId(1);
+
+    TaskWriter<Record> changeWrite = builder.buildWriter(locationKind);
+    for (Record record : records) {
+      changeWrite.write(record);
+    }
+    DataFile[] dataFiles = changeWrite.complete().dataFiles();
+    if (table.isKeyedTable()) {
+      KeyedTable keyedTable = table.asKeyedTable();
+      OverwriteBaseFiles overwriteBaseFiles = keyedTable.newOverwriteBaseFiles();
+      Arrays.stream(dataFiles).forEach(overwriteBaseFiles::addFile);
+      overwriteBaseFiles.withTransactionId(keyedTable.beginTransaction(System.currentTimeMillis() + ""));
+      overwriteBaseFiles.commit();
+    } else if (table.isUnkeyedTable()) {
+      UnkeyedTable unkeyedTable = table.asUnkeyedTable();
+      OverwriteFiles overwriteFiles = unkeyedTable.newOverwrite();
+      Arrays.stream(dataFiles).forEach(overwriteFiles::addFile);
+      overwriteFiles.commit();
+    } else {
+      throw new IllegalStateException("Table is neither keyed nor unkeyed");
+    }
+    return Arrays.asList(dataFiles);
+  }
+
+  public void adaptHiveInsertPosDeleteFiles(long transactionId, List<DataFile> dataFiles, ArcticTable table) throws IOException {
+    Map<StructLike, List<DataFile>> dataFilesPartitionMap =
+        new HashMap<>(dataFiles.stream().collect(Collectors.groupingBy(ContentFile::partition)));
+    List<DeleteFile> deleteFiles = new ArrayList<>();
+    for (Map.Entry<StructLike, List<DataFile>> dataFilePartitionMap : dataFilesPartitionMap.entrySet()) {
+      StructLike partition = dataFilePartitionMap.getKey();
+      List<DataFile> partitionFiles = dataFilePartitionMap.getValue();
+      Map<DataTreeNode, List<DataFile>> nodeFilesPartitionMap = new HashMap<>(partitionFiles.stream()
+          .collect(Collectors.groupingBy(dataFile ->
+              DefaultKeyedFile.parseMetaFromFileName(dataFile.path().toString()).node())));
+      for (Map.Entry<DataTreeNode, List<DataFile>> nodeFilePartitionMap : nodeFilesPartitionMap.entrySet()) {
+        DataTreeNode key = nodeFilePartitionMap.getKey();
+        List<DataFile> nodeFiles = nodeFilePartitionMap.getValue();
+
+        // write pos delete
+        SortedPosDeleteWriter<Record> writer = AdaptHiveGenericTaskWriterBuilder
+            .builderFor(table)
+            .withTransactionId(transactionId).buildBasePosDeleteWriter(key.getMask(), key.getIndex(), partition);
+        for (DataFile nodeFile : nodeFiles) {
+          writer.delete(nodeFile.path(), 0);
+        }
+        deleteFiles.addAll(writer.complete());
+      }
+    }
+    if (table.isKeyedTable()) {
+      RowDelta rowDelta = table.asKeyedTable().baseTable().newRowDelta();
+      deleteFiles.forEach(rowDelta::addDeletes);
+      rowDelta.commit();
+    } else if (table.isUnkeyedTable()) {
+      RowDelta rowDelta = table.asUnkeyedTable().newRowDelta();
+      deleteFiles.forEach(rowDelta::addDeletes);
+      rowDelta.commit();
+    } else {
+      throw new IllegalStateException("Table is neither keyed nor unkeyed");
+    }
+  }
+
 }
diff --git a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/TestHiveTableMergeOnRead.java b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/TestHiveTableMergeOnRead.java
new file mode 100644
index 000000000..9a826f8f9
--- /dev/null
+++ b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/TestHiveTableMergeOnRead.java
@@ -0,0 +1,193 @@
+package com.netease.arctic.spark;
+
+import com.netease.arctic.data.ChangeAction;
+import com.netease.arctic.hive.table.HiveLocationKind;
+import com.netease.arctic.table.BaseLocationKind;
+import com.netease.arctic.table.KeyedTable;
+import com.netease.arctic.table.TableIdentifier;
+import com.netease.arctic.table.UnkeyedTable;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.thrift.TException;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+public class TestHiveTableMergeOnRead extends SparkTestBase {
+  private final String database = "db";
+  private final String table = "testA";
+  private KeyedTable keyedTable;
+
+  private UnkeyedTable unkeyedTable;
+  private final TableIdentifier identifier = TableIdentifier.of(catalogNameHive, database, table);
+
+  @Before
+  public void before() {
+    sql("use " + catalogNameHive);
+    sql("create database if not exists {0}", database);
+  }
+
+  @After
+  public void after() {
+    sql("use " + catalogNameHive);
+    sql("drop table {0}.{1}", database, table);
+    sql("drop database if exists " + database);
+  }
+
+  @Test
+  public void testMergeOnReadKeyedPartition() throws IOException, TException {
+    sql("create table {0}.{1} ( \n" +
+        " id int , \n" +
+        " data string , \n " +
+        " dt string , \n" +
+        " primary key (id) \n" +
+        ") using arctic \n" +
+        " partitioned by ( dt ) \n", database, table);
+    keyedTable = loadTable(identifier).asKeyedTable();
+
+    writeHive(keyedTable, BaseLocationKind.INSTANT, Lists.newArrayList(
+        newRecord(keyedTable, 1, "aaa", "2021-1-1"),
+        newRecord(keyedTable, 2, "bbb", "2021-1-1"),
+        newRecord(keyedTable, 3, "ccc", "2021-1-1"),
+        newRecord(keyedTable, 4, "ddd", "2021-1-2"),
+        newRecord(keyedTable, 5, "eee", "2021-1-2"),
+        newRecord(keyedTable, 6, "fff", "2021-1-2")
+    ));
+    writeHive(keyedTable, HiveLocationKind.INSTANT, Lists.newArrayList(
+        newRecord(keyedTable, 7, "aaa_hive", "2021-1-1"),
+        newRecord(keyedTable, 8, "bbb_hive", "2021-1-1"),
+        newRecord(keyedTable, 9, "ccc_hive", "2021-1-2"),
+        newRecord(keyedTable, 10, "ddd_hive", "2021-1-2")
+    ));
+    writeChange(identifier, ChangeAction.DELETE, Lists.newArrayList(
+        newRecord(keyedTable, 1, "aaa", "2021-1-1")
+    ));
+
+    sql("select * from {0}.{1} order by id", database, table);
+    Assert.assertEquals(9, rows.size());
+    assertContainIdSet(rows, 0, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+
+    List<Partition> partitions = hms.getClient().listPartitions(
+        database,
+        table,
+        (short) -1);
+    Assert.assertEquals(2, partitions.size());
+    sql("use spark_catalog");
+    sql("select * from {0}.{1}", database, table);
+    Assert.assertEquals(4, rows.size());
+    assertContainIdSet(rows, 0, 7, 8, 9, 10);
+  }
+
+  @Test
+  public void testMergeOnReadUnkeyedPartiton() throws IOException, TException {
+    sql("create table {0}.{1} ( \n" +
+        " id int , \n" +
+        " data string , \n " +
+        " dt string \n" +
+        ") using arctic \n" +
+        " partitioned by ( dt ) \n", database, table);
+    unkeyedTable = loadTable(identifier).asUnkeyedTable();
+    writeHive(unkeyedTable, BaseLocationKind.INSTANT, Lists.newArrayList(
+        newRecord(unkeyedTable.schema(), 1, "aaa", "2021-1-1"),
+        newRecord(unkeyedTable.schema(), 2, "bbb", "2021-1-1"),
+        newRecord(unkeyedTable.schema(), 3, "ccc", "2021-1-1"),
+        newRecord(unkeyedTable.schema(), 4, "ddd", "2021-1-2"),
+        newRecord(unkeyedTable.schema(), 5, "eee", "2021-1-2"),
+        newRecord(unkeyedTable.schema(), 6, "fff", "2021-1-2")
+    ));
+    writeHive(unkeyedTable, HiveLocationKind.INSTANT, Lists.newArrayList(
+        newRecord(unkeyedTable.schema(), 7, "aaa_hive", "2021-1-1"),
+        newRecord(unkeyedTable.schema(), 8, "bbb_hive", "2021-1-1"),
+        newRecord(unkeyedTable.schema(), 9, "ccc_hive", "2021-1-2"),
+        newRecord(unkeyedTable.schema(), 10, "ddd_hive", "2021-1-2")
+    ));
+    sql("select * from {0}.{1} order by id", database, table);
+    Assert.assertEquals(10, rows.size());
+    assertContainIdSet(rows, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+
+    List<Partition> partitions = hms.getClient().listPartitions(
+        database,
+        table,
+        (short) -1);
+    Assert.assertEquals(2, partitions.size());
+    sql("use spark_catalog");
+    sql("select * from {0}.{1}", database, table);
+    Assert.assertEquals(4, rows.size());
+    assertContainIdSet(rows, 0, 7, 8, 9, 10);
+  }
+
+  @Test
+  public void testMergeOnReadKeyedUnPartition() throws IOException {
+    sql("create table {0}.{1} ( \n" +
+        " id int , \n" +
+        " data string , \n " +
+        " dt string , \n" +
+        " primary key (id) \n" +
+        ") using arctic \n", database, table);
+    keyedTable = loadTable(identifier).asKeyedTable();
+
+    List<DataFile> dataFiles = writeHive(keyedTable, BaseLocationKind.INSTANT, Lists.newArrayList(
+        newRecord(keyedTable, 1, "aaa", "2021-1-1"),
+        newRecord(keyedTable, 2, "bbb", "2021-1-1"),
+        newRecord(keyedTable, 3, "ccc", "2021-1-1"),
+        newRecord(keyedTable, 4, "ddd", "2021-1-2"),
+        newRecord(keyedTable, 5, "eee", "2021-1-2"),
+        newRecord(keyedTable, 6, "fff", "2021-1-2")
+    ));
+
+    writeHive(keyedTable, HiveLocationKind.INSTANT, Lists.newArrayList(
+        newRecord(keyedTable, 7, "aaa_hive", "2021-1-1"),
+        newRecord(keyedTable, 8, "bbb_hive", "2021-1-1"),
+        newRecord(keyedTable, 9, "ccc_hive", "2021-1-2"),
+        newRecord(keyedTable, 10, "ddd_hive", "2021-1-2")
+    ));
+    writeChange(identifier, ChangeAction.DELETE, Lists.newArrayList(
+        newRecord(keyedTable, 1, "aaa", "2021-1-1")
+    ));
+    sql("select * from {0}.{1} order by id", database, table);
+    Assert.assertEquals(9, rows.size());
+    assertContainIdSet(rows, 0, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+
+    sql("use spark_catalog");
+    sql("select * from {0}.{1}", database, table);
+    Assert.assertEquals(4, rows.size());
+    assertContainIdSet(rows, 0, 7, 8, 9, 10);
+  }
+
+  @Test
+  public void testMergeOnReadUnkeyedUnpartition() throws IOException {
+    sql("create table {0}.{1} ( \n" +
+        " id int , \n" +
+        " data string , \n " +
+        " dt string \n" +
+        ") using arctic \n", database, table);
+    unkeyedTable = loadTable(identifier).asUnkeyedTable();
+    writeHive(unkeyedTable, BaseLocationKind.INSTANT, Lists.newArrayList(
+        newRecord(unkeyedTable.schema(), 1, "aaa", "2021-1-1"),
+        newRecord(unkeyedTable.schema(), 2, "bbb", "2021-1-1"),
+        newRecord(unkeyedTable.schema(), 3, "ccc", "2021-1-1"),
+        newRecord(unkeyedTable.schema(), 4, "ddd", "2021-1-2"),
+        newRecord(unkeyedTable.schema(), 5, "eee", "2021-1-2"),
+        newRecord(unkeyedTable.schema(), 6, "fff", "2021-1-2")
+    ));
+    writeHive(unkeyedTable, HiveLocationKind.INSTANT, Lists.newArrayList(
+        newRecord(unkeyedTable.schema(), 7, "aaa_hive", "2021-1-1"),
+        newRecord(unkeyedTable.schema(), 8, "bbb_hive", "2021-1-1"),
+        newRecord(unkeyedTable.schema(), 9, "ccc_hive", "2021-1-2"),
+        newRecord(unkeyedTable.schema(), 10, "ddd_hive", "2021-1-2")
+    ));
+    sql("select * from {0}.{1} order by id", database, table);
+    Assert.assertEquals(10, rows.size());
+    assertContainIdSet(rows, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
+    sql("use spark_catalog");
+    sql("select * from {0}.{1}", database, table);
+    Assert.assertEquals(4, rows.size());
+    assertContainIdSet(rows, 0, 7, 8, 9, 10);
+  }
+
+}
diff --git a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/hive/TestAdaptHiveWriter.java b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/hive/TestAdaptHiveWriter.java
deleted file mode 100644
index e3afbc131..000000000
--- a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/hive/TestAdaptHiveWriter.java
+++ /dev/null
@@ -1,162 +0,0 @@
-/*
- * 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.spark.hive;
-
-import com.netease.arctic.hive.HiveTableTestBase;
-import com.netease.arctic.hive.table.HiveLocationKind;
-import com.netease.arctic.spark.writer.ArcticSparkBaseTaskWriter;
-import com.netease.arctic.spark.writer.ArcticSparkTaskWriterBuilder;
-import com.netease.arctic.table.ArcticTable;
-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.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZoneOffset;
-import java.time.temporal.ChronoUnit;
-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.io.CloseableIterable;
-import org.apache.iceberg.io.TaskWriter;
-import org.apache.iceberg.io.WriteResult;
-import org.apache.iceberg.parquet.AdaptHiveParquet;
-import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.apache.iceberg.spark.SparkSchemaUtil;
-import org.apache.iceberg.spark.data.SparkParquetReaders;
-import org.apache.spark.sql.catalyst.InternalRow;
-import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
-import org.apache.spark.sql.types.Decimal;
-import org.apache.spark.unsafe.types.UTF8String;
-import org.junit.Assert;
-import org.junit.Test;
-import scala.collection.mutable.ArrayBuffer;
-
-public class TestAdaptHiveWriter extends HiveTableTestBase {
-
-  @Test
-  public void testWriteTypeFromOperateKind(){
-    {
-      ArcticSparkTaskWriterBuilder builder = ArcticSparkTaskWriterBuilder
-          .buildFor(testKeyedHiveTable)
-          .withTransactionId(1);
-
-      Assert.assertTrue(builder.buildWriter(BaseLocationKind.INSTANT) instanceof ArcticSparkBaseTaskWriter);
-      Assert.assertTrue(builder.buildWriter(HiveLocationKind.INSTANT) instanceof ArcticSparkBaseTaskWriter);
-
-      Assert.assertTrue(builder.buildWriter(WriteOperationKind.OVERWRITE) instanceof ArcticSparkBaseTaskWriter);
-    }
-    {
-      ArcticSparkTaskWriterBuilder builder = ArcticSparkTaskWriterBuilder
-          .buildFor(testHiveTable)
-          .withTransactionId(1);
-
-      Assert.assertTrue(builder.buildWriter(BaseLocationKind.INSTANT) instanceof ArcticSparkBaseTaskWriter);
-      Assert.assertTrue(builder.buildWriter(HiveLocationKind.INSTANT) instanceof ArcticSparkBaseTaskWriter);
-
-      Assert.assertTrue(builder.buildWriter(WriteOperationKind.OVERWRITE) instanceof ArcticSparkBaseTaskWriter);
-    }
-  }
-
-  @Test
-  public void testKeyedTableBaseWriteByLocationKind() throws IOException {
-    testWrite(testKeyedHiveTable, BaseLocationKind.INSTANT, generateInternalRow(), "base");
-  }
-
-  @Test
-  public void testKeyedTableHiveWriteByLocationKind() throws IOException {
-    testWrite(testKeyedHiveTable, HiveLocationKind.INSTANT, generateInternalRow(), "hive_data");
-  }
-
-  @Test
-  public void testUnKeyedTableChangeWriteByLocationKind() throws IOException {
-    try {
-      testWrite(testHiveTable, ChangeLocationKind.INSTANT, generateInternalRow(), "change");
-    }catch (Exception e){
-      Assert.assertTrue(e instanceof IllegalArgumentException);
-    }
-  }
-
-  @Test
-  public void testUnKeyedTableBaseWriteByLocationKind() throws IOException {
-    testWrite(testHiveTable, BaseLocationKind.INSTANT, generateInternalRow(), "base");
-  }
-
-  @Test
-  public void testUnKeyedTableHiveWriteByLocationKind() throws IOException {
-    testWrite(testHiveTable, HiveLocationKind.INSTANT, generateInternalRow(), "hive_data");
-  }
-
-  public void testWrite(ArcticTable table, LocationKind locationKind, List<InternalRow> records, String pathFeature) throws IOException {
-    ArcticSparkTaskWriterBuilder builder = ArcticSparkTaskWriterBuilder
-        .buildFor(table)
-        .withDataSourceSchema(SparkSchemaUtil.convert(table.schema()))
-        .withTransactionId(1);
-
-    TaskWriter<InternalRow> changeWrite = builder.buildWriter(locationKind);
-    for (InternalRow record: records) {
-      changeWrite.write(record);
-    }
-    WriteResult complete = changeWrite.complete();
-    Arrays.stream(complete.dataFiles()).forEach(s -> Assert.assertTrue(s.path().toString().contains(pathFeature)));
-    CloseableIterable<InternalRow> concat =
-        CloseableIterable.concat(Arrays.stream(complete.dataFiles()).map(s -> readParquet(
-            table.schema(),
-            s.path().toString())).collect(Collectors.toList()));
-    Set<InternalRow> result = new HashSet<>();
-    Iterators.addAll(result, concat.iterator());
-    Assert.assertEquals(result, records.stream().collect(Collectors.toSet()));
-  }
-
-  private CloseableIterable<InternalRow> readParquet(Schema schema, String path) {
-    AdaptHiveParquet.ReadBuilder builder = AdaptHiveParquet.read(
-            Files.localInput(new File(path)))
-        .project(schema)
-        .createReaderFunc(fileSchema -> SparkParquetReaders.buildReader(schema, fileSchema, new HashMap<>()))
-        .caseSensitive(false);
-
-    CloseableIterable<InternalRow> iterable = builder.build();
-    return iterable;
-  }
-
-  private List<InternalRow> generateInternalRow() {
-    Instant EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC).toInstant();
-    ArrayBuffer seq = new ArrayBuffer<Object>(5);
-    seq = seq.$plus$eq(1);
-    seq = seq.$plus$eq(UTF8String.fromString("jack"));
-    seq = seq.$plus$eq(
-        ChronoUnit.MICROS.between(EPOCH, LocalDateTime.of(2022, 1, 1, 10, 0, 0)
-            .atOffset(ZoneOffset.ofHours(8)).toInstant()));
-    seq = seq.$plus$eq(
-        ChronoUnit.MICROS.between(EPOCH, LocalDateTime.of(2022, 1, 1, 10, 0, 0)
-            .atOffset(ZoneOffset.ofHours(8)).toInstant()));
-    seq = seq.$plus$eq(Decimal.apply("100"));
-    InternalRow internalRow = GenericInternalRow.fromSeq(seq);
-    return Lists.newArrayList(internalRow);
-  }
-}