Repository: NetEase/arctic
Pull request #222 (2022-08-24 07:46:52)

Second phase of reconstruct read/write to adapt hive
Add flink adapt hive UT, Fix core HiveMetastoreClient bug, Resolve jar conflict in trino

link issue #151 


Patch:
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 0f587431b..95b4785f1 100644
--- a/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java
+++ b/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java
@@ -310,7 +310,9 @@ public class BaseArcticCatalog implements ArcticCatalog {
               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);
       }
     }
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 d9778a3de..56b8c4ef0 100644
--- a/core/src/main/java/com/netease/arctic/table/TableMetaStore.java
+++ b/core/src/main/java/com/netease/arctic/table/TableMetaStore.java
@@ -20,6 +20,7 @@ package com.netease.arctic.table;
 
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang.ArrayUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
@@ -371,6 +372,13 @@ public class TableMetaStore implements Serializable {
     configuration.set(CommonConfigurationKeys.IPC_CLIENT_FALLBACK_TO_SIMPLE_AUTH_ALLOWED_KEY, "true");
     //Enforce configuration resolve resources
     configuration.get(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY);
+
+    //Avoid check hive version
+    configuration.set("hive.metastore.schema.verification", "false");
+
+    //It will encounter error(Required table missing : "DBS" in Catalog "" Schema "") when there is not this param
+    configuration.set("datanucleus.schema.autoCreateAll", "true");
+
     return configuration;
   }
 
@@ -430,7 +438,8 @@ public class TableMetaStore implements Serializable {
     }
 
     public Builder withBase64MetaStoreSite(String encodedMetaStoreSite) {
-      this.metaStoreSite = Base64.getDecoder().decode(encodedMetaStoreSite);
+      this.metaStoreSite = StringUtils.isBlank(encodedMetaStoreSite) ? null :
+          Base64.getDecoder().decode(encodedMetaStoreSite);
       return this;
     }
 
@@ -445,7 +454,8 @@ public class TableMetaStore implements Serializable {
     }
 
     public Builder withBase64HdfsSite(String encodedHdfsSite) {
-      this.hdfsSite = Base64.getDecoder().decode(encodedHdfsSite);
+      this.hdfsSite = StringUtils.isBlank(encodedHdfsSite) ? null :
+          Base64.getDecoder().decode(encodedHdfsSite);
       return this;
     }
 
@@ -460,7 +470,8 @@ public class TableMetaStore implements Serializable {
     }
 
     public Builder withBase64CoreSite(String encodedCoreSite) {
-      this.coreSite = Base64.getDecoder().decode(encodedCoreSite);
+      this.coreSite = StringUtils.isBlank(encodedCoreSite) ? null :
+          Base64.getDecoder().decode(encodedCoreSite);
       return this;
     }
 
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/AdaptHiveFlinkParquetWriters.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/AdaptHiveFlinkParquetWriters.java
index 963192063..15ab28a8f 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/AdaptHiveFlinkParquetWriters.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/AdaptHiveFlinkParquetWriters.java
@@ -26,6 +26,7 @@ import org.apache.flink.table.data.StringData;
 import org.apache.flink.table.data.TimestampData;
 import org.apache.flink.table.types.logical.ArrayType;
 import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
 import org.apache.flink.table.types.logical.MapType;
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.table.types.logical.RowType.RowField;
@@ -51,12 +52,15 @@ import org.apache.parquet.schema.Type;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.time.Instant;
+import java.time.ZoneId;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.concurrent.TimeUnit;
 
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE;
+
 public class AdaptHiveFlinkParquetWriters {
   private AdaptHiveFlinkParquetWriters() {
   }
@@ -183,7 +187,12 @@ public class AdaptHiveFlinkParquetWriters {
         case INT64:
           return ParquetValueWriters.longs(desc);
         case INT96:
-          return new TimestampInt96Writer(desc);
+          LogicalTypeRoot typeRoot = logicalType.getTypeRoot();
+          if (typeRoot == TIMESTAMP_WITHOUT_TIME_ZONE) {
+            return new TimestampInt96Writer(desc);
+          } else {
+            return new TimestampTZInt96Writer(desc);
+          }
         case FLOAT:
           return ParquetValueWriters.floats(desc);
         case DOUBLE:
@@ -194,6 +203,42 @@ public class AdaptHiveFlinkParquetWriters {
     }
   }
 
+  private static class TimestampTZInt96Writer extends AdaptHivePrimitiveWriter<TimestampData> {
+
+    private static final long JULIAN_DAY_OF_EPOCH = 2440588L;
+    private static final long MICROS_PER_DAY = 86400000000L;
+    private static final long MILLIS_IN_DAY = TimeUnit.DAYS.toMillis(1);
+    private static final long NANOS_PER_MILLISECOND = TimeUnit.MILLISECONDS.toNanos(1);
+
+    public TimestampTZInt96Writer(ColumnDescriptor descriptor) {
+      super(descriptor);
+    }
+
+    /**
+     * Writes nano timestamps to parquet int96
+     */
+    void writeBinary(int repetitionLevel, int julianDay, long nanosOfDay) {
+      ByteBuffer buf = ByteBuffer.allocate(12);
+      buf.order(ByteOrder.LITTLE_ENDIAN);
+      buf.putLong(nanosOfDay);
+      buf.putInt(julianDay);
+      buf.flip();
+      column.writeBinary(repetitionLevel, Binary.fromConstantByteBuffer(buf));
+    }
+
+    void writeInstant(int repetitionLevel, Instant instant) {
+      long timestamp = instant.toEpochMilli();
+      int julianDay = (int) (timestamp / MILLIS_IN_DAY + 2440588L);
+      long nanosOfDay = timestamp % MILLIS_IN_DAY * NANOS_PER_MILLISECOND + instant.getNano() % NANOS_PER_MILLISECOND;
+      writeBinary(repetitionLevel, julianDay, nanosOfDay);
+    }
+
+    @Override
+    public void write(int repetitionLevel, TimestampData value) {
+      writeInstant(repetitionLevel, value.toInstant());
+    }
+  }
+
   private static class TimestampInt96Writer extends AdaptHivePrimitiveWriter<TimestampData> {
 
     private static final long JULIAN_DAY_OF_EPOCH = 2440588L;
@@ -226,7 +271,7 @@ public class AdaptHiveFlinkParquetWriters {
 
     @Override
     public void write(int repetitionLevel, TimestampData value) {
-      writeInstant(repetitionLevel, value.toInstant());
+      writeInstant(repetitionLevel, value.toLocalDateTime().atZone(ZoneId.systemDefault()).toInstant());
     }
   }
 
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
index 268a1eb49..2ad7177f0 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
@@ -18,11 +18,11 @@
 
 package com.netease.arctic.flink.write;
 
+import com.netease.arctic.hive.io.writer.AdaptHiveOperateToTableRelation;
+import com.netease.arctic.hive.io.writer.AdaptHiveOutputFileFactory;
 import com.netease.arctic.hive.table.HiveLocationKind;
 import com.netease.arctic.hive.table.SupportHive;
 import com.netease.arctic.hive.utils.HiveTableUtil;
-import com.netease.arctic.hive.write.AdaptHiveOperateToTableRelation;
-import com.netease.arctic.hive.write.AdaptHiveOutputFileFactory;
 import com.netease.arctic.io.writer.CommonOutputFileFactory;
 import com.netease.arctic.io.writer.OutputFileFactory;
 import com.netease.arctic.io.writer.SortedPosDeleteWriter;
diff --git a/flink/v1.14/flink/pom.xml b/flink/v1.14/flink/pom.xml
index d42c9febc..946ed2209 100644
--- a/flink/v1.14/flink/pom.xml
+++ b/flink/v1.14/flink/pom.xml
@@ -299,6 +299,19 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>com.netease.arctic</groupId>
+            <artifactId>arctic-hive</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.netease.arctic</groupId>
+            <artifactId>arctic-hive</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+            <classifier>tests</classifier>
+        </dependency>
 
         <dependency>
             <groupId>org.apache.flink</groupId>
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/AdaptHiveFlinkParquetWriters.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/AdaptHiveFlinkParquetWriters.java
index 8c99333f6..c1b7ca803 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/AdaptHiveFlinkParquetWriters.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/AdaptHiveFlinkParquetWriters.java
@@ -26,6 +26,7 @@ import org.apache.flink.table.data.StringData;
 import org.apache.flink.table.data.TimestampData;
 import org.apache.flink.table.types.logical.ArrayType;
 import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
 import org.apache.flink.table.types.logical.MapType;
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.table.types.logical.RowType.RowField;
@@ -51,19 +52,23 @@ import org.apache.parquet.schema.Type;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.time.Instant;
+import java.time.ZoneId;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.concurrent.TimeUnit;
 
+import static org.apache.flink.table.types.logical.LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE;
+
 public class AdaptHiveFlinkParquetWriters {
   private AdaptHiveFlinkParquetWriters() {
   }
 
   @SuppressWarnings("unchecked")
   public static <T> ParquetValueWriter<T> buildWriter(LogicalType schema, MessageType type) {
-    return (ParquetValueWriter<T>) ParquetWithFlinkSchemaVisitor.visit(schema, type, new WriteBuilder(type));
+    return (ParquetValueWriter<T>) ParquetWithFlinkSchemaVisitor.visit(schema, type,
+        new WriteBuilder(type));
   }
 
   private static class WriteBuilder extends ParquetWithFlinkSchemaVisitor<ParquetValueWriter<?>> {
@@ -182,7 +187,12 @@ public class AdaptHiveFlinkParquetWriters {
         case INT64:
           return ParquetValueWriters.longs(desc);
         case INT96:
-          return new TimestampInt96Writer(desc);
+          LogicalTypeRoot typeRoot = logicalType.getTypeRoot();
+          if (typeRoot == TIMESTAMP_WITHOUT_TIME_ZONE) {
+            return new TimestampInt96Writer(desc);
+          } else {
+            return new TimestampTZInt96Writer(desc);
+          }
         case FLOAT:
           return ParquetValueWriters.floats(desc);
         case DOUBLE:
@@ -193,6 +203,42 @@ public class AdaptHiveFlinkParquetWriters {
     }
   }
 
+  private static class TimestampTZInt96Writer extends AdaptHivePrimitiveWriter<TimestampData> {
+
+    private static final long JULIAN_DAY_OF_EPOCH = 2440588L;
+    private static final long MICROS_PER_DAY = 86400000000L;
+    private static final long MILLIS_IN_DAY = TimeUnit.DAYS.toMillis(1);
+    private static final long NANOS_PER_MILLISECOND = TimeUnit.MILLISECONDS.toNanos(1);
+
+    public TimestampTZInt96Writer(ColumnDescriptor descriptor) {
+      super(descriptor);
+    }
+
+    /**
+     * Writes nano timestamps to parquet int96
+     */
+    void writeBinary(int repetitionLevel, int julianDay, long nanosOfDay) {
+      ByteBuffer buf = ByteBuffer.allocate(12);
+      buf.order(ByteOrder.LITTLE_ENDIAN);
+      buf.putLong(nanosOfDay);
+      buf.putInt(julianDay);
+      buf.flip();
+      column.writeBinary(repetitionLevel, Binary.fromConstantByteBuffer(buf));
+    }
+
+    void writeInstant(int repetitionLevel, Instant instant) {
+      long timestamp = instant.toEpochMilli();
+      int julianDay = (int) (timestamp / MILLIS_IN_DAY + 2440588L);
+      long nanosOfDay = timestamp % MILLIS_IN_DAY * NANOS_PER_MILLISECOND + instant.getNano() % NANOS_PER_MILLISECOND;
+      writeBinary(repetitionLevel, julianDay, nanosOfDay);
+    }
+
+    @Override
+    public void write(int repetitionLevel, TimestampData value) {
+      writeInstant(repetitionLevel, value.toInstant());
+    }
+  }
+
   private static class TimestampInt96Writer extends AdaptHivePrimitiveWriter<TimestampData> {
 
     private static final long JULIAN_DAY_OF_EPOCH = 2440588L;
@@ -225,7 +271,7 @@ public class AdaptHiveFlinkParquetWriters {
 
     @Override
     public void write(int repetitionLevel, TimestampData value) {
-      writeInstant(repetitionLevel, value.toInstant());
+      writeInstant(repetitionLevel, value.toLocalDateTime().atZone(ZoneId.systemDefault()).toInstant());
     }
   }
 
diff --git a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
index 268a1eb49..2ad7177f0 100644
--- a/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
+++ b/flink/v1.14/flink/src/main/java/com/netease/arctic/flink/write/FlinkTaskWriterBuilder.java
@@ -18,11 +18,11 @@
 
 package com.netease.arctic.flink.write;
 
+import com.netease.arctic.hive.io.writer.AdaptHiveOperateToTableRelation;
+import com.netease.arctic.hive.io.writer.AdaptHiveOutputFileFactory;
 import com.netease.arctic.hive.table.HiveLocationKind;
 import com.netease.arctic.hive.table.SupportHive;
 import com.netease.arctic.hive.utils.HiveTableUtil;
-import com.netease.arctic.hive.write.AdaptHiveOperateToTableRelation;
-import com.netease.arctic.hive.write.AdaptHiveOutputFileFactory;
 import com.netease.arctic.io.writer.CommonOutputFileFactory;
 import com.netease.arctic.io.writer.OutputFileFactory;
 import com.netease.arctic.io.writer.SortedPosDeleteWriter;
diff --git a/hive/pom.xml b/hive/pom.xml
index 32de95ba0..41acece9a 100644
--- a/hive/pom.xml
+++ b/hive/pom.xml
@@ -106,6 +106,11 @@
             <version>${iceberg.version}</version>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+        </dependency>
     </dependencies>
 
     <build>
diff --git a/hive/src/main/java/com/netease/arctic/hive/ArcticHiveClientPool.java b/hive/src/main/java/com/netease/arctic/hive/ArcticHiveClientPool.java
index 0ba04bb08..cf66f88f8 100644
--- a/hive/src/main/java/com/netease/arctic/hive/ArcticHiveClientPool.java
+++ b/hive/src/main/java/com/netease/arctic/hive/ArcticHiveClientPool.java
@@ -1,6 +1,7 @@
 package com.netease.arctic.hive;
 
 import com.netease.arctic.table.TableMetaStore;
+import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
 import org.apache.iceberg.hive.HiveClientPool;
 import org.slf4j.Logger;
diff --git a/hive/src/main/java/com/netease/arctic/hive/CachedHiveClientPool.java b/hive/src/main/java/com/netease/arctic/hive/CachedHiveClientPool.java
index 6402a60cd..677cf24e3 100644
--- a/hive/src/main/java/com/netease/arctic/hive/CachedHiveClientPool.java
+++ b/hive/src/main/java/com/netease/arctic/hive/CachedHiveClientPool.java
@@ -22,6 +22,7 @@ import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Caffeine;
 import com.netease.arctic.ams.api.properties.CatalogMetaProperties;
 import com.netease.arctic.table.TableMetaStore;
+import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
 import org.apache.iceberg.util.PropertyUtil;
 import org.apache.thrift.TException;
@@ -53,6 +54,7 @@ public class CachedHiveClientPool implements HMSClient, Serializable {
   }
 
   private ArcticHiveClientPool clientPool() {
+    tableMetaStore.getHiveSiteLocation().ifPresent(HiveConf::setHiveSiteLocation);
     return clientPoolCache.get(tableMetaStore, k -> new ArcticHiveClientPool(tableMetaStore, clientPoolSize));
   }
 
diff --git a/hive/src/main/java/com/netease/arctic/hive/io/reader/AdaptHiveGenericArcticDataReader.java b/hive/src/main/java/com/netease/arctic/hive/io/reader/AdaptHiveGenericArcticDataReader.java
new file mode 100644
index 000000000..23b213b9b
--- /dev/null
+++ b/hive/src/main/java/com/netease/arctic/hive/io/reader/AdaptHiveGenericArcticDataReader.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.hive.io.reader;
+
+import com.netease.arctic.data.DataTreeNode;
+import com.netease.arctic.iceberg.optimize.InternalRecordWrapper;
+import com.netease.arctic.io.ArcticFileIO;
+import com.netease.arctic.io.reader.BaseArcticDataReader;
+import com.netease.arctic.table.PrimaryKeySpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.AdaptHiveGenericParquetReaders;
+import org.apache.iceberg.parquet.ParquetValueReader;
+import org.apache.iceberg.types.Type;
+import org.apache.parquet.schema.MessageType;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+
+/**
+ * Implementation of {@link BaseArcticDataReader} with record type {@link Record}.
+ */
+public class AdaptHiveGenericArcticDataReader extends BaseArcticDataReader<Record> {
+
+  public AdaptHiveGenericArcticDataReader(
+      ArcticFileIO fileIO,
+      Schema tableSchema,
+      Schema projectedSchema,
+      PrimaryKeySpec primaryKeySpec,
+      String nameMapping,
+      boolean caseSensitive,
+      BiFunction<Type, Object, Object> convertConstant) {
+    super(fileIO, tableSchema, projectedSchema, primaryKeySpec, nameMapping, caseSensitive, convertConstant, false);
+  }
+
+  public AdaptHiveGenericArcticDataReader(
+      ArcticFileIO fileIO,
+      Schema tableSchema,
+      Schema projectedSchema,
+      PrimaryKeySpec primaryKeySpec,
+      String nameMapping,
+      boolean caseSensitive,
+      BiFunction<Type, Object, Object> convertConstant,
+      Set<DataTreeNode> sourceNodes, boolean reuseContainer) {
+    super(fileIO, tableSchema, projectedSchema, primaryKeySpec,
+        nameMapping, caseSensitive, convertConstant, sourceNodes, reuseContainer);
+  }
+
+  @Override
+  protected Function<MessageType, ParquetValueReader<?>> getNewReaderFunction(
+      Schema projectSchema,
+      Map<Integer, ?> idToConstant) {
+    return fileSchema -> AdaptHiveGenericParquetReaders.buildReader(projectSchema, fileSchema, idToConstant);
+  }
+
+  @Override
+  protected Function<Schema, Function<Record, StructLike>> toStructLikeFunction() {
+    return schema -> record -> new InternalRecordWrapper(schema.asStruct()).wrap(record);
+  }
+}
diff --git a/hive/src/main/java/com/netease/arctic/hive/write/AdaptHiveGenericTaskWriterBuilder.java b/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveGenericTaskWriterBuilder.java
similarity index 99%
rename from hive/src/main/java/com/netease/arctic/hive/write/AdaptHiveGenericTaskWriterBuilder.java
rename to hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveGenericTaskWriterBuilder.java
index cd44b0f0e..d5460e569 100644
--- a/hive/src/main/java/com/netease/arctic/hive/write/AdaptHiveGenericTaskWriterBuilder.java
+++ b/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveGenericTaskWriterBuilder.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package com.netease.arctic.hive.write;
+package com.netease.arctic.hive.io.writer;
 
 import com.netease.arctic.data.ChangeAction;
 import com.netease.arctic.hive.table.HiveLocationKind;
@@ -133,7 +133,6 @@ public class AdaptHiveGenericTaskWriterBuilder implements TaskWriterBuilder<Reco
   }
 
   private GenericBaseTaskWriter buildBaseWriter(LocationKind locationKind) {
-    Preconditions.checkNotNull(transactionId);
     FileFormat fileFormat = FileFormat.valueOf((table.properties().getOrDefault(
         TableProperties.BASE_FILE_FORMAT,
         TableProperties.BASE_FILE_FORMAT_DEFAULT).toUpperCase(Locale.ENGLISH)));
diff --git a/hive/src/main/java/com/netease/arctic/hive/write/AdaptHiveOperateToTableRelation.java b/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOperateToTableRelation.java
similarity index 98%
rename from hive/src/main/java/com/netease/arctic/hive/write/AdaptHiveOperateToTableRelation.java
rename to hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOperateToTableRelation.java
index 612f4df66..9e06d6129 100644
--- a/hive/src/main/java/com/netease/arctic/hive/write/AdaptHiveOperateToTableRelation.java
+++ b/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOperateToTableRelation.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package com.netease.arctic.hive.write;
+package com.netease.arctic.hive.io.writer;
 
 import com.netease.arctic.hive.table.HiveLocationKind;
 import com.netease.arctic.hive.table.KeyedHiveTable;
diff --git a/hive/src/main/java/com/netease/arctic/hive/write/AdaptHiveOutputFileFactory.java b/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOutputFileFactory.java
similarity index 99%
rename from hive/src/main/java/com/netease/arctic/hive/write/AdaptHiveOutputFileFactory.java
rename to hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOutputFileFactory.java
index 265e7c956..1d4f94bbc 100644
--- a/hive/src/main/java/com/netease/arctic/hive/write/AdaptHiveOutputFileFactory.java
+++ b/hive/src/main/java/com/netease/arctic/hive/io/writer/AdaptHiveOutputFileFactory.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package com.netease.arctic.hive.write;
+package com.netease.arctic.hive.io.writer;
 
 import com.netease.arctic.io.ArcticFileIO;
 import com.netease.arctic.io.writer.OutputFileFactory;
diff --git a/hive/src/main/java/com/netease/arctic/hive/write/OperateToTableRelation.java b/hive/src/main/java/com/netease/arctic/hive/io/writer/OperateToTableRelation.java
similarity index 96%
rename from hive/src/main/java/com/netease/arctic/hive/write/OperateToTableRelation.java
rename to hive/src/main/java/com/netease/arctic/hive/io/writer/OperateToTableRelation.java
index 0e7365d53..b1948f8ad 100644
--- a/hive/src/main/java/com/netease/arctic/hive/write/OperateToTableRelation.java
+++ b/hive/src/main/java/com/netease/arctic/hive/io/writer/OperateToTableRelation.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package com.netease.arctic.hive.write;
+package com.netease.arctic.hive.io.writer;
 
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.LocationKind;
diff --git a/hive/src/main/java/org/apache/hadoop/hive/ql/optimizer/ppr/PartitionExpressionForMetastore.java b/hive/src/main/java/org/apache/hadoop/hive/ql/optimizer/ppr/PartitionExpressionForMetastore.java
new file mode 100644
index 000000000..61441d08d
--- /dev/null
+++ b/hive/src/main/java/org/apache/hadoop/hive/ql/optimizer/ppr/PartitionExpressionForMetastore.java
@@ -0,0 +1,64 @@
+/*
+ * 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.hadoop.hive.ql.optimizer.ppr;
+
+import org.apache.hadoop.hive.metastore.FileFormatProxy;
+import org.apache.hadoop.hive.metastore.PartitionExpressionProxy;
+import org.apache.hadoop.hive.metastore.api.FileMetadataExprType;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
+
+import java.util.List;
+
+/**
+ * Copy from package jar "hive-exec" to avoid class check error
+ */
+public class PartitionExpressionForMetastore implements PartitionExpressionProxy {
+
+  @Override
+  public String convertExprToFilter(byte[] expr) throws MetaException {
+    return null;
+  }
+
+  @Override
+  public boolean filterPartitionsByExpr(
+      List<String> partColumnNames,
+      List<PrimitiveTypeInfo> partColumnTypeInfos,
+      byte[] expr,
+      String defaultPartitionName,
+      List<String> partitionNames) throws MetaException {
+    return false;
+  }
+
+  @Override
+  public FileMetadataExprType getMetadataType(String inputFormat) {
+    return null;
+  }
+
+  @Override
+  public FileFormatProxy getFileFormatProxy(FileMetadataExprType type) {
+    return null;
+  }
+
+  @Override
+  public SearchArgument createSarg(byte[] expr) {
+    return null;
+  }
+}
diff --git a/hive/src/main/java/org/apache/hadoop/hive/ql/security/authorization/AuthorizationPreEventListener.java b/hive/src/main/java/org/apache/hadoop/hive/ql/security/authorization/AuthorizationPreEventListener.java
new file mode 100644
index 000000000..ef64db78c
--- /dev/null
+++ b/hive/src/main/java/org/apache/hadoop/hive/ql/security/authorization/AuthorizationPreEventListener.java
@@ -0,0 +1,40 @@
+/*
+ * 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.hadoop.hive.ql.security.authorization;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.MetaStorePreEventListener;
+import org.apache.hadoop.hive.metastore.api.InvalidOperationException;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.hadoop.hive.metastore.events.PreEventContext;
+
+/**
+ * Copy from package jar "hive-exec" to avoid class check error
+ */
+public class AuthorizationPreEventListener extends MetaStorePreEventListener {
+  public AuthorizationPreEventListener(Configuration config) {
+    super(config);
+  }
+
+  @Override
+  public void onEvent(PreEventContext context) throws MetaException, NoSuchObjectException, InvalidOperationException {
+
+  }
+}
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkCatalog.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkCatalog.java
index 76bab0a77..88ee5e5cc 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkCatalog.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/ArcticSparkCatalog.java
@@ -359,7 +359,7 @@ public class ArcticSparkCatalog implements TableCatalog, SupportsNamespaces {
     if (StringUtils.isBlank(catalogUrl)) {
       throw new IllegalArgumentException("lack required properties: url");
     }
-    catalog = CatalogLoader.load(catalogUrl, new HashMap<>());
+    catalog = CatalogLoader.load(catalogUrl, Maps.newHashMap());
   }
 
   @Override
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/io/TaskWriters.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/io/TaskWriters.java
index f229906b1..8fc4634b8 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/io/TaskWriters.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/io/TaskWriters.java
@@ -18,8 +18,8 @@
 
 package com.netease.arctic.spark.io;
 
+import com.netease.arctic.hive.io.writer.AdaptHiveOutputFileFactory;
 import com.netease.arctic.hive.table.SupportHive;
-import com.netease.arctic.hive.write.AdaptHiveOutputFileFactory;
 import com.netease.arctic.io.writer.CommonOutputFileFactory;
 import com.netease.arctic.io.writer.OutputFileFactory;
 import com.netease.arctic.table.ArcticTable;
diff --git a/trino/pom.xml b/trino/pom.xml
index 46a3f42d0..a6fe0a8ed 100644
--- a/trino/pom.xml
+++ b/trino/pom.xml
@@ -305,13 +305,127 @@
             <artifactId>arctic-hive</artifactId>
             <version>${project.version}</version>
             <exclusions>
+<!--                <exclusion>-->
+<!--                    <groupId>org.apache.hive</groupId>-->
+<!--                    <artifactId>hive-metastore</artifactId>-->
+<!--                </exclusion>-->
                 <exclusion>
-                    <groupId>org.apache.hive</groupId>
-                    <artifactId>hive-metastore</artifactId>
+                    <groupId>org.apache.avro</groupId>
+                    <artifactId>avro</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.eclipse.jetty.aggregate</groupId>
+                    <artifactId>jetty-all</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.eclipse.jetty.orbit</groupId>
+                    <artifactId>javax.servlet</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hbase</groupId>
+                    <artifactId>hbase-annotations</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hbase</groupId>
+                    <artifactId>hbase-client</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.sun.jersey</groupId>
+                    <artifactId>jersey-core</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.sun.jersey</groupId>
+                    <artifactId>jersey-server</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.parquet</groupId>
+                    <artifactId>parquet-hadoop-bundle</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-common</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>javax.servlet</groupId>
+                    <artifactId>servlet-api</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>javax.servlet</groupId>
+            <artifactId>javax.servlet-api</artifactId>
+            <version>4.0.1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.netease.arctic</groupId>
+            <artifactId>arctic-hive</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+            <type>test-jar</type>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.avro</groupId>
+                    <artifactId>avro</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.eclipse.jetty.aggregate</groupId>
+                    <artifactId>jetty-all</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.eclipse.jetty.orbit</groupId>
+                    <artifactId>javax.servlet</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hbase</groupId>
+                    <artifactId>hbase-annotations</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hbase</groupId>
+                    <artifactId>hbase-client</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.sun.jersey</groupId>
+                    <artifactId>jersey-core</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>com.sun.jersey</groupId>
+                    <artifactId>jersey-server</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.parquet</groupId>
+                    <artifactId>parquet-hadoop-bundle</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-common</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>javax.servlet</groupId>
+                    <artifactId>servlet-api</artifactId>
                 </exclusion>
             </exclusions>
         </dependency>
 
+        <dependency>
+            <groupId>io.trino</groupId>
+            <artifactId>trino-hive</artifactId>
+            <version>${presto.version}</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.trino.hive</groupId>
+                    <artifactId>hive-apache</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro</artifactId>
+            <version>1.10.1</version>
+        </dependency>
+
         <dependency>
             <groupId>com.sun.jersey</groupId>
             <artifactId>jersey-core</artifactId>
@@ -348,7 +462,7 @@
                     <artifactId>parquet-format</artifactId>
                 </exclusion>
                 <exclusion>
-                    <groupId>com.facebook.presto.hive</groupId>
+                    <groupId>io.trino.hive</groupId>
                     <artifactId>hive-apache</artifactId>
                 </exclusion>
             </exclusions>
@@ -410,10 +524,6 @@
             <artifactId>iceberg-parquet</artifactId>
             <version>${iceberg.version}</version>
             <exclusions>
-                <exclusion>
-                    <groupId>org.apache.parquet</groupId>
-                    <artifactId>parquet-avro</artifactId>
-                </exclusion>
                 <exclusion>
                     <groupId>org.slf4j</groupId>
                     <artifactId>slf4j-api</artifactId>
@@ -453,6 +563,10 @@
                     <groupId>org.apache.iceberg</groupId>
                     <artifactId>*</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>io.trino.hive</groupId>
+                    <artifactId>hive-apache</artifactId>
+                </exclusion>
             </exclusions>
         </dependency>
 
diff --git a/trino/src/main/java/com/netease/arctic/trino/ArcticModule.java b/trino/src/main/java/com/netease/arctic/trino/ArcticModule.java
index f1a3d16c4..a7c96951d 100644
--- a/trino/src/main/java/com/netease/arctic/trino/ArcticModule.java
+++ b/trino/src/main/java/com/netease/arctic/trino/ArcticModule.java
@@ -33,7 +33,6 @@ import io.trino.plugin.hive.FileFormatDataSourceStats;
 import io.trino.plugin.hive.HdfsConfig;
 import io.trino.plugin.hive.HdfsConfiguration;
 import io.trino.plugin.hive.HdfsEnvironment;
-import io.trino.plugin.hive.HiveConfig;
 import io.trino.plugin.hive.NamenodeStats;
 import io.trino.plugin.hive.authentication.HdfsAuthentication;
 import io.trino.plugin.hive.metastore.MetastoreConfig;
@@ -95,7 +94,7 @@ public class ArcticModule implements Module {
 
     configBinder(binder).bindConfig(MetastoreConfig.class);
     configBinder(binder).bindConfig(IcebergConfig.class);
-    configBinder(binder).bindConfig(HiveConfig.class);
+
     newSetBinder(binder, SessionPropertiesProvider.class).addBinding()
         .to(IcebergSessionProperties.class)
         .in(Scopes.SINGLETON);
diff --git a/trino/src/main/java/com/netease/arctic/trino/ArcticSessionProperties.java b/trino/src/main/java/com/netease/arctic/trino/ArcticSessionProperties.java
index efbf1c486..2b1061366 100644
--- a/trino/src/main/java/com/netease/arctic/trino/ArcticSessionProperties.java
+++ b/trino/src/main/java/com/netease/arctic/trino/ArcticSessionProperties.java
@@ -25,7 +25,6 @@ import io.airlift.units.Duration;
 import io.trino.orc.OrcWriteValidation.OrcWriteValidationMode;
 import io.trino.plugin.base.session.SessionPropertiesProvider;
 import io.trino.plugin.hive.HiveCompressionCodec;
-import io.trino.plugin.hive.HiveConfig;
 import io.trino.plugin.hive.parquet.ParquetReaderConfig;
 import io.trino.plugin.hive.parquet.ParquetWriterConfig;
 import io.trino.plugin.iceberg.IcebergConfig;
@@ -83,8 +82,7 @@ public final class ArcticSessionProperties
       IcebergConfig icebergConfig,
       ArcticConfig arcticConfig,
       ParquetReaderConfig parquetReaderConfig,
-      ParquetWriterConfig parquetWriterConfig,
-      HiveConfig hiveConfig) {
+      ParquetWriterConfig parquetWriterConfig) {
     sessionProperties = ImmutableList.<PropertyMetadata<?>>builder()
         //                .add(enumProperty(
         //                        COMPRESSION_CODEC,
@@ -132,11 +130,11 @@ public final class ArcticSessionProperties
         //                        "Read only required fields from a struct",
         //                        icebergConfig.isProjectionPushdownEnabled(),
         //                        false))
-        .add(dataSizeProperty(
-            TARGET_MAX_FILE_SIZE,
-            "Target maximum size of written files; the actual size may be larger",
-            hiveConfig.getTargetMaxFileSize(),
-            false))
+        // .add(dataSizeProperty(
+        //     TARGET_MAX_FILE_SIZE,
+        //     "Target maximum size of written files; the actual size may be larger",
+        //     hiveConfig.getTargetMaxFileSize(),
+        //     false))
         .add(booleanProperty(SYNC_HIVE, "sync hive data to arctic", false, false))
         .add(durationProperty(
             EXPIRE_SNAPSHOTS_MIN_RETENTION,
diff --git a/trino/src/main/java/com/netease/arctic/trino/DefaultArcticCatalogFactory.java b/trino/src/main/java/com/netease/arctic/trino/DefaultArcticCatalogFactory.java
index 8d5f5d14d..8267a5093 100644
--- a/trino/src/main/java/com/netease/arctic/trino/DefaultArcticCatalogFactory.java
+++ b/trino/src/main/java/com/netease/arctic/trino/DefaultArcticCatalogFactory.java
@@ -23,6 +23,8 @@ import com.netease.arctic.catalog.CatalogLoader;
 import io.trino.spi.classloader.ThreadContextClassLoader;
 import javax.inject.Inject;
 
+import java.util.Collections;
+
 /**
  * A factory to generate {@link ArcticCatalog}
  */
@@ -44,7 +46,7 @@ public class DefaultArcticCatalogFactory implements ArcticCatalogFactory {
           try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(this.getClass().getClassLoader())) {
             this.arcticCatalog =
                 new ArcticCatalogSupportTableSuffix(
-                    CatalogLoader.load(arcticConfig.getCatalogUrl(), null));
+                    CatalogLoader.load(arcticConfig.getCatalogUrl(), Collections.emptyMap()));
           }
         }
       }
diff --git a/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedConnectorPageSource.java b/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedConnectorPageSource.java
index 107595aa0..f3954af1a 100644
--- a/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedConnectorPageSource.java
+++ b/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedConnectorPageSource.java
@@ -57,7 +57,6 @@ import static com.google.common.base.Throwables.throwIfInstanceOf;
 import static com.netease.arctic.ArcticErrorCode.ARCTIC_BAD_DATA;
 import static io.trino.plugin.iceberg.IcebergErrorCode.ICEBERG_BAD_DATA;
 import static java.util.Objects.requireNonNull;
-import static org.apache.iceberg.MetadataColumns.FILE_PATH;
 
 /**
  * ConnectorPageSource for Keyed Table
diff --git a/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedConnectorSplitManager.java b/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedConnectorSplitManager.java
index 0fe7811eb..33bd4a935 100644
--- a/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedConnectorSplitManager.java
+++ b/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedConnectorSplitManager.java
@@ -47,9 +47,7 @@ import org.slf4j.LoggerFactory;
 
 import javax.inject.Inject;
 import java.util.ArrayList;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Set;
 import java.util.stream.Collectors;
 
 import static io.trino.plugin.iceberg.ExpressionConverter.toIcebergExpression;
diff --git a/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedDeleteFilter.java b/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedDeleteFilter.java
index 4d46c4765..4767686a6 100644
--- a/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedDeleteFilter.java
+++ b/trino/src/main/java/com/netease/arctic/trino/keyed/KeyedDeleteFilter.java
@@ -18,14 +18,12 @@
 
 package com.netease.arctic.trino.keyed;
 
-import com.netease.arctic.data.PrimaryKeyedFile;
 import com.netease.arctic.io.reader.ArcticDeleteFilter;
 import com.netease.arctic.scan.KeyedTableScanTask;
 import com.netease.arctic.table.PrimaryKeySpec;
 import io.trino.plugin.iceberg.IcebergColumnHandle;
 import io.trino.plugin.iceberg.TypeConverter;
 import io.trino.plugin.iceberg.delete.TrinoRow;
-import io.trino.spi.TrinoException;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.io.FileIO;
@@ -39,7 +37,6 @@ import java.util.Set;
 import java.util.stream.Collectors;
 
 import static com.google.common.collect.ImmutableList.toImmutableList;
-import static com.google.common.collect.ImmutableSet.toImmutableSet;
 
 /**
  * KeyedDeleteFilter is used to do MOR for Keyed Table
diff --git a/trino/src/main/java/com/netease/arctic/trino/unkeyed/IcebergSplitManager.java b/trino/src/main/java/com/netease/arctic/trino/unkeyed/IcebergSplitManager.java
index d3350d228..7c68b4ad6 100644
--- a/trino/src/main/java/com/netease/arctic/trino/unkeyed/IcebergSplitManager.java
+++ b/trino/src/main/java/com/netease/arctic/trino/unkeyed/IcebergSplitManager.java
@@ -19,7 +19,6 @@
 package com.netease.arctic.trino.unkeyed;
 
 import com.google.common.collect.ImmutableList;
-import com.netease.arctic.table.UnkeyedTable;
 import com.netease.arctic.trino.ArcticTransactionManager;
 import com.netease.arctic.trino.TableNameResolve;
 import io.airlift.units.Duration;
diff --git a/trino/src/main/java/io/trino/parquet/reader/TimestampColumnReader.java b/trino/src/main/java/io/trino/parquet/reader/TimestampColumnReader.java
new file mode 100644
index 000000000..bb9ef760b
--- /dev/null
+++ b/trino/src/main/java/io/trino/parquet/reader/TimestampColumnReader.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed 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 io.trino.parquet.reader;
+
+import io.trino.parquet.RichColumnDescriptor;
+import io.trino.plugin.base.type.DecodedTimestamp;
+import io.trino.plugin.base.type.TrinoTimestampEncoder;
+import io.trino.spi.block.BlockBuilder;
+import io.trino.spi.type.LongTimestampWithTimeZone;
+import io.trino.spi.type.TimestampType;
+import io.trino.spi.type.TimestampWithTimeZoneType;
+import io.trino.spi.type.Type;
+import org.joda.time.DateTimeZone;
+
+import static io.trino.parquet.ParquetTimestampUtils.decodeInt96Timestamp;
+import static io.trino.plugin.base.type.TrinoTimestampEncoderFactory.createTimestampEncoder;
+import static io.trino.spi.type.TimeZoneKey.UTC_KEY;
+import static java.util.Objects.requireNonNull;
+
+public class TimestampColumnReader
+    extends PrimitiveColumnReader {
+  private final DateTimeZone timeZone;
+
+  public TimestampColumnReader(RichColumnDescriptor descriptor, DateTimeZone timeZone) {
+    super(descriptor);
+    this.timeZone = requireNonNull(timeZone, "timeZone is null");
+  }
+
+  // TODO: refactor to provide type at construction time (https://github.com/trinodb/trino/issues/5198)
+  @Override
+  protected void readValue(BlockBuilder blockBuilder, Type type) {
+    if (type instanceof TimestampWithTimeZoneType) {
+      DecodedTimestamp decodedTimestamp = decodeInt96Timestamp(valuesReader.readBytes());
+      LongTimestampWithTimeZone longTimestampWithTimeZone =
+          LongTimestampWithTimeZone.fromEpochSecondsAndFraction(decodedTimestamp.getEpochSeconds(),
+              decodedTimestamp.getNanosOfSecond() * 1000L, UTC_KEY);
+      type.writeObject(blockBuilder, longTimestampWithTimeZone);
+    } else {
+      TrinoTimestampEncoder<?> trinoTimestampEncoder = createTimestampEncoder((TimestampType) type, timeZone);
+      trinoTimestampEncoder.write(decodeInt96Timestamp(valuesReader.readBytes()), blockBuilder);
+    }
+  }
+}
diff --git a/trino/src/main/java/io/trino/plugin/iceberg/IcebergSessionProperties.java b/trino/src/main/java/io/trino/plugin/iceberg/IcebergSessionProperties.java
new file mode 100644
index 000000000..2ec162a8a
--- /dev/null
+++ b/trino/src/main/java/io/trino/plugin/iceberg/IcebergSessionProperties.java
@@ -0,0 +1,382 @@
+/*
+ * 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 io.trino.plugin.iceberg;
+
+import com.google.common.collect.ImmutableList;
+import io.airlift.units.DataSize;
+import io.airlift.units.Duration;
+import io.trino.orc.OrcWriteValidation.OrcWriteValidationMode;
+import io.trino.plugin.base.session.SessionPropertiesProvider;
+import io.trino.plugin.hive.HiveCompressionCodec;
+import io.trino.plugin.hive.orc.OrcReaderConfig;
+import io.trino.plugin.hive.orc.OrcWriterConfig;
+import io.trino.plugin.hive.parquet.ParquetReaderConfig;
+import io.trino.plugin.hive.parquet.ParquetWriterConfig;
+import io.trino.spi.TrinoException;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.session.PropertyMetadata;
+
+import javax.inject.Inject;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ThreadLocalRandom;
+
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static io.airlift.units.DataSize.Unit.GIGABYTE;
+import static io.trino.plugin.base.session.PropertyMetadataUtil.dataSizeProperty;
+import static io.trino.plugin.base.session.PropertyMetadataUtil.durationProperty;
+import static io.trino.spi.StandardErrorCode.INVALID_SESSION_PROPERTY;
+import static io.trino.spi.session.PropertyMetadata.booleanProperty;
+import static io.trino.spi.session.PropertyMetadata.doubleProperty;
+import static io.trino.spi.session.PropertyMetadata.enumProperty;
+import static io.trino.spi.session.PropertyMetadata.integerProperty;
+import static io.trino.spi.session.PropertyMetadata.stringProperty;
+import static java.lang.String.format;
+
+/**
+ * Copy from IcebergSessionProperties to remove HaveConfig, because HaveConfig need something in trino-hive
+ */
+public final class IcebergSessionProperties
+    implements SessionPropertiesProvider {
+  private static final String COMPRESSION_CODEC = "compression_codec";
+  private static final String USE_FILE_SIZE_FROM_METADATA = "use_file_size_from_metadata";
+  private static final String ORC_BLOOM_FILTERS_ENABLED = "orc_bloom_filters_enabled";
+  private static final String ORC_MAX_MERGE_DISTANCE = "orc_max_merge_distance";
+  private static final String ORC_MAX_BUFFER_SIZE = "orc_max_buffer_size";
+  private static final String ORC_STREAM_BUFFER_SIZE = "orc_stream_buffer_size";
+  private static final String ORC_TINY_STRIPE_THRESHOLD = "orc_tiny_stripe_threshold";
+  private static final String ORC_MAX_READ_BLOCK_SIZE = "orc_max_read_block_size";
+  private static final String ORC_LAZY_READ_SMALL_RANGES = "orc_lazy_read_small_ranges";
+  private static final String ORC_NESTED_LAZY_ENABLED = "orc_nested_lazy_enabled";
+  private static final String ORC_STRING_STATISTICS_LIMIT = "orc_string_statistics_limit";
+  private static final String ORC_WRITER_VALIDATE_PERCENTAGE = "orc_writer_validate_percentage";
+  private static final String ORC_WRITER_VALIDATE_MODE = "orc_writer_validate_mode";
+  private static final String ORC_WRITER_MIN_STRIPE_SIZE = "orc_writer_min_stripe_size";
+  private static final String ORC_WRITER_MAX_STRIPE_SIZE = "orc_writer_max_stripe_size";
+  private static final String ORC_WRITER_MAX_STRIPE_ROWS = "orc_writer_max_stripe_rows";
+  private static final String ORC_WRITER_MAX_DICTIONARY_MEMORY = "orc_writer_max_dictionary_memory";
+  private static final String PARQUET_MAX_READ_BLOCK_SIZE = "parquet_max_read_block_size";
+  private static final String PARQUET_WRITER_BLOCK_SIZE = "parquet_writer_block_size";
+  private static final String PARQUET_WRITER_PAGE_SIZE = "parquet_writer_page_size";
+  private static final String PARQUET_WRITER_BATCH_SIZE = "parquet_writer_batch_size";
+  private static final String DYNAMIC_FILTERING_WAIT_TIMEOUT = "dynamic_filtering_wait_timeout";
+  private static final String STATISTICS_ENABLED = "statistics_enabled";
+  private static final String PROJECTION_PUSHDOWN_ENABLED = "projection_pushdown_enabled";
+  private static final String TARGET_MAX_FILE_SIZE = "target_max_file_size";
+  private static final String HIVE_CATALOG_NAME = "hive_catalog_name";
+  public static final String EXPIRE_SNAPSHOTS_MIN_RETENTION = "expire_snapshots_min_retention";
+  public static final String DELETE_ORPHAN_FILES_MIN_RETENTION = "delete_orphan_files_min_retention";
+
+  private final List<PropertyMetadata<?>> sessionProperties;
+
+  @Inject
+  public IcebergSessionProperties(
+      IcebergConfig icebergConfig,
+      OrcReaderConfig orcReaderConfig,
+      OrcWriterConfig orcWriterConfig,
+      ParquetReaderConfig parquetReaderConfig,
+      ParquetWriterConfig parquetWriterConfig) {
+    sessionProperties = ImmutableList.<PropertyMetadata<?>>builder()
+        .add(enumProperty(
+            COMPRESSION_CODEC,
+            "Compression codec to use when writing files",
+            HiveCompressionCodec.class,
+            icebergConfig.getCompressionCodec(),
+            false))
+        .add(booleanProperty(
+            USE_FILE_SIZE_FROM_METADATA,
+            "Use file size stored in Iceberg metadata",
+            icebergConfig.isUseFileSizeFromMetadata(),
+            false))
+        .add(booleanProperty(
+            ORC_BLOOM_FILTERS_ENABLED,
+            "ORC: Enable bloom filters for predicate pushdown",
+            orcReaderConfig.isBloomFiltersEnabled(),
+            false))
+        .add(dataSizeProperty(
+            ORC_MAX_MERGE_DISTANCE,
+            "ORC: Maximum size of gap between two reads to merge into a single read",
+            orcReaderConfig.getMaxMergeDistance(),
+            false))
+        .add(dataSizeProperty(
+            ORC_MAX_BUFFER_SIZE,
+            "ORC: Maximum size of a single read",
+            orcReaderConfig.getMaxBufferSize(),
+            false))
+        .add(dataSizeProperty(
+            ORC_STREAM_BUFFER_SIZE,
+            "ORC: Size of buffer for streaming reads",
+            orcReaderConfig.getStreamBufferSize(),
+            false))
+        .add(dataSizeProperty(
+            ORC_TINY_STRIPE_THRESHOLD,
+            "ORC: Threshold below which an ORC stripe or file will read in its entirety",
+            orcReaderConfig.getTinyStripeThreshold(),
+            false))
+        .add(dataSizeProperty(
+            ORC_MAX_READ_BLOCK_SIZE,
+            "ORC: Soft max size of Trino blocks produced by ORC reader",
+            orcReaderConfig.getMaxBlockSize(),
+            false))
+        .add(booleanProperty(
+            ORC_LAZY_READ_SMALL_RANGES,
+            "Experimental: ORC: Read small file segments lazily",
+            orcReaderConfig.isLazyReadSmallRanges(),
+            false))
+        .add(booleanProperty(
+            ORC_NESTED_LAZY_ENABLED,
+            "Experimental: ORC: Lazily read nested data",
+            orcReaderConfig.isNestedLazy(),
+            false))
+        .add(dataSizeProperty(
+            ORC_STRING_STATISTICS_LIMIT,
+            "ORC: Maximum size of string statistics; drop if exceeding",
+            orcWriterConfig.getStringStatisticsLimit(),
+            false))
+        .add(doubleProperty(
+            ORC_WRITER_VALIDATE_PERCENTAGE,
+            "ORC: Percentage of written files to validate by re-reading them",
+            orcWriterConfig.getValidationPercentage(),
+            doubleValue -> {
+              if (doubleValue < 0.0 || doubleValue > 100.0) {
+                throw new TrinoException(INVALID_SESSION_PROPERTY, format(
+                    "%s must be between 0.0 and 100.0 inclusive: %s",
+                    ORC_WRITER_VALIDATE_PERCENTAGE,
+                    doubleValue));
+              }
+            },
+            false))
+        .add(enumProperty(
+            ORC_WRITER_VALIDATE_MODE,
+            "ORC: Level of detail in ORC validation",
+            OrcWriteValidationMode.class,
+            orcWriterConfig.getValidationMode(),
+            false))
+        .add(dataSizeProperty(
+            ORC_WRITER_MIN_STRIPE_SIZE,
+            "ORC: Min stripe size",
+            orcWriterConfig.getStripeMinSize(),
+            false))
+        .add(dataSizeProperty(
+            ORC_WRITER_MAX_STRIPE_SIZE,
+            "ORC: Max stripe size",
+            orcWriterConfig.getStripeMaxSize(),
+            false))
+        .add(integerProperty(
+            ORC_WRITER_MAX_STRIPE_ROWS,
+            "ORC: Max stripe row count",
+            orcWriterConfig.getStripeMaxRowCount(),
+            false))
+        .add(dataSizeProperty(
+            ORC_WRITER_MAX_DICTIONARY_MEMORY,
+            "ORC: Max dictionary memory",
+            orcWriterConfig.getDictionaryMaxMemory(),
+            false))
+        .add(dataSizeProperty(
+            PARQUET_MAX_READ_BLOCK_SIZE,
+            "Parquet: Maximum size of a block to read",
+            parquetReaderConfig.getMaxReadBlockSize(),
+            false))
+        .add(dataSizeProperty(
+            PARQUET_WRITER_BLOCK_SIZE,
+            "Parquet: Writer block size",
+            parquetWriterConfig.getBlockSize(),
+            false))
+        .add(dataSizeProperty(
+            PARQUET_WRITER_PAGE_SIZE,
+            "Parquet: Writer page size",
+            parquetWriterConfig.getPageSize(),
+            false))
+        .add(integerProperty(
+            PARQUET_WRITER_BATCH_SIZE,
+            "Parquet: Maximum number of rows passed to the writer in each batch",
+            parquetWriterConfig.getBatchSize(),
+            false))
+        .add(dataSizeProperty(
+            TARGET_MAX_FILE_SIZE,
+            "Target maximum size of written files; the actual size may be larger",
+            DataSize.of(1, GIGABYTE),
+            false))
+        .add(durationProperty(
+            DYNAMIC_FILTERING_WAIT_TIMEOUT,
+            "Duration to wait for completion of dynamic filters during split generation",
+            icebergConfig.getDynamicFilteringWaitTimeout(),
+            false))
+        .add(booleanProperty(
+            STATISTICS_ENABLED,
+            "Expose table statistics",
+            icebergConfig.isTableStatisticsEnabled(),
+            false))
+        .add(booleanProperty(
+            PROJECTION_PUSHDOWN_ENABLED,
+            "Read only required fields from a struct",
+            icebergConfig.isProjectionPushdownEnabled(),
+            false))
+        // .add(dataSizeProperty(
+        //     TARGET_MAX_FILE_SIZE,
+        //     "Target maximum size of written files; the actual size may be larger",
+        //     hiveConfig.getTargetMaxFileSize(),
+        //     false))
+        .add(stringProperty(
+            HIVE_CATALOG_NAME,
+            "Catalog to redirect to when a Hive table is referenced",
+            icebergConfig.getHiveCatalogName().orElse(null),
+            // Session-level redirections configuration does not work well with views,
+            // as view body is analyzed in context
+            // of a session with properties stripped off. Thus, this property is more
+            // of a test-only, or at most POC usefulness.
+            true))
+        .add(durationProperty(
+            EXPIRE_SNAPSHOTS_MIN_RETENTION,
+            "Minimal retention period for expire_snapshot procedure",
+            icebergConfig.getExpireSnapshotsMinRetention(),
+            false))
+        .add(durationProperty(
+            DELETE_ORPHAN_FILES_MIN_RETENTION,
+            "Minimal retention period for delete_orphan_files procedure",
+            icebergConfig.getDeleteOrphanFilesMinRetention(),
+            false))
+        .build();
+  }
+
+  @Override
+  public List<PropertyMetadata<?>> getSessionProperties() {
+    return sessionProperties;
+  }
+
+  public static boolean isOrcBloomFiltersEnabled(ConnectorSession session) {
+    return session.getProperty(ORC_BLOOM_FILTERS_ENABLED, Boolean.class);
+  }
+
+  public static DataSize getOrcMaxMergeDistance(ConnectorSession session) {
+    return session.getProperty(ORC_MAX_MERGE_DISTANCE, DataSize.class);
+  }
+
+  public static DataSize getOrcMaxBufferSize(ConnectorSession session) {
+    return session.getProperty(ORC_MAX_BUFFER_SIZE, DataSize.class);
+  }
+
+  public static DataSize getOrcStreamBufferSize(ConnectorSession session) {
+    return session.getProperty(ORC_STREAM_BUFFER_SIZE, DataSize.class);
+  }
+
+  public static DataSize getOrcTinyStripeThreshold(ConnectorSession session) {
+    return session.getProperty(ORC_TINY_STRIPE_THRESHOLD, DataSize.class);
+  }
+
+  public static DataSize getOrcMaxReadBlockSize(ConnectorSession session) {
+    return session.getProperty(ORC_MAX_READ_BLOCK_SIZE, DataSize.class);
+  }
+
+  public static boolean getOrcLazyReadSmallRanges(ConnectorSession session) {
+    return session.getProperty(ORC_LAZY_READ_SMALL_RANGES, Boolean.class);
+  }
+
+  public static boolean isOrcNestedLazy(ConnectorSession session) {
+    return session.getProperty(ORC_NESTED_LAZY_ENABLED, Boolean.class);
+  }
+
+  public static DataSize getOrcStringStatisticsLimit(ConnectorSession session) {
+    return session.getProperty(ORC_STRING_STATISTICS_LIMIT, DataSize.class);
+  }
+
+  public static boolean isOrcWriterValidate(ConnectorSession session) {
+    double percentage = session.getProperty(ORC_WRITER_VALIDATE_PERCENTAGE, Double.class);
+    if (percentage == 0.0) {
+      return false;
+    }
+
+    checkArgument(percentage > 0.0 && percentage <= 100.0);
+
+    return ThreadLocalRandom.current().nextDouble(100) < percentage;
+  }
+
+  public static OrcWriteValidationMode getOrcWriterValidateMode(ConnectorSession session) {
+    return session.getProperty(ORC_WRITER_VALIDATE_MODE, OrcWriteValidationMode.class);
+  }
+
+  public static DataSize getOrcWriterMinStripeSize(ConnectorSession session) {
+    return session.getProperty(ORC_WRITER_MIN_STRIPE_SIZE, DataSize.class);
+  }
+
+  public static DataSize getOrcWriterMaxStripeSize(ConnectorSession session) {
+    return session.getProperty(ORC_WRITER_MAX_STRIPE_SIZE, DataSize.class);
+  }
+
+  public static int getOrcWriterMaxStripeRows(ConnectorSession session) {
+    return session.getProperty(ORC_WRITER_MAX_STRIPE_ROWS, Integer.class);
+  }
+
+  public static DataSize getOrcWriterMaxDictionaryMemory(ConnectorSession session) {
+    return session.getProperty(ORC_WRITER_MAX_DICTIONARY_MEMORY, DataSize.class);
+  }
+
+  public static HiveCompressionCodec getCompressionCodec(ConnectorSession session) {
+    return session.getProperty(COMPRESSION_CODEC, HiveCompressionCodec.class);
+  }
+
+  public static boolean isUseFileSizeFromMetadata(ConnectorSession session) {
+    return session.getProperty(USE_FILE_SIZE_FROM_METADATA, Boolean.class);
+  }
+
+  public static DataSize getParquetMaxReadBlockSize(ConnectorSession session) {
+    return session.getProperty(PARQUET_MAX_READ_BLOCK_SIZE, DataSize.class);
+  }
+
+  public static DataSize getParquetWriterPageSize(ConnectorSession session) {
+    return session.getProperty(PARQUET_WRITER_PAGE_SIZE, DataSize.class);
+  }
+
+  public static DataSize getParquetWriterBlockSize(ConnectorSession session) {
+    return session.getProperty(PARQUET_WRITER_BLOCK_SIZE, DataSize.class);
+  }
+
+  public static int getParquetWriterBatchSize(ConnectorSession session) {
+    return session.getProperty(PARQUET_WRITER_BATCH_SIZE, Integer.class);
+  }
+
+  public static Duration getDynamicFilteringWaitTimeout(ConnectorSession session) {
+    return session.getProperty(DYNAMIC_FILTERING_WAIT_TIMEOUT, Duration.class);
+  }
+
+  public static boolean isStatisticsEnabled(ConnectorSession session) {
+    return session.getProperty(STATISTICS_ENABLED, Boolean.class);
+  }
+
+  public static boolean isProjectionPushdownEnabled(ConnectorSession session) {
+    return session.getProperty(PROJECTION_PUSHDOWN_ENABLED, Boolean.class);
+  }
+
+  public static long getTargetMaxFileSize(ConnectorSession session) {
+    return session.getProperty(TARGET_MAX_FILE_SIZE, DataSize.class).toBytes();
+  }
+
+  public static Optional<String> getHiveCatalogName(ConnectorSession session) {
+    return Optional.ofNullable(session.getProperty(HIVE_CATALOG_NAME, String.class));
+  }
+
+  public static Duration getExpireSnapshotMinRetention(ConnectorSession session) {
+    return session.getProperty(EXPIRE_SNAPSHOTS_MIN_RETENTION, Duration.class);
+  }
+
+  public static Duration getDeleteOrphanFilesMinRetention(ConnectorSession session) {
+    return session.getProperty(DELETE_ORPHAN_FILES_MIN_RETENTION, Duration.class);
+  }
+}
diff --git a/trino/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverterUtil.java b/trino/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverterUtil.java
new file mode 100644
index 000000000..d171b4926
--- /dev/null
+++ b/trino/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverterUtil.java
@@ -0,0 +1,34 @@
+package org.apache.parquet.format.converter;
+
+import org.apache.parquet.format.ConvertedType;
+import org.apache.parquet.format.LogicalType;
+import org.apache.parquet.format.SchemaElement;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+
+/**
+ * Copy from hive-apache package, because include hive-apache will cause class conflict
+ */
+public final class ParquetMetadataConverterUtil {
+  private ParquetMetadataConverterUtil() {
+
+  }
+
+  public static LogicalTypeAnnotation getLogicalTypeAnnotation(
+      ParquetMetadataConverter parquetMetadataConverter,
+      ConvertedType convertedType,
+      SchemaElement schemaElement) {
+    return parquetMetadataConverter.getLogicalTypeAnnotation(convertedType, schemaElement);
+  }
+
+  public static LogicalTypeAnnotation getLogicalTypeAnnotation(
+      ParquetMetadataConverter parquetMetadataConverter,
+      LogicalType logicalType) {
+    return parquetMetadataConverter.getLogicalTypeAnnotation(logicalType);
+  }
+
+  public static LogicalType convertToLogicalType(
+      ParquetMetadataConverter parquetMetadataConverter,
+      LogicalTypeAnnotation logicalTypeAnnotation) {
+    return parquetMetadataConverter.convertToLogicalType(logicalTypeAnnotation);
+  }
+}
diff --git a/trino/src/main/java/org/apache/parquet/io/ColumnIOUtil.java b/trino/src/main/java/org/apache/parquet/io/ColumnIOUtil.java
new file mode 100644
index 000000000..4606c4c71
--- /dev/null
+++ b/trino/src/main/java/org/apache/parquet/io/ColumnIOUtil.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed 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.parquet.io;
+
+/**
+ * Copy from hive-apache package, because include hive-apache will cause class conflict
+ */
+public final class ColumnIOUtil {
+  private ColumnIOUtil() {
+
+  }
+
+  public static int columnDefinitionLevel(ColumnIO column) {
+    return column.getDefinitionLevel();
+  }
+
+  public static int columnRepetitionLevel(ColumnIO column) {
+    return column.getRepetitionLevel();
+  }
+}


Test patch:
diff --git a/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java
new file mode 100644
index 000000000..6b39dac7f
--- /dev/null
+++ b/flink/v1.12/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.flink.write;
+
+import com.netease.arctic.flink.read.AdaptHiveFlinkParquetReaders;
+import com.netease.arctic.hive.HiveTableTestBase;
+import com.netease.arctic.hive.io.HiveTestRecords;
+import com.netease.arctic.hive.io.writer.AdaptHiveGenericTaskWriterBuilder;
+import com.netease.arctic.hive.table.HiveLocationKind;
+import com.netease.arctic.io.writer.GenericBaseTaskWriter;
+import com.netease.arctic.io.writer.GenericChangeTaskWriter;
+import com.netease.arctic.table.ArcticTable;
+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.io.File;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+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.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+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.flink.FlinkSchemaUtil;
+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.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class AdaptHiveWriterTest extends HiveTableTestBase {
+
+  @Test
+  public void testWriteTypeFromOperateKind(){
+    {
+      FlinkTaskWriterBuilder builder = FlinkTaskWriterBuilder
+          .buildFor(testKeyedHiveTable)
+          .withFlinkSchema(FlinkSchemaUtil.convert(testKeyedHiveTable.schema()))
+          .withTransactionId(1);
+
+      Assert.assertTrue(builder.buildWriter(ChangeLocationKind.INSTANT) instanceof FlinkChangeTaskWriter);
+      Assert.assertTrue(builder.buildWriter(BaseLocationKind.INSTANT) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(HiveLocationKind.INSTANT) instanceof FlinkBaseTaskWriter);
+
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.APPEND) instanceof FlinkChangeTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.OVERWRITE) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.MINOR_OPTIMIZE) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.MAJOR_OPTIMIZE) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.FULL_OPTIMIZE) instanceof FlinkBaseTaskWriter);
+    }
+    {
+      FlinkTaskWriterBuilder builder = FlinkTaskWriterBuilder
+          .buildFor(testHiveTable)
+          .withFlinkSchema(FlinkSchemaUtil.convert(testHiveTable.schema()))
+          .withTransactionId(1);
+
+      Assert.assertTrue(builder.buildWriter(BaseLocationKind.INSTANT) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(HiveLocationKind.INSTANT) instanceof FlinkBaseTaskWriter);
+
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.APPEND) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.OVERWRITE) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.MAJOR_OPTIMIZE) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.FULL_OPTIMIZE) instanceof FlinkBaseTaskWriter);
+    }
+  }
+
+  @Test
+  public void testKeyedTableChangeWriteByLocationKind() throws IOException {
+    testWrite(testKeyedHiveTable, ChangeLocationKind.INSTANT, geneRowData(), "change");
+  }
+
+  @Test
+  public void testKeyedTableBaseWriteByLocationKind() throws IOException {
+    testWrite(testKeyedHiveTable, BaseLocationKind.INSTANT, geneRowData(), "base");
+  }
+
+  @Test
+  public void testKeyedTableHiveWriteByLocationKind() throws IOException {
+    testWrite(testKeyedHiveTable, HiveLocationKind.INSTANT, geneRowData(), "hive");
+  }
+
+  @Test
+  public void testUnPartitionKeyedTableChangeWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionKeyedHiveTable, ChangeLocationKind.INSTANT, geneRowData(), "change");
+  }
+
+  @Test
+  public void testUnPartitionKeyedTableBaseWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionKeyedHiveTable, BaseLocationKind.INSTANT, geneRowData(), "base");
+  }
+
+  @Test
+  public void testUnPartitionKeyedTableHiveWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionKeyedHiveTable, HiveLocationKind.INSTANT, geneRowData(), "hive");
+  }
+
+  @Test
+  public void testUnKeyedTableChangeWriteByLocationKind() throws IOException {
+    try {
+      testWrite(testHiveTable, ChangeLocationKind.INSTANT, geneRowData(), "change");
+    }catch (Exception e){
+      Assert.assertTrue(e instanceof IllegalArgumentException);
+    }
+  }
+
+  @Test
+  public void testUnKeyedTableBaseWriteByLocationKind() throws IOException {
+    testWrite(testHiveTable, BaseLocationKind.INSTANT, geneRowData(), "base");
+  }
+
+  @Test
+  public void testUnKeyedTableHiveWriteByLocationKind() throws IOException {
+    testWrite(testHiveTable, HiveLocationKind.INSTANT, geneRowData(), "hive");
+  }
+
+  @Test
+  public void testUnPartitionUnKeyedTableChangeWriteByLocationKind() throws IOException {
+    try {
+      testWrite(testUnPartitionHiveTable, ChangeLocationKind.INSTANT, geneRowData(), "change");
+    }catch (Exception e){
+      Assert.assertTrue(e instanceof IllegalArgumentException);
+    }
+  }
+
+  @Test
+  public void testUnPartitionUnKeyedTableBaseWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionHiveTable, BaseLocationKind.INSTANT, geneRowData(), "base");
+  }
+
+  @Test
+  public void testUnPartitionUnKeyedTableHiveWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionHiveTable, HiveLocationKind.INSTANT, geneRowData(), "hive");
+  }
+
+  public void testWrite(ArcticTable table, LocationKind locationKind, List<RowData> records, String pathFeature) throws IOException {
+    FlinkTaskWriterBuilder builder = FlinkTaskWriterBuilder
+        .buildFor(table)
+        .withFlinkSchema(FlinkSchemaUtil.convert(table.schema()))
+        .withTransactionId(1);
+
+    TaskWriter<RowData> changeWrite = builder.buildWriter(locationKind);
+    for (RowData record: records) {
+      changeWrite.write(record);
+    }
+    WriteResult complete = changeWrite.complete();
+    Arrays.stream(complete.dataFiles()).forEach(s -> Assert.assertTrue(s.path().toString().contains(pathFeature)));
+    CloseableIterable<RowData> concat =
+        CloseableIterable.concat(Arrays.stream(complete.dataFiles()).map(s -> readParquet(
+            table.schema(),
+            s.path().toString())).collect(Collectors.toList()));
+    Set<RowData> result = new HashSet<>();
+    Iterators.addAll(result, concat.iterator());
+    Assert.assertEquals(result, records.stream().collect(Collectors.toSet()));
+  }
+
+  private CloseableIterable<RowData> readParquet(Schema schema, String path){
+    AdaptHiveParquet.ReadBuilder builder = AdaptHiveParquet.read(
+            Files.localInput(path))
+        .project(schema)
+        .createReaderFunc(fileSchema -> AdaptHiveFlinkParquetReaders.buildReader(schema, fileSchema, new HashMap<>()))
+        .caseSensitive(false);
+
+    CloseableIterable<RowData> iterable = builder.build();
+    return iterable;
+  }
+
+  private List<RowData> geneRowData(){
+    RowData rowData = GenericRowData.of(
+        1,
+        StringData.fromString("jack"),
+        TimestampData.fromLocalDateTime(LocalDateTime.of(2022, 1, 1 , 10, 0, 0)),
+        TimestampData.fromInstant(LocalDateTime.of(2022, 1, 1 , 10, 0, 0).toInstant(ZoneOffset.ofHours(8))),
+        DecimalData.fromBigDecimal(new BigDecimal("100"), 3, 0)
+    );
+    return Lists.newArrayList(rowData);
+  }
+}
diff --git a/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java
new file mode 100644
index 000000000..0afaf1366
--- /dev/null
+++ b/flink/v1.14/flink/src/test/java/com/netease/arctic/flink/write/AdaptHiveWriterTest.java
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.netease.arctic.flink.write;
+
+import com.netease.arctic.flink.read.AdaptHiveFlinkParquetReaders;
+import com.netease.arctic.hive.HiveTableTestBase;
+import com.netease.arctic.hive.table.HiveLocationKind;
+import com.netease.arctic.table.ArcticTable;
+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.io.File;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+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.flink.table.data.DecimalData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.flink.FlinkSchemaUtil;
+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.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class AdaptHiveWriterTest extends HiveTableTestBase {
+
+  @Test
+  public void testWriteTypeFromOperateKind(){
+    {
+      FlinkTaskWriterBuilder builder = FlinkTaskWriterBuilder
+          .buildFor(testKeyedHiveTable)
+          .withFlinkSchema(FlinkSchemaUtil.convert(testKeyedHiveTable.schema()))
+          .withTransactionId(1);
+
+      Assert.assertTrue(builder.buildWriter(ChangeLocationKind.INSTANT) instanceof FlinkChangeTaskWriter);
+      Assert.assertTrue(builder.buildWriter(BaseLocationKind.INSTANT) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(HiveLocationKind.INSTANT) instanceof FlinkBaseTaskWriter);
+
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.APPEND) instanceof FlinkChangeTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.OVERWRITE) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.MINOR_OPTIMIZE) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.MAJOR_OPTIMIZE) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.FULL_OPTIMIZE) instanceof FlinkBaseTaskWriter);
+    }
+    {
+      FlinkTaskWriterBuilder builder = FlinkTaskWriterBuilder
+          .buildFor(testHiveTable)
+          .withFlinkSchema(FlinkSchemaUtil.convert(testHiveTable.schema()))
+          .withTransactionId(1);
+
+      Assert.assertTrue(builder.buildWriter(BaseLocationKind.INSTANT) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(HiveLocationKind.INSTANT) instanceof FlinkBaseTaskWriter);
+
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.APPEND) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.OVERWRITE) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.MAJOR_OPTIMIZE) instanceof FlinkBaseTaskWriter);
+      Assert.assertTrue(builder.buildWriter(WriteOperationKind.FULL_OPTIMIZE) instanceof FlinkBaseTaskWriter);
+    }
+  }
+
+  @Test
+  public void testKeyedTableChangeWriteByLocationKind() throws IOException {
+    testWrite(testKeyedHiveTable, ChangeLocationKind.INSTANT, geneRowData(), "change");
+  }
+
+  @Test
+  public void testKeyedTableBaseWriteByLocationKind() throws IOException {
+    testWrite(testKeyedHiveTable, BaseLocationKind.INSTANT, geneRowData(), "base");
+  }
+
+  @Test
+  public void testKeyedTableHiveWriteByLocationKind() throws IOException {
+    testWrite(testKeyedHiveTable, HiveLocationKind.INSTANT, geneRowData(), "hive");
+  }
+
+  @Test
+  public void testUnPartitionKeyedTableChangeWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionKeyedHiveTable, ChangeLocationKind.INSTANT, geneRowData(), "change");
+  }
+
+  @Test
+  public void testUnPartitionKeyedTableBaseWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionKeyedHiveTable, BaseLocationKind.INSTANT, geneRowData(), "base");
+  }
+
+  @Test
+  public void testUnPartitionKeyedTableHiveWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionKeyedHiveTable, HiveLocationKind.INSTANT, geneRowData(), "hive");
+  }
+
+  @Test
+  public void testUnKeyedTableChangeWriteByLocationKind() throws IOException {
+    try {
+      testWrite(testHiveTable, ChangeLocationKind.INSTANT, geneRowData(), "change");
+    }catch (Exception e){
+      Assert.assertTrue(e instanceof IllegalArgumentException);
+    }
+  }
+
+  @Test
+  public void testUnKeyedTableBaseWriteByLocationKind() throws IOException {
+    testWrite(testHiveTable, BaseLocationKind.INSTANT, geneRowData(), "base");
+  }
+
+  @Test
+  public void testUnKeyedTableHiveWriteByLocationKind() throws IOException {
+    testWrite(testHiveTable, HiveLocationKind.INSTANT, geneRowData(), "hive");
+  }
+
+  @Test
+  public void testUnPartitionUnKeyedTableChangeWriteByLocationKind() throws IOException {
+    try {
+      testWrite(testUnPartitionHiveTable, ChangeLocationKind.INSTANT, geneRowData(), "change");
+    }catch (Exception e){
+      Assert.assertTrue(e instanceof IllegalArgumentException);
+    }
+  }
+
+  @Test
+  public void testUnPartitionUnKeyedTableBaseWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionHiveTable, BaseLocationKind.INSTANT, geneRowData(), "base");
+  }
+
+  @Test
+  public void testUnPartitionUnKeyedTableHiveWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionHiveTable, HiveLocationKind.INSTANT, geneRowData(), "hive");
+  }
+
+  public void testWrite(ArcticTable table, LocationKind locationKind, List<RowData> records, String pathFeature) throws IOException {
+    FlinkTaskWriterBuilder builder = FlinkTaskWriterBuilder
+        .buildFor(table)
+        .withFlinkSchema(FlinkSchemaUtil.convert(table.schema()))
+        .withTransactionId(1);
+
+    TaskWriter<RowData> changeWrite = builder.buildWriter(locationKind);
+    for (RowData record: records) {
+      changeWrite.write(record);
+    }
+    WriteResult complete = changeWrite.complete();
+    Arrays.stream(complete.dataFiles()).forEach(s -> Assert.assertTrue(s.path().toString().contains(pathFeature)));
+    CloseableIterable<RowData> concat =
+        CloseableIterable.concat(Arrays.stream(complete.dataFiles()).map(s -> readParquet(
+            table.schema(),
+            s.path().toString())).collect(Collectors.toList()));
+    Set<RowData> result = new HashSet<>();
+    Iterators.addAll(result, concat.iterator());
+    Assert.assertEquals(result, records.stream().collect(Collectors.toSet()));
+  }
+
+  private CloseableIterable<RowData> readParquet(Schema schema, String path){
+    AdaptHiveParquet.ReadBuilder builder = AdaptHiveParquet.read(
+            Files.localInput(path))
+        .project(schema)
+        .createReaderFunc(fileSchema -> AdaptHiveFlinkParquetReaders.buildReader(schema, fileSchema, new HashMap<>()))
+        .caseSensitive(false);
+
+    CloseableIterable<RowData> iterable = builder.build();
+    return iterable;
+  }
+
+  private List<RowData> geneRowData(){
+    RowData rowData = GenericRowData.of(
+        1,
+        StringData.fromString("jack"),
+        TimestampData.fromLocalDateTime(LocalDateTime.of(2022, 1, 1 , 10, 0, 0)),
+        TimestampData.fromInstant(LocalDateTime.of(2022, 1, 1 , 10, 0, 0).toInstant(ZoneOffset.ofHours(8))),
+        DecimalData.fromBigDecimal(new BigDecimal("100"), 3, 0)
+    );
+    return Lists.newArrayList(rowData);
+  }
+}
diff --git a/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java b/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java
index bfb30df5c..0b8512646 100644
--- a/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java
+++ b/hive/src/test/java/com/netease/arctic/hive/HiveTableTestBase.java
@@ -29,6 +29,7 @@ import com.netease.arctic.hive.table.KeyedHiveTable;
 import com.netease.arctic.hive.table.UnkeyedHiveTable;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.TableIdentifier;
+import com.netease.arctic.table.TableProperties;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -71,6 +72,11 @@ public class HiveTableTestBase extends TableTestBase {
   protected static final TableIdentifier HIVE_PK_TABLE_ID =
       TableIdentifier.of(HIVE_CATALOG_NAME, HIVE_DB_NAME, "test_pk_hive_table");
 
+  protected static final TableIdentifier UN_PARTITION_HIVE_TABLE_ID =
+      TableIdentifier.of(HIVE_CATALOG_NAME, HIVE_DB_NAME, "un_partition_test_hive_table");
+  protected static final TableIdentifier UN_PARTITION_HIVE_PK_TABLE_ID =
+      TableIdentifier.of(HIVE_CATALOG_NAME, HIVE_DB_NAME, "un_partition_test_pk_hive_table");
+
   public static final Schema HIVE_TABLE_SCHEMA = new Schema(
       Types.NestedField.required(1, "id", Types.IntegerType.get()),
       Types.NestedField.required(2, "name", Types.StringType.get()),
@@ -86,6 +92,9 @@ public class HiveTableTestBase extends TableTestBase {
   protected UnkeyedHiveTable testHiveTable;
   protected KeyedHiveTable testKeyedHiveTable;
 
+  protected UnkeyedHiveTable testUnPartitionHiveTable;
+  protected KeyedHiveTable testUnPartitionKeyedHiveTable;
+
   @BeforeClass
   public static void startMetastore() throws Exception {
     int ref = testCount.incrementAndGet();
@@ -141,20 +150,35 @@ public class HiveTableTestBase extends TableTestBase {
         .withPartitionSpec(HIVE_SPEC)
         .create().asUnkeyedTable();
 
+    testUnPartitionHiveTable = (UnkeyedHiveTable) hiveCatalog
+        .newTableBuilder(UN_PARTITION_HIVE_TABLE_ID, HIVE_TABLE_SCHEMA)
+        .create().asUnkeyedTable();
+
     testKeyedHiveTable = (KeyedHiveTable) hiveCatalog
         .newTableBuilder(HIVE_PK_TABLE_ID, HIVE_TABLE_SCHEMA)
         .withPartitionSpec(HIVE_SPEC)
         .withPrimaryKeySpec(PRIMARY_KEY_SPEC)
         .create().asKeyedTable();
+
+    testUnPartitionKeyedHiveTable = (KeyedHiveTable) hiveCatalog
+        .newTableBuilder(UN_PARTITION_HIVE_PK_TABLE_ID, HIVE_TABLE_SCHEMA)
+        .withPrimaryKeySpec(PRIMARY_KEY_SPEC)
+        .create().asKeyedTable();
   }
 
   @After
   public void clearTable() {
     hiveCatalog.dropTable(HIVE_TABLE_ID, true);
-    AMS.handler().getTableCommitMetas().remove(TABLE_ID.buildTableIdentifier());
+    AMS.handler().getTableCommitMetas().remove(HIVE_TABLE_ID.buildTableIdentifier());
+
+    hiveCatalog.dropTable(UN_PARTITION_HIVE_TABLE_ID, true);
+    AMS.handler().getTableCommitMetas().remove(UN_PARTITION_HIVE_TABLE_ID.buildTableIdentifier());
 
     hiveCatalog.dropTable(HIVE_PK_TABLE_ID, true);
-    AMS.handler().getTableCommitMetas().remove(PK_TABLE_ID.buildTableIdentifier());
+    AMS.handler().getTableCommitMetas().remove(HIVE_PK_TABLE_ID.buildTableIdentifier());
+
+    hiveCatalog.dropTable(UN_PARTITION_HIVE_PK_TABLE_ID, true);
+    AMS.handler().getTableCommitMetas().remove(UN_PARTITION_HIVE_PK_TABLE_ID.buildTableIdentifier());
   }
 
 
diff --git a/hive/src/test/java/com/netease/arctic/hive/io/AdaptHiveWriterTest.java b/hive/src/test/java/com/netease/arctic/hive/io/AdaptHiveWriterTest.java
index 715aa04ee..3b550aa1d 100644
--- a/hive/src/test/java/com/netease/arctic/hive/io/AdaptHiveWriterTest.java
+++ b/hive/src/test/java/com/netease/arctic/hive/io/AdaptHiveWriterTest.java
@@ -20,7 +20,7 @@ package com.netease.arctic.hive.io;
 
 import com.netease.arctic.hive.HiveTableTestBase;
 import com.netease.arctic.hive.table.HiveLocationKind;
-import com.netease.arctic.hive.write.AdaptHiveGenericTaskWriterBuilder;
+import com.netease.arctic.hive.io.writer.AdaptHiveGenericTaskWriterBuilder;
 import com.netease.arctic.io.writer.GenericBaseTaskWriter;
 import com.netease.arctic.io.writer.GenericChangeTaskWriter;
 import com.netease.arctic.table.ArcticTable;
@@ -94,7 +94,22 @@ public class AdaptHiveWriterTest extends HiveTableTestBase {
 
   @Test
   public void testKeyedTableHiveWriteByLocationKind() throws IOException {
-    testWrite(testKeyedHiveTable, HiveLocationKind.INSTANT, HiveTestRecords.baseRecords(), "hive_data");
+    testWrite(testKeyedHiveTable, HiveLocationKind.INSTANT, HiveTestRecords.baseRecords(), "hive");
+  }
+
+  @Test
+  public void testUnPartitionKeyedTableChangeWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionKeyedHiveTable, ChangeLocationKind.INSTANT, HiveTestRecords.changeInsertRecords(), "change");
+  }
+
+  @Test
+  public void testUnPartitionKeyedTableBaseWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionKeyedHiveTable, BaseLocationKind.INSTANT, HiveTestRecords.baseRecords(), "base");
+  }
+
+  @Test
+  public void testUnPartitionKeyedTableHiveWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionKeyedHiveTable, HiveLocationKind.INSTANT, HiveTestRecords.baseRecords(), "hive");
   }
 
   @Test
@@ -113,7 +128,26 @@ public class AdaptHiveWriterTest extends HiveTableTestBase {
 
   @Test
   public void testUnKeyedTableHiveWriteByLocationKind() throws IOException {
-    testWrite(testHiveTable, HiveLocationKind.INSTANT, HiveTestRecords.baseRecords(), "hive_data");
+    testWrite(testHiveTable, HiveLocationKind.INSTANT, HiveTestRecords.baseRecords(), "hive");
+  }
+
+  @Test
+  public void testUnPartitionUnKeyedTableChangeWriteByLocationKind() throws IOException {
+    try {
+      testWrite(testUnPartitionHiveTable, ChangeLocationKind.INSTANT, HiveTestRecords.changeInsertRecords(), "change");
+    }catch (Exception e){
+      Assert.assertTrue(e instanceof IllegalArgumentException);
+    }
+  }
+
+  @Test
+  public void testUnPartitionUnKeyedTableBaseWriteByLocationKind() throws IOException {
+    testWrite(testUnPartitionHiveTable, BaseLocationKind.INSTANT, HiveTestRecords.baseRecords(), "base");
+  }
+
+  @Test
+  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 {
@@ -138,7 +172,7 @@ public class AdaptHiveWriterTest extends HiveTableTestBase {
 
   private CloseableIterable<Record> readParquet(Schema schema, String path){
     AdaptHiveParquet.ReadBuilder builder = AdaptHiveParquet.read(
-            Files.localInput(new File(path)))
+            Files.localInput(path))
         .project(schema)
         .createReaderFunc(fileSchema -> AdaptHiveGenericParquetReaders.buildReader(schema, fileSchema, new HashMap<>()))
         .caseSensitive(false);
diff --git a/hive/src/test/java/com/netease/arctic/hive/io/HiveTestRecords.java b/hive/src/test/java/com/netease/arctic/hive/io/HiveTestRecords.java
index a674eda90..2512ba1ef 100644
--- a/hive/src/test/java/com/netease/arctic/hive/io/HiveTestRecords.java
+++ b/hive/src/test/java/com/netease/arctic/hive/io/HiveTestRecords.java
@@ -36,16 +36,6 @@ public class HiveTestRecords {
     GenericRecord record = GenericRecord.create(HIVE_TABLE_SCHEMA);
 
     ImmutableList.Builder<Record> builder = ImmutableList.builder();
-    builder.add(record.copy(ImmutableMap.of("id", 1, "name", "john",
-        "op_time", LocalDateTime.of(2022, 1, 1, 12, 0, 0),
-        "op_time_with_zone", OffsetDateTime.of(
-            LocalDateTime.of(2022, 1, 1, 12, 0, 0), ZoneOffset.UTC),
-        "d", new BigDecimal("100"))));
-    builder.add(record.copy(ImmutableMap.of("id", 2, "name", "lily",
-        "op_time", LocalDateTime.of(2022, 1, 2, 12, 0, 0),
-        "op_time_with_zone", OffsetDateTime.of(
-            LocalDateTime.of(2022, 1, 2, 12, 0, 0), ZoneOffset.UTC),
-        "d", new BigDecimal("101"))));
     builder.add(record.copy(ImmutableMap.of("id", 3, "name", "jake",
         "op_time", LocalDateTime.of(2022, 1, 3, 12, 0, 0),
         "op_time_with_zone", OffsetDateTime.of(
@@ -60,6 +50,23 @@ public class HiveTestRecords {
     return builder.build();
   }
 
+  public static List<Record> hiveRecords() {
+    GenericRecord record = GenericRecord.create(HIVE_TABLE_SCHEMA);
+
+    ImmutableList.Builder<Record> builder = ImmutableList.builder();
+    builder.add(record.copy(ImmutableMap.of("id", 1, "name", "john",
+        "op_time", LocalDateTime.of(2022, 1, 1, 12, 0, 0),
+        "op_time_with_zone", OffsetDateTime.of(
+            LocalDateTime.of(2022, 1, 1, 12, 0, 0), ZoneOffset.UTC),
+        "d", new BigDecimal("100"))));
+    builder.add(record.copy(ImmutableMap.of("id", 2, "name", "lily",
+        "op_time", LocalDateTime.of(2022, 1, 2, 12, 0, 0),
+        "op_time_with_zone", OffsetDateTime.of(
+            LocalDateTime.of(2022, 1, 2, 12, 0, 0), ZoneOffset.UTC),
+        "d", new BigDecimal("101"))));
+    return builder.build();
+  }
+
   public static List<Record> changeInsertRecords() {
     GenericRecord record = GenericRecord.create(HIVE_TABLE_SCHEMA);
 
@@ -85,6 +92,16 @@ public class HiveTestRecords {
         "op_time_with_zone", OffsetDateTime.of(
             LocalDateTime.of(2022, 1, 1, 12, 0, 0), ZoneOffset.UTC),
         "d", new BigDecimal("104"))));
+    builder.add(record.copy(ImmutableMap.of("id", 1, "name", "john",
+        "op_time", LocalDateTime.of(2022, 1, 1, 12, 0, 0),
+        "op_time_with_zone", OffsetDateTime.of(
+            LocalDateTime.of(2022, 1, 1, 12, 0, 0), ZoneOffset.UTC),
+        "d", new BigDecimal("100"))));
+    builder.add(record.copy(ImmutableMap.of("id", 3, "name", "jake",
+        "op_time", LocalDateTime.of(2022, 1, 3, 12, 0, 0),
+        "op_time_with_zone", OffsetDateTime.of(
+            LocalDateTime.of(2022, 1, 3, 12, 0, 0), ZoneOffset.UTC),
+        "d", new BigDecimal("102"))));
     return builder.build();
   }
 }
diff --git a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
index 5355bad9d..4bf500a7b 100644
--- a/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
+++ b/spark/v3.1/spark/src/test/java/com/netease/arctic/spark/SparkTestContext.java
@@ -29,7 +29,7 @@ import com.netease.arctic.catalog.CatalogLoader;
 import com.netease.arctic.data.ChangeAction;
 import com.netease.arctic.data.DataTreeNode;
 import com.netease.arctic.data.DefaultKeyedFile;
-import com.netease.arctic.hive.write.AdaptHiveGenericTaskWriterBuilder;
+import com.netease.arctic.hive.io.writer.AdaptHiveGenericTaskWriterBuilder;
 import com.netease.arctic.io.writer.GenericTaskWriters;
 import com.netease.arctic.io.writer.SortedPosDeleteWriter;
 import com.netease.arctic.op.OverwriteBaseFiles;
diff --git a/trino/src/test/java/com/netease/arctic/trino/arctic/ArcticQueryRunner.java b/trino/src/test/java/com/netease/arctic/trino/arctic/ArcticQueryRunner.java
index 0ba1694a9..28e3dae62 100644
--- a/trino/src/test/java/com/netease/arctic/trino/arctic/ArcticQueryRunner.java
+++ b/trino/src/test/java/com/netease/arctic/trino/arctic/ArcticQueryRunner.java
@@ -90,7 +90,6 @@ public final class ArcticQueryRunner
                 queryRunner.installPlugin(new ArcticPlugin());
                 Map<String, String> icebergProperties = new HashMap<>(this.icebergProperties.buildOrThrow());
                 queryRunner.createCatalog(ARCTIC_CATALOG, "arctic", icebergProperties);
-
                 return queryRunner;
             }
             catch (Exception e) {
diff --git a/trino/src/test/java/com/netease/arctic/trino/arctic/TableTestBaseForTrino.java b/trino/src/test/java/com/netease/arctic/trino/arctic/TableTestBaseForTrino.java
index dd0027426..218996094 100644
--- a/trino/src/test/java/com/netease/arctic/trino/arctic/TableTestBaseForTrino.java
+++ b/trino/src/test/java/com/netease/arctic/trino/arctic/TableTestBaseForTrino.java
@@ -59,13 +59,14 @@ import java.time.LocalDateTime;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Set;
+import org.junit.rules.TemporaryFolder;
 
 import static com.netease.arctic.ams.api.MockArcticMetastoreServer.TEST_CATALOG_NAME;
 import static com.netease.arctic.ams.api.MockArcticMetastoreServer.TEST_DB_NAME;
 
 public abstract class TableTestBaseForTrino extends AbstractTestQueryFramework {
 
-  protected static File tableDir = new File("./arctic_unit_test");
+  protected static TemporaryFolder tmp = new TemporaryFolder();
 
   protected static final MockArcticMetastoreServer AMS = MockArcticMetastoreServer.getInstance();
 
@@ -111,11 +112,9 @@ public abstract class TableTestBaseForTrino extends AbstractTestQueryFramework {
   protected KeyedTable testKeyedTable;
 
   protected void setupTables() throws Exception {
-    if (tableDir.exists()){
-      FileUtils.deleteDirectory(tableDir);
-    }
     testCatalog = CatalogLoader.load(AMS.getUrl());
 
+    File tableDir = tmp.newFolder();
     testTable = testCatalog
         .newTableBuilder(TABLE_ID, TABLE_SCHEMA)
         .withProperty(TableProperties.LOCATION, tableDir.getPath() + "/table")
diff --git a/trino/src/test/java/com/netease/arctic/trino/arctic/TestBaseArcticPrimaryTable.java b/trino/src/test/java/com/netease/arctic/trino/arctic/TestBaseArcticPrimaryTable.java
index 06c1a0206..cd61ed6d6 100644
--- a/trino/src/test/java/com/netease/arctic/trino/arctic/TestBaseArcticPrimaryTable.java
+++ b/trino/src/test/java/com/netease/arctic/trino/arctic/TestBaseArcticPrimaryTable.java
@@ -32,6 +32,7 @@ public class TestBaseArcticPrimaryTable extends TableTestBaseWithInitDataForTrin
   @Override
   protected QueryRunner createQueryRunner() throws Exception {
     // setupAMS();
+    tmp.create();
     setupTables();
     initData();
     return ArcticQueryRunner.builder()
diff --git a/trino/src/test/java/com/netease/arctic/trino/arctic/TestHiveTableBaseForTrino.java b/trino/src/test/java/com/netease/arctic/trino/arctic/TestHiveTableBaseForTrino.java
new file mode 100644
index 000000000..8cff3cb20
--- /dev/null
+++ b/trino/src/test/java/com/netease/arctic/trino/arctic/TestHiveTableBaseForTrino.java
@@ -0,0 +1,268 @@
+/*
+ * 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.trino.arctic;
+
+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.CatalogLoader;
+import com.netease.arctic.hive.HMSMockServer;
+import com.netease.arctic.hive.catalog.ArcticHiveCatalog;
+import com.netease.arctic.hive.table.KeyedHiveTable;
+import com.netease.arctic.hive.table.UnkeyedHiveTable;
+import com.netease.arctic.table.ArcticTable;
+import com.netease.arctic.table.TableIdentifier;
+import com.netease.arctic.table.TableProperties;
+import io.trino.testing.QueryRunner;
+import java.io.File;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.relocated.com.google.common.base.Joiner;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.thrift.TException;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.rules.TemporaryFolder;
+
+import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_HIVE;
+
+public abstract class TestHiveTableBaseForTrino extends TableTestBaseForTrino {
+  protected static final String HIVE_DB_NAME = "hivedb";
+  protected static final String HIVE_CATALOG_NAME = "hive_catalog";
+  protected static final AtomicInteger testCount = new AtomicInteger(0);
+
+  protected static final TemporaryFolder tempFolder = new TemporaryFolder();
+
+  protected static HMSMockServer hms  ;
+
+  protected static final TableIdentifier HIVE_TABLE_ID =
+      TableIdentifier.of(HIVE_CATALOG_NAME, HIVE_DB_NAME, "test_hive_table");
+  protected static final TableIdentifier HIVE_PK_TABLE_ID =
+      TableIdentifier.of(HIVE_CATALOG_NAME, HIVE_DB_NAME, "test_pk_hive_table");
+
+  protected static final TableIdentifier UN_PARTITION_HIVE_TABLE_ID =
+      TableIdentifier.of(HIVE_CATALOG_NAME, HIVE_DB_NAME, "un_partition_test_hive_table");
+  protected static final TableIdentifier UN_PARTITION_HIVE_PK_TABLE_ID =
+      TableIdentifier.of(HIVE_CATALOG_NAME, HIVE_DB_NAME, "un_partition_test_pk_hive_table");
+
+  public static final Schema HIVE_TABLE_SCHEMA = new Schema(
+      Types.NestedField.required(1, "id", Types.IntegerType.get()),
+      Types.NestedField.required(2, "name", Types.StringType.get()),
+      Types.NestedField.required(3, "op_time", Types.TimestampType.withoutZone()),
+      Types.NestedField.required(4, "op_time_with_zone", Types.TimestampType.withZone()),
+      Types.NestedField.required(5, "d", Types.DecimalType.of(3, 0))
+  );
+
+  protected static final PartitionSpec HIVE_SPEC =
+      PartitionSpec.builderFor(HIVE_TABLE_SCHEMA).identity("name").build();
+
+  protected ArcticHiveCatalog hiveCatalog;
+  protected UnkeyedHiveTable testHiveTable;
+  protected KeyedHiveTable testKeyedHiveTable;
+
+  protected UnkeyedHiveTable testUnPartitionHiveTable;
+  protected KeyedHiveTable testUnPartitionKeyedHiveTable;
+
+
+  protected static void startMetastore() throws Exception {
+    int ref = testCount.incrementAndGet();
+    if (ref == 1){
+      tempFolder.create();
+      hms = new HMSMockServer(tempFolder.newFolder("hive"));
+      hms.start();
+
+      String dbPath = hms.getDatabasePath(HIVE_DB_NAME);
+      Database db = new Database(HIVE_DB_NAME, "description", dbPath, new HashMap<>());
+      hms.getClient().createDatabase(db);
+
+      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"));
+
+      Map<String, String> catalogProperties = new HashMap<>();
+
+      CatalogMeta catalogMeta = new CatalogMeta(HIVE_CATALOG_NAME, CATALOG_TYPE_HIVE,
+          storageConfig, authConfig, catalogProperties);
+      AMS.createCatalogIfAbsent(catalogMeta);
+    }
+  }
+
+
+  protected static void stopMetastore() {
+    int ref = testCount.decrementAndGet();
+    if (ref == 0){
+      hms.stop();
+      hms = null;
+      tempFolder.delete();
+    }
+  }
+
+
+  protected void setupTables() throws Exception {
+    hiveCatalog = (ArcticHiveCatalog) CatalogLoader.load(AMS.getUrl(HIVE_CATALOG_NAME));
+    File tableDir = tmp.newFolder();
+
+    testHiveTable = (UnkeyedHiveTable) hiveCatalog
+        .newTableBuilder(HIVE_TABLE_ID, HIVE_TABLE_SCHEMA)
+        .withProperty(TableProperties.LOCATION, tableDir.getPath() + "/table")
+        .withPartitionSpec(HIVE_SPEC)
+        .create().asUnkeyedTable();
+
+    testUnPartitionHiveTable = (UnkeyedHiveTable) hiveCatalog
+        .newTableBuilder(UN_PARTITION_HIVE_TABLE_ID, HIVE_TABLE_SCHEMA)
+        .withProperty(TableProperties.LOCATION, tableDir.getPath() + "/un_partition_table")
+        .create().asUnkeyedTable();
+
+    testKeyedHiveTable = (KeyedHiveTable) hiveCatalog
+        .newTableBuilder(HIVE_PK_TABLE_ID, HIVE_TABLE_SCHEMA)
+        .withProperty(TableProperties.LOCATION, tableDir.getPath() + "/pk_table")
+        .withPartitionSpec(HIVE_SPEC)
+        .withPrimaryKeySpec(PRIMARY_KEY_SPEC)
+        .create().asKeyedTable();
+
+    testUnPartitionKeyedHiveTable = (KeyedHiveTable) hiveCatalog
+        .newTableBuilder(UN_PARTITION_HIVE_PK_TABLE_ID, HIVE_TABLE_SCHEMA)
+        .withProperty(TableProperties.LOCATION, tableDir.getPath() + "/un_partition_pk_table")
+        .withPrimaryKeySpec(PRIMARY_KEY_SPEC)
+        .create().asKeyedTable();
+  }
+
+
+  protected void clearTable() {
+    hiveCatalog.dropTable(HIVE_TABLE_ID, true);
+    AMS.handler().getTableCommitMetas().remove(HIVE_TABLE_ID.buildTableIdentifier());
+
+    hiveCatalog.dropTable(UN_PARTITION_HIVE_TABLE_ID, true);
+    AMS.handler().getTableCommitMetas().remove(UN_PARTITION_HIVE_TABLE_ID.buildTableIdentifier());
+
+    hiveCatalog.dropTable(HIVE_PK_TABLE_ID, true);
+    AMS.handler().getTableCommitMetas().remove(HIVE_PK_TABLE_ID.buildTableIdentifier());
+
+    hiveCatalog.dropTable(UN_PARTITION_HIVE_PK_TABLE_ID, true);
+    AMS.handler().getTableCommitMetas().remove(UN_PARTITION_HIVE_PK_TABLE_ID.buildTableIdentifier());
+  }
+
+  public static class DataFileBuilder {
+    final TableIdentifier identifier;
+    final Table hiveTable;
+    final ArcticTable table;
+
+    public DataFileBuilder(ArcticTable table) throws TException {
+      identifier = table.id();
+      this.table = table;
+      hiveTable = hms.getClient().getTable(identifier.getDatabase(), identifier.getTableName());
+    }
+
+    public DataFile build(String valuePath, String path) {
+      DataFiles.Builder builder =  DataFiles.builder(table.spec())
+          .withPath(hiveTable.getSd().getLocation() + path)
+          .withFileSizeInBytes(0)
+          .withRecordCount(2);
+
+      if (!StringUtils.isEmpty(valuePath)){
+        builder = builder.withPartitionPath(valuePath);
+      }
+      return builder.build();
+    }
+
+    public List<DataFile> buildList(List<Map.Entry<String, String>> partValueFiles){
+      return partValueFiles.stream().map(
+          kv -> this.build(kv.getKey(), kv.getValue())
+      ).collect(Collectors.toList());
+    }
+  }
+
+  public static String getPartitionPath(List<String> values, PartitionSpec spec) {
+    List<String> nameValues = Lists.newArrayList();
+    for (int i = 0; i < values.size(); i++) {
+      String field = spec.fields().get(i).name();
+      String value = values.get(i);
+      nameValues.add(field + "=" + value);
+    }
+    return Joiner.on("/").join(nameValues);
+  }
+
+  /**
+   * assert hive table partition location as expected
+   * @param partitionLocations
+   * @param table
+   * @throws TException
+   */
+  public static void assertHivePartitionLocations(Map<String, String> partitionLocations, ArcticTable table) throws TException {
+    TableIdentifier identifier = table.id();
+    final String database = identifier.getDatabase();
+    final String tableName = identifier.getTableName();
+
+    List<Partition> partitions = hms.getClient().listPartitions(
+        database,
+        tableName,
+        (short) -1);
+
+    System.out.println("> assert hive partition location as expected");
+    System.out.printf("HiveTable[%s.%s] partition count: %d \n", database, tableName, partitions.size());
+    for (Partition p: partitions){
+      System.out.printf(
+          "HiveTablePartition[%s.%s  %s] location:%s \n", database, tableName,
+          Joiner.on("/").join(p.getValues()), p.getSd().getLocation());
+    }
+
+    Assert.assertEquals("expect " + partitionLocations.size() + " partition after first rewrite partition",
+        partitionLocations.size(), partitions.size());
+
+    for (Partition p : partitions) {
+      String valuePath = getPartitionPath(p.getValues(), table.spec());
+      Assert.assertTrue("partition " + valuePath + " is not expected",
+          partitionLocations.containsKey(valuePath));
+
+      String locationExpect = partitionLocations.get(valuePath);
+      String actualLocation = p.getSd().getLocation();
+      Assert.assertTrue(
+          "partition location is not expected, expect " + actualLocation + " end-with " + locationExpect,
+          actualLocation.contains(locationExpect));
+    }
+  }
+}
diff --git a/trino/src/test/java/com/netease/arctic/trino/arctic/TestHiveTble.java b/trino/src/test/java/com/netease/arctic/trino/arctic/TestHiveTble.java
new file mode 100644
index 000000000..26f9d4578
--- /dev/null
+++ b/trino/src/test/java/com/netease/arctic/trino/arctic/TestHiveTble.java
@@ -0,0 +1,214 @@
+/*
+ * 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.trino.arctic;
+
+import com.google.common.collect.ImmutableMap;
+import com.netease.arctic.data.ChangeAction;
+import com.netease.arctic.hive.io.HiveTestRecords;
+import com.netease.arctic.hive.io.writer.AdaptHiveGenericTaskWriterBuilder;
+import com.netease.arctic.hive.table.HiveLocationKind;
+import com.netease.arctic.table.ArcticTable;
+import com.netease.arctic.table.BaseLocationKind;
+import com.netease.arctic.table.ChangeLocationKind;
+import com.netease.arctic.table.LocationKind;
+import io.trino.testing.QueryRunner;
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.OverwriteFiles;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.parquet.AdaptHiveGenericParquetReaders;
+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.testng.annotations.AfterClass;
+import org.testng.annotations.Test;
+
+public class TestHiveTble extends TestHiveTableBaseForTrino{
+
+  private final String TEST_HIVE_TABLE_FULL_NAME = "arctic." + HIVE_TABLE_ID.getDatabase() + "." + HIVE_TABLE_ID.getTableName();
+
+  private final String TEST_HIVE_PK_TABLE_FULL_NAME =
+      "arctic." + HIVE_PK_TABLE_ID.getDatabase() + "." + HIVE_PK_TABLE_ID.getTableName();
+
+  private final String TEST_HIVE_PK_TABLE_FULL_NAME_BASE =
+      "arctic." + HIVE_PK_TABLE_ID.getDatabase() + "." + "\"" + HIVE_PK_TABLE_ID.getTableName() + "#base\"";
+
+  private final String TEST_UN_PARTITION_HIVE_TABLE_FULL_NAME =
+      "arctic." + UN_PARTITION_HIVE_TABLE_ID.getDatabase() + "." + UN_PARTITION_HIVE_TABLE_ID.getTableName();
+
+  private final String TEST_UN_PARTITION_HIVE_PK_TABLE_FULL_NAME =
+      "arctic." + UN_PARTITION_HIVE_PK_TABLE_ID.getDatabase() + "." + UN_PARTITION_HIVE_PK_TABLE_ID.getTableName();
+
+  private final String TEST_UN_PARTITION_HIVE_PK_TABLE_FULL_NAME_BASE =
+      "arctic." + UN_PARTITION_HIVE_PK_TABLE_ID.getDatabase() + "." + "\"" + UN_PARTITION_HIVE_PK_TABLE_ID.getTableName() + "#base\"";
+
+  private int txid = 1;
+
+  @Override
+  protected QueryRunner createQueryRunner() throws Exception {
+    tmp.create();
+    tempFolder.create();
+    startMetastore();
+    setupTables();
+    initData();
+    return ArcticQueryRunner.builder()
+        .setExtraProperties(ImmutableMap.of("http-server.http.port", "8080"))
+        .setIcebergProperties(ImmutableMap.of("arctic.url",
+            String.format("thrift://localhost:%s/%s",AMS.port(), HIVE_CATALOG_NAME)))
+        .build();
+  }
+
+  private void initData() throws IOException {
+    write(testHiveTable, BaseLocationKind.INSTANT, HiveTestRecords.baseRecords());
+    write(testHiveTable, HiveLocationKind.INSTANT, HiveTestRecords.hiveRecords());
+
+    write(testKeyedHiveTable, ChangeLocationKind.INSTANT, HiveTestRecords.changeInsertRecords());
+    write(testKeyedHiveTable, BaseLocationKind.INSTANT, HiveTestRecords.baseRecords());
+    write(testKeyedHiveTable, HiveLocationKind.INSTANT, HiveTestRecords.hiveRecords());
+    write(testKeyedHiveTable, ChangeLocationKind.INSTANT, HiveTestRecords.changeDeleteRecords(), ChangeAction.DELETE);
+
+    write(testUnPartitionHiveTable, BaseLocationKind.INSTANT, HiveTestRecords.baseRecords());
+    write(testUnPartitionHiveTable, HiveLocationKind.INSTANT, HiveTestRecords.hiveRecords());
+
+    write(testUnPartitionKeyedHiveTable, ChangeLocationKind.INSTANT, HiveTestRecords.changeInsertRecords());
+    write(testUnPartitionKeyedHiveTable, BaseLocationKind.INSTANT, HiveTestRecords.baseRecords());
+    write(testUnPartitionKeyedHiveTable, HiveLocationKind.INSTANT, HiveTestRecords.hiveRecords());
+    write(testUnPartitionKeyedHiveTable, ChangeLocationKind.INSTANT, HiveTestRecords.changeDeleteRecords(), ChangeAction.DELETE);
+  }
+
+  @Test
+  public void testHiveTableMOR() {
+    //H2 unsupported timestamp with time zone, timestamp with time zone need test manually
+    assertQuery("select id, name, op_time, d from " + TEST_HIVE_TABLE_FULL_NAME, "VALUES" +
+        "(1, 'john', TIMESTAMP'2022-01-01 12:00:00.000000', 100)," +
+        "(2, 'lily', TIMESTAMP'2022-01-02 12:00:00.000000', 101)," +
+        "(3, 'jake', TIMESTAMP'2022-01-03 12:00:00.000000', 102)," +
+        "(4, 'sam', TIMESTAMP'2022-01-04 12:00:00.000000', 103)");
+  }
+
+  @Test
+  public void testKeyedHiveTableMOR() {
+    //H2 unsupported timestamp with time zone, timestamp with time zone need test manually
+    assertQuery("select id, name, op_time, d from " + TEST_HIVE_PK_TABLE_FULL_NAME, "VALUES" +
+        "(2, 'lily', TIMESTAMP'2022-01-02 12:00:00.000000', 101)," +
+        "(4, 'sam', TIMESTAMP'2022-01-04 12:00:00.000000', 103)," +
+        "(6, 'mack', TIMESTAMP'2022-01-01 12:00:00.000000', 105)");
+  }
+
+  @Test
+  public void testKeyedHiveTableBase() {
+    //H2 unsupported timestamp with time zone, timestamp with time zone need test manually
+    assertQuery("select id, name, op_time, d from " + TEST_HIVE_PK_TABLE_FULL_NAME_BASE, "VALUES" +
+        "(1, 'john', TIMESTAMP'2022-01-01 12:00:00.000000', 100)," +
+        "(2, 'lily', TIMESTAMP'2022-01-02 12:00:00.000000', 101)," +
+        "(3, 'jake', TIMESTAMP'2022-01-03 12:00:00.000000', 102)," +
+        "(4, 'sam', TIMESTAMP'2022-01-04 12:00:00.000000', 103)");
+  }
+
+  @Test
+  public void testNoPartitionHiveTableMOR() {
+    //H2 unsupported timestamp with time zone, timestamp with time zone need test manually
+    assertQuery("select id, name, op_time, d from " + TEST_UN_PARTITION_HIVE_TABLE_FULL_NAME, "VALUES" +
+        "(1, 'john', TIMESTAMP'2022-01-01 12:00:00.000000', 100)," +
+        "(2, 'lily', TIMESTAMP'2022-01-02 12:00:00.000000', 101)," +
+        "(3, 'jake', TIMESTAMP'2022-01-03 12:00:00.000000', 102)," +
+        "(4, 'sam', TIMESTAMP'2022-01-04 12:00:00.000000', 103)");
+  }
+
+  @Test
+  public void testNoPartitionKeyedHiveTableMOR() {
+    //H2 unsupported timestamp with time zone, timestamp with time zone need test manually
+    assertQuery("select id, name, op_time, d from " + TEST_UN_PARTITION_HIVE_PK_TABLE_FULL_NAME, "VALUES" +
+        "(2, 'lily', TIMESTAMP'2022-01-02 12:00:00.000000', 101)," +
+        "(4, 'sam', TIMESTAMP'2022-01-04 12:00:00.000000', 103)," +
+        "(6, 'mack', TIMESTAMP'2022-01-01 12:00:00.000000', 105)");
+  }
+
+  @Test
+  public void testNoPartitionKeyedHiveTableBase() {
+    //H2 unsupported timestamp with time zone, timestamp with time zone need test manually
+    assertQuery("select id, name, op_time, d from " + TEST_UN_PARTITION_HIVE_PK_TABLE_FULL_NAME_BASE, "VALUES" +
+        "(1, 'john', TIMESTAMP'2022-01-01 12:00:00.000000', 100)," +
+        "(2, 'lily', TIMESTAMP'2022-01-02 12:00:00.000000', 101)," +
+        "(3, 'jake', TIMESTAMP'2022-01-03 12:00:00.000000', 102)," +
+        "(4, 'sam', TIMESTAMP'2022-01-04 12:00:00.000000', 103)");
+  }
+
+  @AfterClass
+  public void clear(){
+    clearTable();
+    stopMetastore();
+  }
+
+  private void write(ArcticTable table, LocationKind locationKind, List<Record> records) throws
+      IOException {
+    write(table, locationKind, records, ChangeAction.INSERT);
+  }
+
+  private void write(ArcticTable table, LocationKind locationKind, List<Record> records, ChangeAction changeAction) throws
+      IOException {
+    AdaptHiveGenericTaskWriterBuilder builder = AdaptHiveGenericTaskWriterBuilder
+        .builderFor(table)
+        .withChangeAction(changeAction)
+        .withTransactionId(txid++);
+
+    TaskWriter<Record> changeWrite = builder.buildWriter(locationKind);
+    for (Record record: records) {
+      changeWrite.write(record);
+    }
+    WriteResult complete = changeWrite.complete();
+    if (locationKind == ChangeLocationKind.INSTANT) {
+      AppendFiles appendFiles = table.asKeyedTable().changeTable().newAppend();
+      Arrays.stream(complete.dataFiles()).forEach(s -> appendFiles.appendFile(s));
+      appendFiles.commit();
+    } else {
+      if (table.isUnkeyedTable()) {
+        OverwriteFiles overwriteFiles = table.asUnkeyedTable().newOverwrite();
+        Arrays.stream(complete.dataFiles()).forEach(s -> overwriteFiles.addFile(s));
+        overwriteFiles.commit();
+      } else {
+        OverwriteFiles overwriteFiles = table.asKeyedTable().baseTable().newOverwrite();
+        Arrays.stream(complete.dataFiles()).forEach(s -> overwriteFiles.addFile(s));
+        overwriteFiles.commit();
+      }
+    }
+  }
+
+  private CloseableIterable<Record> readParquet(Schema schema, String path){
+    AdaptHiveParquet.ReadBuilder builder = AdaptHiveParquet.read(
+            Files.localInput(new File(path)))
+        .project(schema)
+        .createReaderFunc(fileSchema -> AdaptHiveGenericParquetReaders.buildReader(schema, fileSchema, new HashMap<>()))
+        .caseSensitive(false);
+
+    CloseableIterable<Record> iterable = builder.build();
+    return iterable;
+  }
+
+  private String base(String table) {
+    return "\"" + table + "#base\"";
+  }
+}
diff --git a/trino/src/test/java/com/netease/arctic/trino/iceberg/TestArcticConnectorFactory.java b/trino/src/test/java/com/netease/arctic/trino/iceberg/TestArcticConnectorFactory.java
index a02f86c53..513d2bb51 100644
--- a/trino/src/test/java/com/netease/arctic/trino/iceberg/TestArcticConnectorFactory.java
+++ b/trino/src/test/java/com/netease/arctic/trino/iceberg/TestArcticConnectorFactory.java
@@ -22,7 +22,6 @@ import com.google.inject.Injector;
 import com.google.inject.Key;
 import com.google.inject.TypeLiteral;
 import com.netease.arctic.trino.ArcticConnector;
-import com.netease.arctic.trino.ArcticModule;
 import com.netease.arctic.trino.ArcticTransactionManager;
 import io.airlift.bootstrap.Bootstrap;
 import io.airlift.bootstrap.LifeCycleManager;
diff --git a/trino/src/test/java/com/netease/arctic/trino/iceberg/TestUnionModule.java b/trino/src/test/java/com/netease/arctic/trino/iceberg/TestUnionModule.java
index 5dad7f983..aef7d0e33 100644
--- a/trino/src/test/java/com/netease/arctic/trino/iceberg/TestUnionModule.java
+++ b/trino/src/test/java/com/netease/arctic/trino/iceberg/TestUnionModule.java
@@ -96,7 +96,7 @@ public class TestUnionModule implements Module {
         //############# IcebergModule
         configBinder(binder).bindConfig(MetastoreConfig.class);
         configBinder(binder).bindConfig(IcebergConfig.class);
-        configBinder(binder).bindConfig(HiveConfig.class);
+
         newSetBinder(binder, SessionPropertiesProvider.class).addBinding().to(IcebergSessionProperties.class).in(Scopes.SINGLETON);
         binder.bind(IcebergTableProperties.class).in(Scopes.SINGLETON);