Repository: NetEase/arctic
Pull request #193 (2022-08-18 08:13:17)

Check input is validate at OverwriteFiles  && ReplacePartitions
check input is validate at OverwriteFiles  && ReplacePartitions


Patch:
diff --git a/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/CommitMetaProducer.java b/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/CommitMetaProducer.java
new file mode 100644
index 000000000..7e7fea02a
--- /dev/null
+++ b/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/CommitMetaProducer.java
@@ -0,0 +1,43 @@
+/**
+ * Autogenerated by Thrift Compiler (0.13.0)
+ *
+ * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
+ *  @generated
+ */
+package com.netease.arctic.ams.api;
+
+
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2022-08-22")
+public enum CommitMetaProducer implements org.apache.thrift.TEnum {
+  OPTIMIZE(0),
+  INGESTION(1);
+
+  private final int value;
+
+  private CommitMetaProducer(int value) {
+    this.value = value;
+  }
+
+  /**
+   * Get the integer value of this enum value, as defined in the Thrift IDL.
+   */
+  public int getValue() {
+    return value;
+  }
+
+  /**
+   * Find a the enum type by its integer value, as defined in the Thrift IDL.
+   * @return null if the value is not found.
+   */
+  @org.apache.thrift.annotation.Nullable
+  public static CommitMetaProducer findByValue(int value) { 
+    switch (value) {
+      case 0:
+        return OPTIMIZE;
+      case 1:
+        return INGESTION;
+      default:
+        return null;
+    }
+  }
+}
diff --git a/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/TableCommitMeta.java b/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/TableCommitMeta.java
index fd0b288b3..022958f0c 100644
--- a/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/TableCommitMeta.java
+++ b/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/TableCommitMeta.java
@@ -25,7 +25,7 @@
 package com.netease.arctic.ams.api;
 
 @SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
-@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2022-06-30")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2022-08-12")
 public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta, TableCommitMeta._Fields>, java.io.Serializable, Cloneable, Comparable<TableCommitMeta> {
   private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("TableCommitMeta");
 
@@ -34,15 +34,17 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
   private static final org.apache.thrift.protocol.TField CHANGES_FIELD_DESC = new org.apache.thrift.protocol.TField("changes", org.apache.thrift.protocol.TType.LIST, (short)3);
   private static final org.apache.thrift.protocol.TField COMMIT_TIME_FIELD_DESC = new org.apache.thrift.protocol.TField("commitTime", org.apache.thrift.protocol.TType.I64, (short)4);
   private static final org.apache.thrift.protocol.TField PROPERTIES_FIELD_DESC = new org.apache.thrift.protocol.TField("properties", org.apache.thrift.protocol.TType.MAP, (short)5);
+  private static final org.apache.thrift.protocol.TField COMMIT_META_PRODUCER_FIELD_DESC = new org.apache.thrift.protocol.TField("commitMetaProducer", org.apache.thrift.protocol.TType.I32, (short)6);
 
   private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new TableCommitMetaStandardSchemeFactory();
   private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new TableCommitMetaTupleSchemeFactory();
 
-  public @org.apache.thrift.annotation.Nullable com.netease.arctic.ams.api.TableIdentifier tableIdentifier; // required
-  public @org.apache.thrift.annotation.Nullable java.lang.String action; // required
+  public @org.apache.thrift.annotation.Nullable TableIdentifier tableIdentifier; // required
+  public @org.apache.thrift.annotation.Nullable String action; // required
   public @org.apache.thrift.annotation.Nullable java.util.List<TableChange> changes; // required
   public long commitTime; // required
-  public @org.apache.thrift.annotation.Nullable java.util.Map<java.lang.String,java.lang.String> properties; // required
+  public @org.apache.thrift.annotation.Nullable java.util.Map<String, String> properties; // required
+  public @org.apache.thrift.annotation.Nullable CommitMetaProducer commitMetaProducer; // required
 
   /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
   public enum _Fields implements org.apache.thrift.TFieldIdEnum {
@@ -50,9 +52,10 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
     ACTION((short)2, "action"),
     CHANGES((short)3, "changes"),
     COMMIT_TIME((short)4, "commitTime"),
-    PROPERTIES((short)5, "properties");
+    PROPERTIES((short)5, "properties"),
+    COMMIT_META_PRODUCER((short)6, "commitMetaProducer");
 
-    private static final java.util.Map<java.lang.String, _Fields> byName = new java.util.HashMap<java.lang.String, _Fields>();
+    private static final java.util.Map<String, _Fields> byName = new java.util.HashMap<String, _Fields>();
 
     static {
       for (_Fields field : java.util.EnumSet.allOf(_Fields.class)) {
@@ -76,6 +79,8 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
           return COMMIT_TIME;
         case 5: // PROPERTIES
           return PROPERTIES;
+        case 6: // COMMIT_META_PRODUCER
+          return COMMIT_META_PRODUCER;
         default:
           return null;
       }
@@ -87,7 +92,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
      */
     public static _Fields findByThriftIdOrThrow(int fieldId) {
       _Fields fields = findByThriftId(fieldId);
-      if (fields == null) throw new java.lang.IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+      if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
       return fields;
     }
 
@@ -95,14 +100,14 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
      * Find the _Fields constant that matches name, or null if its not found.
      */
     @org.apache.thrift.annotation.Nullable
-    public static _Fields findByName(java.lang.String name) {
+    public static _Fields findByName(String name) {
       return byName.get(name);
     }
 
     private final short _thriftId;
-    private final java.lang.String _fieldName;
+    private final String _fieldName;
 
-    _Fields(short thriftId, java.lang.String fieldName) {
+    _Fields(short thriftId, String fieldName) {
       _thriftId = thriftId;
       _fieldName = fieldName;
     }
@@ -111,7 +116,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
       return _thriftId;
     }
 
-    public java.lang.String getFieldName() {
+    public String getFieldName() {
       return _fieldName;
     }
   }
@@ -123,7 +128,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
   static {
     java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new java.util.EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
     tmpMap.put(_Fields.TABLE_IDENTIFIER, new org.apache.thrift.meta_data.FieldMetaData("tableIdentifier", org.apache.thrift.TFieldRequirementType.DEFAULT, 
-        new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, com.netease.arctic.ams.api.TableIdentifier.class)));
+        new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, TableIdentifier.class)));
     tmpMap.put(_Fields.ACTION, new org.apache.thrift.meta_data.FieldMetaData("action", org.apache.thrift.TFieldRequirementType.DEFAULT, 
         new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
     tmpMap.put(_Fields.CHANGES, new org.apache.thrift.meta_data.FieldMetaData("changes", org.apache.thrift.TFieldRequirementType.DEFAULT, 
@@ -135,6 +140,8 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
         new org.apache.thrift.meta_data.MapMetaData(org.apache.thrift.protocol.TType.MAP, 
             new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING), 
             new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING))));
+    tmpMap.put(_Fields.COMMIT_META_PRODUCER, new org.apache.thrift.meta_data.FieldMetaData("commitMetaProducer", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.ENUM        , "CommitMetaProducer")));
     metaDataMap = java.util.Collections.unmodifiableMap(tmpMap);
     org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(TableCommitMeta.class, metaDataMap);
   }
@@ -143,11 +150,12 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
   }
 
   public TableCommitMeta(
-    com.netease.arctic.ams.api.TableIdentifier tableIdentifier,
-    java.lang.String action,
+    TableIdentifier tableIdentifier,
+    String action,
     java.util.List<TableChange> changes,
     long commitTime,
-    java.util.Map<java.lang.String,java.lang.String> properties)
+    java.util.Map<String, String> properties,
+    CommitMetaProducer commitMetaProducer)
   {
     this();
     this.tableIdentifier = tableIdentifier;
@@ -156,6 +164,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
     this.commitTime = commitTime;
     setCommitTimeIsSet(true);
     this.properties = properties;
+    this.commitMetaProducer = commitMetaProducer;
   }
 
   /**
@@ -164,7 +173,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
   public TableCommitMeta(TableCommitMeta other) {
     __isset_bitfield = other.__isset_bitfield;
     if (other.isSetTableIdentifier()) {
-      this.tableIdentifier = new com.netease.arctic.ams.api.TableIdentifier(other.tableIdentifier);
+      this.tableIdentifier = new TableIdentifier(other.tableIdentifier);
     }
     if (other.isSetAction()) {
       this.action = other.action;
@@ -178,9 +187,12 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
     }
     this.commitTime = other.commitTime;
     if (other.isSetProperties()) {
-      java.util.Map<java.lang.String,java.lang.String> __this__properties = new java.util.HashMap<java.lang.String,java.lang.String>(other.properties);
+      java.util.Map<String, String> __this__properties = new java.util.HashMap<String, String>(other.properties);
       this.properties = __this__properties;
     }
+    if (other.isSetCommitMetaProducer()) {
+      this.commitMetaProducer = other.commitMetaProducer;
+    }
   }
 
   public TableCommitMeta deepCopy() {
@@ -195,14 +207,15 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
     setCommitTimeIsSet(false);
     this.commitTime = 0;
     this.properties = null;
+    this.commitMetaProducer = null;
   }
 
   @org.apache.thrift.annotation.Nullable
-  public com.netease.arctic.ams.api.TableIdentifier getTableIdentifier() {
+  public TableIdentifier getTableIdentifier() {
     return this.tableIdentifier;
   }
 
-  public TableCommitMeta setTableIdentifier(@org.apache.thrift.annotation.Nullable com.netease.arctic.ams.api.TableIdentifier tableIdentifier) {
+  public TableCommitMeta setTableIdentifier(@org.apache.thrift.annotation.Nullable TableIdentifier tableIdentifier) {
     this.tableIdentifier = tableIdentifier;
     return this;
   }
@@ -223,11 +236,11 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
   }
 
   @org.apache.thrift.annotation.Nullable
-  public java.lang.String getAction() {
+  public String getAction() {
     return this.action;
   }
 
-  public TableCommitMeta setAction(@org.apache.thrift.annotation.Nullable java.lang.String action) {
+  public TableCommitMeta setAction(@org.apache.thrift.annotation.Nullable String action) {
     this.action = action;
     return this;
   }
@@ -315,19 +328,19 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
     return (this.properties == null) ? 0 : this.properties.size();
   }
 
-  public void putToProperties(java.lang.String key, java.lang.String val) {
+  public void putToProperties(String key, String val) {
     if (this.properties == null) {
-      this.properties = new java.util.HashMap<java.lang.String,java.lang.String>();
+      this.properties = new java.util.HashMap<String, String>();
     }
     this.properties.put(key, val);
   }
 
   @org.apache.thrift.annotation.Nullable
-  public java.util.Map<java.lang.String,java.lang.String> getProperties() {
+  public java.util.Map<String, String> getProperties() {
     return this.properties;
   }
 
-  public TableCommitMeta setProperties(@org.apache.thrift.annotation.Nullable java.util.Map<java.lang.String,java.lang.String> properties) {
+  public TableCommitMeta setProperties(@org.apache.thrift.annotation.Nullable java.util.Map<String, String> properties) {
     this.properties = properties;
     return this;
   }
@@ -347,13 +360,38 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
     }
   }
 
-  public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) {
+  @org.apache.thrift.annotation.Nullable
+  public CommitMetaProducer getCommitMetaProducer() {
+    return this.commitMetaProducer;
+  }
+
+  public TableCommitMeta setCommitMetaProducer(@org.apache.thrift.annotation.Nullable CommitMetaProducer commitMetaProducer) {
+    this.commitMetaProducer = commitMetaProducer;
+    return this;
+  }
+
+  public void unsetCommitMetaProducer() {
+    this.commitMetaProducer = null;
+  }
+
+  /** Returns true if field commitMetaProducer is set (has been assigned a value) and false otherwise */
+  public boolean isSetCommitMetaProducer() {
+    return this.commitMetaProducer != null;
+  }
+
+  public void setCommitMetaProducerIsSet(boolean value) {
+    if (!value) {
+      this.commitMetaProducer = null;
+    }
+  }
+
+  public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable Object value) {
     switch (field) {
     case TABLE_IDENTIFIER:
       if (value == null) {
         unsetTableIdentifier();
       } else {
-        setTableIdentifier((com.netease.arctic.ams.api.TableIdentifier)value);
+        setTableIdentifier((TableIdentifier)value);
       }
       break;
 
@@ -361,7 +399,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
       if (value == null) {
         unsetAction();
       } else {
-        setAction((java.lang.String)value);
+        setAction((String)value);
       }
       break;
 
@@ -377,7 +415,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
       if (value == null) {
         unsetCommitTime();
       } else {
-        setCommitTime((java.lang.Long)value);
+        setCommitTime((Long)value);
       }
       break;
 
@@ -385,7 +423,15 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
       if (value == null) {
         unsetProperties();
       } else {
-        setProperties((java.util.Map<java.lang.String,java.lang.String>)value);
+        setProperties((java.util.Map<String, String>)value);
+      }
+      break;
+
+    case COMMIT_META_PRODUCER:
+      if (value == null) {
+        unsetCommitMetaProducer();
+      } else {
+        setCommitMetaProducer((CommitMetaProducer)value);
       }
       break;
 
@@ -393,7 +439,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
   }
 
   @org.apache.thrift.annotation.Nullable
-  public java.lang.Object getFieldValue(_Fields field) {
+  public Object getFieldValue(_Fields field) {
     switch (field) {
     case TABLE_IDENTIFIER:
       return getTableIdentifier();
@@ -410,14 +456,17 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
     case PROPERTIES:
       return getProperties();
 
+    case COMMIT_META_PRODUCER:
+      return getCommitMetaProducer();
+
     }
-    throw new java.lang.IllegalStateException();
+    throw new IllegalStateException();
   }
 
   /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
   public boolean isSet(_Fields field) {
     if (field == null) {
-      throw new java.lang.IllegalArgumentException();
+      throw new IllegalArgumentException();
     }
 
     switch (field) {
@@ -431,12 +480,14 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
       return isSetCommitTime();
     case PROPERTIES:
       return isSetProperties();
+    case COMMIT_META_PRODUCER:
+      return isSetCommitMetaProducer();
     }
-    throw new java.lang.IllegalStateException();
+    throw new IllegalStateException();
   }
 
   @Override
-  public boolean equals(java.lang.Object that) {
+  public boolean equals(Object that) {
     if (that == null)
       return false;
     if (that instanceof TableCommitMeta)
@@ -495,6 +546,15 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
         return false;
     }
 
+    boolean this_present_commitMetaProducer = true && this.isSetCommitMetaProducer();
+    boolean that_present_commitMetaProducer = true && that.isSetCommitMetaProducer();
+    if (this_present_commitMetaProducer || that_present_commitMetaProducer) {
+      if (!(this_present_commitMetaProducer && that_present_commitMetaProducer))
+        return false;
+      if (!this.commitMetaProducer.equals(that.commitMetaProducer))
+        return false;
+    }
+
     return true;
   }
 
@@ -520,6 +580,10 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
     if (isSetProperties())
       hashCode = hashCode * 8191 + properties.hashCode();
 
+    hashCode = hashCode * 8191 + ((isSetCommitMetaProducer()) ? 131071 : 524287);
+    if (isSetCommitMetaProducer())
+      hashCode = hashCode * 8191 + commitMetaProducer.getValue();
+
     return hashCode;
   }
 
@@ -531,7 +595,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
 
     int lastComparison = 0;
 
-    lastComparison = java.lang.Boolean.valueOf(isSetTableIdentifier()).compareTo(other.isSetTableIdentifier());
+    lastComparison = Boolean.valueOf(isSetTableIdentifier()).compareTo(other.isSetTableIdentifier());
     if (lastComparison != 0) {
       return lastComparison;
     }
@@ -541,7 +605,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
         return lastComparison;
       }
     }
-    lastComparison = java.lang.Boolean.valueOf(isSetAction()).compareTo(other.isSetAction());
+    lastComparison = Boolean.valueOf(isSetAction()).compareTo(other.isSetAction());
     if (lastComparison != 0) {
       return lastComparison;
     }
@@ -551,7 +615,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
         return lastComparison;
       }
     }
-    lastComparison = java.lang.Boolean.valueOf(isSetChanges()).compareTo(other.isSetChanges());
+    lastComparison = Boolean.valueOf(isSetChanges()).compareTo(other.isSetChanges());
     if (lastComparison != 0) {
       return lastComparison;
     }
@@ -561,7 +625,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
         return lastComparison;
       }
     }
-    lastComparison = java.lang.Boolean.valueOf(isSetCommitTime()).compareTo(other.isSetCommitTime());
+    lastComparison = Boolean.valueOf(isSetCommitTime()).compareTo(other.isSetCommitTime());
     if (lastComparison != 0) {
       return lastComparison;
     }
@@ -571,7 +635,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
         return lastComparison;
       }
     }
-    lastComparison = java.lang.Boolean.valueOf(isSetProperties()).compareTo(other.isSetProperties());
+    lastComparison = Boolean.valueOf(isSetProperties()).compareTo(other.isSetProperties());
     if (lastComparison != 0) {
       return lastComparison;
     }
@@ -581,6 +645,16 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
         return lastComparison;
       }
     }
+    lastComparison = Boolean.valueOf(isSetCommitMetaProducer()).compareTo(other.isSetCommitMetaProducer());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetCommitMetaProducer()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.commitMetaProducer, other.commitMetaProducer);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
     return 0;
   }
 
@@ -598,8 +672,8 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
   }
 
   @Override
-  public java.lang.String toString() {
-    java.lang.StringBuilder sb = new java.lang.StringBuilder("TableCommitMeta(");
+  public String toString() {
+    StringBuilder sb = new StringBuilder("TableCommitMeta(");
     boolean first = true;
 
     sb.append("tableIdentifier:");
@@ -637,6 +711,14 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
       sb.append(this.properties);
     }
     first = false;
+    if (!first) sb.append(", ");
+    sb.append("commitMetaProducer:");
+    if (this.commitMetaProducer == null) {
+      sb.append("null");
+    } else {
+      sb.append(this.commitMetaProducer);
+    }
+    first = false;
     sb.append(")");
     return sb.toString();
   }
@@ -657,7 +739,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
     }
   }
 
-  private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, java.lang.ClassNotFoundException {
+  private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
     try {
       // it doesn't seem like you should have to do this, but java serialization is wacky, and doesn't call the default constructor.
       __isset_bitfield = 0;
@@ -687,7 +769,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
         switch (schemeField.id) {
           case 1: // TABLE_IDENTIFIER
             if (schemeField.type == org.apache.thrift.protocol.TType.STRUCT) {
-              struct.tableIdentifier = new com.netease.arctic.ams.api.TableIdentifier();
+              struct.tableIdentifier = new TableIdentifier();
               struct.tableIdentifier.read(iprot);
               struct.setTableIdentifierIsSet(true);
             } else { 
@@ -733,9 +815,9 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
             if (schemeField.type == org.apache.thrift.protocol.TType.MAP) {
               {
                 org.apache.thrift.protocol.TMap _map67 = iprot.readMapBegin();
-                struct.properties = new java.util.HashMap<java.lang.String,java.lang.String>(2*_map67.size);
-                @org.apache.thrift.annotation.Nullable java.lang.String _key68;
-                @org.apache.thrift.annotation.Nullable java.lang.String _val69;
+                struct.properties = new java.util.HashMap<String, String>(2*_map67.size);
+                @org.apache.thrift.annotation.Nullable String _key68;
+                @org.apache.thrift.annotation.Nullable String _val69;
                 for (int _i70 = 0; _i70 < _map67.size; ++_i70)
                 {
                   _key68 = iprot.readString();
@@ -749,6 +831,14 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
               org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
             }
             break;
+          case 6: // COMMIT_META_PRODUCER
+            if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
+              struct.commitMetaProducer = CommitMetaProducer.findByValue(iprot.readI32());
+              struct.setCommitMetaProducerIsSet(true);
+            } else { 
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+            }
+            break;
           default:
             org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
         }
@@ -793,7 +883,7 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
         oprot.writeFieldBegin(PROPERTIES_FIELD_DESC);
         {
           oprot.writeMapBegin(new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.STRING, struct.properties.size()));
-          for (java.util.Map.Entry<java.lang.String, java.lang.String> _iter72 : struct.properties.entrySet())
+          for (java.util.Map.Entry<String, String> _iter72 : struct.properties.entrySet())
           {
             oprot.writeString(_iter72.getKey());
             oprot.writeString(_iter72.getValue());
@@ -802,6 +892,11 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
         }
         oprot.writeFieldEnd();
       }
+      if (struct.commitMetaProducer != null) {
+        oprot.writeFieldBegin(COMMIT_META_PRODUCER_FIELD_DESC);
+        oprot.writeI32(struct.commitMetaProducer.getValue());
+        oprot.writeFieldEnd();
+      }
       oprot.writeFieldStop();
       oprot.writeStructEnd();
     }
@@ -835,7 +930,10 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
       if (struct.isSetProperties()) {
         optionals.set(4);
       }
-      oprot.writeBitSet(optionals, 5);
+      if (struct.isSetCommitMetaProducer()) {
+        optionals.set(5);
+      }
+      oprot.writeBitSet(optionals, 6);
       if (struct.isSetTableIdentifier()) {
         struct.tableIdentifier.write(oprot);
       }
@@ -857,21 +955,24 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
       if (struct.isSetProperties()) {
         {
           oprot.writeI32(struct.properties.size());
-          for (java.util.Map.Entry<java.lang.String, java.lang.String> _iter74 : struct.properties.entrySet())
+          for (java.util.Map.Entry<String, String> _iter74 : struct.properties.entrySet())
           {
             oprot.writeString(_iter74.getKey());
             oprot.writeString(_iter74.getValue());
           }
         }
       }
+      if (struct.isSetCommitMetaProducer()) {
+        oprot.writeI32(struct.commitMetaProducer.getValue());
+      }
     }
 
     @Override
     public void read(org.apache.thrift.protocol.TProtocol prot, TableCommitMeta struct) throws org.apache.thrift.TException {
       org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
-      java.util.BitSet incoming = iprot.readBitSet(5);
+      java.util.BitSet incoming = iprot.readBitSet(6);
       if (incoming.get(0)) {
-        struct.tableIdentifier = new com.netease.arctic.ams.api.TableIdentifier();
+        struct.tableIdentifier = new TableIdentifier();
         struct.tableIdentifier.read(iprot);
         struct.setTableIdentifierIsSet(true);
       }
@@ -900,9 +1001,9 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
       if (incoming.get(4)) {
         {
           org.apache.thrift.protocol.TMap _map78 = new org.apache.thrift.protocol.TMap(org.apache.thrift.protocol.TType.STRING, org.apache.thrift.protocol.TType.STRING, iprot.readI32());
-          struct.properties = new java.util.HashMap<java.lang.String,java.lang.String>(2*_map78.size);
-          @org.apache.thrift.annotation.Nullable java.lang.String _key79;
-          @org.apache.thrift.annotation.Nullable java.lang.String _val80;
+          struct.properties = new java.util.HashMap<String, String>(2*_map78.size);
+          @org.apache.thrift.annotation.Nullable String _key79;
+          @org.apache.thrift.annotation.Nullable String _val80;
           for (int _i81 = 0; _i81 < _map78.size; ++_i81)
           {
             _key79 = iprot.readString();
@@ -912,6 +1013,10 @@ public class TableCommitMeta implements org.apache.thrift.TBase<TableCommitMeta,
         }
         struct.setPropertiesIsSet(true);
       }
+      if (incoming.get(5)) {
+        struct.commitMetaProducer = CommitMetaProducer.findByValue(iprot.readI32());
+        struct.setCommitMetaProducerIsSet(true);
+      }
     }
   }
 
diff --git a/ams/ams-api/src/main/thrift/arctic_table_metastore.thrift b/ams/ams-api/src/main/thrift/arctic_table_metastore.thrift
index f0335407f..4ca4070ea 100644
--- a/ams/ams-api/src/main/thrift/arctic_table_metastore.thrift
+++ b/ams/ams-api/src/main/thrift/arctic_table_metastore.thrift
@@ -46,10 +46,9 @@ struct TableCommitMeta {
     3: list<TableChange> changes;
     4: i64 commitTime;
     5: map<string, string> properties;
+    6: CommitMetaProducer commitMetaProducer;
 }
 
-
-
 struct TableMeta {
     1:arctic_commons.TableIdentifier tableIdentifier;
     2:PrimaryKeySpec keySpec;
@@ -61,7 +60,10 @@ struct PrimaryKeySpec {
   1: list<string> fields;
 }
 
-
+enum CommitMetaProducer {
+    OPTIMIZE,
+    INGESTION
+}
 
 /**
 * replace TableContainer、ArcticTableItem
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/BaseOptimizeCommit.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/BaseOptimizeCommit.java
index bef1ea963..8bf7972cd 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/BaseOptimizeCommit.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/BaseOptimizeCommit.java
@@ -18,6 +18,7 @@
 
 package com.netease.arctic.ams.server.optimize;
 
+import com.netease.arctic.ams.api.CommitMetaProducer;
 import com.netease.arctic.ams.api.OptimizeType;
 import com.netease.arctic.ams.server.model.BaseOptimizeTask;
 import com.netease.arctic.ams.server.model.TableOptimizeRuntime;
@@ -27,6 +28,7 @@ import com.netease.arctic.data.DefaultKeyedFile;
 import com.netease.arctic.op.OverwriteBaseFiles;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.UnkeyedTable;
+import com.netease.arctic.trace.SnapshotSummary;
 import com.netease.arctic.utils.SerializationUtil;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.iceberg.ContentFile;
@@ -148,6 +150,7 @@ public class BaseOptimizeCommit {
       // commit minor optimize content
       if (CollectionUtils.isNotEmpty(minorAddFiles) || CollectionUtils.isNotEmpty(minorDeleteFiles)) {
         OverwriteBaseFiles overwriteBaseFiles = new OverwriteBaseFiles(arcticTable.asKeyedTable());
+        overwriteBaseFiles.set(SnapshotSummary.SNAPSHOT_PRODUCER, CommitMetaProducer.OPTIMIZE.name());
         AtomicInteger addedPosDeleteFile = new AtomicInteger(0);
         minorAddFiles.forEach(contentFile -> {
           if (contentFile instanceof DataFile) {
@@ -176,6 +179,7 @@ public class BaseOptimizeCommit {
 
         if (CollectionUtils.isNotEmpty(deletedPosDeleteFiles)) {
           RewriteFiles rewriteFiles = baseArcticTable.newRewrite();
+          rewriteFiles.set(SnapshotSummary.SNAPSHOT_PRODUCER, CommitMetaProducer.OPTIMIZE.name());
           rewriteFiles.rewriteFiles(Collections.emptySet(), deletedPosDeleteFiles,
               Collections.emptySet(), Collections.emptySet());
           try {
@@ -230,12 +234,14 @@ public class BaseOptimizeCommit {
         }
         if (deleteDeleteFiles.isEmpty()) {
           OverwriteFiles overwriteFiles = baseArcticTable.newOverwrite();
+          overwriteFiles.set(SnapshotSummary.SNAPSHOT_PRODUCER, CommitMetaProducer.OPTIMIZE.name());
           deleteDataFiles.forEach(overwriteFiles::deleteFile);
           addDataFiles.forEach(overwriteFiles::addFile);
           overwriteFiles.commit();
         } else {
           RewriteFiles rewriteFiles = baseArcticTable.newRewrite()
               .validateFromSnapshot(baseArcticTable.currentSnapshot().snapshotId());
+          rewriteFiles.set(SnapshotSummary.SNAPSHOT_PRODUCER, CommitMetaProducer.OPTIMIZE.name());
           rewriteFiles.rewriteFiles(deleteDataFiles, deleteDeleteFiles, addDataFiles, addDeleteFiles);
           rewriteFiles.commit();
         }
diff --git a/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java b/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java
index 001998058..0f587431b 100644
--- a/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java
+++ b/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java
@@ -560,6 +560,7 @@ public class BaseArcticCatalog implements ArcticCatalog {
       String baseLocation = checkLocation(meta, MetaTableProperties.LOCATION_KEY_BASE);
 
       meta.putToProperties(TableProperties.TABLE_CREATE_TIME, String.valueOf(System.currentTimeMillis()));
+      meta.putToProperties(org.apache.iceberg.TableProperties.FORMAT_VERSION, "2");
       Table table = tableMetaStore.doAs(() -> {
         try {
           return tables.create(schema, partitionSpec, meta.getProperties(), baseLocation);
diff --git a/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java b/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java
index 4dd344483..fcf7b0282 100644
--- a/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java
+++ b/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java
@@ -22,6 +22,7 @@ import com.netease.arctic.scan.CombinedScanTask;
 import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.UnkeyedTable;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.OverwriteFiles;
@@ -126,6 +127,10 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
     if (transactionId != null && transactionId > 0) {
       overwriteFiles.set("txId", transactionId + "");
     }
+
+    if (MapUtils.isNotEmpty(properties)) {
+      properties.forEach(overwriteFiles::set);
+    }
     overwriteFiles.commit();
 
     // step2: RowDelta/Rewrite pos-delete files
@@ -146,6 +151,9 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
           partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
         }
         addDeleteFiles.forEach(rowDelta::addDeletes);
+        if (MapUtils.isNotEmpty(properties)) {
+          properties.forEach(rowDelta::set);
+        }
         rowDelta.commit();
       } else {
         RewriteFiles rewriteFiles = transaction.newRewrite();
@@ -164,6 +172,9 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
         }
         rewriteFiles.rewriteFiles(Collections.emptySet(), new HashSet<>(deleteDeleteFiles),
             Collections.emptySet(), new HashSet<>(addDeleteFiles));
+        if (MapUtils.isNotEmpty(properties)) {
+          properties.forEach(rewriteFiles::set);
+        }
         rewriteFiles.commit();
       }
     }
diff --git a/core/src/main/java/com/netease/arctic/op/PartitionTransactionOperation.java b/core/src/main/java/com/netease/arctic/op/PartitionTransactionOperation.java
index 97c21b5c2..45e547222 100644
--- a/core/src/main/java/com/netease/arctic/op/PartitionTransactionOperation.java
+++ b/core/src/main/java/com/netease/arctic/op/PartitionTransactionOperation.java
@@ -27,6 +27,9 @@ import org.apache.iceberg.Transaction;
 import org.apache.iceberg.UpdateProperties;
 import org.apache.iceberg.util.StructLikeMap;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /**
  * Abstract transaction operation on {@link BaseTable} which will change
  * max transaction id map
@@ -36,9 +39,16 @@ public abstract class PartitionTransactionOperation implements PendingUpdate<Str
   KeyedTable keyedTable;
   private Transaction tx;
 
+  protected final Map<String, String> properties;
 
   public PartitionTransactionOperation(KeyedTable baseTable) {
     this.keyedTable = baseTable;
+    this.properties = new HashMap<>();
+  }
+
+  public PartitionTransactionOperation set(String key, String value) {
+    this.properties.put(key, value);
+    return this;
   }
 
   /**
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 175eac383..6e528ef40 100644
--- a/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java
+++ b/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java
@@ -19,6 +19,7 @@
 package com.netease.arctic.trace;
 
 import com.netease.arctic.AmsClient;
+import com.netease.arctic.ams.api.CommitMetaProducer;
 import com.netease.arctic.ams.api.Constants;
 import com.netease.arctic.ams.api.TableChange;
 import com.netease.arctic.ams.api.TableCommitMeta;
@@ -35,10 +36,12 @@ import org.apache.iceberg.SnapshotSummary;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.PropertyUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -58,6 +61,7 @@ public class AmsTableTracer implements TableTracer {
 
   private String action;
   private Map<String, String> properties;
+  private final Map<String, String> snapshotSummary = new HashMap<>();
   private InternalTableChange defaultTableChange;
   private final Map<Long, AmsTableTracer.InternalTableChange> transactionSnapshotTableChanges = new LinkedHashMap<>();
 
@@ -123,6 +127,10 @@ public class AmsTableTracer implements TableTracer {
     TableCommitMeta commitMeta = new TableCommitMeta();
     commitMeta.setTableIdentifier(table.id().buildTableIdentifier());
     commitMeta.setAction(action);
+    String commitMetaSource = PropertyUtil.propertyAsString(snapshotSummary,
+        com.netease.arctic.trace.SnapshotSummary.SNAPSHOT_PRODUCER,
+        com.netease.arctic.trace.SnapshotSummary.SNAPSHOT_PRODUCER_DEFAULT);
+    commitMeta.setCommitMetaProducer(CommitMetaProducer.valueOf(commitMetaSource));
     commitMeta.setCommitTime(System.currentTimeMillis());
     boolean update = false;
     boolean threw = false;
@@ -176,6 +184,11 @@ public class AmsTableTracer implements TableTracer {
     this.properties = newProperties;
   }
 
+  @Override
+  public void setSnapshotSummary(String key, String value) {
+    snapshotSummary.put(key, value);
+  }
+
   public void setAction(String action) {
     this.action = action;
   }
diff --git a/core/src/main/java/com/netease/arctic/trace/CreateTableTransaction.java b/core/src/main/java/com/netease/arctic/trace/CreateTableTransaction.java
index 6b3561608..a7e1f8742 100644
--- a/core/src/main/java/com/netease/arctic/trace/CreateTableTransaction.java
+++ b/core/src/main/java/com/netease/arctic/trace/CreateTableTransaction.java
@@ -251,6 +251,10 @@ public class CreateTableTransaction implements Transaction {
     @Override
     public void replaceProperties(Map<String, String> newProperties) {
     }
+
+    @Override
+    public void setSnapshotSummary(String key, String value) {
+    }
   }
 
   class TransactionTable implements Table, HasTableOperations, Serializable {
diff --git a/core/src/main/java/com/netease/arctic/trace/SnapshotSummary.java b/core/src/main/java/com/netease/arctic/trace/SnapshotSummary.java
new file mode 100644
index 000000000..5e15d1f11
--- /dev/null
+++ b/core/src/main/java/com/netease/arctic/trace/SnapshotSummary.java
@@ -0,0 +1,26 @@
+/*
+ * 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.trace;
+
+import com.netease.arctic.ams.api.CommitMetaProducer;
+
+public class SnapshotSummary {
+  public static final String SNAPSHOT_PRODUCER = "snapshot.producer";
+  public static final String SNAPSHOT_PRODUCER_DEFAULT = CommitMetaProducer.INGESTION.name();
+}
diff --git a/core/src/main/java/com/netease/arctic/trace/TableTracer.java b/core/src/main/java/com/netease/arctic/trace/TableTracer.java
index 98a3b5642..ebe52806d 100644
--- a/core/src/main/java/com/netease/arctic/trace/TableTracer.java
+++ b/core/src/main/java/com/netease/arctic/trace/TableTracer.java
@@ -62,4 +62,12 @@ public interface TableTracer {
    * @param newProperties properties to replace
    */
   void replaceProperties(Map<String, String> newProperties);
+
+  /**
+   * Set a summary property in the snapshot produced by this update.
+   *
+   * @param key a String property name
+   * @param value a String property value
+   */
+  void setSnapshotSummary(String key, String value);
 }
diff --git a/core/src/main/java/com/netease/arctic/trace/TracedAppendFiles.java b/core/src/main/java/com/netease/arctic/trace/TracedAppendFiles.java
index 00771025b..fe407b8bd 100644
--- a/core/src/main/java/com/netease/arctic/trace/TracedAppendFiles.java
+++ b/core/src/main/java/com/netease/arctic/trace/TracedAppendFiles.java
@@ -55,6 +55,7 @@ public class TracedAppendFiles implements AppendFiles {
   @Override
   public AppendFiles set(String property, String value) {
     appendFiles.set(property, value);
+    tracer.setSnapshotSummary(property, value);
     return this;
   }
 
diff --git a/core/src/main/java/com/netease/arctic/trace/TracedDeleteFiles.java b/core/src/main/java/com/netease/arctic/trace/TracedDeleteFiles.java
index 66272415f..fad3a4b6c 100644
--- a/core/src/main/java/com/netease/arctic/trace/TracedDeleteFiles.java
+++ b/core/src/main/java/com/netease/arctic/trace/TracedDeleteFiles.java
@@ -59,6 +59,7 @@ public class TracedDeleteFiles implements DeleteFiles {
   @Override
   public DeleteFiles set(String property, String value) {
     deleteFiles.set(property, value);
+    tracer.setSnapshotSummary(property, value);
     return this;
   }
 
diff --git a/core/src/main/java/com/netease/arctic/trace/TracedOverwriteFiles.java b/core/src/main/java/com/netease/arctic/trace/TracedOverwriteFiles.java
index 3cc2a7685..becffb386 100644
--- a/core/src/main/java/com/netease/arctic/trace/TracedOverwriteFiles.java
+++ b/core/src/main/java/com/netease/arctic/trace/TracedOverwriteFiles.java
@@ -91,6 +91,7 @@ public class TracedOverwriteFiles implements OverwriteFiles {
   @Override
   public OverwriteFiles set(String property, String value) {
     overwriteFiles.set(property, value);
+    tracer.setSnapshotSummary(property, value);
     return this;
   }
 
diff --git a/core/src/main/java/com/netease/arctic/trace/TracedRewriteFiles.java b/core/src/main/java/com/netease/arctic/trace/TracedRewriteFiles.java
index 5c2fb7bb4..60c2ab29e 100644
--- a/core/src/main/java/com/netease/arctic/trace/TracedRewriteFiles.java
+++ b/core/src/main/java/com/netease/arctic/trace/TracedRewriteFiles.java
@@ -66,6 +66,7 @@ public class TracedRewriteFiles implements RewriteFiles {
   @Override
   public RewriteFiles set(String property, String value) {
     rewriteFiles.set(property, value);
+    tracer.setSnapshotSummary(property, value);
     return this;
   }
 
diff --git a/core/src/main/java/com/netease/arctic/trace/TracedRowDelta.java b/core/src/main/java/com/netease/arctic/trace/TracedRowDelta.java
index 8373fa5c7..af842e0e4 100644
--- a/core/src/main/java/com/netease/arctic/trace/TracedRowDelta.java
+++ b/core/src/main/java/com/netease/arctic/trace/TracedRowDelta.java
@@ -86,6 +86,7 @@ public class TracedRowDelta implements RowDelta {
   @Override
   public RowDelta set(String property, String value) {
     rowDelta.set(property, value);
+    tracer.setSnapshotSummary(property, value);
     return this;
   }
 
diff --git a/core/src/main/java/com/netease/arctic/trace/TracedTransaction.java b/core/src/main/java/com/netease/arctic/trace/TracedTransaction.java
index 492a138eb..9403153a2 100644
--- a/core/src/main/java/com/netease/arctic/trace/TracedTransaction.java
+++ b/core/src/main/java/com/netease/arctic/trace/TracedTransaction.java
@@ -199,6 +199,11 @@ public class TracedTransaction implements Transaction {
     public void replaceProperties(Map<String, String> newProperties) {
       tracer.replaceProperties(newProperties);
     }
+
+    @Override
+    public void setSnapshotSummary(String key, String value) {
+      tracer.setSnapshotSummary(key, value);
+    }
   }
 
   class TransactionTable implements Table, HasTableOperations, Serializable {
diff --git a/dist/src/main/assemblies/bin.xml b/dist/src/main/assemblies/bin.xml
index 489af8fb9..555a86757 100644
--- a/dist/src/main/assemblies/bin.xml
+++ b/dist/src/main/assemblies/bin.xml
@@ -29,7 +29,7 @@
 
     <files>
         <file>
-            <source>../optimizer/target/arctic-optimizer-${project.version}.jar</source>
+            <source>../optimizer/target/arctic-optimizer-${project.version}-jar-with-dependencies.jar</source>
             <outputDirectory>lib/</outputDirectory>
             <destName>CompactJob.jar</destName>
             <fileMode>0644</fileMode>
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/ArcticDynamicSource.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/ArcticDynamicSource.java
index 7d3685c4f..81432030b 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/ArcticDynamicSource.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/ArcticDynamicSource.java
@@ -32,6 +32,7 @@ import org.apache.flink.table.connector.source.DynamicTableSource;
 import org.apache.flink.table.connector.source.ScanTableSource;
 import org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown;
 import org.apache.flink.table.connector.source.abilities.SupportsLimitPushDown;
+import org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown;
 import org.apache.flink.table.connector.source.abilities.SupportsWatermarkPushDown;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.expressions.ResolvedExpression;
@@ -60,7 +61,7 @@ import static com.netease.arctic.table.TableProperties.READ_DISTRIBUTION_MODE_DE
  * Flink table api that generates source operators.
  */
 public class ArcticDynamicSource implements ScanTableSource, SupportsFilterPushDown,
-    SupportsLimitPushDown, SupportsWatermarkPushDown {
+    SupportsProjectionPushDown, SupportsLimitPushDown, SupportsWatermarkPushDown {
 
   public static final Logger LOG = LoggerFactory.getLogger(ArcticDynamicSource.class);
 
@@ -221,6 +222,22 @@ public class ArcticDynamicSource implements ScanTableSource, SupportsFilterPushD
     }
   }
 
+  @Override
+  public boolean supportsNestedProjection() {
+    if (arcticDynamicSource instanceof SupportsProjectionPushDown) {
+      return ((SupportsProjectionPushDown) arcticDynamicSource).supportsNestedProjection();
+    } else {
+      return false;
+    }
+  }
+
+  @Override
+  public void applyProjection(int[][] projectedFields) {
+    if (arcticDynamicSource instanceof SupportsProjectionPushDown) {
+      ((SupportsProjectionPushDown) arcticDynamicSource).applyProjection(projectedFields);
+    }
+  }
+
   @Override
   public void applyLimit(long newLimit) {
     if (arcticDynamicSource instanceof SupportsLimitPushDown) {
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/KafkaDynamicSource.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/KafkaDynamicSource.java
index 23bd7961f..fe629c7ab 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/KafkaDynamicSource.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/KafkaDynamicSource.java
@@ -19,6 +19,7 @@
 package com.netease.arctic.flink.table;
 
 import com.netease.arctic.flink.read.FlinkKafkaConsumer;
+import com.netease.arctic.flink.util.Projection;
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.serialization.DeserializationSchema;
@@ -33,7 +34,7 @@ import org.apache.flink.table.connector.format.DecodingFormat;
 import org.apache.flink.table.connector.source.DataStreamScanProvider;
 import org.apache.flink.table.connector.source.DynamicTableSource;
 import org.apache.flink.table.connector.source.ScanTableSource;
-import org.apache.flink.table.connector.source.SourceFunctionProvider;
+import org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown;
 import org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata;
 import org.apache.flink.table.connector.source.abilities.SupportsWatermarkPushDown;
 import org.apache.flink.table.data.GenericMapData;
@@ -56,6 +57,7 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Properties;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -66,7 +68,7 @@ import java.util.stream.Stream;
  */
 @Internal
 public class KafkaDynamicSource
-    implements ScanTableSource, SupportsReadingMetadata, SupportsWatermarkPushDown {
+    implements ScanTableSource, SupportsReadingMetadata, SupportsProjectionPushDown, SupportsWatermarkPushDown {
 
   // --------------------------------------------------------------------------------------------
   // Mutable attributes
@@ -82,6 +84,11 @@ public class KafkaDynamicSource
    */
   protected List<String> metadataKeys;
 
+  /**
+   * Field index paths of all fields that must be present in the physically produced data.
+   */
+  protected int[] projectedFields;
+
   /**
    * Watermark strategy that is used to generate per-partition watermark.
    */
@@ -199,6 +206,7 @@ public class KafkaDynamicSource
     // Mutable attributes
     this.producedDataType = physicalDataType;
     this.metadataKeys = Collections.emptyList();
+    this.projectedFields = null;
     this.watermarkStrategy = null;
     // Kafka-specific attributes
     Preconditions.checkArgument(
@@ -232,7 +240,9 @@ public class KafkaDynamicSource
         createDeserialization(context, valueDecodingFormat, valueProjection, null);
 
     final TypeInformation<RowData> producedTypeInfo =
-        context.createTypeInformation(producedDataType);
+        context.createTypeInformation(Optional.ofNullable(projectedFields)
+            .map(Projection::of)
+            .map(p -> p.project(producedDataType)).orElse(producedDataType));
 
     final FlinkKafkaConsumer<RowData> kafkaConsumer =
         createKafkaConsumer(keyDeserialization, valueDeserialization, producedTypeInfo);
@@ -294,6 +304,23 @@ public class KafkaDynamicSource
     this.producedDataType = producedDataType;
   }
 
+  @Override
+  public boolean supportsNestedProjection() {
+    // TODO: support nested projection
+    return false;
+  }
+
+  @Override
+  public void applyProjection(int[][] projectFields) {
+    this.projectedFields = new int[projectFields.length];
+    for (int i = 0; i < projectFields.length; i++) {
+      Preconditions.checkArgument(
+          projectFields[i].length == 1,
+          "Don't support nested projection now.");
+      this.projectedFields[i] = projectFields[i][0];
+    }
+  }
+
   @Override
   public void applyWatermark(WatermarkStrategy<RowData> watermarkStrategy) {
     this.watermarkStrategy = watermarkStrategy;
@@ -352,6 +379,7 @@ public class KafkaDynamicSource
         Objects.equals(specificStartupOffsets, that.specificStartupOffsets) &&
         startupTimestampMillis == that.startupTimestampMillis &&
         Objects.equals(upsertMode, that.upsertMode) &&
+        Arrays.equals(projectedFields, that.projectedFields) &&
         Objects.equals(watermarkStrategy, that.watermarkStrategy);
   }
 
@@ -373,6 +401,7 @@ public class KafkaDynamicSource
         specificStartupOffsets,
         startupTimestampMillis,
         upsertMode,
+        projectedFields,
         watermarkStrategy);
   }
 
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/LogDynamicSource.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/LogDynamicSource.java
index 83ade0854..6be4d1698 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/LogDynamicSource.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/table/LogDynamicSource.java
@@ -33,12 +33,15 @@ import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.types.RowKind;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.types.Types;
 
 import javax.annotation.Nullable;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_LOG_CONSISTENCY_GUARANTEE_ENABLE;
 import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_LOG_CONSUMER_CHANGELOG_MODE;
@@ -113,15 +116,20 @@ public class LogDynamicSource extends KafkaDynamicSource {
 
     final FlinkKafkaConsumer<RowData> kafkaConsumer;
 
-    KafkaDeserializationSchemaWrapper<RowData> deserializationSchemaWrapper =
+    final KafkaDeserializationSchemaWrapper<RowData> deserializationSchemaWrapper =
         new KafkaDeserializationSchemaWrapper<>(valueDeserialization);
+    Schema projectedSchema = schema;
+    if (projectedFields != null) {
+      final List<Types.NestedField> columns = schema.columns();
+      projectedSchema = new Schema(Arrays.stream(projectedFields).mapToObj(columns::get).collect(Collectors.toList()));
+    }
     if (topics != null) {
       kafkaConsumer =
           new LogKafkaConsumer(
               topics,
               deserializationSchemaWrapper,
               properties,
-              schema,
+              projectedSchema,
               tableOptions);
     } else {
       kafkaConsumer =
@@ -129,7 +137,7 @@ public class LogDynamicSource extends KafkaDynamicSource {
               topicPattern,
               deserializationSchemaWrapper,
               properties,
-              schema,
+              projectedSchema,
               tableOptions);
     }
 
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/util/Projection.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/util/Projection.java
new file mode 100644
index 000000000..729f6826b
--- /dev/null
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/util/Projection.java
@@ -0,0 +1,444 @@
+/*
+ * 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.flink.util;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.FieldsDataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.utils.TypeConversions;
+import org.apache.flink.util.Preconditions;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.ROW;
+
+/**
+ * {@link Projection} represents a list of (possibly nested) indexes that can be used to project
+ * data types. A row projection includes both reducing the accessible fields and reordering them.
+ *
+ * <p>NOTE: Copied from Flink.
+ */
+public abstract class Projection {
+
+  // sealed class
+  private Projection() {
+  }
+
+  /**
+   * Projects a (possibly nested) row data type by returning a new data type that only includes
+   * fields of the given index paths.
+   *
+   * <p>When extracting nested fields, the name of the resulting fields is the full path of the
+   * field separated by {@code _}. For example, the field {@code b} inside the row field {@code a}
+   * of the root {@link DataType} is named {@code a_b} in the result {@link DataType}. In case of
+   * naming conflicts the postfix notation '_$%d' is used, where {@code %d} is an arbitrary
+   * number, in order to generate a unique field name. For example if the root {@link DataType}
+   * includes both a field {@code a_b} and a nested row {@code a} with field {@code b}, the result
+   * {@link DataType} will contain one field named {@code a_b} and the other named {@code a_b_1}.
+   */
+  public abstract DataType project(DataType dataType);
+
+  /**
+   * Same as {@link #project(DataType)}, but accepting and returning {@link LogicalType}.
+   */
+  public LogicalType project(LogicalType logicalType) {
+    return this.project(TypeConversions.fromLogicalToDataType(logicalType)).getLogicalType();
+  }
+
+  /**
+   * @return {@code true} whether this projection is nested or not.
+   */
+  public abstract boolean isNested();
+
+  /**
+   * Perform a difference of this {@link Projection} with another {@link Projection}. The result
+   * of this operation is a new {@link Projection} retaining the same ordering of this instance
+   * but with the indexes from {@code other} removed. For example:
+   *
+   * <pre>
+   * <code>
+   * [4, 1, 0, 3, 2] - [4, 2] = [1, 0, 2]
+   * </code>
+   * </pre>
+   *
+   * <p>Note how the index {@code 3} in the minuend becomes {@code 2} because it's rescaled to
+   * project correctly a {@link RowData} or arity 3.
+   *
+   * @param other the subtrahend
+   * @throws IllegalArgumentException when {@code other} is nested.
+   */
+  public abstract Projection difference(Projection other);
+
+  /**
+   * Complement this projection. The returned projection is an ordered projection of fields from 0
+   * to {@code fieldsNumber} except the indexes in this {@link Projection}. For example:
+   *
+   * <pre>
+   * <code>
+   * [4, 2].complement(5) = [0, 1, 3]
+   * </code>
+   * </pre>
+   *
+   * @param fieldsNumber the size of the universe
+   * @throws IllegalStateException if this projection is nested.
+   */
+  public abstract Projection complement(int fieldsNumber);
+
+  /**
+   * Like {@link #complement(int)}, using the {@code dataType} fields count.
+   */
+  public Projection complement(DataType dataType) {
+    return complement(dataType.getLogicalType().getChildren().size());
+  }
+
+  /**
+   * Convert this instance to a projection of top level indexes. The array represents the mapping
+   * of the fields of the original {@link DataType}. For example, {@code [0, 2, 1]} specifies to
+   * include in the following order the 1st field, the 3rd field and the 2nd field of the row.
+   *
+   * @throws IllegalStateException if this projection is nested.
+   */
+  public abstract int[] toTopLevelIndexes();
+
+  /**
+   * Convert this instance to a nested projection index paths. The array represents the mapping of
+   * the fields of the original {@link DataType}, including nested rows. For example, {@code [[0,
+   * 2, 1], ...]} specifies to include the 2nd field of the 3rd field of the 1st field in the
+   * top-level row.
+   */
+  public abstract int[][] toNestedIndexes();
+
+  /**
+   * Create an empty {@link Projection}, that is a projection that projects no fields, returning
+   * an empty {@link DataType}.
+   */
+  public static Projection empty() {
+    return EmptyProjection.INSTANCE;
+  }
+
+  /**
+   * Create a {@link Projection} of the provided {@code indexes}.
+   *
+   * @see #toTopLevelIndexes()
+   */
+  public static Projection of(int[] indexes) {
+    if (indexes.length == 0) {
+      return empty();
+    }
+    return new TopLevelProjection(indexes);
+  }
+
+  /**
+   * Create a {@link Projection} of the provided {@code indexes}.
+   *
+   * @see #toNestedIndexes()
+   */
+  public static Projection of(int[][] indexes) {
+    if (indexes.length == 0) {
+      return empty();
+    }
+    return new NestedProjection(indexes);
+  }
+
+  /**
+   * Create a {@link Projection} of a field range.
+   */
+  public static Projection range(int startInclusive, int endExclusive) {
+    return new TopLevelProjection(IntStream.range(startInclusive, endExclusive).toArray());
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof Projection)) {
+      return false;
+    }
+    Projection other = (Projection) o;
+    if (!this.isNested() && !other.isNested()) {
+      return Arrays.equals(this.toTopLevelIndexes(), other.toTopLevelIndexes());
+    }
+    return Arrays.deepEquals(this.toNestedIndexes(), other.toNestedIndexes());
+  }
+
+  @Override
+  public int hashCode() {
+    if (isNested()) {
+      return Arrays.deepHashCode(toNestedIndexes());
+    }
+    return Arrays.hashCode(toTopLevelIndexes());
+  }
+
+  @Override
+  public String toString() {
+    if (isNested()) {
+      return "Nested projection = " + Arrays.deepToString(toNestedIndexes());
+    }
+    return "Top level projection = " + Arrays.toString(toTopLevelIndexes());
+  }
+
+  private static class EmptyProjection extends Projection {
+
+    static final EmptyProjection INSTANCE = new EmptyProjection();
+
+    private EmptyProjection() {
+    }
+
+    @Override
+    public DataType project(DataType dataType) {
+      return new NestedProjection(toNestedIndexes()).project(dataType);
+    }
+
+    @Override
+    public boolean isNested() {
+      return false;
+    }
+
+    @Override
+    public Projection difference(Projection projection) {
+      return this;
+    }
+
+    @Override
+    public Projection complement(int fieldsNumber) {
+      return new TopLevelProjection(IntStream.range(0, fieldsNumber).toArray());
+    }
+
+    @Override
+    public int[] toTopLevelIndexes() {
+      return new int[0];
+    }
+
+    @Override
+    public int[][] toNestedIndexes() {
+      return new int[0][];
+    }
+  }
+
+  private static class NestedProjection extends Projection {
+
+    final int[][] projection;
+    final boolean nested;
+
+    NestedProjection(int[][] projection) {
+      this.projection = projection;
+      this.nested = Arrays.stream(projection).anyMatch(arr -> arr.length > 1);
+    }
+
+    @Override
+    public DataType project(DataType dataType) {
+      final List<RowType.RowField> updatedFields = new ArrayList<>();
+      final List<DataType> updatedChildren = new ArrayList<>();
+      Set<String> nameDomain = new HashSet<>();
+      int duplicateCount = 0;
+      for (int[] indexPath : this.projection) {
+        DataType fieldType = dataType.getChildren().get(indexPath[0]);
+        LogicalType fieldLogicalType = fieldType.getLogicalType();
+        StringBuilder builder =
+            new StringBuilder(
+                ((RowType) dataType.getLogicalType())
+                    .getFieldNames()
+                    .get(indexPath[0]));
+        for (int index = 1; index < indexPath.length; index++) {
+          Preconditions.checkArgument(
+              fieldLogicalType.getTypeRoot() == ROW, "Row data type expected.");
+          RowType rowtype = ((RowType) fieldLogicalType);
+          builder.append("_").append(rowtype.getFieldNames().get(indexPath[index]));
+          fieldLogicalType = rowtype.getFields().get(indexPath[index]).getType();
+          fieldType = fieldType.getChildren().get(indexPath[index]);
+        }
+        String path = builder.toString();
+        while (nameDomain.contains(path)) {
+          path = builder.append("_$").append(duplicateCount++).toString();
+        }
+        updatedFields.add(new RowType.RowField(path, fieldLogicalType));
+        updatedChildren.add(fieldType);
+        nameDomain.add(path);
+      }
+      return new FieldsDataType(
+          new RowType(dataType.getLogicalType().isNullable(), updatedFields),
+          dataType.getConversionClass(),
+          updatedChildren);
+    }
+
+    @Override
+    public boolean isNested() {
+      return nested;
+    }
+
+    @Override
+    public Projection difference(Projection other) {
+      if (other.isNested()) {
+        throw new IllegalArgumentException(
+            "Cannot perform difference between nested projection and nested projection");
+      }
+      if (other instanceof EmptyProjection) {
+        return this;
+      }
+      if (!this.isNested()) {
+        return new TopLevelProjection(toTopLevelIndexes()).difference(other);
+      }
+
+      // Extract the indexes to exclude and sort them
+      int[] indexesToExclude = other.toTopLevelIndexes();
+      indexesToExclude = Arrays.copyOf(indexesToExclude, indexesToExclude.length);
+      Arrays.sort(indexesToExclude);
+
+      List<int[]> resultProjection =
+          Arrays.stream(projection).collect(Collectors.toCollection(ArrayList::new));
+
+      ListIterator<int[]> resultProjectionIterator = resultProjection.listIterator();
+      while (resultProjectionIterator.hasNext()) {
+        int[] indexArr = resultProjectionIterator.next();
+
+        // Let's check if the index is inside the indexesToExclude array
+        int searchResult = Arrays.binarySearch(indexesToExclude, indexArr[0]);
+        if (searchResult >= 0) {
+          // Found, we need to remove it
+          resultProjectionIterator.remove();
+        } else {
+          // Not found, let's compute the offset.
+          // Offset is the index where the projection index should be inserted in the
+          // indexesToExclude array
+          int offset = (-(searchResult) - 1);
+          if (offset != 0) {
+            indexArr[0] = indexArr[0] - offset;
+          }
+        }
+      }
+
+      return new NestedProjection(resultProjection.toArray(new int[0][]));
+    }
+
+    @Override
+    public Projection complement(int fieldsNumber) {
+      if (isNested()) {
+        throw new IllegalStateException("Cannot perform complement of a nested projection");
+      }
+      return new TopLevelProjection(toTopLevelIndexes()).complement(fieldsNumber);
+    }
+
+    @Override
+    public int[] toTopLevelIndexes() {
+      if (isNested()) {
+        throw new IllegalStateException(
+            "Cannot convert a nested projection to a top level projection");
+      }
+      return Arrays.stream(projection).mapToInt(arr -> arr[0]).toArray();
+    }
+
+    @Override
+    public int[][] toNestedIndexes() {
+      return projection;
+    }
+  }
+
+  private static class TopLevelProjection extends Projection {
+
+    final int[] projection;
+
+    TopLevelProjection(int[] projection) {
+      this.projection = projection;
+    }
+
+    @Override
+    public DataType project(DataType dataType) {
+      return new NestedProjection(toNestedIndexes()).project(dataType);
+    }
+
+    @Override
+    public boolean isNested() {
+      return false;
+    }
+
+    @Override
+    public Projection difference(Projection other) {
+      if (other.isNested()) {
+        throw new IllegalArgumentException(
+            "Cannot perform difference between top level projection and nested projection");
+      }
+      if (other instanceof EmptyProjection) {
+        return this;
+      }
+
+      // Extract the indexes to exclude and sort them
+      int[] indexesToExclude = other.toTopLevelIndexes();
+      indexesToExclude = Arrays.copyOf(indexesToExclude, indexesToExclude.length);
+      Arrays.sort(indexesToExclude);
+
+      List<Integer> resultProjection =
+          Arrays.stream(projection)
+              .boxed()
+              .collect(Collectors.toCollection(ArrayList::new));
+
+      ListIterator<Integer> resultProjectionIterator = resultProjection.listIterator();
+      while (resultProjectionIterator.hasNext()) {
+        int index = resultProjectionIterator.next();
+
+        // Let's check if the index is inside the indexesToExclude array
+        int searchResult = Arrays.binarySearch(indexesToExclude, index);
+        if (searchResult >= 0) {
+          // Found, we need to remove it
+          resultProjectionIterator.remove();
+        } else {
+          // Not found, let's compute the offset.
+          // Offset is the index where the projection index should be inserted in the
+          // indexesToExclude array
+          int offset = (-(searchResult) - 1);
+          if (offset != 0) {
+            resultProjectionIterator.set(index - offset);
+          }
+        }
+      }
+
+      return new TopLevelProjection(resultProjection.stream().mapToInt(i -> i).toArray());
+    }
+
+    @Override
+    public Projection complement(int fieldsNumber) {
+      int[] indexesToExclude = Arrays.copyOf(projection, projection.length);
+      Arrays.sort(indexesToExclude);
+
+      return new TopLevelProjection(
+          IntStream.range(0, fieldsNumber)
+              .filter(i -> Arrays.binarySearch(indexesToExclude, i) < 0)
+              .toArray());
+    }
+
+    @Override
+    public int[] toTopLevelIndexes() {
+      return projection;
+    }
+
+    @Override
+    public int[][] toNestedIndexes() {
+      return Arrays.stream(projection).mapToObj(i -> new int[] {i}).toArray(int[][]::new);
+    }
+  }
+}
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/ArcticDynamicSource.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/ArcticDynamicSource.java
index 7d3685c4f..81432030b 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/ArcticDynamicSource.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/ArcticDynamicSource.java
@@ -32,6 +32,7 @@ import org.apache.flink.table.connector.source.DynamicTableSource;
 import org.apache.flink.table.connector.source.ScanTableSource;
 import org.apache.flink.table.connector.source.abilities.SupportsFilterPushDown;
 import org.apache.flink.table.connector.source.abilities.SupportsLimitPushDown;
+import org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown;
 import org.apache.flink.table.connector.source.abilities.SupportsWatermarkPushDown;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.expressions.ResolvedExpression;
@@ -60,7 +61,7 @@ import static com.netease.arctic.table.TableProperties.READ_DISTRIBUTION_MODE_DE
  * Flink table api that generates source operators.
  */
 public class ArcticDynamicSource implements ScanTableSource, SupportsFilterPushDown,
-    SupportsLimitPushDown, SupportsWatermarkPushDown {
+    SupportsProjectionPushDown, SupportsLimitPushDown, SupportsWatermarkPushDown {
 
   public static final Logger LOG = LoggerFactory.getLogger(ArcticDynamicSource.class);
 
@@ -221,6 +222,22 @@ public class ArcticDynamicSource implements ScanTableSource, SupportsFilterPushD
     }
   }
 
+  @Override
+  public boolean supportsNestedProjection() {
+    if (arcticDynamicSource instanceof SupportsProjectionPushDown) {
+      return ((SupportsProjectionPushDown) arcticDynamicSource).supportsNestedProjection();
+    } else {
+      return false;
+    }
+  }
+
+  @Override
+  public void applyProjection(int[][] projectedFields) {
+    if (arcticDynamicSource instanceof SupportsProjectionPushDown) {
+      ((SupportsProjectionPushDown) arcticDynamicSource).applyProjection(projectedFields);
+    }
+  }
+
   @Override
   public void applyLimit(long newLimit) {
     if (arcticDynamicSource instanceof SupportsLimitPushDown) {
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/KafkaDynamicSource.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/KafkaDynamicSource.java
index 9cb093eb2..ab0514cf7 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/KafkaDynamicSource.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/KafkaDynamicSource.java
@@ -19,6 +19,7 @@
 package com.netease.arctic.flink.table;
 
 import com.netease.arctic.flink.read.FlinkKafkaConsumer;
+import com.netease.arctic.flink.util.Projection;
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
 import org.apache.flink.api.common.serialization.DeserializationSchema;
@@ -33,6 +34,7 @@ import org.apache.flink.table.connector.format.DecodingFormat;
 import org.apache.flink.table.connector.source.DataStreamScanProvider;
 import org.apache.flink.table.connector.source.DynamicTableSource;
 import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.abilities.SupportsProjectionPushDown;
 import org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata;
 import org.apache.flink.table.connector.source.abilities.SupportsWatermarkPushDown;
 import org.apache.flink.table.data.GenericMapData;
@@ -55,6 +57,7 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Properties;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -65,7 +68,7 @@ import java.util.stream.Stream;
  */
 @Internal
 public class KafkaDynamicSource
-    implements ScanTableSource, SupportsReadingMetadata, SupportsWatermarkPushDown {
+    implements ScanTableSource, SupportsReadingMetadata, SupportsProjectionPushDown, SupportsWatermarkPushDown {
 
   // --------------------------------------------------------------------------------------------
   // Mutable attributes
@@ -81,6 +84,11 @@ public class KafkaDynamicSource
    */
   protected List<String> metadataKeys;
 
+  /**
+   * Field index paths of all fields that must be present in the physically produced data.
+   */
+  protected int[] projectedFields;
+
   /**
    * Watermark strategy that is used to generate per-partition watermark.
    */
@@ -198,6 +206,7 @@ public class KafkaDynamicSource
     // Mutable attributes
     this.producedDataType = physicalDataType;
     this.metadataKeys = Collections.emptyList();
+    this.projectedFields = null;
     this.watermarkStrategy = null;
     // Kafka-specific attributes
     Preconditions.checkArgument(
@@ -231,7 +240,9 @@ public class KafkaDynamicSource
         createDeserialization(context, valueDecodingFormat, valueProjection, null);
 
     final TypeInformation<RowData> producedTypeInfo =
-        context.createTypeInformation(producedDataType);
+        context.createTypeInformation(Optional.ofNullable(projectedFields)
+            .map(Projection::of)
+            .map(p -> p.project(producedDataType)).orElse(producedDataType));
 
     final FlinkKafkaConsumer<RowData> kafkaConsumer =
         createKafkaConsumer(keyDeserialization, valueDeserialization, producedTypeInfo);
@@ -293,6 +304,23 @@ public class KafkaDynamicSource
     this.producedDataType = producedDataType;
   }
 
+  @Override
+  public boolean supportsNestedProjection() {
+    // TODO: support nested projection
+    return false;
+  }
+
+  @Override
+  public void applyProjection(int[][] projectFields) {
+    this.projectedFields = new int[projectFields.length];
+    for (int i = 0; i < projectFields.length; i++) {
+      Preconditions.checkArgument(
+          projectFields[i].length == 1,
+          "Don't support nested projection now.");
+      this.projectedFields[i] = projectFields[i][0];
+    }
+  }
+
   @Override
   public void applyWatermark(WatermarkStrategy<RowData> watermarkStrategy) {
     this.watermarkStrategy = watermarkStrategy;
@@ -318,6 +346,7 @@ public class KafkaDynamicSource
             sourceName);
     copy.producedDataType = producedDataType;
     copy.metadataKeys = metadataKeys;
+    copy.projectedFields = projectedFields;
     copy.watermarkStrategy = watermarkStrategy;
     return copy;
   }
@@ -351,6 +380,7 @@ public class KafkaDynamicSource
         Objects.equals(specificStartupOffsets, that.specificStartupOffsets) &&
         startupTimestampMillis == that.startupTimestampMillis &&
         Objects.equals(upsertMode, that.upsertMode) &&
+        Arrays.equals(projectedFields, that.projectedFields) &&
         Objects.equals(watermarkStrategy, that.watermarkStrategy);
   }
 
@@ -372,6 +402,7 @@ public class KafkaDynamicSource
         specificStartupOffsets,
         startupTimestampMillis,
         upsertMode,
+        projectedFields,
         watermarkStrategy);
   }
 
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/LogDynamicSource.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/LogDynamicSource.java
index 83ade0854..6be4d1698 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/LogDynamicSource.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/table/LogDynamicSource.java
@@ -33,12 +33,15 @@ import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.types.RowKind;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.types.Types;
 
 import javax.annotation.Nullable;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_LOG_CONSISTENCY_GUARANTEE_ENABLE;
 import static com.netease.arctic.flink.table.descriptors.ArcticValidator.ARCTIC_LOG_CONSUMER_CHANGELOG_MODE;
@@ -113,15 +116,20 @@ public class LogDynamicSource extends KafkaDynamicSource {
 
     final FlinkKafkaConsumer<RowData> kafkaConsumer;
 
-    KafkaDeserializationSchemaWrapper<RowData> deserializationSchemaWrapper =
+    final KafkaDeserializationSchemaWrapper<RowData> deserializationSchemaWrapper =
         new KafkaDeserializationSchemaWrapper<>(valueDeserialization);
+    Schema projectedSchema = schema;
+    if (projectedFields != null) {
+      final List<Types.NestedField> columns = schema.columns();
+      projectedSchema = new Schema(Arrays.stream(projectedFields).mapToObj(columns::get).collect(Collectors.toList()));
+    }
     if (topics != null) {
       kafkaConsumer =
           new LogKafkaConsumer(
               topics,
               deserializationSchemaWrapper,
               properties,
-              schema,
+              projectedSchema,
               tableOptions);
     } else {
       kafkaConsumer =
@@ -129,7 +137,7 @@ public class LogDynamicSource extends KafkaDynamicSource {
               topicPattern,
               deserializationSchemaWrapper,
               properties,
-              schema,
+              projectedSchema,
               tableOptions);
     }
 
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/util/Projection.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/util/Projection.java
new file mode 100644
index 000000000..89964ed62
--- /dev/null
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/util/Projection.java
@@ -0,0 +1,434 @@
+/*
+ * 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.flink.util;
+
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.FieldsDataType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.utils.TypeConversions;
+import org.apache.flink.util.Preconditions;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.ROW;
+
+/**
+ * {@link Projection} represents a list of (possibly nested) indexes that can be used to project
+ * data types. A row projection includes both reducing the accessible fields and reordering them.
+ *
+ * <p>NOTE: Copied from Flink.
+ */
+public abstract class Projection {
+
+  // sealed class
+  private Projection() {
+  }
+
+  /**
+   * Projects a (possibly nested) row data type by returning a new data type that only includes
+   * fields of the given index paths.
+   *
+   * <p>When extracting nested fields, the name of the resulting fields is the full path of the
+   * field separated by {@code _}. For example, the field {@code b} inside the row field {@code a}
+   * of the root {@link DataType} is named {@code a_b} in the result {@link DataType}. In case of
+   * naming conflicts the postfix notation '_$%d' is used, where {@code %d} is an arbitrary
+   * number, in order to generate a unique field name. For example if the root {@link DataType}
+   * includes both a field {@code a_b} and a nested row {@code a} with field {@code b}, the result
+   * {@link DataType} will contain one field named {@code a_b} and the other named {@code a_b_1}.
+   */
+  public abstract DataType project(DataType dataType);
+
+  /**
+   * Same as {@link #project(DataType)}, but accepting and returning {@link LogicalType}.
+   */
+  public LogicalType project(LogicalType logicalType) {
+    return this.project(TypeConversions.fromLogicalToDataType(logicalType)).getLogicalType();
+  }
+
+  /**
+   * @return {@code true} whether this projection is nested or not.
+   */
+  public abstract boolean isNested();
+
+  /**
+   * Perform a difference of this {@link Projection} with another {@link Projection}. The result
+   * of this operation is a new {@link Projection} retaining the same ordering of this instance
+   * but with the indexes from {@code other} removed. For example:
+   *
+   * <pre>
+   * <code>
+   * [4, 1, 0, 3, 2] - [4, 2] = [1, 0, 2]
+   * </code>
+   * </pre>
+   *
+   * <p>Note how the index {@code 3} in the minuend becomes {@code 2} because it's rescaled to
+   * project correctly a {@link RowData} or arity 3.
+   *
+   * @param other the subtrahend
+   * @throws IllegalArgumentException when {@code other} is nested.
+   */
+  public abstract Projection difference(Projection other);
+
+  /**
+   * Complement this projection. The returned projection is an ordered projection of fields from 0
+   * to {@code fieldsNumber} except the indexes in this {@link Projection}. For example:
+   *
+   * <pre>
+   * <code>
+   * [4, 2].complement(5) = [0, 1, 3]
+   * </code>
+   * </pre>
+   *
+   * @param fieldsNumber the size of the universe
+   * @throws IllegalStateException if this projection is nested.
+   */
+  public abstract Projection complement(int fieldsNumber);
+
+  /**
+   * Like {@link #complement(int)}, using the {@code dataType} fields count.
+   */
+  public Projection complement(DataType dataType) {
+    return complement(dataType.getLogicalType().getChildren().size());
+  }
+
+  /**
+   * Convert this instance to a projection of top level indexes. The array represents the mapping
+   * of the fields of the original {@link DataType}. For example, {@code [0, 2, 1]} specifies to
+   * include in the following order the 1st field, the 3rd field and the 2nd field of the row.
+   *
+   * @throws IllegalStateException if this projection is nested.
+   */
+  public abstract int[] toTopLevelIndexes();
+
+  /**
+   * Convert this instance to a nested projection index paths. The array represents the mapping of
+   * the fields of the original {@link DataType}, including nested rows. For example, {@code [[0,
+   * 2, 1], ...]} specifies to include the 2nd field of the 3rd field of the 1st field in the
+   * top-level row.
+   */
+  public abstract int[][] toNestedIndexes();
+
+  /**
+   * Create an empty {@link Projection}, that is a projection that projects no fields, returning
+   * an empty {@link DataType}.
+   */
+  public static Projection empty() {
+    return EmptyProjection.INSTANCE;
+  }
+
+  /**
+   * Create a {@link Projection} of the provided {@code indexes}.
+   *
+   * @see #toTopLevelIndexes()
+   */
+  public static Projection of(int[] indexes) {
+    if (indexes.length == 0) {
+      return empty();
+    }
+    return new TopLevelProjection(indexes);
+  }
+
+  /**
+   * Create a {@link Projection} of the provided {@code indexes}.
+   *
+   * @see #toNestedIndexes()
+   */
+  public static Projection of(int[][] indexes) {
+    if (indexes.length == 0) {
+      return empty();
+    }
+    return new NestedProjection(indexes);
+  }
+
+  /**
+   * Create a {@link Projection} of a field range.
+   */
+  public static Projection range(int startInclusive, int endExclusive) {
+    return new TopLevelProjection(IntStream.range(startInclusive, endExclusive).toArray());
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof Projection)) {
+      return false;
+    }
+    Projection other = (Projection) o;
+    if (!this.isNested() && !other.isNested()) {
+      return Arrays.equals(this.toTopLevelIndexes(), other.toTopLevelIndexes());
+    }
+    return Arrays.deepEquals(this.toNestedIndexes(), other.toNestedIndexes());
+  }
+
+  @Override
+  public int hashCode() {
+    if (isNested()) {
+      return Arrays.deepHashCode(toNestedIndexes());
+    }
+    return Arrays.hashCode(toTopLevelIndexes());
+  }
+
+  @Override
+  public String toString() {
+    if (isNested()) {
+      return "Nested projection = " + Arrays.deepToString(toNestedIndexes());
+    }
+    return "Top level projection = " + Arrays.toString(toTopLevelIndexes());
+  }
+
+  private static class EmptyProjection extends Projection {
+
+    static final EmptyProjection INSTANCE = new EmptyProjection();
+
+    private EmptyProjection() {
+    }
+
+    @Override
+    public DataType project(DataType dataType) {
+      return new NestedProjection(toNestedIndexes()).project(dataType);
+    }
+
+    @Override
+    public boolean isNested() {
+      return false;
+    }
+
+    @Override
+    public Projection difference(Projection projection) {
+      return this;
+    }
+
+    @Override
+    public Projection complement(int fieldsNumber) {
+      return new TopLevelProjection(IntStream.range(0, fieldsNumber).toArray());
+    }
+
+    @Override
+    public int[] toTopLevelIndexes() {
+      return new int[0];
+    }
+
+    @Override
+    public int[][] toNestedIndexes() {
+      return new int[0][];
+    }
+  }
+
+  private static class NestedProjection extends Projection {
+
+    final int[][] projection;
+    final boolean nested;
+
+    NestedProjection(int[][] projection) {
+      this.projection = projection;
+      this.nested = Arrays.stream(projection).anyMatch(arr -> arr.length > 1);
+    }
+
+    @Override
+    public DataType project(DataType dataType) {
+      final List<RowType.RowField> updatedFields = new ArrayList<>();
+      final List<DataType> updatedChildren = new ArrayList<>();
+      Set<String> nameDomain = new HashSet<>();
+      int duplicateCount = 0;
+      for (int[] indexPath : this.projection) {
+        DataType fieldType = dataType.getChildren().get(indexPath[0]);
+        LogicalType fieldLogicalType = fieldType.getLogicalType();
+        StringBuilder builder =
+            new StringBuilder(((RowType) dataType.getLogicalType()).getFieldNames().get(indexPath[0]));
+        for (int index = 1; index < indexPath.length; index++) {
+          Preconditions.checkArgument(fieldLogicalType.getTypeRoot() == ROW, "Row data type expected.");
+          RowType rowtype = ((RowType) fieldLogicalType);
+          builder.append("_").append(rowtype.getFieldNames().get(indexPath[index]));
+          fieldLogicalType = rowtype.getFields().get(indexPath[index]).getType();
+          fieldType = fieldType.getChildren().get(indexPath[index]);
+        }
+        String path = builder.toString();
+        while (nameDomain.contains(path)) {
+          path = builder.append("_$").append(duplicateCount++).toString();
+        }
+        updatedFields.add(new RowType.RowField(path, fieldLogicalType));
+        updatedChildren.add(fieldType);
+        nameDomain.add(path);
+      }
+      return new FieldsDataType(
+          new RowType(dataType.getLogicalType().isNullable(), updatedFields),
+          dataType.getConversionClass(),
+          updatedChildren);
+    }
+
+    @Override
+    public boolean isNested() {
+      return nested;
+    }
+
+    @Override
+    public Projection difference(Projection other) {
+      if (other.isNested()) {
+        throw new IllegalArgumentException("Cannot perform difference between nested projection and nested projection");
+      }
+      if (other instanceof EmptyProjection) {
+        return this;
+      }
+      if (!this.isNested()) {
+        return new TopLevelProjection(toTopLevelIndexes()).difference(other);
+      }
+
+      // Extract the indexes to exclude and sort them
+      int[] indexesToExclude = other.toTopLevelIndexes();
+      indexesToExclude = Arrays.copyOf(indexesToExclude, indexesToExclude.length);
+      Arrays.sort(indexesToExclude);
+
+      List<int[]> resultProjection = Arrays.stream(projection).collect(Collectors.toCollection(ArrayList::new));
+
+      ListIterator<int[]> resultProjectionIterator = resultProjection.listIterator();
+      while (resultProjectionIterator.hasNext()) {
+        int[] indexArr = resultProjectionIterator.next();
+
+        // Let's check if the index is inside the indexesToExclude array
+        int searchResult = Arrays.binarySearch(indexesToExclude, indexArr[0]);
+        if (searchResult >= 0) {
+          // Found, we need to remove it
+          resultProjectionIterator.remove();
+        } else {
+          // Not found, let's compute the offset.
+          // Offset is the index where the projection index should be inserted in the
+          // indexesToExclude array
+          int offset = (-(searchResult) - 1);
+          if (offset != 0) {
+            indexArr[0] = indexArr[0] - offset;
+          }
+        }
+      }
+
+      return new NestedProjection(resultProjection.toArray(new int[0][]));
+    }
+
+    @Override
+    public Projection complement(int fieldsNumber) {
+      if (isNested()) {
+        throw new IllegalStateException("Cannot perform complement of a nested projection");
+      }
+      return new TopLevelProjection(toTopLevelIndexes()).complement(fieldsNumber);
+    }
+
+    @Override
+    public int[] toTopLevelIndexes() {
+      if (isNested()) {
+        throw new IllegalStateException("Cannot convert a nested projection to a top level projection");
+      }
+      return Arrays.stream(projection).mapToInt(arr -> arr[0]).toArray();
+    }
+
+    @Override
+    public int[][] toNestedIndexes() {
+      return projection;
+    }
+  }
+
+  private static class TopLevelProjection extends Projection {
+
+    final int[] projection;
+
+    TopLevelProjection(int[] projection) {
+      this.projection = projection;
+    }
+
+    @Override
+    public DataType project(DataType dataType) {
+      return new NestedProjection(toNestedIndexes()).project(dataType);
+    }
+
+    @Override
+    public boolean isNested() {
+      return false;
+    }
+
+    @Override
+    public Projection difference(Projection other) {
+      if (other.isNested()) {
+        throw new IllegalArgumentException(
+            "Cannot perform difference between top level projection and nested projection");
+      }
+      if (other instanceof EmptyProjection) {
+        return this;
+      }
+
+      // Extract the indexes to exclude and sort them
+      int[] indexesToExclude = other.toTopLevelIndexes();
+      indexesToExclude = Arrays.copyOf(indexesToExclude, indexesToExclude.length);
+      Arrays.sort(indexesToExclude);
+
+      List<Integer> resultProjection =
+          Arrays.stream(projection).boxed().collect(Collectors.toCollection(ArrayList::new));
+
+      ListIterator<Integer> resultProjectionIterator = resultProjection.listIterator();
+      while (resultProjectionIterator.hasNext()) {
+        int index = resultProjectionIterator.next();
+
+        // Let's check if the index is inside the indexesToExclude array
+        int searchResult = Arrays.binarySearch(indexesToExclude, index);
+        if (searchResult >= 0) {
+          // Found, we need to remove it
+          resultProjectionIterator.remove();
+        } else {
+          // Not found, let's compute the offset.
+          // Offset is the index where the projection index should be inserted in the
+          // indexesToExclude array
+          int offset = (-(searchResult) - 1);
+          if (offset != 0) {
+            resultProjectionIterator.set(index - offset);
+          }
+        }
+      }
+
+      return new TopLevelProjection(resultProjection.stream().mapToInt(i -> i).toArray());
+    }
+
+    @Override
+    public Projection complement(int fieldsNumber) {
+      int[] indexesToExclude = Arrays.copyOf(projection, projection.length);
+      Arrays.sort(indexesToExclude);
+
+      return new TopLevelProjection(IntStream.range(0, fieldsNumber)
+          .filter(i -> Arrays.binarySearch(indexesToExclude, i) < 0)
+          .toArray());
+    }
+
+    @Override
+    public int[] toTopLevelIndexes() {
+      return projection;
+    }
+
+    @Override
+    public int[][] toNestedIndexes() {
+      return Arrays.stream(projection).mapToObj(i -> new int[] {i}).toArray(int[][]::new);
+    }
+  }
+}
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 356037228..502ad330a 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
@@ -255,6 +255,7 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
 
       meta.putToProperties(TableProperties.TABLE_CREATE_TIME, String.valueOf(System.currentTimeMillis()));
       meta.putToProperties(TableProperties.BASE_HIVE_LOCATION_ROOT, hiveLocation);
+      meta.putToProperties(org.apache.iceberg.TableProperties.FORMAT_VERSION, "2");
 
       Table table = tableMetaStore.doAs(() -> {
         try {
diff --git a/hive/src/main/java/com/netease/arctic/hive/exceptions/CannotAlterHiveLocationException.java b/hive/src/main/java/com/netease/arctic/hive/exceptions/CannotAlterHiveLocationException.java
new file mode 100644
index 000000000..ede35623e
--- /dev/null
+++ b/hive/src/main/java/com/netease/arctic/hive/exceptions/CannotAlterHiveLocationException.java
@@ -0,0 +1,28 @@
+/*
+ * 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.exceptions;
+
+/**
+ * Exception will be thrown when alter hive location check failed
+ */
+public class CannotAlterHiveLocationException extends IllegalStateException {
+  public CannotAlterHiveLocationException(String message) {
+    super(message);
+  }
+}
diff --git a/hive/src/main/java/com/netease/arctic/hive/op/OverwriteHiveFiles.java b/hive/src/main/java/com/netease/arctic/hive/op/OverwriteHiveFiles.java
index 1afb44c10..88f36be89 100644
--- a/hive/src/main/java/com/netease/arctic/hive/op/OverwriteHiveFiles.java
+++ b/hive/src/main/java/com/netease/arctic/hive/op/OverwriteHiveFiles.java
@@ -3,9 +3,12 @@ package com.netease.arctic.hive.op;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import com.netease.arctic.hive.HMSClient;
+import com.netease.arctic.hive.exceptions.CannotAlterHiveLocationException;
 import com.netease.arctic.hive.table.UnkeyedHiveTable;
 import com.netease.arctic.hive.utils.HivePartitionUtil;
 import com.netease.arctic.utils.FileUtil;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.metastore.PartitionDropOptions;
 import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
 import org.apache.hadoop.hive.metastore.api.Partition;
@@ -78,10 +81,9 @@ public class OverwriteHiveFiles implements OverwriteFiles {
   @Override
   public OverwriteFiles addFile(DataFile file) {
     delegate.addFile(file);
-
-    String tableLocation = hiveTable.getSd().getLocation();
+    String hiveLocationRoot = table.hiveLocation();
     String dataFileLocation = file.path().toString();
-    if (dataFileLocation.toLowerCase().contains(tableLocation.toLowerCase())) {
+    if (dataFileLocation.toLowerCase().contains(hiveLocationRoot.toLowerCase())) {
       // only handle file in hive location
       this.addFiles.add(file);
     }
@@ -91,9 +93,9 @@ public class OverwriteHiveFiles implements OverwriteFiles {
   @Override
   public OverwriteFiles deleteFile(DataFile file) {
     delegate.deleteFile(file);
-    String tableLocation = hiveTable.getSd().getLocation();
+    String hiveLocation = table.hiveLocation();
     String dataFileLocation = file.path().toString();
-    if (dataFileLocation.toLowerCase().contains(tableLocation.toLowerCase())) {
+    if (dataFileLocation.toLowerCase().contains(hiveLocation.toLowerCase())) {
       // only handle file in hive location
       this.deleteFiles.add(file);
     }
@@ -161,7 +163,7 @@ public class OverwriteHiveFiles implements OverwriteFiles {
   public void commit() {
     if (!table.spec().isUnpartitioned()) {
       this.partitionToDelete = getDeletePartition();
-      this.partitionToCreate = getCreatePartition();
+      this.partitionToCreate = getCreatePartition(this.partitionToDelete);
     }
 
     delegate.commit();
@@ -178,7 +180,7 @@ public class OverwriteHiveFiles implements OverwriteFiles {
     return delegate.updateEvent();
   }
 
-  protected List<Partition> getCreatePartition() {
+  protected List<Partition> getCreatePartition(List<Partition> partitionToDelete) {
     if (this.addFiles.isEmpty()) {
       return Lists.newArrayList();
     }
@@ -209,6 +211,13 @@ public class OverwriteHiveFiles implements OverwriteFiles {
       Partition p = HivePartitionUtil.newPartition(hiveTable, values, location, dataFiles);
       partitions.add(p);
     }
+    for (String val : partitionLocationMap.keySet()) {
+      String partitionLocation = partitionLocationMap.get(val);
+      List<DataFile> dataFiles = partitionDataFileMap.get(val);
+      checkCreatePartitionDataFiles(dataFiles, partitionLocation);
+    }
+
+    partitions = filterNewPartitionNonExists(partitions, partitionToDelete);
     return partitions;
   }
 
@@ -226,20 +235,107 @@ public class OverwriteHiveFiles implements OverwriteFiles {
     Types.StructType partitionSchema = table.spec().partitionType();
 
     Set<String> checkedPartitionValues = Sets.newHashSet();
+    Set<Path> deleteFileLocations = Sets.newHashSet();
 
     for (DataFile dataFile : deleteFiles) {
       List<String> values = HivePartitionUtil.partitionValuesAsList(dataFile.partition(), partitionSchema);
       String pathValue = Joiner.on("/").join(values);
+      deleteFileLocations.add(new Path(dataFile.path().toString()));
       if (checkedPartitionValues.contains(pathValue)) {
         continue;
       }
-
       try {
         Partition partition = hmsClient.run(c -> c.getPartition(db, tableName, values));
         partitions.add(partition);
-        checkedPartitionValues.add(pathValue);
       } catch (NoSuchObjectException e) {
-        checkedPartitionValues.add(pathValue);
+        // pass do nothing
+      } catch (TException | InterruptedException e) {
+        throw new RuntimeException(e);
+      }
+      checkedPartitionValues.add(pathValue);
+    }
+
+    partitions.forEach(p -> checkPartitionDelete(deleteFileLocations, p));
+    return partitions;
+  }
+
+  private void checkPartitionDelete(Set<Path> deleteFiles, Partition partition) {
+    String partitionLocation = partition.getSd().getLocation();
+    List<FileStatus> files = table.io().list(partitionLocation);
+    for (FileStatus f : files) {
+      Path filePath = f.getPath();
+      if (!deleteFiles.contains(filePath)) {
+        throw new CannotAlterHiveLocationException(
+            "can't delete hive partition: " + partitionToString(partition) + ", file under partition is not deleted: " +
+                filePath.toString());
+      }
+    }
+  }
+
+  /**
+   * check all file with same partition key under same path
+   */
+  private void checkCreatePartitionDataFiles(List<DataFile> addFiles, String partitionLocation) {
+    Path partitionPath = new Path(partitionLocation);
+    for (DataFile df : addFiles) {
+      String fileDir = FileUtil.getFileDir(df.path().toString());
+      Path dirPath = new Path(fileDir);
+      if (!partitionPath.equals(dirPath)) {
+        throw new CannotAlterHiveLocationException(
+            "can't create new hive location: " + partitionLocation + " for data file: " + df.path().toString() +
+                " is not under partition location path"
+        );
+      }
+    }
+  }
+
+  /**
+   * filter partitionToCreate. make sure all partition non-exist in hive. or
+   * 0. partition is able to delete.
+   * 0.1 - not same location, allow to create
+   * 0.2 - same location, can't create ( delete partition will not delete files )
+   * 1. exists but location is same. skip
+   * 2. exists but location is not same, throw {@link CannotAlterHiveLocationException}
+   */
+  private List<Partition> filterNewPartitionNonExists(
+      List<Partition> partitionToCreate,
+      List<Partition> partitionToDelete) {
+    List<Partition> partitions = Lists.newArrayList();
+    Map<String, Partition> deletePartitionValueMap = Maps.newHashMap();
+    for (Partition p : partitionToDelete) {
+      String partValue = Joiner.on("/").join(p.getValues());
+      deletePartitionValueMap.put(partValue, p);
+    }
+
+    for (Partition p : partitionToCreate) {
+      String partValue = Joiner.on("/").join(p.getValues());
+      String location = p.getSd().getLocation();
+      Partition toDelete = deletePartitionValueMap.get(partValue);
+      if (toDelete != null) {
+        String deleteLocation = toDelete.getSd().getLocation();
+        // if exists partition to delete with same value
+        // make sure location is different
+        if (isPathEquals(location, deleteLocation)) {
+          throw new CannotAlterHiveLocationException("can't create new partition: " + partitionToString(p) + ", this " +
+              "partition will be " +
+              "delete and re-create with same location");
+        } else {
+          partitions.add(p);
+          continue;
+        }
+      }
+
+      try {
+        Partition partitionInHive = hmsClient.run(c -> c.getPartition(db, tableName, p.getValues()));
+        String locationInHive = partitionInHive.getSd().getLocation();
+        if (isPathEquals(location, locationInHive)) {
+          // exists same location, skip create operation
+          continue;
+        }
+        throw new CannotAlterHiveLocationException("can't create new partition: " + partitionToString(p) +
+            ", this partition exists in hive with different location: " + locationInHive);
+      } catch (NoSuchObjectException e) {
+        partitions.add(p);
       } catch (TException | InterruptedException e) {
         throw new RuntimeException(e);
       }
@@ -322,4 +418,15 @@ public class OverwriteHiveFiles implements OverwriteFiles {
       throw new RuntimeException(e);
     }
   }
+
+  private boolean isPathEquals(String pathA, String pathB) {
+    Path path1 = new Path(pathA);
+    Path path2 = new Path(pathB);
+    return path1.equals(path2);
+  }
+
+  private String partitionToString(Partition p) {
+    return "Partition(values: [" + Joiner.on("/").join(p.getValues()) +
+        "], location: " + p.getSd().getLocation() + ")";
+  }
 }
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 bb56d3f44..4b159dfbd 100644
--- a/hive/src/main/java/com/netease/arctic/hive/op/ReplaceHivePartitions.java
+++ b/hive/src/main/java/com/netease/arctic/hive/op/ReplaceHivePartitions.java
@@ -1,9 +1,11 @@
 package com.netease.arctic.hive.op;
 
 import com.netease.arctic.hive.HMSClient;
+import com.netease.arctic.hive.exceptions.CannotAlterHiveLocationException;
 import com.netease.arctic.hive.table.UnkeyedHiveTable;
 import com.netease.arctic.hive.utils.HivePartitionUtil;
 import com.netease.arctic.utils.FileUtil;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
 import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.Table;
@@ -61,7 +63,7 @@ public class ReplaceHivePartitions implements ReplacePartitions {
   @Override
   public ReplacePartitions addFile(DataFile file) {
     delegate.addFile(file);
-    String tableLocation = hiveTable.getSd().getLocation();
+    String tableLocation = table.hiveLocation();
     String dataFileLocation = file.path().toString();
     if (dataFileLocation.toLowerCase().contains(tableLocation.toLowerCase())) {
       // only handle file in hive location
@@ -139,6 +141,7 @@ public class ReplaceHivePartitions implements ReplacePartitions {
       partitionDataFileMap.get(value).add(d);
       partitionValueMap.put(value, partitionValues);
     }
+    partitionLocationMap.forEach((k, v) -> checkDataFileInSameLocation(v, partitionDataFileMap.get(k)));
 
     for (String val : partitionValueMap.keySet()) {
       List<String> values = partitionValueMap.get(val);
@@ -190,6 +193,20 @@ public class ReplaceHivePartitions implements ReplacePartitions {
     }
   }
 
+  private void checkDataFileInSameLocation(String partitionLocation, List<DataFile> files) {
+    Path partitionPath = new Path(partitionLocation);
+    for (DataFile df : files) {
+      String fileDir = FileUtil.getFileDir(df.path().toString());
+      Path dirPath = new Path(fileDir);
+      if (!partitionPath.equals(dirPath)) {
+        throw new CannotAlterHiveLocationException(
+            "can't create new hive location: " + partitionLocation + " for data file: " + df.path().toString() +
+                " is not under partition location path"
+        );
+      }
+    }
+  }
+
   private static void rewriteHivePartitions(Partition partition, String location, List<DataFile> dataFiles) {
     partition.getSd().setLocation(location);
     int lastAccessTime = (int) (System.currentTimeMillis() / 1000);
diff --git a/spark/v2.3/spark/pom.xml b/spark/v2.3/spark/pom.xml
index 21e1a6c9a..65ab698ef 100644
--- a/spark/v2.3/spark/pom.xml
+++ b/spark/v2.3/spark/pom.xml
@@ -175,17 +175,11 @@
             </exclusions>
         </dependency>
 
-<!--        <dependency>-->
-<!--            <groupId>org.apache.iceberg</groupId>-->
-<!--            <artifactId>iceberg-spark3-extensions</artifactId>-->
-<!--            <version>${iceberg.version}</version>-->
-<!--            <exclusions>-->
-<!--                <exclusion>-->
-<!--                    <groupId>org.apache.parquet</groupId>-->
-<!--                    <artifactId>parquet-column</artifactId>-->
-<!--                </exclusion>-->
-<!--            </exclusions>-->
-<!--        </dependency>-->
+        <dependency>
+            <groupId>com.netease.arctic</groupId>
+            <artifactId>arctic-hive</artifactId>
+            <version>${project.version}</version>
+        </dependency>
 
         <!-- test dependency -->
         <dependency>
@@ -223,6 +217,14 @@
             <version>2.3.7</version>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>com.netease.arctic</groupId>
+            <artifactId>arctic-hive</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/spark/v3.1/spark/pom.xml b/spark/v3.1/spark/pom.xml
index 5f36c2243..10ef83d34 100644
--- a/spark/v3.1/spark/pom.xml
+++ b/spark/v3.1/spark/pom.xml
@@ -219,6 +219,8 @@
             <groupId>com.netease.arctic</groupId>
             <artifactId>arctic-hive</artifactId>
             <version>${project.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
         </dependency>
     </dependencies>
 
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkTable.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkTable.java
index 3d243b9df..ef89e95af 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkTable.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkTable.java
@@ -172,13 +172,13 @@ public class ArcticSparkTable implements org.apache.spark.sql.connector.catalog.
 
     // use only name in order to correctly invalidate Spark cache
     ArcticSparkTable that = (ArcticSparkTable) other;
-    return arcticTable.name().equals(that.arcticTable.name());
+    return arcticTable.id().equals(that.arcticTable.id());
   }
 
   @Override
   public int hashCode() {
     // use only name in order to correctly invalidate Spark cache
-    return arcticTable.name().hashCode();
+    return arcticTable.id().hashCode();
   }
 
   @Override


Test patch:
diff --git a/core/src/test/java/com/netease/arctic/trace/TableTracerTest.java b/core/src/test/java/com/netease/arctic/trace/TableTracerTest.java
index e239df0d5..fb80022ab 100644
--- a/core/src/test/java/com/netease/arctic/trace/TableTracerTest.java
+++ b/core/src/test/java/com/netease/arctic/trace/TableTracerTest.java
@@ -20,6 +20,7 @@ package com.netease.arctic.trace;
 
 import com.google.common.collect.Lists;
 import com.netease.arctic.TableTestBase;
+import com.netease.arctic.ams.api.CommitMetaProducer;
 import com.netease.arctic.ams.api.Constants;
 import com.netease.arctic.ams.api.DataFile;
 import com.netease.arctic.ams.api.TableChange;
@@ -79,6 +80,22 @@ public class TableTracerTest extends TableTestBase {
         new org.apache.iceberg.DataFile[]{});
   }
 
+  @Test
+  public void testTraceAppendFilesByOptimize() {
+    testTable.newAppend()
+        .appendFile(FILE_A)
+        .appendFile(FILE_B)
+        .set(SnapshotSummary.SNAPSHOT_PRODUCER, CommitMetaProducer.OPTIMIZE.name())
+        .commit();
+
+    List<TableCommitMeta> TableCommitMetas = AMS.handler().getTableCommitMetas().get(TABLE_ID.buildTableIdentifier());
+    Assert.assertEquals(1, TableCommitMetas.size());
+    TableCommitMeta commitMeta = TableCommitMetas.get(0);
+    Assert.assertSame(commitMeta.getCommitMetaProducer(), CommitMetaProducer.OPTIMIZE);
+    validateCommitMeta(commitMeta, DataOperations.APPEND, new org.apache.iceberg.DataFile[]{FILE_A, FILE_B},
+        new org.apache.iceberg.DataFile[]{});
+  }
+
   @Test
   public void testTraceFastAppend() {
     testTable.newFastAppend()
@@ -111,6 +128,22 @@ public class TableTracerTest extends TableTestBase {
         new org.apache.iceberg.DataFile[]{});
   }
 
+  @Test
+  public void testTraceFastAppendByOptimize() {
+    testTable.newFastAppend()
+        .appendFile(FILE_A)
+        .appendFile(FILE_B)
+        .set(SnapshotSummary.SNAPSHOT_PRODUCER, CommitMetaProducer.OPTIMIZE.name())
+        .commit();
+
+    List<TableCommitMeta> TableCommitMetas = AMS.handler().getTableCommitMetas().get(TABLE_ID.buildTableIdentifier());
+    Assert.assertEquals(1, TableCommitMetas.size());
+    TableCommitMeta commitMeta = TableCommitMetas.get(0);
+    Assert.assertSame(commitMeta.getCommitMetaProducer(), CommitMetaProducer.OPTIMIZE);
+    validateCommitMeta(commitMeta, DataOperations.APPEND, new org.apache.iceberg.DataFile[]{FILE_A, FILE_B},
+        new org.apache.iceberg.DataFile[]{});
+  }
+
   @Test
   public void testTraceOverwrite() {
     testTable.newFastAppend()
@@ -155,6 +188,28 @@ public class TableTracerTest extends TableTestBase {
         new org.apache.iceberg.DataFile[]{FILE_A, FILE_B});
   }
 
+  @Test
+  public void testTraceOverwriteByOptimize() {
+    testTable.newFastAppend()
+        .appendFile(FILE_A)
+        .appendFile(FILE_B)
+        .commit();
+
+    testTable.newOverwrite()
+        .deleteFile(FILE_A)
+        .deleteFile(FILE_B)
+        .addFile(FILE_C)
+        .set(SnapshotSummary.SNAPSHOT_PRODUCER, CommitMetaProducer.OPTIMIZE.name())
+        .commit();
+
+    List<TableCommitMeta> TableCommitMetas = AMS.handler().getTableCommitMetas().get(TABLE_ID.buildTableIdentifier());
+    Assert.assertEquals(2, TableCommitMetas.size());
+    TableCommitMeta commitMeta = TableCommitMetas.get(1);
+    Assert.assertSame(commitMeta.getCommitMetaProducer(), CommitMetaProducer.OPTIMIZE);
+    validateCommitMeta(commitMeta, DataOperations.OVERWRITE, new org.apache.iceberg.DataFile[]{FILE_C},
+        new org.apache.iceberg.DataFile[]{FILE_A, FILE_B});
+  }
+
   @Test
   public void testTraceRewrite() {
     testTable.newFastAppend()
@@ -195,6 +250,26 @@ public class TableTracerTest extends TableTestBase {
         new org.apache.iceberg.DataFile[]{FILE_A, FILE_B});
   }
 
+  @Test
+  public void testTraceRewriteByOptimize() {
+    testTable.newFastAppend()
+        .appendFile(FILE_A)
+        .appendFile(FILE_B)
+        .commit();
+
+    testTable.newRewrite()
+        .rewriteFiles(Sets.newHashSet(FILE_A, FILE_B), Sets.newHashSet(FILE_C))
+        .set(SnapshotSummary.SNAPSHOT_PRODUCER, CommitMetaProducer.OPTIMIZE.name())
+        .commit();
+
+    List<TableCommitMeta> TableCommitMetas = AMS.handler().getTableCommitMetas().get(TABLE_ID.buildTableIdentifier());
+    Assert.assertEquals(2, TableCommitMetas.size());
+    TableCommitMeta commitMeta = TableCommitMetas.get(1);
+    Assert.assertSame(commitMeta.getCommitMetaProducer(), CommitMetaProducer.OPTIMIZE);
+    validateCommitMeta(commitMeta, DataOperations.REPLACE, new org.apache.iceberg.DataFile[]{FILE_C},
+        new org.apache.iceberg.DataFile[]{FILE_A, FILE_B});
+  }
+
   @Test
   public void testMultipleOperationInTx() {
     Transaction transaction = testTable.newTransaction();
@@ -224,6 +299,38 @@ public class TableTracerTest extends TableTestBase {
         new org.apache.iceberg.DataFile[]{FILE_C}, new org.apache.iceberg.DataFile[]{FILE_A});
   }
 
+  @Test
+  public void testMultipleOperationInTxByOptimize() {
+    Transaction transaction = testTable.newTransaction();
+    transaction.newAppend()
+        .appendFile(FILE_A)
+        .appendFile(FILE_B)
+        .set(SnapshotSummary.SNAPSHOT_PRODUCER, CommitMetaProducer.OPTIMIZE.name())
+        .commit();
+
+    transaction.newOverwrite()
+        .deleteFile(FILE_A)
+        .addFile(FILE_C)
+        .set(SnapshotSummary.SNAPSHOT_PRODUCER, CommitMetaProducer.OPTIMIZE.name())
+        .commit();
+
+    Assert.assertFalse(AMS.handler().getTableCommitMetas().containsKey(TABLE_ID.buildTableIdentifier()));
+
+    transaction.commitTransaction();
+
+    List<Snapshot> snapshots = Lists.newArrayList(testTable.snapshots());
+    Assert.assertEquals(2, snapshots.size());
+    List<TableCommitMeta> TableCommitMetas = AMS.handler().getTableCommitMetas().get(TABLE_ID.buildTableIdentifier());
+    Assert.assertEquals(1, TableCommitMetas.size());
+    TableCommitMeta commitMeta = TableCommitMetas.get(0);
+    Assert.assertSame(commitMeta.getCommitMetaProducer(), CommitMetaProducer.OPTIMIZE);
+    Assert.assertEquals(2, commitMeta.getChanges().size());
+    validateTableChange(snapshots.get(0), commitMeta.getChanges().get(0),
+        new org.apache.iceberg.DataFile[]{FILE_A, FILE_B}, new org.apache.iceberg.DataFile[]{});
+    validateTableChange(snapshots.get(1), commitMeta.getChanges().get(1),
+        new org.apache.iceberg.DataFile[]{FILE_C}, new org.apache.iceberg.DataFile[]{FILE_A});
+  }
+
   @Test
   public void testTraceRemovePosDeleteInternal() throws Exception {
     testKeyedTable.baseTable().newAppend().appendFile(FILE_A).commit();
diff --git a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/util/ProjectionTest.java b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/util/ProjectionTest.java
new file mode 100644
index 000000000..692e6cbe1
--- /dev/null
+++ b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/util/ProjectionTest.java
@@ -0,0 +1,172 @@
+/*
+ * 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.flink.util;
+
+import org.apache.flink.table.types.DataType;
+
+import org.junit.jupiter.api.Test;
+
+import static org.apache.flink.table.api.DataTypes.BIGINT;
+import static org.apache.flink.table.api.DataTypes.BOOLEAN;
+import static org.apache.flink.table.api.DataTypes.DOUBLE;
+import static org.apache.flink.table.api.DataTypes.FIELD;
+import static org.apache.flink.table.api.DataTypes.INT;
+import static org.apache.flink.table.api.DataTypes.ROW;
+import static org.apache.flink.table.api.DataTypes.STRING;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class ProjectionTest {
+
+  @Test
+  void testTopLevelProject() {
+    assertThat(
+        Projection.of(new int[] {2, 1})
+            .project(
+                ROW(
+                    FIELD("f0", BIGINT()),
+                    FIELD("f1", STRING()),
+                    FIELD("f2", INT()))))
+        .isEqualTo(ROW(FIELD("f2", INT()), FIELD("f1", STRING())));
+  }
+
+  @Test
+  void testNestedProject() {
+    final DataType thirdLevelRow =
+        ROW(FIELD("c0", BOOLEAN()), FIELD("c1", DOUBLE()), FIELD("c2", INT()));
+    final DataType secondLevelRow =
+        ROW(FIELD("b0", BOOLEAN()), FIELD("b1", thirdLevelRow), FIELD("b2", INT()));
+    final DataType topLevelRow =
+        ROW(FIELD("a0", INT()), FIELD("a1", secondLevelRow), FIELD("a1_b1_c0", INT()));
+
+    assertThat(Projection.of(new int[][] {{0}, {1, 1, 0}}).project(topLevelRow))
+        .isEqualTo(ROW(FIELD("a0", INT()), FIELD("a1_b1_c0", BOOLEAN())));
+    assertThat(Projection.of(new int[][] {{1, 1}, {0}}).project(topLevelRow))
+        .isEqualTo(ROW(FIELD("a1_b1", thirdLevelRow), FIELD("a0", INT())));
+    assertThat(
+        Projection.of(new int[][] {{1, 1, 2}, {1, 1, 1}, {1, 1, 0}})
+            .project(topLevelRow))
+        .isEqualTo(
+            ROW(
+                FIELD("a1_b1_c2", INT()),
+                FIELD("a1_b1_c1", DOUBLE()),
+                FIELD("a1_b1_c0", BOOLEAN())));
+    assertThat(Projection.of(new int[][] {{1, 1, 0}, {2}}).project(topLevelRow))
+        .isEqualTo(ROW(FIELD("a1_b1_c0", BOOLEAN()), FIELD("a1_b1_c0_$0", INT())));
+  }
+
+  @Test
+  void testIsNested() {
+    assertThat(Projection.of(new int[] {2, 1}).isNested()).isFalse();
+    assertThat(Projection.of(new int[][] {new int[] {1}, new int[] {3}}).isNested()).isFalse();
+    assertThat(
+        Projection.of(new int[][] {new int[] {1}, new int[] {1, 2}, new int[] {3}})
+            .isNested())
+        .isTrue();
+  }
+
+  @Test
+  void testDifference() {
+    assertThat(
+        Projection.of(new int[] {4, 1, 0, 3, 2})
+            .difference(Projection.of(new int[] {4, 2})))
+        .isEqualTo(Projection.of(new int[] {1, 0, 2}));
+
+    assertThat(
+        Projection.of(
+                new int[][] {
+                    new int[] {4},
+                    new int[] {1, 3},
+                    new int[] {0},
+                    new int[] {3, 1},
+                    new int[] {2}
+                })
+            .difference(Projection.of(new int[] {4, 2})))
+        .isEqualTo(
+            Projection.of(
+                new int[][] {new int[] {1, 3}, new int[] {0}, new int[] {2, 1}}));
+
+    assertThatThrownBy(
+        () ->
+            Projection.of(new int[] {1, 2, 3, 4})
+                .difference(
+                    Projection.of(
+                        new int[][] {
+                            new int[] {2}, new int[] {3, 4}
+                        })))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void testComplement() {
+    assertThat(Projection.of(new int[] {4, 1, 2}).complement(5))
+        .isEqualTo(Projection.of(new int[] {0, 3}));
+
+    assertThat(
+        Projection.of(new int[][] {new int[] {4}, new int[] {1}, new int[] {2}})
+            .complement(5))
+        .isEqualTo(Projection.of(new int[] {0, 3}));
+
+    assertThatThrownBy(
+        () ->
+            Projection.of(
+                    new int[][] {
+                        new int[] {4}, new int[] {1, 3}, new int[] {2}
+                    })
+                .complement(10))
+        .isInstanceOf(IllegalStateException.class);
+  }
+
+  @Test
+  void testToTopLevelIndexes() {
+    assertThat(Projection.of(new int[] {1, 2, 3, 4}).toTopLevelIndexes())
+        .isEqualTo(new int[] {1, 2, 3, 4});
+
+    assertThat(
+        Projection.of(new int[][] {new int[] {4}, new int[] {1}, new int[] {2}})
+            .toTopLevelIndexes())
+        .isEqualTo(new int[] {4, 1, 2});
+
+    assertThatThrownBy(
+        () ->
+            Projection.of(
+                    new int[][] {
+                        new int[] {4}, new int[] {1, 3}, new int[] {2}
+                    })
+                .toTopLevelIndexes())
+        .isInstanceOf(IllegalStateException.class);
+  }
+
+  @Test
+  void testToNestedIndexes() {
+    assertThat(Projection.of(new int[] {1, 2, 3, 4}).toNestedIndexes())
+        .isEqualTo(
+            new int[][] {new int[] {1}, new int[] {2}, new int[] {3}, new int[] {4}});
+    assertThat(
+        Projection.of(new int[][] {new int[] {4}, new int[] {1, 3}, new int[] {2}})
+            .toNestedIndexes())
+        .isEqualTo(new int[][] {new int[] {4}, new int[] {1, 3}, new int[] {2}});
+  }
+
+  @Test
+  void testEquals() {
+    assertThat(Projection.of(new int[][] {new int[] {1}, new int[] {2}, new int[] {3}}))
+        .isEqualTo(Projection.of(new int[] {1, 2, 3}));
+  }
+}
diff --git a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/util/ProjectionTest.java b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/util/ProjectionTest.java
new file mode 100644
index 000000000..f81ffb8e8
--- /dev/null
+++ b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/util/ProjectionTest.java
@@ -0,0 +1,171 @@
+/*
+ * 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.flink.util;
+
+import org.apache.flink.table.types.DataType;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.flink.table.api.DataTypes.BIGINT;
+import static org.apache.flink.table.api.DataTypes.BOOLEAN;
+import static org.apache.flink.table.api.DataTypes.DOUBLE;
+import static org.apache.flink.table.api.DataTypes.FIELD;
+import static org.apache.flink.table.api.DataTypes.INT;
+import static org.apache.flink.table.api.DataTypes.ROW;
+import static org.apache.flink.table.api.DataTypes.STRING;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class ProjectionTest {
+
+  @Test
+  void testTopLevelProject() {
+    assertThat(
+        Projection.of(new int[] {2, 1})
+            .project(
+                ROW(
+                    FIELD("f0", BIGINT()),
+                    FIELD("f1", STRING()),
+                    FIELD("f2", INT()))))
+        .isEqualTo(ROW(FIELD("f2", INT()), FIELD("f1", STRING())));
+  }
+
+  @Test
+  void testNestedProject() {
+    final DataType thirdLevelRow =
+        ROW(FIELD("c0", BOOLEAN()), FIELD("c1", DOUBLE()), FIELD("c2", INT()));
+    final DataType secondLevelRow =
+        ROW(FIELD("b0", BOOLEAN()), FIELD("b1", thirdLevelRow), FIELD("b2", INT()));
+    final DataType topLevelRow =
+        ROW(FIELD("a0", INT()), FIELD("a1", secondLevelRow), FIELD("a1_b1_c0", INT()));
+
+    assertThat(Projection.of(new int[][] {{0}, {1, 1, 0}}).project(topLevelRow))
+        .isEqualTo(ROW(FIELD("a0", INT()), FIELD("a1_b1_c0", BOOLEAN())));
+    assertThat(Projection.of(new int[][] {{1, 1}, {0}}).project(topLevelRow))
+        .isEqualTo(ROW(FIELD("a1_b1", thirdLevelRow), FIELD("a0", INT())));
+    assertThat(
+        Projection.of(new int[][] {{1, 1, 2}, {1, 1, 1}, {1, 1, 0}})
+            .project(topLevelRow))
+        .isEqualTo(
+            ROW(
+                FIELD("a1_b1_c2", INT()),
+                FIELD("a1_b1_c1", DOUBLE()),
+                FIELD("a1_b1_c0", BOOLEAN())));
+    assertThat(Projection.of(new int[][] {{1, 1, 0}, {2}}).project(topLevelRow))
+        .isEqualTo(ROW(FIELD("a1_b1_c0", BOOLEAN()), FIELD("a1_b1_c0_$0", INT())));
+  }
+
+  @Test
+  void testIsNested() {
+    assertThat(Projection.of(new int[] {2, 1}).isNested()).isFalse();
+    assertThat(Projection.of(new int[][] {new int[] {1}, new int[] {3}}).isNested()).isFalse();
+    assertThat(
+        Projection.of(new int[][] {new int[] {1}, new int[] {1, 2}, new int[] {3}})
+            .isNested())
+        .isTrue();
+  }
+
+  @Test
+  void testDifference() {
+    assertThat(
+        Projection.of(new int[] {4, 1, 0, 3, 2})
+            .difference(Projection.of(new int[] {4, 2})))
+        .isEqualTo(Projection.of(new int[] {1, 0, 2}));
+
+    assertThat(
+        Projection.of(
+                new int[][] {
+                    new int[] {4},
+                    new int[] {1, 3},
+                    new int[] {0},
+                    new int[] {3, 1},
+                    new int[] {2}
+                })
+            .difference(Projection.of(new int[] {4, 2})))
+        .isEqualTo(
+            Projection.of(
+                new int[][] {new int[] {1, 3}, new int[] {0}, new int[] {2, 1}}));
+
+    assertThatThrownBy(
+        () ->
+            Projection.of(new int[] {1, 2, 3, 4})
+                .difference(
+                    Projection.of(
+                        new int[][] {
+                            new int[] {2}, new int[] {3, 4}
+                        })))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void testComplement() {
+    assertThat(Projection.of(new int[] {4, 1, 2}).complement(5))
+        .isEqualTo(Projection.of(new int[] {0, 3}));
+
+    assertThat(
+        Projection.of(new int[][] {new int[] {4}, new int[] {1}, new int[] {2}})
+            .complement(5))
+        .isEqualTo(Projection.of(new int[] {0, 3}));
+
+    assertThatThrownBy(
+        () ->
+            Projection.of(
+                    new int[][] {
+                        new int[] {4}, new int[] {1, 3}, new int[] {2}
+                    })
+                .complement(10))
+        .isInstanceOf(IllegalStateException.class);
+  }
+
+  @Test
+  void testToTopLevelIndexes() {
+    assertThat(Projection.of(new int[] {1, 2, 3, 4}).toTopLevelIndexes())
+        .isEqualTo(new int[] {1, 2, 3, 4});
+
+    assertThat(
+        Projection.of(new int[][] {new int[] {4}, new int[] {1}, new int[] {2}})
+            .toTopLevelIndexes())
+        .isEqualTo(new int[] {4, 1, 2});
+
+    assertThatThrownBy(
+        () ->
+            Projection.of(
+                    new int[][] {
+                        new int[] {4}, new int[] {1, 3}, new int[] {2}
+                    })
+                .toTopLevelIndexes())
+        .isInstanceOf(IllegalStateException.class);
+  }
+
+  @Test
+  void testToNestedIndexes() {
+    assertThat(Projection.of(new int[] {1, 2, 3, 4}).toNestedIndexes())
+        .isEqualTo(
+            new int[][] {new int[] {1}, new int[] {2}, new int[] {3}, new int[] {4}});
+    assertThat(
+        Projection.of(new int[][] {new int[] {4}, new int[] {1, 3}, new int[] {2}})
+            .toNestedIndexes())
+        .isEqualTo(new int[][] {new int[] {4}, new int[] {1, 3}, new int[] {2}});
+  }
+
+  @Test
+  void testEquals() {
+    assertThat(Projection.of(new int[][] {new int[] {1}, new int[] {2}, new int[] {3}}))
+        .isEqualTo(Projection.of(new int[] {1, 2, 3}));
+  }
+}
diff --git a/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java b/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java
index c0822ceac..80c0dbc40 100644
--- a/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java
+++ b/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java
@@ -29,22 +29,15 @@ import com.netease.arctic.hive.table.KeyedHiveTable;
 import com.netease.arctic.hive.table.UnkeyedHiveTable;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.TableIdentifier;
-import com.netease.arctic.table.TableProperties;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.Collectors;
-import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.hive.metastore.api.Database;
 import org.apache.hadoop.hive.metastore.api.Partition;
-import org.apache.hadoop.hive.metastore.api.Table;
-import org.apache.iceberg.DataFile;
-import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.relocated.com.google.common.base.Joiner;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.thrift.TException;
 import org.apache.iceberg.Schema;
-import org.apache.iceberg.hive.TestHiveMetastore;
 import org.apache.iceberg.types.Types;
 import org.junit.After;
 import org.junit.AfterClass;
@@ -66,9 +59,6 @@ public class HiveTableTestBase extends TableTestBase {
   protected static final TemporaryFolder tempFolder = new TemporaryFolder();
 
   protected static HMSMockServer hms  ;
- // protected static TestHiveMetastore metastore;
-  //protected static HiveConf hiveConf;
-  // protected static HiveMetaStoreClient metastoreClient;
 
   protected static final TableIdentifier HIVE_TABLE_ID =
       TableIdentifier.of(HIVE_CATALOG_NAME, HIVE_DB_NAME, "test_hive_table");
@@ -142,13 +132,11 @@ public class HiveTableTestBase extends TableTestBase {
 
     testHiveTable = (UnkeyedHiveTable) hiveCatalog
         .newTableBuilder(HIVE_TABLE_ID, HIVE_TABLE_SCHEMA)
-        .withProperty(TableProperties.LOCATION, tableDir.getPath() + "/table")
         .withPartitionSpec(HIVE_SPEC)
         .create().asUnkeyedTable();
 
     testKeyedHiveTable = (KeyedHiveTable) hiveCatalog
         .newTableBuilder(HIVE_PK_TABLE_ID, HIVE_TABLE_SCHEMA)
-        .withProperty(TableProperties.LOCATION, tableDir.getPath() + "/pk_table")
         .withPartitionSpec(HIVE_SPEC)
         .withPrimaryKeySpec(PRIMARY_KEY_SPEC)
         .create().asKeyedTable();
@@ -164,38 +152,6 @@ public class HiveTableTestBase extends TableTestBase {
   }
 
 
-
-
-  public static class DataFileBuilder {
-    final TableIdentifier identifier;
-    final Table hiveTable;
-    final ArcticTable table;
-
-    public DataFileBuilder(ArcticTable table) throws TException {
-      identifier = table.id();
-      this.table = table;
-      hiveTable = hms.getClient().getTable(identifier.getDatabase(), identifier.getTableName());
-    }
-
-    public DataFile build(String valuePath, String path) {
-      DataFiles.Builder builder =  DataFiles.builder(table.spec())
-          .withPath(hiveTable.getSd().getLocation() + path)
-          .withFileSizeInBytes(0)
-          .withRecordCount(2);
-
-      if (!StringUtils.isEmpty(valuePath)){
-        builder = builder.withPartitionPath(valuePath);
-      }
-      return builder.build();
-    }
-
-    public List<DataFile> buildList(List<Map.Entry<String, String>> partValueFiles){
-      return partValueFiles.stream().map(
-          kv -> this.build(kv.getKey(), kv.getValue())
-      ).collect(Collectors.toList());
-    }
-  }
-
   public static String getPartitionPath(List<String> values, PartitionSpec spec) {
     List<String> nameValues = Lists.newArrayList();
     for (int i = 0; i < values.size(); i++) {
@@ -208,9 +164,6 @@ public class HiveTableTestBase extends TableTestBase {
 
   /**
    * assert hive table partition location as expected
-   * @param partitionLocations
-   * @param table
-   * @throws TException
    */
   public static void assertHivePartitionLocations(Map<String, String> partitionLocations, ArcticTable table) throws TException {
     TableIdentifier identifier = table.id();
diff --git a/hive/src/test/java/com/netease/arctic/hive/MockDataFileBuilder.java b/hive/src/test/java/com/netease/arctic/hive/MockDataFileBuilder.java
new file mode 100644
index 000000000..667ee4152
--- /dev/null
+++ b/hive/src/test/java/com/netease/arctic/hive/MockDataFileBuilder.java
@@ -0,0 +1,56 @@
+package com.netease.arctic.hive;
+
+import com.netease.arctic.hive.table.SupportHive;
+import com.netease.arctic.table.ArcticTable;
+import com.netease.arctic.table.TableIdentifier;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.thrift.TException;
+
+public class MockDataFileBuilder {
+  final TableIdentifier identifier;
+  final Table hiveTable;
+  final ArcticTable table;
+
+  public MockDataFileBuilder(ArcticTable table, HiveMetaStoreClient client) throws TException {
+    identifier = table.id();
+    this.table = table;
+    hiveTable = client.getTable(identifier.getDatabase(), identifier.getTableName());
+  }
+
+  public DataFile build(String valuePath, String path) {
+    String hiveLocation = ((SupportHive) table).hiveLocation();
+
+    String finalPath = hiveLocation + path;
+    DataFiles.Builder builder =  DataFiles.builder(table.spec())
+        .withPath(finalPath)
+        .withFileSizeInBytes(0)
+        .withRecordCount(2);
+
+    if (!StringUtils.isEmpty(valuePath)){
+      builder = builder.withPartitionPath(valuePath);
+    }
+    if (!table.io().exists(finalPath)){
+      // create a temp file for test
+      try {
+        table.io().newOutputFile(finalPath).createOrOverwrite().close();
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
+    }
+    return builder.build();
+  }
+
+  public List<DataFile> buildList(List<Map.Entry<String, String>> partValueFiles){
+    return partValueFiles.stream().map(
+        kv -> this.build(kv.getKey(), kv.getValue())
+    ).collect(Collectors.toList());
+  }
+}
diff --git a/hive/src/test/java/com/netease/arctic/hive/op/TestOverwriteFiles.java b/hive/src/test/java/com/netease/arctic/hive/op/TestOverwriteFiles.java
index 5688e9cf8..ea11aada5 100644
--- a/hive/src/test/java/com/netease/arctic/hive/op/TestOverwriteFiles.java
+++ b/hive/src/test/java/com/netease/arctic/hive/op/TestOverwriteFiles.java
@@ -1,6 +1,8 @@
 package com.netease.arctic.hive.op;
 
 import com.netease.arctic.hive.HiveTableTestBase;
+import com.netease.arctic.hive.MockDataFileBuilder;
+import com.netease.arctic.hive.exceptions.CannotAlterHiveLocationException;
 import com.netease.arctic.hive.table.UnkeyedHiveTable;
 import com.netease.arctic.op.OverwriteBaseFiles;
 import com.netease.arctic.table.KeyedTable;
@@ -36,7 +38,7 @@ public class TestOverwriteFiles extends HiveTableTestBase {
         Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a2.parquet"),
         Maps.immutableEntry("name=bbb", "/test_path/partition2/data-a2.parquet")
     );
-    DataFileBuilder dataFileBuilder = new DataFileBuilder(table);
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
     List<DataFile> dataFiles = dataFileBuilder.buildList(files);
 
     OverwriteFiles overwriteFiles = table.newOverwrite();
@@ -62,7 +64,6 @@ public class TestOverwriteFiles extends HiveTableTestBase {
     assertHivePartitionLocations(partitionAndLocations, table);
   }
 
-
   @Test
   public void testOverwriteKeyedPartitionTable() throws TException {
     KeyedTable table = testKeyedHiveTable;
@@ -73,7 +74,7 @@ public class TestOverwriteFiles extends HiveTableTestBase {
         Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a2.parquet"),
         Maps.immutableEntry("name=bbb", "/test_path/partition2/data-a2.parquet")
     );
-    DataFileBuilder dataFileBuilder = new DataFileBuilder(table);
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
     List<DataFile> dataFiles = dataFileBuilder.buildList(files);
 
     OverwriteBaseFiles overwriteBaseFiles = table.newOverwriteBaseFiles();
@@ -102,7 +103,6 @@ public class TestOverwriteFiles extends HiveTableTestBase {
     assertHivePartitionLocations(partitionAndLocations, table);
   }
 
-
   @Test
   public void testOverwriteOperationTransaction() throws TException {
     UnkeyedTable table = testHiveTable;
@@ -113,7 +113,7 @@ public class TestOverwriteFiles extends HiveTableTestBase {
         Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a2.parquet"),
         Maps.immutableEntry("name=bbb", "/test_path/partition2/data-a2.parquet")
     );
-    DataFileBuilder dataFileBuilder = new DataFileBuilder(table);
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
     List<DataFile> dataFiles = dataFileBuilder.buildList(files);
 
     Transaction tx = table.newTransaction();
@@ -154,7 +154,7 @@ public class TestOverwriteFiles extends HiveTableTestBase {
         Maps.immutableEntry(null, "/test_path/hive_data_location/data-a2.parquet"),
         Maps.immutableEntry(null, "/test_path/hive_data_location/data-a2.parquet")
     );
-    DataFileBuilder dataFileBuilder = new DataFileBuilder(table);
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
     List<DataFile> dataFiles = dataFileBuilder.buildList(files);
 
     OverwriteFiles overwriteFiles = table.newOverwrite();
@@ -164,7 +164,8 @@ public class TestOverwriteFiles extends HiveTableTestBase {
     // assert no hive partition
     assertHivePartitionLocations(partitionAndLocations, table);
     Table hiveTable = hms.getClient().getTable(table.id().getDatabase(), table.name());
-    Assert.assertTrue("table location to new path",
+    Assert.assertTrue(
+        "table location to new path",
         hiveTable.getSd().getLocation().endsWith("/test_path/hive_data_location"));
 
     // =================== test delete all files and add no file to un-partitioned table ===================
@@ -174,11 +175,13 @@ public class TestOverwriteFiles extends HiveTableTestBase {
 
     assertHivePartitionLocations(partitionAndLocations, table);
     hiveTable = hms.getClient().getTable(table.id().getDatabase(), table.name());
-    Assert.assertFalse("table location to new path",
+    Assert.assertFalse(
+        "table location to new path",
         hiveTable.getSd().getLocation().endsWith("/test_path/hive_data_location"));
 
     String hiveLocation = hiveTable.getSd().getLocation();
-    Assert.assertTrue("table location change to hive location",
+    Assert.assertTrue(
+        "table location change to hive location",
         hiveLocation.startsWith(table.hiveLocation()));
     System.out.println(hiveLocation);
 
@@ -196,7 +199,7 @@ public class TestOverwriteFiles extends HiveTableTestBase {
         Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a2.parquet"),
         Maps.immutableEntry("name=bbb", "/test_path/partition2/data-a2.parquet")
     );
-    DataFileBuilder dataFileBuilder = new DataFileBuilder(table);
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
     List<DataFile> dataFiles = dataFileBuilder.buildList(files);
 
     OverwriteFiles overwriteFiles = table.newOverwrite();
@@ -221,11 +224,156 @@ public class TestOverwriteFiles extends HiveTableTestBase {
     assertHivePartitionLocations(partitionAndLocations, table);
   }
 
+  /**
+   * failed when not delete all files in partition
+   */
+  @Test
+  public void testExceptionNotDeleteAllPartitionFiles() throws TException {
+    UnkeyedTable table = testHiveTable;
+    Map<String, String> partitionAndLocations = Maps.newHashMap();
+
+    List<Map.Entry<String, String>> files = Lists.newArrayList(
+        Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a1.parquet"),
+        Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a2.parquet"),
+        Maps.immutableEntry("name=bbb", "/test_path/partition2/data-a2.parquet")
+    );
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
+    List<DataFile> dataFiles = dataFileBuilder.buildList(files);
+
+    OverwriteFiles overwriteFiles = table.newOverwrite();
+    dataFiles.forEach(overwriteFiles::addFile);
+    overwriteFiles.commit();
+
+    applyOverwrite(partitionAndLocations, s -> false, files);
+    assertHivePartitionLocations(partitionAndLocations, table);
+
+    OverwriteFiles newOverwrite = table.newOverwrite();
+    // delete one file under partition[name=aaa]
+    dataFiles.stream()
+        .filter(d -> d.path().toString().endsWith("/partition1/data-a1.parquet"))
+        .forEach(newOverwrite::deleteFile);
+
+    files = Lists.newArrayList(
+        Maps.immutableEntry("name=ddd", "/test_path/partition3/data-a3.parquet"),
+        Maps.immutableEntry("name=ccc", "/test_path/partition4/data-c.parquet")
+    );
+    dataFiles = dataFileBuilder.buildList(files);
+    dataFiles.forEach(newOverwrite::addFile);
+    Assert.assertThrows(CannotAlterHiveLocationException.class, newOverwrite::commit);
+  }
+
+  /**
+   * failed when add two file in different location but with same partition key
+   */
+  @Test
+  public void testExceptionAddFileInDifferentLocation() throws TException {
+    UnkeyedTable table = testHiveTable;
+    List<Map.Entry<String, String>> files = Lists.newArrayList(
+        Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a1.parquet"),
+        Maps.immutableEntry("name=aaa", "/test_path/partition2/data-a2.parquet"),
+        Maps.immutableEntry("name=bbb", "/test_path/partition2/data-a2.parquet")
+    );
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
+    List<DataFile> dataFiles = dataFileBuilder.buildList(files);
+
+    OverwriteFiles overwriteFiles = table.newOverwrite();
+    dataFiles.forEach(overwriteFiles::addFile);
+    Assert.assertThrows(CannotAlterHiveLocationException.class, overwriteFiles::commit);
+  }
+
+  /**
+   * failed when add partition exist with different location
+   */
+  @Test
+  public void testExceptionAddFileWithExistPartition() throws TException {
+    UnkeyedTable table = testHiveTable;
+    List<Map.Entry<String, String>> files = Lists.newArrayList(
+        Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a1.parquet"),
+        Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a2.parquet"),
+        Maps.immutableEntry("name=bbb", "/test_path/partition2/data-a2.parquet")
+    );
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
+    List<DataFile> dataFiles = dataFileBuilder.buildList(files);
+    OverwriteFiles overwriteFiles = table.newOverwrite();
+    dataFiles.forEach(overwriteFiles::addFile);
+    overwriteFiles.commit();
+
+    // add file with exist partition[name=aaa], but in different location, can't create new partition
+    OverwriteFiles newOverwrite = table.newOverwrite();
+    files = Lists.newArrayList(
+        Maps.immutableEntry("name=aaa", "/test_path/partition3/data-a3.parquet"),
+        Maps.immutableEntry("name=ccc", "/test_path/partition4/data-c.parquet")
+    );
+    dataFiles = dataFileBuilder.buildList(files);
+    dataFiles.forEach(newOverwrite::addFile);
+    Assert.assertThrows(CannotAlterHiveLocationException.class, newOverwrite::commit);
+  }
+
+  /**
+   * failed when delete and add partition with same location
+   */
+  @Test
+  public void testExceptionDeleteCreateSamePartition() throws TException {
+    UnkeyedTable table = testHiveTable;
+    List<Map.Entry<String, String>> files = Lists.newArrayList(
+        Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a1.parquet"),
+        Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a2.parquet"),
+        Maps.immutableEntry("name=bbb", "/test_path/partition2/data-a2.parquet")
+    );
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
+    List<DataFile> dataFiles = dataFileBuilder.buildList(files);
+    OverwriteFiles overwriteFiles = table.newOverwrite();
+    dataFiles.forEach(overwriteFiles::addFile);
+    overwriteFiles.commit();
+
+    OverwriteFiles newOverwrite = table.newOverwrite();
+    files = Lists.newArrayList(
+        Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a3.parquet"),
+        Maps.immutableEntry("name=ccc", "/test_path/partition4/data-c.parquet")
+    );
+    dataFiles = dataFileBuilder.buildList(files);
+    dataFiles.forEach(newOverwrite::addFile);
+    // delete partition[name=aaa], add partition[name=aaa] in same location
+    newOverwrite.overwriteByRowFilter(Expressions.equal("name", "aaa"));
+    Assert.assertThrows(CannotAlterHiveLocationException.class, newOverwrite::commit);
+  }
+
+  /**
+   * add file to exist partition, overwrite success without create partition
+   */
+  @Test
+  public void testWhenAddFileToExistPartition() throws TException {
+    UnkeyedTable table = testHiveTable;
+    List<Map.Entry<String, String>> files = Lists.newArrayList(
+        Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a1.parquet"),
+        Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a2.parquet"),
+        Maps.immutableEntry("name=bbb", "/test_path/partition2/data-a2.parquet")
+    );
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
+    List<DataFile> dataFiles = dataFileBuilder.buildList(files);
+    OverwriteFiles overwriteFiles = table.newOverwrite();
+    dataFiles.forEach(overwriteFiles::addFile);
+    overwriteFiles.commit();
+
+    OverwriteFiles newOverwrite = table.newOverwrite();
+    files = Lists.newArrayList(
+        Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a3.parquet"),
+        Maps.immutableEntry("name=ccc", "/test_path/partition4/data-c.parquet")
+    );
+    dataFiles = dataFileBuilder.buildList(files);
+    dataFiles.forEach(newOverwrite::addFile);
+    newOverwrite.commit();
+    Map<String, String> partitions = Maps.newHashMap();
+    partitions.put("name=aaa", "/test_path/partition1");
+    partitions.put("name=bbb", "/test_path/partition2");
+    partitions.put("name=ccc", "/test_path/partition4");
+    assertHivePartitionLocations(partitions, table);
+  }
 
   private void applyOverwrite(
       Map<String, String> partitionAndLocations,
       Predicate<String> deleteFunc,
-      List<Map.Entry<String, String>> addFiles){
+      List<Map.Entry<String, String>> addFiles) {
     Set<String> deleteLocations = partitionAndLocations.keySet()
         .stream().filter(deleteFunc).collect(Collectors.toSet());
 
@@ -238,6 +386,5 @@ public class TestOverwriteFiles extends HiveTableTestBase {
           partLocation
       );
     });
-
   }
 }
diff --git a/hive/src/test/java/com/netease/arctic/hive/op/TestRewritePartitions.java b/hive/src/test/java/com/netease/arctic/hive/op/TestRewritePartitions.java
index 043dfd00e..1a8f127cb 100644
--- a/hive/src/test/java/com/netease/arctic/hive/op/TestRewritePartitions.java
+++ b/hive/src/test/java/com/netease/arctic/hive/op/TestRewritePartitions.java
@@ -19,6 +19,8 @@
 package com.netease.arctic.hive.op;
 
 import com.netease.arctic.hive.HiveTableTestBase;
+import com.netease.arctic.hive.MockDataFileBuilder;
+import com.netease.arctic.hive.exceptions.CannotAlterHiveLocationException;
 import com.netease.arctic.op.RewritePartitions;
 import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.UnkeyedTable;
@@ -47,7 +49,7 @@ public class TestRewritePartitions extends HiveTableTestBase {
         Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a2.parquet"),
         Maps.immutableEntry("name=bbb", "/test_path/partition2/data-a2.parquet")
     );
-    DataFileBuilder dataFileBuilder = new DataFileBuilder(table);
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
     List<DataFile> initDataFiles = dataFileBuilder.buildList(overwriteFiles);
 
     ReplacePartitions replacePartitions = table.newReplacePartitions();
@@ -58,8 +60,6 @@ public class TestRewritePartitions extends HiveTableTestBase {
     //assert hive partition equal with expect.
     assertHivePartitionLocations(partitionAndLocations, table);
 
-
-
     // =================== overwrite table ========================
     overwriteFiles = Lists.newArrayList(
         Maps.immutableEntry("name=aaa", "/test_path/partition3/data-a3.parquet"),
@@ -87,7 +87,7 @@ public class TestRewritePartitions extends HiveTableTestBase {
         Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a2.parquet"),
         Maps.immutableEntry("name=bbb", "/test_path/partition2/data-a2.parquet")
     );
-    DataFileBuilder dataFileBuilder = new DataFileBuilder(table);
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
     List<DataFile> initDataFiles = dataFileBuilder.buildList(overwriteFiles);
 
     RewritePartitions rewritePartitions = table.newRewritePartitions();
@@ -99,7 +99,6 @@ public class TestRewritePartitions extends HiveTableTestBase {
     //assert hive partition equal with expect.
     assertHivePartitionLocations(partitionAndLocations, table);
 
-
     // =================== overwrite table ========================
     overwriteFiles = Lists.newArrayList(
         Maps.immutableEntry("name=aaa", "/test_path/partition3/data-a3.parquet"),
@@ -115,7 +114,6 @@ public class TestRewritePartitions extends HiveTableTestBase {
     applyRewritePartitions(partitionAndLocations, overwriteFiles);
     //assert hive partition equal with expect.
     assertHivePartitionLocations(partitionAndLocations, table);
-
   }
 
   @Test
@@ -128,7 +126,7 @@ public class TestRewritePartitions extends HiveTableTestBase {
         Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a2.parquet"),
         Maps.immutableEntry("name=bbb", "/test_path/partition2/data-a2.parquet")
     );
-    DataFileBuilder dataFileBuilder = new DataFileBuilder(table);
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
     List<DataFile> initDataFiles = dataFileBuilder.buildList(overwriteFiles);
 
     Transaction tx = table.newTransaction();
@@ -148,24 +146,39 @@ public class TestRewritePartitions extends HiveTableTestBase {
     tx.commitTransaction();
     applyRewritePartitions(partitionAndLocations, overwriteFiles);
 
-
     assertHivePartitionLocations(partitionAndLocations, table);
     Assert.assertTrue(table.properties().containsKey("test-rewrite-partition"));
     Assert.assertEquals("test-rewrite-partition-value", table.properties().get("test-rewrite-partition"));
-
   }
 
+  /**
+   * failed then add file of same partition with different location
+   */
+  @Test
+  public void testExceptionAddFileWithDifferentLocation() throws TException {
+    UnkeyedTable table = testHiveTable;
 
+    List<Map.Entry<String, String>> overwriteFiles = Lists.newArrayList(
+        Maps.immutableEntry("name=aaa", "/test_path/partition1/data-a1.parquet"),
+        Maps.immutableEntry("name=aaa", "/test_path/partition2/data-a2.parquet"),
+        Maps.immutableEntry("name=bbb", "/test_path/partition2/data-a2.parquet")
+    );
+    MockDataFileBuilder dataFileBuilder = new MockDataFileBuilder(table, hms.getClient());
+    List<DataFile> initDataFiles = dataFileBuilder.buildList(overwriteFiles);
+
+    ReplacePartitions replacePartitions = table.newReplacePartitions();
+    initDataFiles.forEach(replacePartitions::addFile);
+
+    Assert.assertThrows(CannotAlterHiveLocationException.class, replacePartitions::commit);
+  }
 
 
-  private void applyRewritePartitions(Map<String, String> partitionLocations,
-      List<Map.Entry<String, String>> overwriteFiles){
+  private void applyRewritePartitions(
+      Map<String, String> partitionLocations,
+      List<Map.Entry<String, String>> overwriteFiles) {
     overwriteFiles.forEach(kv -> {
       String partLocation = FileUtil.getFileDir(kv.getValue());
       partitionLocations.put(kv.getKey(), partLocation);
     });
   }
-
-
-
 }
diff --git a/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java b/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
index 9664a04d6..910b2cf01 100644
--- a/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
+++ b/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
@@ -85,17 +85,6 @@ public class SparkTestContext extends ExternalResource {
   protected static String catalogName;
   protected List<Object[]> rows;
 
-  private static SparkTestContext sparkTestContext;
-
-  private static int refCount = 0;
-
-  public static SparkTestContext getSparkTestContext () {
-    if (refCount == 0) {
-      sparkTestContext = new SparkTestContext();
-    }
-    return sparkTestContext;
-  }
-
   public static
   void cleanUpAdditionSparkConfigs() {
     additionSparkConfigs.clear();
diff --git a/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/hive/HMSMockServer.java b/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/hive/HMSMockServer.java
deleted file mode 100644
index 6ba955ba5..000000000
--- a/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/hive/HMSMockServer.java
+++ /dev/null
@@ -1,267 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.netease.arctic.spark.hive;
-
-import org.apache.commons.io.FileUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.metastore.*;
-import org.apache.iceberg.common.DynConstructors;
-import org.apache.iceberg.common.DynMethods;
-import org.apache.iceberg.hadoop.Util;
-import org.apache.iceberg.hive.HiveClientPool;
-import org.apache.thrift.protocol.TBinaryProtocol;
-import org.apache.thrift.server.TServer;
-import org.apache.thrift.server.TThreadPoolServer;
-import org.apache.thrift.transport.TServerSocket;
-import org.apache.thrift.transport.TTransportFactory;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.File;
-import java.util.Random;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
-public class HMSMockServer {
-
-  public static final Logger LOG = LoggerFactory.getLogger(HMSMockServer.class);
-
-  private static final String DEFAULT_DATABASE_NAME = "default";
-  private static final int DEFAULT_POOL_SIZE = 5;
-
-
-
-  // create the metastore handlers based on whether we're working with Hive2 or Hive3 dependencies
-  // we need to do this because there is a breaking API change between Hive2 and Hive3
-  private static final DynConstructors.Ctor<HiveMetaStore.HMSHandler> HMS_HANDLER_CTOR = DynConstructors.builder()
-      .impl(HiveMetaStore.HMSHandler.class, String.class, Configuration.class)
-      .impl(HiveMetaStore.HMSHandler.class, String.class, HiveConf.class)
-      .build();
-
-  private static final DynMethods.StaticMethod GET_BASE_HMS_HANDLER = DynMethods.builder("getProxy")
-      .impl(RetryingHMSHandler.class, Configuration.class, IHMSHandler.class, boolean.class)
-      .impl(RetryingHMSHandler.class, HiveConf.class, IHMSHandler.class, boolean.class)
-      .buildStatic();
-
-  // Hive3 introduces background metastore tasks (MetastoreTaskThread) for performing various cleanup duties. These
-  // threads are scheduled and executed in a static thread pool (org.apache.hadoop.com.netease.spark.hive.metastore.ThreadPool).
-  // This thread pool is shut down normally as part of the JVM shutdown hook, but since we're creating and tearing down
-  // multiple metastore instances within the same JVM, we have to call this cleanup method manually, otherwise
-  // threads from our previous test suite will be stuck in the pool with stale config, and keep on being scheduled.
-  // This can lead to issues, e.g. accidental Persistence Manager closure by ScheduledQueryExecutionsMaintTask.
-  private static final DynMethods.StaticMethod METASTORE_THREADS_SHUTDOWN = DynMethods.builder("shutdown")
-      .impl("org.apache.hadoop.com.netease.spark.hive.metastore.ThreadPool")
-      .orNoop()
-      .buildStatic();
-
-  private File hiveLocalDir;
-  private HiveConf hiveConf;
-  private ExecutorService executorService;
-  private TServer server;
-  private HiveMetaStore.HMSHandler baseHandler;
-  private HiveClientPool clientPool;
-  private int port ;
-  private HiveMetaStoreClient client ;
-
-  public HMSMockServer(){
-    this(new File("hive_warehouse"));
-  }
-
-  public HMSMockServer(File file){
-    this.hiveLocalDir = file;
-    // if (file.exists()){
-    //
-    // }
-    int port = new Random().nextInt(4000);
-    this.port = port + 24000;
-    this.hiveConf = newHiveConf(this.port);
-  }
-
-
-  /**
-   * Starts a TestHiveMetastore with the default connection pool size (5).
-   */
-  public void start() {
-    LOG.info("-------------------------------------------------------------------------");
-    LOG.info("    Starting HiveMetastoreServer ");
-    LOG.info("-------------------------------------------------------------------------");
-
-    start(DEFAULT_POOL_SIZE);
-  }
-
-  /**
-   * Starts a TestHiveMetastore with a provided connection pool size.
-   * @param poolSize The number of threads in the executor pool
-   */
-  public void start(int poolSize) {
-    try {
-      LOG.info("com.netease.spark.hive local dir: " + hiveLocalDir.getAbsolutePath());
-      FileUtils.deleteQuietly(hiveLocalDir);
-
-      File derbyLogFile = new File(hiveLocalDir, "derby.log");
-      System.setProperty("derby.stream.error.file", derbyLogFile.getAbsolutePath());
-
-
-      TServerSocket socket = new TServerSocket(port);
-      this.server = newThriftServer(socket, poolSize, hiveConf);
-      this.executorService = Executors.newSingleThreadExecutor();
-      this.executorService.submit(() -> server.serve());
-
-      // in Hive3, setting this as a system prop ensures that it will be picked up whenever a new HiveConf is created
-      System.setProperty(HiveConf.ConfVars.METASTOREURIS.varname, hiveConf.getVar(HiveConf.ConfVars.METASTOREURIS));
-
-      this.clientPool = new HiveClientPool(1, hiveConf);
-    } catch (Exception e) {
-      throw new RuntimeException("Cannot start TestHiveMetastore", e);
-    }
-  }
-
-  public int getMetastorePort(){
-    return this.port;
-  }
-
-  public void stop() {
-    LOG.info("-------------------------------------------------------------------------");
-    LOG.info("    Shutdown HiveMetastoreServer ");
-    LOG.info("-------------------------------------------------------------------------");
-
-    if (clientPool != null) {
-      clientPool.close();
-    }
-    if (server != null) {
-      server.stop();
-    }
-    if (executorService != null) {
-      executorService.shutdown();
-    }
-    if (hiveLocalDir != null) {
-      hiveLocalDir.delete();
-    }
-    if (baseHandler != null) {
-      baseHandler.shutdown();
-    }
-    METASTORE_THREADS_SHUTDOWN.invoke();
-
-    LOG.info("-------------------------------------------------------------------------");
-    LOG.info("    HiveMetastoreServer finished");
-    LOG.info("-------------------------------------------------------------------------");
-  }
-
-  public HiveConf hiveConf() {
-    return hiveConf;
-  }
-
-  public HiveClientPool clientPool() {
-    return clientPool;
-  }
-
-  public String getDatabasePath(String dbName) {
-    File dbDir = new File(hiveLocalDir, dbName );
-    return dbDir.getAbsolutePath().replace("\\", "/");
-  }
-
-  public String getTablePath(String db, String table){
-    File dbDir = new File(hiveLocalDir, db);
-    File tableDir = new File(dbDir, table);
-    return tableDir.getAbsolutePath().replace("\\", "/");
-  }
-
-  public void reset() throws Exception {
-    for (String dbName : clientPool.run(HiveMetaStoreClient::getAllDatabases)) {
-      for (String tblName : clientPool.run(client -> client.getAllTables(dbName))) {
-        clientPool.run(client -> {
-          client.dropTable(dbName, tblName, true, true, true);
-          return null;
-        });
-      }
-
-      if (!DEFAULT_DATABASE_NAME.equals(dbName)) {
-        // Drop cascade, functions dropped by cascade
-        clientPool.run(client -> {
-          client.dropDatabase(dbName, true, true, true);
-          return null;
-        });
-      }
-    }
-
-    Path warehouseRoot = new Path(hiveLocalDir.getAbsolutePath());
-    FileSystem fs = Util.getFs(warehouseRoot, hiveConf);
-    for (FileStatus fileStatus : fs.listStatus(warehouseRoot)) {
-      if (!fileStatus.getPath().getName().equals("derby.log") &&
-          !fileStatus.getPath().getName().equals("metastore_db")) {
-        fs.delete(fileStatus.getPath(), true);
-      }
-    }
-  }
-
-  public HiveMetaStoreClient getClient(){
-    if (client == null){
-      try {
-        this.client = new HiveMetaStoreClient(hiveConf);
-      }catch (Exception e){
-        throw new RuntimeException(e);
-      }
-    }
-    return this.client;
-  }
-
-  private TServer newThriftServer(TServerSocket socket, int poolSize, HiveConf conf) throws Exception {
-    HiveConf serverConf = new HiveConf(conf);
-    String derbyPath = getDerbyPath();
-    LOG.info("DerbyPath: " + derbyPath);
-    serverConf.set(HiveConf.ConfVars.METASTORECONNECTURLKEY.varname, "jdbc:derby:" + derbyPath + ";create=true");
-    baseHandler = HMS_HANDLER_CTOR.newInstance("new db based metaserver", serverConf);
-    IHMSHandler handler = GET_BASE_HMS_HANDLER.invoke(serverConf, baseHandler, false);
-
-    TThreadPoolServer.Args args = new TThreadPoolServer.Args(socket)
-        .processor(new TSetIpAddressProcessor<>(handler))
-        .transportFactory(new TTransportFactory())
-        .protocolFactory(new TBinaryProtocol.Factory())
-        .minWorkerThreads(poolSize)
-        .maxWorkerThreads(poolSize);
-
-    return new TThreadPoolServer(args);
-  }
-
-  private HiveConf newHiveConf(int port) {
-    Configuration conf = new Configuration(false);
-    HiveConf newHiveConf = new HiveConf(conf, HMSMockServer.class);
-    newHiveConf.set(HiveConf.ConfVars.METASTOREURIS.varname, "thrift://localhost:" + port);
-    newHiveConf.set(HiveConf.ConfVars.METASTOREWAREHOUSE.varname,
-        "file:///" + hiveLocalDir.getAbsolutePath().replace("\\", "/"));
-    newHiveConf.set(HiveConf.ConfVars.METASTORE_TRY_DIRECT_SQL.varname, "false");
-    newHiveConf.set(HiveConf.ConfVars.METASTORE_DISALLOW_INCOMPATIBLE_COL_TYPE_CHANGES.varname, "false");
-    newHiveConf.set(HiveConf.ConfVars.METASTORE_SCHEMA_VERIFICATION.varname, "false");
-    newHiveConf.setBoolVar(HiveConf.ConfVars.HIVE_IN_TEST, true);
-    newHiveConf.set("datanucleus.schema.autoCreateTables", "true");
-    newHiveConf.set("com.netease.spark.hive.metastore.client.capability.check", "false");
-    newHiveConf.set("iceberg.com.netease.spark.hive.client-pool-size", "2");
-    return newHiveConf;
-  }
-
-  private String getDerbyPath() {
-    File metastoreDB = new File(hiveLocalDir, "metastore_db");
-    return  metastoreDB.getAbsolutePath().replace("\\", "/");
-  }
-
-}
diff --git a/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/hive/SparkHiveTestContext.java b/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/hive/SparkHiveTestContext.java
index e62ac1c8d..9c0590809 100644
--- a/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/hive/SparkHiveTestContext.java
+++ b/spark/v2.3/spark/src/test/java/com/netease/arctic/spark/hive/SparkHiveTestContext.java
@@ -19,6 +19,7 @@
 package com.netease.arctic.spark.hive;
 
 
+import com.netease.arctic.hive.HMSMockServer;
 import com.netease.arctic.spark.SparkTestContext;
 
 import java.io.File;
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 7bb69af8c..2a5c0ea57 100644
--- a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
+++ b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
@@ -27,8 +27,8 @@ import com.netease.arctic.ams.api.TableMeta;
 import com.netease.arctic.catalog.ArcticCatalog;
 import com.netease.arctic.catalog.CatalogLoader;
 import com.netease.arctic.data.ChangeAction;
+import com.netease.arctic.hive.HMSMockServer;
 import com.netease.arctic.io.writer.GenericTaskWriters;
-import com.netease.arctic.spark.hive.HMSMockServer;
 import com.netease.arctic.spark.hive.HiveCatalogMetaTestUtil;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.KeyedTable;
@@ -100,8 +100,6 @@ public class SparkTestContext extends ExternalResource {
   protected static String catalogNameArctic;
   protected static String catalogNameHive;
   protected List<Object[]> rows;
-  private static SparkTestContext sparkTestContext;
-  private static int refCount = 0;
   static final File hmsDir = new File(testBaseDir, "hive");
   protected static final HMSMockServer hms = new HMSMockServer(hmsDir);