Repository: NetEase/arctic
Pull request #201 (2022-08-22 03:13:12)

Feature unkeyed table format v2
To support merge-on-read delete/update operation on unkeyed table，change iceberg fromat to v2.


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/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 {


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();