Repository: NetEase/arctic
Pull request #550 (2022-11-01 07:14:54)

[CORE]Make BaseIcebergReader can read "transaction_id, offset,action"
## Why are the changes needed?
spark need read change table with "_transaction_id, _file_offset,_action" columns.
subtask of #521 
## Brief change log
com.netease.arctic.io.reader.BaseIcebergDataReader can read  "_transaction_id, _file_offset,_action" columns

## How was this patch tested?
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] Run test locally before making a pull request

## Documentation

  - Does this pull request introduces a new feature? (yes / **no**)
  - If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)


shidayang: @hellojinsilei 

Patch:
diff --git a/LICENSE b/LICENSE
index cdf9cdd72..5bf7cf772 100644
--- a/LICENSE
+++ b/LICENSE
@@ -223,6 +223,12 @@ This product includes code from Apache Iceberg.
 * org.apache.iceberg.parquet.ParquetWriter copied and modified to org.apache.iceberg.parquet.AdaptHiveParquetWriter
 * org.apache.iceberg.parquet.PrimitiveWriter copied and modified to org.apache.iceberg.parquet.AdaptHivePrimitiveWriter
 * org.apache.iceberg.parquet.TypeToMessageType copied and modified to org.apache.iceberg.parquet.AdaptHiveTypeToMessageType
+* org.apache.iceberg.parquet.ParquetConversions copied and modified to org.apache.iceberg.parquet.AdaptHiveParquetConversions
+* org.apache.iceberg.parquet.ParquetDictionaryRowGroupFilter copied and modified to org.apache.iceberg.parquet.AdaptHiveParquetDictionaryRowGroupFilter
+* org.apache.iceberg.parquet.ParquetMetricsRowGroupFilter copied and modified to org.apache.iceberg.parquet.AdaptHiveParquetMetricsRowGroupFilter
+* org.apache.iceberg.parquet.ParquetReader copied and modified to org.apache.iceberg.parquet.AdaptHiveParquetReader
+* org.apache.iceberg.parquet.ReadConf copied and modified to org.apache.iceberg.parquet.AdaptHiveReadConf
+
 
 Copyright: 2011-2019 The Apache Software Foundation
 Home page: http://iceberg.apache.org/
diff --git a/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizeManager.java b/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizeManager.java
index 2f3ae2a62..5385d0308 100644
--- a/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizeManager.java
+++ b/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizeManager.java
@@ -7,7 +7,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-08-25")
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2022-09-19")
 public class OptimizeManager {
 
   /**
@@ -24,6 +24,8 @@ public class OptimizeManager {
 
     public void reportOptimizerState(OptimizerStateReport reportData) throws org.apache.thrift.TException;
 
+    public OptimizerDescriptor registerOptimizer(OptimizerRegisterInfo registerInfo) throws org.apache.thrift.TException;
+
   }
 
   public interface AsyncIface {
@@ -36,6 +38,8 @@ public class OptimizeManager {
 
     public void reportOptimizerState(OptimizerStateReport reportData, org.apache.thrift.async.AsyncMethodCallback<Void> resultHandler) throws org.apache.thrift.TException;
 
+    public void registerOptimizer(OptimizerRegisterInfo registerInfo, org.apache.thrift.async.AsyncMethodCallback<OptimizerDescriptor> resultHandler) throws org.apache.thrift.TException;
+
   }
 
   public static class Client extends org.apache.thrift.TServiceClient implements Iface {
@@ -146,6 +150,29 @@ public class OptimizeManager {
       return;
     }
 
+    public OptimizerDescriptor registerOptimizer(OptimizerRegisterInfo registerInfo) throws org.apache.thrift.TException
+    {
+      send_registerOptimizer(registerInfo);
+      return recv_registerOptimizer();
+    }
+
+    public void send_registerOptimizer(OptimizerRegisterInfo registerInfo) throws org.apache.thrift.TException
+    {
+      registerOptimizer_args args = new registerOptimizer_args();
+      args.setRegisterInfo(registerInfo);
+      sendBase("registerOptimizer", args);
+    }
+
+    public OptimizerDescriptor recv_registerOptimizer() throws org.apache.thrift.TException
+    {
+      registerOptimizer_result result = new registerOptimizer_result();
+      receiveBase(result, "registerOptimizer");
+      if (result.isSetSuccess()) {
+        return result.success;
+      }
+      throw new org.apache.thrift.TApplicationException(org.apache.thrift.TApplicationException.MISSING_RESULT, "registerOptimizer failed: unknown result");
+    }
+
   }
   public static class AsyncClient extends org.apache.thrift.async.TAsyncClient implements AsyncIface {
     public static class Factory implements org.apache.thrift.async.TAsyncClientFactory<AsyncClient> {
@@ -298,6 +325,38 @@ public class OptimizeManager {
       }
     }
 
+    public void registerOptimizer(OptimizerRegisterInfo registerInfo, org.apache.thrift.async.AsyncMethodCallback<OptimizerDescriptor> resultHandler) throws org.apache.thrift.TException {
+      checkReady();
+      registerOptimizer_call method_call = new registerOptimizer_call(registerInfo, resultHandler, this, ___protocolFactory, ___transport);
+      this.___currentMethod = method_call;
+      ___manager.call(method_call);
+    }
+
+    public static class registerOptimizer_call extends org.apache.thrift.async.TAsyncMethodCall<OptimizerDescriptor> {
+      private OptimizerRegisterInfo registerInfo;
+      public registerOptimizer_call(OptimizerRegisterInfo registerInfo, org.apache.thrift.async.AsyncMethodCallback<OptimizerDescriptor> resultHandler, org.apache.thrift.async.TAsyncClient client, org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.transport.TNonblockingTransport transport) throws org.apache.thrift.TException {
+        super(client, protocolFactory, transport, resultHandler, false);
+        this.registerInfo = registerInfo;
+      }
+
+      public void write_args(org.apache.thrift.protocol.TProtocol prot) throws org.apache.thrift.TException {
+        prot.writeMessageBegin(new org.apache.thrift.protocol.TMessage("registerOptimizer", org.apache.thrift.protocol.TMessageType.CALL, 0));
+        registerOptimizer_args args = new registerOptimizer_args();
+        args.setRegisterInfo(registerInfo);
+        args.write(prot);
+        prot.writeMessageEnd();
+      }
+
+      public OptimizerDescriptor getResult() throws org.apache.thrift.TException {
+        if (getState() != org.apache.thrift.async.TAsyncMethodCall.State.RESPONSE_READ) {
+          throw new java.lang.IllegalStateException("Method call not finished!");
+        }
+        org.apache.thrift.transport.TMemoryInputTransport memoryTransport = new org.apache.thrift.transport.TMemoryInputTransport(getFrameBuffer().array());
+        org.apache.thrift.protocol.TProtocol prot = client.getProtocolFactory().getProtocol(memoryTransport);
+        return (new Client(prot)).recv_registerOptimizer();
+      }
+    }
+
   }
 
   public static class Processor<I extends Iface> extends org.apache.thrift.TBaseProcessor<I> implements org.apache.thrift.TProcessor {
@@ -315,6 +374,7 @@ public class OptimizeManager {
       processMap.put("pollTask", new pollTask());
       processMap.put("reportOptimizeResult", new reportOptimizeResult());
       processMap.put("reportOptimizerState", new reportOptimizerState());
+      processMap.put("registerOptimizer", new registerOptimizer());
       return processMap;
     }
 
@@ -422,6 +482,31 @@ public class OptimizeManager {
       }
     }
 
+    public static class registerOptimizer<I extends Iface> extends org.apache.thrift.ProcessFunction<I, registerOptimizer_args> {
+      public registerOptimizer() {
+        super("registerOptimizer");
+      }
+
+      public registerOptimizer_args getEmptyArgsInstance() {
+        return new registerOptimizer_args();
+      }
+
+      protected boolean isOneway() {
+        return false;
+      }
+
+      @Override
+      protected boolean rethrowUnhandledExceptions() {
+        return false;
+      }
+
+      public registerOptimizer_result getResult(I iface, registerOptimizer_args args) throws org.apache.thrift.TException {
+        registerOptimizer_result result = new registerOptimizer_result();
+        result.success = iface.registerOptimizer(args.registerInfo);
+        return result;
+      }
+    }
+
   }
 
   public static class AsyncProcessor<I extends AsyncIface> extends org.apache.thrift.TBaseAsyncProcessor<I> {
@@ -439,6 +524,7 @@ public class OptimizeManager {
       processMap.put("pollTask", new pollTask());
       processMap.put("reportOptimizeResult", new reportOptimizeResult());
       processMap.put("reportOptimizerState", new reportOptimizerState());
+      processMap.put("registerOptimizer", new registerOptimizer());
       return processMap;
     }
 
@@ -687,6 +773,67 @@ public class OptimizeManager {
       }
     }
 
+    public static class registerOptimizer<I extends AsyncIface> extends org.apache.thrift.AsyncProcessFunction<I, registerOptimizer_args, OptimizerDescriptor> {
+      public registerOptimizer() {
+        super("registerOptimizer");
+      }
+
+      public registerOptimizer_args getEmptyArgsInstance() {
+        return new registerOptimizer_args();
+      }
+
+      public org.apache.thrift.async.AsyncMethodCallback<OptimizerDescriptor> getResultHandler(final org.apache.thrift.server.AbstractNonblockingServer.AsyncFrameBuffer fb, final int seqid) {
+        final org.apache.thrift.AsyncProcessFunction fcall = this;
+        return new org.apache.thrift.async.AsyncMethodCallback<OptimizerDescriptor>() { 
+          public void onComplete(OptimizerDescriptor o) {
+            registerOptimizer_result result = new registerOptimizer_result();
+            result.success = o;
+            try {
+              fcall.sendResponse(fb, result, org.apache.thrift.protocol.TMessageType.REPLY,seqid);
+            } catch (org.apache.thrift.transport.TTransportException e) {
+              _LOGGER.error("TTransportException writing to internal frame buffer", e);
+              fb.close();
+            } catch (java.lang.Exception e) {
+              _LOGGER.error("Exception writing to internal frame buffer", e);
+              onError(e);
+            }
+          }
+          public void onError(java.lang.Exception e) {
+            byte msgType = org.apache.thrift.protocol.TMessageType.REPLY;
+            org.apache.thrift.TSerializable msg;
+            registerOptimizer_result result = new registerOptimizer_result();
+            if (e instanceof org.apache.thrift.transport.TTransportException) {
+              _LOGGER.error("TTransportException inside handler", e);
+              fb.close();
+              return;
+            } else if (e instanceof org.apache.thrift.TApplicationException) {
+              _LOGGER.error("TApplicationException inside handler", e);
+              msgType = org.apache.thrift.protocol.TMessageType.EXCEPTION;
+              msg = (org.apache.thrift.TApplicationException)e;
+            } else {
+              _LOGGER.error("Exception inside handler", e);
+              msgType = org.apache.thrift.protocol.TMessageType.EXCEPTION;
+              msg = new org.apache.thrift.TApplicationException(org.apache.thrift.TApplicationException.INTERNAL_ERROR, e.getMessage());
+            }
+            try {
+              fcall.sendResponse(fb,msg,msgType,seqid);
+            } catch (java.lang.Exception ex) {
+              _LOGGER.error("Exception writing to internal frame buffer", ex);
+              fb.close();
+            }
+          }
+        };
+      }
+
+      protected boolean isOneway() {
+        return false;
+      }
+
+      public void start(I iface, registerOptimizer_args args, org.apache.thrift.async.AsyncMethodCallback<OptimizerDescriptor> resultHandler) throws org.apache.thrift.TException {
+        iface.registerOptimizer(args.registerInfo,resultHandler);
+      }
+    }
+
   }
 
   public static class ping_args implements org.apache.thrift.TBase<ping_args, ping_args._Fields>, java.io.Serializable, Cloneable, Comparable<ping_args>   {
@@ -3601,4 +3748,748 @@ public class OptimizeManager {
     }
   }
 
+  public static class registerOptimizer_args implements org.apache.thrift.TBase<registerOptimizer_args, registerOptimizer_args._Fields>, java.io.Serializable, Cloneable, Comparable<registerOptimizer_args>   {
+    private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("registerOptimizer_args");
+
+    private static final org.apache.thrift.protocol.TField REGISTER_INFO_FIELD_DESC = new org.apache.thrift.protocol.TField("registerInfo", org.apache.thrift.protocol.TType.STRUCT, (short)1);
+
+    private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new registerOptimizer_argsStandardSchemeFactory();
+    private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new registerOptimizer_argsTupleSchemeFactory();
+
+    public @org.apache.thrift.annotation.Nullable OptimizerRegisterInfo registerInfo; // 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 {
+      REGISTER_INFO((short)1, "registerInfo");
+
+      private static final java.util.Map<java.lang.String, _Fields> byName = new java.util.HashMap<java.lang.String, _Fields>();
+
+      static {
+        for (_Fields field : java.util.EnumSet.allOf(_Fields.class)) {
+          byName.put(field.getFieldName(), field);
+        }
+      }
+
+      /**
+       * Find the _Fields constant that matches fieldId, or null if its not found.
+       */
+      @org.apache.thrift.annotation.Nullable
+      public static _Fields findByThriftId(int fieldId) {
+        switch(fieldId) {
+          case 1: // REGISTER_INFO
+            return REGISTER_INFO;
+          default:
+            return null;
+        }
+      }
+
+      /**
+       * Find the _Fields constant that matches fieldId, throwing an exception
+       * if it is not found.
+       */
+      public static _Fields findByThriftIdOrThrow(int fieldId) {
+        _Fields fields = findByThriftId(fieldId);
+        if (fields == null) throw new java.lang.IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+        return fields;
+      }
+
+      /**
+       * 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) {
+        return byName.get(name);
+      }
+
+      private final short _thriftId;
+      private final java.lang.String _fieldName;
+
+      _Fields(short thriftId, java.lang.String fieldName) {
+        _thriftId = thriftId;
+        _fieldName = fieldName;
+      }
+
+      public short getThriftFieldId() {
+        return _thriftId;
+      }
+
+      public java.lang.String getFieldName() {
+        return _fieldName;
+      }
+    }
+
+    // isset id assignments
+    public static final java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+    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.REGISTER_INFO, new org.apache.thrift.meta_data.FieldMetaData("registerInfo", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+          new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, OptimizerRegisterInfo.class)));
+      metaDataMap = java.util.Collections.unmodifiableMap(tmpMap);
+      org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(registerOptimizer_args.class, metaDataMap);
+    }
+
+    public registerOptimizer_args() {
+    }
+
+    public registerOptimizer_args(
+      OptimizerRegisterInfo registerInfo)
+    {
+      this();
+      this.registerInfo = registerInfo;
+    }
+
+    /**
+     * Performs a deep copy on <i>other</i>.
+     */
+    public registerOptimizer_args(registerOptimizer_args other) {
+      if (other.isSetRegisterInfo()) {
+        this.registerInfo = new OptimizerRegisterInfo(other.registerInfo);
+      }
+    }
+
+    public registerOptimizer_args deepCopy() {
+      return new registerOptimizer_args(this);
+    }
+
+    @Override
+    public void clear() {
+      this.registerInfo = null;
+    }
+
+    @org.apache.thrift.annotation.Nullable
+    public OptimizerRegisterInfo getRegisterInfo() {
+      return this.registerInfo;
+    }
+
+    public registerOptimizer_args setRegisterInfo(@org.apache.thrift.annotation.Nullable OptimizerRegisterInfo registerInfo) {
+      this.registerInfo = registerInfo;
+      return this;
+    }
+
+    public void unsetRegisterInfo() {
+      this.registerInfo = null;
+    }
+
+    /** Returns true if field registerInfo is set (has been assigned a value) and false otherwise */
+    public boolean isSetRegisterInfo() {
+      return this.registerInfo != null;
+    }
+
+    public void setRegisterInfoIsSet(boolean value) {
+      if (!value) {
+        this.registerInfo = null;
+      }
+    }
+
+    public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) {
+      switch (field) {
+      case REGISTER_INFO:
+        if (value == null) {
+          unsetRegisterInfo();
+        } else {
+          setRegisterInfo((OptimizerRegisterInfo)value);
+        }
+        break;
+
+      }
+    }
+
+    @org.apache.thrift.annotation.Nullable
+    public java.lang.Object getFieldValue(_Fields field) {
+      switch (field) {
+      case REGISTER_INFO:
+        return getRegisterInfo();
+
+      }
+      throw new java.lang.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();
+      }
+
+      switch (field) {
+      case REGISTER_INFO:
+        return isSetRegisterInfo();
+      }
+      throw new java.lang.IllegalStateException();
+    }
+
+    @Override
+    public boolean equals(java.lang.Object that) {
+      if (that == null)
+        return false;
+      if (that instanceof registerOptimizer_args)
+        return this.equals((registerOptimizer_args)that);
+      return false;
+    }
+
+    public boolean equals(registerOptimizer_args that) {
+      if (that == null)
+        return false;
+      if (this == that)
+        return true;
+
+      boolean this_present_registerInfo = true && this.isSetRegisterInfo();
+      boolean that_present_registerInfo = true && that.isSetRegisterInfo();
+      if (this_present_registerInfo || that_present_registerInfo) {
+        if (!(this_present_registerInfo && that_present_registerInfo))
+          return false;
+        if (!this.registerInfo.equals(that.registerInfo))
+          return false;
+      }
+
+      return true;
+    }
+
+    @Override
+    public int hashCode() {
+      int hashCode = 1;
+
+      hashCode = hashCode * 8191 + ((isSetRegisterInfo()) ? 131071 : 524287);
+      if (isSetRegisterInfo())
+        hashCode = hashCode * 8191 + registerInfo.hashCode();
+
+      return hashCode;
+    }
+
+    @Override
+    public int compareTo(registerOptimizer_args other) {
+      if (!getClass().equals(other.getClass())) {
+        return getClass().getName().compareTo(other.getClass().getName());
+      }
+
+      int lastComparison = 0;
+
+      lastComparison = java.lang.Boolean.valueOf(isSetRegisterInfo()).compareTo(other.isSetRegisterInfo());
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+      if (isSetRegisterInfo()) {
+        lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.registerInfo, other.registerInfo);
+        if (lastComparison != 0) {
+          return lastComparison;
+        }
+      }
+      return 0;
+    }
+
+    @org.apache.thrift.annotation.Nullable
+    public _Fields fieldForId(int fieldId) {
+      return _Fields.findByThriftId(fieldId);
+    }
+
+    public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+      scheme(iprot).read(iprot, this);
+    }
+
+    public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+      scheme(oprot).write(oprot, this);
+    }
+
+    @Override
+    public java.lang.String toString() {
+      java.lang.StringBuilder sb = new java.lang.StringBuilder("registerOptimizer_args(");
+      boolean first = true;
+
+      sb.append("registerInfo:");
+      if (this.registerInfo == null) {
+        sb.append("null");
+      } else {
+        sb.append(this.registerInfo);
+      }
+      first = false;
+      sb.append(")");
+      return sb.toString();
+    }
+
+    public void validate() throws org.apache.thrift.TException {
+      // check for required fields
+      // check for sub-struct validity
+      if (registerInfo != null) {
+        registerInfo.validate();
+      }
+    }
+
+    private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+      try {
+        write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+      } catch (org.apache.thrift.TException te) {
+        throw new java.io.IOException(te);
+      }
+    }
+
+    private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, java.lang.ClassNotFoundException {
+      try {
+        read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+      } catch (org.apache.thrift.TException te) {
+        throw new java.io.IOException(te);
+      }
+    }
+
+    private static class registerOptimizer_argsStandardSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+      public registerOptimizer_argsStandardScheme getScheme() {
+        return new registerOptimizer_argsStandardScheme();
+      }
+    }
+
+    private static class registerOptimizer_argsStandardScheme extends org.apache.thrift.scheme.StandardScheme<registerOptimizer_args> {
+
+      public void read(org.apache.thrift.protocol.TProtocol iprot, registerOptimizer_args struct) throws org.apache.thrift.TException {
+        org.apache.thrift.protocol.TField schemeField;
+        iprot.readStructBegin();
+        while (true)
+        {
+          schemeField = iprot.readFieldBegin();
+          if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { 
+            break;
+          }
+          switch (schemeField.id) {
+            case 1: // REGISTER_INFO
+              if (schemeField.type == org.apache.thrift.protocol.TType.STRUCT) {
+                struct.registerInfo = new OptimizerRegisterInfo();
+                struct.registerInfo.read(iprot);
+                struct.setRegisterInfoIsSet(true);
+              } else { 
+                org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+              }
+              break;
+            default:
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+          }
+          iprot.readFieldEnd();
+        }
+        iprot.readStructEnd();
+
+        // check for required fields of primitive type, which can't be checked in the validate method
+        struct.validate();
+      }
+
+      public void write(org.apache.thrift.protocol.TProtocol oprot, registerOptimizer_args struct) throws org.apache.thrift.TException {
+        struct.validate();
+
+        oprot.writeStructBegin(STRUCT_DESC);
+        if (struct.registerInfo != null) {
+          oprot.writeFieldBegin(REGISTER_INFO_FIELD_DESC);
+          struct.registerInfo.write(oprot);
+          oprot.writeFieldEnd();
+        }
+        oprot.writeFieldStop();
+        oprot.writeStructEnd();
+      }
+
+    }
+
+    private static class registerOptimizer_argsTupleSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+      public registerOptimizer_argsTupleScheme getScheme() {
+        return new registerOptimizer_argsTupleScheme();
+      }
+    }
+
+    private static class registerOptimizer_argsTupleScheme extends org.apache.thrift.scheme.TupleScheme<registerOptimizer_args> {
+
+      @Override
+      public void write(org.apache.thrift.protocol.TProtocol prot, registerOptimizer_args struct) throws org.apache.thrift.TException {
+        org.apache.thrift.protocol.TTupleProtocol oprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+        java.util.BitSet optionals = new java.util.BitSet();
+        if (struct.isSetRegisterInfo()) {
+          optionals.set(0);
+        }
+        oprot.writeBitSet(optionals, 1);
+        if (struct.isSetRegisterInfo()) {
+          struct.registerInfo.write(oprot);
+        }
+      }
+
+      @Override
+      public void read(org.apache.thrift.protocol.TProtocol prot, registerOptimizer_args struct) throws org.apache.thrift.TException {
+        org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+        java.util.BitSet incoming = iprot.readBitSet(1);
+        if (incoming.get(0)) {
+          struct.registerInfo = new OptimizerRegisterInfo();
+          struct.registerInfo.read(iprot);
+          struct.setRegisterInfoIsSet(true);
+        }
+      }
+    }
+
+    private static <S extends org.apache.thrift.scheme.IScheme> S scheme(org.apache.thrift.protocol.TProtocol proto) {
+      return (org.apache.thrift.scheme.StandardScheme.class.equals(proto.getScheme()) ? STANDARD_SCHEME_FACTORY : TUPLE_SCHEME_FACTORY).getScheme();
+    }
+  }
+
+  public static class registerOptimizer_result implements org.apache.thrift.TBase<registerOptimizer_result, registerOptimizer_result._Fields>, java.io.Serializable, Cloneable, Comparable<registerOptimizer_result>   {
+    private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("registerOptimizer_result");
+
+    private static final org.apache.thrift.protocol.TField SUCCESS_FIELD_DESC = new org.apache.thrift.protocol.TField("success", org.apache.thrift.protocol.TType.STRUCT, (short)0);
+
+    private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new registerOptimizer_resultStandardSchemeFactory();
+    private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new registerOptimizer_resultTupleSchemeFactory();
+
+    public @org.apache.thrift.annotation.Nullable OptimizerDescriptor success; // 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 {
+      SUCCESS((short)0, "success");
+
+      private static final java.util.Map<java.lang.String, _Fields> byName = new java.util.HashMap<java.lang.String, _Fields>();
+
+      static {
+        for (_Fields field : java.util.EnumSet.allOf(_Fields.class)) {
+          byName.put(field.getFieldName(), field);
+        }
+      }
+
+      /**
+       * Find the _Fields constant that matches fieldId, or null if its not found.
+       */
+      @org.apache.thrift.annotation.Nullable
+      public static _Fields findByThriftId(int fieldId) {
+        switch(fieldId) {
+          case 0: // SUCCESS
+            return SUCCESS;
+          default:
+            return null;
+        }
+      }
+
+      /**
+       * Find the _Fields constant that matches fieldId, throwing an exception
+       * if it is not found.
+       */
+      public static _Fields findByThriftIdOrThrow(int fieldId) {
+        _Fields fields = findByThriftId(fieldId);
+        if (fields == null) throw new java.lang.IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+        return fields;
+      }
+
+      /**
+       * 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) {
+        return byName.get(name);
+      }
+
+      private final short _thriftId;
+      private final java.lang.String _fieldName;
+
+      _Fields(short thriftId, java.lang.String fieldName) {
+        _thriftId = thriftId;
+        _fieldName = fieldName;
+      }
+
+      public short getThriftFieldId() {
+        return _thriftId;
+      }
+
+      public java.lang.String getFieldName() {
+        return _fieldName;
+      }
+    }
+
+    // isset id assignments
+    public static final java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+    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.SUCCESS, new org.apache.thrift.meta_data.FieldMetaData("success", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+          new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, OptimizerDescriptor.class)));
+      metaDataMap = java.util.Collections.unmodifiableMap(tmpMap);
+      org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(registerOptimizer_result.class, metaDataMap);
+    }
+
+    public registerOptimizer_result() {
+    }
+
+    public registerOptimizer_result(
+      OptimizerDescriptor success)
+    {
+      this();
+      this.success = success;
+    }
+
+    /**
+     * Performs a deep copy on <i>other</i>.
+     */
+    public registerOptimizer_result(registerOptimizer_result other) {
+      if (other.isSetSuccess()) {
+        this.success = new OptimizerDescriptor(other.success);
+      }
+    }
+
+    public registerOptimizer_result deepCopy() {
+      return new registerOptimizer_result(this);
+    }
+
+    @Override
+    public void clear() {
+      this.success = null;
+    }
+
+    @org.apache.thrift.annotation.Nullable
+    public OptimizerDescriptor getSuccess() {
+      return this.success;
+    }
+
+    public registerOptimizer_result setSuccess(@org.apache.thrift.annotation.Nullable OptimizerDescriptor success) {
+      this.success = success;
+      return this;
+    }
+
+    public void unsetSuccess() {
+      this.success = null;
+    }
+
+    /** Returns true if field success is set (has been assigned a value) and false otherwise */
+    public boolean isSetSuccess() {
+      return this.success != null;
+    }
+
+    public void setSuccessIsSet(boolean value) {
+      if (!value) {
+        this.success = null;
+      }
+    }
+
+    public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) {
+      switch (field) {
+      case SUCCESS:
+        if (value == null) {
+          unsetSuccess();
+        } else {
+          setSuccess((OptimizerDescriptor)value);
+        }
+        break;
+
+      }
+    }
+
+    @org.apache.thrift.annotation.Nullable
+    public java.lang.Object getFieldValue(_Fields field) {
+      switch (field) {
+      case SUCCESS:
+        return getSuccess();
+
+      }
+      throw new java.lang.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();
+      }
+
+      switch (field) {
+      case SUCCESS:
+        return isSetSuccess();
+      }
+      throw new java.lang.IllegalStateException();
+    }
+
+    @Override
+    public boolean equals(java.lang.Object that) {
+      if (that == null)
+        return false;
+      if (that instanceof registerOptimizer_result)
+        return this.equals((registerOptimizer_result)that);
+      return false;
+    }
+
+    public boolean equals(registerOptimizer_result that) {
+      if (that == null)
+        return false;
+      if (this == that)
+        return true;
+
+      boolean this_present_success = true && this.isSetSuccess();
+      boolean that_present_success = true && that.isSetSuccess();
+      if (this_present_success || that_present_success) {
+        if (!(this_present_success && that_present_success))
+          return false;
+        if (!this.success.equals(that.success))
+          return false;
+      }
+
+      return true;
+    }
+
+    @Override
+    public int hashCode() {
+      int hashCode = 1;
+
+      hashCode = hashCode * 8191 + ((isSetSuccess()) ? 131071 : 524287);
+      if (isSetSuccess())
+        hashCode = hashCode * 8191 + success.hashCode();
+
+      return hashCode;
+    }
+
+    @Override
+    public int compareTo(registerOptimizer_result other) {
+      if (!getClass().equals(other.getClass())) {
+        return getClass().getName().compareTo(other.getClass().getName());
+      }
+
+      int lastComparison = 0;
+
+      lastComparison = java.lang.Boolean.valueOf(isSetSuccess()).compareTo(other.isSetSuccess());
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+      if (isSetSuccess()) {
+        lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.success, other.success);
+        if (lastComparison != 0) {
+          return lastComparison;
+        }
+      }
+      return 0;
+    }
+
+    @org.apache.thrift.annotation.Nullable
+    public _Fields fieldForId(int fieldId) {
+      return _Fields.findByThriftId(fieldId);
+    }
+
+    public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+      scheme(iprot).read(iprot, this);
+    }
+
+    public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+      scheme(oprot).write(oprot, this);
+      }
+
+    @Override
+    public java.lang.String toString() {
+      java.lang.StringBuilder sb = new java.lang.StringBuilder("registerOptimizer_result(");
+      boolean first = true;
+
+      sb.append("success:");
+      if (this.success == null) {
+        sb.append("null");
+      } else {
+        sb.append(this.success);
+      }
+      first = false;
+      sb.append(")");
+      return sb.toString();
+    }
+
+    public void validate() throws org.apache.thrift.TException {
+      // check for required fields
+      // check for sub-struct validity
+      if (success != null) {
+        success.validate();
+      }
+    }
+
+    private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+      try {
+        write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+      } catch (org.apache.thrift.TException te) {
+        throw new java.io.IOException(te);
+      }
+    }
+
+    private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, java.lang.ClassNotFoundException {
+      try {
+        read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+      } catch (org.apache.thrift.TException te) {
+        throw new java.io.IOException(te);
+      }
+    }
+
+    private static class registerOptimizer_resultStandardSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+      public registerOptimizer_resultStandardScheme getScheme() {
+        return new registerOptimizer_resultStandardScheme();
+      }
+    }
+
+    private static class registerOptimizer_resultStandardScheme extends org.apache.thrift.scheme.StandardScheme<registerOptimizer_result> {
+
+      public void read(org.apache.thrift.protocol.TProtocol iprot, registerOptimizer_result struct) throws org.apache.thrift.TException {
+        org.apache.thrift.protocol.TField schemeField;
+        iprot.readStructBegin();
+        while (true)
+        {
+          schemeField = iprot.readFieldBegin();
+          if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { 
+            break;
+          }
+          switch (schemeField.id) {
+            case 0: // SUCCESS
+              if (schemeField.type == org.apache.thrift.protocol.TType.STRUCT) {
+                struct.success = new OptimizerDescriptor();
+                struct.success.read(iprot);
+                struct.setSuccessIsSet(true);
+              } else { 
+                org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+              }
+              break;
+            default:
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+          }
+          iprot.readFieldEnd();
+        }
+        iprot.readStructEnd();
+
+        // check for required fields of primitive type, which can't be checked in the validate method
+        struct.validate();
+      }
+
+      public void write(org.apache.thrift.protocol.TProtocol oprot, registerOptimizer_result struct) throws org.apache.thrift.TException {
+        struct.validate();
+
+        oprot.writeStructBegin(STRUCT_DESC);
+        if (struct.success != null) {
+          oprot.writeFieldBegin(SUCCESS_FIELD_DESC);
+          struct.success.write(oprot);
+          oprot.writeFieldEnd();
+        }
+        oprot.writeFieldStop();
+        oprot.writeStructEnd();
+      }
+
+    }
+
+    private static class registerOptimizer_resultTupleSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+      public registerOptimizer_resultTupleScheme getScheme() {
+        return new registerOptimizer_resultTupleScheme();
+      }
+    }
+
+    private static class registerOptimizer_resultTupleScheme extends org.apache.thrift.scheme.TupleScheme<registerOptimizer_result> {
+
+      @Override
+      public void write(org.apache.thrift.protocol.TProtocol prot, registerOptimizer_result struct) throws org.apache.thrift.TException {
+        org.apache.thrift.protocol.TTupleProtocol oprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+        java.util.BitSet optionals = new java.util.BitSet();
+        if (struct.isSetSuccess()) {
+          optionals.set(0);
+        }
+        oprot.writeBitSet(optionals, 1);
+        if (struct.isSetSuccess()) {
+          struct.success.write(oprot);
+        }
+      }
+
+      @Override
+      public void read(org.apache.thrift.protocol.TProtocol prot, registerOptimizer_result struct) throws org.apache.thrift.TException {
+        org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+        java.util.BitSet incoming = iprot.readBitSet(1);
+        if (incoming.get(0)) {
+          struct.success = new OptimizerDescriptor();
+          struct.success.read(iprot);
+          struct.setSuccessIsSet(true);
+        }
+      }
+    }
+
+    private static <S extends org.apache.thrift.scheme.IScheme> S scheme(org.apache.thrift.protocol.TProtocol proto) {
+      return (org.apache.thrift.scheme.StandardScheme.class.equals(proto.getScheme()) ? STANDARD_SCHEME_FACTORY : TUPLE_SCHEME_FACTORY).getScheme();
+    }
+  }
+
 }
diff --git a/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizerDescriptor.java b/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizerDescriptor.java
new file mode 100644
index 000000000..a93027ebf
--- /dev/null
+++ b/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizerDescriptor.java
@@ -0,0 +1,1071 @@
+/**
+ * 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;
+
+@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2022-09-19")
+public class OptimizerDescriptor implements org.apache.thrift.TBase<OptimizerDescriptor, OptimizerDescriptor._Fields>, java.io.Serializable, Cloneable, Comparable<OptimizerDescriptor> {
+  private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("OptimizerDescriptor");
+
+  private static final org.apache.thrift.protocol.TField OPTIMIZER_ID_FIELD_DESC = new org.apache.thrift.protocol.TField("optimizerId", org.apache.thrift.protocol.TType.I32, (short)1);
+  private static final org.apache.thrift.protocol.TField GROUP_ID_FIELD_DESC = new org.apache.thrift.protocol.TField("groupId", org.apache.thrift.protocol.TType.I32, (short)2);
+  private static final org.apache.thrift.protocol.TField GROUP_NAME_FIELD_DESC = new org.apache.thrift.protocol.TField("groupName", org.apache.thrift.protocol.TType.STRING, (short)3);
+  private static final org.apache.thrift.protocol.TField CORE_NUMBER_FIELD_DESC = new org.apache.thrift.protocol.TField("coreNumber", org.apache.thrift.protocol.TType.I32, (short)4);
+  private static final org.apache.thrift.protocol.TField MEMORY_SIZE_FIELD_DESC = new org.apache.thrift.protocol.TField("memorySize", org.apache.thrift.protocol.TType.I64, (short)5);
+  private static final org.apache.thrift.protocol.TField CONTAINER_FIELD_DESC = new org.apache.thrift.protocol.TField("container", org.apache.thrift.protocol.TType.STRING, (short)6);
+  private static final org.apache.thrift.protocol.TField STATUS_FIELD_DESC = new org.apache.thrift.protocol.TField("status", org.apache.thrift.protocol.TType.STRING, (short)7);
+  private static final org.apache.thrift.protocol.TField UPDATE_TIME_FIELD_DESC = new org.apache.thrift.protocol.TField("updateTime", org.apache.thrift.protocol.TType.I64, (short)8);
+
+  private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new OptimizerDescriptorStandardSchemeFactory();
+  private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new OptimizerDescriptorTupleSchemeFactory();
+
+  public int optimizerId; // required
+  public int groupId; // required
+  public @org.apache.thrift.annotation.Nullable java.lang.String groupName; // required
+  public int coreNumber; // required
+  public long memorySize; // required
+  public @org.apache.thrift.annotation.Nullable java.lang.String container; // required
+  public @org.apache.thrift.annotation.Nullable java.lang.String status; // required
+  public long updateTime; // 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 {
+    OPTIMIZER_ID((short)1, "optimizerId"),
+    GROUP_ID((short)2, "groupId"),
+    GROUP_NAME((short)3, "groupName"),
+    CORE_NUMBER((short)4, "coreNumber"),
+    MEMORY_SIZE((short)5, "memorySize"),
+    CONTAINER((short)6, "container"),
+    STATUS((short)7, "status"),
+    UPDATE_TIME((short)8, "updateTime");
+
+    private static final java.util.Map<java.lang.String, _Fields> byName = new java.util.HashMap<java.lang.String, _Fields>();
+
+    static {
+      for (_Fields field : java.util.EnumSet.allOf(_Fields.class)) {
+        byName.put(field.getFieldName(), field);
+      }
+    }
+
+    /**
+     * Find the _Fields constant that matches fieldId, or null if its not found.
+     */
+    @org.apache.thrift.annotation.Nullable
+    public static _Fields findByThriftId(int fieldId) {
+      switch(fieldId) {
+        case 1: // OPTIMIZER_ID
+          return OPTIMIZER_ID;
+        case 2: // GROUP_ID
+          return GROUP_ID;
+        case 3: // GROUP_NAME
+          return GROUP_NAME;
+        case 4: // CORE_NUMBER
+          return CORE_NUMBER;
+        case 5: // MEMORY_SIZE
+          return MEMORY_SIZE;
+        case 6: // CONTAINER
+          return CONTAINER;
+        case 7: // STATUS
+          return STATUS;
+        case 8: // UPDATE_TIME
+          return UPDATE_TIME;
+        default:
+          return null;
+      }
+    }
+
+    /**
+     * Find the _Fields constant that matches fieldId, throwing an exception
+     * if it is not found.
+     */
+    public static _Fields findByThriftIdOrThrow(int fieldId) {
+      _Fields fields = findByThriftId(fieldId);
+      if (fields == null) throw new java.lang.IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+      return fields;
+    }
+
+    /**
+     * 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) {
+      return byName.get(name);
+    }
+
+    private final short _thriftId;
+    private final java.lang.String _fieldName;
+
+    _Fields(short thriftId, java.lang.String fieldName) {
+      _thriftId = thriftId;
+      _fieldName = fieldName;
+    }
+
+    public short getThriftFieldId() {
+      return _thriftId;
+    }
+
+    public java.lang.String getFieldName() {
+      return _fieldName;
+    }
+  }
+
+  // isset id assignments
+  private static final int __OPTIMIZERID_ISSET_ID = 0;
+  private static final int __GROUPID_ISSET_ID = 1;
+  private static final int __CORENUMBER_ISSET_ID = 2;
+  private static final int __MEMORYSIZE_ISSET_ID = 3;
+  private static final int __UPDATETIME_ISSET_ID = 4;
+  private byte __isset_bitfield = 0;
+  public static final java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+  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.OPTIMIZER_ID, new org.apache.thrift.meta_data.FieldMetaData("optimizerId", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I32)));
+    tmpMap.put(_Fields.GROUP_ID, new org.apache.thrift.meta_data.FieldMetaData("groupId", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I32)));
+    tmpMap.put(_Fields.GROUP_NAME, new org.apache.thrift.meta_data.FieldMetaData("groupName", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
+    tmpMap.put(_Fields.CORE_NUMBER, new org.apache.thrift.meta_data.FieldMetaData("coreNumber", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I32)));
+    tmpMap.put(_Fields.MEMORY_SIZE, new org.apache.thrift.meta_data.FieldMetaData("memorySize", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I64)));
+    tmpMap.put(_Fields.CONTAINER, new org.apache.thrift.meta_data.FieldMetaData("container", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
+    tmpMap.put(_Fields.STATUS, new org.apache.thrift.meta_data.FieldMetaData("status", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
+    tmpMap.put(_Fields.UPDATE_TIME, new org.apache.thrift.meta_data.FieldMetaData("updateTime", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I64)));
+    metaDataMap = java.util.Collections.unmodifiableMap(tmpMap);
+    org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(OptimizerDescriptor.class, metaDataMap);
+  }
+
+  public OptimizerDescriptor() {
+  }
+
+  public OptimizerDescriptor(
+    int optimizerId,
+    int groupId,
+    java.lang.String groupName,
+    int coreNumber,
+    long memorySize,
+    java.lang.String container,
+    java.lang.String status,
+    long updateTime)
+  {
+    this();
+    this.optimizerId = optimizerId;
+    setOptimizerIdIsSet(true);
+    this.groupId = groupId;
+    setGroupIdIsSet(true);
+    this.groupName = groupName;
+    this.coreNumber = coreNumber;
+    setCoreNumberIsSet(true);
+    this.memorySize = memorySize;
+    setMemorySizeIsSet(true);
+    this.container = container;
+    this.status = status;
+    this.updateTime = updateTime;
+    setUpdateTimeIsSet(true);
+  }
+
+  /**
+   * Performs a deep copy on <i>other</i>.
+   */
+  public OptimizerDescriptor(OptimizerDescriptor other) {
+    __isset_bitfield = other.__isset_bitfield;
+    this.optimizerId = other.optimizerId;
+    this.groupId = other.groupId;
+    if (other.isSetGroupName()) {
+      this.groupName = other.groupName;
+    }
+    this.coreNumber = other.coreNumber;
+    this.memorySize = other.memorySize;
+    if (other.isSetContainer()) {
+      this.container = other.container;
+    }
+    if (other.isSetStatus()) {
+      this.status = other.status;
+    }
+    this.updateTime = other.updateTime;
+  }
+
+  public OptimizerDescriptor deepCopy() {
+    return new OptimizerDescriptor(this);
+  }
+
+  @Override
+  public void clear() {
+    setOptimizerIdIsSet(false);
+    this.optimizerId = 0;
+    setGroupIdIsSet(false);
+    this.groupId = 0;
+    this.groupName = null;
+    setCoreNumberIsSet(false);
+    this.coreNumber = 0;
+    setMemorySizeIsSet(false);
+    this.memorySize = 0;
+    this.container = null;
+    this.status = null;
+    setUpdateTimeIsSet(false);
+    this.updateTime = 0;
+  }
+
+  public int getOptimizerId() {
+    return this.optimizerId;
+  }
+
+  public OptimizerDescriptor setOptimizerId(int optimizerId) {
+    this.optimizerId = optimizerId;
+    setOptimizerIdIsSet(true);
+    return this;
+  }
+
+  public void unsetOptimizerId() {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.clearBit(__isset_bitfield, __OPTIMIZERID_ISSET_ID);
+  }
+
+  /** Returns true if field optimizerId is set (has been assigned a value) and false otherwise */
+  public boolean isSetOptimizerId() {
+    return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield, __OPTIMIZERID_ISSET_ID);
+  }
+
+  public void setOptimizerIdIsSet(boolean value) {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.setBit(__isset_bitfield, __OPTIMIZERID_ISSET_ID, value);
+  }
+
+  public int getGroupId() {
+    return this.groupId;
+  }
+
+  public OptimizerDescriptor setGroupId(int groupId) {
+    this.groupId = groupId;
+    setGroupIdIsSet(true);
+    return this;
+  }
+
+  public void unsetGroupId() {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.clearBit(__isset_bitfield, __GROUPID_ISSET_ID);
+  }
+
+  /** Returns true if field groupId is set (has been assigned a value) and false otherwise */
+  public boolean isSetGroupId() {
+    return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield, __GROUPID_ISSET_ID);
+  }
+
+  public void setGroupIdIsSet(boolean value) {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.setBit(__isset_bitfield, __GROUPID_ISSET_ID, value);
+  }
+
+  @org.apache.thrift.annotation.Nullable
+  public java.lang.String getGroupName() {
+    return this.groupName;
+  }
+
+  public OptimizerDescriptor setGroupName(@org.apache.thrift.annotation.Nullable java.lang.String groupName) {
+    this.groupName = groupName;
+    return this;
+  }
+
+  public void unsetGroupName() {
+    this.groupName = null;
+  }
+
+  /** Returns true if field groupName is set (has been assigned a value) and false otherwise */
+  public boolean isSetGroupName() {
+    return this.groupName != null;
+  }
+
+  public void setGroupNameIsSet(boolean value) {
+    if (!value) {
+      this.groupName = null;
+    }
+  }
+
+  public int getCoreNumber() {
+    return this.coreNumber;
+  }
+
+  public OptimizerDescriptor setCoreNumber(int coreNumber) {
+    this.coreNumber = coreNumber;
+    setCoreNumberIsSet(true);
+    return this;
+  }
+
+  public void unsetCoreNumber() {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.clearBit(__isset_bitfield, __CORENUMBER_ISSET_ID);
+  }
+
+  /** Returns true if field coreNumber is set (has been assigned a value) and false otherwise */
+  public boolean isSetCoreNumber() {
+    return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield, __CORENUMBER_ISSET_ID);
+  }
+
+  public void setCoreNumberIsSet(boolean value) {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.setBit(__isset_bitfield, __CORENUMBER_ISSET_ID, value);
+  }
+
+  public long getMemorySize() {
+    return this.memorySize;
+  }
+
+  public OptimizerDescriptor setMemorySize(long memorySize) {
+    this.memorySize = memorySize;
+    setMemorySizeIsSet(true);
+    return this;
+  }
+
+  public void unsetMemorySize() {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.clearBit(__isset_bitfield, __MEMORYSIZE_ISSET_ID);
+  }
+
+  /** Returns true if field memorySize is set (has been assigned a value) and false otherwise */
+  public boolean isSetMemorySize() {
+    return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield, __MEMORYSIZE_ISSET_ID);
+  }
+
+  public void setMemorySizeIsSet(boolean value) {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.setBit(__isset_bitfield, __MEMORYSIZE_ISSET_ID, value);
+  }
+
+  @org.apache.thrift.annotation.Nullable
+  public java.lang.String getContainer() {
+    return this.container;
+  }
+
+  public OptimizerDescriptor setContainer(@org.apache.thrift.annotation.Nullable java.lang.String container) {
+    this.container = container;
+    return this;
+  }
+
+  public void unsetContainer() {
+    this.container = null;
+  }
+
+  /** Returns true if field container is set (has been assigned a value) and false otherwise */
+  public boolean isSetContainer() {
+    return this.container != null;
+  }
+
+  public void setContainerIsSet(boolean value) {
+    if (!value) {
+      this.container = null;
+    }
+  }
+
+  @org.apache.thrift.annotation.Nullable
+  public java.lang.String getStatus() {
+    return this.status;
+  }
+
+  public OptimizerDescriptor setStatus(@org.apache.thrift.annotation.Nullable java.lang.String status) {
+    this.status = status;
+    return this;
+  }
+
+  public void unsetStatus() {
+    this.status = null;
+  }
+
+  /** Returns true if field status is set (has been assigned a value) and false otherwise */
+  public boolean isSetStatus() {
+    return this.status != null;
+  }
+
+  public void setStatusIsSet(boolean value) {
+    if (!value) {
+      this.status = null;
+    }
+  }
+
+  public long getUpdateTime() {
+    return this.updateTime;
+  }
+
+  public OptimizerDescriptor setUpdateTime(long updateTime) {
+    this.updateTime = updateTime;
+    setUpdateTimeIsSet(true);
+    return this;
+  }
+
+  public void unsetUpdateTime() {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.clearBit(__isset_bitfield, __UPDATETIME_ISSET_ID);
+  }
+
+  /** Returns true if field updateTime is set (has been assigned a value) and false otherwise */
+  public boolean isSetUpdateTime() {
+    return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield, __UPDATETIME_ISSET_ID);
+  }
+
+  public void setUpdateTimeIsSet(boolean value) {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.setBit(__isset_bitfield, __UPDATETIME_ISSET_ID, value);
+  }
+
+  public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) {
+    switch (field) {
+    case OPTIMIZER_ID:
+      if (value == null) {
+        unsetOptimizerId();
+      } else {
+        setOptimizerId((java.lang.Integer)value);
+      }
+      break;
+
+    case GROUP_ID:
+      if (value == null) {
+        unsetGroupId();
+      } else {
+        setGroupId((java.lang.Integer)value);
+      }
+      break;
+
+    case GROUP_NAME:
+      if (value == null) {
+        unsetGroupName();
+      } else {
+        setGroupName((java.lang.String)value);
+      }
+      break;
+
+    case CORE_NUMBER:
+      if (value == null) {
+        unsetCoreNumber();
+      } else {
+        setCoreNumber((java.lang.Integer)value);
+      }
+      break;
+
+    case MEMORY_SIZE:
+      if (value == null) {
+        unsetMemorySize();
+      } else {
+        setMemorySize((java.lang.Long)value);
+      }
+      break;
+
+    case CONTAINER:
+      if (value == null) {
+        unsetContainer();
+      } else {
+        setContainer((java.lang.String)value);
+      }
+      break;
+
+    case STATUS:
+      if (value == null) {
+        unsetStatus();
+      } else {
+        setStatus((java.lang.String)value);
+      }
+      break;
+
+    case UPDATE_TIME:
+      if (value == null) {
+        unsetUpdateTime();
+      } else {
+        setUpdateTime((java.lang.Long)value);
+      }
+      break;
+
+    }
+  }
+
+  @org.apache.thrift.annotation.Nullable
+  public java.lang.Object getFieldValue(_Fields field) {
+    switch (field) {
+    case OPTIMIZER_ID:
+      return getOptimizerId();
+
+    case GROUP_ID:
+      return getGroupId();
+
+    case GROUP_NAME:
+      return getGroupName();
+
+    case CORE_NUMBER:
+      return getCoreNumber();
+
+    case MEMORY_SIZE:
+      return getMemorySize();
+
+    case CONTAINER:
+      return getContainer();
+
+    case STATUS:
+      return getStatus();
+
+    case UPDATE_TIME:
+      return getUpdateTime();
+
+    }
+    throw new java.lang.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();
+    }
+
+    switch (field) {
+    case OPTIMIZER_ID:
+      return isSetOptimizerId();
+    case GROUP_ID:
+      return isSetGroupId();
+    case GROUP_NAME:
+      return isSetGroupName();
+    case CORE_NUMBER:
+      return isSetCoreNumber();
+    case MEMORY_SIZE:
+      return isSetMemorySize();
+    case CONTAINER:
+      return isSetContainer();
+    case STATUS:
+      return isSetStatus();
+    case UPDATE_TIME:
+      return isSetUpdateTime();
+    }
+    throw new java.lang.IllegalStateException();
+  }
+
+  @Override
+  public boolean equals(java.lang.Object that) {
+    if (that == null)
+      return false;
+    if (that instanceof OptimizerDescriptor)
+      return this.equals((OptimizerDescriptor)that);
+    return false;
+  }
+
+  public boolean equals(OptimizerDescriptor that) {
+    if (that == null)
+      return false;
+    if (this == that)
+      return true;
+
+    boolean this_present_optimizerId = true;
+    boolean that_present_optimizerId = true;
+    if (this_present_optimizerId || that_present_optimizerId) {
+      if (!(this_present_optimizerId && that_present_optimizerId))
+        return false;
+      if (this.optimizerId != that.optimizerId)
+        return false;
+    }
+
+    boolean this_present_groupId = true;
+    boolean that_present_groupId = true;
+    if (this_present_groupId || that_present_groupId) {
+      if (!(this_present_groupId && that_present_groupId))
+        return false;
+      if (this.groupId != that.groupId)
+        return false;
+    }
+
+    boolean this_present_groupName = true && this.isSetGroupName();
+    boolean that_present_groupName = true && that.isSetGroupName();
+    if (this_present_groupName || that_present_groupName) {
+      if (!(this_present_groupName && that_present_groupName))
+        return false;
+      if (!this.groupName.equals(that.groupName))
+        return false;
+    }
+
+    boolean this_present_coreNumber = true;
+    boolean that_present_coreNumber = true;
+    if (this_present_coreNumber || that_present_coreNumber) {
+      if (!(this_present_coreNumber && that_present_coreNumber))
+        return false;
+      if (this.coreNumber != that.coreNumber)
+        return false;
+    }
+
+    boolean this_present_memorySize = true;
+    boolean that_present_memorySize = true;
+    if (this_present_memorySize || that_present_memorySize) {
+      if (!(this_present_memorySize && that_present_memorySize))
+        return false;
+      if (this.memorySize != that.memorySize)
+        return false;
+    }
+
+    boolean this_present_container = true && this.isSetContainer();
+    boolean that_present_container = true && that.isSetContainer();
+    if (this_present_container || that_present_container) {
+      if (!(this_present_container && that_present_container))
+        return false;
+      if (!this.container.equals(that.container))
+        return false;
+    }
+
+    boolean this_present_status = true && this.isSetStatus();
+    boolean that_present_status = true && that.isSetStatus();
+    if (this_present_status || that_present_status) {
+      if (!(this_present_status && that_present_status))
+        return false;
+      if (!this.status.equals(that.status))
+        return false;
+    }
+
+    boolean this_present_updateTime = true;
+    boolean that_present_updateTime = true;
+    if (this_present_updateTime || that_present_updateTime) {
+      if (!(this_present_updateTime && that_present_updateTime))
+        return false;
+      if (this.updateTime != that.updateTime)
+        return false;
+    }
+
+    return true;
+  }
+
+  @Override
+  public int hashCode() {
+    int hashCode = 1;
+
+    hashCode = hashCode * 8191 + optimizerId;
+
+    hashCode = hashCode * 8191 + groupId;
+
+    hashCode = hashCode * 8191 + ((isSetGroupName()) ? 131071 : 524287);
+    if (isSetGroupName())
+      hashCode = hashCode * 8191 + groupName.hashCode();
+
+    hashCode = hashCode * 8191 + coreNumber;
+
+    hashCode = hashCode * 8191 + org.apache.thrift.TBaseHelper.hashCode(memorySize);
+
+    hashCode = hashCode * 8191 + ((isSetContainer()) ? 131071 : 524287);
+    if (isSetContainer())
+      hashCode = hashCode * 8191 + container.hashCode();
+
+    hashCode = hashCode * 8191 + ((isSetStatus()) ? 131071 : 524287);
+    if (isSetStatus())
+      hashCode = hashCode * 8191 + status.hashCode();
+
+    hashCode = hashCode * 8191 + org.apache.thrift.TBaseHelper.hashCode(updateTime);
+
+    return hashCode;
+  }
+
+  @Override
+  public int compareTo(OptimizerDescriptor other) {
+    if (!getClass().equals(other.getClass())) {
+      return getClass().getName().compareTo(other.getClass().getName());
+    }
+
+    int lastComparison = 0;
+
+    lastComparison = java.lang.Boolean.valueOf(isSetOptimizerId()).compareTo(other.isSetOptimizerId());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetOptimizerId()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.optimizerId, other.optimizerId);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
+    lastComparison = java.lang.Boolean.valueOf(isSetGroupId()).compareTo(other.isSetGroupId());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetGroupId()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.groupId, other.groupId);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
+    lastComparison = java.lang.Boolean.valueOf(isSetGroupName()).compareTo(other.isSetGroupName());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetGroupName()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.groupName, other.groupName);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
+    lastComparison = java.lang.Boolean.valueOf(isSetCoreNumber()).compareTo(other.isSetCoreNumber());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetCoreNumber()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.coreNumber, other.coreNumber);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
+    lastComparison = java.lang.Boolean.valueOf(isSetMemorySize()).compareTo(other.isSetMemorySize());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetMemorySize()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.memorySize, other.memorySize);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
+    lastComparison = java.lang.Boolean.valueOf(isSetContainer()).compareTo(other.isSetContainer());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetContainer()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.container, other.container);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
+    lastComparison = java.lang.Boolean.valueOf(isSetStatus()).compareTo(other.isSetStatus());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetStatus()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.status, other.status);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
+    lastComparison = java.lang.Boolean.valueOf(isSetUpdateTime()).compareTo(other.isSetUpdateTime());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetUpdateTime()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.updateTime, other.updateTime);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
+    return 0;
+  }
+
+  @org.apache.thrift.annotation.Nullable
+  public _Fields fieldForId(int fieldId) {
+    return _Fields.findByThriftId(fieldId);
+  }
+
+  public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+    scheme(iprot).read(iprot, this);
+  }
+
+  public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+    scheme(oprot).write(oprot, this);
+  }
+
+  @Override
+  public java.lang.String toString() {
+    java.lang.StringBuilder sb = new java.lang.StringBuilder("OptimizerDescriptor(");
+    boolean first = true;
+
+    sb.append("optimizerId:");
+    sb.append(this.optimizerId);
+    first = false;
+    if (!first) sb.append(", ");
+    sb.append("groupId:");
+    sb.append(this.groupId);
+    first = false;
+    if (!first) sb.append(", ");
+    sb.append("groupName:");
+    if (this.groupName == null) {
+      sb.append("null");
+    } else {
+      sb.append(this.groupName);
+    }
+    first = false;
+    if (!first) sb.append(", ");
+    sb.append("coreNumber:");
+    sb.append(this.coreNumber);
+    first = false;
+    if (!first) sb.append(", ");
+    sb.append("memorySize:");
+    sb.append(this.memorySize);
+    first = false;
+    if (!first) sb.append(", ");
+    sb.append("container:");
+    if (this.container == null) {
+      sb.append("null");
+    } else {
+      sb.append(this.container);
+    }
+    first = false;
+    if (!first) sb.append(", ");
+    sb.append("status:");
+    if (this.status == null) {
+      sb.append("null");
+    } else {
+      sb.append(this.status);
+    }
+    first = false;
+    if (!first) sb.append(", ");
+    sb.append("updateTime:");
+    sb.append(this.updateTime);
+    first = false;
+    sb.append(")");
+    return sb.toString();
+  }
+
+  public void validate() throws org.apache.thrift.TException {
+    // check for required fields
+    // check for sub-struct validity
+  }
+
+  private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+    try {
+      write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+    } catch (org.apache.thrift.TException te) {
+      throw new java.io.IOException(te);
+    }
+  }
+
+  private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, java.lang.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;
+      read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+    } catch (org.apache.thrift.TException te) {
+      throw new java.io.IOException(te);
+    }
+  }
+
+  private static class OptimizerDescriptorStandardSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+    public OptimizerDescriptorStandardScheme getScheme() {
+      return new OptimizerDescriptorStandardScheme();
+    }
+  }
+
+  private static class OptimizerDescriptorStandardScheme extends org.apache.thrift.scheme.StandardScheme<OptimizerDescriptor> {
+
+    public void read(org.apache.thrift.protocol.TProtocol iprot, OptimizerDescriptor struct) throws org.apache.thrift.TException {
+      org.apache.thrift.protocol.TField schemeField;
+      iprot.readStructBegin();
+      while (true)
+      {
+        schemeField = iprot.readFieldBegin();
+        if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { 
+          break;
+        }
+        switch (schemeField.id) {
+          case 1: // OPTIMIZER_ID
+            if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
+              struct.optimizerId = iprot.readI32();
+              struct.setOptimizerIdIsSet(true);
+            } else { 
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+            }
+            break;
+          case 2: // GROUP_ID
+            if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
+              struct.groupId = iprot.readI32();
+              struct.setGroupIdIsSet(true);
+            } else { 
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+            }
+            break;
+          case 3: // GROUP_NAME
+            if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
+              struct.groupName = iprot.readString();
+              struct.setGroupNameIsSet(true);
+            } else { 
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+            }
+            break;
+          case 4: // CORE_NUMBER
+            if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
+              struct.coreNumber = iprot.readI32();
+              struct.setCoreNumberIsSet(true);
+            } else { 
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+            }
+            break;
+          case 5: // MEMORY_SIZE
+            if (schemeField.type == org.apache.thrift.protocol.TType.I64) {
+              struct.memorySize = iprot.readI64();
+              struct.setMemorySizeIsSet(true);
+            } else { 
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+            }
+            break;
+          case 6: // CONTAINER
+            if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
+              struct.container = iprot.readString();
+              struct.setContainerIsSet(true);
+            } else { 
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+            }
+            break;
+          case 7: // STATUS
+            if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
+              struct.status = iprot.readString();
+              struct.setStatusIsSet(true);
+            } else { 
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+            }
+            break;
+          case 8: // UPDATE_TIME
+            if (schemeField.type == org.apache.thrift.protocol.TType.I64) {
+              struct.updateTime = iprot.readI64();
+              struct.setUpdateTimeIsSet(true);
+            } else { 
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+            }
+            break;
+          default:
+            org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+        }
+        iprot.readFieldEnd();
+      }
+      iprot.readStructEnd();
+
+      // check for required fields of primitive type, which can't be checked in the validate method
+      struct.validate();
+    }
+
+    public void write(org.apache.thrift.protocol.TProtocol oprot, OptimizerDescriptor struct) throws org.apache.thrift.TException {
+      struct.validate();
+
+      oprot.writeStructBegin(STRUCT_DESC);
+      oprot.writeFieldBegin(OPTIMIZER_ID_FIELD_DESC);
+      oprot.writeI32(struct.optimizerId);
+      oprot.writeFieldEnd();
+      oprot.writeFieldBegin(GROUP_ID_FIELD_DESC);
+      oprot.writeI32(struct.groupId);
+      oprot.writeFieldEnd();
+      if (struct.groupName != null) {
+        oprot.writeFieldBegin(GROUP_NAME_FIELD_DESC);
+        oprot.writeString(struct.groupName);
+        oprot.writeFieldEnd();
+      }
+      oprot.writeFieldBegin(CORE_NUMBER_FIELD_DESC);
+      oprot.writeI32(struct.coreNumber);
+      oprot.writeFieldEnd();
+      oprot.writeFieldBegin(MEMORY_SIZE_FIELD_DESC);
+      oprot.writeI64(struct.memorySize);
+      oprot.writeFieldEnd();
+      if (struct.container != null) {
+        oprot.writeFieldBegin(CONTAINER_FIELD_DESC);
+        oprot.writeString(struct.container);
+        oprot.writeFieldEnd();
+      }
+      if (struct.status != null) {
+        oprot.writeFieldBegin(STATUS_FIELD_DESC);
+        oprot.writeString(struct.status);
+        oprot.writeFieldEnd();
+      }
+      oprot.writeFieldBegin(UPDATE_TIME_FIELD_DESC);
+      oprot.writeI64(struct.updateTime);
+      oprot.writeFieldEnd();
+      oprot.writeFieldStop();
+      oprot.writeStructEnd();
+    }
+
+  }
+
+  private static class OptimizerDescriptorTupleSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+    public OptimizerDescriptorTupleScheme getScheme() {
+      return new OptimizerDescriptorTupleScheme();
+    }
+  }
+
+  private static class OptimizerDescriptorTupleScheme extends org.apache.thrift.scheme.TupleScheme<OptimizerDescriptor> {
+
+    @Override
+    public void write(org.apache.thrift.protocol.TProtocol prot, OptimizerDescriptor struct) throws org.apache.thrift.TException {
+      org.apache.thrift.protocol.TTupleProtocol oprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+      java.util.BitSet optionals = new java.util.BitSet();
+      if (struct.isSetOptimizerId()) {
+        optionals.set(0);
+      }
+      if (struct.isSetGroupId()) {
+        optionals.set(1);
+      }
+      if (struct.isSetGroupName()) {
+        optionals.set(2);
+      }
+      if (struct.isSetCoreNumber()) {
+        optionals.set(3);
+      }
+      if (struct.isSetMemorySize()) {
+        optionals.set(4);
+      }
+      if (struct.isSetContainer()) {
+        optionals.set(5);
+      }
+      if (struct.isSetStatus()) {
+        optionals.set(6);
+      }
+      if (struct.isSetUpdateTime()) {
+        optionals.set(7);
+      }
+      oprot.writeBitSet(optionals, 8);
+      if (struct.isSetOptimizerId()) {
+        oprot.writeI32(struct.optimizerId);
+      }
+      if (struct.isSetGroupId()) {
+        oprot.writeI32(struct.groupId);
+      }
+      if (struct.isSetGroupName()) {
+        oprot.writeString(struct.groupName);
+      }
+      if (struct.isSetCoreNumber()) {
+        oprot.writeI32(struct.coreNumber);
+      }
+      if (struct.isSetMemorySize()) {
+        oprot.writeI64(struct.memorySize);
+      }
+      if (struct.isSetContainer()) {
+        oprot.writeString(struct.container);
+      }
+      if (struct.isSetStatus()) {
+        oprot.writeString(struct.status);
+      }
+      if (struct.isSetUpdateTime()) {
+        oprot.writeI64(struct.updateTime);
+      }
+    }
+
+    @Override
+    public void read(org.apache.thrift.protocol.TProtocol prot, OptimizerDescriptor struct) throws org.apache.thrift.TException {
+      org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+      java.util.BitSet incoming = iprot.readBitSet(8);
+      if (incoming.get(0)) {
+        struct.optimizerId = iprot.readI32();
+        struct.setOptimizerIdIsSet(true);
+      }
+      if (incoming.get(1)) {
+        struct.groupId = iprot.readI32();
+        struct.setGroupIdIsSet(true);
+      }
+      if (incoming.get(2)) {
+        struct.groupName = iprot.readString();
+        struct.setGroupNameIsSet(true);
+      }
+      if (incoming.get(3)) {
+        struct.coreNumber = iprot.readI32();
+        struct.setCoreNumberIsSet(true);
+      }
+      if (incoming.get(4)) {
+        struct.memorySize = iprot.readI64();
+        struct.setMemorySizeIsSet(true);
+      }
+      if (incoming.get(5)) {
+        struct.container = iprot.readString();
+        struct.setContainerIsSet(true);
+      }
+      if (incoming.get(6)) {
+        struct.status = iprot.readString();
+        struct.setStatusIsSet(true);
+      }
+      if (incoming.get(7)) {
+        struct.updateTime = iprot.readI64();
+        struct.setUpdateTimeIsSet(true);
+      }
+    }
+  }
+
+  private static <S extends org.apache.thrift.scheme.IScheme> S scheme(org.apache.thrift.protocol.TProtocol proto) {
+    return (org.apache.thrift.scheme.StandardScheme.class.equals(proto.getScheme()) ? STANDARD_SCHEME_FACTORY : TUPLE_SCHEME_FACTORY).getScheme();
+  }
+}
+
diff --git a/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizerRegisterInfo.java b/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizerRegisterInfo.java
new file mode 100644
index 000000000..84065d7f7
--- /dev/null
+++ b/ams/ams-api/src/main/gen-java/com/netease/arctic/ams/api/OptimizerRegisterInfo.java
@@ -0,0 +1,573 @@
+/**
+ * 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;
+
+@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked", "unused"})
+@javax.annotation.Generated(value = "Autogenerated by Thrift Compiler (0.13.0)", date = "2022-09-19")
+public class OptimizerRegisterInfo implements org.apache.thrift.TBase<OptimizerRegisterInfo, OptimizerRegisterInfo._Fields>, java.io.Serializable, Cloneable, Comparable<OptimizerRegisterInfo> {
+  private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("OptimizerRegisterInfo");
+
+  private static final org.apache.thrift.protocol.TField OPTIMIZER_GROUP_NAME_FIELD_DESC = new org.apache.thrift.protocol.TField("optimizerGroupName", org.apache.thrift.protocol.TType.STRING, (short)1);
+  private static final org.apache.thrift.protocol.TField CORE_NUMBER_FIELD_DESC = new org.apache.thrift.protocol.TField("coreNumber", org.apache.thrift.protocol.TType.I32, (short)2);
+  private static final org.apache.thrift.protocol.TField MEMORY_SIZE_FIELD_DESC = new org.apache.thrift.protocol.TField("memorySize", org.apache.thrift.protocol.TType.I64, (short)3);
+
+  private static final org.apache.thrift.scheme.SchemeFactory STANDARD_SCHEME_FACTORY = new OptimizerRegisterInfoStandardSchemeFactory();
+  private static final org.apache.thrift.scheme.SchemeFactory TUPLE_SCHEME_FACTORY = new OptimizerRegisterInfoTupleSchemeFactory();
+
+  public @org.apache.thrift.annotation.Nullable java.lang.String optimizerGroupName; // required
+  public int coreNumber; // required
+  public long memorySize; // 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 {
+    OPTIMIZER_GROUP_NAME((short)1, "optimizerGroupName"),
+    CORE_NUMBER((short)2, "coreNumber"),
+    MEMORY_SIZE((short)3, "memorySize");
+
+    private static final java.util.Map<java.lang.String, _Fields> byName = new java.util.HashMap<java.lang.String, _Fields>();
+
+    static {
+      for (_Fields field : java.util.EnumSet.allOf(_Fields.class)) {
+        byName.put(field.getFieldName(), field);
+      }
+    }
+
+    /**
+     * Find the _Fields constant that matches fieldId, or null if its not found.
+     */
+    @org.apache.thrift.annotation.Nullable
+    public static _Fields findByThriftId(int fieldId) {
+      switch(fieldId) {
+        case 1: // OPTIMIZER_GROUP_NAME
+          return OPTIMIZER_GROUP_NAME;
+        case 2: // CORE_NUMBER
+          return CORE_NUMBER;
+        case 3: // MEMORY_SIZE
+          return MEMORY_SIZE;
+        default:
+          return null;
+      }
+    }
+
+    /**
+     * Find the _Fields constant that matches fieldId, throwing an exception
+     * if it is not found.
+     */
+    public static _Fields findByThriftIdOrThrow(int fieldId) {
+      _Fields fields = findByThriftId(fieldId);
+      if (fields == null) throw new java.lang.IllegalArgumentException("Field " + fieldId + " doesn't exist!");
+      return fields;
+    }
+
+    /**
+     * 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) {
+      return byName.get(name);
+    }
+
+    private final short _thriftId;
+    private final java.lang.String _fieldName;
+
+    _Fields(short thriftId, java.lang.String fieldName) {
+      _thriftId = thriftId;
+      _fieldName = fieldName;
+    }
+
+    public short getThriftFieldId() {
+      return _thriftId;
+    }
+
+    public java.lang.String getFieldName() {
+      return _fieldName;
+    }
+  }
+
+  // isset id assignments
+  private static final int __CORENUMBER_ISSET_ID = 0;
+  private static final int __MEMORYSIZE_ISSET_ID = 1;
+  private byte __isset_bitfield = 0;
+  public static final java.util.Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
+  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.OPTIMIZER_GROUP_NAME, new org.apache.thrift.meta_data.FieldMetaData("optimizerGroupName", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
+    tmpMap.put(_Fields.CORE_NUMBER, new org.apache.thrift.meta_data.FieldMetaData("coreNumber", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I32)));
+    tmpMap.put(_Fields.MEMORY_SIZE, new org.apache.thrift.meta_data.FieldMetaData("memorySize", org.apache.thrift.TFieldRequirementType.DEFAULT, 
+        new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I64)));
+    metaDataMap = java.util.Collections.unmodifiableMap(tmpMap);
+    org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(OptimizerRegisterInfo.class, metaDataMap);
+  }
+
+  public OptimizerRegisterInfo() {
+  }
+
+  public OptimizerRegisterInfo(
+    java.lang.String optimizerGroupName,
+    int coreNumber,
+    long memorySize)
+  {
+    this();
+    this.optimizerGroupName = optimizerGroupName;
+    this.coreNumber = coreNumber;
+    setCoreNumberIsSet(true);
+    this.memorySize = memorySize;
+    setMemorySizeIsSet(true);
+  }
+
+  /**
+   * Performs a deep copy on <i>other</i>.
+   */
+  public OptimizerRegisterInfo(OptimizerRegisterInfo other) {
+    __isset_bitfield = other.__isset_bitfield;
+    if (other.isSetOptimizerGroupName()) {
+      this.optimizerGroupName = other.optimizerGroupName;
+    }
+    this.coreNumber = other.coreNumber;
+    this.memorySize = other.memorySize;
+  }
+
+  public OptimizerRegisterInfo deepCopy() {
+    return new OptimizerRegisterInfo(this);
+  }
+
+  @Override
+  public void clear() {
+    this.optimizerGroupName = null;
+    setCoreNumberIsSet(false);
+    this.coreNumber = 0;
+    setMemorySizeIsSet(false);
+    this.memorySize = 0;
+  }
+
+  @org.apache.thrift.annotation.Nullable
+  public java.lang.String getOptimizerGroupName() {
+    return this.optimizerGroupName;
+  }
+
+  public OptimizerRegisterInfo setOptimizerGroupName(@org.apache.thrift.annotation.Nullable java.lang.String optimizerGroupName) {
+    this.optimizerGroupName = optimizerGroupName;
+    return this;
+  }
+
+  public void unsetOptimizerGroupName() {
+    this.optimizerGroupName = null;
+  }
+
+  /** Returns true if field optimizerGroupName is set (has been assigned a value) and false otherwise */
+  public boolean isSetOptimizerGroupName() {
+    return this.optimizerGroupName != null;
+  }
+
+  public void setOptimizerGroupNameIsSet(boolean value) {
+    if (!value) {
+      this.optimizerGroupName = null;
+    }
+  }
+
+  public int getCoreNumber() {
+    return this.coreNumber;
+  }
+
+  public OptimizerRegisterInfo setCoreNumber(int coreNumber) {
+    this.coreNumber = coreNumber;
+    setCoreNumberIsSet(true);
+    return this;
+  }
+
+  public void unsetCoreNumber() {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.clearBit(__isset_bitfield, __CORENUMBER_ISSET_ID);
+  }
+
+  /** Returns true if field coreNumber is set (has been assigned a value) and false otherwise */
+  public boolean isSetCoreNumber() {
+    return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield, __CORENUMBER_ISSET_ID);
+  }
+
+  public void setCoreNumberIsSet(boolean value) {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.setBit(__isset_bitfield, __CORENUMBER_ISSET_ID, value);
+  }
+
+  public long getMemorySize() {
+    return this.memorySize;
+  }
+
+  public OptimizerRegisterInfo setMemorySize(long memorySize) {
+    this.memorySize = memorySize;
+    setMemorySizeIsSet(true);
+    return this;
+  }
+
+  public void unsetMemorySize() {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.clearBit(__isset_bitfield, __MEMORYSIZE_ISSET_ID);
+  }
+
+  /** Returns true if field memorySize is set (has been assigned a value) and false otherwise */
+  public boolean isSetMemorySize() {
+    return org.apache.thrift.EncodingUtils.testBit(__isset_bitfield, __MEMORYSIZE_ISSET_ID);
+  }
+
+  public void setMemorySizeIsSet(boolean value) {
+    __isset_bitfield = org.apache.thrift.EncodingUtils.setBit(__isset_bitfield, __MEMORYSIZE_ISSET_ID, value);
+  }
+
+  public void setFieldValue(_Fields field, @org.apache.thrift.annotation.Nullable java.lang.Object value) {
+    switch (field) {
+    case OPTIMIZER_GROUP_NAME:
+      if (value == null) {
+        unsetOptimizerGroupName();
+      } else {
+        setOptimizerGroupName((java.lang.String)value);
+      }
+      break;
+
+    case CORE_NUMBER:
+      if (value == null) {
+        unsetCoreNumber();
+      } else {
+        setCoreNumber((java.lang.Integer)value);
+      }
+      break;
+
+    case MEMORY_SIZE:
+      if (value == null) {
+        unsetMemorySize();
+      } else {
+        setMemorySize((java.lang.Long)value);
+      }
+      break;
+
+    }
+  }
+
+  @org.apache.thrift.annotation.Nullable
+  public java.lang.Object getFieldValue(_Fields field) {
+    switch (field) {
+    case OPTIMIZER_GROUP_NAME:
+      return getOptimizerGroupName();
+
+    case CORE_NUMBER:
+      return getCoreNumber();
+
+    case MEMORY_SIZE:
+      return getMemorySize();
+
+    }
+    throw new java.lang.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();
+    }
+
+    switch (field) {
+    case OPTIMIZER_GROUP_NAME:
+      return isSetOptimizerGroupName();
+    case CORE_NUMBER:
+      return isSetCoreNumber();
+    case MEMORY_SIZE:
+      return isSetMemorySize();
+    }
+    throw new java.lang.IllegalStateException();
+  }
+
+  @Override
+  public boolean equals(java.lang.Object that) {
+    if (that == null)
+      return false;
+    if (that instanceof OptimizerRegisterInfo)
+      return this.equals((OptimizerRegisterInfo)that);
+    return false;
+  }
+
+  public boolean equals(OptimizerRegisterInfo that) {
+    if (that == null)
+      return false;
+    if (this == that)
+      return true;
+
+    boolean this_present_optimizerGroupName = true && this.isSetOptimizerGroupName();
+    boolean that_present_optimizerGroupName = true && that.isSetOptimizerGroupName();
+    if (this_present_optimizerGroupName || that_present_optimizerGroupName) {
+      if (!(this_present_optimizerGroupName && that_present_optimizerGroupName))
+        return false;
+      if (!this.optimizerGroupName.equals(that.optimizerGroupName))
+        return false;
+    }
+
+    boolean this_present_coreNumber = true;
+    boolean that_present_coreNumber = true;
+    if (this_present_coreNumber || that_present_coreNumber) {
+      if (!(this_present_coreNumber && that_present_coreNumber))
+        return false;
+      if (this.coreNumber != that.coreNumber)
+        return false;
+    }
+
+    boolean this_present_memorySize = true;
+    boolean that_present_memorySize = true;
+    if (this_present_memorySize || that_present_memorySize) {
+      if (!(this_present_memorySize && that_present_memorySize))
+        return false;
+      if (this.memorySize != that.memorySize)
+        return false;
+    }
+
+    return true;
+  }
+
+  @Override
+  public int hashCode() {
+    int hashCode = 1;
+
+    hashCode = hashCode * 8191 + ((isSetOptimizerGroupName()) ? 131071 : 524287);
+    if (isSetOptimizerGroupName())
+      hashCode = hashCode * 8191 + optimizerGroupName.hashCode();
+
+    hashCode = hashCode * 8191 + coreNumber;
+
+    hashCode = hashCode * 8191 + org.apache.thrift.TBaseHelper.hashCode(memorySize);
+
+    return hashCode;
+  }
+
+  @Override
+  public int compareTo(OptimizerRegisterInfo other) {
+    if (!getClass().equals(other.getClass())) {
+      return getClass().getName().compareTo(other.getClass().getName());
+    }
+
+    int lastComparison = 0;
+
+    lastComparison = java.lang.Boolean.valueOf(isSetOptimizerGroupName()).compareTo(other.isSetOptimizerGroupName());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetOptimizerGroupName()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.optimizerGroupName, other.optimizerGroupName);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
+    lastComparison = java.lang.Boolean.valueOf(isSetCoreNumber()).compareTo(other.isSetCoreNumber());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetCoreNumber()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.coreNumber, other.coreNumber);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
+    lastComparison = java.lang.Boolean.valueOf(isSetMemorySize()).compareTo(other.isSetMemorySize());
+    if (lastComparison != 0) {
+      return lastComparison;
+    }
+    if (isSetMemorySize()) {
+      lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.memorySize, other.memorySize);
+      if (lastComparison != 0) {
+        return lastComparison;
+      }
+    }
+    return 0;
+  }
+
+  @org.apache.thrift.annotation.Nullable
+  public _Fields fieldForId(int fieldId) {
+    return _Fields.findByThriftId(fieldId);
+  }
+
+  public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
+    scheme(iprot).read(iprot, this);
+  }
+
+  public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
+    scheme(oprot).write(oprot, this);
+  }
+
+  @Override
+  public java.lang.String toString() {
+    java.lang.StringBuilder sb = new java.lang.StringBuilder("OptimizerRegisterInfo(");
+    boolean first = true;
+
+    sb.append("optimizerGroupName:");
+    if (this.optimizerGroupName == null) {
+      sb.append("null");
+    } else {
+      sb.append(this.optimizerGroupName);
+    }
+    first = false;
+    if (!first) sb.append(", ");
+    sb.append("coreNumber:");
+    sb.append(this.coreNumber);
+    first = false;
+    if (!first) sb.append(", ");
+    sb.append("memorySize:");
+    sb.append(this.memorySize);
+    first = false;
+    sb.append(")");
+    return sb.toString();
+  }
+
+  public void validate() throws org.apache.thrift.TException {
+    // check for required fields
+    // check for sub-struct validity
+  }
+
+  private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
+    try {
+      write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
+    } catch (org.apache.thrift.TException te) {
+      throw new java.io.IOException(te);
+    }
+  }
+
+  private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, java.lang.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;
+      read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
+    } catch (org.apache.thrift.TException te) {
+      throw new java.io.IOException(te);
+    }
+  }
+
+  private static class OptimizerRegisterInfoStandardSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+    public OptimizerRegisterInfoStandardScheme getScheme() {
+      return new OptimizerRegisterInfoStandardScheme();
+    }
+  }
+
+  private static class OptimizerRegisterInfoStandardScheme extends org.apache.thrift.scheme.StandardScheme<OptimizerRegisterInfo> {
+
+    public void read(org.apache.thrift.protocol.TProtocol iprot, OptimizerRegisterInfo struct) throws org.apache.thrift.TException {
+      org.apache.thrift.protocol.TField schemeField;
+      iprot.readStructBegin();
+      while (true)
+      {
+        schemeField = iprot.readFieldBegin();
+        if (schemeField.type == org.apache.thrift.protocol.TType.STOP) { 
+          break;
+        }
+        switch (schemeField.id) {
+          case 1: // OPTIMIZER_GROUP_NAME
+            if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
+              struct.optimizerGroupName = iprot.readString();
+              struct.setOptimizerGroupNameIsSet(true);
+            } else { 
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+            }
+            break;
+          case 2: // CORE_NUMBER
+            if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
+              struct.coreNumber = iprot.readI32();
+              struct.setCoreNumberIsSet(true);
+            } else { 
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+            }
+            break;
+          case 3: // MEMORY_SIZE
+            if (schemeField.type == org.apache.thrift.protocol.TType.I64) {
+              struct.memorySize = iprot.readI64();
+              struct.setMemorySizeIsSet(true);
+            } else { 
+              org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+            }
+            break;
+          default:
+            org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
+        }
+        iprot.readFieldEnd();
+      }
+      iprot.readStructEnd();
+
+      // check for required fields of primitive type, which can't be checked in the validate method
+      struct.validate();
+    }
+
+    public void write(org.apache.thrift.protocol.TProtocol oprot, OptimizerRegisterInfo struct) throws org.apache.thrift.TException {
+      struct.validate();
+
+      oprot.writeStructBegin(STRUCT_DESC);
+      if (struct.optimizerGroupName != null) {
+        oprot.writeFieldBegin(OPTIMIZER_GROUP_NAME_FIELD_DESC);
+        oprot.writeString(struct.optimizerGroupName);
+        oprot.writeFieldEnd();
+      }
+      oprot.writeFieldBegin(CORE_NUMBER_FIELD_DESC);
+      oprot.writeI32(struct.coreNumber);
+      oprot.writeFieldEnd();
+      oprot.writeFieldBegin(MEMORY_SIZE_FIELD_DESC);
+      oprot.writeI64(struct.memorySize);
+      oprot.writeFieldEnd();
+      oprot.writeFieldStop();
+      oprot.writeStructEnd();
+    }
+
+  }
+
+  private static class OptimizerRegisterInfoTupleSchemeFactory implements org.apache.thrift.scheme.SchemeFactory {
+    public OptimizerRegisterInfoTupleScheme getScheme() {
+      return new OptimizerRegisterInfoTupleScheme();
+    }
+  }
+
+  private static class OptimizerRegisterInfoTupleScheme extends org.apache.thrift.scheme.TupleScheme<OptimizerRegisterInfo> {
+
+    @Override
+    public void write(org.apache.thrift.protocol.TProtocol prot, OptimizerRegisterInfo struct) throws org.apache.thrift.TException {
+      org.apache.thrift.protocol.TTupleProtocol oprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+      java.util.BitSet optionals = new java.util.BitSet();
+      if (struct.isSetOptimizerGroupName()) {
+        optionals.set(0);
+      }
+      if (struct.isSetCoreNumber()) {
+        optionals.set(1);
+      }
+      if (struct.isSetMemorySize()) {
+        optionals.set(2);
+      }
+      oprot.writeBitSet(optionals, 3);
+      if (struct.isSetOptimizerGroupName()) {
+        oprot.writeString(struct.optimizerGroupName);
+      }
+      if (struct.isSetCoreNumber()) {
+        oprot.writeI32(struct.coreNumber);
+      }
+      if (struct.isSetMemorySize()) {
+        oprot.writeI64(struct.memorySize);
+      }
+    }
+
+    @Override
+    public void read(org.apache.thrift.protocol.TProtocol prot, OptimizerRegisterInfo struct) throws org.apache.thrift.TException {
+      org.apache.thrift.protocol.TTupleProtocol iprot = (org.apache.thrift.protocol.TTupleProtocol) prot;
+      java.util.BitSet incoming = iprot.readBitSet(3);
+      if (incoming.get(0)) {
+        struct.optimizerGroupName = iprot.readString();
+        struct.setOptimizerGroupNameIsSet(true);
+      }
+      if (incoming.get(1)) {
+        struct.coreNumber = iprot.readI32();
+        struct.setCoreNumberIsSet(true);
+      }
+      if (incoming.get(2)) {
+        struct.memorySize = iprot.readI64();
+        struct.setMemorySizeIsSet(true);
+      }
+    }
+  }
+
+  private static <S extends org.apache.thrift.scheme.IScheme> S scheme(org.apache.thrift.protocol.TProtocol proto) {
+    return (org.apache.thrift.scheme.StandardScheme.class.equals(proto.getScheme()) ? STANDARD_SCHEME_FACTORY : TUPLE_SCHEME_FACTORY).getScheme();
+  }
+}
+
diff --git a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/CatalogMetaProperties.java b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/CatalogMetaProperties.java
index 0db48d13a..bfd6f9cc4 100644
--- a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/CatalogMetaProperties.java
+++ b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/CatalogMetaProperties.java
@@ -41,6 +41,10 @@ public class CatalogMetaProperties {
 
     public static final String CATALOG_TYPE_HADOOP = "hadoop";
     public static final String CATALOG_TYPE_HIVE = "hive";
+    public static final String CATALOG_TYPE_AMS = "ams";
+    public static final String CATALOG_TYPE_CUSTOM = "custom";
+
+    public static final String TABLE_FORMATS = "table-formats";
 
     public static final String CLIENT_POOL_SIZE = "clients";
     public static final int CLIENT_POOL_SIZE_DEFAULT = 2;
diff --git a/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/TableFormat.java b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/TableFormat.java
new file mode 100644
index 000000000..9b4e2a1c7
--- /dev/null
+++ b/ams/ams-api/src/main/java/com/netease/arctic/ams/api/properties/TableFormat.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.ams.api.properties;
+
+/**
+ * Table formats Arctic supported
+ *
+ * @since 0.4.0
+ */
+public enum TableFormat {
+  ICEBERG, HIVE;
+}
diff --git a/ams/ams-api/src/main/thrift/arctic_optimize_manager.thrift b/ams/ams-api/src/main/thrift/arctic_optimize_manager.thrift
index ae7dd34cb..013a24483 100644
--- a/ams/ams-api/src/main/thrift/arctic_optimize_manager.thrift
+++ b/ams/ams-api/src/main/thrift/arctic_optimize_manager.thrift
@@ -94,6 +94,23 @@ struct DataFileInfo {
     9: i64 recordCount;
 }
 
+struct OptimizerRegisterInfo {
+    1: string optimizerGroupName;
+    2: i32 coreNumber;
+    3: i64 memorySize;
+}
+
+struct OptimizerDescriptor {
+    1: i32 optimizerId;
+    2: i32 groupId;
+    3: string groupName;
+    4: i32 coreNumber;
+    5: i64 memorySize;
+    6: string container;
+    7: string status;
+    8: i64 updateTime;
+}
+
 /**
 * replace TableContainer
 **/
@@ -107,4 +124,6 @@ service OptimizeManager {
     void reportOptimizeResult(1:OptimizeTaskStat optimizeTaskStat)
 
     void reportOptimizerState(1: OptimizerStateReport reportData)
+
+    OptimizerDescriptor registerOptimizer(1: OptimizerRegisterInfo registerInfo)
 }
\ No newline at end of file
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/ArcticMetaStore.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/ArcticMetaStore.java
index a931390c6..6cd568f07 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/ArcticMetaStore.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/ArcticMetaStore.java
@@ -114,6 +114,7 @@ public class ArcticMetaStore {
         startMetaStore(conf);
       }
     } catch (Throwable t) {
+      failover();
       LOG.error("MetaStore Thrift Server threw an exception...", t);
     }
   }
@@ -586,7 +587,9 @@ public class ArcticMetaStore {
       Container container = new Container();
       container.setName(optimize.getString(ConfigFileProperties.CONTAINER_NAME));
       container.setType(optimize.getString(ConfigFileProperties.CONTAINER_TYPE));
-      container.setProperties(optimize.getObject(ConfigFileProperties.CONTAINER_PROPERTIES, Map.class));
+      if (optimize.containsKey(ConfigFileProperties.CONTAINER_PROPERTIES)) {
+        container.setProperties(optimize.getObject(ConfigFileProperties.CONTAINER_PROPERTIES, Map.class));
+      }
 
       ServiceContainer.getOptimizeQueueService().insertContainer(container);
     }
@@ -612,7 +615,10 @@ public class ArcticMetaStore {
             "can not find such container config named" +
                 optimizeGroup.getString(ConfigFileProperties.OPTIMIZE_GROUP_CONTAINER));
       }
-      optimizeQueueMeta.properties = optimizeGroup.getObject(ConfigFileProperties.OPTIMIZE_GROUP_PROPERTIES, Map.class);
+      if (optimizeGroup.containsKey(ConfigFileProperties.OPTIMIZE_GROUP_PROPERTIES)) {
+        optimizeQueueMeta.properties =
+            optimizeGroup.getObject(ConfigFileProperties.OPTIMIZE_GROUP_PROPERTIES, Map.class);
+      }
       boolean updated = false;
       for (OptimizeQueueMeta meta : optimizeQueueMetas) {
         if (meta.name.equals(optimizeQueueMeta.name)) {
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ConfigFileProperties.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ConfigFileProperties.java
index 90f91949a..788c51e71 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ConfigFileProperties.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/config/ConfigFileProperties.java
@@ -50,6 +50,7 @@ public class ConfigFileProperties {
   public static final String CONTAINER_LIST = "containers";
   public static final String CONTAINER_NAME = "name";
   public static final String CONTAINER_TYPE = "type";
+  public static final String EXTERNAL_CONTAINER_TYPE = "external";
   public static final String CONTAINER_PROPERTIES = "properties";
 
   //optimize config
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/OptimizerController.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/OptimizerController.java
index 17cee928a..1f09e7a17 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/OptimizerController.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/controller/OptimizerController.java
@@ -48,7 +48,9 @@ import java.util.Date;
 import java.util.List;
 import java.util.Map;
 
-/** optimize controller.
+/**
+ * optimize controller.
+ *
  * @Description: optimizer is a task to compact small files in arctic table.
  * OptimizerController is the optimizer interface's controller, through this interface, you can get the optimized table,
  * optimizer task, optimizer group information, scale out or release optimizer, etc.
@@ -96,11 +98,11 @@ public class OptimizerController extends RestBaseController {
       arcticTableItemList.sort((o1, o2) -> {
         // first we compare the status , and then we compare the start time when status are equal;
         int statDiff = o1.getTableOptimizeRuntime().getOptimizeStatus()
-                .compareTo(o2.getTableOptimizeRuntime().getOptimizeStatus());
+            .compareTo(o2.getTableOptimizeRuntime().getOptimizeStatus());
         // status order is asc, startTime order is desc
         if (statDiff == 0) {
           long timeDiff = o1.getTableOptimizeRuntime().getOptimizeStatusStartTime() -
-                  o2.getTableOptimizeRuntime().getOptimizeStatusStartTime();
+              o2.getTableOptimizeRuntime().getOptimizeStatusStartTime();
           return timeDiff >= 0 ? (timeDiff == 0 ? 0 : -1) : 1;
         } else if (statDiff < 0) {
           return -1;
@@ -109,7 +111,7 @@ public class OptimizerController extends RestBaseController {
         }
       });
       PageResult<TableOptimizeItem, TableOptimizeInfo> amsPageResult = PageResult.of(arcticTableItemList,
-              offset, pageSize, TableOptimizeItem::buildTableOptimizeInfo);
+          offset, pageSize, TableOptimizeItem::buildTableOptimizeInfo);
       ctx.json(OkResponse.of(amsPageResult));
     } catch (Exception e) {
       LOG.error("Failed to get optimizerGroup tables", e);
@@ -137,12 +139,12 @@ public class OptimizerController extends RestBaseController {
       }
 
       PageResult<Optimizer, Optimizer> amsPageResult = PageResult.of(optimizers,
-              offset, pageSize);
+          offset, pageSize);
       ctx.json(OkResponse.of(amsPageResult));
     } catch (Exception e) {
       LOG.error("Failed to get optimizer", e);
       ctx.json(new ErrorResponse(HttpCode.BAD_REQUEST,
-              "Failed to get optimizer", ""));
+          "Failed to get optimizer", ""));
     }
   }
 
@@ -157,7 +159,7 @@ public class OptimizerController extends RestBaseController {
     } catch (Exception e) {
       LOG.error("Failed to get optimizerGroups", e);
       ctx.json(new ErrorResponse(HttpCode.BAD_REQUEST,
-              "Failed to get optimizerGroups", ""));
+          "Failed to get optimizerGroups", ""));
     }
   }
 
@@ -193,6 +195,7 @@ public class OptimizerController extends RestBaseController {
 
   /**
    * release optimizer.
+   *
    * @pathParam jobId
    */
   public static void releaseOptimizer(Context ctx) {
@@ -226,20 +229,24 @@ public class OptimizerController extends RestBaseController {
       if (containerType.equals("local")) {
         String memory = optimizerGroupInfo.getProperties().get("memory");
         optimizerService.insertOptimizer(optimizerName, optimizerGroupInfo.getId(), optimizerGroupInfo.getName(),
-                TableTaskStatus.STARTING,
-                new SimpleDateFormat("yyyy-MM-dd-HH:mm:ss").format(new Date()),
-                parallelism + 1, Long.parseLong(memory), parallelism, container);
+            TableTaskStatus.STARTING,
+            new SimpleDateFormat("yyyy-MM-dd-HH:mm:ss").format(new Date()),
+            parallelism + 1, Long.parseLong(memory), parallelism, container);
       } else if (containerType.equals("flink")) {
         int tmMemory = Integer.parseInt(optimizerGroupInfo.getProperties().get("taskmanager.memory"));
         int jmMemory = Integer.parseInt(optimizerGroupInfo.getProperties().get("jobmanager.memory"));
         long memory = jmMemory + (long) tmMemory * parallelism;
 
         optimizerService.insertOptimizer(optimizerName,
-                optimizerGroupInfo.getId(),
-                optimizerGroupInfo.getName(),
-                TableTaskStatus.STARTING,
-                new SimpleDateFormat("yyyy-MM-dd-HH:mm:ss").format(new Date()),
-                parallelism + 1, memory, parallelism, container);
+            optimizerGroupInfo.getId(),
+            optimizerGroupInfo.getName(),
+            TableTaskStatus.STARTING,
+            new SimpleDateFormat("yyyy-MM-dd-HH:mm:ss").format(new Date()),
+            parallelism + 1, memory, parallelism, container);
+      } else {
+        ctx.json(new ErrorResponse(HttpCode.BAD_REQUEST, containerType + " type container not support scaleout by ams",
+            ""));
+        return;
       }
 
       String optimizerId = optimizerService.selectOptimizerIdByOptimizerName(optimizerName);
@@ -250,7 +257,7 @@ public class OptimizerController extends RestBaseController {
         LOG.error("Failed to scaleOut optimizer", e);
         optimizerService.deleteOptimizerByName(optimizerName);
         ctx.json(new ErrorResponse(HttpCode.BAD_REQUEST,
-                "Failed to scaleOut optimizer", ""));
+            "Failed to scaleOut optimizer", ""));
         return;
       }
       ctx.json(OkResponse.of("success to scaleOut optimizer"));
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/handler/impl/OptimizeManagerHandler.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/handler/impl/OptimizeManagerHandler.java
index 6bfcb4447..6d187919d 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/handler/impl/OptimizeManagerHandler.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/handler/impl/OptimizeManagerHandler.java
@@ -23,6 +23,8 @@ import com.netease.arctic.ams.api.NoSuchObjectException;
 import com.netease.arctic.ams.api.OptimizeManager;
 import com.netease.arctic.ams.api.OptimizeTask;
 import com.netease.arctic.ams.api.OptimizeTaskStat;
+import com.netease.arctic.ams.api.OptimizerDescriptor;
+import com.netease.arctic.ams.api.OptimizerRegisterInfo;
 import com.netease.arctic.ams.api.OptimizerStateReport;
 import com.netease.arctic.ams.server.service.ServiceContainer;
 import org.apache.thrift.TException;
@@ -49,4 +51,9 @@ public class OptimizeManagerHandler implements OptimizeManager.Iface {
   public void reportOptimizerState(OptimizerStateReport reportData) throws TException {
     ServiceContainer.getOptimizerService().updateOptimizerState(reportData);
   }
+
+  @Override
+  public OptimizerDescriptor registerOptimizer(OptimizerRegisterInfo registerInfo) throws TException {
+    return ServiceContainer.getOptimizerService().registerOptimizer(registerInfo);
+  }
 }
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/OptimizerMapper.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/OptimizerMapper.java
index e75a2870d..7365c07f9 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/OptimizerMapper.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/mapper/OptimizerMapper.java
@@ -93,8 +93,27 @@ public interface OptimizerMapper {
   void deleteOptimizerByName(String optimizerName);
 
   @Select("select optimizer_id, queue_name, queue_id, optimizer_status, core_number, " +
-      "memory, parallelism, container, jobmanager_url, optimizer_instance, optimizer_state_info " +
+      "memory, parallelism, container, jobmanager_url, optimizer_instance, optimizer_state_info ,update_time " +
       "from " + TABLE_NAME + " where optimizer_id = #{optimizerId}")
+  @Results({
+          @Result(property = "jobId", column = "optimizer_id"),
+          @Result(property = "groupName", column = "queue_name"),
+          @Result(property = "queueId", column = "queue_id"),
+          @Result(property = "jobStatus", column = "optimizer_status"),
+          @Result(property = "coreNumber", column = "core_number"),
+          @Result(property = "memory", column = "memory"),
+          @Result(property = "parallelism", column = "parallelism"),
+          @Result(property = "container", column = "container"),
+          @Result(property = "jobmanagerUrl", column = "jobmanager_url"),
+          @Result(property = "instance", column = "optimizer_instance"),
+          @Result(property = "stateInfo", column = "optimizer_state_info", typeHandler = Map2StringConverter.class),
+          @Result(property = "updateTime", column = "update_time")
+  })
+  Optimizer selectOptimizer(@Param("optimizerId") Long optimizerId);
+
+  @Select("select optimizer_id, queue_name, queue_id, optimizer_status, core_number, " +
+      "memory, parallelism, container, jobmanager_url, optimizer_instance, optimizer_state_info ,update_time " +
+      "from " + TABLE_NAME + " where optimizer_name = #{optimizerName}")
   @Results({
       @Result(property = "jobId", column = "optimizer_id"),
       @Result(property = "groupName", column = "queue_name"),
@@ -106,9 +125,11 @@ public interface OptimizerMapper {
       @Result(property = "container", column = "container"),
       @Result(property = "jobmanagerUrl", column = "jobmanager_url"),
       @Result(property = "instance", column = "optimizer_instance"),
-      @Result(property = "stateInfo", column = "optimizer_state_info", typeHandler = Map2StringConverter.class)
+      @Result(property = "stateInfo", column = "optimizer_state_info", typeHandler = Map2StringConverter.class),
+      @Result(property = "updateTime", column = "update_time")
   })
-  Optimizer selectOptimizer(@Param("optimizerId") Long optimizerId);
+  Optimizer selectOptimizerByName(@Param("optimizerName") String optimizerName);
+
 
   @Update("update " + TABLE_NAME + " set" +
       " optimizer_instance = #{instance}, optimizer_state_info = #{state,typeHandler=com.netease.arctic.ams" +
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/Optimizer.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/Optimizer.java
index 380cca423..3094207bd 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/Optimizer.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/model/Optimizer.java
@@ -18,6 +18,7 @@
 
 package com.netease.arctic.ams.server.model;
 
+import com.netease.arctic.ams.api.OptimizerDescriptor;
 import java.sql.Timestamp;
 import java.util.Map;
 
@@ -36,6 +37,13 @@ public class Optimizer {
 
   private String container;
 
+  private String containerType;
+
+  public OptimizerDescriptor convertToDescriptor() {
+    return new OptimizerDescriptor(jobId, Integer.parseInt(queueId), groupName, coreNumber, memory, container,
+        jobStatus.name(), updateTime.getTime());
+  }
+
   public String getGroupName() {
     return groupName;
   }
@@ -131,4 +139,12 @@ public class Optimizer {
   public void setUpdateTime(Timestamp updateTime) {
     this.updateTime = updateTime;
   }
+
+  public String getContainerType() {
+    return containerType;
+  }
+
+  public void setContainerType(String containerType) {
+    this.containerType = containerType;
+  }
 }
\ No newline at end of file
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizeExecuteService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizeExecuteService.java
index 2cb957d3a..46382ba0b 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizeExecuteService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizeExecuteService.java
@@ -22,6 +22,7 @@ import com.alibaba.fastjson.JSONObject;
 import com.netease.arctic.ams.api.NoSuchObjectException;
 import com.netease.arctic.ams.server.ArcticMetaStore;
 import com.netease.arctic.ams.server.config.ArcticMetaStoreConf;
+import com.netease.arctic.ams.server.config.ConfigFileProperties;
 import com.netease.arctic.ams.server.model.Container;
 import com.netease.arctic.ams.server.model.OptimizerGroupInfo;
 import com.netease.arctic.ams.server.model.TableTaskStatus;
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizerService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizerService.java
index e2099962b..42dbc7f74 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizerService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizerService.java
@@ -19,8 +19,12 @@
 package com.netease.arctic.ams.server.service.impl;
 
 import com.alibaba.fastjson.JSONObject;
+import com.netease.arctic.ams.api.InvalidObjectException;
 import com.netease.arctic.ams.api.NoSuchObjectException;
+import com.netease.arctic.ams.api.OptimizerDescriptor;
+import com.netease.arctic.ams.api.OptimizerRegisterInfo;
 import com.netease.arctic.ams.api.OptimizerStateReport;
+import com.netease.arctic.ams.server.config.ConfigFileProperties;
 import com.netease.arctic.ams.server.mapper.OptimizerGroupMapper;
 import com.netease.arctic.ams.server.mapper.OptimizerMapper;
 import com.netease.arctic.ams.server.model.Container;
@@ -37,6 +41,8 @@ import org.apache.ibatis.session.SqlSession;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.text.SimpleDateFormat;
+import java.util.Date;
 import java.util.List;
 import java.util.Map;
 
@@ -48,14 +54,18 @@ public class OptimizerService extends IJDBCService {
   public List<Optimizer> getOptimizers(String optimizerGroup) {
     try (SqlSession sqlSession = getSqlSession(true)) {
       OptimizerMapper optimizerMapper = getMapper(sqlSession, OptimizerMapper.class);
-      return optimizerMapper.selectOptimizersByGroupName(optimizerGroup);
+      List<Optimizer> optimizers = optimizerMapper.selectOptimizersByGroupName(optimizerGroup);
+      optimizers.forEach(this::fillContainerType);
+      return optimizers;
     }
   }
 
   public List<Optimizer> getOptimizers() {
     try (SqlSession sqlSession = getSqlSession(true)) {
       OptimizerMapper optimizerMapper = getMapper(sqlSession, OptimizerMapper.class);
-      return optimizerMapper.selectOptimizers();
+      List<Optimizer> optimizers = optimizerMapper.selectOptimizers();
+      optimizers.forEach(this::fillContainerType);
+      return optimizers;
     }
   }
 
@@ -66,6 +76,13 @@ public class OptimizerService extends IJDBCService {
     }
   }
 
+  public Optimizer getOptimizer(String jobName) {
+    try (SqlSession sqlSession = getSqlSession(true)) {
+      OptimizerMapper optimizerMapper = getMapper(sqlSession, OptimizerMapper.class);
+      return optimizerMapper.selectOptimizerByName(jobName);
+    }
+  }
+
   public void deleteOptimizer(Long optimizerId) {
     try (SqlSession sqlSession = getSqlSession(true)) {
       OptimizerMapper optimizerMapper = getMapper(sqlSession, OptimizerMapper.class);
@@ -89,6 +106,10 @@ public class OptimizerService extends IJDBCService {
     if (optimizer != null) {
       OptimizerGroupInfo optimizerGroupInfo = getOptimizerGroupInfo(optimizer.getGroupName());
       Container container = ServiceContainer.getContainerMetaService().getContainer(optimizerGroupInfo.getContainer());
+      if (container.getType().equals(ConfigFileProperties.EXTERNAL_CONTAINER_TYPE)) {
+        updateOptimizerStatus(reportData.getOptimizerId(), TableTaskStatus.RUNNING);
+        return;
+      }
       OptimizerFactory factory =
           ServiceContainer.getOptimizeExecuteService().findOptimizerFactory(container.getType());
       com.netease.arctic.optimizer.Optimizer instance = factory.deserialize(optimizer.getInstance());
@@ -175,5 +196,28 @@ public class OptimizerService extends IJDBCService {
       optimizerMapper.deleteOptimizerByName(optimizerName);
     }
   }
+
+  public OptimizerDescriptor registerOptimizer(OptimizerRegisterInfo registerInfo) throws InvalidObjectException {
+    OptimizerGroupInfo optimizerGroupInfo = getOptimizerGroupInfo(registerInfo.getOptimizerGroupName());
+    if (optimizerGroupInfo == null) {
+      throw new InvalidObjectException("optimizer group not found");
+    }
+    String currentTime = new SimpleDateFormat("yyyy-MM-dd-HH:mm:ss").format(new Date());
+    String optimizerName = "arctic_optimizer_" + currentTime;
+    insertOptimizer(optimizerName, optimizerGroupInfo.getId(), optimizerGroupInfo.getName(),
+        TableTaskStatus.STARTING, currentTime, registerInfo.getCoreNumber(), registerInfo.getMemorySize(),
+        registerInfo.getCoreNumber(), optimizerGroupInfo.getContainer());
+    return getOptimizer(optimizerName).convertToDescriptor();
+  }
+
+  private Optimizer fillContainerType(Optimizer optimizer) {
+    if (optimizer == null) {
+      return null;
+    }
+    OptimizerGroupInfo optimizerGroupInfo = getOptimizerGroupInfo(optimizer.getGroupName());
+    Container container = ServiceContainer.getContainerMetaService().getContainer(optimizerGroupInfo.getContainer());
+    optimizer.setContainerType(container.getType());
+    return optimizer;
+  }
 }
 
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 c0dad395c..4a368f0d5 100644
--- a/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java
+++ b/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java
@@ -42,9 +42,9 @@ import com.netease.arctic.table.TableMetaStore;
 import com.netease.arctic.table.TableProperties;
 import com.netease.arctic.table.UnkeyedTable;
 import com.netease.arctic.trace.CreateTableTransaction;
+import com.netease.arctic.utils.CatalogUtil;
 import com.netease.arctic.utils.ConvertStructUtil;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
@@ -56,13 +56,11 @@ import org.apache.iceberg.Tables;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.Transactions;
 import org.apache.iceberg.exceptions.NoSuchTableException;
-import org.apache.iceberg.hadoop.HadoopTableOperations;
 import org.apache.iceberg.hadoop.HadoopTables;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.util.PropertyUtil;
 import org.apache.thrift.TException;
-import org.mortbay.log.Log;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -108,8 +106,7 @@ public class BaseArcticCatalog implements ArcticCatalog {
       }
     }
 
-    TableMetaStore.Builder builder = getMetaStoreBuilder(mergeMetaProperties(meta, properties));
-    tableMetaStore = builder.build();
+    tableMetaStore = CatalogUtil.buildMetaStore(meta);
     tables = new HadoopTables(tableMetaStore.getConfiguration());
   }
 
@@ -189,12 +186,13 @@ public class BaseArcticCatalog implements ArcticCatalog {
     ArcticFileIO fileIO = new ArcticHadoopFileIO(tableMetaStore);
     Table baseIcebergTable = tableMetaStore.doAs(() -> tables.load(baseLocation));
     BaseTable baseTable = new BaseKeyedTable.BaseInternalTable(tableIdentifier,
-        useArcticTableOperations(baseIcebergTable, baseLocation, fileIO, tableMetaStore.getConfiguration()),
+        CatalogUtil.useArcticTableOperations(baseIcebergTable, baseLocation, fileIO, tableMetaStore.getConfiguration()),
         fileIO, client);
 
     Table changeIcebergTable = tableMetaStore.doAs(() -> tables.load(changeLocation));
     ChangeTable changeTable = new BaseKeyedTable.ChangeInternalTable(tableIdentifier,
-        useArcticTableOperations(changeIcebergTable, changeLocation, fileIO, tableMetaStore.getConfiguration()),
+        CatalogUtil.useArcticTableOperations(changeIcebergTable, changeLocation, fileIO,
+            tableMetaStore.getConfiguration()),
         fileIO, client);
     return new BaseKeyedTable(tableMeta, tableLocation,
         buildPrimaryKeySpec(baseTable.schema(), tableMeta), client, baseTable, changeTable);
@@ -217,7 +215,7 @@ public class BaseArcticCatalog implements ArcticCatalog {
     String baseLocation = checkLocation(tableMeta, MetaTableProperties.LOCATION_KEY_BASE);
     Table table = tableMetaStore.doAs(() -> tables.load(baseLocation));
     ArcticFileIO arcticFileIO = new ArcticHadoopFileIO(tableMetaStore);
-    return new BaseUnkeyedTable(tableIdentifier, useArcticTableOperations(table, baseLocation,
+    return new BaseUnkeyedTable(tableIdentifier, CatalogUtil.useArcticTableOperations(table, baseLocation,
         arcticFileIO, tableMetaStore.getConfiguration()), arcticFileIO, client);
   }
 
@@ -227,18 +225,6 @@ public class BaseArcticCatalog implements ArcticCatalog {
     return location;
   }
 
-  protected Table useArcticTableOperations(
-      Table table, String tableLocation,
-      ArcticFileIO arcticFileIO, Configuration configuration) {
-    if (table instanceof org.apache.iceberg.BaseTable) {
-      org.apache.iceberg.BaseTable baseTable = (org.apache.iceberg.BaseTable) table;
-      if (baseTable.operations() instanceof HadoopTableOperations) {
-        return new org.apache.iceberg.BaseTable(new ArcticHadoopTableOperations(new Path(tableLocation),
-            arcticFileIO, configuration), table.name());
-      }
-    }
-    return table;
-  }
 
   @Override
   public void renameTable(TableIdentifier from, String newTableName) {
@@ -581,7 +567,8 @@ public class BaseArcticCatalog implements ArcticCatalog {
       });
 
       BaseTable baseTable = new BaseKeyedTable.BaseInternalTable(tableIdentifier,
-          useArcticTableOperations(baseIcebergTable, baseLocation, fileIO, tableMetaStore.getConfiguration()),
+          CatalogUtil.useArcticTableOperations(baseIcebergTable, baseLocation, fileIO,
+              tableMetaStore.getConfiguration()),
           fileIO, client);
 
       Table changeIcebergTable = tableMetaStore.doAs(() -> {
@@ -592,7 +579,8 @@ public class BaseArcticCatalog implements ArcticCatalog {
         }
       });
       ChangeTable changeTable = new BaseKeyedTable.ChangeInternalTable(tableIdentifier,
-          useArcticTableOperations(changeIcebergTable, changeLocation, fileIO, tableMetaStore.getConfiguration()),
+          CatalogUtil.useArcticTableOperations(changeIcebergTable, changeLocation, fileIO,
+              tableMetaStore.getConfiguration()),
           fileIO, client);
       return new BaseKeyedTable(meta, tableLocation,
           primaryKeySpec, client, baseTable, changeTable);
@@ -612,7 +600,7 @@ public class BaseArcticCatalog implements ArcticCatalog {
         }
       });
       ArcticFileIO fileIO = new ArcticHadoopFileIO(tableMetaStore);
-      return new BaseUnkeyedTable(tableIdentifier, useArcticTableOperations(table, baseLocation, fileIO,
+      return new BaseUnkeyedTable(tableIdentifier, CatalogUtil.useArcticTableOperations(table, baseLocation, fileIO,
           tableMetaStore.getConfiguration()), fileIO, client);
     }
 
diff --git a/core/src/main/java/com/netease/arctic/catalog/BaseIcebergCatalog.java b/core/src/main/java/com/netease/arctic/catalog/BaseIcebergCatalog.java
new file mode 100644
index 000000000..636b9e286
--- /dev/null
+++ b/core/src/main/java/com/netease/arctic/catalog/BaseIcebergCatalog.java
@@ -0,0 +1,155 @@
+/*
+ * 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.catalog;
+
+import com.netease.arctic.AmsClient;
+import com.netease.arctic.ams.api.CatalogMeta;
+import com.netease.arctic.io.ArcticFileIO;
+import com.netease.arctic.io.ArcticHadoopFileIO;
+import com.netease.arctic.table.ArcticTable;
+import com.netease.arctic.table.BaseUnkeyedTable;
+import com.netease.arctic.table.TableBuilder;
+import com.netease.arctic.table.TableIdentifier;
+import com.netease.arctic.table.TableMetaStore;
+import com.netease.arctic.utils.CatalogUtil;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SupportsNamespaces;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * A wrapper class around {@link Catalog} and implement {@link ArcticCatalog}.
+ */
+public class BaseIcebergCatalog implements ArcticCatalog {
+
+  private CatalogMeta meta;
+  private transient TableMetaStore tableMetaStore;
+  private transient Catalog icebergCatalog;
+
+  @Override
+  public String name() {
+    return meta.getCatalogName();
+  }
+
+  @Override
+  public void initialize(
+      AmsClient client, CatalogMeta meta, Map<String, String> properties) {
+    this.meta = meta;
+    meta.putToCatalogProperties(org.apache.iceberg.CatalogUtil.ICEBERG_CATALOG_TYPE,
+        meta.getCatalogType());
+    tableMetaStore = CatalogUtil.buildMetaStore(meta);
+    icebergCatalog = org.apache.iceberg.CatalogUtil.buildIcebergCatalog(name(),
+        meta.getCatalogProperties(), tableMetaStore.getConfiguration());
+  }
+
+  @Override
+  public List<String> listDatabases() {
+    if (icebergCatalog instanceof SupportsNamespaces) {
+      return tableMetaStore.doAs(() -> ((SupportsNamespaces)icebergCatalog).listNamespaces(Namespace.empty()).stream()
+          .map(namespace -> namespace.level(0)).distinct().collect(Collectors.toList()));
+    } else {
+      throw new UnsupportedOperationException(String.format("Iceberg catalog: %s do now implement SupportsNamespaces",
+       icebergCatalog.getClass().getName()));
+    }
+  }
+
+  @Override
+  public void createDatabase(String databaseName) {
+    if (icebergCatalog instanceof SupportsNamespaces) {
+      tableMetaStore.doAs(() -> {
+        ((SupportsNamespaces)icebergCatalog).createNamespace(Namespace.of(databaseName));
+        return null;
+      });
+    } else {
+      throw new UnsupportedOperationException(String.format("Iceberg catalog: %s do now implement SupportsNamespaces",
+          icebergCatalog.getClass().getName()));
+    }
+  }
+
+  @Override
+  public void dropDatabase(String databaseName) {
+    if (icebergCatalog instanceof SupportsNamespaces) {
+      tableMetaStore.doAs(() -> {
+        ((SupportsNamespaces)icebergCatalog).dropNamespace(Namespace.of(databaseName));
+        return null;
+      });
+    } else {
+      throw new UnsupportedOperationException(String.format("Iceberg catalog: %s do now implement SupportsNamespaces",
+          icebergCatalog.getClass().getName()));
+    }
+  }
+
+  @Override
+  public List<TableIdentifier> listTables(String database) {
+    return tableMetaStore.doAs(() -> icebergCatalog.listTables(Namespace.of(database)).stream()
+      .filter(tableIdentifier -> tableIdentifier.namespace().levels().length == 1)
+      .map(tableIdentifier -> TableIdentifier.of(name(), database, tableIdentifier.name()))
+      .collect(Collectors.toList()));
+  }
+
+  @Override
+  public ArcticTable loadTable(TableIdentifier tableIdentifier) {
+    Table icebergTable = tableMetaStore.doAs(() -> icebergCatalog
+        .loadTable(toIcebergTableIdentifier(tableIdentifier)));
+    ArcticFileIO arcticFileIO = new ArcticHadoopFileIO(tableMetaStore);
+    return new BaseIcebergTable(tableIdentifier, CatalogUtil.useArcticTableOperations(icebergTable,
+        icebergTable.location(), arcticFileIO, tableMetaStore.getConfiguration()), arcticFileIO);
+  }
+
+  @Override
+  public void renameTable(TableIdentifier from, String newTableName) {
+    tableMetaStore.doAs(() -> {
+      icebergCatalog.renameTable(toIcebergTableIdentifier(from),
+          org.apache.iceberg.catalog.TableIdentifier.of(Namespace.of(from.getDatabase()),
+              newTableName));
+      return null;
+    });
+  }
+
+  @Override
+  public boolean dropTable(TableIdentifier tableIdentifier, boolean purge) {
+    return tableMetaStore.doAs(() -> icebergCatalog.dropTable(toIcebergTableIdentifier(tableIdentifier), purge));
+  }
+
+  @Override
+  public TableBuilder newTableBuilder(
+      TableIdentifier identifier, Schema schema) {
+    throw new UnsupportedOperationException("unsupported new iceberg table for now.");
+  }
+
+  private org.apache.iceberg.catalog.TableIdentifier toIcebergTableIdentifier(TableIdentifier tableIdentifier) {
+    return org.apache.iceberg.catalog.TableIdentifier.of(Namespace.of(tableIdentifier.getDatabase()),
+        tableIdentifier.getTableName());
+  }
+
+  public class BaseIcebergTable extends BaseUnkeyedTable {
+
+    public BaseIcebergTable(
+        TableIdentifier tableIdentifier,
+        Table icebergTable,
+        ArcticFileIO arcticFileIO) {
+      super(tableIdentifier, icebergTable, arcticFileIO);
+    }
+  }
+}
diff --git a/core/src/main/java/com/netease/arctic/catalog/CatalogLoader.java b/core/src/main/java/com/netease/arctic/catalog/CatalogLoader.java
index d00c7c9d5..2774ff1da 100644
--- a/core/src/main/java/com/netease/arctic/catalog/CatalogLoader.java
+++ b/core/src/main/java/com/netease/arctic/catalog/CatalogLoader.java
@@ -25,14 +25,21 @@ import com.netease.arctic.ams.api.CatalogMeta;
 import com.netease.arctic.ams.api.NoSuchObjectException;
 import com.netease.arctic.ams.api.client.AmsClientPools;
 import com.netease.arctic.ams.api.client.ArcticThriftUrl;
+import com.netease.arctic.ams.api.properties.CatalogMetaProperties;
+import com.netease.arctic.ams.api.properties.TableFormat;
+import com.netease.arctic.utils.CatalogUtil;
 import org.apache.iceberg.common.DynConstructors;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.thrift.TException;
 
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 
+import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_AMS;
+import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_CUSTOM;
 import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_HADOOP;
 import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_HIVE;
 
@@ -41,7 +48,8 @@ import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALO
  */
 public class CatalogLoader {
 
-  public static final String HADOOP_CATALOG_IMPL = BaseArcticCatalog.class.getName();
+  public static final String AMS_CATALOG_IMPL = BaseArcticCatalog.class.getName();
+  public static final String ICEBERG_CATALOG_IMPL = BaseIcebergCatalog.class.getName();
   public static final String HIVE_CATALOG_IMPL = "com.netease.arctic.hive.catalog.ArcticHiveCatalog";
 
   /**
@@ -93,18 +101,43 @@ public class CatalogLoader {
     try {
       CatalogMeta catalogMeta = client.getCatalog(catalogName);
       String type = catalogMeta.getCatalogType();
+      CatalogUtil.mergeCatalogProperties(catalogMeta, props);
       String catalogImpl;
+      Set<TableFormat> tableFormats = CatalogUtil.tableFormats(catalogMeta);
+      Preconditions.checkArgument(tableFormats.size() == 1, "Catalog support only one table format now.");
+      TableFormat tableFormat = tableFormats.iterator().next();
       switch (type) {
         case CATALOG_TYPE_HADOOP:
-          catalogImpl = HADOOP_CATALOG_IMPL;
+          Preconditions.checkArgument(tableFormat.equals(TableFormat.ICEBERG),
+              "Hadoop catalog support iceberg table only.");
+          if (catalogMeta.getCatalogProperties().containsKey(CatalogMetaProperties.TABLE_FORMATS)) {
+            catalogImpl = ICEBERG_CATALOG_IMPL;
+          } else {
+            // Compatibility with older versions
+            catalogImpl = AMS_CATALOG_IMPL;
+          }
           break;
         case CATALOG_TYPE_HIVE:
-          catalogImpl = HIVE_CATALOG_IMPL;
+          if (tableFormat.equals(TableFormat.ICEBERG)) {
+            catalogImpl = ICEBERG_CATALOG_IMPL;
+          } else if (tableFormat.equals(TableFormat.HIVE)) {
+            catalogImpl = HIVE_CATALOG_IMPL;
+          } else {
+            throw new IllegalArgumentException("Hive Catalog support iceberg table and hive table only");
+          }
+          break;
+        case CATALOG_TYPE_AMS:
+          Preconditions.checkArgument(tableFormat.equals(TableFormat.ICEBERG),
+              "AMS catalog support iceberg table only.");
+          catalogImpl = AMS_CATALOG_IMPL;
+          break;
+        case CATALOG_TYPE_CUSTOM:
+          Preconditions.checkArgument(tableFormat.equals(TableFormat.ICEBERG),
+              "Custom catalog support iceberg table only.");
+          catalogImpl = ICEBERG_CATALOG_IMPL;
           break;
         default:
-          throw new IllegalStateException(
-              "unsupported catalog type:" + type
-          );
+          throw new IllegalStateException("unsupported catalog type:" + type);
       }
       ArcticCatalog catalog = buildCatalog(catalogImpl);
       catalog.initialize(client, catalogMeta, props);
diff --git a/core/src/main/java/com/netease/arctic/io/reader/DataReaderCommon.java b/core/src/main/java/com/netease/arctic/io/reader/DataReaderCommon.java
index ac5539f60..81ed8b037 100644
--- a/core/src/main/java/com/netease/arctic/io/reader/DataReaderCommon.java
+++ b/core/src/main/java/com/netease/arctic/io/reader/DataReaderCommon.java
@@ -17,7 +17,7 @@ import java.util.function.BiFunction;
 
 public class DataReaderCommon {
 
-  protected static Map<Integer, ?> getIdToConstant(ArcticFileScanTask task, Schema projectedSchema,
+  protected static Map<Integer, ?> getIdToConstant(FileScanTask task, Schema projectedSchema,
       BiFunction<Type, Object, Object> convertConstant) {
     Schema partitionSchema = TypeUtil.select(projectedSchema, task.spec().identitySourceIds());
     Map<Integer, Object> idToConstant = new HashMap<>();
@@ -26,38 +26,28 @@ public class DataReaderCommon {
     }
     idToConstant.put(org.apache.iceberg.MetadataColumns.FILE_PATH.fieldId(),
         convertConstant.apply(Types.StringType.get(), task.file().path().toString()));
-    idToConstant.put(
-        MetadataColumns.TRANSACTION_ID_FILED_ID,
-        convertConstant.apply(Types.LongType.get(), task.file().transactionId()));
 
-    if (task.fileType() == DataFileType.BASE_FILE) {
+    if (task instanceof ArcticFileScanTask) {
+      ArcticFileScanTask arcticFileScanTask = (ArcticFileScanTask)task;
       idToConstant.put(
-          MetadataColumns.FILE_OFFSET_FILED_ID,
-          convertConstant.apply(Types.LongType.get(), Long.MAX_VALUE));
-    }
-    if (task.fileType() == DataFileType.EQ_DELETE_FILE) {
-      idToConstant.put(MetadataColumns.CHANGE_ACTION_ID, convertConstant.apply(
-          Types.StringType.get(),
-          ChangeAction.DELETE.toString()));
-    } else if (task.fileType() == DataFileType.INSERT_FILE) {
-      idToConstant.put(MetadataColumns.CHANGE_ACTION_ID, convertConstant.apply(
-          Types.StringType.get(),
-          ChangeAction.INSERT.toString()));
+          MetadataColumns.TRANSACTION_ID_FILED_ID,
+          convertConstant.apply(Types.LongType.get(), arcticFileScanTask.file().transactionId()));
+
+      if (arcticFileScanTask.fileType() == DataFileType.BASE_FILE) {
+        idToConstant.put(
+            MetadataColumns.FILE_OFFSET_FILED_ID,
+            convertConstant.apply(Types.LongType.get(), Long.MAX_VALUE));
+      }
+      if (arcticFileScanTask.fileType() == DataFileType.EQ_DELETE_FILE) {
+        idToConstant.put(MetadataColumns.CHANGE_ACTION_ID, convertConstant.apply(
+            Types.StringType.get(),
+            ChangeAction.DELETE.toString()));
+      } else if (arcticFileScanTask.fileType() == DataFileType.INSERT_FILE) {
+        idToConstant.put(MetadataColumns.CHANGE_ACTION_ID, convertConstant.apply(
+            Types.StringType.get(),
+            ChangeAction.INSERT.toString()));
+      }
     }
     return idToConstant;
   }
-
-  protected static Map<Integer, ?> getIdToConstant(FileScanTask task, Schema projectedSchema,
-                                                   BiFunction<Type, Object, Object> convertConstant) {
-    Schema partitionSchema = TypeUtil.select(projectedSchema, task.spec().identitySourceIds());
-    Map<Integer, Object> idToConstant = new HashMap<>();
-    if (!partitionSchema.columns().isEmpty()) {
-      idToConstant.putAll(PartitionUtil.constantsMap(task, convertConstant));
-    }
-    idToConstant.put(org.apache.iceberg.MetadataColumns.FILE_PATH.fieldId(),
-        convertConstant.apply(Types.StringType.get(), task.file().path().toString()));
-
-    return idToConstant;
-  }
-
 }
diff --git a/core/src/main/java/com/netease/arctic/io/reader/GenericIcebergDataReader.java b/core/src/main/java/com/netease/arctic/io/reader/GenericIcebergDataReader.java
new file mode 100644
index 000000000..4403b8dd6
--- /dev/null
+++ b/core/src/main/java/com/netease/arctic/io/reader/GenericIcebergDataReader.java
@@ -0,0 +1,82 @@
+/*
+ * 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.io.reader;
+
+import com.netease.arctic.data.DataTreeNode;
+import com.netease.arctic.iceberg.optimize.InternalRecordWrapper;
+import com.netease.arctic.io.ArcticFileIO;
+import com.netease.arctic.table.PrimaryKeySpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetReaders;
+import org.apache.iceberg.parquet.ParquetValueReader;
+import org.apache.iceberg.types.Type;
+import org.apache.parquet.schema.MessageType;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+
+public class GenericIcebergDataReader extends BaseIcebergDataReader<Record> {
+  public GenericIcebergDataReader(
+      ArcticFileIO fileIO,
+      Schema tableSchema,
+      Schema projectedSchema,
+      String nameMapping,
+      boolean caseSensitive,
+      BiFunction<Type, Object, Object> convertConstant,
+      boolean reuseContainer) {
+    super(fileIO, tableSchema, projectedSchema, nameMapping, caseSensitive, convertConstant, reuseContainer);
+  }
+
+  public GenericIcebergDataReader(
+      ArcticFileIO fileIO,
+      Schema tableSchema,
+      Schema projectedSchema,
+      PrimaryKeySpec primaryKeySpec,
+      String nameMapping,
+      boolean caseSensitive,
+      BiFunction<Type, Object, Object> convertConstant,
+      Set<DataTreeNode> sourceNodes,
+      boolean reuseContainer) {
+    super(
+        fileIO,
+        tableSchema,
+        projectedSchema,
+        primaryKeySpec,
+        nameMapping,
+        caseSensitive,
+        convertConstant,
+        sourceNodes,
+        reuseContainer);
+  }
+
+  @Override
+  protected Function<MessageType, ParquetValueReader<?>> getNewReaderFunction(
+      Schema projectedSchema, Map<Integer, ?> idToConstant) {
+    return fileSchema -> GenericParquetReaders.buildReader(projectedSchema, fileSchema, idToConstant);
+  }
+
+  @Override
+  protected Function<Schema, Function<Record, StructLike>> toStructLikeFunction() {
+    return schema -> record -> new InternalRecordWrapper(schema.asStruct()).wrap(record);
+  }
+}
diff --git a/core/src/main/java/com/netease/arctic/op/ArcticTableOperations.java b/core/src/main/java/com/netease/arctic/op/ArcticTableOperations.java
new file mode 100644
index 000000000..27993a494
--- /dev/null
+++ b/core/src/main/java/com/netease/arctic/op/ArcticTableOperations.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.op;
+
+import com.netease.arctic.io.ArcticFileIO;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
+
+/**
+ * A wrapper class around {@link TableOperations}. Add authentication for all methods.
+ */
+public class ArcticTableOperations implements TableOperations {
+
+  private final TableOperations ops;
+  private final ArcticFileIO arcticFileIO;
+
+  public ArcticTableOperations(TableOperations ops, ArcticFileIO arcticFileIO) {
+    this.ops = ops;
+    this.arcticFileIO = arcticFileIO;
+  }
+
+  @Override
+  public TableMetadata current() {
+    return arcticFileIO.doAs(() -> ops.current());
+  }
+
+  @Override
+  public TableMetadata refresh() {
+    return arcticFileIO.doAs(() -> ops.refresh());
+  }
+
+  @Override
+  public void commit(TableMetadata base, TableMetadata metadata) {
+    arcticFileIO.doAs(() -> {
+      ops.commit(base, metadata);
+      return null;
+    });
+  }
+
+  @Override
+  public FileIO io() {
+    return arcticFileIO;
+  }
+
+  @Override
+  public String metadataFileLocation(String fileName) {
+    return ops.metadataFileLocation(fileName);
+  }
+
+  @Override
+  public LocationProvider locationProvider() {
+    return ops.locationProvider();
+  }
+
+  @Override
+  public TableOperations temp(TableMetadata uncommittedMetadata) {
+    return ops.temp(uncommittedMetadata);
+  }
+}
diff --git a/core/src/main/java/com/netease/arctic/table/TableMetaStore.java b/core/src/main/java/com/netease/arctic/table/TableMetaStore.java
index fb3ce3ed2..7a6d52b1c 100644
--- a/core/src/main/java/com/netease/arctic/table/TableMetaStore.java
+++ b/core/src/main/java/com/netease/arctic/table/TableMetaStore.java
@@ -381,6 +381,9 @@ public class TableMetaStore implements Serializable {
     Configuration configuration = new Configuration();
     configuration.addResource(new ByteArrayInputStream(metaStore.getCoreSite()));
     configuration.addResource(new ByteArrayInputStream(metaStore.getHdfsSite()));
+    if (!ArrayUtils.isEmpty(metaStore.getMetaStoreSite())) {
+      configuration.addResource(new ByteArrayInputStream(metaStore.getMetaStoreSite()));
+    }
     configuration.set(CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_KEY, "true");
     //Enforce configuration resolve resources
     configuration.get(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY);
diff --git a/core/src/main/java/com/netease/arctic/utils/CatalogUtil.java b/core/src/main/java/com/netease/arctic/utils/CatalogUtil.java
new file mode 100644
index 000000000..afe961493
--- /dev/null
+++ b/core/src/main/java/com/netease/arctic/utils/CatalogUtil.java
@@ -0,0 +1,167 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.utils;
+
+import com.netease.arctic.ams.api.CatalogMeta;
+import com.netease.arctic.ams.api.properties.CatalogMetaProperties;
+import com.netease.arctic.ams.api.properties.TableFormat;
+import com.netease.arctic.io.ArcticFileIO;
+import com.netease.arctic.op.ArcticHadoopTableOperations;
+import com.netease.arctic.op.ArcticTableOperations;
+import com.netease.arctic.table.TableMetaStore;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.curator.shaded.com.google.common.collect.Maps;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.hadoop.HadoopTableOperations;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.PropertyUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_AMS;
+import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_CUSTOM;
+import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_HADOOP;
+import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_HIVE;
+
+public class CatalogUtil {
+
+  private static final Logger LOG = LoggerFactory.getLogger(CatalogUtil.class);
+
+  /**
+   * Return table format set catalog supported.
+   */
+  public static Set<TableFormat> tableFormats(CatalogMeta meta) {
+    if (meta.getCatalogProperties().containsKey(CatalogMetaProperties.TABLE_FORMATS)) {
+      String tableFormatsProperty = meta.getCatalogProperties().get(CatalogMetaProperties.TABLE_FORMATS);
+      return Arrays.stream(tableFormatsProperty.split(",")).map(tableFormatString
+          -> TableFormat.valueOf(tableFormatString.trim().toUpperCase(Locale.ROOT))).collect(Collectors.toSet());
+    } else {
+      // Generate table format from catalog type for compatibility with older versions
+      switch (meta.getCatalogType()) {
+        case CATALOG_TYPE_AMS:
+        case CATALOG_TYPE_CUSTOM:
+        case CATALOG_TYPE_HADOOP:
+          return Sets.newHashSet(TableFormat.ICEBERG);
+        case CATALOG_TYPE_HIVE:
+          return Sets.newHashSet(TableFormat.HIVE);
+        default:
+          throw new IllegalArgumentException("Unsupported catalog type:" +  meta.getCatalogType());
+      }
+    }
+  }
+
+  /**
+   * Merge catalog properties in client side into catalog meta.
+   */
+  public static void mergeCatalogProperties(CatalogMeta meta, Map<String, String> properties) {
+    if (meta.getCatalogProperties() == null) {
+      meta.setCatalogProperties(Maps.newHashMap());
+    }
+    if (properties != null) {
+      properties.forEach(meta::putToCatalogProperties);
+    }
+  }
+
+  /**
+   * Build {@link TableMetaStore} from catalog meta.
+   */
+  public static TableMetaStore buildMetaStore(CatalogMeta catalogMeta) {
+    // load storage configs
+    TableMetaStore.Builder builder = TableMetaStore.builder();
+    if (catalogMeta.getStorageConfigs() != null) {
+      Map<String, String> storageConfigs = catalogMeta.getStorageConfigs();
+      if (CatalogMetaProperties.STORAGE_CONFIGS_VALUE_TYPE_HDFS
+          .equalsIgnoreCase(
+              storageConfigs.get(CatalogMetaProperties.STORAGE_CONFIGS_KEY_TYPE))) {
+        String coreSite = storageConfigs.get(CatalogMetaProperties.STORAGE_CONFIGS_KEY_CORE_SITE);
+        String hdfsSite = storageConfigs.get(CatalogMetaProperties.STORAGE_CONFIGS_KEY_HDFS_SITE);
+        String hiveSite = storageConfigs.get(CatalogMetaProperties.STORAGE_CONFIGS_KEY_HIVE_SITE);
+        builder.withBase64CoreSite(coreSite)
+            .withBase64MetaStoreSite(hiveSite)
+            .withBase64HdfsSite(hdfsSite);
+      }
+    }
+
+    boolean loadAuthFromAMS = PropertyUtil.propertyAsBoolean(catalogMeta.getCatalogProperties(),
+        CatalogMetaProperties.LOAD_AUTH_FROM_AMS, CatalogMetaProperties.LOAD_AUTH_FROM_AMS_DEFAULT);
+    // load auth configs from ams
+    if (loadAuthFromAMS) {
+      if (catalogMeta.getAuthConfigs() != null) {
+        Map<String, String> authConfigs = catalogMeta.getAuthConfigs();
+        String authType = authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_TYPE);
+        LOG.info("TableMetaStore use auth config in catalog meta, authType is {}", authType);
+        if (CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_SIMPLE.equalsIgnoreCase(authType)) {
+          String hadoopUsername = authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_HADOOP_USERNAME);
+          builder.withSimpleAuth(hadoopUsername);
+        } else if (CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_KERBEROS.equalsIgnoreCase(authType)) {
+          String krb5 = authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_KRB5);
+          String keytab = authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_KEYTAB);
+          String principal = authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_PRINCIPAL);
+          builder.withBase64KrbAuth(keytab, krb5, principal);
+        }
+      }
+    }
+
+    // cover auth configs from ams with auth configs in properties
+    String authType = catalogMeta.getCatalogProperties().get(CatalogMetaProperties.AUTH_CONFIGS_KEY_TYPE);
+    if (StringUtils.isNotEmpty(authType)) {
+      LOG.info("TableMetaStore use auth config in properties, authType is {}", authType);
+      if (CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_SIMPLE.equalsIgnoreCase(authType)) {
+        String hadoopUsername = catalogMeta.getCatalogProperties()
+            .get(CatalogMetaProperties.AUTH_CONFIGS_KEY_HADOOP_USERNAME);
+        builder.withSimpleAuth(hadoopUsername);
+      } else if (CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_KERBEROS.equalsIgnoreCase(authType)) {
+        String krb5 = catalogMeta.getCatalogProperties().get(CatalogMetaProperties.AUTH_CONFIGS_KEY_KRB5);
+        String keytab = catalogMeta.getCatalogProperties().get(CatalogMetaProperties.AUTH_CONFIGS_KEY_KEYTAB);
+        String principal = catalogMeta.getCatalogProperties().get(CatalogMetaProperties.AUTH_CONFIGS_KEY_PRINCIPAL);
+        builder.withBase64KrbAuth(keytab, krb5, principal);
+      }
+    }
+
+    return builder.build();
+  }
+
+  /**
+   * Wrap table operation with arctic authorization logic for {@link Table}.
+   */
+  public static Table useArcticTableOperations(
+      Table table, String tableLocation,
+      ArcticFileIO arcticFileIO, Configuration configuration) {
+    if (table instanceof org.apache.iceberg.BaseTable) {
+      org.apache.iceberg.BaseTable baseTable = (org.apache.iceberg.BaseTable) table;
+      if (baseTable.operations() instanceof HadoopTableOperations) {
+        return new org.apache.iceberg.BaseTable(new ArcticHadoopTableOperations(new Path(tableLocation),
+            arcticFileIO, configuration), table.name());
+      } else {
+        return new org.apache.iceberg.BaseTable(new ArcticTableOperations(((BaseTable) table).operations(),
+            arcticFileIO), table.name());
+      }
+    }
+    return table;
+  }
+}
diff --git a/hive/src/main/java/com/netease/arctic/hive/HMSClientImpl.java b/hive/src/main/java/com/netease/arctic/hive/HMSClientImpl.java
index bc2fe3647..92543d4bd 100644
--- a/hive/src/main/java/com/netease/arctic/hive/HMSClientImpl.java
+++ b/hive/src/main/java/com/netease/arctic/hive/HMSClientImpl.java
@@ -18,8 +18,6 @@
 
 package com.netease.arctic.hive;
 
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
 import org.apache.hadoop.hive.metastore.PartitionDropOptions;
 import org.apache.hadoop.hive.metastore.api.Database;
@@ -27,7 +25,6 @@ import org.apache.hadoop.hive.metastore.api.EnvironmentContext;
 import org.apache.hadoop.hive.metastore.api.MetaException;
 import org.apache.hadoop.hive.metastore.api.Partition;
 import org.apache.hadoop.hive.metastore.api.Table;
-import org.apache.iceberg.common.DynConstructors;
 import org.apache.iceberg.common.DynMethods;
 import org.apache.thrift.TException;
 
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 870ff54a2..8b4f01596 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
@@ -33,12 +33,12 @@ import com.netease.arctic.hive.utils.HiveSchemaUtil;
 import com.netease.arctic.hive.utils.HiveTableUtil;
 import com.netease.arctic.io.ArcticFileIO;
 import com.netease.arctic.io.ArcticHadoopFileIO;
-import com.netease.arctic.table.BaseKeyedTable;
 import com.netease.arctic.table.ChangeTable;
 import com.netease.arctic.table.PrimaryKeySpec;
 import com.netease.arctic.table.TableBuilder;
 import com.netease.arctic.table.TableIdentifier;
 import com.netease.arctic.table.TableProperties;
+import com.netease.arctic.utils.CatalogUtil;
 import org.apache.hadoop.hive.metastore.TableType;
 import org.apache.hadoop.hive.metastore.api.Database;
 import org.apache.iceberg.FileFormat;
@@ -160,12 +160,13 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
     ArcticFileIO fileIO = new ArcticHadoopFileIO(tableMetaStore);
     Table baseIcebergTable = tableMetaStore.doAs(() -> tables.load(baseLocation));
     UnkeyedHiveTable baseTable = new UnkeyedHiveTable(tableIdentifier,
-        useArcticTableOperations(baseIcebergTable, baseLocation, fileIO, tableMetaStore.getConfiguration()),
+        CatalogUtil.useArcticTableOperations(baseIcebergTable, baseLocation, fileIO, tableMetaStore.getConfiguration()),
         fileIO, tableLocation, client, hiveClientPool, false);
 
     Table changeIcebergTable = tableMetaStore.doAs(() -> tables.load(changeLocation));
     ChangeTable changeTable = new KeyedHiveTable.HiveChangeInternalTable(tableIdentifier,
-        useArcticTableOperations(changeIcebergTable, changeLocation, fileIO, tableMetaStore.getConfiguration()),
+        CatalogUtil.useArcticTableOperations(changeIcebergTable, changeLocation, fileIO,
+            tableMetaStore.getConfiguration()),
         fileIO, client);
     return new KeyedHiveTable(tableMeta, tableLocation,
         buildPrimaryKeySpec(baseTable.schema(), tableMeta), client, hiveClientPool, baseTable, changeTable);
@@ -178,7 +179,7 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
     String tableLocation = checkLocation(tableMeta, MetaTableProperties.LOCATION_KEY_TABLE);
     Table table = tableMetaStore.doAs(() -> tables.load(baseLocation));
     ArcticFileIO arcticFileIO = new ArcticHadoopFileIO(tableMetaStore);
-    return new UnkeyedHiveTable(tableIdentifier, useArcticTableOperations(table, baseLocation,
+    return new UnkeyedHiveTable(tableIdentifier, CatalogUtil.useArcticTableOperations(table, baseLocation,
         arcticFileIO, tableMetaStore.getConfiguration()), arcticFileIO, tableLocation, client, hiveClientPool);
   }
 
@@ -290,7 +291,8 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
         }
       });
       UnkeyedHiveTable baseTable = new UnkeyedHiveTable(tableIdentifier,
-          useArcticTableOperations(baseIcebergTable, baseLocation, fileIO, tableMetaStore.getConfiguration()),
+          CatalogUtil.useArcticTableOperations(baseIcebergTable, baseLocation, fileIO,
+              tableMetaStore.getConfiguration()),
           fileIO, tableLocation, client, hiveClientPool, false);
 
       Table changeIcebergTable = tableMetaStore.doAs(() -> {
@@ -304,7 +306,8 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
         }
       });
       ChangeTable changeTable = new KeyedHiveTable.HiveChangeInternalTable(tableIdentifier,
-          useArcticTableOperations(changeIcebergTable, changeLocation, fileIO, tableMetaStore.getConfiguration()),
+          CatalogUtil.useArcticTableOperations(changeIcebergTable, changeLocation, fileIO,
+              tableMetaStore.getConfiguration()),
           fileIO, client);
 
       Map<String, String> metaProperties = meta.properties;
@@ -376,7 +379,7 @@ public class ArcticHiveCatalog extends BaseArcticCatalog {
         throw new RuntimeException("Failed to create hive table:" + meta.getTableIdentifier(), e);
       }
       ArcticFileIO fileIO = new ArcticHadoopFileIO(tableMetaStore);
-      return new UnkeyedHiveTable(tableIdentifier, useArcticTableOperations(table, baseLocation, fileIO,
+      return new UnkeyedHiveTable(tableIdentifier, CatalogUtil.useArcticTableOperations(table, baseLocation, fileIO,
           tableMetaStore.getConfiguration()), fileIO, tableLocation, client, hiveClientPool);
     }
 
diff --git a/hive/src/main/java/org/apache/iceberg/data/AdaptHiveGenericAppenderFactory.java b/hive/src/main/java/org/apache/iceberg/data/AdaptHiveGenericAppenderFactory.java
index f34e426ac..cc26ae653 100644
--- a/hive/src/main/java/org/apache/iceberg/data/AdaptHiveGenericAppenderFactory.java
+++ b/hive/src/main/java/org/apache/iceberg/data/AdaptHiveGenericAppenderFactory.java
@@ -97,7 +97,7 @@ public class AdaptHiveGenericAppenderFactory implements FileAppenderFactory<Reco
               .setAll(config)
               .overwrite()
               .build();
-
+        //Change For Arctic ⬇
         case PARQUET:
           return AdaptHiveParquet.write(outputFile)
               .schema(schema)
@@ -106,7 +106,7 @@ public class AdaptHiveGenericAppenderFactory implements FileAppenderFactory<Reco
               .metricsConfig(metricsConfig)
               .overwrite()
               .build();
-
+        //Change For Arctic ⬆
         case ORC:
           return ORC.write(outputFile)
               .schema(schema)
diff --git a/hive/src/main/java/org/apache/iceberg/data/parquet/AdaptHiveBaseParquetReaders.java b/hive/src/main/java/org/apache/iceberg/data/parquet/AdaptHiveBaseParquetReaders.java
index daa66a9b0..2bf5472ce 100644
--- a/hive/src/main/java/org/apache/iceberg/data/parquet/AdaptHiveBaseParquetReaders.java
+++ b/hive/src/main/java/org/apache/iceberg/data/parquet/AdaptHiveBaseParquetReaders.java
@@ -310,6 +310,7 @@ public abstract class AdaptHiveBaseParquetReaders<T> {
         case INT64:
         case DOUBLE:
           return new ParquetValueReaders.UnboxedReader<>(desc);
+        //Change For Arctic ⬇
         case INT96:
           // Impala & Spark used to write timestamps as INT96 without a logical type. For backwards
           // compatibility we try to read INT96 as timestamps.
@@ -319,6 +320,7 @@ public abstract class AdaptHiveBaseParquetReaders<T> {
           } else {
             return new TimestampIntWithOutTZ96Reader(desc);
           }
+        //Change For Arctic ⬆
         default:
           throw new UnsupportedOperationException("Unsupported type: " + primitive);
       }
@@ -365,6 +367,7 @@ public abstract class AdaptHiveBaseParquetReaders<T> {
     }
   }
 
+  //Change For Arctic ⬇
   private static class TimestampIntWithOutTZ96Reader extends ParquetValueReaders.PrimitiveReader<LocalDateTime> {
     private static final long UNIX_EPOCH_JULIAN = 2_440_588L;
 
@@ -403,6 +406,7 @@ public abstract class AdaptHiveBaseParquetReaders<T> {
           .plusNanos(timeOfDayNanos).atOffset(ZoneOffset.UTC);
     }
   }
+  //Change For Arctic ⬆
 
   private static class TimestamptzReader extends ParquetValueReaders.PrimitiveReader<OffsetDateTime> {
     private TimestamptzReader(ColumnDescriptor desc) {
diff --git a/hive/src/main/java/org/apache/iceberg/data/parquet/AdaptHiveBaseParquetWriter.java b/hive/src/main/java/org/apache/iceberg/data/parquet/AdaptHiveBaseParquetWriter.java
index c94fa76d8..3d4c44ccb 100644
--- a/hive/src/main/java/org/apache/iceberg/data/parquet/AdaptHiveBaseParquetWriter.java
+++ b/hive/src/main/java/org/apache/iceberg/data/parquet/AdaptHiveBaseParquetWriter.java
@@ -149,8 +149,10 @@ public abstract class AdaptHiveBaseParquetWriter<T> {
           return ParquetValueWriters.floats(desc);
         case DOUBLE:
           return ParquetValueWriters.doubles(desc);
+        //Change For Arctic ⬇
         case INT96:
           return new TimestampInt96Writer(desc);
+        //Change For Arctic ⬆
         default:
           throw new UnsupportedOperationException("Unsupported type: " + primitive);
       }
@@ -305,6 +307,7 @@ public abstract class AdaptHiveBaseParquetWriter<T> {
     }
   }
 
+  //Change For Arctic ⬇
   private static class TimestampInt96Writer<T> extends AdaptHivePrimitiveWriter<T> {
 
     private static final long JULIAN_DAY_OF_EPOCH = 2440588L;
@@ -371,4 +374,5 @@ public abstract class AdaptHiveBaseParquetWriter<T> {
       }
     }
   }
+  //Change For Arctic ⬆
 }
diff --git a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveColumnWriter.java b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveColumnWriter.java
index 98257ef77..ba539e3aa 100644
--- a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveColumnWriter.java
+++ b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveColumnWriter.java
@@ -47,6 +47,7 @@ public abstract class AdaptHiveColumnWriter<T> implements TripleWriter<T> {
             writeLong(rl, value);
           }
         };
+      //Change For Arctic
       case INT96:
         return (AdaptHiveColumnWriter<T>) new AdaptHiveColumnWriter<Binary>(desc) {
           @Override
@@ -54,6 +55,7 @@ public abstract class AdaptHiveColumnWriter<T> implements TripleWriter<T> {
             writeBinary(rl, value);
           }
         };
+      //Change For Arctic
       case FLOAT:
         return (AdaptHiveColumnWriter<T>) new AdaptHiveColumnWriter<Float>(desc) {
           @Override
diff --git a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquet.java b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquet.java
index 777fb4103..b2afdcc61 100644
--- a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquet.java
+++ b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquet.java
@@ -265,9 +265,11 @@ public class AdaptHiveParquet {
             .withDictionaryPageSize(dictionaryPageSize)
             .build();
 
+        //Change For Arctic
         return new org.apache.iceberg.parquet.AdaptHiveParquetWriter<>(
             conf, file, schema, rowGroupSize, metadata, createWriterFunc, codec,
             parquetProperties, metricsConfig, writeMode);
+        //Change For Arctic
       } else {
         return new ParquetWriteAdapter<>(new ParquetWriteBuilder<D>(ParquetIO.file(file))
             .withWriterVersion(writerVersion)
@@ -810,8 +812,10 @@ public class AdaptHiveParquet {
           return new VectorizedParquetReader<>(file, schema, options, batchedReaderFunc, nameMapping, filter,
               reuseContainers, caseSensitive, maxRecordsPerBatch);
         } else {
-          return new org.apache.iceberg.parquet.ParquetReader<>(
+          //Change For Arctic
+          return new AdaptHiveParquetReader<>(
               file, schema, options, readerFunc, nameMapping, filter, reuseContainers, caseSensitive);
+          //Change For Arctic
         }
       }
 
diff --git a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetConversions.java b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetConversions.java
new file mode 100644
index 000000000..770c915ed
--- /dev/null
+++ b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetConversions.java
@@ -0,0 +1,150 @@
+/*
+ * 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 org.apache.iceberg.parquet;
+
+import org.apache.iceberg.expressions.Literal;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import static java.time.Instant.EPOCH;
+
+/**
+ * Copy from iceberg {@link org.apache.iceberg.parquet.ParquetConversions} to resolve int96 type.
+ */
+class AdaptHiveParquetConversions {
+  private AdaptHiveParquetConversions() {
+  }
+
+  @SuppressWarnings("unchecked")
+  static <T> Literal<T> fromParquetPrimitive(Type type, PrimitiveType parquetType, Object value) {
+    switch (type.typeId()) {
+      case BOOLEAN:
+        return (Literal<T>) Literal.of((Boolean) value);
+      case INTEGER:
+      case DATE:
+        return (Literal<T>) Literal.of((Integer) value);
+      case LONG:
+      case TIME:
+      case TIMESTAMP:
+        return (Literal<T>) Literal.of((Long) value);
+      case FLOAT:
+        return (Literal<T>) Literal.of((Float) value);
+      case DOUBLE:
+        return (Literal<T>) Literal.of((Double) value);
+      case STRING:
+        Function<Object, Object> stringConversion = converterFromParquet(parquetType);
+        return (Literal<T>) Literal.of((CharSequence) stringConversion.apply(value));
+      case UUID:
+        Function<Object, Object> uuidConversion = converterFromParquet(parquetType);
+        return (Literal<T>) Literal.of((UUID) uuidConversion.apply(value));
+      case FIXED:
+      case BINARY:
+        Function<Object, Object> binaryConversion = converterFromParquet(parquetType);
+        return (Literal<T>) Literal.of((ByteBuffer) binaryConversion.apply(value));
+      case DECIMAL:
+        Function<Object, Object> decimalConversion = converterFromParquet(parquetType);
+        return (Literal<T>) Literal.of((BigDecimal) decimalConversion.apply(value));
+      default:
+        throw new IllegalArgumentException("Unsupported primitive type: " + type);
+    }
+  }
+
+  static Function<Object, Object> converterFromParquet(PrimitiveType parquetType, Type icebergType) {
+    Function<Object, Object> fromParquet = converterFromParquet(parquetType);
+
+    //Change For Arctic:Adapt int 96
+    if (parquetType.getPrimitiveTypeName() == PrimitiveType.PrimitiveTypeName.INT96) {
+      return binary -> {
+        final ByteBuffer byteBuffer = ByteBuffer.wrap(((Binary) binary).getBytes()).order(ByteOrder.LITTLE_ENDIAN);
+        final long timeOfDayNanos = byteBuffer.getLong();
+        final int julianDay = byteBuffer.getInt();
+        Instant instant = Instant
+            .ofEpochMilli(TimeUnit.DAYS.toMillis(julianDay - 2_440_588L))
+            .plusNanos(timeOfDayNanos);
+
+        if (!((Types.TimestampType) icebergType).shouldAdjustToUTC()) {
+          //iceberg org.apache.iceberg.expressions.Literals resolve timestamp without tz use UTC, but in fact it is
+          // local time zone, so the Literals time will plus 8 hours
+          instant = instant.plus(8, ChronoUnit.HOURS);
+        }
+        return ChronoUnit.MICROS.between(EPOCH, instant);
+      };
+    }
+    //Change For Arctic
+
+    if (icebergType != null) {
+      if (icebergType.typeId() == Type.TypeID.LONG &&
+          parquetType.getPrimitiveTypeName() == PrimitiveType.PrimitiveTypeName.INT32) {
+        return value -> ((Integer) fromParquet.apply(value)).longValue();
+      } else if (icebergType.typeId() == Type.TypeID.DOUBLE &&
+          parquetType.getPrimitiveTypeName() == PrimitiveType.PrimitiveTypeName.FLOAT) {
+        return value -> ((Float) fromParquet.apply(value)).doubleValue();
+      }
+    }
+
+    return fromParquet;
+  }
+
+  static Function<Object, Object> converterFromParquet(PrimitiveType type) {
+    if (type.getOriginalType() != null) {
+      switch (type.getOriginalType()) {
+        case UTF8:
+          // decode to CharSequence to avoid copying into a new String
+          return binary -> StandardCharsets.UTF_8.decode(((Binary) binary).toByteBuffer());
+        case DECIMAL:
+          int scale = type.getDecimalMetadata().getScale();
+          switch (type.getPrimitiveTypeName()) {
+            case INT32:
+            case INT64:
+              return num -> BigDecimal.valueOf(((Number) num).longValue(), scale);
+            case FIXED_LEN_BYTE_ARRAY:
+            case BINARY:
+              return bin -> new BigDecimal(new BigInteger(((Binary) bin).getBytes()), scale);
+            default:
+              throw new IllegalArgumentException(
+                  "Unsupported primitive type for decimal: " + type.getPrimitiveTypeName());
+          }
+        default:
+      }
+    }
+
+    switch (type.getPrimitiveTypeName()) {
+      case FIXED_LEN_BYTE_ARRAY:
+      case BINARY:
+        return binary -> ByteBuffer.wrap(((Binary) binary).getBytes());
+      default:
+    }
+
+    return obj -> obj;
+  }
+}
diff --git a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetDictionaryRowGroupFilter.java b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetDictionaryRowGroupFilter.java
new file mode 100644
index 000000000..842ebdd94
--- /dev/null
+++ b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetDictionaryRowGroupFilter.java
@@ -0,0 +1,445 @@
+/*
+ * 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 org.apache.iceberg.parquet;
+
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.expressions.Binder;
+import org.apache.iceberg.expressions.BoundReference;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.ExpressionVisitors;
+import org.apache.iceberg.expressions.ExpressionVisitors.BoundExpressionVisitor;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.Literal;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types.StructType;
+import org.apache.iceberg.util.NaNUtil;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.Dictionary;
+import org.apache.parquet.column.page.DictionaryPage;
+import org.apache.parquet.column.page.DictionaryPageReadStore;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.io.IOException;
+import java.util.Comparator;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+
+/**
+ * Copy from iceberg {@link org.apache.iceberg.parquet.ParquetDictionaryRowGroupFilter} to resolve int96 type.
+ */
+public class AdaptHiveParquetDictionaryRowGroupFilter {
+  private final Schema schema;
+  private final Expression expr;
+
+  public AdaptHiveParquetDictionaryRowGroupFilter(Schema schema, Expression unbound) {
+    this(schema, unbound, true);
+  }
+
+  public AdaptHiveParquetDictionaryRowGroupFilter(Schema schema, Expression unbound, boolean caseSensitive) {
+    this.schema = schema;
+    StructType struct = schema.asStruct();
+    this.expr = Binder.bind(struct, Expressions.rewriteNot(unbound), caseSensitive);
+  }
+
+  /**
+   * Test whether the dictionaries for a row group may contain records that match the expression.
+   *
+   * @param fileSchema schema for the Parquet file
+   * @param dictionaries a dictionary page read store
+   * @return false if the file cannot contain rows that match the expression, true otherwise.
+   */
+  public boolean shouldRead(MessageType fileSchema, BlockMetaData rowGroup,
+      DictionaryPageReadStore dictionaries) {
+    return new EvalVisitor().eval(fileSchema, rowGroup, dictionaries);
+  }
+
+  private static final boolean ROWS_MIGHT_MATCH = true;
+  private static final boolean ROWS_CANNOT_MATCH = false;
+
+  private class EvalVisitor extends BoundExpressionVisitor<Boolean> {
+    private DictionaryPageReadStore dictionaries = null;
+    private Map<Integer, Set<?>> dictCache = null;
+    private Map<Integer, Boolean> isFallback = null;
+    private Map<Integer, Boolean> mayContainNulls = null;
+    private Map<Integer, ColumnDescriptor> cols = null;
+    private Map<Integer, Function<Object, Object>> conversions = null;
+
+    private boolean eval(MessageType fileSchema, BlockMetaData rowGroup,
+        DictionaryPageReadStore dictionaryReadStore) {
+      this.dictionaries = dictionaryReadStore;
+      this.dictCache = Maps.newHashMap();
+      this.isFallback = Maps.newHashMap();
+      this.mayContainNulls = Maps.newHashMap();
+      this.cols = Maps.newHashMap();
+      this.conversions = Maps.newHashMap();
+
+      for (ColumnDescriptor desc : fileSchema.getColumns()) {
+        PrimitiveType colType = fileSchema.getType(desc.getPath()).asPrimitiveType();
+        if (colType.getId() != null) {
+          int id = colType.getId().intValue();
+          Type icebergType = schema.findType(id);
+          cols.put(id, desc);
+          //Change For Arctic
+          conversions.put(id, AdaptHiveParquetConversions.converterFromParquet(colType, icebergType));
+          //Change For Arctic:
+        }
+      }
+
+      for (ColumnChunkMetaData meta : rowGroup.getColumns()) {
+        PrimitiveType colType = fileSchema.getType(meta.getPath().toArray()).asPrimitiveType();
+        if (colType.getId() != null) {
+          int id = colType.getId().intValue();
+          isFallback.put(id, ParquetUtil.hasNonDictionaryPages(meta));
+          mayContainNulls.put(id, mayContainNull(meta));
+        }
+      }
+
+      return ExpressionVisitors.visitEvaluator(expr, this);
+    }
+
+    @Override
+    public Boolean alwaysTrue() {
+      return ROWS_MIGHT_MATCH; // all rows match
+    }
+
+    @Override
+    public Boolean alwaysFalse() {
+      return ROWS_CANNOT_MATCH; // all rows fail
+    }
+
+    @Override
+    public Boolean not(Boolean result) {
+      return !result;
+    }
+
+    @Override
+    public Boolean and(Boolean leftResult, Boolean rightResult) {
+      return leftResult && rightResult;
+    }
+
+    @Override
+    public Boolean or(Boolean leftResult, Boolean rightResult) {
+      return leftResult || rightResult;
+    }
+
+    @Override
+    public <T> Boolean isNull(BoundReference<T> ref) {
+      // dictionaries only contain non-nulls and cannot eliminate based on isNull or NotNull
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean notNull(BoundReference<T> ref) {
+      // dictionaries only contain non-nulls and cannot eliminate based on isNull or NotNull
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean isNaN(BoundReference<T> ref) {
+      int id = ref.fieldId();
+
+      Boolean hasNonDictPage = isFallback.get(id);
+      if (hasNonDictPage == null || hasNonDictPage) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Set<T> dictionary = dict(id, comparatorForNaNPredicate(ref));
+      return dictionary.stream().anyMatch(NaNUtil::isNaN) ? ROWS_MIGHT_MATCH : ROWS_CANNOT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean notNaN(BoundReference<T> ref) {
+      int id = ref.fieldId();
+
+      Boolean hasNonDictPage = isFallback.get(id);
+      if (hasNonDictPage == null || hasNonDictPage) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Set<T> dictionary = dict(id, comparatorForNaNPredicate(ref));
+      return dictionary.stream().allMatch(NaNUtil::isNaN) ? ROWS_CANNOT_MATCH : ROWS_MIGHT_MATCH;
+    }
+
+    private <T> Comparator<T> comparatorForNaNPredicate(BoundReference<T> ref) {
+      // Construct the same comparator as in ComparableLiteral.comparator, ignoring null value order as dictionary
+      // cannot contain null values.
+      // No need to check type: incompatible types will be handled during expression binding.
+      return Comparators.forType(ref.type().asPrimitiveType());
+    }
+
+    @Override
+    public <T> Boolean lt(BoundReference<T> ref, Literal<T> lit) {
+      int id = ref.fieldId();
+
+      Boolean hasNonDictPage = isFallback.get(id);
+      if (hasNonDictPage == null || hasNonDictPage) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Set<T> dictionary = dict(id, lit.comparator());
+
+      // if any item in the dictionary matches the predicate, then at least one row does
+      for (T item : dictionary) {
+        int cmp = lit.comparator().compare(item, lit.value());
+        if (cmp < 0) {
+          return ROWS_MIGHT_MATCH;
+        }
+      }
+
+      return ROWS_CANNOT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean ltEq(BoundReference<T> ref, Literal<T> lit) {
+      int id = ref.fieldId();
+
+      Boolean hasNonDictPage = isFallback.get(id);
+      if (hasNonDictPage == null || hasNonDictPage) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Set<T> dictionary = dict(id, lit.comparator());
+
+      // if any item in the dictionary matches the predicate, then at least one row does
+      for (T item : dictionary) {
+        int cmp = lit.comparator().compare(item, lit.value());
+        if (cmp <= 0) {
+          return ROWS_MIGHT_MATCH;
+        }
+      }
+
+      return ROWS_CANNOT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean gt(BoundReference<T> ref, Literal<T> lit) {
+      int id = ref.fieldId();
+
+      Boolean hasNonDictPage = isFallback.get(id);
+      if (hasNonDictPage == null || hasNonDictPage) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Set<T> dictionary = dict(id, lit.comparator());
+
+      // if any item in the dictionary matches the predicate, then at least one row does
+      for (T item : dictionary) {
+        int cmp = lit.comparator().compare(item, lit.value());
+        if (cmp > 0) {
+          return ROWS_MIGHT_MATCH;
+        }
+      }
+
+      return ROWS_CANNOT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean gtEq(BoundReference<T> ref, Literal<T> lit) {
+      int id = ref.fieldId();
+
+      Boolean hasNonDictPage = isFallback.get(id);
+      if (hasNonDictPage == null || hasNonDictPage) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Set<T> dictionary = dict(id, lit.comparator());
+
+      // if any item in the dictionary matches the predicate, then at least one row does
+      for (T item : dictionary) {
+        int cmp = lit.comparator().compare(item, lit.value());
+        if (cmp >= 0) {
+          return ROWS_MIGHT_MATCH;
+        }
+      }
+
+      return ROWS_CANNOT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean eq(BoundReference<T> ref, Literal<T> lit) {
+      int id = ref.fieldId();
+
+      Boolean hasNonDictPage = isFallback.get(id);
+      if (hasNonDictPage == null || hasNonDictPage) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Set<T> dictionary = dict(id, lit.comparator());
+
+      return dictionary.contains(lit.value()) ? ROWS_MIGHT_MATCH : ROWS_CANNOT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean notEq(BoundReference<T> ref, Literal<T> lit) {
+      int id = ref.fieldId();
+
+      Boolean hasNonDictPage = isFallback.get(id);
+      if (hasNonDictPage == null || hasNonDictPage) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Set<T> dictionary = dict(id, lit.comparator());
+      if (dictionary.size() > 1 || mayContainNulls.get(id)) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      return dictionary.contains(lit.value()) ? ROWS_CANNOT_MATCH : ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean in(BoundReference<T> ref, Set<T> literalSet) {
+      int id = ref.fieldId();
+
+      Boolean hasNonDictPage = isFallback.get(id);
+      if (hasNonDictPage == null || hasNonDictPage) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Set<T> dictionary = dict(id, ref.comparator());
+
+      // we need to find out the smaller set to iterate through
+      Set<T> smallerSet;
+      Set<T> biggerSet;
+
+      if (literalSet.size() < dictionary.size()) {
+        smallerSet = literalSet;
+        biggerSet = dictionary;
+      } else {
+        smallerSet = dictionary;
+        biggerSet = literalSet;
+      }
+
+      for (T e : smallerSet) {
+        if (biggerSet.contains(e)) {
+          // value sets intersect so rows match
+          return ROWS_MIGHT_MATCH;
+        }
+      }
+
+      // value sets are disjoint so rows don't match
+      return ROWS_CANNOT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean notIn(BoundReference<T> ref, Set<T> literalSet) {
+      int id = ref.fieldId();
+
+      Boolean hasNonDictPage = isFallback.get(id);
+      if (hasNonDictPage == null || hasNonDictPage) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Set<T> dictionary = dict(id, ref.comparator());
+      if (dictionary.size() > literalSet.size() || mayContainNulls.get(id)) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      // ROWS_CANNOT_MATCH if no values in the dictionary that are not also in the set (the difference is empty)
+      return Sets.difference(dictionary, literalSet).isEmpty() ? ROWS_CANNOT_MATCH : ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean startsWith(BoundReference<T> ref, Literal<T> lit) {
+      int id = ref.fieldId();
+
+      Boolean hasNonDictPage = isFallback.get(id);
+      if (hasNonDictPage == null || hasNonDictPage) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Set<T> dictionary = dict(id, lit.comparator());
+      for (T item : dictionary) {
+        if (item.toString().startsWith(lit.value().toString())) {
+          return ROWS_MIGHT_MATCH;
+        }
+      }
+
+      return ROWS_CANNOT_MATCH;
+    }
+
+    @SuppressWarnings("unchecked")
+    private <T> Set<T> dict(int id, Comparator<T> comparator) {
+      Preconditions.checkNotNull(dictionaries, "Dictionary is required");
+
+      Set<?> cached = dictCache.get(id);
+      if (cached != null) {
+        return (Set<T>) cached;
+      }
+
+      ColumnDescriptor col = cols.get(id);
+      DictionaryPage page = dictionaries.readDictionaryPage(col);
+      // may not be dictionary-encoded
+      if (page == null) {
+        throw new IllegalStateException("Failed to read required dictionary page for id: " + id);
+      }
+
+      Function<Object, Object> conversion = conversions.get(id);
+
+      Dictionary dict;
+      try {
+        dict = page.getEncoding().initDictionary(col, page);
+      } catch (IOException e) {
+        throw new RuntimeIOException("Failed to create reader for dictionary page");
+      }
+
+      Set<T> dictSet = Sets.newTreeSet(comparator);
+
+      for (int i = 0; i <= dict.getMaxId(); i++) {
+        switch (col.getPrimitiveType().getPrimitiveTypeName()) {
+          case FIXED_LEN_BYTE_ARRAY: dictSet.add((T) conversion.apply(dict.decodeToBinary(i)));
+            break;
+          case BINARY: dictSet.add((T) conversion.apply(dict.decodeToBinary(i)));
+            break;
+          case INT32: dictSet.add((T) conversion.apply(dict.decodeToInt(i)));
+            break;
+          case INT64: dictSet.add((T) conversion.apply(dict.decodeToLong(i)));
+            break;
+          case FLOAT: dictSet.add((T) conversion.apply(dict.decodeToFloat(i)));
+            break;
+          case DOUBLE: dictSet.add((T) conversion.apply(dict.decodeToDouble(i)));
+            break;
+          case INT96: dictSet.add((T) conversion.apply(dict.decodeToBinary(i)));
+            break;
+          default:
+            throw new IllegalArgumentException(
+                "Cannot decode dictionary of type: " + col.getPrimitiveType().getPrimitiveTypeName());
+        }
+      }
+
+      dictCache.put(id, dictSet);
+
+      return dictSet;
+    }
+  }
+
+  private static boolean mayContainNull(ColumnChunkMetaData meta) {
+    return meta.getStatistics() == null || meta.getStatistics().getNumNulls() != 0;
+  }
+
+}
diff --git a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetMetricsRowGroupFilter.java b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetMetricsRowGroupFilter.java
new file mode 100644
index 000000000..b0c7ae345
--- /dev/null
+++ b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetMetricsRowGroupFilter.java
@@ -0,0 +1,527 @@
+/*
+ * 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 org.apache.iceberg.parquet;
+
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.expressions.Binder;
+import org.apache.iceberg.expressions.BoundReference;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.ExpressionVisitors;
+import org.apache.iceberg.expressions.ExpressionVisitors.BoundExpressionVisitor;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.Literal;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types.StructType;
+import org.apache.iceberg.util.BinaryUtil;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * Copy from iceberg {@link org.apache.iceberg.parquet.ParquetMetricsRowGroupFilter} to resolve int96 type.
+ */
+public class AdaptHiveParquetMetricsRowGroupFilter {
+  private static final int IN_PREDICATE_LIMIT = 200;
+
+  private final Schema schema;
+  private final Expression expr;
+
+  public AdaptHiveParquetMetricsRowGroupFilter(Schema schema, Expression unbound) {
+    this(schema, unbound, true);
+  }
+
+  public AdaptHiveParquetMetricsRowGroupFilter(Schema schema, Expression unbound, boolean caseSensitive) {
+    this.schema = schema;
+    StructType struct = schema.asStruct();
+    this.expr = Binder.bind(struct, Expressions.rewriteNot(unbound), caseSensitive);
+  }
+
+  /**
+   * Test whether the file may contain records that match the expression.
+   *
+   * @param fileSchema schema for the Parquet file
+   * @param rowGroup metadata for a row group
+   * @return false if the file cannot contain rows that match the expression, true otherwise.
+   */
+  public boolean shouldRead(MessageType fileSchema, BlockMetaData rowGroup) {
+    return new MetricsEvalVisitor().eval(fileSchema, rowGroup);
+  }
+
+  private static final boolean ROWS_MIGHT_MATCH = true;
+  private static final boolean ROWS_CANNOT_MATCH = false;
+
+  private class MetricsEvalVisitor extends BoundExpressionVisitor<Boolean> {
+    private Map<Integer, Statistics<?>> stats = null;
+    private Map<Integer, Long> valueCounts = null;
+    private Map<Integer, Function<Object, Object>> conversions = null;
+
+    private boolean eval(MessageType fileSchema, BlockMetaData rowGroup) {
+      if (rowGroup.getRowCount() <= 0) {
+        return ROWS_CANNOT_MATCH;
+      }
+
+      this.stats = Maps.newHashMap();
+      this.valueCounts = Maps.newHashMap();
+      this.conversions = Maps.newHashMap();
+      for (ColumnChunkMetaData col : rowGroup.getColumns()) {
+        PrimitiveType colType = fileSchema.getType(col.getPath().toArray()).asPrimitiveType();
+        if (colType.getId() != null) {
+          int id = colType.getId().intValue();
+          Type icebergType = schema.findType(id);
+          stats.put(id, col.getStatistics());
+          valueCounts.put(id, col.getValueCount());
+          //Change For Arctic
+          conversions.put(id, AdaptHiveParquetConversions.converterFromParquet(colType, icebergType));
+          //Change For Arctic
+        }
+      }
+
+      return ExpressionVisitors.visitEvaluator(expr, this);
+    }
+
+    @Override
+    public Boolean alwaysTrue() {
+      return ROWS_MIGHT_MATCH; // all rows match
+    }
+
+    @Override
+    public Boolean alwaysFalse() {
+      return ROWS_CANNOT_MATCH; // all rows fail
+    }
+
+    @Override
+    public Boolean not(Boolean result) {
+      return !result;
+    }
+
+    @Override
+    public Boolean and(Boolean leftResult, Boolean rightResult) {
+      return leftResult && rightResult;
+    }
+
+    @Override
+    public Boolean or(Boolean leftResult, Boolean rightResult) {
+      return leftResult || rightResult;
+    }
+
+    @Override
+    public <T> Boolean isNull(BoundReference<T> ref) {
+      // no need to check whether the field is required because binding evaluates that case
+      // if the column has no null values, the expression cannot match
+      Integer id = ref.fieldId();
+
+      Long valueCount = valueCounts.get(id);
+      if (valueCount == null) {
+        // the column is not present and is all nulls
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Statistics<?> colStats = stats.get(id);
+      if (colStats != null && !colStats.isEmpty() && colStats.getNumNulls() == 0) {
+        // there are stats and no values are null => all values are non-null
+        return ROWS_CANNOT_MATCH;
+      }
+
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean notNull(BoundReference<T> ref) {
+      // no need to check whether the field is required because binding evaluates that case
+      // if the column has no non-null values, the expression cannot match
+      Integer id = ref.fieldId();
+
+      // When filtering nested types notNull() is implicit filter passed even though complex
+      // filters aren't pushed down in Parquet. Leave all nested column type filters to be
+      // evaluated post scan.
+      if (schema.findType(id) instanceof Type.NestedType) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Long valueCount = valueCounts.get(id);
+      if (valueCount == null) {
+        // the column is not present and is all nulls
+        return ROWS_CANNOT_MATCH;
+      }
+
+      Statistics<?> colStats = stats.get(id);
+      if (colStats != null && valueCount - colStats.getNumNulls() == 0) {
+        // (num nulls == value count) => all values are null => no non-null values
+        return ROWS_CANNOT_MATCH;
+      }
+
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean isNaN(BoundReference<T> ref) {
+      int id = ref.fieldId();
+
+      Long valueCount = valueCounts.get(id);
+      if (valueCount == null) {
+        // the column is not present and is all nulls
+        return ROWS_CANNOT_MATCH;
+      }
+
+      Statistics<?> colStats = stats.get(id);
+      if (colStats != null && valueCount - colStats.getNumNulls() == 0) {
+        // (num nulls == value count) => all values are null => no nan values
+        return ROWS_CANNOT_MATCH;
+      }
+
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean notNaN(BoundReference<T> ref) {
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean lt(BoundReference<T> ref, Literal<T> lit) {
+      Integer id = ref.fieldId();
+
+      Long valueCount = valueCounts.get(id);
+      if (valueCount == null) {
+        // the column is not present and is all nulls
+        return ROWS_CANNOT_MATCH;
+      }
+
+      Statistics<?> colStats = stats.get(id);
+      if (colStats != null && !colStats.isEmpty()) {
+        if (hasNonNullButNoMinMax(colStats, valueCount)) {
+          return ROWS_MIGHT_MATCH;
+        }
+
+        if (!colStats.hasNonNullValue()) {
+          return ROWS_CANNOT_MATCH;
+        }
+
+        T lower = min(colStats, id);
+        int cmp = lit.comparator().compare(lower, lit.value());
+        if (cmp >= 0) {
+          return ROWS_CANNOT_MATCH;
+        }
+      }
+
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean ltEq(BoundReference<T> ref, Literal<T> lit) {
+      Integer id = ref.fieldId();
+
+      Long valueCount = valueCounts.get(id);
+      if (valueCount == null) {
+        // the column is not present and is all nulls
+        return ROWS_CANNOT_MATCH;
+      }
+
+      Statistics<?> colStats = stats.get(id);
+      if (colStats != null && !colStats.isEmpty()) {
+        if (hasNonNullButNoMinMax(colStats, valueCount)) {
+          return ROWS_MIGHT_MATCH;
+        }
+
+        if (!colStats.hasNonNullValue()) {
+          return ROWS_CANNOT_MATCH;
+        }
+
+        T lower = min(colStats, id);
+        int cmp = lit.comparator().compare(lower, lit.value());
+        if (cmp > 0) {
+          return ROWS_CANNOT_MATCH;
+        }
+      }
+
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean gt(BoundReference<T> ref, Literal<T> lit) {
+      Integer id = ref.fieldId();
+
+      Long valueCount = valueCounts.get(id);
+      if (valueCount == null) {
+        // the column is not present and is all nulls
+        return ROWS_CANNOT_MATCH;
+      }
+
+      Statistics<?> colStats = stats.get(id);
+      if (colStats != null && !colStats.isEmpty()) {
+        if (hasNonNullButNoMinMax(colStats, valueCount)) {
+          return ROWS_MIGHT_MATCH;
+        }
+
+        if (!colStats.hasNonNullValue()) {
+          return ROWS_CANNOT_MATCH;
+        }
+
+        T upper = max(colStats, id);
+        int cmp = lit.comparator().compare(upper, lit.value());
+        if (cmp <= 0) {
+          return ROWS_CANNOT_MATCH;
+        }
+      }
+
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean gtEq(BoundReference<T> ref, Literal<T> lit) {
+      Integer id = ref.fieldId();
+
+      Long valueCount = valueCounts.get(id);
+      if (valueCount == null) {
+        // the column is not present and is all nulls
+        return ROWS_CANNOT_MATCH;
+      }
+
+      Statistics<?> colStats = stats.get(id);
+      if (colStats != null && !colStats.isEmpty()) {
+        if (hasNonNullButNoMinMax(colStats, valueCount)) {
+          return ROWS_MIGHT_MATCH;
+        }
+
+        if (!colStats.hasNonNullValue()) {
+          return ROWS_CANNOT_MATCH;
+        }
+
+        T upper = max(colStats, id);
+        int cmp = lit.comparator().compare(upper, lit.value());
+        if (cmp < 0) {
+          return ROWS_CANNOT_MATCH;
+        }
+      }
+
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean eq(BoundReference<T> ref, Literal<T> lit) {
+      Integer id = ref.fieldId();
+
+      // When filtering nested types notNull() is implicit filter passed even though complex
+      // filters aren't pushed down in Parquet. Leave all nested column type filters to be
+      // evaluated post scan.
+      if (schema.findType(id) instanceof Type.NestedType) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Long valueCount = valueCounts.get(id);
+      if (valueCount == null) {
+        // the column is not present and is all nulls
+        return ROWS_CANNOT_MATCH;
+      }
+
+      Statistics<?> colStats = stats.get(id);
+      if (colStats != null && !colStats.isEmpty()) {
+        if (hasNonNullButNoMinMax(colStats, valueCount)) {
+          return ROWS_MIGHT_MATCH;
+        }
+
+        if (!colStats.hasNonNullValue()) {
+          return ROWS_CANNOT_MATCH;
+        }
+
+        T lower = min(colStats, id);
+        int cmp = lit.comparator().compare(lower, lit.value());
+        if (cmp > 0) {
+          return ROWS_CANNOT_MATCH;
+        }
+
+        T upper = max(colStats, id);
+        cmp = lit.comparator().compare(upper, lit.value());
+        if (cmp < 0) {
+          return ROWS_CANNOT_MATCH;
+        }
+      }
+
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean notEq(BoundReference<T> ref, Literal<T> lit) {
+      // because the bounds are not necessarily a min or max value, this cannot be answered using
+      // them. notEq(col, X) with (X, Y) doesn't guarantee that X is a value in col.
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean in(BoundReference<T> ref, Set<T> literalSet) {
+      Integer id = ref.fieldId();
+
+      // When filtering nested types notNull() is implicit filter passed even though complex
+      // filters aren't pushed down in Parquet. Leave all nested column type filters to be
+      // evaluated post scan.
+      if (schema.findType(id) instanceof Type.NestedType) {
+        return ROWS_MIGHT_MATCH;
+      }
+
+      Long valueCount = valueCounts.get(id);
+      if (valueCount == null) {
+        // the column is not present and is all nulls
+        return ROWS_CANNOT_MATCH;
+      }
+
+      Statistics<?> colStats = stats.get(id);
+      if (colStats != null && !colStats.isEmpty()) {
+        if (hasNonNullButNoMinMax(colStats, valueCount)) {
+          return ROWS_MIGHT_MATCH;
+        }
+
+        if (!colStats.hasNonNullValue()) {
+          return ROWS_CANNOT_MATCH;
+        }
+
+        Collection<T> literals = literalSet;
+
+        if (literals.size() > IN_PREDICATE_LIMIT) {
+          // skip evaluating the predicate if the number of values is too big
+          return ROWS_MIGHT_MATCH;
+        }
+
+        T lower = min(colStats, id);
+        literals = literals.stream().filter(v -> ref.comparator().compare(lower, v) <= 0).collect(Collectors.toList());
+        if (literals.isEmpty()) {  // if all values are less than lower bound, rows cannot match.
+          return ROWS_CANNOT_MATCH;
+        }
+
+        T upper = max(colStats, id);
+        literals = literals.stream().filter(v -> ref.comparator().compare(upper, v) >= 0).collect(Collectors.toList());
+        if (literals.isEmpty()) { // if all remaining values are greater than upper bound, rows cannot match.
+          return ROWS_CANNOT_MATCH;
+        }
+      }
+
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    public <T> Boolean notIn(BoundReference<T> ref, Set<T> literalSet) {
+      // because the bounds are not necessarily a min or max value, this cannot be answered using
+      // them. notIn(col, {X, ...}) with (X, Y) doesn't guarantee that X is a value in col.
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public <T> Boolean startsWith(BoundReference<T> ref, Literal<T> lit) {
+      int id = ref.fieldId();
+
+      Long valueCount = valueCounts.get(id);
+      if (valueCount == null) {
+        // the column is not present and is all nulls
+        return ROWS_CANNOT_MATCH;
+      }
+
+      Statistics<Binary> colStats = (Statistics<Binary>) stats.get(id);
+      if (colStats != null && !colStats.isEmpty()) {
+        if (hasNonNullButNoMinMax(colStats, valueCount)) {
+          return ROWS_MIGHT_MATCH;
+        }
+
+        if (!colStats.hasNonNullValue()) {
+          return ROWS_CANNOT_MATCH;
+        }
+
+        ByteBuffer prefixAsBytes = lit.toByteBuffer();
+
+        Comparator<ByteBuffer> comparator = Comparators.unsignedBytes();
+
+        Binary lower = colStats.genericGetMin();
+        // truncate lower bound so that its length in bytes is not greater than the length of prefix
+        int lowerLength = Math.min(prefixAsBytes.remaining(), lower.length());
+        int lowerCmp = comparator.compare(BinaryUtil.truncateBinary(lower.toByteBuffer(), lowerLength), prefixAsBytes);
+        if (lowerCmp > 0) {
+          return ROWS_CANNOT_MATCH;
+        }
+
+        Binary upper = colStats.genericGetMax();
+        // truncate upper bound so that its length in bytes is not greater than the length of prefix
+        int upperLength = Math.min(prefixAsBytes.remaining(), upper.length());
+        int upperCmp = comparator.compare(BinaryUtil.truncateBinary(upper.toByteBuffer(), upperLength), prefixAsBytes);
+        if (upperCmp < 0) {
+          return ROWS_CANNOT_MATCH;
+        }
+      }
+
+      return ROWS_MIGHT_MATCH;
+    }
+
+    @SuppressWarnings("unchecked")
+    private <T> T min(Statistics<?> statistics, int id) {
+      return (T) conversions.get(id).apply(statistics.genericGetMin());
+    }
+
+    @SuppressWarnings("unchecked")
+    private <T> T max(Statistics<?> statistics, int id) {
+      return (T) conversions.get(id).apply(statistics.genericGetMax());
+    }
+  }
+
+  /**
+   * Checks against older versions of Parquet statistics which may have a null count but undefined min and max
+   * statistics. Returns true if nonNull values exist in the row group but no further statistics are available.
+   * <p>
+   * We can't use {@code  statistics.hasNonNullValue()} because it is inaccurate with older files and will return
+   * false if min and max are not set.
+   * <p>
+   * This is specifically for 1.5.0-CDH Parquet builds and later which contain the different unusual hasNonNull
+   * behavior. OSS Parquet builds are not effected because PARQUET-251 prohibits the reading of these statistics
+   * from versions of Parquet earlier than 1.8.0.
+   *
+   * @param statistics Statistics to check
+   * @param valueCount Number of values in the row group
+   * @return true if nonNull values exist and no other stats can be used
+   */
+  static boolean hasNonNullButNoMinMax(Statistics statistics, long valueCount) {
+    return statistics.getNumNulls() < valueCount &&
+        (statistics.getMaxBytes() == null || statistics.getMinBytes() == null);
+  }
+
+  private static Function<Object, Object> converterFor(PrimitiveType parquetType, Type icebergType) {
+    Function<Object, Object> fromParquet = ParquetConversions.converterFromParquet(parquetType);
+    if (icebergType != null) {
+      if (icebergType.typeId() == Type.TypeID.LONG &&
+          parquetType.getPrimitiveTypeName() == PrimitiveType.PrimitiveTypeName.INT32) {
+        return value -> ((Integer) fromParquet.apply(value)).longValue();
+      } else if (icebergType.typeId() == Type.TypeID.DOUBLE &&
+          parquetType.getPrimitiveTypeName() == PrimitiveType.PrimitiveTypeName.FLOAT) {
+        return value -> ((Float) fromParquet.apply(value)).doubleValue();
+      }
+    }
+
+    return fromParquet;
+  }
+}
diff --git a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetReader.java b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetReader.java
new file mode 100644
index 000000000..8477d7b10
--- /dev/null
+++ b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetReader.java
@@ -0,0 +1,156 @@
+/*
+ * 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 org.apache.iceberg.parquet;
+
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableGroup;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.mapping.NameMapping;
+import org.apache.parquet.ParquetReadOptions;
+import org.apache.parquet.column.page.PageReadStore;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.schema.MessageType;
+
+import java.io.IOException;
+import java.util.function.Function;
+
+/**
+ * Copy from iceberg {@link org.apache.iceberg.parquet.ParquetReader} to use {@link AdaptHiveReadConf}
+ */
+public class AdaptHiveParquetReader<T> extends CloseableGroup implements CloseableIterable<T> {
+  private final InputFile input;
+  private final Schema expectedSchema;
+  private final ParquetReadOptions options;
+  private final Function<MessageType, ParquetValueReader<?>> readerFunc;
+  private final Expression filter;
+  private final boolean reuseContainers;
+  private final boolean caseSensitive;
+  private final NameMapping nameMapping;
+
+  public AdaptHiveParquetReader(InputFile input, Schema expectedSchema, ParquetReadOptions options,
+      Function<MessageType, ParquetValueReader<?>> readerFunc, NameMapping nameMapping,
+      Expression filter, boolean reuseContainers, boolean caseSensitive) {
+    this.input = input;
+    this.expectedSchema = expectedSchema;
+    this.options = options;
+    this.readerFunc = readerFunc;
+    // replace alwaysTrue with null to avoid extra work evaluating a trivial filter
+    this.filter = filter == Expressions.alwaysTrue() ? null : filter;
+    this.reuseContainers = reuseContainers;
+    this.caseSensitive = caseSensitive;
+    this.nameMapping = nameMapping;
+  }
+
+  private AdaptHiveReadConf<T> conf = null;
+
+  private AdaptHiveReadConf<T> init() {
+    if (conf == null) {
+      //Change For Arctic
+      AdaptHiveReadConf<T> adaptHiveReadConf = new AdaptHiveReadConf<>(
+          input, options, expectedSchema, filter, readerFunc, null, nameMapping, reuseContainers,
+          caseSensitive, null);
+      //Change For Arctic
+      this.conf = adaptHiveReadConf.copy();
+      return adaptHiveReadConf;
+    }
+    return conf;
+  }
+
+  @Override
+  public CloseableIterator<T> iterator() {
+    FileIterator<T> iter = new FileIterator<>(init());
+    addCloseable(iter);
+    return iter;
+  }
+
+  private static class FileIterator<T> implements CloseableIterator<T> {
+    private final ParquetFileReader reader;
+    private final boolean[] shouldSkip;
+    private final ParquetValueReader<T> model;
+    private final long totalValues;
+    private final boolean reuseContainers;
+    private final long[] rowGroupsStartRowPos;
+
+    private int nextRowGroup = 0;
+    private long nextRowGroupStart = 0;
+    private long valuesRead = 0;
+    private T last = null;
+
+    FileIterator(AdaptHiveReadConf<T> conf) {
+      this.reader = conf.reader();
+      this.shouldSkip = conf.shouldSkip();
+      this.model = conf.model();
+      this.totalValues = conf.totalValues();
+      this.reuseContainers = conf.reuseContainers();
+      this.rowGroupsStartRowPos = conf.startRowPositions();
+    }
+
+    @Override
+    public boolean hasNext() {
+      return valuesRead < totalValues;
+    }
+
+    @Override
+    public T next() {
+      if (valuesRead >= nextRowGroupStart) {
+        advance();
+      }
+
+      if (reuseContainers) {
+        this.last = model.read(last);
+      } else {
+        this.last = model.read(null);
+      }
+      valuesRead += 1;
+
+      return last;
+    }
+
+    private void advance() {
+      while (shouldSkip[nextRowGroup]) {
+        nextRowGroup += 1;
+        reader.skipNextRowGroup();
+      }
+
+      PageReadStore pages;
+      try {
+        pages = reader.readNextRowGroup();
+      } catch (IOException e) {
+        throw new RuntimeIOException(e);
+      }
+
+      long rowPosition = rowGroupsStartRowPos[nextRowGroup];
+      nextRowGroupStart += pages.getRowCount();
+      nextRowGroup += 1;
+
+      model.setPageSource(pages, rowPosition);
+    }
+
+    @Override
+    public void close() throws IOException {
+      reader.close();
+    }
+  }
+}
diff --git a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetSchemaUtil.java b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetSchemaUtil.java
index a651e5f76..8e66081d6 100644
--- a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetSchemaUtil.java
+++ b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetSchemaUtil.java
@@ -38,6 +38,7 @@ public class AdaptHiveParquetSchemaUtil {
   }
 
   public static MessageType convert(Schema schema, String name) {
+    //Change For Arctic
     return new AdaptHiveTypeToMessageType().convert(schema, name);
   }
 
diff --git a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetWriter.java b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetWriter.java
index 7362faf24..4c58dbd5c 100644
--- a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetWriter.java
+++ b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveParquetWriter.java
@@ -94,7 +94,9 @@ class AdaptHiveParquetWriter<T> implements FileAppender<T>, Closeable {
     this.props = properties;
     this.metadata = ImmutableMap.copyOf(metadata);
     this.compressor = new CodecFactory(conf, props.getPageSizeThreshold()).getCompressor(codec);
+    //Change For Arctic
     this.parquetSchema = AdaptHiveParquetSchemaUtil.convert(schema, "table");
+    //Change For Arctic
     this.model = (ParquetValueWriter<T>) createWriterFunc.apply(parquetSchema);
     this.metricsConfig = metricsConfig;
     this.columnIndexTruncateLength = conf.getInt(COLUMN_INDEX_TRUNCATE_LENGTH, DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH);
diff --git a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHivePrimitiveWriter.java b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHivePrimitiveWriter.java
index fd7247155..a45d43d05 100644
--- a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHivePrimitiveWriter.java
+++ b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHivePrimitiveWriter.java
@@ -31,7 +31,9 @@ public class AdaptHivePrimitiveWriter<T> implements ParquetValueWriter<T> {
   private final List<TripleWriter<?>> children;
 
   protected AdaptHivePrimitiveWriter(ColumnDescriptor desc) {
+    //Change For Arctic
     this.column = AdaptHiveColumnWriter.newWriter(desc);
+    //Change For Arctic
     this.children = ImmutableList.of(column);
   }
 
diff --git a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveReadConf.java b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveReadConf.java
new file mode 100644
index 000000000..7fb2c0198
--- /dev/null
+++ b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveReadConf.java
@@ -0,0 +1,244 @@
+/*
+ * 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 org.apache.iceberg.parquet;
+
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.exceptions.RuntimeIOException;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.mapping.NameMapping;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.parquet.ParquetReadOptions;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnPath;
+import org.apache.parquet.schema.MessageType;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * Copy from iceberg {@ling org.apache.iceberg.parquet.ReadConf} to use {@link AdaptHiveParquetMetricsRowGroupFilter}
+ * and {@link AdaptHiveParquetDictionaryRowGroupFilter}
+ */
+class AdaptHiveReadConf<T> {
+  private final ParquetFileReader reader;
+  private final InputFile file;
+  private final ParquetReadOptions options;
+  private final MessageType projection;
+  private final ParquetValueReader<T> model;
+  private final VectorizedReader<T> vectorizedModel;
+  private final List<BlockMetaData> rowGroups;
+  private final boolean[] shouldSkip;
+  private final long totalValues;
+  private final boolean reuseContainers;
+  private final Integer batchSize;
+  private final long[] startRowPositions;
+
+  // List of column chunk metadata for each row group
+  private final List<Map<ColumnPath, ColumnChunkMetaData>> columnChunkMetaDataForRowGroups;
+
+  @SuppressWarnings("unchecked")
+  AdaptHiveReadConf(InputFile file, ParquetReadOptions options, Schema expectedSchema, Expression filter,
+      Function<MessageType, ParquetValueReader<?>> readerFunc, Function<MessageType,
+      VectorizedReader<?>> batchedReaderFunc, NameMapping nameMapping, boolean reuseContainers,
+      boolean caseSensitive, Integer batchSize) {
+    this.file = file;
+    this.options = options;
+    this.reader = newReader(file, options);
+    MessageType fileSchema = reader.getFileMetaData().getSchema();
+
+    MessageType typeWithIds;
+    if (ParquetSchemaUtil.hasIds(fileSchema)) {
+      typeWithIds = fileSchema;
+      this.projection = ParquetSchemaUtil.pruneColumns(fileSchema, expectedSchema);
+    } else if (nameMapping != null) {
+      typeWithIds = ParquetSchemaUtil.applyNameMapping(fileSchema, nameMapping);
+      this.projection = ParquetSchemaUtil.pruneColumns(typeWithIds, expectedSchema);
+    } else {
+      typeWithIds = ParquetSchemaUtil.addFallbackIds(fileSchema);
+      this.projection = ParquetSchemaUtil.pruneColumnsFallback(fileSchema, expectedSchema);
+    }
+
+    this.rowGroups = reader.getRowGroups();
+    this.shouldSkip = new boolean[rowGroups.size()];
+    this.startRowPositions = new long[rowGroups.size()];
+
+    // Fetch all row groups starting positions to compute the row offsets of the filtered row groups
+    Map<Long, Long> offsetToStartPos = generateOffsetToStartPos(expectedSchema);
+
+    //Change For Arctic: use arctic filter
+    AdaptHiveParquetMetricsRowGroupFilter statsFilter = null;
+    AdaptHiveParquetDictionaryRowGroupFilter dictFilter = null;
+    if (filter != null) {
+      statsFilter = new AdaptHiveParquetMetricsRowGroupFilter(expectedSchema, filter, caseSensitive);
+      dictFilter = new AdaptHiveParquetDictionaryRowGroupFilter(expectedSchema, filter, caseSensitive);
+    }
+    //Change For Arctic
+
+    long computedTotalValues = 0L;
+    for (int i = 0; i < shouldSkip.length; i += 1) {
+      BlockMetaData rowGroup = rowGroups.get(i);
+      startRowPositions[i] = offsetToStartPos == null ? 0 : offsetToStartPos.get(rowGroup.getStartingPos());
+      boolean shouldRead = filter == null || (
+          statsFilter.shouldRead(typeWithIds, rowGroup) &&
+              dictFilter.shouldRead(typeWithIds, rowGroup, reader.getDictionaryReader(rowGroup)));
+      this.shouldSkip[i] = !shouldRead;
+      if (shouldRead) {
+        computedTotalValues += rowGroup.getRowCount();
+      }
+    }
+
+    this.totalValues = computedTotalValues;
+    if (readerFunc != null) {
+      this.model = (ParquetValueReader<T>) readerFunc.apply(typeWithIds);
+      this.vectorizedModel = null;
+      this.columnChunkMetaDataForRowGroups = null;
+    } else {
+      this.model = null;
+      this.vectorizedModel = (VectorizedReader<T>) batchedReaderFunc.apply(typeWithIds);
+      this.columnChunkMetaDataForRowGroups = getColumnChunkMetadataForRowGroups();
+    }
+
+    this.reuseContainers = reuseContainers;
+    this.batchSize = batchSize;
+  }
+
+  private AdaptHiveReadConf(AdaptHiveReadConf<T> toCopy) {
+    this.reader = null;
+    this.file = toCopy.file;
+    this.options = toCopy.options;
+    this.projection = toCopy.projection;
+    this.model = toCopy.model;
+    this.rowGroups = toCopy.rowGroups;
+    this.shouldSkip = toCopy.shouldSkip;
+    this.totalValues = toCopy.totalValues;
+    this.reuseContainers = toCopy.reuseContainers;
+    this.batchSize = toCopy.batchSize;
+    this.vectorizedModel = toCopy.vectorizedModel;
+    this.columnChunkMetaDataForRowGroups = toCopy.columnChunkMetaDataForRowGroups;
+    this.startRowPositions = toCopy.startRowPositions;
+  }
+
+  ParquetFileReader reader() {
+    if (reader != null) {
+      reader.setRequestedSchema(projection);
+      return reader;
+    }
+
+    ParquetFileReader newReader = newReader(file, options);
+    newReader.setRequestedSchema(projection);
+    return newReader;
+  }
+
+  ParquetValueReader<T> model() {
+    return model;
+  }
+
+  VectorizedReader<T> vectorizedModel() {
+    return vectorizedModel;
+  }
+
+  boolean[] shouldSkip() {
+    return shouldSkip;
+  }
+
+  private Map<Long, Long> generateOffsetToStartPos(Schema schema) {
+    if (schema.findField(MetadataColumns.ROW_POSITION.fieldId()) == null) {
+      return null;
+    }
+
+    try (ParquetFileReader fileReader = newReader(file, ParquetReadOptions.builder().build())) {
+      Map<Long, Long> offsetToStartPos = new HashMap<>();
+
+      long curRowCount = 0;
+      for (int i = 0; i < fileReader.getRowGroups().size(); i += 1) {
+        BlockMetaData meta = fileReader.getRowGroups().get(i);
+        offsetToStartPos.put(meta.getStartingPos(), curRowCount);
+        curRowCount += meta.getRowCount();
+      }
+
+      return offsetToStartPos;
+
+    } catch (IOException e) {
+      throw new UncheckedIOException("Failed to create/close reader for file: " + file, e);
+    }
+  }
+
+  long[] startRowPositions() {
+    return startRowPositions;
+  }
+
+  long totalValues() {
+    return totalValues;
+  }
+
+  boolean reuseContainers() {
+    return reuseContainers;
+  }
+
+  Integer batchSize() {
+    return batchSize;
+  }
+
+  List<Map<ColumnPath, ColumnChunkMetaData>> columnChunkMetadataForRowGroups() {
+    return columnChunkMetaDataForRowGroups;
+  }
+
+  AdaptHiveReadConf<T> copy() {
+    return new AdaptHiveReadConf<>(this);
+  }
+
+  private static ParquetFileReader newReader(InputFile file, ParquetReadOptions options) {
+    try {
+      return ParquetFileReader.open(ParquetIO.file(file), options);
+    } catch (IOException e) {
+      throw new RuntimeIOException(e, "Failed to open Parquet file: %s", file.location());
+    }
+  }
+
+  private List<Map<ColumnPath, ColumnChunkMetaData>> getColumnChunkMetadataForRowGroups() {
+    Set<ColumnPath> projectedColumns = projection.getColumns().stream()
+        .map(columnDescriptor -> ColumnPath.get(columnDescriptor.getPath())).collect(Collectors.toSet());
+    ImmutableList.Builder<Map<ColumnPath, ColumnChunkMetaData>> listBuilder = ImmutableList.builder();
+    for (int i = 0; i < rowGroups.size(); i++) {
+      if (!shouldSkip[i]) {
+        BlockMetaData blockMetaData = rowGroups.get(i);
+        ImmutableMap.Builder<ColumnPath, ColumnChunkMetaData> mapBuilder = ImmutableMap.builder();
+        blockMetaData.getColumns().stream()
+            .filter(columnChunkMetaData -> projectedColumns.contains(columnChunkMetaData.getPath()))
+            .forEach(columnChunkMetaData -> mapBuilder.put(columnChunkMetaData.getPath(), columnChunkMetaData));
+        listBuilder.add(mapBuilder.build());
+      } else {
+        listBuilder.add(ImmutableMap.of());
+      }
+    }
+    return listBuilder.build();
+  }
+}
diff --git a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveTypeToMessageType.java b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveTypeToMessageType.java
index 9601e8497..473154540 100644
--- a/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveTypeToMessageType.java
+++ b/hive/src/main/java/org/apache/iceberg/parquet/AdaptHiveTypeToMessageType.java
@@ -127,8 +127,10 @@ public class AdaptHiveTypeToMessageType {
         return Types.primitive(INT32, repetition).as(DATE).id(id).named(name);
       case TIME:
         return Types.primitive(INT64, repetition).as(TIME_MICROS).id(id).named(name);
+      //Change For Arctic
       case TIMESTAMP:
         return Types.primitive(INT96, repetition).id(id).named(name);
+      //Change For Arctic
       case STRING:
         return Types.primitive(BINARY, repetition).as(STRING).id(id).named(name);
       case BINARY:
diff --git a/optimizer/src/main/java/com/netease/arctic/optimizer/OptimizerConfig.java b/optimizer/src/main/java/com/netease/arctic/optimizer/OptimizerConfig.java
index 5fd7c46e6..1dbc8a31a 100644
--- a/optimizer/src/main/java/com/netease/arctic/optimizer/OptimizerConfig.java
+++ b/optimizer/src/main/java/com/netease/arctic/optimizer/OptimizerConfig.java
@@ -39,6 +39,13 @@ public class OptimizerConfig implements Serializable {
   @Option(name = "-hb", aliases = "--heart-beat", usage = "heart beat interval (ms)")
   private long heartBeat = 10000; // 10 s
 
+  // support register optimizer
+  @Option(name = "-qn", aliases = "--queue-name", usage = "queue name")
+  private String queueName;
+
+  @Option(name = "-m", aliases = "--memory-size", usage = "memory size")
+  private int executorMemory;
+
   public OptimizerConfig() {
   }
 
@@ -87,6 +94,22 @@ public class OptimizerConfig implements Serializable {
     this.optimizerId = optimizerId;
   }
 
+  public String getQueueName() {
+    return queueName;
+  }
+
+  public void setQueueName(String queueName) {
+    this.queueName = queueName;
+  }
+
+  public int getExecutorMemory() {
+    return executorMemory;
+  }
+
+  public void setExecutorMemory(int executorMemory) {
+    this.executorMemory = executorMemory;
+  }
+
   @Override
   public String toString() {
     return "OptimizerConfig{" +
diff --git a/optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkOptimizer.java b/optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkOptimizer.java
index 8b2808692..151408e6b 100644
--- a/optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkOptimizer.java
+++ b/optimizer/src/main/java/com/netease/arctic/optimizer/flink/FlinkOptimizer.java
@@ -23,6 +23,7 @@ import com.netease.arctic.ams.api.properties.OptimizerProperties;
 import com.netease.arctic.optimizer.OptimizerConfig;
 import com.netease.arctic.optimizer.StatefulOptimizer;
 import com.netease.arctic.optimizer.operator.BaseToucher;
+import com.netease.arctic.optimizer.util.OptimizerUtil;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.configuration.Configuration;
@@ -225,6 +226,11 @@ public class FlinkOptimizer implements StatefulOptimizer {
     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());
     OptimizerConfig optimizerConfig = new OptimizerConfig(args);
 
+    if (optimizerConfig.getOptimizerId() == null || optimizerConfig.getOptimizerId().isEmpty() ||
+        "unknown".equals(optimizerConfig.getOptimizerId())) {
+      OptimizerUtil.register(optimizerConfig);
+    }
+
     env.addSource(new FlinkConsumer(optimizerConfig))
         .setParallelism(optimizerConfig.getExecutorParallel())
         .process(new FlinkExecuteFunction(optimizerConfig))
diff --git a/optimizer/src/main/java/com/netease/arctic/optimizer/local/LocalOptimizer.java b/optimizer/src/main/java/com/netease/arctic/optimizer/local/LocalOptimizer.java
index 3af4dacd7..0b8ac45b9 100644
--- a/optimizer/src/main/java/com/netease/arctic/optimizer/local/LocalOptimizer.java
+++ b/optimizer/src/main/java/com/netease/arctic/optimizer/local/LocalOptimizer.java
@@ -29,6 +29,7 @@ import com.netease.arctic.optimizer.operator.BaseTaskConsumer;
 import com.netease.arctic.optimizer.operator.BaseTaskExecutor;
 import com.netease.arctic.optimizer.operator.BaseTaskReporter;
 import com.netease.arctic.optimizer.operator.BaseToucher;
+import com.netease.arctic.optimizer.util.OptimizerUtil;
 import org.kohsuke.args4j.CmdLineException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -150,6 +151,10 @@ public class LocalOptimizer implements StatefulOptimizer {
 
   public static void main(String[] args) throws CmdLineException {
     OptimizerConfig optimizerConfig = new OptimizerConfig(args);
+    if (optimizerConfig.getOptimizerId() == null || optimizerConfig.getOptimizerId().isEmpty() ||
+        "unknown".equals(optimizerConfig.getOptimizerId())) {
+      OptimizerUtil.register(optimizerConfig);
+    }
     new LocalOptimizer().init(optimizerConfig);
     LOG.info("init LocalOptimizer with {}", optimizerConfig);
   }
diff --git a/optimizer/src/main/java/com/netease/arctic/optimizer/util/OptimizerUtil.java b/optimizer/src/main/java/com/netease/arctic/optimizer/util/OptimizerUtil.java
new file mode 100644
index 000000000..4fb21e3d5
--- /dev/null
+++ b/optimizer/src/main/java/com/netease/arctic/optimizer/util/OptimizerUtil.java
@@ -0,0 +1,51 @@
+/*
+ * 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.optimizer.util;
+
+import com.netease.arctic.ams.api.OptimizeManager;
+import com.netease.arctic.ams.api.OptimizerDescriptor;
+import com.netease.arctic.ams.api.OptimizerRegisterInfo;
+import com.netease.arctic.ams.api.client.OptimizeManagerClientPools;
+import com.netease.arctic.optimizer.OptimizerConfig;
+import org.apache.iceberg.exceptions.ValidationException;
+
+public class OptimizerUtil {
+
+  public static void register(OptimizerConfig config) {
+    ValidationException.check(config.getAmsUrl() != null && !config.getAmsUrl().isEmpty(), "ams url is empty");
+    ValidationException.check(
+        config.getQueueName() != null && !config.getQueueName().isEmpty(),
+        "optimizer name is empty");
+    ValidationException.check(config.getExecutorParallel() > 0, "optimizer executor parallel must be greater than 0");
+    ValidationException.check(config.getExecutorMemory() > 0, "optimizer executor memory must be greater than 0");
+
+    OptimizeManager.Iface client = OptimizeManagerClientPools.getClient(config.getAmsUrl());
+    OptimizerRegisterInfo info = new OptimizerRegisterInfo();
+    info.setOptimizerGroupName(config.getQueueName());
+    info.setCoreNumber(config.getExecutorParallel());
+    info.setMemorySize(config.getExecutorMemory());
+    try {
+      OptimizerDescriptor descriptor = client.registerOptimizer(info);
+      config.setOptimizerId(String.valueOf(descriptor.getOptimizerId()));
+      config.setQueueId(descriptor.getGroupId());
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+}
diff --git a/pom.xml b/pom.xml
index 677703f4f..8855dc95b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -282,6 +282,11 @@
                     <groupId>org.apache.maven.plugins</groupId>
                     <artifactId>maven-surefire-plugin</artifactId>
                 </plugin>
+                <plugin>
+                    <groupId>org.apache.maven.plugins</groupId>
+                    <artifactId>maven-shade-plugin</artifactId>
+                    <version>3.2.4</version>
+                </plugin>
             </plugins>
         </pluginManagement>
 
diff --git a/site/docs/ch/optimizers.md b/site/docs/ch/optimizers.md
index c7be2df13..71e0b9ac5 100644
--- a/site/docs/ch/optimizers.md
+++ b/site/docs/ch/optimizers.md
@@ -127,3 +127,28 @@ AMS 中添加自定义的 Optimizer Container，需要在`conf/config.yaml`中`c
 
 用户可以灵活定义 Optimizer Container 和 Optimizer Group 中的 properties，在构造 Optimizer 时，这些 properties 都会传递到 Optimizer，具体可以参考`com.netease.arctic.ams.server.service.impl.OptimizeExecuteService.startOptimizer`。
 
+### 自定义 Optimizer 的生命周期管理
+
+首先用户需要在AMS中配置Container和Optimizer group(参考[新增 optimizer group](meta-service/dashboard.md#flink))，其中container类型为external，
+且Container以及Optimizer group的properties无需配置。
+
+用户可以在本身已有的任务调度系统中管理Optimizer生命周期，只需提供以下Optimizer启动所需参数即可自动注册并启动Optimizer：
+```text
+AMS_THRIFT_SERVER_URL：AMS Thrift Server地址
+OPTIMIZE_GROUP_NAME：Optimizer所属Optimizer Group名称，AMS中可见
+EXECUTOR_PARALLELISM：Optimizer并行度
+EXECUTOR_MEMORY：Optimizer内存，一般在flink场景下EXECUTOR_MEMORY = EXECUTOR_JOBMANAGER_MEMORY + EXECUTOR_PARALLELISM * EXECUTOR_TASKMANAGER_MEMORY
+```
+
+下面以FlinkOptimizer为例介绍如何通过用户任务调度系统管理和注册Optimizer。
+
+用户可以通过已有调度系统启动FlinkOptimizer，例如flink run方式提交任务如下：
+
+```shell
+flink run -m yarn-cluster  -ytm {EXECUTOR_TASKMANAGER_MEMORY} 
+-yjm {EXECUTOR_JOBMANAGER_MEMORY} 
+-c com.netease.arctic.optimizer.flink.FlinkOptimizer 
+/ARCTIC_HOME/plugin/optimize/OptimizeJob.jar -a {AMS_THRIFT_SERVER_URL} 
+-qn {OPTIMIZE_GROUP_NAME} -p {EXECUTOR_PARALLELISM} -m {EXECUTOR_MEMORY} 
+--heart-beat 60000
+```


Test patch:
diff --git a/ams/ams-api/src/test/java/com/netease/arctic/ams/api/MockArcticMetastoreServer.java b/ams/ams-api/src/test/java/com/netease/arctic/ams/api/MockArcticMetastoreServer.java
index cafc676b8..3df4aa0da 100644
--- a/ams/ams-api/src/test/java/com/netease/arctic/ams/api/MockArcticMetastoreServer.java
+++ b/ams/ams-api/src/test/java/com/netease/arctic/ams/api/MockArcticMetastoreServer.java
@@ -54,6 +54,8 @@ public class MockArcticMetastoreServer implements Runnable {
   private final Object lock = new Object();
   private final AmsHandler amsHandler = new AmsHandler();
 
+  private final OptimizeManagerHandler optimizeManagerHandler = new OptimizeManagerHandler();
+
   private TServer server;
 
   private static final MockArcticMetastoreServer INSTANCE = new MockArcticMetastoreServer();
@@ -153,6 +155,10 @@ public class MockArcticMetastoreServer implements Runnable {
           new ArcticTableMetastore.Processor<>(amsHandler);
       processor.registerProcessor("TableMetastore", amsProcessor);
 
+      OptimizeManager.Processor<OptimizeManagerHandler> optimizerManProcessor =
+          new OptimizeManager.Processor<>(optimizeManagerHandler);
+      processor.registerProcessor("OptimizeManager", optimizerManProcessor);
+
       TThreadedSelectorServer.Args args = new TThreadedSelectorServer.Args(serverTransport)
           .processor(processor)
           .transportFactory(new TFramedTransport.Factory())
@@ -322,4 +328,36 @@ public class MockArcticMetastoreServer implements Runnable {
       }
     }
   }
+
+  public class OptimizeManagerHandler implements OptimizeManager.Iface {
+
+    public void cleanUp() {
+    }
+
+    @Override
+    public void ping() throws TException {
+
+    }
+
+    @Override
+    public OptimizeTask pollTask(int queueId, JobId jobId, String attemptId, long waitTime)
+        throws NoSuchObjectException, TException {
+      return null;
+    }
+
+    @Override
+    public void reportOptimizeResult(OptimizeTaskStat optimizeTaskStat) throws TException {
+
+    }
+
+    @Override
+    public void reportOptimizerState(OptimizerStateReport reportData) throws TException {
+
+    }
+
+    @Override
+    public OptimizerDescriptor registerOptimizer(OptimizerRegisterInfo registerInfo) throws TException {
+      return new OptimizerDescriptor();
+    }
+  }
 }
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java
index 11bbcc2f4..791de87a0 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java
@@ -42,6 +42,7 @@ import com.netease.arctic.ams.server.service.MetaService;
 import com.netease.arctic.ams.server.service.ServiceContainer;
 import com.netease.arctic.ams.server.service.TestDDLTracerService;
 import com.netease.arctic.ams.server.service.TestFileInfoCacheService;
+import com.netease.arctic.ams.server.service.TestOptimizerService;
 import com.netease.arctic.ams.server.service.impl.AdaptHiveService;
 import com.netease.arctic.ams.server.service.TestSupportHiveSyncService;
 import com.netease.arctic.ams.server.service.impl.ArcticTransactionService;
@@ -49,6 +50,8 @@ import com.netease.arctic.ams.server.service.impl.CatalogMetadataService;
 import com.netease.arctic.ams.server.service.impl.DDLTracerService;
 import com.netease.arctic.ams.server.service.impl.FileInfoCacheService;
 import com.netease.arctic.ams.server.service.impl.JDBCMetaService;
+import com.netease.arctic.ams.server.service.impl.OptimizeQueueService;
+import com.netease.arctic.ams.server.service.impl.OptimizerService;
 import com.netease.arctic.ams.server.util.DerbyTestUtil;
 import com.netease.arctic.ams.server.utils.CatalogUtil;
 import com.netease.arctic.ams.server.utils.JDBCSqlSessionFactoryProvider;
@@ -100,7 +103,8 @@ import static org.powermock.api.mockito.PowerMockito.when;
     TestSupportHiveMajorOptimizeCommit.class,
     TestSupportHiveSyncService.class,
     TestExpiredFileCleanSupportHive.class,
-    TestOrphanFileCleanSupportHive.class})
+    TestOrphanFileCleanSupportHive.class,
+    TestOptimizerService.class})
 @PrepareForTest({
     JDBCSqlSessionFactoryProvider.class,
     ArcticMetaStore.class,
@@ -170,6 +174,11 @@ public class AmsTestBase {
     configuration.setString(ArcticMetaStoreConf.DB_TYPE, "derby");
     ArcticMetaStore.conf = configuration;
 
+    OptimizeQueueService optimizeQueueService = new OptimizeQueueService();
+    when(ServiceContainer.getOptimizeQueueService()).thenReturn(optimizeQueueService);
+    OptimizerService optimizerService = new OptimizerService();
+    when(ServiceContainer.getOptimizerService()).thenReturn(optimizerService);
+
     //create
     createCatalog();
   }
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestOptimizerService.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestOptimizerService.java
new file mode 100644
index 000000000..591ea0a6f
--- /dev/null
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestOptimizerService.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *  *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *  *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.ams.server.service;
+
+import com.google.common.collect.Maps;
+import com.netease.arctic.ams.api.InvalidObjectException;
+import com.netease.arctic.ams.api.MetaException;
+import com.netease.arctic.ams.api.NoSuchObjectException;
+import com.netease.arctic.ams.api.OptimizerDescriptor;
+import com.netease.arctic.ams.api.OptimizerRegisterInfo;
+import com.netease.arctic.ams.server.ArcticMetaStore;
+import com.netease.arctic.ams.server.model.OptimizeQueueMeta;
+import com.netease.arctic.ams.server.model.TableTaskStatus;
+import com.netease.arctic.ams.server.service.impl.OptimizeQueueService;
+import com.netease.arctic.ams.server.service.impl.OptimizerService;
+import com.netease.arctic.ams.server.utils.JDBCSqlSessionFactoryProvider;
+import java.util.List;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+
+import java.util.Map;
+
+import static org.powermock.api.mockito.PowerMockito.when;
+
+@PowerMockIgnore({"javax.management.*"})
+@PrepareForTest({
+    ServiceContainer.class,
+    JDBCSqlSessionFactoryProvider.class,
+    OptimizerService.class,
+    ArcticMetaStore.class
+})
+public class TestOptimizerService {
+
+  private static final String optimizerGroupName = "testGroup";
+  private static final String containerName = "testContainer";
+
+  @BeforeClass
+  public static void before() throws MetaException, NoSuchObjectException {
+    OptimizeQueueMeta optimizeQueueMeta = new OptimizeQueueMeta();
+    optimizeQueueMeta.setName(optimizerGroupName);
+    optimizeQueueMeta.setContainer(containerName);
+    Map<String, String> properties = Maps.newHashMap();
+    optimizeQueueMeta.setProperties(properties);
+    ServiceContainer.getOptimizeQueueService().createQueue(optimizeQueueMeta);
+  }
+
+  @Test
+  public void testRegisterOptimizer() throws MetaException, InvalidObjectException {
+    OptimizerRegisterInfo optimizerRegisterInfo = new OptimizerRegisterInfo();
+    optimizerRegisterInfo.setOptimizerGroupName(optimizerGroupName);
+    optimizerRegisterInfo.setCoreNumber(1);
+    optimizerRegisterInfo.setMemorySize(1);
+    OptimizerDescriptor descriptor = ServiceContainer.getOptimizerService().registerOptimizer(optimizerRegisterInfo);
+    int groupId = ServiceContainer.getOptimizerService().getOptimizerGroupInfo(optimizerGroupName).getId();
+    Assert.assertEquals(descriptor.getGroupId(), groupId);
+    Assert.assertEquals(descriptor.getGroupName(), optimizerGroupName);
+    Assert.assertEquals(descriptor.getContainer(), containerName);
+    Assert.assertEquals(descriptor.getStatus(), TableTaskStatus.STARTING.name());
+  }
+}
diff --git a/core/src/test/java/com/netease/arctic/catalog/IcebergHadoopCatalogTest.java b/core/src/test/java/com/netease/arctic/catalog/IcebergHadoopCatalogTest.java
new file mode 100644
index 000000000..6ab416961
--- /dev/null
+++ b/core/src/test/java/com/netease/arctic/catalog/IcebergHadoopCatalogTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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.catalog;
+
+import com.google.common.collect.Maps;
+import com.netease.arctic.TableTestBase;
+import com.netease.arctic.ams.api.CatalogMeta;
+import com.netease.arctic.ams.api.MockArcticMetastoreServer;
+import com.netease.arctic.ams.api.properties.CatalogMetaProperties;
+import com.netease.arctic.table.ArcticTable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.thrift.TException;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_HADOOP;
+import static org.apache.iceberg.CatalogUtil.ICEBERG_CATALOG_TYPE;
+import static org.apache.iceberg.CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP;
+
+public class IcebergHadoopCatalogTest {
+
+  @ClassRule
+  public static final TemporaryFolder tempFolder = new TemporaryFolder();
+  private static final String ICEBERG_HADOOP_CATALOG_NAME = "iceberg_hadoop";
+
+  @BeforeClass
+  public static void createIcebergCatalog() throws IOException {
+    Map<String, String> storageConfig = new HashMap<>();
+    storageConfig.put(
+        CatalogMetaProperties.STORAGE_CONFIGS_KEY_TYPE,
+        CatalogMetaProperties.STORAGE_CONFIGS_VALUE_TYPE_HDFS);
+    storageConfig.put(CatalogMetaProperties.STORAGE_CONFIGS_KEY_CORE_SITE, MockArcticMetastoreServer.getHadoopSite());
+    storageConfig.put(CatalogMetaProperties.STORAGE_CONFIGS_KEY_HDFS_SITE, MockArcticMetastoreServer.getHadoopSite());
+
+    Map<String, String> authConfig = new HashMap<>();
+    authConfig.put(CatalogMetaProperties.AUTH_CONFIGS_KEY_TYPE,
+        CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_SIMPLE);
+    authConfig.put(CatalogMetaProperties.AUTH_CONFIGS_KEY_HADOOP_USERNAME,
+        System.getProperty("user.name"));
+
+    tempFolder.create();
+    Map<String, String> catalogProperties = new HashMap<>();
+    catalogProperties.put(CatalogProperties.WAREHOUSE_LOCATION, tempFolder.newFolder().getPath());
+    catalogProperties.put(CatalogMetaProperties.TABLE_FORMATS, "iceberg");
+
+    CatalogMeta catalogMeta = new CatalogMeta(ICEBERG_HADOOP_CATALOG_NAME, CATALOG_TYPE_HADOOP,
+        storageConfig, authConfig, catalogProperties);
+    MockArcticMetastoreServer.getInstance().handler().createCatalog(catalogMeta);
+
+    MockArcticMetastoreServer.getInstance().createCatalogIfAbsent(catalogMeta);
+  }
+
+  @Test
+  public void testLoadCatalog() {
+    ArcticCatalog icebergCatalog =
+        CatalogLoader.load(MockArcticMetastoreServer.getInstance().getUrl(ICEBERG_HADOOP_CATALOG_NAME));
+    Assert.assertEquals(ICEBERG_HADOOP_CATALOG_NAME, icebergCatalog.name());
+    Assert.assertEquals(0, icebergCatalog.listDatabases().size());
+  }
+
+  @Test
+  public void testCreateAndDropDatabase() {
+    ArcticCatalog icebergCatalog =
+        CatalogLoader.load(MockArcticMetastoreServer.getInstance().getUrl(ICEBERG_HADOOP_CATALOG_NAME));
+    icebergCatalog.createDatabase("db1");
+    Assert.assertEquals(1, icebergCatalog.listDatabases().size());
+    Assert.assertEquals("db1", icebergCatalog.listDatabases().get(0));
+    icebergCatalog.dropDatabase("db1");
+    Assert.assertEquals(0, icebergCatalog.listDatabases().size());
+  }
+
+  @Test
+  public void testLoadIcebergTable() throws TException {
+    ArcticCatalog icebergCatalog =
+        CatalogLoader.load(MockArcticMetastoreServer.getInstance().getUrl(ICEBERG_HADOOP_CATALOG_NAME));
+    icebergCatalog.createDatabase("db2");
+    CatalogMeta catalogMeta = MockArcticMetastoreServer.getInstance().handler().getCatalog(ICEBERG_HADOOP_CATALOG_NAME);
+    Map<String, String> catalogProperties = Maps.newHashMap(catalogMeta.getCatalogProperties());
+    catalogProperties.put(ICEBERG_CATALOG_TYPE, ICEBERG_CATALOG_TYPE_HADOOP);
+    Catalog nativeIcebergTable = org.apache.iceberg.CatalogUtil.buildIcebergCatalog(ICEBERG_HADOOP_CATALOG_NAME,
+        catalogProperties, new Configuration());
+    nativeIcebergTable.createTable(TableIdentifier.of("db2", "tb1"), TableTestBase.TABLE_SCHEMA);
+    ArcticTable table = icebergCatalog.loadTable(
+        com.netease.arctic.table.TableIdentifier.of(ICEBERG_HADOOP_CATALOG_NAME, "db2", "tb1"));
+    Assert.assertTrue(table instanceof BaseIcebergCatalog.BaseIcebergTable);
+    Assert.assertEquals(true, table.isUnkeyedTable());
+    Assert.assertEquals(TableTestBase.TABLE_SCHEMA.asStruct(), table.schema().asStruct());
+    nativeIcebergTable.dropTable(TableIdentifier.of("db2", "tb1"), true);
+    icebergCatalog.dropDatabase("db2");
+  }
+
+}
diff --git a/core/src/test/java/com/netease/arctic/io/TaskReaderTest.java b/core/src/test/java/com/netease/arctic/io/TaskReaderTest.java
index f726a307e..d38d61382 100644
--- a/core/src/test/java/com/netease/arctic/io/TaskReaderTest.java
+++ b/core/src/test/java/com/netease/arctic/io/TaskReaderTest.java
@@ -19,17 +19,30 @@
 package com.netease.arctic.io;
 
 import com.google.common.collect.Sets;
+import com.netease.arctic.io.reader.BaseIcebergDataReader;
 import com.netease.arctic.io.reader.BaseIcebergPosDeleteReader;
 import com.netease.arctic.io.reader.GenericArcticDataReader;
+import com.netease.arctic.io.reader.GenericIcebergDataReader;
+import com.netease.arctic.scan.ArcticFileScanTask;
+import com.netease.arctic.scan.BaseArcticFileScanTask;
 import com.netease.arctic.scan.CombinedScanTask;
 import com.netease.arctic.scan.KeyedTableScanTask;
+import java.util.Map;
+import java.util.function.Function;
+import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.MetadataColumns;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
 import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.IdentityPartitionConverters;
 import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.GenericParquetReaders;
 import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.parquet.ParquetValueReader;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.types.Types;
+import org.apache.parquet.schema.MessageType;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -66,6 +79,40 @@ public class TaskReaderTest extends TableTestBaseWithInitData {
     Assert.assertEquals(rightIds, resultIds);
   }
 
+  @Test
+  public void testReadChange(){
+    Table changeTable = testKeyedTable.changeTable();
+    CloseableIterable<FileScanTask> fileScanTasks = changeTable.newScan().planFiles();
+    CloseableIterable<ArcticFileScanTask> arcticFileScanTasks = CloseableIterable.transform(
+        fileScanTasks, fileScanTask -> new BaseArcticFileScanTask(fileScanTask)
+    );
+    Schema schema = changeTable.schema();
+    List<Types.NestedField> columns = schema.columns().stream().collect(Collectors.toList());
+    columns.add(com.netease.arctic.table.MetadataColumns.TRANSACTION_ID_FILED);
+    columns.add(com.netease.arctic.table.MetadataColumns.FILE_OFFSET_FILED);
+    columns.add(com.netease.arctic.table.MetadataColumns.CHANGE_ACTION_FIELD);
+    Schema externalSchema = new Schema(columns);
+
+    GenericIcebergDataReader genericIcebergDataReader = new GenericIcebergDataReader(
+        testKeyedTable.io(),
+        externalSchema,
+        externalSchema,
+        null,
+        false,
+        IdentityPartitionConverters::convertConstant,
+        false
+    );
+
+    ImmutableList.Builder<Record> builder = ImmutableList.builder();
+    for (ArcticFileScanTask arcticFileScanTask: arcticFileScanTasks){
+      builder.addAll(genericIcebergDataReader.readData(arcticFileScanTask));
+    }
+    List<Record> records = builder.build();
+    for (Record record: records) {
+      Assert.assertTrue(record.size() == 6);
+    }
+  }
+
   @Test
   public void testReadPosDelete() {
     BaseIcebergPosDeleteReader baseIcebergPosDeleteReader =
diff --git a/hive/src/test/java/com/netease/arctic/hive/catalog/IcebergHiveCatalogTest.java b/hive/src/test/java/com/netease/arctic/hive/catalog/IcebergHiveCatalogTest.java
new file mode 100644
index 000000000..e3f4e87ba
--- /dev/null
+++ b/hive/src/test/java/com/netease/arctic/hive/catalog/IcebergHiveCatalogTest.java
@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.hive.catalog;
+
+import com.google.common.collect.Maps;
+import com.netease.arctic.CatalogMetaTestUtil;
+import com.netease.arctic.TableTestBase;
+import com.netease.arctic.ams.api.CatalogMeta;
+import com.netease.arctic.ams.api.MockArcticMetastoreServer;
+import com.netease.arctic.ams.api.properties.CatalogMetaProperties;
+import com.netease.arctic.catalog.ArcticCatalog;
+import com.netease.arctic.catalog.BaseIcebergCatalog;
+import com.netease.arctic.catalog.CatalogLoader;
+import com.netease.arctic.hive.HiveTableTestBase;
+import com.netease.arctic.table.ArcticTable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.thrift.TException;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_HIVE;
+import static org.apache.iceberg.CatalogUtil.ICEBERG_CATALOG_TYPE;
+import static org.apache.iceberg.CatalogUtil.ICEBERG_CATALOG_TYPE_HIVE;
+
+public class IcebergHiveCatalogTest extends HiveTableTestBase {
+
+  private static final String ICEBERG_HIVE_CATALOG_NAME = "iceberg_hive";
+
+  @BeforeClass
+  public static void createIcebergCatalog() throws IOException {
+    Map<String, String> storageConfig = new HashMap<>();
+    storageConfig.put(
+        CatalogMetaProperties.STORAGE_CONFIGS_KEY_TYPE,
+        CatalogMetaProperties.STORAGE_CONFIGS_VALUE_TYPE_HDFS);
+    storageConfig.put(CatalogMetaProperties.STORAGE_CONFIGS_KEY_CORE_SITE, MockArcticMetastoreServer.getHadoopSite());
+    storageConfig.put(CatalogMetaProperties.STORAGE_CONFIGS_KEY_HDFS_SITE, MockArcticMetastoreServer.getHadoopSite());
+    storageConfig.put(CatalogMetaProperties.STORAGE_CONFIGS_KEY_HIVE_SITE,
+        CatalogMetaTestUtil.encodingSite(hms.hiveConf()));
+
+    Map<String, String> authConfig = new HashMap<>();
+    authConfig.put(CatalogMetaProperties.AUTH_CONFIGS_KEY_TYPE,
+        CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_SIMPLE);
+    authConfig.put(CatalogMetaProperties.AUTH_CONFIGS_KEY_HADOOP_USERNAME,
+        System.getProperty("user.name"));
+
+    tempFolder.create();
+    Map<String, String> catalogProperties = new HashMap<>();
+    catalogProperties.put(CatalogMetaProperties.TABLE_FORMATS, "iceberg");
+
+    CatalogMeta catalogMeta = new CatalogMeta(ICEBERG_HIVE_CATALOG_NAME, CATALOG_TYPE_HIVE,
+        storageConfig, authConfig, catalogProperties);
+    MockArcticMetastoreServer.getInstance().handler().createCatalog(catalogMeta);
+
+    MockArcticMetastoreServer.getInstance().createCatalogIfAbsent(catalogMeta);
+  }
+
+  @Test
+  public void testLoadCatalog() {
+    ArcticCatalog icebergCatalog =
+        CatalogLoader.load(MockArcticMetastoreServer.getInstance().getUrl(ICEBERG_HIVE_CATALOG_NAME));
+    Assert.assertEquals(ICEBERG_HIVE_CATALOG_NAME, icebergCatalog.name());
+    Assert.assertEquals(2, icebergCatalog.listDatabases().size());
+  }
+
+  @Test
+  public void testCreateAndDropDatabase() {
+    ArcticCatalog icebergCatalog =
+        CatalogLoader.load(MockArcticMetastoreServer.getInstance().getUrl(ICEBERG_HIVE_CATALOG_NAME));
+    icebergCatalog.createDatabase("db1");
+    Assert.assertEquals(3, icebergCatalog.listDatabases().size());
+    Assert.assertEquals("db1", icebergCatalog.listDatabases().get(0));
+    icebergCatalog.dropDatabase("db1");
+    Assert.assertEquals(2, icebergCatalog.listDatabases().size());
+  }
+
+  @Test
+  public void testLoadIcebergTable() throws TException {
+    ArcticCatalog icebergCatalog =
+        CatalogLoader.load(MockArcticMetastoreServer.getInstance().getUrl(ICEBERG_HIVE_CATALOG_NAME));
+    icebergCatalog.createDatabase("db2");
+    CatalogMeta catalogMeta = MockArcticMetastoreServer.getInstance().handler().getCatalog(ICEBERG_HIVE_CATALOG_NAME);
+    Map<String, String> catalogProperties = Maps.newHashMap(catalogMeta.getCatalogProperties());
+    catalogProperties.put(ICEBERG_CATALOG_TYPE, ICEBERG_CATALOG_TYPE_HIVE);
+    Catalog nativeIcebergTable = org.apache.iceberg.CatalogUtil.buildIcebergCatalog(
+        ICEBERG_HIVE_CATALOG_NAME,
+        catalogProperties, new Configuration());
+    nativeIcebergTable.createTable(TableIdentifier.of("db2", "tb1"), TableTestBase.TABLE_SCHEMA);
+    ArcticTable table = icebergCatalog.loadTable(
+        com.netease.arctic.table.TableIdentifier.of(ICEBERG_HIVE_CATALOG_NAME, "db2", "tb1"));
+    Assert.assertTrue(table instanceof BaseIcebergCatalog.BaseIcebergTable);
+    Assert.assertEquals(true, table.isUnkeyedTable());
+    Assert.assertEquals(TableTestBase.TABLE_SCHEMA.asStruct(), table.schema().asStruct());
+    nativeIcebergTable.dropTable(TableIdentifier.of("db2", "tb1"), true);
+    icebergCatalog.dropDatabase("db2");
+  }
+
+}
diff --git a/hive/src/test/java/com/netease/arctic/hive/io/TestAdaptHiveReader.java b/hive/src/test/java/com/netease/arctic/hive/io/TestAdaptHiveReader.java
new file mode 100644
index 000000000..a4a2bccb8
--- /dev/null
+++ b/hive/src/test/java/com/netease/arctic/hive/io/TestAdaptHiveReader.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.hive.io;
+
+import com.netease.arctic.hive.HiveTableTestBase;
+import com.netease.arctic.hive.table.HiveLocationKind;
+import java.io.IOException;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.junit.Test;
+
+import static com.netease.arctic.hive.io.TestIOUtils.testWrite;
+
+public class TestAdaptHiveReader extends HiveTableTestBase {
+
+  @Test
+  public void testInt96WithoutTZPredicatePushDown() throws IOException {
+    testWrite(testKeyedHiveTable, HiveLocationKind.INSTANT, HiveTestRecords.baseRecords(), "hive",
+        Expressions.equal("op_time", "2022-01-04T12:00:00"),
+        ImmutableList.of(HiveTestRecords.baseRecords().get(1)));
+  }
+
+  @Test
+  public void testInt96WithTZPredicatePushDown() throws IOException {
+    testWrite(testKeyedHiveTable, HiveLocationKind.INSTANT, HiveTestRecords.baseRecords(), "hive",
+        Expressions.equal("op_time_with_zone", "2022-01-04T13:00:00+01:00"),
+        ImmutableList.of(HiveTestRecords.baseRecords().get(1)));
+  }
+
+}
diff --git a/hive/src/test/java/com/netease/arctic/hive/io/TestAdaptHiveWriter.java b/hive/src/test/java/com/netease/arctic/hive/io/TestAdaptHiveWriter.java
index ec5c4c462..915919ab4 100644
--- a/hive/src/test/java/com/netease/arctic/hive/io/TestAdaptHiveWriter.java
+++ b/hive/src/test/java/com/netease/arctic/hive/io/TestAdaptHiveWriter.java
@@ -28,14 +28,18 @@ import com.netease.arctic.table.BaseLocationKind;
 import com.netease.arctic.table.ChangeLocationKind;
 import com.netease.arctic.table.LocationKind;
 import com.netease.arctic.table.WriteOperationKind;
+import java.time.LocalDateTime;
 import org.apache.iceberg.Files;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.data.parquet.AdaptHiveGenericParquetReaders;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.TaskWriter;
 import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.parquet.AdaptHiveParquet;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
 import org.junit.Assert;
 import org.junit.Test;
@@ -48,6 +52,8 @@ import java.util.List;
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import static com.netease.arctic.hive.io.TestIOUtils.testWrite;
+
 public class TestAdaptHiveWriter extends HiveTableTestBase {
 
   @Test
@@ -148,35 +154,4 @@ public class TestAdaptHiveWriter extends HiveTableTestBase {
   public void testUnPartitionUnKeyedTableHiveWriteByLocationKind() throws IOException {
     testWrite(testUnPartitionHiveTable, HiveLocationKind.INSTANT, HiveTestRecords.baseRecords(), "hive");
   }
-
-  public void testWrite(ArcticTable table, LocationKind locationKind, List<Record> records, String pathFeature) throws IOException {
-    AdaptHiveGenericTaskWriterBuilder builder = AdaptHiveGenericTaskWriterBuilder
-        .builderFor(table)
-        .withTransactionId(table.isKeyedTable() ? 1L : null);
-
-    TaskWriter<Record> changeWrite = builder.buildWriter(locationKind);
-    for (Record record: records) {
-      changeWrite.write(record);
-    }
-    WriteResult complete = changeWrite.complete();
-    Arrays.stream(complete.dataFiles()).forEach(s -> Assert.assertTrue(s.path().toString().contains(pathFeature)));
-    CloseableIterable<Record> concat =
-        CloseableIterable.concat(Arrays.stream(complete.dataFiles()).map(s -> readParquet(
-            table.schema(),
-            s.path().toString())).collect(Collectors.toList()));
-    Set<Record> result = new HashSet<>();
-    Iterators.addAll(result, concat.iterator());
-    Assert.assertEquals(result, records.stream().collect(Collectors.toSet()));
-  }
-
-  private CloseableIterable<Record> readParquet(Schema schema, String path){
-    AdaptHiveParquet.ReadBuilder builder = AdaptHiveParquet.read(
-        Files.localInput(path))
-        .project(schema)
-        .createReaderFunc(fileSchema -> AdaptHiveGenericParquetReaders.buildReader(schema, fileSchema, new HashMap<>()))
-        .caseSensitive(false);
-
-    CloseableIterable<Record> iterable = builder.build();
-    return iterable;
-  }
 }
diff --git a/hive/src/test/java/com/netease/arctic/hive/io/TestIOUtils.java b/hive/src/test/java/com/netease/arctic/hive/io/TestIOUtils.java
new file mode 100644
index 000000000..09601509e
--- /dev/null
+++ b/hive/src/test/java/com/netease/arctic/hive/io/TestIOUtils.java
@@ -0,0 +1,87 @@
+/*
+ * 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.io;
+
+import com.netease.arctic.hive.io.writer.AdaptHiveGenericTaskWriterBuilder;
+import com.netease.arctic.table.ArcticTable;
+import com.netease.arctic.table.LocationKind;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.AdaptHiveGenericParquetReaders;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.TaskWriter;
+import org.apache.iceberg.io.WriteResult;
+import org.apache.iceberg.parquet.AdaptHiveParquet;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
+import org.junit.Assert;
+
+public class TestIOUtils {
+
+  public static void testWrite(ArcticTable table, LocationKind locationKind, List<Record> records, String pathFeature) throws IOException {
+    testWrite(table, locationKind, records, pathFeature, null, null);
+  }
+
+  public static void testWrite(
+      ArcticTable table, LocationKind locationKind, List<Record> records, String pathFeature,
+      Expression expression, List<Record> readRecords) throws IOException {
+    AdaptHiveGenericTaskWriterBuilder builder = AdaptHiveGenericTaskWriterBuilder
+        .builderFor(table)
+        .withTransactionId(table.isKeyedTable() ? 1L : null);
+
+    TaskWriter<Record> changeWrite = builder.buildWriter(locationKind);
+    for (Record record: records) {
+      changeWrite.write(record);
+    }
+    WriteResult complete = changeWrite.complete();
+    Arrays.stream(complete.dataFiles()).forEach(s -> Assert.assertTrue(s.path().toString().contains(pathFeature)));
+    CloseableIterable<Record> concat =
+        CloseableIterable.concat(Arrays.stream(complete.dataFiles()).map(s -> readParquet(
+            table.schema(),
+            s.path().toString(), expression)).collect(Collectors.toList()));
+    Set<Record> result = new HashSet<>();
+    Iterators.addAll(result, concat.iterator());
+    if (readRecords == null) {
+      Assert.assertEquals(result, new HashSet<>(records));
+    } else {
+      Assert.assertEquals(result, new HashSet<>(readRecords));
+    }
+  }
+
+  private static CloseableIterable<Record> readParquet(Schema schema, String path, Expression expression){
+    AdaptHiveParquet.ReadBuilder builder = AdaptHiveParquet.read(
+            Files.localInput(path))
+        .project(schema)
+        .filter(expression == null ? Expressions.alwaysTrue() : expression)
+        .createReaderFunc(fileSchema -> AdaptHiveGenericParquetReaders.buildReader(schema, fileSchema, new HashMap<>()))
+        .caseSensitive(false);
+
+    CloseableIterable<Record> iterable = builder.build();
+    return iterable;
+  }
+}