Repository: NetEase/arctic
Pull request #148 (2022-08-09 11:12:25)

remove method in BaseTable interface. new operations from KeyedTable


Patch:
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/MinorOptimizePlan.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/MinorOptimizePlan.java
index 4b45e7315..f6d071696 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/MinorOptimizePlan.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/MinorOptimizePlan.java
@@ -274,7 +274,7 @@ public class MinorOptimizePlan extends BaseOptimizePlan {
   private long getBaseMaxTransactionId(String partition) {
     if (baseTableMaxTransactionId == null) {
       baseTableMaxTransactionId = new HashMap<>();
-      baseTableMaxTransactionId.putAll(arcticTable.asKeyedTable().baseTable().maxTransactionId());
+      baseTableMaxTransactionId.putAll(arcticTable.asKeyedTable().maxTransactionId());
       LOG.debug("{} ==== get base table max transaction id: {}", tableId(), baseTableMaxTransactionId);
     }
     Long maxTransactionId = baseTableMaxTransactionId.get(partition);
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/TableOptimizeItem.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/TableOptimizeItem.java
index 621184465..05d817d77 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/TableOptimizeItem.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/TableOptimizeItem.java
@@ -539,7 +539,7 @@ public class TableOptimizeItem extends IJDBCService {
     if (isKeyedTable()) {
       KeyedTable keyedHiveTable = getArcticTable(true).asKeyedTable();
       record.setSnapshotInfo(TableStatCollector.buildBaseTableSnapshotInfo(keyedHiveTable.baseTable()));
-      record.setBaseTableMaxTransactionId(keyedHiveTable.baseTable().maxTransactionId().toString());
+      record.setBaseTableMaxTransactionId(keyedHiveTable.maxTransactionId().toString());
     } else {
       getArcticTable(true);
       record.setSnapshotInfo(TableStatCollector.buildBaseTableSnapshotInfo(getArcticTable(true).asUnkeyedTable()));
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/TableExpireService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/TableExpireService.java
index 81d457336..62802a834 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/TableExpireService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/TableExpireService.java
@@ -32,8 +32,6 @@ import com.netease.arctic.catalog.CatalogLoader;
 import com.netease.arctic.data.DefaultKeyedFile;
 import com.netease.arctic.data.PrimaryKeyedFile;
 import com.netease.arctic.table.ArcticTable;
-import com.netease.arctic.table.BaseTable;
-import com.netease.arctic.table.ChangeTable;
 import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.TableIdentifier;
 import com.netease.arctic.table.TableProperties;
@@ -117,7 +115,7 @@ public class TableExpireService implements ITableExpireService {
         if (arcticTable.isKeyedTable()) {
           KeyedTable keyedArcticTable = arcticTable.asKeyedTable();
           keyedArcticTable.io().doAs(() -> {
-            BaseTable baseTable = keyedArcticTable.baseTable();
+            UnkeyedTable baseTable = keyedArcticTable.baseTable();
             if (baseTable == null) {
               LOG.warn("[{}] Base table is null: {} ", traceId, tableIdentifier);
               return null;
@@ -129,7 +127,7 @@ public class TableExpireService implements ITableExpireService {
             long baseCleanedTime = System.currentTimeMillis();
             LOG.info("[{}] {} base expire cost {} ms", traceId, arcticTable.id(), baseCleanedTime - startTime);
 
-            ChangeTable changeTable = keyedArcticTable.changeTable();
+            UnkeyedTable changeTable = keyedArcticTable.changeTable();
             if (changeTable == null) {
               LOG.warn("[{}] Change table is null: {}", traceId, tableIdentifier);
               return null;
@@ -160,7 +158,7 @@ public class TableExpireService implements ITableExpireService {
   }
 
   public static void deleteChangeFile(KeyedTable keyedTable, List<DataFileInfo> changeDataFiles) {
-    StructLikeMap<Long> baseMaxTransactionId = keyedTable.baseTable().partitionMaxTransactionId();
+    StructLikeMap<Long> baseMaxTransactionId = keyedTable.partitionMaxTransactionId();
     if (MapUtils.isEmpty(baseMaxTransactionId)) {
       LOG.info("table {} not contains max transaction id", keyedTable.id());
       return;
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/TableStatCollector.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/TableStatCollector.java
index d54cbade9..261b88ee8 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/TableStatCollector.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/utils/TableStatCollector.java
@@ -23,8 +23,6 @@ import com.netease.arctic.ams.server.model.FilesStatistics;
 import com.netease.arctic.ams.server.model.SnapshotInfo;
 import com.netease.arctic.ams.server.model.TableStatistics;
 import com.netease.arctic.table.ArcticTable;
-import com.netease.arctic.table.BaseTable;
-import com.netease.arctic.table.ChangeTable;
 import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.TableIdentifier;
 import com.netease.arctic.table.UnkeyedTable;
@@ -46,7 +44,7 @@ public class TableStatCollector {
   @Nonnull
   public static TableStatistics collectChangeTableInfo(KeyedTable table) {
     try {
-      ChangeTable internalTable = null;
+      UnkeyedTable internalTable = null;
       try {
         internalTable = table.changeTable();
       } catch (NoSuchTableException e) {
@@ -65,7 +63,7 @@ public class TableStatCollector {
   public static TableStatistics collectBaseTableInfo(KeyedTable table) {
     try {
       LOG.info("start collect {} table info of {}", Constants.INNER_TABLE_BASE, table.id());
-      BaseTable internalTable = null;
+      UnkeyedTable internalTable = null;
       try {
         internalTable = table.baseTable();
       } catch (org.apache.iceberg.exceptions.NoSuchTableException e) {
diff --git a/core/src/main/java/com/netease/arctic/op/ArcticOperations.java b/core/src/main/java/com/netease/arctic/op/ArcticOperations.java
deleted file mode 100644
index 88c7f393d..000000000
--- a/core/src/main/java/com/netease/arctic/op/ArcticOperations.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.netease.arctic.op;
-
-import com.netease.arctic.table.KeyedTable;
-
-/**
- * Factory to build arctic operations.
- */
-public class ArcticOperations {
-  public static OverwriteBaseFiles newOverwriteBaseFiles(KeyedTable keyedTable) {
-    return new OverwriteBaseFiles(keyedTable);
-  }
-
-  public static RewritePartitions newRewritePartitions(KeyedTable keyedTable) {
-    return new RewritePartitions(keyedTable.baseTable());
-  }
-}
diff --git a/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java b/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java
index 58acb0102..40e5d58f1 100644
--- a/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java
+++ b/core/src/main/java/com/netease/arctic/op/OverwriteBaseFiles.java
@@ -20,6 +20,7 @@ package com.netease.arctic.op;
 
 import com.netease.arctic.scan.CombinedScanTask;
 import com.netease.arctic.table.KeyedTable;
+import com.netease.arctic.table.UnkeyedTable;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
@@ -44,7 +45,6 @@ import java.util.List;
  */
 public class OverwriteBaseFiles extends PartitionTransactionOperation {
 
-  private final KeyedTable table;
   private final List<DataFile> deleteFiles;
   private final List<DataFile> addFiles;
   private final List<DeleteFile> deleteDeleteFiles;
@@ -55,13 +55,12 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
   private Long transactionId;
 
   public OverwriteBaseFiles(KeyedTable table) {
-    super(table.baseTable());
-    this.table = table;
+    super(table);
     this.deleteFiles = Lists.newArrayList();
     this.addFiles = Lists.newArrayList();
     this.deleteDeleteFiles = Lists.newArrayList();
     this.addDeleteFiles = Lists.newArrayList();
-    this.maxTransactionId = StructLikeMap.create(baseTable.spec().partitionType());
+    this.maxTransactionId = StructLikeMap.create(table.spec().partitionType());
   }
 
   public OverwriteBaseFiles overwriteByRowFilter(Expression expr) {
@@ -106,6 +105,7 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
     applyDeleteExpression();
 
     StructLike partitionData = null;
+    UnkeyedTable baseTable = keyedTable.baseTable();
 
     // step1: overwrite data files
     OverwriteFiles overwriteFiles = transaction.newOverwrite();
@@ -114,13 +114,13 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
     }
     for (DataFile d : this.addFiles) {
       overwriteFiles.addFile(d);
-      partitionData = table.spec().isUnpartitioned() ? null : d.partition();
+      partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
       partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
     }
     // 为什么？
     for (DataFile d : this.deleteFiles) {
       overwriteFiles.deleteFile(d);
-      partitionData = table.spec().isUnpartitioned() ? null : d.partition();
+      partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
       partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
     }
     overwriteFiles.commit();
@@ -134,12 +134,12 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
         }
 
         for (DeleteFile d : this.addDeleteFiles) {
-          partitionData = table.spec().isUnpartitioned() ? null : d.partition();
+          partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
           partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
         }
         // 为什么？
         for (DataFile d : this.deleteFiles) {
-          partitionData = table.spec().isUnpartitioned() ? null : d.partition();
+          partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
           partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
         }
         addDeleteFiles.forEach(rowDelta::addDeletes);
@@ -151,12 +151,12 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
         }
 
         for (DeleteFile d : this.addDeleteFiles) {
-          partitionData = table.spec().isUnpartitioned() ? null : d.partition();
+          partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
           partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
         }
         // 为什么？
         for (DataFile d : this.deleteFiles) {
-          partitionData = table.spec().isUnpartitioned() ? null : d.partition();
+          partitionData = keyedTable.spec().isUnpartitioned() ? null : d.partition();
           partitionMaxTxId.put(partitionData, getPartitionMaxTxId(partitionData));
         }
         rewriteFiles.rewriteFiles(Collections.emptySet(), new HashSet<>(deleteDeleteFiles),
@@ -166,7 +166,7 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
     }
 
     // step3: set max transaction id
-    if (table.spec().isUnpartitioned()) {
+    if (keyedTable.spec().isUnpartitioned()) {
       long maxTransactionId = partitionMaxTxId.get(partitionData);
       partitionMaxTxId.put(partitionData, Math.max(maxTransactionId, this.maxTransactionId.get(partitionData)));
     } else {
@@ -186,7 +186,7 @@ public class OverwriteBaseFiles extends PartitionTransactionOperation {
       return;
     }
     try (CloseableIterable<CombinedScanTask> combinedScanTasks
-        = table.newScan().filter(deleteExpression).planTasks()) {
+        = keyedTable.newScan().filter(deleteExpression).planTasks()) {
       combinedScanTasks.forEach(combinedTask -> combinedTask.tasks().forEach(
           t -> {
             t.dataTasks().forEach(ft -> deleteFiles.add(ft.file()));
diff --git a/core/src/main/java/com/netease/arctic/op/PartitionTransactionOperation.java b/core/src/main/java/com/netease/arctic/op/PartitionTransactionOperation.java
index a47bf8025..2dbb6c883 100644
--- a/core/src/main/java/com/netease/arctic/op/PartitionTransactionOperation.java
+++ b/core/src/main/java/com/netease/arctic/op/PartitionTransactionOperation.java
@@ -19,6 +19,7 @@
 package com.netease.arctic.op;
 
 import com.netease.arctic.table.BaseTable;
+import com.netease.arctic.table.KeyedTable;
 import com.netease.arctic.table.TableProperties;
 import com.netease.arctic.utils.TablePropertyUtil;
 import org.apache.iceberg.Transaction;
@@ -31,10 +32,10 @@ import org.apache.iceberg.util.StructLikeMap;
  */
 public abstract class PartitionTransactionOperation {
 
-  BaseTable baseTable;
+  KeyedTable keyedTable;
 
-  public PartitionTransactionOperation(BaseTable baseTable) {
-    this.baseTable = baseTable;
+  public PartitionTransactionOperation(KeyedTable baseTable) {
+    this.keyedTable = baseTable;
   }
 
   /**
@@ -47,11 +48,11 @@ public abstract class PartitionTransactionOperation {
   protected abstract StructLikeMap<Long> apply(Transaction transaction, StructLikeMap<Long> partitionMaxTxId);
 
   public void commit() {
-    Transaction tx = baseTable.newTransaction();
+    Transaction tx = keyedTable.baseTable().newTransaction();
 
-    StructLikeMap<Long> partitionMaxTxId = apply(tx, baseTable.partitionMaxTransactionId());
+    StructLikeMap<Long> partitionMaxTxId = apply(tx, keyedTable.partitionMaxTransactionId());
 
-    String propertyValue = TablePropertyUtil.encodePartitionMaxTxId(baseTable.spec(), partitionMaxTxId);
+    String propertyValue = TablePropertyUtil.encodePartitionMaxTxId(keyedTable.spec(), partitionMaxTxId);
     UpdateProperties updateProperties = tx.updateProperties();
     updateProperties.set(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID, propertyValue);
     updateProperties.commit();
diff --git a/core/src/main/java/com/netease/arctic/op/RewritePartitions.java b/core/src/main/java/com/netease/arctic/op/RewritePartitions.java
index 9a2478b82..0b6926207 100644
--- a/core/src/main/java/com/netease/arctic/op/RewritePartitions.java
+++ b/core/src/main/java/com/netease/arctic/op/RewritePartitions.java
@@ -19,6 +19,7 @@
 package com.netease.arctic.op;
 
 import com.netease.arctic.table.BaseTable;
+import com.netease.arctic.table.KeyedTable;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.ReplacePartitions;
 import org.apache.iceberg.StructLike;
@@ -37,8 +38,8 @@ public class RewritePartitions extends PartitionTransactionOperation {
   private List<DataFile> addFiles = Lists.newArrayList();
   private Long transactionId;
   
-  public RewritePartitions(BaseTable baseTable) {
-    super(baseTable);
+  public RewritePartitions(KeyedTable keyedTable) {
+    super(keyedTable);
   }
 
   public RewritePartitions addDataFile(DataFile dataFile) {
diff --git a/core/src/main/java/com/netease/arctic/scan/BaseKeyedTableScan.java b/core/src/main/java/com/netease/arctic/scan/BaseKeyedTableScan.java
index c1aca3916..a439bfeb2 100644
--- a/core/src/main/java/com/netease/arctic/scan/BaseKeyedTableScan.java
+++ b/core/src/main/java/com/netease/arctic/scan/BaseKeyedTableScan.java
@@ -230,7 +230,7 @@ public class BaseKeyedTableScan implements KeyedTableScan {
       List<ArcticFileScanTask> baseTasks) {
     ListMultimap<StructLike, ArcticFileScanTask> filesGroupedByPartition
         = Multimaps.newListMultimap(Maps.newHashMap(), Lists::newArrayList);
-    StructLikeMap<Long> partitionMaxTxId = table.baseTable().partitionMaxTransactionId();
+    StructLikeMap<Long> partitionMaxTxId = table.partitionMaxTransactionId();
 
     // filter change files according to max transaction id
     changeTasks.forEach(task -> {
diff --git a/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java b/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
index 8cc802aa8..dc4add212 100644
--- a/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
@@ -24,6 +24,8 @@ import com.netease.arctic.AmsClient;
 import com.netease.arctic.ams.api.TableMeta;
 import com.netease.arctic.io.ArcticFileIO;
 import com.netease.arctic.op.KeyedSchemaUpdate;
+import com.netease.arctic.op.OverwriteBaseFiles;
+import com.netease.arctic.op.RewritePartitions;
 import com.netease.arctic.op.UpdateKeyedTableProperties;
 import com.netease.arctic.scan.BaseKeyedTableScan;
 import com.netease.arctic.scan.KeyedTableScan;
@@ -177,6 +179,45 @@ public class BaseKeyedTable implements KeyedTable {
     return name();
   }
 
+  @Override
+  public Map<String, Long> maxTransactionId() {
+    String s = baseTable.properties().get(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID);
+    if (s != null) {
+      try {
+        Map<String, Long> results = Maps.newHashMap();
+        Map map = new ObjectMapper().readValue(s, Map.class);
+        for (Object key : map.keySet()) {
+          results.put(key.toString(), Long.parseLong(map.get(key).toString()));
+        }
+        return results;
+      } catch (JsonProcessingException e) {
+        throw new UnsupportedOperationException("Failed to get " + TableProperties.BASE_TABLE_MAX_TRANSACTION_ID, e);
+      }
+    } else {
+      return Collections.emptyMap();
+    }
+  }
+
+  @Override
+  public StructLikeMap<Long> partitionMaxTransactionId() {
+    String s = baseTable.properties().get(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID);
+    if (s != null) {
+      return TablePropertyUtil.decodePartitionMaxTxId(spec(), s);
+    } else {
+      return StructLikeMap.create(spec().partitionType());
+    }
+  }
+
+  @Override
+  public OverwriteBaseFiles newOverwriteBaseFiles() {
+    return new OverwriteBaseFiles(this);
+  }
+
+  @Override
+  public RewritePartitions newRewritePartitions() {
+    return new RewritePartitions(this);
+  }
+
   public static class BaseInternalTable extends BaseUnkeyedTable implements BaseTable {
 
     public BaseInternalTable(
@@ -185,34 +226,6 @@ public class BaseKeyedTable implements KeyedTable {
       super(tableIdentifier, baseIcebergTable, arcticFileIO, client);
     }
 
-    @Override
-    public Map<String, Long> maxTransactionId() {
-      String s = properties().get(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID);
-      if (s != null) {
-        try {
-          Map<String, Long> results = Maps.newHashMap();
-          Map map = new ObjectMapper().readValue(s, Map.class);
-          for (Object key : map.keySet()) {
-            results.put(key.toString(), Long.parseLong(map.get(key).toString()));
-          }
-          return results;
-        } catch (JsonProcessingException e) {
-          throw new UnsupportedOperationException("Failed to get " + TableProperties.BASE_TABLE_MAX_TRANSACTION_ID, e);
-        }
-      } else {
-        return Collections.emptyMap();
-      }
-    }
-
-    @Override
-    public StructLikeMap<Long> partitionMaxTransactionId() {
-      String s = properties().get(TableProperties.BASE_TABLE_MAX_TRANSACTION_ID);
-      if (s != null) {
-        return TablePropertyUtil.decodePartitionMaxTxId(spec(), s);
-      } else {
-        return StructLikeMap.create(spec().partitionType());
-      }
-    }
   }
 
   public static class ChangeInternalTable extends BaseUnkeyedTable implements ChangeTable {
diff --git a/core/src/main/java/com/netease/arctic/table/BaseTable.java b/core/src/main/java/com/netease/arctic/table/BaseTable.java
index d0611775a..b6fa6096b 100644
--- a/core/src/main/java/com/netease/arctic/table/BaseTable.java
+++ b/core/src/main/java/com/netease/arctic/table/BaseTable.java
@@ -18,17 +18,10 @@
 
 package com.netease.arctic.table;
 
-import org.apache.iceberg.util.StructLikeMap;
-
-import java.util.Map;
 
 /**
  * Base table store of an {@link KeyedTable}, storing base records in it.
  */
 public interface BaseTable extends UnkeyedTable {
 
-  @Deprecated
-  Map<String, Long> maxTransactionId();
-
-  StructLikeMap<Long> partitionMaxTransactionId();
 }
diff --git a/core/src/main/java/com/netease/arctic/table/KeyedTable.java b/core/src/main/java/com/netease/arctic/table/KeyedTable.java
index 1a2fe12a6..2b7fa7e53 100644
--- a/core/src/main/java/com/netease/arctic/table/KeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/KeyedTable.java
@@ -18,9 +18,13 @@
 
 package com.netease.arctic.table;
 
+import com.netease.arctic.op.OverwriteBaseFiles;
+import com.netease.arctic.op.RewritePartitions;
 import com.netease.arctic.scan.KeyedTableScan;
 import org.apache.iceberg.UpdateProperties;
 import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.util.StructLikeMap;
+import java.util.Map;
 
 /**
  * Represents an arctic table with keys supported, consist of one {@link ChangeTable} and one {@link BaseTable}.
@@ -79,12 +83,28 @@ public interface KeyedTable extends ArcticTable {
 
   /**
    * Allocate a new transaction id from this table
+   *
    * @param signature signature for this request. Requests with the same signature will get the same transaction id.
    *                  Requests with signature NULL will always get different transaction id.
    * @return a new transaction id
    */
   long beginTransaction(String signature);
 
+  /**
+   * get max transactionId of each partition. use {@link #partitionMaxTransactionId()} instead.
+   *
+   * @return map of max transactionId of each partition
+   */
+  @Deprecated
+  Map<String, Long> maxTransactionId();
+
+  /**
+   * get max transactionId of each partition
+   *
+   * @return map of max transactionId of each partition
+   */
+  StructLikeMap<Long> partitionMaxTransactionId();
+
   @Override
   default boolean isKeyedTable() {
     return true;
@@ -94,4 +114,8 @@ public interface KeyedTable extends ArcticTable {
   default KeyedTable asKeyedTable() {
     return this;
   }
+
+  RewritePartitions newRewritePartitions();
+
+  OverwriteBaseFiles newOverwriteBaseFiles();
 }
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 3cc123419..a4ce7c46a 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
@@ -137,7 +137,7 @@ public class ArcticSparkCatalog implements TableCatalog, SupportsNamespaces {
     if (type != null) {
       switch (type) {
         case CHANGE:
-          ChangeTable changeTable = table.asKeyedTable().changeTable();
+          UnkeyedTable changeTable = table.asKeyedTable().changeTable();
           return new ArcticSparkChangeTable(changeTable);
         default:
           throw new IllegalArgumentException("Unknown inner table type: " + type);
diff --git a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkWrite.java b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkWrite.java
index 740d2ffb5..b5cf423b0 100644
--- a/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkWrite.java
+++ b/spark/v3.1/spark/src/main/java/com/netease/arctic/spark/writer/SparkWrite.java
@@ -18,11 +18,9 @@
 
 package com.netease.arctic.spark.writer;
 
-import com.netease.arctic.op.ArcticOperations;
 import com.netease.arctic.op.OverwriteBaseFiles;
 import com.netease.arctic.op.RewritePartitions;
 import com.netease.arctic.table.KeyedTable;
-import com.netease.arctic.table.TableProperties;
 import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.expressions.Expression;
@@ -117,7 +115,7 @@ public class SparkWrite {
   private class DynamicOverwrite extends BaseBatchWrite {
     @Override
     public void commit(WriterCommitMessage[] messages) {
-      RewritePartitions rewritePartitions = ArcticOperations.newRewritePartitions(table);
+      RewritePartitions rewritePartitions = table.newRewritePartitions();
       rewritePartitions.withTransactionId(transactionId);
 
       for (DataFile file : files(messages)) {
@@ -136,7 +134,7 @@ public class SparkWrite {
 
     @Override
     public void commit(WriterCommitMessage[] messages) {
-      OverwriteBaseFiles overwriteBaseFiles = ArcticOperations.newOverwriteBaseFiles(table);
+      OverwriteBaseFiles overwriteBaseFiles = table.newOverwriteBaseFiles();
       overwriteBaseFiles.overwriteByRowFilter(overwriteExpr);
       overwriteBaseFiles.withTransactionId(transactionId);
 


Test patch:
diff --git a/core/src/test/java/com/netease/arctic/op/OverwriteBaseFileTest.java b/core/src/test/java/com/netease/arctic/op/OverwriteBaseFileTest.java
index 86deb9cdc..9fc4457fc 100644
--- a/core/src/test/java/com/netease/arctic/op/OverwriteBaseFileTest.java
+++ b/core/src/test/java/com/netease/arctic/op/OverwriteBaseFileTest.java
@@ -74,7 +74,7 @@ public class OverwriteBaseFileTest extends TableTestBase {
     ));
     this.initTxId = txId;
 
-    RewritePartitions overwrite = ArcticOperations.newRewritePartitions(testKeyedTable);
+    RewritePartitions overwrite = testKeyedTable.newRewritePartitions();
     files.forEach(overwrite::addDataFile);
     overwrite.withTransactionId(txId);
     overwrite.commit();
@@ -87,7 +87,7 @@ public class OverwriteBaseFileTest extends TableTestBase {
     ));
 
     // init. 3 partition with init txId
-    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.baseTable().partitionMaxTransactionId();
+    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.partitionMaxTransactionId();
     Assert.assertEquals(initTxId, partitionMaxTxId.get(
         partitionData(TABLE_SCHEMA, SPEC, "2020-1-1")
     ).longValue());
@@ -118,14 +118,14 @@ public class OverwriteBaseFileTest extends TableTestBase {
         newGenericRecord(TABLE_SCHEMA, 9, "999", "2020-1-1")
     );
     List<DataFile> newFiles = writeBaseNoCommit(testKeyedTable, txId, newRecords);
-    OverwriteBaseFiles overwrite = ArcticOperations.newOverwriteBaseFiles(testKeyedTable);
+    OverwriteBaseFiles overwrite = testKeyedTable.newOverwriteBaseFiles();
     newFiles.forEach(overwrite::addFile);
     overwrite.overwriteByRowFilter(Expressions.alwaysTrue())
         .withTransactionId(txId)
         .commit();
     // overwrite all partition and add new data file
 
-    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.baseTable().partitionMaxTransactionId();
+    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.partitionMaxTransactionId();
     // expect result: all partition with new txId
     Assert.assertEquals(txId, partitionMaxTxId.get(
         partitionData(TABLE_SCHEMA, SPEC, "2020-1-1")
@@ -161,7 +161,7 @@ public class OverwriteBaseFileTest extends TableTestBase {
         newGenericRecord(TABLE_SCHEMA, 9, "999", "2020-1-1")
     );
     List<DataFile> newFiles = writeBaseNoCommit(testKeyedTable, txId, newRecords);
-    OverwriteBaseFiles overwrite = ArcticOperations.newOverwriteBaseFiles(testKeyedTable);
+    OverwriteBaseFiles overwrite = testKeyedTable.newOverwriteBaseFiles();
     newFiles.forEach(overwrite::addFile);
     overwrite.withTransactionId(txId);
     overwrite.overwriteByRowFilter(
@@ -177,7 +177,7 @@ public class OverwriteBaseFileTest extends TableTestBase {
     overwrite.commit();
     // overwrite all partition and add new data file
 
-    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.baseTable().partitionMaxTransactionId();
+    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.partitionMaxTransactionId();
     // expect result: 1,2 partition with new txId, 3 partition use old txId
     Assert.assertEquals(txId, partitionMaxTxId.get(
         partitionData(TABLE_SCHEMA, SPEC, "2020-1-1")
diff --git a/core/src/test/java/com/netease/arctic/op/RewritePartitionsTest.java b/core/src/test/java/com/netease/arctic/op/RewritePartitionsTest.java
index a0663fa0c..86f3ccf9b 100644
--- a/core/src/test/java/com/netease/arctic/op/RewritePartitionsTest.java
+++ b/core/src/test/java/com/netease/arctic/op/RewritePartitionsTest.java
@@ -46,7 +46,7 @@ public class RewritePartitionsTest extends TableTestBase {
     ));
     this.initTxId = txId;
 
-    RewritePartitions overwrite = ArcticOperations.newRewritePartitions(testKeyedTable);
+    RewritePartitions overwrite = testKeyedTable.newRewritePartitions();
     files.forEach(overwrite::addDataFile);
     overwrite.withTransactionId(txId);
     overwrite.commit();
@@ -58,7 +58,7 @@ public class RewritePartitionsTest extends TableTestBase {
     ));
 
     // init. 3 partition with init txId
-    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.baseTable().partitionMaxTransactionId();
+    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.partitionMaxTransactionId();
     Assert.assertEquals(initTxId, partitionMaxTxId.get(
         partitionData(TABLE_SCHEMA, SPEC, quickDate(1))
     ).longValue());
@@ -89,13 +89,13 @@ public class RewritePartitionsTest extends TableTestBase {
         newGenericRecord(TABLE_SCHEMA, 9, "999", quickDate(1))
     );
     List<DataFile> newFiles = writeBaseNoCommit(testKeyedTable, txId, newRecords);
-    RewritePartitions overwrite = ArcticOperations.newRewritePartitions(testKeyedTable);
+    RewritePartitions overwrite = testKeyedTable.newRewritePartitions();
     newFiles.forEach(overwrite::addDataFile);
     overwrite.withTransactionId(txId);
     overwrite.commit();
     // overwrite 1 partition by data file
 
-    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.baseTable().partitionMaxTransactionId();
+    StructLikeMap<Long> partitionMaxTxId = testKeyedTable.partitionMaxTransactionId();
     // expect result: 1 partition with new txId, 2,3 partition use old txId
     Assert.assertEquals(txId, partitionMaxTxId.get(
         partitionData(TABLE_SCHEMA, SPEC, quickDate(1))