Repository: apache/iceberg (language: python)

Issue:
Flink: Maintenance - TableChange refactor

Flink: Maintenance - TableChange refactor



Discussion / hints:

Fix patch:
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/JdbcLockFactory.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/JdbcLockFactory.java
index 21c8935abede..f22be33aeae6 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/JdbcLockFactory.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/JdbcLockFactory.java
@@ -105,12 +105,12 @@ void open(JdbcLockFactory other) {
 
   @Override
   public Lock createLock() {
-    return new Lock(pool, lockId, Type.MAINTENANCE);
+    return new JdbcLock(pool, lockId, Type.MAINTENANCE);
   }
 
   @Override
   public Lock createRecoveryLock() {
-    return new Lock(pool, lockId, Type.RECOVERY);
+    return new JdbcLock(pool, lockId, Type.RECOVERY);
   }
 
   @Override
@@ -153,12 +153,12 @@ private void initializeLockTables() {
     }
   }
 
-  public static class Lock implements TriggerLockFactory.Lock {
+  private static class JdbcLock implements TriggerLockFactory.Lock {
     private final JdbcClientPool pool;
     private final String lockId;
     private final Type type;
 
-    public Lock(JdbcClientPool pool, String lockId, Type type) {
+    private JdbcLock(JdbcClientPool pool, String lockId, Type type) {
       this.pool = pool;
       this.lockId = lockId;
       this.type = type;
@@ -221,7 +221,7 @@ public boolean isHeld() {
         throw new UncheckedInterruptedException(e, "Interrupted during isHeld");
       } catch (SQLException e) {
         // SQL exception happened when getting lock information
-        throw new UncheckedSQLException(e, "Failed to get lock information for %s", this);
+        throw new UncheckedSQLException(e, "Failed to check the state of the lock %s", this);
       }
     }
 
@@ -266,8 +266,6 @@ public void unlock() {
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
         throw new UncheckedInterruptedException(e, "Interrupted during unlock");
-      } catch (UncheckedSQLException e) {
-        throw e;
       } catch (SQLException e) {
         // SQL exception happened when getting/updating lock information
         throw new UncheckedSQLException(e, "Failed to remove lock %s", this);
@@ -312,7 +310,7 @@ private enum Type {
     MAINTENANCE("m"),
     RECOVERY("r");
 
-    private String key;
+    private final String key;
 
     Type(String key) {
       this.key = key;
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableChange.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableChange.java
index 7d0b94e97da7..5252cf61b031 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableChange.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TableChange.java
@@ -29,19 +29,29 @@
 /** Event describing changes in an Iceberg table */
 @Internal
 class TableChange {
-  private int dataFileNum;
-  private int deleteFileNum;
-  private long dataFileSize;
-  private long deleteFileSize;
-  private int commitNum;
-
-  private TableChange(
-      int dataFileNum, int deleteFileNum, long dataFileSize, long deleteFileSize, int commitNum) {
-    this.dataFileNum = dataFileNum;
-    this.deleteFileNum = deleteFileNum;
-    this.dataFileSize = dataFileSize;
-    this.deleteFileSize = deleteFileSize;
-    this.commitNum = commitNum;
+  private int dataFileCount;
+  private long dataFileSizeInBytes;
+  private int posDeleteFileCount;
+  private long posDeleteRecordCount;
+  private int eqDeleteFileCount;
+  private long eqDeleteRecordCount;
+  private int commitCount;
+
+  TableChange(
+      int dataFileCount,
+      long dataFileSizeInBytes,
+      int posDeleteFileCount,
+      long posDeleteRecordCount,
+      int eqDeleteFileCount,
+      long eqDeleteRecordCount,
+      int commitCount) {
+    this.dataFileCount = dataFileCount;
+    this.dataFileSizeInBytes = dataFileSizeInBytes;
+    this.posDeleteFileCount = posDeleteFileCount;
+    this.posDeleteRecordCount = posDeleteRecordCount;
+    this.eqDeleteFileCount = eqDeleteFileCount;
+    this.eqDeleteRecordCount = eqDeleteRecordCount;
+    this.commitCount = commitCount;
   }
 
   TableChange(Snapshot snapshot, FileIO io) {
@@ -50,67 +60,96 @@ private TableChange(
 
     dataFiles.forEach(
         dataFile -> {
-          this.dataFileNum++;
-          this.dataFileSize += dataFile.fileSizeInBytes();
+          this.dataFileCount++;
+          this.dataFileSizeInBytes += dataFile.fileSizeInBytes();
         });
 
     deleteFiles.forEach(
         deleteFile -> {
-          this.deleteFileNum++;
-          this.deleteFileSize += deleteFile.fileSizeInBytes();
+          switch (deleteFile.content()) {
+            case POSITION_DELETES:
+              this.posDeleteFileCount++;
+              this.posDeleteRecordCount += deleteFile.recordCount();
+              break;
+            case EQUALITY_DELETES:
+              this.eqDeleteFileCount++;
+              this.eqDeleteRecordCount += deleteFile.recordCount();
+              break;
+            default:
+              throw new IllegalArgumentException("Unexpected delete file content: " + deleteFile);
+          }
         });
 
-    this.commitNum = 1;
+    this.commitCount = 1;
   }
 
   static TableChange empty() {
-    return new TableChange(0, 0, 0L, 0L, 0);
+    return new TableChange(0, 0L, 0, 0L, 0, 0L, 0);
   }
 
   static Builder builder() {
     return new Builder();
   }
 
-  int dataFileNum() {
-    return dataFileNum;
+  int dataFileCount() {
+    return dataFileCount;
+  }
+
+  long dataFileSizeInBytes() {
+    return dataFileSizeInBytes;
   }
 
-  int deleteFileNum() {
-    return deleteFileNum;
+  int posDeleteFileCount() {
+    return posDeleteFileCount;
   }
 
-  long dataFileSize() {
-    return dataFileSize;
+  long posDeleteRecordCount() {
+    return posDeleteRecordCount;
   }
 
-  long deleteFileSize() {
-    return deleteFileSize;
+  int eqDeleteFileCount() {
+    return eqDeleteFileCount;
   }
 
-  public int commitNum() {
-    return commitNum;
+  long eqDeleteRecordCount() {
+    return eqDeleteRecordCount;
+  }
+
+  public int commitCount() {
+    return commitCount;
   }
 
   public void merge(TableChange other) {
-    this.dataFileNum += other.dataFileNum;
-    this.deleteFileNum += other.deleteFileNum;
-    this.dataFileSize += other.dataFileSize;
-    this.deleteFileSize += other.deleteFileSize;
-    this.commitNum += other.commitNum;
+    this.dataFileCount += other.dataFileCount;
+    this.dataFileSizeInBytes += other.dataFileSizeInBytes;
+    this.posDeleteFileCount += other.posDeleteFileCount;
+    this.posDeleteRecordCount += other.posDeleteRecordCount;
+    this.eqDeleteFileCount += other.eqDeleteFileCount;
+    this.eqDeleteRecordCount += other.eqDeleteRecordCount;
+    this.commitCount += other.commitCount;
   }
 
   TableChange copy() {
-    return new TableChange(dataFileNum, deleteFileNum, dataFileSize, deleteFileSize, commitNum);
+    return new TableChange(
+        dataFileCount,
+        dataFileSizeInBytes,
+        posDeleteFileCount,
+        posDeleteRecordCount,
+        eqDeleteFileCount,
+        eqDeleteRecordCount,
+        commitCount);
   }
 
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
-        .add("dataFileNum", dataFileNum)
-        .add("deleteFileNum", deleteFileNum)
-        .add("dataFileSize", dataFileSize)
-        .add("deleteFileSize", deleteFileSize)
-        .add("commitNum", commitNum)
+        .add("dataFileCount", dataFileCount)
+        .add("dataFileSizeInBytes", dataFileSizeInBytes)
+        .add("posDeleteFileCount", posDeleteFileCount)
+        .add("posDeleteRecordCount", posDeleteRecordCount)
+        .add("eqDeleteFileCount", eqDeleteFileCount)
+        .add("eqDeleteRecordCount", eqDeleteRecordCount)
+        .add("commitCount", commitCount)
         .toString();
   }
 
@@ -123,52 +162,80 @@ public boolean equals(Object other) {
     }
 
     TableChange that = (TableChange) other;
-    return dataFileNum == that.dataFileNum
-        && deleteFileNum == that.deleteFileNum
-        && dataFileSize == that.dataFileSize
-        && deleteFileSize == that.deleteFileSize
-        && commitNum == that.commitNum;
+    return dataFileCount == that.dataFileCount
+        && dataFileSizeInBytes == that.dataFileSizeInBytes
+        && posDeleteFileCount == that.posDeleteFileCount
+        && posDeleteRecordCount == that.posDeleteRecordCount
+        && eqDeleteFileCount == that.eqDeleteFileCount
+        && eqDeleteRecordCount == that.eqDeleteRecordCount
+        && commitCount == that.commitCount;
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(dataFileNum, deleteFileNum, dataFileSize, deleteFileSize, commitNum);
+    return Objects.hash(
+        dataFileCount,
+        dataFileSizeInBytes,
+        posDeleteFileCount,
+        posDeleteRecordCount,
+        eqDeleteFileCount,
+        eqDeleteRecordCount,
+        commitCount);
   }
 
   static class Builder {
-    private int dataFileNum = 0;
-    private int deleteFileNum = 0;
-    private long dataFileSize = 0L;
-    private long deleteFileSize = 0L;
-    private int commitNum = 0;
-
-    public Builder dataFileNum(int newDataFileNum) {
-      this.dataFileNum = newDataFileNum;
+    private int dataFileCount = 0;
+    private long dataFileSizeInBytes = 0L;
+    private int posDeleteFileCount = 0;
+    private long posDeleteRecordCount = 0L;
+    private int eqDeleteFileCount = 0;
+    private long eqDeleteRecordCount = 0L;
+    private int commitCount = 0;
+
+    public Builder dataFileCount(int newDataFileCount) {
+      this.dataFileCount = newDataFileCount;
+      return this;
+    }
+
+    public Builder dataFileSizeInBytes(long newDataFileSizeInBytes) {
+      this.dataFileSizeInBytes = newDataFileSizeInBytes;
+      return this;
+    }
+
+    public Builder posDeleteFileCount(int newPosDeleteFileCount) {
+      this.posDeleteFileCount = newPosDeleteFileCount;
       return this;
     }
 
-    public Builder deleteFileNum(int newDeleteFileNum) {
-      this.deleteFileNum = newDeleteFileNum;
+    public Builder posDeleteRecordCount(long newPosDeleteRecordCount) {
+      this.posDeleteRecordCount = newPosDeleteRecordCount;
       return this;
     }
 
-    public Builder dataFileSize(long newDataFileSize) {
-      this.dataFileSize = newDataFileSize;
+    public Builder eqDeleteFileCount(int newEqDeleteFileCount) {
+      this.eqDeleteFileCount = newEqDeleteFileCount;
       return this;
     }
 
-    public Builder deleteFileSize(long newDeleteFileSize) {
-      this.deleteFileSize = newDeleteFileSize;
+    public Builder eqDeleteRecordCount(long newEqDeleteRecordCount) {
+      this.eqDeleteRecordCount = newEqDeleteRecordCount;
       return this;
     }
 
-    public Builder commitNum(int newCommitNum) {
-      this.commitNum = newCommitNum;
+    public Builder commitCount(int newCommitCount) {
+      this.commitCount = newCommitCount;
       return this;
     }
 
     public TableChange build() {
-      return new TableChange(dataFileNum, deleteFileNum, dataFileSize, deleteFileSize, commitNum);
+      return new TableChange(
+          dataFileCount,
+          dataFileSizeInBytes,
+          posDeleteFileCount,
+          posDeleteRecordCount,
+          eqDeleteFileCount,
+          eqDeleteRecordCount,
+          commitCount);
     }
   }
 }
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerEvaluator.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerEvaluator.java
index 37e4e3afd4bd..dba33b22a42a 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerEvaluator.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/TriggerEvaluator.java
@@ -40,15 +40,7 @@ private TriggerEvaluator(List<Predicate> predicates) {
 
   boolean check(TableChange event, long lastTimeMs, long currentTimeMs) {
     boolean result =
-        predicates.stream()
-            .anyMatch(
-                p -> {
-                  try {
-                    return p.evaluate(event, lastTimeMs, currentTimeMs);
-                  } catch (Exception e) {
-                    throw new RuntimeException("Error accessing state", e);
-                  }
-                });
+        predicates.stream().anyMatch(p -> p.evaluate(event, lastTimeMs, currentTimeMs));
     LOG.debug(
         "Checking event: {}, at {}, last: {} with result: {}",
         event,
@@ -59,29 +51,47 @@ boolean check(TableChange event, long lastTimeMs, long currentTimeMs) {
   }
 
   static class Builder implements Serializable {
-    private Integer commitNumber;
-    private Integer fileNumber;
-    private Long fileSize;
-    private Integer deleteFileNumber;
+    private Integer dataFileCount;
+    private Long dataFileSizeInBytes;
+    private Integer posDeleteFileCount;
+    private Long posDeleteRecordCount;
+    private Integer eqDeleteFileCount;
+    private Long eqDeleteRecordCount;
+    private Integer commitCount;
     private Duration timeout;
 
-    Builder commitNumber(int newCommitNumber) {
-      this.commitNumber = newCommitNumber;
+    public Builder dataFileCount(int newDataFileCount) {
+      this.dataFileCount = newDataFileCount;
       return this;
     }
 
-    Builder fileNumber(int newFileNumber) {
-      this.fileNumber = newFileNumber;
+    public Builder dataFileSizeInBytes(long neDataFileSizeInBytes) {
+      this.dataFileSizeInBytes = neDataFileSizeInBytes;
       return this;
     }
 
-    Builder fileSize(long newFileSize) {
-      this.fileSize = newFileSize;
+    public Builder posDeleteFileCount(int newPosDeleteFileCount) {
+      this.posDeleteFileCount = newPosDeleteFileCount;
       return this;
     }
 
-    Builder deleteFileNumber(int newDeleteFileNumber) {
-      this.deleteFileNumber = newDeleteFileNumber;
+    public Builder posDeleteRecordCount(long newPosDeleteRecordCount) {
+      this.posDeleteRecordCount = newPosDeleteRecordCount;
+      return this;
+    }
+
+    public Builder eqDeleteFileCount(int newEqDeleteFileCount) {
+      this.eqDeleteFileCount = newEqDeleteFileCount;
+      return this;
+    }
+
+    public Builder eqDeleteRecordCount(long newEqDeleteRecordCount) {
+      this.eqDeleteRecordCount = newEqDeleteRecordCount;
+      return this;
+    }
+
+    public Builder commitCount(int newCommitCount) {
+      this.commitCount = newCommitCount;
       return this;
     }
 
@@ -92,24 +102,37 @@ Builder timeout(Duration newTimeout) {
 
     TriggerEvaluator build() {
       List<Predicate> predicates = Lists.newArrayList();
-      if (commitNumber != null) {
-        predicates.add((change, unused, unused2) -> change.commitNum() >= commitNumber);
+      if (dataFileCount != null) {
+        predicates.add((change, unused, unused2) -> change.dataFileCount() >= dataFileCount);
+      }
+
+      if (dataFileSizeInBytes != null) {
+        predicates.add(
+            (change, unused, unused2) -> change.dataFileSizeInBytes() >= dataFileSizeInBytes);
+      }
+
+      if (posDeleteFileCount != null) {
+        predicates.add(
+            (change, unused, unused2) -> change.posDeleteFileCount() >= posDeleteFileCount);
+      }
+
+      if (posDeleteRecordCount != null) {
+        predicates.add(
+            (change, unused, unused2) -> change.posDeleteRecordCount() >= posDeleteRecordCount);
       }
 
-      if (fileNumber != null) {
+      if (eqDeleteFileCount != null) {
         predicates.add(
-            (change, unused, unused2) ->
-                change.dataFileNum() + change.deleteFileNum() >= fileNumber);
+            (change, unused, unused2) -> change.eqDeleteFileCount() >= eqDeleteFileCount);
       }
 
-      if (fileSize != null) {
+      if (eqDeleteRecordCount != null) {
         predicates.add(
-            (change, unused, unused2) ->
-                change.dataFileSize() + change.deleteFileSize() >= fileSize);
+            (change, unused, unused2) -> change.eqDeleteRecordCount() >= eqDeleteRecordCount);
       }
 
-      if (deleteFileNumber != null) {
-        predicates.add((change, unused, unused2) -> change.deleteFileNum() >= deleteFileNumber);
+      if (commitCount != null) {
+        predicates.add((change, unused, unused2) -> change.commitCount() >= commitCount);
       }
 
       if (timeout != null) {


Test patch:
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java
index 8c02601025ae..3aee05322561 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestMonitorSource.java
@@ -162,7 +162,11 @@ void testSource() throws Exception {
 
                 // The first non-empty event should contain the expected value
                 return newEvent.equals(
-                    TableChange.builder().dataFileNum(1).dataFileSize(size).commitNum(1).build());
+                    TableChange.builder()
+                        .dataFileCount(1)
+                        .dataFileSizeInBytes(size)
+                        .commitCount(1)
+                        .build());
               });
     } finally {
       closeJobClient(jobClient);
@@ -298,17 +302,17 @@ void testMaxReadBack() {
         new MonitorSource.TableChangeIterator(tableLoader, null, 1);
 
     // For a single maxReadBack we only get a single change
-    assertThat(iterator.next().commitNum()).isEqualTo(1);
+    assertThat(iterator.next().commitCount()).isEqualTo(1);
 
     iterator = new MonitorSource.TableChangeIterator(tableLoader, null, 2);
 
     // Expecting 2 commits/snapshots for maxReadBack=2
-    assertThat(iterator.next().commitNum()).isEqualTo(2);
+    assertThat(iterator.next().commitCount()).isEqualTo(2);
 
     iterator = new MonitorSource.TableChangeIterator(tableLoader, null, Long.MAX_VALUE);
 
     // For maxReadBack Long.MAX_VALUE we get every change
-    assertThat(iterator.next().commitNum()).isEqualTo(3);
+    assertThat(iterator.next().commitCount()).isEqualTo(3);
   }
 
   @Test
@@ -323,7 +327,7 @@ void testSkipReplace() {
         new MonitorSource.TableChangeIterator(tableLoader, null, Long.MAX_VALUE);
 
     // Read the current snapshot
-    assertThat(iterator.next().commitNum()).isEqualTo(1);
+    assertThat(iterator.next().commitCount()).isEqualTo(1);
 
     // Create a DataOperations.REPLACE snapshot
     Table table = tableLoader.loadTable();
@@ -350,16 +354,17 @@ private static TableChange tableChangeWithLastSnapshot(Table table, TableChange
         Lists.newArrayList(table.currentSnapshot().addedDeleteFiles(table.io()).iterator());
 
     long dataSize = dataFiles.stream().mapToLong(ContentFile::fileSizeInBytes).sum();
-    long deleteSize = deleteFiles.stream().mapToLong(ContentFile::fileSizeInBytes).sum();
+    long deleteRecordCount = deleteFiles.stream().mapToLong(DeleteFile::recordCount).sum();
 
     TableChange newChange = previous.copy();
     newChange.merge(
         TableChange.builder()
-            .dataFileNum(dataFiles.size())
-            .dataFileSize(dataSize)
-            .deleteFileNum(deleteFiles.size())
-            .deleteFileSize(deleteSize)
-            .commitNum(1)
+            .dataFileCount(dataFiles.size())
+            .dataFileSizeInBytes(dataSize)
+            // Currently we only test with equality deletes
+            .eqDeleteFileCount(deleteFiles.size())
+            .eqDeleteRecordCount(deleteRecordCount)
+            .commitCount(1)
             .build());
     return newChange;
   }
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestTriggerManager.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestTriggerManager.java
index 55e64f3e84bf..fba4a12d9c6b 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestTriggerManager.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestTriggerManager.java
@@ -81,95 +81,159 @@ void after() throws IOException {
   }
 
   @Test
-  void testCommitNumber() throws Exception {
+  void testCommitCount() throws Exception {
+    TriggerManager manager =
+        manager(sql.tableLoader(TABLE_NAME), new TriggerEvaluator.Builder().commitCount(3).build());
+    try (KeyedOneInputStreamOperatorTestHarness<Boolean, TableChange, Trigger> testHarness =
+        harness(manager)) {
+      testHarness.open();
+
+      addEventAndCheckResult(testHarness, TableChange.builder().commitCount(1).build(), 0);
+      addEventAndCheckResult(testHarness, TableChange.builder().commitCount(2).build(), 1);
+      addEventAndCheckResult(testHarness, TableChange.builder().commitCount(3).build(), 2);
+      addEventAndCheckResult(testHarness, TableChange.builder().commitCount(10).build(), 3);
+
+      // No trigger in this case
+      addEventAndCheckResult(testHarness, TableChange.builder().commitCount(1).build(), 3);
+      addEventAndCheckResult(testHarness, TableChange.builder().commitCount(1).build(), 3);
+
+      addEventAndCheckResult(testHarness, TableChange.builder().commitCount(1).build(), 4);
+    }
+  }
+
+  @Test
+  void testDataFileCount() throws Exception {
     TriggerManager manager =
         manager(
-            sql.tableLoader(TABLE_NAME), new TriggerEvaluator.Builder().commitNumber(3).build());
+            sql.tableLoader(TABLE_NAME), new TriggerEvaluator.Builder().dataFileCount(3).build());
     try (KeyedOneInputStreamOperatorTestHarness<Boolean, TableChange, Trigger> testHarness =
         harness(manager)) {
       testHarness.open();
 
-      addEventAndCheckResult(testHarness, TableChange.builder().commitNum(1).build(), 0);
-      addEventAndCheckResult(testHarness, TableChange.builder().commitNum(2).build(), 1);
-      addEventAndCheckResult(testHarness, TableChange.builder().commitNum(3).build(), 2);
-      addEventAndCheckResult(testHarness, TableChange.builder().commitNum(10).build(), 3);
+      addEventAndCheckResult(testHarness, TableChange.builder().dataFileCount(1).build(), 0);
+
+      addEventAndCheckResult(testHarness, TableChange.builder().dataFileCount(2).build(), 1);
+      addEventAndCheckResult(testHarness, TableChange.builder().dataFileCount(3).build(), 2);
+      addEventAndCheckResult(testHarness, TableChange.builder().dataFileCount(5).build(), 3);
 
       // No trigger in this case
-      addEventAndCheckResult(testHarness, TableChange.builder().commitNum(1).build(), 3);
-      addEventAndCheckResult(testHarness, TableChange.builder().commitNum(1).build(), 3);
+      addEventAndCheckResult(testHarness, TableChange.builder().dataFileCount(1).build(), 3);
 
-      addEventAndCheckResult(testHarness, TableChange.builder().commitNum(1).build(), 4);
+      addEventAndCheckResult(testHarness, TableChange.builder().dataFileCount(2).build(), 4);
     }
   }
 
   @Test
-  void testFileNumber() throws Exception {
+  void testDataFileSizeInBytes() throws Exception {
     TriggerManager manager =
-        manager(sql.tableLoader(TABLE_NAME), new TriggerEvaluator.Builder().fileNumber(3).build());
+        manager(
+            sql.tableLoader(TABLE_NAME),
+            new TriggerEvaluator.Builder().dataFileSizeInBytes(3).build());
     try (KeyedOneInputStreamOperatorTestHarness<Boolean, TableChange, Trigger> testHarness =
         harness(manager)) {
       testHarness.open();
 
-      addEventAndCheckResult(testHarness, TableChange.builder().dataFileNum(1).build(), 0);
+      addEventAndCheckResult(testHarness, TableChange.builder().dataFileSizeInBytes(1L).build(), 0);
+      addEventAndCheckResult(testHarness, TableChange.builder().dataFileSizeInBytes(2L).build(), 1);
+      addEventAndCheckResult(testHarness, TableChange.builder().dataFileSizeInBytes(5L).build(), 2);
 
-      addEventAndCheckResult(
-          testHarness, TableChange.builder().dataFileNum(1).deleteFileNum(1).build(), 1);
-      addEventAndCheckResult(testHarness, TableChange.builder().deleteFileNum(3).build(), 2);
-      addEventAndCheckResult(
-          testHarness, TableChange.builder().dataFileNum(5).deleteFileNum(7).build(), 3);
+      // No trigger in this case
+      addEventAndCheckResult(testHarness, TableChange.builder().dataFileSizeInBytes(1L).build(), 2);
+
+      addEventAndCheckResult(testHarness, TableChange.builder().dataFileSizeInBytes(2L).build(), 3);
+    }
+  }
+
+  @Test
+  void testPosDeleteFileCount() throws Exception {
+    TriggerManager manager =
+        manager(
+            sql.tableLoader(TABLE_NAME),
+            new TriggerEvaluator.Builder().posDeleteFileCount(3).build());
+    try (KeyedOneInputStreamOperatorTestHarness<Boolean, TableChange, Trigger> testHarness =
+        harness(manager)) {
+      testHarness.open();
+
+      addEventAndCheckResult(testHarness, TableChange.builder().posDeleteFileCount(1).build(), 0);
+      addEventAndCheckResult(testHarness, TableChange.builder().posDeleteFileCount(2).build(), 1);
+      addEventAndCheckResult(testHarness, TableChange.builder().posDeleteFileCount(3).build(), 2);
+      addEventAndCheckResult(testHarness, TableChange.builder().posDeleteFileCount(10).build(), 3);
 
       // No trigger in this case
-      addEventAndCheckResult(testHarness, TableChange.builder().dataFileNum(1).build(), 3);
-      addEventAndCheckResult(testHarness, TableChange.builder().deleteFileNum(1).build(), 3);
+      addEventAndCheckResult(testHarness, TableChange.builder().posDeleteFileCount(1).build(), 3);
+      addEventAndCheckResult(testHarness, TableChange.builder().posDeleteFileCount(1).build(), 3);
 
-      addEventAndCheckResult(testHarness, TableChange.builder().dataFileNum(1).build(), 4);
+      addEventAndCheckResult(testHarness, TableChange.builder().posDeleteFileCount(1).build(), 4);
     }
   }
 
   @Test
-  void testFileSize() throws Exception {
+  void testPosDeleteRecordCount() throws Exception {
     TriggerManager manager =
-        manager(sql.tableLoader(TABLE_NAME), new TriggerEvaluator.Builder().fileSize(3).build());
+        manager(
+            sql.tableLoader(TABLE_NAME),
+            new TriggerEvaluator.Builder().posDeleteRecordCount(3).build());
     try (KeyedOneInputStreamOperatorTestHarness<Boolean, TableChange, Trigger> testHarness =
         harness(manager)) {
       testHarness.open();
 
-      addEventAndCheckResult(testHarness, TableChange.builder().dataFileSize(1L).build(), 0);
       addEventAndCheckResult(
-          testHarness, TableChange.builder().dataFileSize(1L).deleteFileSize(1L).build(), 1);
-      addEventAndCheckResult(testHarness, TableChange.builder().deleteFileSize(3L).build(), 2);
+          testHarness, TableChange.builder().posDeleteRecordCount(1L).build(), 0);
       addEventAndCheckResult(
-          testHarness, TableChange.builder().dataFileSize(5L).deleteFileSize(7L).build(), 3);
+          testHarness, TableChange.builder().posDeleteRecordCount(2L).build(), 1);
+      addEventAndCheckResult(
+          testHarness, TableChange.builder().posDeleteRecordCount(5L).build(), 2);
 
       // No trigger in this case
-      addEventAndCheckResult(testHarness, TableChange.builder().dataFileSize(1L).build(), 3);
-      addEventAndCheckResult(testHarness, TableChange.builder().deleteFileSize(1L).build(), 3);
+      addEventAndCheckResult(
+          testHarness, TableChange.builder().posDeleteRecordCount(1L).build(), 2);
 
-      addEventAndCheckResult(testHarness, TableChange.builder().dataFileSize(1L).build(), 4);
+      addEventAndCheckResult(
+          testHarness, TableChange.builder().posDeleteRecordCount(2L).build(), 3);
     }
   }
 
   @Test
-  void testDeleteFileNumber() throws Exception {
+  void testEqDeleteFileCount() throws Exception {
     TriggerManager manager =
         manager(
             sql.tableLoader(TABLE_NAME),
-            new TriggerEvaluator.Builder().deleteFileNumber(3).build());
+            new TriggerEvaluator.Builder().eqDeleteFileCount(3).build());
     try (KeyedOneInputStreamOperatorTestHarness<Boolean, TableChange, Trigger> testHarness =
         harness(manager)) {
       testHarness.open();
 
-      addEventAndCheckResult(
-          testHarness, TableChange.builder().dataFileNum(3).deleteFileNum(1).build(), 0);
-      addEventAndCheckResult(testHarness, TableChange.builder().deleteFileNum(2).build(), 1);
-      addEventAndCheckResult(testHarness, TableChange.builder().deleteFileNum(3).build(), 2);
-      addEventAndCheckResult(testHarness, TableChange.builder().deleteFileNum(10).build(), 3);
+      addEventAndCheckResult(testHarness, TableChange.builder().eqDeleteFileCount(1).build(), 0);
+      addEventAndCheckResult(testHarness, TableChange.builder().eqDeleteFileCount(2).build(), 1);
+      addEventAndCheckResult(testHarness, TableChange.builder().eqDeleteFileCount(3).build(), 2);
+      addEventAndCheckResult(testHarness, TableChange.builder().eqDeleteFileCount(10).build(), 3);
+
+      // No trigger in this case
+      addEventAndCheckResult(testHarness, TableChange.builder().eqDeleteFileCount(1).build(), 3);
+      addEventAndCheckResult(testHarness, TableChange.builder().eqDeleteFileCount(1).build(), 3);
+
+      addEventAndCheckResult(testHarness, TableChange.builder().eqDeleteFileCount(1).build(), 4);
+    }
+  }
+
+  @Test
+  void testEqDeleteRecordCount() throws Exception {
+    TriggerManager manager =
+        manager(
+            sql.tableLoader(TABLE_NAME),
+            new TriggerEvaluator.Builder().eqDeleteRecordCount(3).build());
+    try (KeyedOneInputStreamOperatorTestHarness<Boolean, TableChange, Trigger> testHarness =
+        harness(manager)) {
+      testHarness.open();
+
+      addEventAndCheckResult(testHarness, TableChange.builder().eqDeleteRecordCount(1L).build(), 0);
+      addEventAndCheckResult(testHarness, TableChange.builder().eqDeleteRecordCount(2L).build(), 1);
+      addEventAndCheckResult(testHarness, TableChange.builder().eqDeleteRecordCount(5L).build(), 2);
 
       // No trigger in this case
-      addEventAndCheckResult(testHarness, TableChange.builder().deleteFileNum(1).build(), 3);
-      addEventAndCheckResult(testHarness, TableChange.builder().deleteFileNum(1).build(), 3);
+      addEventAndCheckResult(testHarness, TableChange.builder().eqDeleteRecordCount(1L).build(), 2);
 
-      addEventAndCheckResult(testHarness, TableChange.builder().deleteFileNum(1).build(), 4);
+      addEventAndCheckResult(testHarness, TableChange.builder().eqDeleteRecordCount(2L).build(), 3);
     }
   }
 
@@ -183,7 +247,7 @@ void testTimeout() throws Exception {
         harness(manager)) {
       testHarness.open();
 
-      TableChange event = TableChange.builder().dataFileSize(1).commitNum(1).build();
+      TableChange event = TableChange.builder().dataFileCount(1).commitCount(1).build();
 
       // Wait for some time
       testHarness.processElement(event, EVENT_TIME);
@@ -225,7 +289,7 @@ void testStateRestore() throws Exception {
       testHarness.open();
 
       testHarness.processElement(
-          TableChange.builder().dataFileSize(1).commitNum(1).build(), EVENT_TIME);
+          TableChange.builder().dataFileCount(1).commitCount(1).build(), EVENT_TIME);
 
       assertThat(testHarness.extractOutputValues()).isEmpty();
 
@@ -240,7 +304,7 @@ void testStateRestore() throws Exception {
       testHarness.open();
 
       // Arrives the first real change which triggers the recovery process
-      testHarness.processElement(TableChange.builder().commitNum(1).build(), EVENT_TIME_2);
+      testHarness.processElement(TableChange.builder().commitCount(1).build(), EVENT_TIME_2);
       assertTriggers(
           testHarness.extractOutputValues(),
           Lists.newArrayList(Trigger.recovery(testHarness.getProcessingTime())));
@@ -261,11 +325,11 @@ void testMinFireDelay() throws Exception {
         harness(manager)) {
       testHarness.open();
 
-      addEventAndCheckResult(testHarness, TableChange.builder().commitNum(2).build(), 1);
+      addEventAndCheckResult(testHarness, TableChange.builder().commitCount(2).build(), 1);
       long currentTime = testHarness.getProcessingTime();
 
       // No new fire yet
-      addEventAndCheckResult(testHarness, TableChange.builder().commitNum(2).build(), 1);
+      addEventAndCheckResult(testHarness, TableChange.builder().commitCount(2).build(), 1);
 
       // Check that the trigger fired after the delay
       testHarness.setProcessingTime(currentTime + DELAY);
@@ -281,11 +345,11 @@ void testLockCheckDelay() throws Exception {
         harness(manager)) {
       testHarness.open();
 
-      addEventAndCheckResult(testHarness, TableChange.builder().commitNum(2).build(), 1);
+      addEventAndCheckResult(testHarness, TableChange.builder().commitCount(2).build(), 1);
 
       // Create a lock to prevent execution, and check that there is no result
       assertThat(lock.tryLock()).isTrue();
-      addEventAndCheckResult(testHarness, TableChange.builder().commitNum(2).build(), 1);
+      addEventAndCheckResult(testHarness, TableChange.builder().commitCount(2).build(), 1);
       long currentTime = testHarness.getProcessingTime();
 
       // Remove the lock, and still no trigger
@@ -331,7 +395,7 @@ void testRecovery(boolean locked, boolean runningTask) throws Exception {
       ++processingTime;
       expected.add(Trigger.recovery(processingTime));
       testHarness.setProcessingTime(processingTime);
-      testHarness.processElement(TableChange.builder().commitNum(2).build(), processingTime);
+      testHarness.processElement(TableChange.builder().commitCount(2).build(), processingTime);
       assertTriggers(testHarness.extractOutputValues(), expected);
 
       // Nothing happens until the recovery is finished
@@ -347,7 +411,7 @@ void testRecovery(boolean locked, boolean runningTask) throws Exception {
       // Still no results as the recovery is ongoing
       ++processingTime;
       testHarness.setProcessingTime(processingTime);
-      testHarness.processElement(TableChange.builder().commitNum(2).build(), processingTime);
+      testHarness.processElement(TableChange.builder().commitCount(2).build(), processingTime);
       assertTriggers(testHarness.extractOutputValues(), expected);
 
       // Simulate the action of removing lock and recoveryLock by downstream lock cleaner when it
@@ -383,8 +447,8 @@ void testTriggerMetrics() throws Exception {
             lockFactory,
             Lists.newArrayList(NAME_1, NAME_2),
             Lists.newArrayList(
-                new TriggerEvaluator.Builder().commitNumber(2).build(),
-                new TriggerEvaluator.Builder().commitNumber(4).build()),
+                new TriggerEvaluator.Builder().commitCount(2).build(),
+                new TriggerEvaluator.Builder().commitCount(4).build()),
             1L,
             1L);
     source
@@ -400,7 +464,7 @@ void testTriggerMetrics() throws Exception {
       jobClient = env.executeAsync();
 
       // This one doesn't trigger - tests NOTHING_TO_TRIGGER
-      source.sendRecord(TableChange.builder().commitNum(1).build());
+      source.sendRecord(TableChange.builder().commitCount(1).build());
 
       Awaitility.await()
           .until(
@@ -412,7 +476,7 @@ void testTriggerMetrics() throws Exception {
               });
 
       // Trigger one of the tasks - tests TRIGGERED
-      source.sendRecord(TableChange.builder().commitNum(1).build());
+      source.sendRecord(TableChange.builder().commitCount(1).build());
       // Wait until we receive the trigger
       assertThat(sink.poll(Duration.ofSeconds(5))).isNotNull();
       assertThat(
@@ -421,7 +485,7 @@ void testTriggerMetrics() throws Exception {
       lock.unlock();
 
       // Trigger both of the tasks - tests TRIGGERED
-      source.sendRecord(TableChange.builder().commitNum(2).build());
+      source.sendRecord(TableChange.builder().commitCount(2).build());
       // Wait until we receive the trigger
       assertThat(sink.poll(Duration.ofSeconds(5))).isNotNull();
       lock.unlock();
@@ -472,14 +536,14 @@ void testRateLimiterMetrics() throws Exception {
       jobClient = env.executeAsync();
 
       // Start the first trigger
-      source.sendRecord(TableChange.builder().commitNum(2).build());
+      source.sendRecord(TableChange.builder().commitCount(2).build());
       assertThat(sink.poll(Duration.ofSeconds(5))).isNotNull();
 
       // Remove the lock to allow the next trigger
       lock.unlock();
 
       // The second trigger will be blocked
-      source.sendRecord(TableChange.builder().commitNum(2).build());
+      source.sendRecord(TableChange.builder().commitCount(2).build());
       Awaitility.await()
           .until(
               () ->
@@ -518,11 +582,11 @@ void testConcurrentRunMetrics() throws Exception {
       jobClient = env.executeAsync();
 
       // Start the first trigger - notice that we do not remove the lock after the trigger
-      source.sendRecord(TableChange.builder().commitNum(2).build());
+      source.sendRecord(TableChange.builder().commitCount(2).build());
       assertThat(sink.poll(Duration.ofSeconds(5))).isNotNull();
 
       // The second trigger will be blocked by the lock
-      source.sendRecord(TableChange.builder().commitNum(2).build());
+      source.sendRecord(TableChange.builder().commitCount(2).build());
       Awaitility.await()
           .until(
               () ->
@@ -589,13 +653,13 @@ private TriggerManager manager(
         tableLoader,
         lockFactory,
         Lists.newArrayList(NAME_1),
-        Lists.newArrayList(new TriggerEvaluator.Builder().commitNumber(2).build()),
+        Lists.newArrayList(new TriggerEvaluator.Builder().commitCount(2).build()),
         minFireDelayMs,
         lockCheckDelayMs);
   }
 
   private TriggerManager manager(TableLoader tableLoader) {
-    return manager(tableLoader, new TriggerEvaluator.Builder().commitNumber(2).build());
+    return manager(tableLoader, new TriggerEvaluator.Builder().commitCount(2).build());
   }
 
   private static void assertTriggers(List<Trigger> expected, List<Trigger> actual) {