Repository: NetEase/arctic
Pull request #156 (2022-08-10 13:10:38)

Impelenet UpdateSchema/UpdateProperties for arctic table with hive base store
for https://github.com/NetEase/arctic/issues/42
add logic:
1.update hive table schema
2.set properties to hive table


Patch:
diff --git a/.github/ISSUE_TEMPLATE/bug_report.yml b/.github/ISSUE_TEMPLATE/bug_report.yml
index 7ddfec969..faa4f7b6a 100644
--- a/.github/ISSUE_TEMPLATE/bug_report.yml
+++ b/.github/ISSUE_TEMPLATE/bug_report.yml
@@ -1,21 +1,20 @@
-/*
-  * 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.
-  */
-name: Bug Report
+#
+# 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.
+#
+name: Arctic Bug Report
 description: File a bug report
 title: "[Bug]: "
 labels: ["bug"]
diff --git a/.github/ISSUE_TEMPLATE/config.yml b/.github/ISSUE_TEMPLATE/config.yml
index 71b5a27c1..97357ad86 100644
--- a/.github/ISSUE_TEMPLATE/config.yml
+++ b/.github/ISSUE_TEMPLATE/config.yml
@@ -1,21 +1,20 @@
-/*
-  * 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.
-  */
-blank_issues_enabled: true
+#
+# 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.
+#
+blank_issues_enabled: false
 contact_links:
   - name: GitHub Community Support
     url: https://github.com/NetEase/arctic/discussions
diff --git a/.github/ISSUE_TEMPLATE/feature_request.yml b/.github/ISSUE_TEMPLATE/feature_request.yml
index deac672c9..3f77e986b 100644
--- a/.github/ISSUE_TEMPLATE/feature_request.yml
+++ b/.github/ISSUE_TEMPLATE/feature_request.yml
@@ -1,21 +1,20 @@
-/*
-  * 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.
-  */
-name: Arctic feature request
+#
+# 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.
+#
+name: Arctic Feature Request
 description: Suggest an idea for this project
 title: "[Feature]: "
 labels: ["feature"]
diff --git a/.github/ISSUE_TEMPLATE/improvement.yml b/.github/ISSUE_TEMPLATE/improvement.yml
index 160f8ed33..52b360d63 100644
--- a/.github/ISSUE_TEMPLATE/improvement.yml
+++ b/.github/ISSUE_TEMPLATE/improvement.yml
@@ -1,20 +1,19 @@
-/*
-  * 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.
-  */
+#
+# 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.
+#
 name: Arctic Improvement
 title: "[Improvement]: "
 description: Suggest an improvement on performance, code quality, user experience, etc
diff --git a/.github/PULL_REQUEST_TEMPLATE b/.github/PULL_REQUEST_TEMPLATE
index 6d02bc6a7..96d2039a8 100644
--- a/.github/PULL_REQUEST_TEMPLATE
+++ b/.github/PULL_REQUEST_TEMPLATE
@@ -1,20 +1,3 @@
-/*
-  * 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.
-  */
 <!--
 Thanks for sending a pull request!
 
@@ -24,23 +7,30 @@ Here are some tips for you:
   3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][ARCTIC #XXXX] Your PR title ...'.
 -->
 
-## _Why are the changes needed?_
+## Why are the changes needed?
 <!--
 Please clarify why the changes are needed. For instance,
-  1. If you add a feature, you can talk about the use case of it.
+  1. If you add a feature, you can talk about its use case.
   2. If you fix a bug, you can clarify why it is a bug.
 -->
 
+*(For example: The Arctic Flink dataStream API has already supported this configuration: 'scan.startup.mode', but it is not available with Flink SQL. We should expose this configuration to users when they are using Flink SQL.)*
+
 ## Brief change log
 
-*(for example:)*
-  - *The TaskInfo is stored in the blob store on job creation time as a persistent artifact*
-  - *Deployments RPC transmits only the blob storage reference*
-  - *TaskManagers retrieve the TaskInfo from the blob cache*
+*(For example:)*
+  - *Add the 'scan.startup.mode' configuration to the ArcticScanContext in the flink modules*
+  - *The flink 1.12 and 1.14 versions are affected.*
+  - *Add the documentation in the flink chapter*
 
-## _How was this patch tested?_
+## How was this patch tested?
 - [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible
 
 - [ ] Add screenshots for manual tests if appropriate
 
-- [ ] Run test locally before make a pull request
\ No newline at end of file
+- [ ] Run test locally before making a pull request
+
+## Documentation
+
+  - Does this pull request introduces a new feature? (yes / no)
+  - If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
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 cf49ecc46..e37e744a6 100644
--- a/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
@@ -54,7 +54,7 @@ import java.util.stream.Collectors;
 public class BaseKeyedTable implements KeyedTable {
   private final String tableLocation;
   private final PrimaryKeySpec primaryKeySpec;
-  private final AmsClient client;
+  protected final AmsClient client;
 
   protected final BaseTable baseTable;
   protected final ChangeTable changeTable;
@@ -160,13 +160,7 @@ public class BaseKeyedTable implements KeyedTable {
 
   @Override
   public UpdateProperties updateProperties() {
-    UpdateProperties updateProperties = new UpdateKeyedTableProperties(this, tableMeta);
-    if (client != null) {
-      AmsTableTracer tracer = new AmsTableTracer(this, TrackerOperations.UPDATE_PROPERTIES, client);
-      return new TracedUpdateProperties(updateProperties, tracer);
-    } else {
-      return updateProperties;
-    }
+    return new UpdateKeyedTableProperties(this, tableMeta);
   }
 
   @Override
diff --git a/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java b/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java
index 4244c1d17..d2beb7bd6 100644
--- a/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java
+++ b/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java
@@ -171,7 +171,7 @@ public class AmsTableTracer implements TableTracer {
       commitMeta.setSchemaUpdateMeta(ddlCommitMeta);
       update = true;
     }
-    if (this.properties != null) {
+    if (this.properties != null && Constants.INNER_TABLE_BASE.equals(innerTable)) {
       commitMeta.setProperties(this.properties);
       update = true;
     }
diff --git a/hive/src/main/java/com/netease/arctic/hive/HiveTableProperties.java b/hive/src/main/java/com/netease/arctic/hive/HiveTableProperties.java
new file mode 100644
index 000000000..6c472b4b8
--- /dev/null
+++ b/hive/src/main/java/com/netease/arctic/hive/HiveTableProperties.java
@@ -0,0 +1,11 @@
+package com.netease.arctic.hive;
+
+
+public class HiveTableProperties {
+
+  public static final String ARCTIC_TABLE_FLAG = "arctic.enable";
+
+  public static final String ARCTIC_TABLE_PRIMARY_KEYS = "arctic.table.primary-keys";
+
+}
+
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 9dc100cc4..dba0db00b 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
@@ -24,6 +24,7 @@ import com.netease.arctic.ams.api.TableMeta;
 import com.netease.arctic.ams.api.properties.MetaTableProperties;
 import com.netease.arctic.catalog.BaseArcticCatalog;
 import com.netease.arctic.hive.CachedHiveClientPool;
+import com.netease.arctic.hive.HiveTableProperties;
 import com.netease.arctic.hive.table.KeyedHiveTable;
 import com.netease.arctic.hive.table.UnkeyedHiveTable;
 import com.netease.arctic.hive.utils.HiveSchemaUtil;
@@ -236,6 +237,7 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
               hiveLocation,
               FileFormat.valueOf(PropertyUtil.propertyAsString(properties, TableProperties.DEFAULT_FILE_FORMAT,
                   TableProperties.DEFAULT_FILE_FORMAT_DEFAULT).toUpperCase(Locale.ENGLISH))));
+          setProToHive(hiveTable);
           client.createTable(hiveTable);
           return null;
         });
@@ -347,5 +349,14 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
         LOG.warn("Failed to drop hive table while rolling back create table operation", e);
       }
     }
+
+
+    private void setProToHive(org.apache.hadoop.hive.metastore.api.Table hiveTable) {
+      Map<String, String> parameters = new HashMap<>();
+      parameters.put(HiveTableProperties.ARCTIC_TABLE_FLAG, "true");
+      parameters.put(HiveTableProperties.ARCTIC_TABLE_PRIMARY_KEYS, primaryKeySpec.description());
+      hiveTable.setParameters(parameters);
+    }
   }
+
 }
diff --git a/hive/src/main/java/com/netease/arctic/hive/op/HiveSchemaUpdate.java b/hive/src/main/java/com/netease/arctic/hive/op/HiveSchemaUpdate.java
new file mode 100644
index 000000000..cadb93cd3
--- /dev/null
+++ b/hive/src/main/java/com/netease/arctic/hive/op/HiveSchemaUpdate.java
@@ -0,0 +1,178 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *  *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *  *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.hive.op;
+
+import com.netease.arctic.hive.HMSClient;
+import com.netease.arctic.hive.utils.HiveSchemaUtil;
+import com.netease.arctic.hive.utils.HiveTableUtil;
+import com.netease.arctic.table.ArcticTable;
+import com.netease.arctic.table.KeyedTable;
+import org.apache.iceberg.PartitionField;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+
+import java.util.Collection;
+
+/**
+ * Schema evolution API implementation for {@link KeyedTable}.
+ */
+public class HiveSchemaUpdate implements UpdateSchema {
+  private ArcticTable arcticTable;
+  private Table baseTable;
+  private HMSClient hiveClient;
+  private UpdateSchema updateSchema;
+
+  public HiveSchemaUpdate(ArcticTable arcticTable, HMSClient hiveClient, UpdateSchema updateSchema) {
+    this.arcticTable = arcticTable;
+    if (arcticTable.isKeyedTable()) {
+      this.baseTable = arcticTable.asKeyedTable().baseTable();
+    } else {
+      this.baseTable = arcticTable.asUnkeyedTable();
+    }
+    this.hiveClient = hiveClient;
+    this.updateSchema = updateSchema;
+  }
+
+  @Override
+  public Schema apply() {
+    return this.updateSchema.apply();
+  }
+
+  @Override
+  public void commit() {
+    this.updateSchema.commit();
+    if (HiveTableUtil.loadHmsTable(hiveClient, arcticTable) == null) {
+      throw new RuntimeException(String.format("there is no such hive table named %s", arcticTable.id().toString()));
+    }
+    baseTable.refresh();
+    syncSchemaToHive();
+  }
+
+  private void syncSchemaToHive() {
+    org.apache.hadoop.hive.metastore.api.Table tbl = HiveTableUtil.loadHmsTable(hiveClient, arcticTable);
+    if (tbl == null) {
+      throw new RuntimeException(String.format("there is no such hive table named %s", arcticTable.id().toString()));
+    }
+    tbl.setSd(HiveSchemaUtil.storageDescriptor(baseTable.schema(), baseTable.spec()));
+    HiveTableUtil.persistTable(hiveClient, tbl);
+  }
+
+  @Override
+  public UpdateSchema allowIncompatibleChanges() {
+    throw new UnsupportedOperationException("hive table not support allowIncompatibleChanges");
+  }
+
+  @Override
+  public UpdateSchema addColumn(String name, Type type, String doc) {
+    this.updateSchema.addColumn(name, type, doc);
+    //It is strictly required that all non-partitioned columns precede partitioned columns in the schema.
+    if (!baseTable.spec().isUnpartitioned()) {
+      int parFieldMinIndex = Integer.MAX_VALUE;
+      Types.NestedField firstParField = null;
+      for (PartitionField partitionField : baseTable.spec().fields()) {
+        Types.NestedField sourceField = baseTable.schema().findField(partitionField.sourceId());
+        if (baseTable.schema().columns().indexOf(sourceField) < parFieldMinIndex) {
+          parFieldMinIndex = baseTable.schema().columns().indexOf(sourceField);
+          firstParField = sourceField;
+        }
+      }
+      if (firstParField != null) {
+        this.updateSchema.moveBefore(name, firstParField.name());
+      }
+    }
+    return this;
+  }
+
+  @Override
+  public UpdateSchema addColumn(String parent, String name, Type type, String doc) {
+    this.updateSchema.addColumn(parent, name, type, doc);
+    return this;
+  }
+
+  @Override
+  public UpdateSchema addRequiredColumn(String name, Type type, String doc) {
+    throw new UnsupportedOperationException("hive table not support addRequiredColumn");
+  }
+
+  @Override
+  public UpdateSchema addRequiredColumn(String parent, String name, Type type, String doc) {
+    throw new UnsupportedOperationException("hive table not support addRequiredColumn");
+  }
+
+  @Override
+  public UpdateSchema renameColumn(String name, String newName) {
+    throw new UnsupportedOperationException("not support renameColumn now, there will be error when hive stored as " +
+        "parquet and we rename the column");
+  }
+
+  @Override
+  public UpdateSchema updateColumn(String name, Type.PrimitiveType newType) {
+    this.updateSchema.updateColumn(name, newType);
+    return this;
+  }
+
+  @Override
+  public UpdateSchema updateColumnDoc(String name, String newDoc) {
+    this.updateSchema.updateColumnDoc(name, newDoc);
+    return this;
+  }
+
+  @Override
+  public UpdateSchema makeColumnOptional(String name) {
+    throw new UnsupportedOperationException("hive table not support makeColumnOptional");
+  }
+
+  @Override
+  public UpdateSchema requireColumn(String name) {
+    throw new UnsupportedOperationException("hive table not support requireColumn");
+  }
+
+  @Override
+  public UpdateSchema deleteColumn(String name) {
+    throw new UnsupportedOperationException("hive table not support deleteColumn");
+  }
+
+  @Override
+  public UpdateSchema moveFirst(String name) {
+    throw new UnsupportedOperationException("hive table not support moveFirst");
+  }
+
+  @Override
+  public UpdateSchema moveBefore(String name, String beforeName) {
+    throw new UnsupportedOperationException("hive table not support moveBefore");
+  }
+
+  @Override
+  public UpdateSchema moveAfter(String name, String afterName) {
+    throw new UnsupportedOperationException("hive table not support moveAfter");
+  }
+
+  @Override
+  public UpdateSchema unionByNameWith(Schema newSchema) {
+    throw new UnsupportedOperationException("hive table not support unionByNameWith");
+  }
+
+  @Override
+  public UpdateSchema setIdentifierFields(Collection<String> names) {
+    throw new UnsupportedOperationException("hive table not support setIdentifierFields");
+  }
+}
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 6be8e1d66..da0b49a05 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
@@ -216,7 +216,7 @@ public class ReplaceHivePartitions implements ReplacePartitions {
     try {
       transactionalHMSClient.run(c -> {
         if (!rewritePartitions.isEmpty()) {
-          c.alter_partitions(db, tableName, rewritePartitions.values().stream().collect(Collectors.toList()));
+          c.alter_partitions(db, tableName, rewritePartitions.values().stream().collect(Collectors.toList()), null);
         }
         if (!newPartitions.isEmpty()) {
           c.add_partitions(newPartitions.values().stream().collect(Collectors.toList()));
diff --git a/hive/src/main/java/com/netease/arctic/hive/table/KeyedHiveTable.java b/hive/src/main/java/com/netease/arctic/hive/table/KeyedHiveTable.java
index 80afee72a..f5f0ad03a 100644
--- a/hive/src/main/java/com/netease/arctic/hive/table/KeyedHiveTable.java
+++ b/hive/src/main/java/com/netease/arctic/hive/table/KeyedHiveTable.java
@@ -21,15 +21,11 @@ package com.netease.arctic.hive.table;
 import com.netease.arctic.AmsClient;
 import com.netease.arctic.ams.api.TableMeta;
 import com.netease.arctic.hive.HMSClient;
-import com.netease.arctic.hive.utils.HiveSchemaUtil;
-import com.netease.arctic.io.ArcticFileIO;
+import com.netease.arctic.hive.op.HiveSchemaUpdate;
 import com.netease.arctic.table.BaseKeyedTable;
-import com.netease.arctic.table.BaseTable;
 import com.netease.arctic.table.ChangeTable;
 import com.netease.arctic.table.PrimaryKeySpec;
-import com.netease.arctic.table.TableIdentifier;
-import org.apache.iceberg.Schema;
-import org.apache.iceberg.Table;
+import org.apache.iceberg.UpdateSchema;
 
 /**
  * Implementation of {@link com.netease.arctic.table.KeyedTable} with Hive table as base store.
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 401e7cb85..22dd8b7d9 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
@@ -21,6 +21,7 @@ package com.netease.arctic.hive.table;
 import com.netease.arctic.AmsClient;
 import com.netease.arctic.hive.HMSClient;
 import com.netease.arctic.hive.op.HiveOperationTransaction;
+import com.netease.arctic.hive.op.HiveSchemaUpdate;
 import com.netease.arctic.hive.op.OverwriteHiveFiles;
 import com.netease.arctic.hive.op.ReplaceHivePartitions;
 import com.netease.arctic.io.ArcticFileIO;
@@ -31,6 +32,7 @@ import org.apache.iceberg.OverwriteFiles;
 import org.apache.iceberg.ReplacePartitions;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateSchema;
 
 import static com.netease.arctic.table.TableProperties.BASE_HIVE_LOCATION_ROOT;
 
@@ -82,4 +84,9 @@ public class UnkeyedHiveTable extends BaseUnkeyedTable implements BaseTable, Sup
     Transaction transaction = super.newTransaction();
     return new HiveOperationTransaction(this, transaction, hiveClient);
   }
+
+  @Override
+  public UpdateSchema updateSchema() {
+    return new HiveSchemaUpdate(this, hiveClient, super.updateSchema());
+  }
 }
diff --git a/hive/src/main/java/com/netease/arctic/hive/utils/HiveSchemaUtil.java b/hive/src/main/java/com/netease/arctic/hive/utils/HiveSchemaUtil.java
index 020ceedc1..51798ba40 100644
--- a/hive/src/main/java/com/netease/arctic/hive/utils/HiveSchemaUtil.java
+++ b/hive/src/main/java/com/netease/arctic/hive/utils/HiveSchemaUtil.java
@@ -19,6 +19,8 @@
 package com.netease.arctic.hive.utils;
 
 import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.types.TypeUtil;
@@ -69,4 +71,11 @@ public class HiveSchemaUtil {
     return org.apache.iceberg.hive.HiveSchemaUtil.convert(TypeUtil.select(schema, spec.identitySourceIds()));
   }
 
+  public static StorageDescriptor storageDescriptor(Schema schema, PartitionSpec partitionSpec) {
+    final StorageDescriptor storageDescriptor = new StorageDescriptor();
+    storageDescriptor.setCols(HiveSchemaUtil.hiveTableFields(schema, partitionSpec));
+    SerDeInfo serDeInfo = new SerDeInfo();
+    storageDescriptor.setSerdeInfo(serDeInfo);
+    return storageDescriptor;
+  }
 }
diff --git a/hive/src/main/java/com/netease/arctic/hive/utils/HiveTableUtil.java b/hive/src/main/java/com/netease/arctic/hive/utils/HiveTableUtil.java
index 8b2dbf6bd..1c1cb5550 100644
--- a/hive/src/main/java/com/netease/arctic/hive/utils/HiveTableUtil.java
+++ b/hive/src/main/java/com/netease/arctic/hive/utils/HiveTableUtil.java
@@ -18,12 +18,47 @@
 
 package com.netease.arctic.hive.utils;
 
+import com.netease.arctic.hive.HMSClient;
 import com.netease.arctic.hive.table.SupportHive;
 import com.netease.arctic.table.ArcticTable;
+import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class HiveTableUtil {
 
+  private static final Logger LOG = LoggerFactory.getLogger(HiveTableUtil.class);
+
   public static boolean isHive(ArcticTable arcticTable) {
     return arcticTable instanceof SupportHive;
   }
+
+  public static org.apache.hadoop.hive.metastore.api.Table loadHmsTable(
+      HMSClient hiveClient, ArcticTable arcticTable) {
+    try {
+      return hiveClient.run(client -> client.getTable(
+          arcticTable.id().getDatabase(),
+          arcticTable.id().getTableName()));
+    } catch (NoSuchObjectException nte) {
+      LOG.trace("Table not found {}", arcticTable.id().toString(), nte);
+      return null;
+    } catch (TException e) {
+      throw new RuntimeException(String.format("Metastore operation failed for %s", arcticTable.id().toString()), e);
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      throw new RuntimeException("Interrupted during commit", e);
+    }
+  }
+
+  public static void persistTable(HMSClient hiveClient, org.apache.hadoop.hive.metastore.api.Table tbl) {
+    try {
+      hiveClient.run(client -> {
+        client.alter_table(tbl.getDbName(), tbl.getTableName(), tbl);
+        return null;
+      });
+    } catch (TException | InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+  }
 }
diff --git a/pom.xml b/pom.xml
index b9c86cd7b..16e2df92d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -64,7 +64,6 @@
         <cglib.version>2.2.2</cglib.version>
         <mockito.version>2.23.4</mockito.version>
         <parquet.version>1.12.0</parquet.version>
-        <hive.version>2.3.7</hive.version>
     </properties>
 
     <dependencyManagement>
@@ -352,5 +351,21 @@
             <url>http://mvn.hz.netease.com/artifactory/libs-snapshots</url>
         </snapshotRepository>
     </distributionManagement>
-
+    <profiles>
+        <profile>
+            <id>hive-v2.3</id>
+            <properties>
+                <hive.version>2.3.7</hive.version>
+            </properties>
+            <activation>
+                <activeByDefault>true</activeByDefault>
+            </activation>
+        </profile>
+        <profile>
+            <id>hive-v2.1</id>
+            <properties>
+                <hive.version>2.1.1</hive.version>
+            </properties>
+        </profile>
+    </profiles>
 </project>
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
index 8fc4634b8..58c927488 100644
--- 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
@@ -18,10 +18,13 @@
 
 package com.netease.arctic.spark.io;
 
+import com.netease.arctic.data.ChangeAction;
 import com.netease.arctic.hive.io.writer.AdaptHiveOutputFileFactory;
 import com.netease.arctic.hive.table.SupportHive;
+import com.netease.arctic.io.writer.ChangeTaskWriter;
 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.table.ArcticTable;
 import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.PrimaryKeySpec;
@@ -40,18 +43,17 @@ import org.apache.spark.sql.types.StructType;
 
 import java.util.Locale;
 
-
 public class TaskWriters {
-  private ArcticTable table;
+  private final 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;
+  private final boolean isHiveTable;
+  private final FileFormat fileFormat;
+  private final long fileSize;
+  private final long mask;
 
   protected TaskWriters(ArcticTable table) {
     this.table = table;
@@ -135,6 +137,18 @@ public class TaskWriters {
         table.io(), fileSize, mask, schema, table.spec(), primaryKeySpec);
   }
 
+  public ChangeTaskWriter<InternalRow> newChangeWriter() {
+    preconditions();
+    // TODO: issues-173 - support change writer for upsert
+    return null;
+  }
+
+  public SortedPosDeleteWriter newBasePosDeleteWriter() {
+    preconditions();
+    // TODO: issues-173 - support change writer for upsert
+    return null;
+  }
+
   private void preconditions() {
     if (table.isKeyedTable()) {
       Preconditions.checkState(transactionId != null, "Transaction id is not set");
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 9a523ea9d..6f38ab366 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,9 @@
 
 package com.netease.arctic.spark.reader;
 
+import com.netease.arctic.spark.table.SupportsExtendIdentColumns;
 import com.netease.arctic.table.ArcticTable;
+import org.apache.iceberg.MetadataColumns;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.expressions.Binder;
@@ -27,6 +29,8 @@ import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.spark.SparkFilters;
 import org.apache.iceberg.spark.SparkSchemaUtil;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
 import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.connector.read.Scan;
 import org.apache.spark.sql.connector.read.ScanBuilder;
@@ -37,13 +41,18 @@ import org.apache.spark.sql.types.StructType;
 import org.apache.spark.sql.util.CaseInsensitiveStringMap;
 
 import java.util.List;
+import java.util.stream.Collectors;
 
-public class SparkScanBuilder implements ScanBuilder, SupportsPushDownFilters, SupportsPushDownRequiredColumns {
+
+public class SparkScanBuilder implements ScanBuilder, SupportsExtendIdentColumns, SupportsPushDownFilters,
+    SupportsPushDownRequiredColumns {
   private static final Filter[] NO_FILTERS = new Filter[0];
 
   private final ArcticTable table;
   private final CaseInsensitiveStringMap options;
 
+  private final List<String> metaColumns = Lists.newArrayList();
+
   private Schema schema = null;
   private StructType requestedProjection;
   private final boolean caseSensitive;
@@ -56,7 +65,7 @@ public class SparkScanBuilder implements ScanBuilder, SupportsPushDownFilters, S
     this.caseSensitive = Boolean.parseBoolean(spark.conf().get("spark.sql.caseSensitive"));
   }
 
-  private Schema lazySchema() {
+  private Schema lazySchemaWithRowIdent() {
     if (schema == null) {
       if (requestedProjection != null) {
         // the projection should include all columns that will be returned,
@@ -67,7 +76,15 @@ public class SparkScanBuilder implements ScanBuilder, SupportsPushDownFilters, S
         this.schema = table.schema();
       }
     }
-    return schema;
+
+    // metadata columns
+    List<Types.NestedField> fields = metaColumns.stream()
+        .distinct()
+        .map(MetadataColumns::get)
+        .collect(Collectors.toList());
+    Schema meta = new Schema(fields);
+
+    return TypeUtil.join(schema, meta);
   }
 
   private Expression filterExpression() {
@@ -116,12 +133,32 @@ public class SparkScanBuilder implements ScanBuilder, SupportsPushDownFilters, S
   @Override
   public Scan build() {
     if (table.isKeyedTable()) {
-      return new KeyedSparkBatchScan(table.asKeyedTable(), caseSensitive, lazySchema(), filterExpressions, options);
+      return new KeyedSparkBatchScan(
+          table.asKeyedTable(),
+          caseSensitive,
+          lazySchemaWithRowIdent(),
+          filterExpressions,
+          options);
     } else if (table.isUnkeyedTable()) {
-      return new UnkeyedSparkBatchScan(table.asUnkeyedTable(), caseSensitive, lazySchema(), filterExpressions, options);
+      return new UnkeyedSparkBatchScan(
+          table.asUnkeyedTable(),
+          caseSensitive,
+          lazySchemaWithRowIdent(),
+          filterExpressions,
+          options);
     } else {
       throw new IllegalStateException("Unable to build scan for table: " + table.id().toString() + ", unknown table " +
           "type");
     }
   }
+
+  @Override
+  public SupportsExtendIdentColumns withIdentifierColumns() {
+    if (table.isUnkeyedTable()) {
+      this.metaColumns.addAll(UnkeyedSparkBatchScan.rowIdColumns);
+    } else if (table.isKeyedTable()) {
+      this.metaColumns.addAll(table.asKeyedTable().primaryKeySpec().fieldNames());
+    }
+    return this;
+  }
 }
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
index 5d7e4e724..9c1e4afcb 100644
--- 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
@@ -54,10 +54,13 @@ 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);
 
+  public static final String FILE_NAME_COL = "_file";
+  public static final String ROW_POS_COL = "_pos";
+  public static final List<String> rowIdColumns = Lists.newArrayList(FILE_NAME_COL, ROW_POS_COL);
+
   private final UnkeyedTable table;
   private final boolean caseSensitive;
   private final Schema expectedSchema;
@@ -69,7 +72,6 @@ public class UnkeyedSparkBatchScan implements Scan, Batch, SupportsReportStatist
   private StructType readSchema = null;
   private List<CombinedScanTask> tasks = null;
 
-
   UnkeyedSparkBatchScan(
       UnkeyedTable table, boolean caseSensitive,
       Schema expectedSchema, List<Expression> filters, CaseInsensitiveStringMap options) {
@@ -292,5 +294,4 @@ public class UnkeyedSparkBatchScan implements Scan, Batch, SupportsReportStatist
         "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/ArcticSparkTable.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/ArcticSparkTable.java
index b717dae4c..082ee0d99 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/ArcticSparkTable.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/ArcticSparkTable.java
@@ -20,8 +20,7 @@ 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.writer.KeyedSparkWriteBuilder;
-import com.netease.arctic.spark.writer.UnkeyedSparkWriteBuilder;
+import com.netease.arctic.spark.writer.ArcticSparkWriteBuilder;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.TableProperties;
 import org.apache.iceberg.Schema;
@@ -45,7 +44,7 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap;
 import java.util.Map;
 import java.util.Set;
 
-public class ArcticSparkTable implements org.apache.spark.sql.connector.catalog.Table, SupportsRead, SupportsWrite {
+public class ArcticSparkTable implements Table, SupportsRead, SupportsWrite {
   private static final Set<String> RESERVED_PROPERTIES = Sets.newHashSet("provider", "format", "current-snapshot-id");
   private static final Set<TableCapability> CAPABILITIES = ImmutableSet.of(
       TableCapability.BATCH_READ,
@@ -146,17 +145,6 @@ public class ArcticSparkTable implements org.apache.spark.sql.connector.catalog.
     return CAPABILITIES;
   }
 
-  @Override
-  public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) {
-    SparkScanBuilder scanBuilder = new SparkScanBuilder(sparkSession(), arcticTable, options);
-
-    if (requestedSchema != null) {
-      scanBuilder.pruneColumns(requestedSchema);
-    }
-
-    return scanBuilder;
-  }
-
   @Override
   public String toString() {
     return arcticTable.toString();
@@ -182,13 +170,26 @@ public class ArcticSparkTable implements org.apache.spark.sql.connector.catalog.
   }
 
   @Override
-  public WriteBuilder newWriteBuilder(LogicalWriteInfo 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());
+  public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) {
+    SparkScanBuilder scanBuilder = new SparkScanBuilder(sparkSession(), arcticTable, options);
+
+    if (requestedSchema != null) {
+      scanBuilder.pruneColumns(requestedSchema);
     }
+
+    return scanBuilder;
+  }
+
+  @Override
+  public WriteBuilder newWriteBuilder(LogicalWriteInfo info) {
+    return new ArcticSparkWriteBuilder(arcticTable, info);
+  }
+
+
+  //@Override
+  public boolean appendAsUpsert() {
+    return arcticTable.isKeyedTable() &&
+        Boolean.parseBoolean(arcticTable.properties().getOrDefault(
+                TableProperties.UPSERT_ENABLED, "false"));
   }
 }
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkWriterUtils.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/SupportsExtendIdentColumns.java
similarity index 65%
rename from spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkWriterUtils.java
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/SupportsExtendIdentColumns.java
index c63dbe13f..087cdcd8d 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkWriterUtils.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/SupportsExtendIdentColumns.java
@@ -6,9 +6,9 @@
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
- *
+ *  *
  *     http://www.apache.org/licenses/LICENSE-2.0
- *
+ *  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -16,25 +16,11 @@
  * limitations under the License.
  */
 
-package com.netease.arctic.spark.writer;
+package com.netease.arctic.spark.table;
 
-import org.apache.iceberg.DataFile;
-import org.apache.spark.sql.connector.write.WriterCommitMessage;
+import org.apache.spark.sql.connector.read.ScanBuilder;
 
+public interface SupportsExtendIdentColumns extends ScanBuilder {
 
-public class SparkWriterUtils {
-
-
-  public static class TaskCommit implements WriterCommitMessage {
-    private final DataFile[] taskFiles;
-
-    TaskCommit(DataFile[] taskFiles) {
-      this.taskFiles = taskFiles;
-    }
-
-    DataFile[] files() {
-      return taskFiles;
-    }
-  }
-
+  SupportsExtendIdentColumns withIdentifierColumns();
 }
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/SupportsUpsert.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/SupportsUpsert.java
new file mode 100644
index 000000000..735389b9f
--- /dev/null
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/table/SupportsUpsert.java
@@ -0,0 +1,46 @@
+
+/*
+ * 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.table;
+
+import org.apache.spark.sql.connector.catalog.Table;
+
+/**
+ * A mix-in interface of {@link Table}, to indicate that can handle update or delete by upsert.
+ */
+public interface SupportsUpsert extends Table {
+
+  String UPSERT_OP_COLUMN_NAME = "_arctic_upsert_op";
+  String UPSERT_OP_VALUE_INSERT = "I";
+  String UPSERT_OP_VALUE_DELETE = "D";
+
+
+
+  SupportsExtendIdentColumns newScanBuilder();
+
+  boolean requireAdditionIdentifierColumns();
+
+
+  /**
+   * will table handle insert as upsert
+   *
+   * @return true if table require insert as upsert
+   */
+  boolean appendAsUpsert();
+}
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/ArcticSparkWriteBuilder.java
similarity index 53%
rename from spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/UnkeyedSparkWriteBuilder.java
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/ArcticSparkWriteBuilder.java
index 52207280d..a203e353d 100644
--- 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/ArcticSparkWriteBuilder.java
@@ -6,9 +6,9 @@
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
- *
+ *  *
  *     http://www.apache.org/licenses/LICENSE-2.0
- *
+ *  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -18,7 +18,7 @@
 
 package com.netease.arctic.spark.writer;
 
-import com.netease.arctic.table.UnkeyedTable;
+import com.netease.arctic.table.ArcticTable;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -29,60 +29,73 @@ 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;
-import org.apache.spark.sql.types.StructType;
 import org.apache.spark.sql.util.CaseInsensitiveStringMap;
 
-import java.util.Locale;
+public class ArcticSparkWriteBuilder implements WriteBuilder, SupportsDynamicOverwrite, SupportsOverwrite {
 
-public class UnkeyedSparkWriteBuilder implements WriteBuilder, SupportsDynamicOverwrite, SupportsOverwrite {
+  public interface ArcticWrite {
 
-  private final UnkeyedTable table;
-  private final CaseInsensitiveStringMap options;
-  private final String overwriteMode;
-  private boolean overwriteDynamic = false;
-  private boolean overwriteByFilter = false;
-  private Expression overwriteExpr = null;
+    BatchWrite asBatchAppend();
 
-  private StructType dsSchema = null;
+    BatchWrite asDynamicOverwrite();
 
-  public UnkeyedSparkWriteBuilder(UnkeyedTable table, LogicalWriteInfo info) {
-    this.table = table;
+    BatchWrite asOverwriteByFilter(Expression overwriteExpr);
+
+    BatchWrite asUpsertWrite();
+  }
+
+  protected final CaseInsensitiveStringMap options;
+
+  protected Expression overwriteExpr = null;
+
+  private WriteMode writeMode = WriteMode.APPEND;
+  private final ArcticWrite write;
+
+  public ArcticSparkWriteBuilder(ArcticTable table, LogicalWriteInfo info) {
     this.options = info.options();
-    this.overwriteMode = options.containsKey("overwrite-mode") ?
-        options.get("overwrite-mode").toLowerCase(Locale.ROOT) : null;
-    this.overwriteDynamic = "dynamic".equals(overwriteMode);
-    this.dsSchema = info.schema();
+    if (options.containsKey(WriteMode.WRITE_MODE_KEY)) {
+      this.writeMode = WriteMode.getWriteMode(options.get(WriteMode.WRITE_MODE_KEY));
+    }
+
+    if (table.isKeyedTable()) {
+      write = new KeyedSparkBatchWrite(table.asKeyedTable(), info.schema());
+    } else {
+      write = new UnkeyedSparkBatchWrite(table.asUnkeyedTable(), info.schema());
+    }
   }
 
   @Override
   public WriteBuilder overwriteDynamicPartitions() {
-    Preconditions.checkState(!overwriteByFilter, "Cannot overwrite dynamically and by filter: %s", overwriteExpr);
-    this.overwriteDynamic = true;
+    Preconditions.checkState(overwriteExpr == null, "Cannot overwrite dynamically and by filter: %s", overwriteExpr);
+    writeMode = WriteMode.OVERWRITE_DYNAMIC;
     return this;
   }
 
   @Override
   public WriteBuilder overwrite(Filter[] filters) {
     this.overwriteExpr = SparkFilters.convert(filters);
+    String overwriteMode = options.getOrDefault("overwrite-mode", "null");
     if (overwriteExpr == Expressions.alwaysTrue() && "dynamic".equals(overwriteMode)) {
-      // use the write option to override truncating the table. use dynamic overwrite instead.
-      this.overwriteDynamic = true;
+      writeMode = WriteMode.OVERWRITE_DYNAMIC;
     } else {
-      Preconditions.checkState(!overwriteDynamic, "Cannot overwrite dynamically and by filter: %s", overwriteExpr);
-      this.overwriteByFilter = true;
+      writeMode = WriteMode.OVERWRITE_BY_FILTER;
     }
     return this;
   }
 
   @Override
   public BatchWrite buildForBatch() {
-    UnkeyedSparkBatchWrite write = new UnkeyedSparkBatchWrite(table, dsSchema);
-    if (overwriteByFilter) {
-      return write.asOverwriteByFilter(overwriteExpr);
-    } else if (overwriteDynamic) {
-      return write.asDynamicOverwrite();
-    } else {
-      throw new UnsupportedOperationException("support insert overwrite only");
+    switch (writeMode) {
+      case APPEND:
+        return write.asBatchAppend();
+      case OVERWRITE_BY_FILTER:
+        return write.asOverwriteByFilter(overwriteExpr);
+      case OVERWRITE_DYNAMIC:
+        return write.asDynamicOverwrite();
+      case UPSERT:
+        return write.asUpsertWrite();
+      default:
+        throw new UnsupportedOperationException("unsupported write mode: " + writeMode);
     }
   }
 }
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/KeyedSparkBatchWrite.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/KeyedSparkBatchWrite.java
index 44a820007..3ff943fc9 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/KeyedSparkBatchWrite.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/KeyedSparkBatchWrite.java
@@ -26,8 +26,6 @@ import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.io.TaskWriter;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
-import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.util.PropertyUtil;
 import org.apache.iceberg.util.Tasks;
 import org.apache.spark.sql.catalyst.InternalRow;
@@ -39,9 +37,9 @@ import org.apache.spark.sql.connector.write.WriterCommitMessage;
 import org.apache.spark.sql.types.StructType;
 
 import java.io.Serializable;
-import java.util.Arrays;
 import java.util.Map;
 
+import static com.netease.arctic.spark.writer.WriteTaskCommit.files;
 import static org.apache.iceberg.TableProperties.COMMIT_MAX_RETRY_WAIT_MS;
 import static org.apache.iceberg.TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT;
 import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS;
@@ -51,7 +49,7 @@ 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 KeyedSparkBatchWrite {
+public class KeyedSparkBatchWrite implements ArcticSparkWriteBuilder.ArcticWrite {
   private final KeyedTable table;
   private final StructType dsSchema;
 
@@ -63,18 +61,26 @@ public class KeyedSparkBatchWrite {
     this.transactionId = table.beginTransaction(null);
   }
 
-  BatchWrite asBatchAppend() {
-    return new BatchAppend();
+  @Override
+  public BatchWrite asBatchAppend() {
+    return new AppendWrite();
   }
 
-  BatchWrite asDynamicOverwrite() {
+  @Override
+  public BatchWrite asDynamicOverwrite() {
     return new DynamicOverwrite();
   }
 
-  BatchWrite asOverwriteByFilter(Expression overwriteExpr) {
+  @Override
+  public BatchWrite asOverwriteByFilter(Expression overwriteExpr) {
     return new OverwriteByFilter(overwriteExpr);
   }
 
+  @Override
+  public BatchWrite asUpsertWrite() {
+    return new UpsertWrite();
+  }
+
   private abstract class BaseBatchWrite implements BatchWrite {
     private boolean isOverwrite;
 
@@ -82,11 +88,6 @@ public class KeyedSparkBatchWrite {
       this.isOverwrite = isOverwrite;
     }
 
-    @Override
-    public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
-      return new WriterFactory(table, dsSchema, transactionId, this.isOverwrite);
-    }
-
     @Override
     public void abort(WriterCommitMessage[] messages) {
       Map<String, String> props = table.properties();
@@ -104,20 +105,23 @@ public class KeyedSparkBatchWrite {
     }
   }
 
-  private class BatchAppend extends BaseBatchWrite {
+  private class AppendWrite extends BaseBatchWrite {
 
-    BatchAppend() {
+    AppendWrite() {
       super(false);
     }
 
     @Override
-    public void commit(WriterCommitMessage[] messages) {
-      AppendFiles append = table.baseTable().newAppend();
+    public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
+      return new ChangeWriteFactory(table, dsSchema, transactionId);
+    }
 
+    @Override
+    public void commit(WriterCommitMessage[] messages) {
+      AppendFiles append = table.changeTable().newAppend();
       for (DataFile file : files(messages)) {
         append.appendFile(file);
       }
-
       append.commit();
     }
   }
@@ -128,6 +132,11 @@ public class KeyedSparkBatchWrite {
       super(true);
     }
 
+    @Override
+    public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
+      return new BaseWriterFactory(table, dsSchema, transactionId);
+    }
+
     @Override
     public void commit(WriterCommitMessage[] messages) {
       RewritePartitions rewritePartitions = table.newRewritePartitions();
@@ -148,6 +157,11 @@ public class KeyedSparkBatchWrite {
       this.overwriteExpr = overwriteExpr;
     }
 
+    @Override
+    public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
+      return new BaseWriterFactory(table, dsSchema, transactionId);
+    }
+
     @Override
     public void commit(WriterCommitMessage[] messages) {
       OverwriteBaseFiles overwriteBaseFiles = table.newOverwriteBaseFiles();
@@ -162,18 +176,42 @@ public class KeyedSparkBatchWrite {
     }
   }
 
-  private static class WriterFactory implements DataWriterFactory, Serializable {
-    private final KeyedTable table;
-    private final StructType dsSchema;
-    private final long transactionId;
+  private class UpsertWrite extends BaseBatchWrite {
+    UpsertWrite() {
+      super(true);
+    }
 
-    private boolean isOverwrite;
+    @Override
+    public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
+      return new ChangeWriteFactory(table, dsSchema, transactionId);
+    }
 
-    WriterFactory(KeyedTable table, StructType dsSchema, long transactionId, boolean isOverwrite) {
+    @Override
+    public void commit(WriterCommitMessage[] messages) {
+      AppendFiles append = table.changeTable().newAppend();
+      for (DataFile file : files(messages)) {
+        append.appendFile(file);
+      }
+      append.commit();
+    }
+  }
+
+  private abstract static class AbstractWriterFactory implements DataWriterFactory, Serializable {
+    protected final KeyedTable table;
+    protected final StructType dsSchema;
+    protected final long transactionId;
+
+    AbstractWriterFactory(KeyedTable table, StructType dsSchema, long transactionId) {
       this.table = table;
       this.dsSchema = dsSchema;
       this.transactionId = transactionId;
-      this.isOverwrite = isOverwrite;
+    }
+  }
+
+  private static class BaseWriterFactory extends AbstractWriterFactory {
+
+    BaseWriterFactory(KeyedTable table, StructType dsSchema, long transactionId) {
+      super(table, dsSchema, transactionId);
     }
 
     @Override
@@ -183,17 +221,26 @@ public class KeyedSparkBatchWrite {
           .withPartitionId(partitionId)
           .withTaskId(taskId)
           .withDataSourceSchema(dsSchema)
-          .newBaseWriter(this.isOverwrite);
-      return new InternalRowDataWriter(writer);
+          .newBaseWriter(true);
+      return new SimpleInternalRowDataWriter(writer);
     }
   }
 
-  private static Iterable<DataFile> files(WriterCommitMessage[] messages) {
-    if (messages.length > 0) {
-      return Iterables.concat(Iterables.transform(Arrays.asList(messages), message -> message != null ?
-          ImmutableList.copyOf(((SparkWriterUtils.TaskCommit) message).files()) :
-          ImmutableList.of()));
+  private static class ChangeWriteFactory extends AbstractWriterFactory {
+
+    ChangeWriteFactory(KeyedTable table, StructType dsSchema, long transactionId) {
+      super(table, dsSchema, transactionId);
+    }
+
+    @Override
+    public DataWriter<InternalRow> createWriter(int partitionId, long taskId) {
+      TaskWriter<InternalRow> writer = TaskWriters.of(table)
+          .withTransactionId(transactionId)
+          .withPartitionId(partitionId)
+          .withTaskId(taskId)
+          .withDataSourceSchema(dsSchema)
+          .newChangeWriter();
+      return new SimpleInternalRowDataWriter(writer);
     }
-    return ImmutableList.of();
   }
 }
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/KeyedSparkWriteBuilder.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/KeyedSparkWriteBuilder.java
deleted file mode 100644
index 596142912..000000000
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/KeyedSparkWriteBuilder.java
+++ /dev/null
@@ -1,87 +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.table.KeyedTable;
-import org.apache.iceberg.expressions.Expression;
-import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.spark.SparkFilters;
-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;
-import org.apache.spark.sql.types.StructType;
-import org.apache.spark.sql.util.CaseInsensitiveStringMap;
-
-import java.util.Locale;
-
-public class KeyedSparkWriteBuilder implements WriteBuilder, SupportsDynamicOverwrite, SupportsOverwrite {
-  private final KeyedTable table;
-  private final CaseInsensitiveStringMap options;
-  private final String overwriteMode;
-  private boolean overwriteDynamic = false;
-  private boolean overwriteByFilter = false;
-  private Expression overwriteExpr = null;
-
-  private StructType dsSchema = null;
-
-  public KeyedSparkWriteBuilder(KeyedTable table, LogicalWriteInfo info) {
-    this.table = table;
-    this.options = info.options();
-    this.overwriteMode = options.containsKey("overwrite-mode") ?
-        options.get("overwrite-mode").toLowerCase(Locale.ROOT) : null;
-    this.overwriteDynamic = "dynamic".equals(overwriteMode);
-    this.dsSchema = info.schema();
-  }
-
-  @Override
-  public WriteBuilder overwriteDynamicPartitions() {
-    Preconditions.checkState(!overwriteByFilter, "Cannot overwrite dynamically and by filter: %s", overwriteExpr);
-    this.overwriteDynamic = true;
-    return this;
-  }
-
-  @Override
-  public WriteBuilder overwrite(Filter[] filters) {
-    this.overwriteExpr = SparkFilters.convert(filters);
-    if (overwriteExpr == Expressions.alwaysTrue() && "dynamic".equals(overwriteMode)) {
-      // use the write option to override truncating the table. use dynamic overwrite instead.
-      this.overwriteDynamic = true;
-    } else {
-      Preconditions.checkState(!overwriteDynamic, "Cannot overwrite dynamically and by filter: %s", overwriteExpr);
-      this.overwriteByFilter = true;
-    }
-    return this;
-  }
-
-  @Override
-  public BatchWrite buildForBatch() {
-    KeyedSparkBatchWrite write = new KeyedSparkBatchWrite(table, dsSchema);
-    if (overwriteByFilter) {
-      return write.asOverwriteByFilter(overwriteExpr);
-    } else if (overwriteDynamic) {
-      return write.asDynamicOverwrite();
-    } else {
-      throw new UnsupportedOperationException("support insert overwrite only");
-    }
-  }
-}
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/InternalRowDataWriter.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SimpleInternalRowDataWriter.java
similarity index 89%
rename from spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/InternalRowDataWriter.java
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SimpleInternalRowDataWriter.java
index 51b369308..8eb1709d5 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/InternalRowDataWriter.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SimpleInternalRowDataWriter.java
@@ -26,10 +26,10 @@ import org.apache.spark.sql.connector.write.WriterCommitMessage;
 
 import java.io.IOException;
 
-public class InternalRowDataWriter implements DataWriter<InternalRow> {
+public class SimpleInternalRowDataWriter implements DataWriter<InternalRow> {
   final TaskWriter<InternalRow> writer;
 
-  public InternalRowDataWriter(TaskWriter<InternalRow> writer) {
+  public SimpleInternalRowDataWriter(TaskWriter<InternalRow> writer) {
     this.writer = writer;
   }
 
@@ -41,7 +41,7 @@ public class InternalRowDataWriter implements DataWriter<InternalRow> {
   @Override
   public WriterCommitMessage commit() throws IOException {
     WriteResult result = writer.complete();
-    return new SparkWriterUtils.TaskCommit(result.dataFiles());
+    return new WriteTaskCommit(result.dataFiles());
   }
 
   @Override
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/UnkeyedSparkBatchWrite.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/UnkeyedSparkBatchWrite.java
index 3de8cb1ba..bdb532adb 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/UnkeyedSparkBatchWrite.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/UnkeyedSparkBatchWrite.java
@@ -18,16 +18,14 @@
 
 package com.netease.arctic.spark.writer;
 
-
 import com.netease.arctic.spark.io.TaskWriters;
 import com.netease.arctic.table.UnkeyedTable;
+import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.OverwriteFiles;
 import org.apache.iceberg.ReplacePartitions;
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.io.TaskWriter;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
-import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.util.PropertyUtil;
 import org.apache.iceberg.util.Tasks;
 import org.apache.spark.sql.catalyst.InternalRow;
@@ -39,9 +37,9 @@ import org.apache.spark.sql.connector.write.WriterCommitMessage;
 import org.apache.spark.sql.types.StructType;
 
 import java.io.Serializable;
-import java.util.Arrays;
 import java.util.Map;
 
+import static com.netease.arctic.spark.writer.WriteTaskCommit.files;
 import static org.apache.iceberg.TableProperties.COMMIT_MAX_RETRY_WAIT_MS;
 import static org.apache.iceberg.TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT;
 import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS;
@@ -51,7 +49,7 @@ 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 UnkeyedSparkBatchWrite {
+public class UnkeyedSparkBatchWrite implements ArcticSparkWriteBuilder.ArcticWrite {
 
   private final UnkeyedTable table;
   private final StructType dsSchema;
@@ -61,25 +59,27 @@ public class UnkeyedSparkBatchWrite {
     this.dsSchema = dsSchema;
   }
 
-  BatchWrite asDynamicOverwrite() {
+  @Override
+  public BatchWrite asBatchAppend() {
+    return new AppendWrite();
+  }
+
+  @Override
+  public BatchWrite asDynamicOverwrite() {
     return new DynamicOverwrite();
   }
 
-  BatchWrite asOverwriteByFilter(Expression overwriteExpr) {
+  @Override
+  public BatchWrite asOverwriteByFilter(Expression overwriteExpr) {
     return new OverwriteByFilter(overwriteExpr);
   }
 
-  private abstract class BaseBatchWrite implements BatchWrite {
-    private boolean isOverwrite;
-
-    BaseBatchWrite(boolean isOverwrite) {
-      this.isOverwrite = isOverwrite;
-    }
+  @Override
+  public BatchWrite asUpsertWrite() {
+    return new UpsertWrite();
+  }
 
-    @Override
-    public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
-      return new WriterFactory(table, dsSchema, this.isOverwrite);
-    }
+  private abstract class BaseBatchWrite implements BatchWrite {
 
     @Override
     public void abort(WriterCommitMessage[] messages) {
@@ -98,10 +98,28 @@ public class UnkeyedSparkBatchWrite {
     }
   }
 
+  private class AppendWrite extends BaseBatchWrite {
+
+    @Override
+    public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
+      return new WriterFactory(table, dsSchema, false);
+    }
+
+    @Override
+    public void commit(WriterCommitMessage[] messages) {
+      AppendFiles appendFiles = table.newAppend();
+      for (DataFile file : files(messages)) {
+        appendFiles.appendFile(file);
+      }
+      appendFiles.commit();
+    }
+  }
+
   private class DynamicOverwrite extends BaseBatchWrite {
 
-    DynamicOverwrite() {
-      super(true);
+    @Override
+    public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
+      return new WriterFactory(table, dsSchema, true);
     }
 
     @Override
@@ -118,10 +136,14 @@ public class UnkeyedSparkBatchWrite {
     private final Expression overwriteExpr;
 
     private OverwriteByFilter(Expression overwriteExpr) {
-      super(true);
       this.overwriteExpr = overwriteExpr;
     }
 
+    @Override
+    public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
+      return new WriterFactory(table, dsSchema, true);
+    }
+
     @Override
     public void commit(WriterCommitMessage[] messages) {
       OverwriteFiles overwriteFiles = table.newOverwrite();
@@ -133,6 +155,19 @@ public class UnkeyedSparkBatchWrite {
     }
   }
 
+  private class UpsertWrite extends BaseBatchWrite {
+    @Override
+    public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
+      return new DeltaUpsertWriteFactory(table, dsSchema);
+    }
+
+    @Override
+    public void commit(WriterCommitMessage[] messages) {
+      // TODO: issue #173 - implement upsert commit
+      throw new UnsupportedOperationException("Upsert write is not supported");
+    }
+  }
+
   private static class WriterFactory implements DataWriterFactory, Serializable {
     private final UnkeyedTable table;
     private final StructType dsSchema;
@@ -152,17 +187,20 @@ public class UnkeyedSparkBatchWrite {
           .withTaskId(taskId)
           .withDataSourceSchema(dsSchema)
           .newBaseWriter(this.isOverwrite);
-      return new InternalRowDataWriter(writer);
+      return new SimpleInternalRowDataWriter(writer);
     }
   }
 
-  private static Iterable<DataFile> files(WriterCommitMessage[] messages) {
-    if (messages.length > 0) {
-      return Iterables.concat(Iterables.transform(Arrays.asList(messages), message -> message != null ?
-          ImmutableList.copyOf(((SparkWriterUtils.TaskCommit) message).files()) :
-          ImmutableList.of()));
+  private static class DeltaUpsertWriteFactory extends WriterFactory {
+
+    DeltaUpsertWriteFactory(UnkeyedTable table, StructType dsSchema) {
+      super(table, dsSchema, false);
     }
-    return ImmutableList.of();
-  }
 
+    @Override
+    public DataWriter<InternalRow> createWriter(int partitionId, long taskId) {
+      // TODO: issues-173 - support upsert data writer
+      return null;
+    }
+  }
 }
diff --git a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/ExtendedArcticCommandStrategy.scala b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/WriteMode.java
similarity index 56%
rename from spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/ExtendedArcticCommandStrategy.scala
rename to spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/WriteMode.java
index adb441243..f54955afb 100644
--- a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/ExtendedArcticCommandStrategy.scala
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/WriteMode.java
@@ -6,9 +6,9 @@
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
- *
+ *  *
  *     http://www.apache.org/licenses/LICENSE-2.0
- *
+ *  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -16,21 +16,27 @@
  * limitations under the License.
  */
 
-package com.netease.arctic.spark.sql.execution
+package com.netease.arctic.spark.writer;
 
-import com.netease.arctic.spark.sql.catalyst.plans.MigrateToArcticLogicalPlan
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.execution.SparkPlan
-import org.apache.spark.sql.{SparkSession, Strategy}
+public enum WriteMode {
+  OVERWRITE_BY_FILTER("overwrite-by-filter"),
+  OVERWRITE_DYNAMIC("overwrite-dynamic"),
+  APPEND("append"),
+  UPSERT("upsert");
 
-case class ExtendedArcticCommandStrategy(spark: SparkSession) extends Strategy {
+  public static final String WRITE_MODE_KEY = "write-mode";
 
-  override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
-    case MigrateToArcticLogicalPlan(command) =>
-      println("create migrate to arctic command logical")
-      MigrateToArcticExec(command)::Nil
-
-    case _ => Nil
+  public final String mode;
+  WriteMode(String mode) {
+    this.mode = mode;
   }
 
+  public static WriteMode getWriteMode(String mode) {
+    for (WriteMode m : values()) {
+      if (m.mode.equalsIgnoreCase(mode)) {
+        return m;
+      }
+    }
+    throw new IllegalArgumentException("Invalid write mode: " + mode);
+  }
 }
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/WriteTaskCommit.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/WriteTaskCommit.java
new file mode 100644
index 000000000..d3834296f
--- /dev/null
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/WriteTaskCommit.java
@@ -0,0 +1,47 @@
+/*
+ * 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 org.apache.iceberg.DataFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.spark.sql.connector.write.WriterCommitMessage;
+
+import java.util.Arrays;
+
+public class WriteTaskCommit implements WriterCommitMessage {
+  private final DataFile[] taskFiles;
+
+  WriteTaskCommit(DataFile[] taskFiles) {
+    this.taskFiles = taskFiles;
+  }
+
+  DataFile[] files() {
+    return taskFiles;
+  }
+
+  public static Iterable<DataFile> files(WriterCommitMessage[] messages) {
+    if (messages.length > 0) {
+      return Iterables.concat(Iterables.transform(Arrays.asList(messages), message -> message != null ?
+          ImmutableList.copyOf(((WriteTaskCommit) message).files()) :
+          ImmutableList.of()));
+    }
+    return ImmutableList.of();
+  }
+}
diff --git a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/ArcticSparkExtensions.scala b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/ArcticSparkExtensions.scala
index 25bb3a91d..07ababffc 100644
--- a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/ArcticSparkExtensions.scala
+++ b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/ArcticSparkExtensions.scala
@@ -22,7 +22,7 @@ package com.netease.arctic.spark
 import com.netease.arctic.spark.sql.catalyst.analysis.ResolveArcticCommand
 import com.netease.arctic.spark.sql.catalyst.parser.ArcticSqlExtensionsParser
 import com.netease.arctic.spark.sql.execution
-import com.netease.arctic.spark.sql.optimize.OptimizeWriteRule
+import com.netease.arctic.spark.sql.optimize.{OptimizeWriteRule, RewriteAppendArcticTable, RewriteDeleteFromArcticTable, RewriteUpdateArcticTable}
 import org.apache.spark.sql.SparkSessionExtensions
 import org.apache.spark.sql.catalyst.analysis.{AlignRowLevelOperations, RowLevelOperationsPredicateCheck}
 import org.apache.spark.sql.catalyst.optimizer._
@@ -33,21 +33,32 @@ class ArcticSparkExtensions extends (SparkSessionExtensions => Unit) {
     extensions.injectParser {
       case (_, parser) => new ArcticSqlExtensionsParser(parser)
     }
-    // analyzer extensions
+    // resolve arctic command
+    extensions.injectResolutionRule { spark => ResolveArcticCommand(spark) }
+
+    // iceberg analyzer rules
     extensions.injectPostHocResolutionRule { _ => AlignRowLevelOperations }
-    extensions.injectResolutionRule{ spark => ResolveArcticCommand(spark) }
     extensions.injectCheckRule { _ => RowLevelOperationsPredicateCheck }
 
-    // optimizer extensions
+    // iceberg optimizer rules
     extensions.injectOptimizerRule { _ => OptimizeConditionsInRowLevelOperations }
     extensions.injectOptimizerRule { _ => PullupCorrelatedPredicatesInRowLevelOperations }
     extensions.injectOptimizerRule { spark => RewriteDelete(spark) }
     extensions.injectOptimizerRule { spark => RewriteUpdate(spark) }
     extensions.injectOptimizerRule { spark => RewriteMergeInto(spark) }
+    // arctic optimizer rules
+    extensions.injectOptimizerRule { spark => RewriteAppendArcticTable(spark) }
+    extensions.injectOptimizerRule { spark => RewriteDeleteFromArcticTable(spark) }
+    extensions.injectOptimizerRule { spark => RewriteUpdateArcticTable(spark) }
+
+    // arctic optimizer rules
     extensions.injectPreCBORule(OptimizeWriteRule)
 
-    // planner extensions
-    extensions.injectPlannerStrategy { spark => execution.ExtendedArcticUnkeyedStrategy(spark) }
-    extensions.injectPlannerStrategy { spark => execution.ExtendedArcticCommandStrategy(spark) }
+    // iceberg strategy rules
+    extensions.injectPlannerStrategy { spark => execution.ExtendedIcebergStrategy(spark) }
+
+    // arctic strategy rules
+    extensions.injectPlannerStrategy { spark => execution.ExtendedArcticStrategy(spark) }
   }
+
 }
diff --git a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/ArcticExtensionUtils.scala b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/ArcticExtensionUtils.scala
new file mode 100644
index 000000000..c795b1215
--- /dev/null
+++ b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/ArcticExtensionUtils.scala
@@ -0,0 +1,75 @@
+/*
+ * 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.sql
+
+import com.netease.arctic.spark.table.{ArcticSparkTable, SupportsUpsert}
+import org.apache.spark.sql.connector.catalog.Table
+import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, SubqueryAlias}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+import scala.annotation.tailrec
+
+
+object ArcticExtensionUtils {
+
+  implicit class ArcticTableHelper(table: Table) {
+    def asArcticTable: ArcticSparkTable = {
+      table match {
+        case arcticTable: ArcticSparkTable => arcticTable
+        case _ => throw new IllegalArgumentException(s"$table is not an arctic table")
+      }
+    }
+
+    def asUpsertWrite: SupportsUpsert = {
+      table match {
+        case arcticTable: SupportsUpsert => arcticTable
+        case _ => throw new IllegalArgumentException(s"$table is not an upsert-able table")
+      }
+    }
+  }
+
+  implicit class ArcticRelationHelper(plan: LogicalPlan) {
+    def asTableRelation: DataSourceV2Relation = {
+      ArcticExtensionUtils.asTableRelation(plan)
+    }
+  }
+
+  @tailrec
+  def isArcticRelation(plan: LogicalPlan): Boolean = {
+    def isArcticTable(relation: DataSourceV2Relation): Boolean = relation.table match {
+      case _: ArcticSparkTable => true
+      case _ => false
+    }
+
+    plan match {
+      case s: SubqueryAlias => isArcticRelation(s.child)
+      case r: DataSourceV2Relation => isArcticTable(r)
+      case _ => false
+    }
+  }
+
+  def asTableRelation(plan: LogicalPlan): DataSourceV2Relation = {
+    plan match {
+      case s: SubqueryAlias => asTableRelation(s.child)
+      case r: DataSourceV2Relation => r
+      case _ => throw new IllegalArgumentException("Expected a DataSourceV2Relation")
+    }
+  }
+
+}
diff --git a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/ExtendedArcticStrategy.scala b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/ExtendedArcticStrategy.scala
new file mode 100644
index 000000000..058c2af07
--- /dev/null
+++ b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/ExtendedArcticStrategy.scala
@@ -0,0 +1,66 @@
+/*
+ * 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.sql.execution
+
+import com.netease.arctic.spark.sql.catalyst.plans.MigrateToArcticLogicalPlan
+import com.netease.arctic.spark.writer.WriteMode
+import org.apache.spark.sql.catalyst.analysis.ResolvedTable
+import org.apache.spark.sql.catalyst.plans.logical.{CreateTableAsSelect, DescribeRelation, LogicalPlan}
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.command.CreateTableLikeCommand
+import org.apache.spark.sql.execution.datasources.v2.CreateTableAsSelectExec
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.spark.sql.{SparkSession, Strategy}
+
+import scala.collection.JavaConverters
+
+case class ExtendedArcticStrategy(spark: SparkSession) extends Strategy {
+
+  override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
+    case CreateTableAsSelect(catalog, ident, parts, query, props, options, ifNotExists) =>
+      var propertiesMap: Map[String, String] = props
+      var optionsMap: Map[String, String] = options
+      if (options.contains("primary.keys")) {
+        propertiesMap += ("primary.keys" -> options("primary.keys"))
+      }
+      if(propertiesMap.contains("primary.keys")) {
+        optionsMap += (WriteMode.WRITE_MODE_KEY -> WriteMode.OVERWRITE_DYNAMIC.mode)
+      }
+
+      val writeOptions = new CaseInsensitiveStringMap(JavaConverters.mapAsJavaMap(optionsMap))
+      CreateTableAsSelectExec(catalog, ident, parts, query, planLater(query),
+        propertiesMap, writeOptions, ifNotExists) :: Nil
+
+    case CreateTableLikeCommand(targetTable, sourceTable, storage, provider, properties, ifNotExists) =>
+      CreateArcticTableLikeExec(spark, targetTable, sourceTable, storage, provider, properties, ifNotExists) :: Nil
+
+    case DescribeRelation(r: ResolvedTable, partitionSpec, isExtended) =>
+      if (partitionSpec.nonEmpty) {
+        throw new RuntimeException("DESCRIBE does not support partition for v2 tables.")
+      }
+      DescribeKeyedTableExec(r.table, r.catalog, r.identifier, isExtended) :: Nil
+
+    case MigrateToArcticLogicalPlan(command) =>
+      println("create migrate to arctic command logical")
+      MigrateToArcticExec(command)::Nil
+
+    case _ => Nil
+  }
+
+}
diff --git a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/ExtendedArcticUnkeyedStrategy.scala b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/ExtendedIcebergStrategy.scala
similarity index 79%
rename from spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/ExtendedArcticUnkeyedStrategy.scala
rename to spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/ExtendedIcebergStrategy.scala
index 021a844be..a9445d028 100644
--- a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/ExtendedArcticUnkeyedStrategy.scala
+++ b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/execution/ExtendedIcebergStrategy.scala
@@ -35,29 +35,8 @@ import org.apache.spark.sql.{SparkSession, Strategy}
 import scala.collection.JavaConverters
 import scala.collection.JavaConverters.seqAsJavaList
 
-case class ExtendedArcticUnkeyedStrategy(spark: SparkSession) extends Strategy {
+case class ExtendedIcebergStrategy(spark: SparkSession) extends Strategy {
   override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match {
-    case CreateTableAsSelect(catalog, ident, parts, query, props, options, ifNotExists) =>
-      var propertiesMap: Map[String, String] = props
-      var optionsMap: Map[String, String] = options
-      if (options.contains("primary.keys")) {
-        propertiesMap += ("primary.keys" -> options("primary.keys"))
-        optionsMap += ("overwrite-mode" -> "dynamic")
-      }
-      val writeOptions = new CaseInsensitiveStringMap(JavaConverters.mapAsJavaMap(optionsMap))
-      CreateTableAsSelectExec(catalog, ident, parts, query, planLater(query),
-        propertiesMap, writeOptions, ifNotExists) :: Nil
-
-    case CreateTableLikeCommand(targetTable, sourceTable, storage, provider, properties, ifNotExists) =>
-      CreateArcticTableLikeExec(spark, targetTable, sourceTable, storage, provider, properties, ifNotExists) :: Nil
-
-    case DescribeRelation(r: ResolvedTable, partitionSpec, isExtended) =>
-      if (partitionSpec.nonEmpty) {
-        throw new RuntimeException("DESCRIBE does not support partition for v2 tables.")
-      }
-      DescribeKeyedTableExec(r.table, r.catalog, r.identifier, isExtended) :: Nil
-
-
     case DynamicFileFilter(scanPlan, fileFilterPlan, filterable) =>
       DynamicFileFilterExec(planLater(scanPlan), planLater(fileFilterPlan), filterable) :: Nil
 
diff --git a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/optimize/RewriteAppendArcticTable.scala b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/optimize/RewriteAppendArcticTable.scala
new file mode 100644
index 000000000..0b4926a44
--- /dev/null
+++ b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/optimize/RewriteAppendArcticTable.scala
@@ -0,0 +1,54 @@
+/*
+ * 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.sql.optimize
+
+import com.netease.arctic.spark.table.SupportsUpsert
+import com.netease.arctic.spark.writer.WriteMode
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.Inner
+import org.apache.spark.sql.catalyst.plans.logical.{AppendData, Join, JoinHint, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+case class RewriteAppendArcticTable(spark: SparkSession) extends Rule[LogicalPlan] {
+
+  import com.netease.arctic.spark.sql.ArcticExtensionUtils._
+
+  override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+    case i@AppendData(r: DataSourceV2Relation, query, writeOptions, isByName) if isArcticRelation(r) =>
+      val upsertWrite = r.table.asUpsertWrite
+      val (newQuery, options) = if (upsertWrite.appendAsUpsert()) {
+        val upsertQuery = rewriteAppendAsUpsertQuery(r, upsertWrite, query)
+        val upsertOptions = writeOptions + (WriteMode.WRITE_MODE_KEY -> WriteMode.UPSERT.mode)
+        (upsertQuery, upsertOptions)
+      } else {
+        (query, writeOptions)
+      }
+      i.copy(query = newQuery, writeOptions = options)
+  }
+
+  def rewriteAppendAsUpsertQuery(r: DataSourceV2Relation,
+                                 upsertWrite: SupportsUpsert,
+                                 query: LogicalPlan): LogicalPlan = {
+    // val joinCondition = upsertWrite.joinCondition()
+    val join = Join(r, query, Inner, None, JoinHint.NONE)
+    query
+  }
+
+}
diff --git a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/optimize/RewriteDeleteFromArcticTable.scala b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/optimize/RewriteDeleteFromArcticTable.scala
new file mode 100644
index 000000000..c5250f771
--- /dev/null
+++ b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/optimize/RewriteDeleteFromArcticTable.scala
@@ -0,0 +1,79 @@
+/*
+ * 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.sql.optimize
+
+import com.netease.arctic.spark.sql.ArcticExtensionUtils.{ArcticTableHelper, asTableRelation, isArcticRelation}
+import com.netease.arctic.spark.table.SupportsUpsert
+import com.netease.arctic.spark.writer.WriteMode
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, Expression, Literal}
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation}
+import org.apache.spark.sql.types.StructType
+
+case class RewriteDeleteFromArcticTable(spark: SparkSession) extends Rule[LogicalPlan] {
+
+  private val opCol = SupportsUpsert.UPSERT_OP_COLUMN_NAME
+  private val opDel = SupportsUpsert.UPSERT_OP_VALUE_DELETE
+
+  override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+    case _@DeleteFromTable(table, condition) if isArcticRelation(table) =>
+      val r = asTableRelation(table)
+      val upsert = r.table.asUpsertWrite
+      val query = buildUpsertQuery(r, upsert, condition)
+//      val options = Map.empty[String, String] ++ (WriteMode.WRITE_MODE_KEY, WriteMode.UPSERT.mode)
+      val options = Map.empty()
+      AppendData.byPosition(r, query, options)
+  }
+
+  def buildUpsertQuery(r: DataSourceV2Relation, upsert: SupportsUpsert, condition: Option[Expression]): LogicalPlan = {
+    val scanBuilder = upsert.newScanBuilder()
+    if (upsert.requireAdditionIdentifierColumns()) {
+      scanBuilder.withIdentifierColumns()
+    }
+    val scan = scanBuilder.build()
+    val outputAttr = toOutputAttrs(scan.readSchema(), r.output)
+    val valuesRelation = DataSourceV2ScanRelation(r, scan, outputAttr)
+
+    val matchValueQuery = if (condition.isDefined) {
+      Filter(condition.get, valuesRelation)
+    } else {
+      valuesRelation
+    }
+    val withOperation = Seq(Alias(Literal(opDel), opCol)()) ++ matchValueQuery.output
+    val deleteQuery = Project(withOperation, matchValueQuery)
+    deleteQuery
+  }
+
+  def toOutputAttrs(schema: StructType, attrs: Seq[AttributeReference]): Seq[AttributeReference] = {
+    val nameToAttr = attrs.map(_.name).zip(attrs).toMap
+    schema.map(f => AttributeReference(f.name, f.dataType, f.nullable, f.metadata)()).map {
+      a =>
+        nameToAttr.get(a.name) match {
+          case Some(ref) =>
+            // keep the attribute id if it was present in the relation
+            a.withExprId(ref.exprId)
+          case _ =>
+            // if the field is new, create a new attribute
+            AttributeReference(a.name, a.dataType, a.nullable, a.metadata)()
+        }
+    }
+  }
+}
diff --git a/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/optimize/RewriteUpdateArcticTable.scala b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/optimize/RewriteUpdateArcticTable.scala
new file mode 100644
index 000000000..fe77b0d13
--- /dev/null
+++ b/spark/v3.1/spark/src/main/scala/com/netease/arctic/spark/sql/optimize/RewriteUpdateArcticTable.scala
@@ -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.sql.optimize
+
+import com.netease.arctic.spark.sql.ArcticExtensionUtils.{ArcticTableHelper, asTableRelation, isArcticRelation}
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.expressions.Expression
+import org.apache.spark.sql.catalyst.plans.logical.{AppendData, Assignment, LogicalPlan, UpdateTable}
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+
+
+/**
+ * rewrite update table plan as append upsert data.
+ */
+case class RewriteUpdateArcticTable(spark: SparkSession) extends Rule[LogicalPlan] {
+  override def apply(plan: LogicalPlan): LogicalPlan = plan transform {
+    case u@UpdateTable(table, assignments, condition) if isArcticRelation(table) =>
+      val arcticRelation = asTableRelation(table)
+      val upsertWrite = arcticRelation.table.asUpsertWrite
+      val upsertQuery = buildUpsertQuery(arcticRelation, assignments, condition)
+      val options = Map.empty[String, String]
+      AppendData.byPosition(arcticRelation, upsertQuery, options)
+      u
+  }
+
+
+  def buildUpsertQuery(r: DataSourceV2Relation, assignments: Seq[Assignment],
+                       condition: Option[Expression]): LogicalPlan = {
+    // TODO: support arctic upsert query
+    r
+  }
+
+}
diff --git a/spark/v3.1/spark/src/main/scala/org/apache/spark/sql/catalyst/parser/ArcticExtendSparkSqlAstBuilder.scala b/spark/v3.1/spark/src/main/scala/org/apache/spark/sql/catalyst/parser/ArcticExtendSparkSqlAstBuilder.scala
index 68353379a..4d17aa507 100644
--- a/spark/v3.1/spark/src/main/scala/org/apache/spark/sql/catalyst/parser/ArcticExtendSparkSqlAstBuilder.scala
+++ b/spark/v3.1/spark/src/main/scala/org/apache/spark/sql/catalyst/parser/ArcticExtendSparkSqlAstBuilder.scala
@@ -125,7 +125,6 @@ class ArcticExtendSparkSqlAstBuilder(delegate: ParserInterface)
       case Some(query) =>
         val propertiesMap = buildProperties(primaryForCtas, properties)
         var writeOptions: Map[String, String] = Map.empty
-        writeOptions += ("overwrite-mode" -> "dynamic")
         CreateTableAsSelectStatement(
           table, query, partitioning, bucketSpec, propertiesMap, provider, options, location, comment,
           writeOptions, serdeInfo, external = external, ifNotExists = ifNotExists)


Test patch:
diff --git a/hive/src/test/java/com/netease/arctic/hive/ArcticHiveTestMain.java b/hive/src/test/java/com/netease/arctic/hive/ArcticHiveTestMain.java
index 026296329..0b690c4ee 100644
--- a/hive/src/test/java/com/netease/arctic/hive/ArcticHiveTestMain.java
+++ b/hive/src/test/java/com/netease/arctic/hive/ArcticHiveTestMain.java
@@ -19,6 +19,7 @@
 package com.netease.arctic.hive;
 
 import com.netease.arctic.hive.catalog.TestArcticHiveCatalog;
+import com.netease.arctic.hive.op.TestHiveSchemaUpdate;
 import com.netease.arctic.hive.op.TestOverwriteFiles;
 import com.netease.arctic.hive.op.TestRewritePartitions;
 import org.junit.AfterClass;
@@ -30,7 +31,8 @@ import org.junit.runners.Suite;
 @Suite.SuiteClasses({
     TestArcticHiveCatalog.class,
     TestRewritePartitions.class,
-    TestOverwriteFiles.class
+    TestOverwriteFiles.class,
+    TestHiveSchemaUpdate.class
 })
 public class ArcticHiveTestMain {
 
diff --git a/hive/src/test/java/com/netease/arctic/hive/op/TestHiveSchemaUpdate.java b/hive/src/test/java/com/netease/arctic/hive/op/TestHiveSchemaUpdate.java
new file mode 100644
index 000000000..1114b7de1
--- /dev/null
+++ b/hive/src/test/java/com/netease/arctic/hive/op/TestHiveSchemaUpdate.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.hive.op;
+
+import com.netease.arctic.hive.HiveTableTestBase;
+import com.netease.arctic.hive.utils.HiveSchemaUtil;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.types.Types;
+import org.apache.thrift.TException;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.List;
+
+public class TestHiveSchemaUpdate extends HiveTableTestBase {
+
+  @Test
+  public void testKeyedAdd() throws TException {
+    String testAddCol = "testAdd";
+    String testDoc = "test Doc";
+    testKeyedHiveTable.updateSchema().addColumn(testAddCol, Types.IntegerType.get(), testDoc).commit();
+    List<FieldSchema> fieldSchemas = hms.getClient().getFields(HIVE_DB_NAME, "test_pk_hive_table");
+    boolean isExpect = false;
+    for (FieldSchema fieldSchema : fieldSchemas) {
+      if (fieldSchema.getName().equalsIgnoreCase(testAddCol) && fieldSchema.getComment().equalsIgnoreCase(testDoc) &&
+          fieldSchema.getType().equals("int")) {
+        isExpect = true;
+      }
+    }
+    Assert.assertTrue(isExpect);
+    Assert.assertTrue(compareSchema(testKeyedHiveTable.schema(), testKeyedHiveTable.spec(), fieldSchemas));
+  }
+
+  @Test
+  public void testKeyedUpdate() throws TException {
+    String testUpdateCol = "testUpdate";
+    String testDoc = "test Doc";
+    testKeyedHiveTable.updateSchema().addColumn(testUpdateCol, Types.FloatType.get(), "init doc").commit();
+    testKeyedHiveTable.updateSchema().updateColumn(testUpdateCol, Types.DoubleType.get(), testDoc).commit();
+    List<FieldSchema> fieldSchemas = hms.getClient().getFields(HIVE_DB_NAME, "test_pk_hive_table");
+    boolean isExpect = false;
+    for (FieldSchema fieldSchema : fieldSchemas) {
+      if (fieldSchema.getName().equalsIgnoreCase(testUpdateCol) && fieldSchema.getComment().equalsIgnoreCase(testDoc) &&
+          fieldSchema.getType().equals("double")) {
+        isExpect = true;
+      }
+    }
+    Assert.assertTrue(isExpect);
+    Assert.assertTrue(compareSchema(testKeyedHiveTable.schema(), testKeyedHiveTable.spec(), fieldSchemas));
+  }
+
+  @Test
+  public void testUnKeyedAdd() throws TException {
+    String testAddCol = "testAdd";
+    String testDoc = "test Doc";
+    testHiveTable.updateSchema().addColumn(testAddCol, Types.IntegerType.get(), testDoc).commit();
+    List<FieldSchema> fieldSchemas = hms.getClient().getFields(HIVE_DB_NAME, "test_hive_table");
+    boolean isExpect = false;
+    for (FieldSchema fieldSchema : fieldSchemas) {
+      if (fieldSchema.getName().equalsIgnoreCase(testAddCol) && fieldSchema.getComment().equalsIgnoreCase(testDoc) &&
+          fieldSchema.getType().equals("int")) {
+        isExpect = true;
+      }
+    }
+    Assert.assertTrue(isExpect);
+    Assert.assertTrue(compareSchema(testHiveTable.schema(), testHiveTable.spec(), fieldSchemas));
+  }
+
+  @Test
+  public void testUnKeyedUpdate() throws TException {
+    String testUpdateCol = "testUpdate";
+    String testDoc = "test Doc";
+    testHiveTable.updateSchema().addColumn(testUpdateCol, Types.FloatType.get(), "init doc").commit();
+    testHiveTable.updateSchema().updateColumn(testUpdateCol, Types.DoubleType.get(), testDoc).commit();
+    List<FieldSchema> fieldSchemas = hms.getClient().getFields(HIVE_DB_NAME, "test_hive_table");
+    boolean isExpect = false;
+    for (FieldSchema fieldSchema : fieldSchemas) {
+      if (fieldSchema.getName().equalsIgnoreCase(testUpdateCol) && fieldSchema.getComment().equalsIgnoreCase(testDoc) &&
+          fieldSchema.getType().equals("double")) {
+        isExpect = true;
+      }
+    }
+    Assert.assertTrue(isExpect);
+    Assert.assertTrue(compareSchema(testHiveTable.schema(), testHiveTable.spec(), fieldSchemas));
+  }
+
+  boolean compareSchema(Schema schema, PartitionSpec spec, List<FieldSchema> hiveSchema) {
+    List<FieldSchema> convertFields = HiveSchemaUtil.hiveTableFields(schema, spec);
+    convertFields.forEach(fieldSchema -> {
+      fieldSchema.setName(fieldSchema.getName().toLowerCase());
+    });
+    if (convertFields.size() != hiveSchema.size()) {
+      return false;
+    }
+    for (FieldSchema fieldSchema : hiveSchema) {
+      if (!convertFields.contains(fieldSchema)) {
+        return false;
+      }
+    }
+    return 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 774254bc0..689754e92 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
@@ -53,7 +53,7 @@ import java.io.IOException;
 public class ArcticSparkHiveMainTest {
 
   @BeforeClass
-  public static void suiteSetup() throws IOException {
+  public static void suiteSetup() throws IOException, ClassNotFoundException {
     SparkTestContext.startAll();
   }
 
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 4bf500a7b..dd9f6bd04 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
@@ -121,7 +121,7 @@ public class SparkTestContext extends ExternalResource {
   }
 
   @BeforeClass
-  public static void startAll() throws IOException {
+  public static void startAll() throws IOException, ClassNotFoundException {
     int ref = testCount.incrementAndGet();
     if (ref == 1) {
       setUpTestDirAndArctic();
@@ -157,12 +157,17 @@ public class SparkTestContext extends ExternalResource {
     ams.handler().createCatalog(arctic);
   }
 
-  public static void setUpHMS() throws IOException {
+  public static void setUpHMS() throws IOException, ClassNotFoundException {
     System.out.println("======================== start hive metastore ========================= ");
     hms.start();
     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");
+    String hiveVersion = SparkTestContext.class.getClassLoader()
+        .loadClass("org.apache.hadoop.hive.metastore.HiveMetaStoreClient")
+        .getPackage()
+        .getImplementationVersion();
+    additionSparkConfigs.put("spark.sql.hive.metastore.version", hiveVersion);
+    additionSparkConfigs.put("spark.sql.hive.metastore.jars", "maven");
     //hive.metastore.client.capability.check
     additionSparkConfigs.put("hive.metastore.client.capability.check", "false");
 
diff --git a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/hive/TestHive.java b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/hive/TestHive.java
index 8d01868c8..fb69a6d81 100644
--- a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/hive/TestHive.java
+++ b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/hive/TestHive.java
@@ -17,7 +17,7 @@ public class TestHive extends SparkTestContext {
 
 
   @BeforeClass
-  public static void setup() throws IOException {
+  public static void setup() throws IOException, ClassNotFoundException {
     setUpTestDirAndArctic();
     setUpHMS();
     setUpSparkSession();