Repository: NetEase/arctic
Pull request #1065 (2023-02-07 06:50:41)

[Improvement][Core]: ams tracer should report snapshot information even if no new file is submitted
<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://arctic.netease.com/ch/contribute/
  2. If the PR is related to an issue in https://github.com/NetEase/arctic/issues, add '[ARCTIC-XXXX]' in your PR title, e.g., '[ARCTIC-XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][ARCTIC #XXXX] Your PR title ...'.
-->

## Why are the changes needed?
resolve https://github.com/NetEase/arctic/issues/1013

## Brief change log
AMS tracer report snapshot information even if no new file is submitted

## How was this patch tested?
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [ ] Run test locally before making a pull request

## Documentation

  - Does this pull request introduces a new feature? (no)
  - If yes, how is the feature documented? (not documented)



Patch:
diff --git a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/FileInfoCacheService.java b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/FileInfoCacheService.java
index e32d13832..86c58d43d 100644
--- a/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/FileInfoCacheService.java
+++ b/ams/ams-server/src/main/java/com/netease/arctic/ams/server/service/impl/FileInfoCacheService.java
@@ -490,9 +490,11 @@ public class FileInfoCacheService extends IJDBCService {
         cache.setProducer(tableCommitMeta.getCommitMetaProducer().name());
         long fileSize = 0L;
         int fileCount = 0;
-        for (DataFile file : tableChange.getAddFiles()) {
-          fileSize += file.getFileSize();
-          fileCount++;
+        if (CollectionUtils.isNotEmpty(tableChange.getAddFiles())) {
+          for (DataFile file : tableChange.getAddFiles()) {
+            fileSize += file.getFileSize();
+            fileCount++;
+          }
         }
         cache.setFileSize(fileSize);
         cache.setFileCount(fileCount);
diff --git a/core/src/main/java/com/netease/arctic/table/BaseUnkeyedTable.java b/core/src/main/java/com/netease/arctic/table/BaseUnkeyedTable.java
index f1035accc..08f69ef57 100644
--- a/core/src/main/java/com/netease/arctic/table/BaseUnkeyedTable.java
+++ b/core/src/main/java/com/netease/arctic/table/BaseUnkeyedTable.java
@@ -172,7 +172,7 @@ public class BaseUnkeyedTable implements UnkeyedTable, HasTableOperations {
 
   @Override
   public UpdateSchema updateSchema() {
-    return new TracedSchemaUpdate(icebergTable.updateSchema(), new AmsTableTracer(this, null, client));
+    return new TracedSchemaUpdate(icebergTable.updateSchema(), new AmsTableTracer(this, null, client, false));
   }
 
   @Override
@@ -184,7 +184,7 @@ public class BaseUnkeyedTable implements UnkeyedTable, HasTableOperations {
   public UpdateProperties updateProperties() {
     UpdateProperties updateProperties = icebergTable.updateProperties();
     if (client != null) {
-      AmsTableTracer tracer = new AmsTableTracer(this, TrackerOperations.UPDATE_PROPERTIES, client);
+      AmsTableTracer tracer = new AmsTableTracer(this, TrackerOperations.UPDATE_PROPERTIES, client, false);
       return new TracedUpdateProperties(updateProperties, tracer);
     } else {
       return updateProperties;
@@ -205,7 +205,7 @@ public class BaseUnkeyedTable implements UnkeyedTable, HasTableOperations {
   public AppendFiles newAppend() {
     AppendFiles appendFiles = icebergTable.newAppend();
     if (client != null) {
-      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.APPEND, client);
+      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.APPEND, client, true);
       return new TracedAppendFiles(appendFiles, tracer);
     } else {
       return appendFiles;
@@ -216,7 +216,7 @@ public class BaseUnkeyedTable implements UnkeyedTable, HasTableOperations {
   public AppendFiles newFastAppend() {
     AppendFiles appendFiles = icebergTable.newFastAppend();
     if (client != null) {
-      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.APPEND, client);
+      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.APPEND, client, true);
       return new TracedAppendFiles(appendFiles, tracer);
     } else {
       return appendFiles;
@@ -227,7 +227,7 @@ public class BaseUnkeyedTable implements UnkeyedTable, HasTableOperations {
   public RewriteFiles newRewrite() {
     RewriteFiles rewriteFiles = icebergTable.newRewrite();
     if (client != null) {
-      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.REPLACE, client);
+      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.REPLACE, client, true);
       return new TracedRewriteFiles(rewriteFiles, tracer);
     } else {
       return rewriteFiles;
@@ -243,7 +243,7 @@ public class BaseUnkeyedTable implements UnkeyedTable, HasTableOperations {
   public OverwriteFiles newOverwrite() {
     OverwriteFiles overwriteFiles = icebergTable.newOverwrite();
     if (client != null) {
-      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.OVERWRITE, client);
+      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.OVERWRITE, client, true);
       return new TracedOverwriteFiles(overwriteFiles, tracer);
     } else {
       return overwriteFiles;
@@ -254,7 +254,7 @@ public class BaseUnkeyedTable implements UnkeyedTable, HasTableOperations {
   public RowDelta newRowDelta() {
     RowDelta rowDelta = icebergTable.newRowDelta();
     if (client != null) {
-      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.OVERWRITE, client);
+      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.OVERWRITE, client, true);
       return new TracedRowDelta(rowDelta, tracer);
     } else {
       return rowDelta;
@@ -265,7 +265,7 @@ public class BaseUnkeyedTable implements UnkeyedTable, HasTableOperations {
   public ReplacePartitions newReplacePartitions() {
     ReplacePartitions replacePartitions = icebergTable.newReplacePartitions();
     if (client != null) {
-      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.OVERWRITE, client);
+      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.OVERWRITE, client, true);
       return new TracedReplacePartitions(replacePartitions, tracer);
     } else {
       return replacePartitions;
@@ -276,7 +276,7 @@ public class BaseUnkeyedTable implements UnkeyedTable, HasTableOperations {
   public DeleteFiles newDelete() {
     DeleteFiles deleteFiles = icebergTable.newDelete();
     if (client != null) {
-      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.DELETE, client);
+      TableTracer tracer = new AmsTableTracer(this, TrackerOperations.DELETE, client, true);
       return new TracedDeleteFiles(deleteFiles, tracer);
     } else {
       return deleteFiles;
@@ -302,7 +302,7 @@ public class BaseUnkeyedTable implements UnkeyedTable, HasTableOperations {
   public Transaction newTransaction() {
     Transaction transaction = icebergTable.newTransaction();
     if (client != null) {
-      return new TracedTransaction(transaction, new AmsTableTracer(this, client));
+      return new TracedTransaction(transaction, new AmsTableTracer(this, client, false));
     } else {
       return transaction;
     }
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 a992933d6..1eefdd8b9 100644
--- a/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java
+++ b/core/src/main/java/com/netease/arctic/trace/AmsTableTracer.java
@@ -66,12 +66,23 @@ public class AmsTableTracer implements TableTracer {
   private final Map<Long, AmsTableTracer.InternalTableChange> transactionSnapshotTableChanges = new LinkedHashMap<>();
   private final List<UpdateColumn> updateColumns = new ArrayList<>();
 
-  public AmsTableTracer(UnkeyedTable table, String action, AmsClient client) {
+  /**
+   * Construct a new AmsTableTracer
+   *
+   * @param table             table object
+   * @param action            Table change operation name
+   * @param client            AMS client
+   * @param commitNewSnapshot Whether to submit a new snapshot(but not a transaction)
+   */
+  public AmsTableTracer(UnkeyedTable table, String action, AmsClient client, boolean commitNewSnapshot) {
     this.innerTable = table instanceof ChangeTable ?
         Constants.INNER_TABLE_CHANGE : Constants.INNER_TABLE_BASE;
     this.table = table;
     this.action = action;
     this.client = client;
+    if (commitNewSnapshot) {
+      this.defaultTableChange = new InternalTableChange();
+    }
   }
 
   public AmsTableTracer(KeyedTable table, String action, AmsClient client) {
@@ -80,8 +91,8 @@ public class AmsTableTracer implements TableTracer {
     this.client = client;
   }
 
-  public AmsTableTracer(UnkeyedTable table, AmsClient client) {
-    this(table, null, client);
+  public AmsTableTracer(UnkeyedTable table, AmsClient client, boolean commitNewSnapshot) {
+    this(table, null, client, commitNewSnapshot);
   }
 
   @Override
@@ -106,7 +117,7 @@ public class AmsTableTracer implements TableTracer {
 
   private InternalTableChange getDefaultChange() {
     if (defaultTableChange == null) {
-      defaultTableChange = new InternalTableChange();
+      throw new RuntimeException("This operation should not result in changes to files or snapshots");
     }
     return defaultTableChange;
   }
@@ -128,14 +139,24 @@ public class AmsTableTracer implements TableTracer {
     TableCommitMeta commitMeta = new TableCommitMeta();
     commitMeta.setTableIdentifier(table.id().buildTableIdentifier());
     commitMeta.setAction(action);
-    String commitMetaSource = PropertyUtil.propertyAsString(snapshotSummary,
+    String commitMetaSource = PropertyUtil.propertyAsString(
+        snapshotSummary,
         com.netease.arctic.trace.SnapshotSummary.SNAPSHOT_PRODUCER,
         com.netease.arctic.trace.SnapshotSummary.SNAPSHOT_PRODUCER_DEFAULT);
     commitMeta.setCommitMetaProducer(CommitMetaProducer.valueOf(commitMetaSource));
     commitMeta.setCommitTime(System.currentTimeMillis());
     boolean update = false;
 
-    if (defaultTableChange != null) {
+    if (transactionSnapshotTableChanges.size() > 0) {
+      transactionSnapshotTableChanges.forEach((snapshotId, internalTableChange) -> {
+        if (table.isUnkeyedTable()) {
+          Snapshot snapshot = table.asUnkeyedTable().snapshot(snapshotId);
+          TableChange tableChange = internalTableChange.toTableChange(table, snapshot, innerTable);
+          commitMeta.addToChanges(tableChange);
+        }
+      });
+      update = true;
+    } else if (defaultTableChange != null) {
       Table traceTable;
       if (table.isUnkeyedTable()) {
         traceTable = table.asUnkeyedTable();
@@ -143,21 +164,9 @@ public class AmsTableTracer implements TableTracer {
         throw new IllegalStateException("can't apply table change on keyed table.");
       }
 
-      Optional<TableChange> tableChange =
+      TableChange tableChange =
           defaultTableChange.toTableChange(table, traceTable.currentSnapshot(), innerTable);
-      if (tableChange.isPresent()) {
-        commitMeta.addToChanges(tableChange.get());
-        update = true;
-      }
-    }
-    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);
-        }
-      });
+      commitMeta.addToChanges(tableChange);
       update = true;
     }
     if (updateColumns.size() > 0 && Constants.INNER_TABLE_BASE.equals(innerTable)) {
@@ -233,49 +242,44 @@ public class AmsTableTracer implements TableTracer {
      * Build {@link TableChange} to report to ams.
      *
      * @param arcticTable arctic table which table change belongs
-     * @param snapshot  the snapshot produced in this operation
+     * @param snapshot    the snapshot produced in this operation
      * @param innerTable  inner table name
      * @return table change
      */
-    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 = snapshot.snapshotId();
-        long parentSnapshotId =
-            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));
-        long realDeletedDataFiles = summary.get(SnapshotSummary.DELETED_FILES_PROP) == null ?
-            0 : Long.parseLong(summary.get(SnapshotSummary.DELETED_FILES_PROP));
-        long realAddedDeleteFiles = summary.get(SnapshotSummary.ADDED_DELETE_FILES_PROP) == null ?
-            0 : Long.parseLong(summary.get(SnapshotSummary.ADDED_DELETE_FILES_PROP));
-        long readRemovedDeleteFiles = summary.get(SnapshotSummary.REMOVED_DELETE_FILES_PROP) == null ?
-            0 : Long.parseLong(summary.get(SnapshotSummary.REMOVED_DELETE_FILES_PROP));
-
-        List<com.netease.arctic.ams.api.DataFile> addFiles = new ArrayList<>();
-        List<com.netease.arctic.ams.api.DataFile> deleteFiles = new ArrayList<>();
-        if (realAddedDataFiles == addedFiles.size() && realDeletedDataFiles == deletedFiles.size() &&
-            realAddedDeleteFiles == addedDeleteFiles.size() && readRemovedDeleteFiles == deletedDeleteFiles.size()) {
-          addFiles =
-              addedFiles.stream().map(file -> ConvertStructUtil.convertToAmsDatafile(file, arcticTable))
-                  .collect(Collectors.toList());
-          deleteFiles =
-              deletedFiles.stream().map(file -> ConvertStructUtil.convertToAmsDatafile(file, arcticTable))
-                  .collect(Collectors.toList());
-          addFiles.addAll(addedDeleteFiles.stream()
-              .map(file -> ConvertStructUtil.convertToAmsDatafile(file, arcticTable)).collect(Collectors.toList()));
-          deleteFiles.addAll(deletedDeleteFiles.stream()
-              .map(file -> ConvertStructUtil.convertToAmsDatafile(file, arcticTable)).collect(Collectors.toList()));
-        } else {
-          // tracer file change info is different from iceberg snapshot, should get iceberg real file change info
-          SnapshotFileUtil.getSnapshotFiles(arcticTable, snapshot, addFiles, deleteFiles);
-        }
-
-        return Optional.of(new TableChange(innerTable, addFiles, deleteFiles, currentSnapshotId, parentSnapshotId));
+    public TableChange toTableChange(ArcticTable arcticTable, Snapshot snapshot, String innerTable) {
+      long currentSnapshotId = snapshot.snapshotId();
+      long parentSnapshotId =
+          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));
+      long realDeletedDataFiles = summary.get(SnapshotSummary.DELETED_FILES_PROP) == null ?
+          0 : Long.parseLong(summary.get(SnapshotSummary.DELETED_FILES_PROP));
+      long realAddedDeleteFiles = summary.get(SnapshotSummary.ADDED_DELETE_FILES_PROP) == null ?
+          0 : Long.parseLong(summary.get(SnapshotSummary.ADDED_DELETE_FILES_PROP));
+      long readRemovedDeleteFiles = summary.get(SnapshotSummary.REMOVED_DELETE_FILES_PROP) == null ?
+          0 : Long.parseLong(summary.get(SnapshotSummary.REMOVED_DELETE_FILES_PROP));
+
+      List<com.netease.arctic.ams.api.DataFile> addFiles = new ArrayList<>();
+      List<com.netease.arctic.ams.api.DataFile> deleteFiles = new ArrayList<>();
+      if (realAddedDataFiles == addedFiles.size() && realDeletedDataFiles == deletedFiles.size() &&
+          realAddedDeleteFiles == addedDeleteFiles.size() && readRemovedDeleteFiles == deletedDeleteFiles.size()) {
+        addFiles =
+            addedFiles.stream().map(file -> ConvertStructUtil.convertToAmsDatafile(file, arcticTable))
+                .collect(Collectors.toList());
+        deleteFiles =
+            deletedFiles.stream().map(file -> ConvertStructUtil.convertToAmsDatafile(file, arcticTable))
+                .collect(Collectors.toList());
+        addFiles.addAll(addedDeleteFiles.stream()
+            .map(file -> ConvertStructUtil.convertToAmsDatafile(file, arcticTable)).collect(Collectors.toList()));
+        deleteFiles.addAll(deletedDeleteFiles.stream()
+            .map(file -> ConvertStructUtil.convertToAmsDatafile(file, arcticTable)).collect(Collectors.toList()));
       } else {
-        return Optional.empty();
+        // tracer file change info is different from iceberg snapshot, should get iceberg real file change info
+        SnapshotFileUtil.getSnapshotFiles(arcticTable, snapshot, addFiles, deleteFiles);
       }
+
+      return new TableChange(innerTable, addFiles, deleteFiles, currentSnapshotId, parentSnapshotId);
     }
   }
 
diff --git a/hive/src/main/java/com/netease/arctic/hive/op/UpdateHiveFiles.java b/hive/src/main/java/com/netease/arctic/hive/op/UpdateHiveFiles.java
index 72183c8da..5fd0bb021 100644
--- a/hive/src/main/java/com/netease/arctic/hive/op/UpdateHiveFiles.java
+++ b/hive/src/main/java/com/netease/arctic/hive/op/UpdateHiveFiles.java
@@ -119,10 +119,14 @@ public abstract class UpdateHiveFiles<T extends SnapshotUpdate<T>> implements Sn
       return;
     }
 
-    if (table.spec().isUnpartitioned()) {
-      commitNonPartitionedTable();
-    } else {
-      commitPartitionedTable();
+    try {
+      if (table.spec().isUnpartitioned()) {
+        commitNonPartitionedTable();
+      } else {
+        commitPartitionedTable();
+      }
+    } catch (Exception e) {
+      LOG.warn("Commit operation to HMS failed.", e);
     }
   }
 


Test patch:
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java
index 1c17d3ea3..a42f86d9d 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/AmsTestBase.java
@@ -102,7 +102,8 @@ import static org.powermock.api.mockito.PowerMockito.when;
     TestSupportHiveMajorOptimizeCommit.class,
     TestSupportHiveSyncService.class,
     TestExpiredFileCleanSupportHive.class,
-    TestOrphanFileCleanSupportHive.class})
+    TestOrphanFileCleanSupportHive.class
+})
 @PrepareForTest({
     JDBCSqlSessionFactoryProvider.class,
     ArcticMetaStore.class,
diff --git a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestFileInfoCacheService.java b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestFileInfoCacheService.java
index 9f592e2c5..9782a3e03 100644
--- a/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestFileInfoCacheService.java
+++ b/ams/ams-server/src/test/java/com/netease/arctic/ams/server/service/TestFileInfoCacheService.java
@@ -92,6 +92,34 @@ public class TestFileInfoCacheService extends TableTestBase {
     Assert.assertEquals(snapshotId, transactionsOfTables.get(1).getTransactionId());
   }
 
+  @Test
+  public void testNoNewFilesCommit() throws MetaException {
+    TableCommitMeta meta = new TableCommitMeta();
+    meta.setAction("append");
+    meta.setCommitTime(System.currentTimeMillis());
+    meta.setCommitMetaProducer(CommitMetaProducer.INGESTION);
+    TableIdentifier tableIdentifier = new TableIdentifier(AMS_TEST_CATALOG_NAME, "test", "testNoNewFilesCommit");
+    meta.setTableIdentifier(tableIdentifier);
+    List<TableChange> changes = new ArrayList<>();
+    TableChange change = new TableChange();
+    change.setParentSnapshotId(-1);
+    change.setInnerTable("base");
+    long snapshotId = 1L;
+    change.setSnapshotId(snapshotId);
+
+    changes.add(change);
+    meta.setChanges(changes);
+    Map<String, String> properties = new HashMap<>();
+    properties.put(TableProperties.TABLE_EVENT_TIME_FIELD, "eventTime");
+    meta.setProperties(properties);
+    ServiceContainer.getFileInfoCacheService().commitCacheFileInfo(meta);
+
+    List<TransactionsOfTable> transactionsOfTables =
+        ServiceContainer.getFileInfoCacheService().getTxExcludeOptimize(tableIdentifier);
+    Assert.assertEquals(1, transactionsOfTables.size());
+    Assert.assertEquals(snapshotId, transactionsOfTables.get(0).getTransactionId());
+  }
+
   @Test
   public void testNeedFixCacheWhenParentNotCached() throws MetaException {
     TableCommitMeta meta = new TableCommitMeta();
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 7ba9f4e7c..435f15d6c 100644
--- a/core/src/test/java/com/netease/arctic/trace/TableTracerTest.java
+++ b/core/src/test/java/com/netease/arctic/trace/TableTracerTest.java
@@ -110,6 +110,45 @@ public class TableTracerTest extends TableTestBase {
         new org.apache.iceberg.DataFile[]{});
   }
 
+  @Test
+  public void testTraceAppendNoneFiles() {
+    testTable.newAppend()
+        .commit();
+    List<TableCommitMeta> appendTableCommitMetas =
+        AMS.handler().getTableCommitMetas().get(TABLE_ID.buildTableIdentifier());
+    Assert.assertEquals(1, appendTableCommitMetas.size());
+    Assert.assertNotNull(appendTableCommitMetas.get(0).getChanges());
+
+    Transaction overwritetransaction = testTable.newTransaction();
+    overwritetransaction.newOverwrite()
+        .commit();
+    overwritetransaction.commitTransaction();
+    List<TableCommitMeta> overwriteTableCommitMetas =
+        AMS.handler().getTableCommitMetas().get(TABLE_ID.buildTableIdentifier());
+    Assert.assertEquals(2, overwriteTableCommitMetas.size());
+    Assert.assertEquals(1, overwriteTableCommitMetas.get(1).getChanges().size());
+
+    Transaction rewriteTransaction = testTable.newTransaction();
+    rewriteTransaction.newRewrite()
+        .commit();
+    rewriteTransaction.commitTransaction();
+    List<TableCommitMeta> rewriteTableCommitMetas =
+        AMS.handler().getTableCommitMetas().get(TABLE_ID.buildTableIdentifier());
+    Assert.assertEquals(3, rewriteTableCommitMetas.size());
+    Assert.assertEquals(1, rewriteTableCommitMetas.get(2).getChanges().size());
+
+    testTable.updateSchema().commit();
+    List<TableCommitMeta> updateSchemaCommitMetas =
+        AMS.handler().getTableCommitMetas().get(TABLE_ID.buildTableIdentifier());
+    Assert.assertEquals(3, updateSchemaCommitMetas.size());
+
+    testTable.updateProperties().commit();
+    List<TableCommitMeta> updatePropertiesCommitMetas =
+        AMS.handler().getTableCommitMetas().get(TABLE_ID.buildTableIdentifier());
+    Assert.assertEquals(4, updatePropertiesCommitMetas.size());
+    Assert.assertNull(rewriteTableCommitMetas.get(3).getChanges());
+  }
+
   @Test
   public void testTraceFastAppendInTx() {
     Transaction transaction = testTable.newTransaction();