Repository: NetEase/arctic
Pull request #47 (2022-07-22 05:42:43)

Optimize module core


Patch:
diff --git a/README.md b/README.md
index c539e42ff..6ca0bb331 100644
--- a/README.md
+++ b/README.md
@@ -42,10 +42,10 @@ Arctic contains modules as below:
 
 Arctic is built using Maven with Java 1.8 and Java 11(only for `trino` module).
 
-* To build Trino module need config toolchains.xml in ${user.home}/.m2/ dir, the content is
+* To build Trino module need config `toolchains.xml` in `${user.home}/.m2/` dir, the content is
 ```
+<?xml version="1.0" encoding="UTF-8"?>
 <toolchains>
-<!--  JDK toolchains  -->
     <toolchain>
         <type>jdk</type>
         <provides>
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 5c67883cb..fbcfe4610 100644
--- a/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java
+++ b/core/src/main/java/com/netease/arctic/catalog/BaseArcticCatalog.java
@@ -29,7 +29,6 @@ import com.netease.arctic.ams.api.properties.MetaTableProperties;
 import com.netease.arctic.io.ArcticFileIO;
 import com.netease.arctic.io.ArcticHadoopFileIO;
 import com.netease.arctic.op.ArcticHadoopTableOperations;
-import com.netease.arctic.op.ArcticTableOperations;
 import com.netease.arctic.table.ArcticTable;
 import com.netease.arctic.table.BaseKeyedTable;
 import com.netease.arctic.table.BaseTable;
diff --git a/core/src/main/java/com/netease/arctic/catalog/CatalogLoader.java b/core/src/main/java/com/netease/arctic/catalog/CatalogLoader.java
index 3c7c1a7d8..d629ca7a2 100644
--- a/core/src/main/java/com/netease/arctic/catalog/CatalogLoader.java
+++ b/core/src/main/java/com/netease/arctic/catalog/CatalogLoader.java
@@ -35,7 +35,7 @@ import java.util.stream.Collectors;
 import static com.netease.arctic.ams.api.properties.CatalogMetaProperties.CATALOG_TYPE_HADOOP;
 
 /**
- * Catalogs,  create catalog from arctic metastore thrift url.
+ * Catalogs, create catalog from arctic metastore thrift url.
  */
 public class CatalogLoader {
 
diff --git a/core/src/main/java/com/netease/arctic/data/ChangedLsn.java b/core/src/main/java/com/netease/arctic/data/ChangedLsn.java
index 490499666..e58616647 100644
--- a/core/src/main/java/com/netease/arctic/data/ChangedLsn.java
+++ b/core/src/main/java/com/netease/arctic/data/ChangedLsn.java
@@ -18,14 +18,10 @@
 
 package com.netease.arctic.data;
 
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-
 import java.io.Serializable;
-import java.nio.ByteBuffer;
 import java.util.Objects;
 
 /**
- * @TODO rename to GlobalRowSequence
  * Global row sequence number.
  * <p>
  * Consist of two parts:
@@ -37,32 +33,13 @@ import java.util.Objects;
 
 public class ChangedLsn implements Comparable<ChangedLsn>, Serializable {
 
-  public static final int RECORD_LSN_LENGTH = Long.BYTES * 2;
-  public static final long minimumLsn = 0L;
-
   private final long transactionId;
-  //TODO rename to rowSequence
   private final long fileOffset;
 
   public static ChangedLsn of(long transactionId, long fileOffset) {
     return new ChangedLsn(transactionId, fileOffset);
   }
 
-  public static ChangedLsn from(ByteBuffer byteBuffer) {
-    Preconditions.checkArgument(byteBuffer.remaining() == RECORD_LSN_LENGTH);
-    ChangedLsn recordLsn = ChangedLsn.of(byteBuffer.getLong(), byteBuffer.getLong());
-    byteBuffer.rewind();
-    return recordLsn;
-  }
-
-  /**
-   * TODO deleted
-   * @return minimum lsn, like base file data lsn
-   */
-  public static ChangedLsn minimum() {
-    return new ChangedLsn(minimumLsn, minimumLsn);
-  }
-
   private ChangedLsn(long transactionId, long fileOffset) {
     this.transactionId = transactionId;
     this.fileOffset = fileOffset;
@@ -76,9 +53,6 @@ public class ChangedLsn implements Comparable<ChangedLsn>, Serializable {
     return fileOffset;
   }
 
-  public ByteBuffer byteBuffer() {
-    return ByteBuffer.allocate(RECORD_LSN_LENGTH).putLong(transactionId).putLong(fileOffset);
-  }
 
   @Override
   public int compareTo(ChangedLsn another) {
diff --git a/core/src/main/java/com/netease/arctic/data/DataTreeNode.java b/core/src/main/java/com/netease/arctic/data/DataTreeNode.java
index 46aa9f213..6f36356ca 100644
--- a/core/src/main/java/com/netease/arctic/data/DataTreeNode.java
+++ b/core/src/main/java/com/netease/arctic/data/DataTreeNode.java
@@ -18,6 +18,7 @@
 
 package com.netease.arctic.data;
 
+import com.netease.arctic.ams.api.TreeNode;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.jetbrains.annotations.NotNull;
 
@@ -104,10 +105,6 @@ public final class DataTreeNode implements Serializable {
     return DataTreeNode.of(parentMask, parentIndex);
   }
 
-  public static DataTreeNode fromFile(PrimaryKeyedFile file) {
-    return of(file.mask(), file.index());
-  }
-
   @Override
   public boolean equals(Object o) {
     if (this == o) return true;
@@ -138,4 +135,8 @@ public final class DataTreeNode implements Serializable {
   public long getId() {
     return mask + 1 + index;
   }
+
+  public TreeNode toAmsTreeNode() {
+    return new TreeNode(mask, index);
+  }
 }
diff --git a/core/src/main/java/com/netease/arctic/data/DefaultKeyedFile.java b/core/src/main/java/com/netease/arctic/data/DefaultKeyedFile.java
index f3002ff0c..1e9143120 100644
--- a/core/src/main/java/com/netease/arctic/data/DefaultKeyedFile.java
+++ b/core/src/main/java/com/netease/arctic/data/DefaultKeyedFile.java
@@ -191,19 +191,11 @@ public class DefaultKeyedFile implements PrimaryKeyedFile, Serializable {
   }
 
   @Override
-  public long mask() {
+  public DataTreeNode node() {
     if (meta == null) {
       parse();
     }
-    return meta.node().mask();
-  }
-
-  @Override
-  public long index() {
-    if (meta == null) {
-      parse();
-    }
-    return meta.node().index();
+    return meta.node();
   }
 
   @Override
diff --git a/core/src/main/java/com/netease/arctic/data/PrimaryKeyedFile.java b/core/src/main/java/com/netease/arctic/data/PrimaryKeyedFile.java
index 5b0c0f7cc..5811a146d 100644
--- a/core/src/main/java/com/netease/arctic/data/PrimaryKeyedFile.java
+++ b/core/src/main/java/com/netease/arctic/data/PrimaryKeyedFile.java
@@ -30,10 +30,10 @@ public interface PrimaryKeyedFile extends DataFile {
    */
   DataFileType type();
 
-  //TODO change to DataTreeNode
-  long mask();
-
-  long index();
+  /**
+   * Returns the {@link DataTreeNode} file belong
+   */
+  DataTreeNode node();
 
   /**
    * Returns the id of transaction which file added into table
diff --git a/core/src/main/java/com/netease/arctic/io/ArcticFileIO.java b/core/src/main/java/com/netease/arctic/io/ArcticFileIO.java
index 4dbb7935b..75ae7c599 100644
--- a/core/src/main/java/com/netease/arctic/io/ArcticFileIO.java
+++ b/core/src/main/java/com/netease/arctic/io/ArcticFileIO.java
@@ -46,20 +46,17 @@ public interface ArcticFileIO extends FileIO {
    */
   boolean exists(String path);
 
-  //TODO wait to delete
-  boolean rename(String src, String dts);
-
-  //TODO wait to delete
-  boolean mkdirs(String src);
-
-  //TODO wait to delete
-  boolean deleteFileWithResult(String path);
-
-  //TODO wait to delete
+  /** Delete a file.
+   *
+   * @param path the path to delete.
+   * @param recursive if path is a directory and set to
+   * true, the directory is deleted else throws an exception. In
+   * case of a file the recursive can be set to either true or false.
+   * @return  true if delete is successful else false.
+   */
   boolean deleteFileWithResult(String path, boolean recursive);
 
   //TODO FileStatus is a hadoop object, need to be replaced
-
   List<FileStatus> list(String location);
 
   /**
diff --git a/core/src/main/java/com/netease/arctic/io/ArcticHadoopFileIO.java b/core/src/main/java/com/netease/arctic/io/ArcticHadoopFileIO.java
index 384dc2740..21c94d98f 100644
--- a/core/src/main/java/com/netease/arctic/io/ArcticHadoopFileIO.java
+++ b/core/src/main/java/com/netease/arctic/io/ArcticHadoopFileIO.java
@@ -70,27 +70,12 @@ public class ArcticHadoopFileIO extends HadoopFileIO implements ArcticFileIO {
       try {
         fs.delete(toDelete, false);
       } catch (IOException e) {
-        throw new UncheckedIOException("Failed to delete file: " + new Object[]{path}, e);
+        throw new UncheckedIOException("Failed to delete file: " + path, e);
       }
       return null;
     });
   }
 
-  @Override
-  public boolean deleteFileWithResult(String path) {
-    return tableMetaStore.doAs(() -> {
-      Path toDelete = new Path(path);
-      FileSystem fs = getFs(toDelete);
-      boolean result;
-      try {
-        result = fs.delete(toDelete, false /* not recursive */);
-      } catch (IOException e) {
-        result = false;
-      }
-      return result;
-    });
-  }
-
   @Override
   public boolean deleteFileWithResult(String path, boolean recursive) {
     return tableMetaStore.doAs(() -> {
@@ -195,34 +180,6 @@ public class ArcticHadoopFileIO extends HadoopFileIO implements ArcticFileIO {
     });
   }
 
-  @Override
-  public boolean rename(String src, String dts) {
-    return tableMetaStore.doAs(() -> {
-      Path srcPath = new Path(src);
-      Path dtsPath = new Path(dts);
-      FileSystem fs = getFs(srcPath);
-      try {
-        return fs.rename(srcPath, dtsPath);
-      } catch (IOException e) {
-        throw new UncheckedIOException("Failed to rename: from " + src + " to " + dts, e);
-      }
-    });
-  }
-
-  @Override
-  public boolean mkdirs(String src) {
-    return tableMetaStore.doAs(() -> {
-      Path path = new Path(src);
-      FileSystem fs = getFs(path);
-      try {
-        return fs.mkdirs(path);
-      } catch (IOException e) {
-        LOG.error("Failed to mkdirs: path {}, exception {}", path, e);
-        throw new UncheckedIOException("Failed to mkdirs: path " + src, e);
-      }
-    });
-  }
-
   private FileSystem getFs(Path path) {
     return Util.getFs(path, conf());
   }
diff --git a/core/src/main/java/com/netease/arctic/io/writer/ChangeTaskWriter.java b/core/src/main/java/com/netease/arctic/io/writer/ChangeTaskWriter.java
index a1a679fd0..c091b0b2e 100644
--- a/core/src/main/java/com/netease/arctic/io/writer/ChangeTaskWriter.java
+++ b/core/src/main/java/com/netease/arctic/io/writer/ChangeTaskWriter.java
@@ -36,7 +36,6 @@ import java.io.IOException;
  */
 public abstract class ChangeTaskWriter<T> extends BaseTaskWriter<T> {
 
-  //TODO rename to rowSequence
   private long fileOffset = 0L;
 
   protected ChangeTaskWriter(FileFormat format, FileAppenderFactory<T> appenderFactory,
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 2df45f90e..c1aca3916 100644
--- a/core/src/main/java/com/netease/arctic/scan/BaseKeyedTableScan.java
+++ b/core/src/main/java/com/netease/arctic/scan/BaseKeyedTableScan.java
@@ -18,7 +18,6 @@
 
 package com.netease.arctic.scan;
 
-import com.netease.arctic.data.DataFileType;
 import com.netease.arctic.data.DataTreeNode;
 import com.netease.arctic.table.BaseKeyedTable;
 import com.netease.arctic.table.TableProperties;
@@ -193,7 +192,7 @@ public class BaseKeyedTableScan implements KeyedTableScan {
     keyedTableTasks.stream().forEach(task -> {
       if (!pathSets.contains(task.file().path().toString())) {
         pathSets.add(task.file().path().toString());
-        DataTreeNode treeNode = DataTreeNode.fromFile(task.file());
+        DataTreeNode treeNode = task.file().node();
         NodeFileScanTask nodeFileScanTask = nodeFileScanTaskMap.getOrDefault(treeNode, new NodeFileScanTask(treeNode));
         nodeFileScanTask.addFile(task);
         nodeFileScanTaskMap.put(treeNode, nodeFileScanTask);
@@ -208,7 +207,7 @@ public class BaseKeyedTableScan implements KeyedTableScan {
       nodeFileScanTaskMap.forEach((treeNode1, nodeFileScanTask1) -> {
         if (!treeNode1.equals(treeNode) && (treeNode1.isSonOf(treeNode) || treeNode.isSonOf(treeNode1))) {
           List<ArcticFileScanTask> deletes = nodeFileScanTask1.arcticEquityDeletes().stream()
-              .filter(file -> file.file().mask() == treeNode1.mask() && file.file().index() == treeNode1.index())
+              .filter(file -> file.file().node().equals(treeNode1))
               .collect(Collectors.toList());
 
           nodeFileScanTask.addTasks(deletes);
diff --git a/core/src/main/java/com/netease/arctic/scan/NodeFileScanTask.java b/core/src/main/java/com/netease/arctic/scan/NodeFileScanTask.java
index 169e726da..9e6e8c27e 100644
--- a/core/src/main/java/com/netease/arctic/scan/NodeFileScanTask.java
+++ b/core/src/main/java/com/netease/arctic/scan/NodeFileScanTask.java
@@ -20,7 +20,6 @@ package com.netease.arctic.scan;
 
 import com.netease.arctic.data.DataFileType;
 import com.netease.arctic.data.DataTreeNode;
-import com.netease.arctic.data.PrimaryKeyedFile;
 import com.netease.arctic.table.TableProperties;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -128,40 +127,6 @@ public class NodeFileScanTask implements KeyedTableScanTask {
     }
   }
 
-  //fixme 为了optimize不报错，后面要删除此方法的
-  public void addFile(PrimaryKeyedFile file) {
-
-    DataFileType fileType = file.type();
-    if (fileType == null) {
-      LOG.warn("file type is null");
-      return;
-    }
-    cost = cost + Math.max(file.fileSizeInBytes(), openFileCost);
-    rowNums = rowNums + file.recordCount();
-
-    //fixme 为了optimize不报错，后面要删除此方法的
-    BaseArcticFileScanTask task = new BaseArcticFileScanTask(file, null, null);
-    switch (fileType) {
-      case BASE_FILE:
-        baseTasks.add(task);
-        break;
-      case INSERT_FILE:
-        insertTasks.add(task);
-        break;
-      case EQ_DELETE_FILE:
-        deleteFiles.add(task);
-        break;
-      default:
-        LOG.warn("file type is {}, not add in node", fileType);
-        // ignore the object
-    }
-  }
-
-  //fixme 为了optimize不报错，后面要删除此方法的
-  public void addFiles(List<PrimaryKeyedFile> files) {
-    files.forEach(this::addFile);
-  }
-
   public void addTasks(List<ArcticFileScanTask> files) {
     files.forEach(this::addFile);
   }
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 7ef1cf888..7a1ca828d 100644
--- a/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/BaseKeyedTable.java
@@ -28,7 +28,6 @@ import com.netease.arctic.op.UpdateKeyedTableProperties;
 import com.netease.arctic.scan.BaseKeyedTableScan;
 import com.netease.arctic.scan.KeyedTableScan;
 import com.netease.arctic.trace.AmsTableTracer;
-import com.netease.arctic.trace.TableTracer;
 import com.netease.arctic.trace.TracedUpdateProperties;
 import com.netease.arctic.trace.TrackerOperations;
 import com.netease.arctic.utils.TablePropertyUtil;
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/PartitionAndNodeGroup.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/PartitionAndNodeGroup.java
index acedc4f10..deaabf416 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/PartitionAndNodeGroup.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/PartitionAndNodeGroup.java
@@ -85,7 +85,7 @@ public class PartitionAndNodeGroup {
 
     tasks.forEach(task -> {
       String partitionKey = task.file().partition().toString();
-      Long index = task.file().index();
+      Long index = task.file().node().index();
       Map<Long, Node> indexNodes = nodes.getOrDefault(partitionKey, new HashMap<>());
       Node node = indexNodes.getOrDefault(index, new Node());
       if (insert) {
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/assigner/ShuffleSplitAssigner.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/assigner/ShuffleSplitAssigner.java
index ad2603dcc..fe9c6fd79 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/assigner/ShuffleSplitAssigner.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/assigner/ShuffleSplitAssigner.java
@@ -139,7 +139,7 @@ public class ShuffleSplitAssigner implements SplitAssigner {
 
     int subtaskId = partitionIndexSubtaskMap.computeIfAbsent(
         partitionIndexKey, key -> (partitionIndexSubtaskMap.size() + 1) % totalParallelism);
-    LOG.info("partition = {}, index = {}, subtaskId = {}", file.partition().toString(), file.index(), subtaskId);
+    LOG.info("partition = {}, index = {}, subtaskId = {}", file.partition().toString(), file.node().index(), subtaskId);
     return subtaskId;
   }
 
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java
index 7d6ae2520..1855f49b0 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ArcticSplit.java
@@ -97,8 +97,8 @@ public abstract class ArcticSplit implements SourceSplit, Serializable, Comparab
             MoreObjects.toStringHelper(primaryKeyedFile)
                 .add("file", primaryKeyedFile.path().toString())
                 .add("type", primaryKeyedFile.type().shortName())
-                .add("mask", primaryKeyedFile.mask())
-                .add("index", primaryKeyedFile.index())
+                .add("mask", primaryKeyedFile.node().mask())
+                .add("index", primaryKeyedFile.node().index())
                 .add("transactionId", primaryKeyedFile.transactionId())
                 .toString()).collect(Collectors.toList()));
   }
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java
index 69bf22880..3df8a1f33 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/ChangelogSplit.java
@@ -51,7 +51,7 @@ public class ChangelogSplit extends ArcticSplit {
     this.deleteScanTasks = deleteScanTasks;
     Optional<ArcticFileScanTask> task = insertScanTasks.stream().findFirst();
     PrimaryKeyedFile file = task.isPresent() ? task.get().file() : deleteScanTasks.stream().findFirst().get().file();
-    this.dataTreeNode = DataTreeNode.of(file.mask(), file.index());
+    this.dataTreeNode = file.node();
   }
 
   @Override
diff --git a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java
index 5f35f25e6..f0a8e5947 100644
--- a/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java
+++ b/flink/v1.12/flink/src/main/java/com/netease/arctic/flink/read/hybrid/split/SnapshotSplit.java
@@ -44,7 +44,7 @@ public class SnapshotSplit extends ArcticSplit {
     this.insertScanTasks = insertScanTasks;
     this.taskIndex = taskIndex;
     PrimaryKeyedFile file = insertScanTasks.stream().findFirst().get().file();
-    this.dataTreeNode = DataTreeNode.of(file.mask(), file.index());
+    this.dataTreeNode = file.node();
   }
 
   @Override
diff --git a/optimizer/src/main/java/com/netease/arctic/optimizer/operator/executor/MajorExecutor.java b/optimizer/src/main/java/com/netease/arctic/optimizer/operator/executor/MajorExecutor.java
index c85a7bcac..1e5c4740e 100644
--- a/optimizer/src/main/java/com/netease/arctic/optimizer/operator/executor/MajorExecutor.java
+++ b/optimizer/src/main/java/com/netease/arctic/optimizer/operator/executor/MajorExecutor.java
@@ -206,7 +206,7 @@ public class MajorExecutor extends BaseExecutor<DataFile> {
             return new BaseArcticFileScanTask(defaultKeyedFile, null, table.spec());
           } else {
             return new BaseArcticFileScanTask(defaultKeyedFile,
-                deleteFileMap.get(DataTreeNode.of(defaultKeyedFile.mask(), defaultKeyedFile.index())), table.spec());
+                deleteFileMap.get(defaultKeyedFile.node()), table.spec());
           }
         })
         .collect(Collectors.toList());
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 e43e7ef31..740d2ffb5 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
@@ -96,7 +96,7 @@ public class SparkWrite {
               2.0 /* exponential */)
           .throwFailureWhenFinished()
           .run(file -> {
-            table.io().deleteFileWithResult(file.path().toString());
+            table.io().deleteFile(file.path().toString());
           });
     }
   }


Test patch:
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestMajorOptimizeCommit.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestMajorOptimizeCommit.java
index 43146132e..d5c3b2c30 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestMajorOptimizeCommit.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/optimize/TestMajorOptimizeCommit.java
@@ -101,7 +101,7 @@ public class TestMajorOptimizeCommit extends TestMajorOptimizePlan {
     List<DataFile> dataFiles = insertKeyedTableBaseDataFiles(3);
     return dataFiles.stream().collect(Collectors.groupingBy(dataFile ->  {
       DefaultKeyedFile keyedFile = new DefaultKeyedFile(dataFile);
-      return new TreeNode(keyedFile.mask(), keyedFile.index());
+      return keyedFile.node().toAmsTreeNode();
     }));
   }
 }
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/util/DataFileInfoUtils.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/util/DataFileInfoUtils.java
index 7db5eef89..77423a4e8 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/util/DataFileInfoUtils.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/util/DataFileInfoUtils.java
@@ -47,8 +47,8 @@ public class DataFileInfoUtils {
     if (arcticTable.isKeyedTable()) {
       PrimaryKeyedFile keyedTableFile = new DefaultKeyedFile(dataFile);
       dataFileInfo.setType(keyedTableFile.type().name());
-      dataFileInfo.setIndex(keyedTableFile.index());
-      dataFileInfo.setMask(keyedTableFile.mask());
+      dataFileInfo.setIndex(keyedTableFile.node().index());
+      dataFileInfo.setMask(keyedTableFile.node().mask());
     } else {
       dataFileInfo.setType(DataFileType.BASE_FILE.name());
       dataFileInfo.setIndex(0);
diff --git a/core/src/test/java/com/netease/arctic/data/KeyedDataFileTest.java b/core/src/test/java/com/netease/arctic/data/KeyedDataFileTest.java
index d568a252f..0b1cf91ba 100644
--- a/core/src/test/java/com/netease/arctic/data/KeyedDataFileTest.java
+++ b/core/src/test/java/com/netease/arctic/data/KeyedDataFileTest.java
@@ -39,9 +39,9 @@ public class KeyedDataFileTest extends TableTestBase {
     Assert.assertEquals(1, writeFiles.size());
     DefaultKeyedFile defaultKeyedFile = new DefaultKeyedFile(writeFiles.get(0));
     Assert.assertEquals(DataFileType.INSERT_FILE, defaultKeyedFile.type());
-    Assert.assertEquals(3, defaultKeyedFile.mask());
+    Assert.assertEquals(3, defaultKeyedFile.node().mask());
     Long txId = AMS.handler().getTableCurrentTxId(PK_TABLE_ID.buildTableIdentifier());
-    Assert.assertEquals(0, defaultKeyedFile.index());
+    Assert.assertEquals(0, defaultKeyedFile.node().index());
     Assert.assertEquals(txId, defaultKeyedFile.transactionId());
     Assert.assertEquals(ChangedLsn.of(txId, 1), defaultKeyedFile.minLsn());
     Assert.assertEquals(ChangedLsn.of(txId,2), defaultKeyedFile.maxLsn());
diff --git a/core/src/test/java/com/netease/arctic/io/TableTestBaseWithInitData.java b/core/src/test/java/com/netease/arctic/io/TableTestBaseWithInitData.java
index 573b9834e..3c436b74e 100644
--- a/core/src/test/java/com/netease/arctic/io/TableTestBaseWithInitData.java
+++ b/core/src/test/java/com/netease/arctic/io/TableTestBaseWithInitData.java
@@ -25,15 +25,9 @@ import com.netease.arctic.io.writer.GenericBaseTaskWriter;
 import com.netease.arctic.io.writer.GenericChangeTaskWriter;
 import com.netease.arctic.io.writer.GenericTaskWriters;
 import com.netease.arctic.io.writer.SortedPosDeleteWriter;
-import java.io.IOException;
-import java.time.LocalDateTime;
-import java.util.Arrays;
-import java.util.List;
 import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.MetadataColumns;
-import org.apache.iceberg.RowDelta;
 import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.io.WriteResult;
@@ -41,6 +35,11 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.junit.Before;
 
+import java.io.IOException;
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.List;
+
 public class TableTestBaseWithInitData extends TableTestBase {
 
   public List<Record> baseRecords() {
diff --git a/core/src/test/java/com/netease/arctic/io/TaskReaderTest.java b/core/src/test/java/com/netease/arctic/io/TaskReaderTest.java
index 45bc8f9c3..f726a307e 100644
--- a/core/src/test/java/com/netease/arctic/io/TaskReaderTest.java
+++ b/core/src/test/java/com/netease/arctic/io/TaskReaderTest.java
@@ -19,31 +19,25 @@
 package com.netease.arctic.io;
 
 import com.google.common.collect.Sets;
-import com.netease.arctic.TableTestBase;
-import com.netease.arctic.data.ChangeAction;
 import com.netease.arctic.io.reader.BaseIcebergPosDeleteReader;
 import com.netease.arctic.io.reader.GenericArcticDataReader;
 import com.netease.arctic.scan.CombinedScanTask;
 import com.netease.arctic.scan.KeyedTableScanTask;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Set;
-import java.util.stream.Collectors;
 import org.apache.iceberg.MetadataColumns;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.IdentityPartitionConverters;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.io.CloseableIterable;
-import org.apache.iceberg.io.CloseableIterator;
-import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.junit.Assert;
-import org.junit.Before;
 import org.junit.Test;
 
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
 public class TaskReaderTest extends TableTestBaseWithInitData {
 
   @Test
diff --git a/core/src/test/java/com/netease/arctic/io/TaskWriterTest.java b/core/src/test/java/com/netease/arctic/io/TaskWriterTest.java
index afc879ad8..8af83f8e6 100644
--- a/core/src/test/java/com/netease/arctic/io/TaskWriterTest.java
+++ b/core/src/test/java/com/netease/arctic/io/TaskWriterTest.java
@@ -26,19 +26,16 @@ import com.netease.arctic.io.writer.SortedPosDeleteWriter;
 import com.netease.arctic.utils.ManifestEntryFields;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.ManifestFile;
 import org.apache.iceberg.MetadataColumns;
 import org.apache.iceberg.RowDelta;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.IcebergGenerics;
 import org.apache.iceberg.data.Record;
-import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.hadoop.HadoopTables;
 import org.apache.iceberg.io.WriteResult;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
-import org.checkerframework.checker.index.qual.LowerBoundUnknown;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -47,7 +44,6 @@ import java.nio.ByteBuffer;
 import java.time.LocalDateTime;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 
 public class TaskWriterTest extends TableTestBase {
 
diff --git a/core/src/test/java/com/netease/arctic/trace/TableTracerTest.java b/core/src/test/java/com/netease/arctic/trace/TableTracerTest.java
index 7c327edfd..e239df0d5 100644
--- a/core/src/test/java/com/netease/arctic/trace/TableTracerTest.java
+++ b/core/src/test/java/com/netease/arctic/trace/TableTracerTest.java
@@ -40,7 +40,6 @@ import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.junit.Assert;
 import org.junit.Test;
 
-import java.io.File;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
diff --git a/optimizer/src/test/java/com/netease/arctic/optimizer/util/DataFileInfoUtils.java b/optimizer/src/test/java/com/netease/arctic/optimizer/util/DataFileInfoUtils.java
index 45abf1c0b..2137aa275 100644
--- a/optimizer/src/test/java/com/netease/arctic/optimizer/util/DataFileInfoUtils.java
+++ b/optimizer/src/test/java/com/netease/arctic/optimizer/util/DataFileInfoUtils.java
@@ -47,8 +47,8 @@ public class DataFileInfoUtils {
     if (arcticTable.isKeyedTable()) {
       PrimaryKeyedFile keyedTableFile = new DefaultKeyedFile(dataFile);
       dataFileInfo.setType(keyedTableFile.type().name());
-      dataFileInfo.setIndex(keyedTableFile.index());
-      dataFileInfo.setMask(keyedTableFile.mask());
+      dataFileInfo.setIndex(keyedTableFile.node().index());
+      dataFileInfo.setMask(keyedTableFile.node().mask());
     } else {
       dataFileInfo.setType(DataFileType.BASE_FILE.name());
       dataFileInfo.setIndex(0);