Repository: NetEase/arctic
Pull request #549 (2022-11-01 05:52:21)

[ARCTIC-260][Core]Support loading Iceberg catalog
<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://arctic.netease.com/ch/contribute/
  2. If the PR is related to an issue in https://github.com/NetEase/arctic/issues, add '[ARCTIC-XXXX]' in your PR title, e.g., '[ARCTIC-XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][ARCTIC #XXXX] Your PR title ...'.
-->

## Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about its use case.
  2. If you fix a bug, you can clarify why it is a bug.
-->

Subtask of #260.
Support loading iceberg catalog.

## Brief change log

  - *Add BaseIcebergCatalog implementation of ArcticCatalog for native iceberg usage*
  - *CatalogLoader support loading BaseIcebergCatalog*
  - *Add authrozation for BaseIcebergCatalog*

## 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**)



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


Test patch:
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/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;
+  }
+}