Repository: NetEase/arctic
Pull request #3 (2022-07-19 03:35:19)

Add JAVA_11_HOME to ReadMe


Patch:
diff --git a/README.md b/README.md
index 622058795..c539e42ff 100644
--- a/README.md
+++ b/README.md
@@ -42,8 +42,24 @@ Arctic contains modules as below:
 
 Arctic is built using Maven with Java 1.8 and Java 11(only for `trino` module).
 
-* To invoke a build and run tests: `mvn package`
-* To skip tests: `mvn -DskipTests package`
+* To build Trino module need config toolchains.xml in ${user.home}/.m2/ dir, the content is
+```
+<toolchains>
+<!--  JDK toolchains  -->
+    <toolchain>
+        <type>jdk</type>
+        <provides>
+            <version>11</version>
+            <vendor>sun</vendor>
+        </provides>
+        <configuration>
+            <jdkHome>${yourJdk11Home}</jdkHome>
+        </configuration>
+    </toolchain>
+</toolchains>
+```
+* To invoke a build and run tests: `mvn package -P toolchain`
+* To skip tests: `mvn -DskipTests package -P toolchain`
 
 ## Engines supported
 
diff --git a/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java b/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java
index 827137220..0bd8f84e2 100644
--- a/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java
+++ b/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java
@@ -39,6 +39,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -58,7 +59,7 @@ public class AmsTableTracer implements TableTracer {
   private String action;
   private Map<String, String> properties;
   private InternalTableChange defaultTableChange;
-  private List<TableChange> transactionTableChanges = Lists.newArrayList();
+  private final Map<Long, AmsTableTracer.InternalTableChange> transactionSnapshotTableChanges = new LinkedHashMap<>();
 
   public AmsTableTracer(UnkeyedTable table, String action, AmsClient client) {
     this.innerTable = table instanceof ChangeTable ?
@@ -105,8 +106,8 @@ public class AmsTableTracer implements TableTracer {
     return defaultTableChange;
   }
 
-  public void addTransactionTableChange(TableChange tableChange) {
-    transactionTableChanges.add(tableChange);
+  public void addTransactionTableSnapshot(Long snapshotId, AmsTableTracer.InternalTableChange internalTableChange) {
+    transactionSnapshotTableChanges.put(snapshotId, internalTableChange);
   }
 
   public ArcticTable table() {
@@ -127,21 +128,28 @@ public class AmsTableTracer implements TableTracer {
     boolean threw = false;
 
     if (defaultTableChange != null) {
-      Table traceTable = null;
+      Table traceTable;
       if (table.isUnkeyedTable()) {
         traceTable = table.asUnkeyedTable();
       } else {
         throw new IllegalStateException("can't apply table change on keyed table.");
       }
 
-      Optional<TableChange> tableChange = defaultTableChange.toTableChange(table, traceTable, innerTable);
+      Optional<TableChange> tableChange =
+          defaultTableChange.toTableChange(table, traceTable.currentSnapshot(), innerTable);
       if (tableChange.isPresent()) {
         commitMeta.addToChanges(tableChange.get());
         update = true;
       }
     }
-    if (transactionTableChanges.size() > 0) {
-      transactionTableChanges.forEach(commitMeta::addToChanges);
+    if (transactionSnapshotTableChanges.size() > 0) {
+      transactionSnapshotTableChanges.forEach((snapshotId, internalTableChange) -> {
+        if (table.isUnkeyedTable()) {
+          Snapshot snapshot = table.asUnkeyedTable().snapshot(snapshotId);
+          Optional<TableChange> tableChange = internalTableChange.toTableChange(table, snapshot, innerTable);
+          tableChange.ifPresent(commitMeta::addToChanges);
+        }
+      });
       update = true;
     }
     if (this.properties != null) {
@@ -201,17 +209,16 @@ public class AmsTableTracer implements TableTracer {
      * Build {@link TableChange} to report to ams.
      *
      * @param arcticTable arctic table which table change belongs
-     * @param traceTable  iceberg table produce the snapshot
+     * @param snapshot  the snapshot produced in this operation
      * @param innerTable  inner table name
      * @return table change
      */
-    public Optional<TableChange> toTableChange(ArcticTable arcticTable, Table traceTable, String innerTable) {
+    public Optional<TableChange> toTableChange(ArcticTable arcticTable, Snapshot snapshot, String innerTable) {
       if (addedFiles.size() > 0 || deletedFiles.size() > 0 || addedDeleteFiles.size() > 0 ||
           deletedDeleteFiles.size() > 0) {
-        long currentSnapshotId = traceTable.currentSnapshot().snapshotId();
+        long currentSnapshotId = snapshot.snapshotId();
         long parentSnapshotId =
-            traceTable.currentSnapshot().parentId() == null ? -1 : traceTable.currentSnapshot().parentId();
-        Snapshot snapshot = traceTable.currentSnapshot();
+            snapshot.parentId() == null ? -1 : snapshot.parentId();
         Map<String, String> summary = snapshot.summary();
         long realAddedDataFiles = summary.get(SnapshotSummary.ADDED_FILES_PROP) == null ?
             0 : Long.parseLong(summary.get(SnapshotSummary.ADDED_FILES_PROP));
diff --git a/core/src/main/java/com/netease/arctic/trace/TracedTransaction.java b/core/src/main/java/com/netease/arctic/trace/TracedTransaction.java
index beb3cc7b3..ac08b6b36 100644
--- a/core/src/main/java/com/netease/arctic/trace/TracedTransaction.java
+++ b/core/src/main/java/com/netease/arctic/trace/TracedTransaction.java
@@ -18,7 +18,6 @@
 
 package com.netease.arctic.trace;
 
-import com.netease.arctic.ams.api.TableChange;
 import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataOperations;
@@ -54,7 +53,6 @@ import org.apache.iceberg.io.LocationProvider;
 import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 
 /**
  * Wrap {@link Transaction} with {@link TableTracer}.
@@ -147,7 +145,8 @@ public class TracedTransaction implements Transaction {
 
   @Override
   public DeleteFiles newDelete() {
-    return transaction.newDelete();
+    tracer.setAction(DataOperations.DELETE);
+    return new TracedDeleteFiles(transaction.newDelete(), new TransactionTracker());
   }
 
   @Override
@@ -191,9 +190,7 @@ public class TracedTransaction implements Transaction {
 
     @Override
     public void commit() {
-      Optional<TableChange> tableChange =
-          internalTableChange.toTableChange(tracer.table(), transaction.table(), tracer.innerTable());
-      tableChange.ifPresent(tracer::addTransactionTableChange);
+      tracer.addTransactionTableSnapshot(transaction.table().currentSnapshot().snapshotId(), internalTableChange);
     }
 
     @Override
diff --git a/core/src/main/java/com/netease/arctic/utils/SnapshotFileUtil.java b/core/src/main/java/com/netease/arctic/utils/SnapshotFileUtil.java
index 18280d55b..e304dd89c 100644
--- a/core/src/main/java/com/netease/arctic/utils/SnapshotFileUtil.java
+++ b/core/src/main/java/com/netease/arctic/utils/SnapshotFileUtil.java
@@ -59,6 +59,7 @@ public class SnapshotFileUtil {
     HadoopTables tables = new HadoopTables();
     Table entriesTable = tables.load(table.location() + "#ENTRIES");
     IcebergGenerics.read(entriesTable)
+        .useSnapshot(snapshot.snapshotId())
         .where(Expressions.equal(ManifestEntryFields.SNAPSHOT_ID.name(), snapshot.snapshotId()))
         .build()
         .forEach(record -> {
diff --git a/trino/pom.xml b/trino/pom.xml
index 6ad7637aa..9bb4a982c 100644
--- a/trino/pom.xml
+++ b/trino/pom.xml
@@ -518,4 +518,34 @@
             </plugin>
         </plugins>
     </build>
+
+    <profiles>
+        <profile>
+            <id>toolchain</id>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-toolchains-plugin</artifactId>
+                        <version>3.1.0</version>
+                        <executions>
+                            <execution>
+                                <goals>
+                                    <goal>toolchain</goal>
+                                </goals>
+                            </execution>
+                        </executions>
+                        <configuration>
+                            <toolchains>
+                                <jdk>
+                                    <version>11</version>
+                                    <vendor>sun</vendor>
+                                </jdk>
+                            </toolchains>
+                        </configuration>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
 </project>
\ No newline at end of file


Test patch:
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 2a56adef9..7c327edfd 100644
--- a/core/src/test/java/com/netease/arctic/trace/TableTracerTest.java
+++ b/core/src/test/java/com/netease/arctic/trace/TableTracerTest.java
@@ -29,6 +29,7 @@ import com.netease.arctic.io.writer.GenericTaskWriters;
 import com.netease.arctic.io.writer.SortedPosDeleteWriter;
 import org.apache.iceberg.DataOperations;
 import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.DeleteFiles;
 import org.apache.iceberg.OverwriteFiles;
 import org.apache.iceberg.RowDelta;
 import org.apache.iceberg.Snapshot;
@@ -68,7 +69,7 @@ public class TableTracerTest extends TableTestBase {
         .appendFile(FILE_B)
         .commit();
 
-    Assert.assertTrue(!AMS.handler().getTableCommitMetas().containsKey(TABLE_ID.buildTableIdentifier()));
+    Assert.assertFalse(AMS.handler().getTableCommitMetas().containsKey(TABLE_ID.buildTableIdentifier()));
 
     transaction.commitTransaction();
 
@@ -101,7 +102,7 @@ public class TableTracerTest extends TableTestBase {
         .appendFile(FILE_B)
         .commit();
 
-    Assert.assertTrue(!AMS.handler().getTableCommitMetas().containsKey(TABLE_ID.buildTableIdentifier()));
+    Assert.assertFalse(AMS.handler().getTableCommitMetas().containsKey(TABLE_ID.buildTableIdentifier()));
 
     transaction.commitTransaction();
     List<TableCommitMeta> TableCommitMetas = AMS.handler().getTableCommitMetas().get(TABLE_ID.buildTableIdentifier());
@@ -252,6 +253,39 @@ public class TableTracerTest extends TableTestBase {
     Assert.assertEquals(2, tableChange.deleteFiles.size());
   }
 
+  @Test
+  public void testTraceRemovePosDeleteInternalInTransaction() throws Exception {
+    testKeyedTable.baseTable().newAppend().appendFile(FILE_A).commit();
+
+    SortedPosDeleteWriter<Record> writer = GenericTaskWriters.builderFor(testKeyedTable)
+        .withTransactionId(1).buildBasePosDeleteWriter(2, 1, FILE_A.partition());
+    writer.delete(FILE_A.path(), 1);
+    writer.delete(FILE_A.path(), 3);
+    writer.delete(FILE_A.path(), 5);
+    List<DeleteFile> result = writer.complete();
+    RowDelta rowDelta = testKeyedTable.baseTable().newRowDelta();
+    result.forEach(rowDelta::addDeletes);
+    rowDelta.commit();
+
+    testKeyedTable.baseTable().newAppend().appendFile(FILE_C).commit();
+    Transaction transaction = testKeyedTable.baseTable().newTransaction();
+    OverwriteFiles overwriteFiles = transaction.newOverwrite();
+    overwriteFiles.deleteFile(FILE_A);
+    overwriteFiles.addFile(FILE_B);
+    overwriteFiles.commit();
+    DeleteFiles deleteFiles = transaction.newDelete();
+    deleteFiles.deleteFile(FILE_C);
+    deleteFiles.commit();
+    transaction.commitTransaction();
+
+    List<TableCommitMeta> tableCommitMetas = AMS.handler().getTableCommitMetas().get(PK_TABLE_ID.buildTableIdentifier());
+    Assert.assertEquals(4, tableCommitMetas.size());
+    TableCommitMeta commitMeta = tableCommitMetas.get(tableCommitMetas.size() - 1);
+    Assert.assertEquals(2, commitMeta.getChanges().size());
+    TableChange tableChange = commitMeta.getChanges().get(0);
+    Assert.assertEquals(2, tableChange.deleteFiles.size());
+  }
+
   private void validateTableChange(Snapshot snapshot, TableChange tableChange,
                                    org.apache.iceberg.DataFile[] addFiles,
                                    org.apache.iceberg.DataFile[] deleteFiles) {