expire since latest_snapshot or current_timestamp

ams/server/src/main/java/com/netease/arctic/server/optimizing/maintainer/IcebergTableMaintainer.java

@@ -21,9 +21,11 @@
 import static org.apache.iceberg.relocated.com.google.common.primitives.Longs.min;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Optional;
 import com.google.common.base.Strings;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Maps;
+import com.netease.arctic.ams.api.CommitMetaProducer;
 import com.netease.arctic.io.ArcticFileIO;
 import com.netease.arctic.io.PathInfo;
 import com.netease.arctic.io.SupportsFileSystemOperations;
@@ -156,8 +158,17 @@ public void expireData(TableRuntime tableRuntime) {
       if (!expirationConfig.isValid(field, table.name())) {
         return;
       }
+      Instant startInstant;
+      if (expirationConfig.getSince() == DataExpirationConfig.Since.CURRENT_TIMESTAMP) {
+        startInstant = Instant.now().atZone(getDefaultZoneId(field)).toInstant();
+      } else {
+        startInstant =
+            Instant.ofEpochMilli(fetchLatestNonOptimizedSnapshotTime(table))
+                .atZone(getDefaultZoneId(field))
+                .toInstant();
+      }
 
-      expireDataFrom(expirationConfig, Instant.now().atZone(getDefaultZoneId(field)).toInstant());
+      expireDataFrom(expirationConfig, startInstant);
     } catch (Throwable t) {
       LOG.error("Unexpected purge error for table {} ", tableRuntime.getTableIdentifier(), t);
     }
@@ -265,8 +276,10 @@ protected long olderThanSnapshotNeedToExpire(long mustOlderThan) {
     // Latest checkpoint of flink need retain. If Flink does not continuously commit new snapshots,
     // it can lead to issues with table partitions not expiring.
     long latestFlinkCommitTime = fetchLatestFlinkCommittedSnapshotTime(table);
+    // Retain the latest non-optimized snapshot for remember the real latest update
+    long latestNonOptimizedTime = fetchLatestNonOptimizedSnapshotTime(table);
     long olderThan = System.currentTimeMillis() - baseSnapshotsKeepTime;
-    return min(latestFlinkCommitTime, mustOlderThan, olderThan);
+    return min(latestFlinkCommitTime, latestNonOptimizedTime, mustOlderThan, olderThan);
   }
 
   protected Set<String> expireSnapshotNeedToExcludeFiles() {
@@ -388,6 +401,20 @@ public static long fetchOptimizingSnapshotTime(Table table, TableRuntime tableRu
     return Long.MAX_VALUE;
   }
 
+  /**
+   * When expiring historic data and `data-expire.since` is `CURRENT_SNAPSHOT`, the latest snapshot
+   * should not be produced by Amoro.
+   *
+   * @param table iceberg table
+   * @return the latest non-optimized snapshot timestamp
+   */
+  public static long fetchLatestNonOptimizedSnapshotTime(Table table) {
+    Optional<Snapshot> snapshot =
+        IcebergTableUtil.findSnapshot(
+            table, s -> s.summary().containsValue(CommitMetaProducer.OPTIMIZE.name()));
+    return snapshot.isPresent() ? snapshot.get().timestampMillis() : Long.MAX_VALUE;
+  }
+
   private static int deleteInvalidFilesInFs(
       SupportsFileSystemOperations fio, String location, long lastTime, Set<String> excludes) {
     if (!fio.exists(location)) {

ams/server/src/main/java/com/netease/arctic/server/optimizing/maintainer/MixedTableMaintainer.java

@@ -145,8 +145,22 @@ public void expireData(TableRuntime tableRuntime) {
         return;
       }
       ZoneId defaultZone = IcebergTableMaintainer.getDefaultZoneId(field);
-      Instant instant = Instant.now().atZone(defaultZone).toInstant();
-      expireDataFrom(expirationConfig, instant);
+      Instant startInstant;
+      if (expirationConfig.getSince() == DataExpirationConfig.Since.CURRENT_TIMESTAMP) {
+        startInstant = Instant.now().atZone(defaultZone).toInstant();
+      } else {
+        long latestBaseTs =
+            IcebergTableMaintainer.fetchLatestNonOptimizedSnapshotTime(baseMaintainer.getTable());
+        long latestChangeTs =
+            changeMaintainer == null
+                ? Long.MAX_VALUE
+                : IcebergTableMaintainer.fetchLatestNonOptimizedSnapshotTime(
+                    changeMaintainer.getTable());
+        long latestNonOptimizedTs = Longs.min(latestChangeTs, latestBaseTs);
+
+        startInstant = Instant.ofEpochMilli(latestNonOptimizedTs).atZone(defaultZone).toInstant();
+      }
+      expireDataFrom(expirationConfig, startInstant);
     } catch (Throwable t) {
       LOG.error("Unexpected purge error for table {} ", tableRuntime.getTableIdentifier(), t);
     }

ams/server/src/main/java/com/netease/arctic/server/table/DataExpirationConfig.java

@@ -33,6 +33,8 @@ public class DataExpirationConfig {
   private String dateTimePattern;
   // data-expire.datetime-number-format
   private String numberDateFormat;
+  // data-expire.since
+  private Since since;
 
   @VisibleForTesting
   public enum ExpireLevel {
@@ -49,6 +51,22 @@ public static ExpireLevel fromString(String level) {
     }
   }
 
+  @VisibleForTesting
+  public enum Since {
+    LATEST_SNAPSHOT,
+    CURRENT_TIMESTAMP;
+
+    public static Since fromString(String since) {
+      Preconditions.checkArgument(null != since, "data-expire.since is invalid: null");
+      try {
+        return Since.valueOf(since.toUpperCase(Locale.ENGLISH));
+      } catch (IllegalArgumentException e) {
+        throw new IllegalArgumentException(
+            String.format("Unable to expire data since: %s", since), e);
+      }
+    }
+  }
+
   public static final Set<Type.TypeID> FIELD_TYPES =
       Sets.newHashSet(Type.TypeID.TIMESTAMP, Type.TypeID.STRING, Type.TypeID.LONG);
 
@@ -62,13 +80,15 @@ public DataExpirationConfig(
       ExpireLevel expirationLevel,
       long retentionTime,
       String dateTimePattern,
-      String numberDateFormat) {
+      String numberDateFormat,
+      Since since) {
     this.enabled = enabled;
     this.expirationField = expirationField;
     this.expirationLevel = expirationLevel;
     this.retentionTime = retentionTime;
     this.dateTimePattern = dateTimePattern;
     this.numberDateFormat = numberDateFormat;
+    this.since = since;
   }
 
   public DataExpirationConfig(ArcticTable table) {
@@ -113,6 +133,12 @@ public DataExpirationConfig(ArcticTable table) {
             properties,
             TableProperties.DATA_EXPIRATION_DATE_NUMBER_FORMAT,
             TableProperties.DATA_EXPIRATION_DATE_NUMBER_FORMAT_DEFAULT);
+    since =
+        Since.fromString(
+            CompatiblePropertyUtil.propertyAsString(
+                properties,
+                TableProperties.DATA_EXPIRATION_SINCE,
+                TableProperties.DATA_EXPIRATION_SINCE_DEFAULT));
   }
 
   public static DataExpirationConfig parse(Map<String, String> properties) {
@@ -141,7 +167,13 @@ public static DataExpirationConfig parse(Map<String, String> properties) {
                 CompatiblePropertyUtil.propertyAsString(
                     properties,
                     TableProperties.DATA_EXPIRATION_DATE_NUMBER_FORMAT,
-                    TableProperties.DATA_EXPIRATION_DATE_NUMBER_FORMAT_DEFAULT));
+                    TableProperties.DATA_EXPIRATION_DATE_NUMBER_FORMAT_DEFAULT))
+            .setSince(
+                Since.fromString(
+                    CompatiblePropertyUtil.propertyAsString(
+                        properties,
+                        TableProperties.DATA_EXPIRATION_SINCE,
+                        TableProperties.DATA_EXPIRATION_SINCE_DEFAULT)));
     String retention =
         CompatiblePropertyUtil.propertyAsString(
             properties, TableProperties.DATA_EXPIRATION_RETENTION_TIME, null);
@@ -206,6 +238,15 @@ public DataExpirationConfig setNumberDateFormat(String numberDateFormat) {
     return this;
   }
 
+  public Since getSince() {
+    return since;
+  }
+
+  public DataExpirationConfig setSince(Since since) {
+    this.since = since;
+    return this;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -220,7 +261,8 @@ public boolean equals(Object o) {
         && Objects.equal(expirationField, config.expirationField)
         && expirationLevel == config.expirationLevel
         && Objects.equal(dateTimePattern, config.dateTimePattern)
-        && Objects.equal(numberDateFormat, config.numberDateFormat);
+        && Objects.equal(numberDateFormat, config.numberDateFormat)
+        && since == config.since;
   }
 
   @Override
@@ -231,7 +273,8 @@ public int hashCode() {
         expirationLevel,
         retentionTime,
         dateTimePattern,
-        numberDateFormat);
+        numberDateFormat,
+        since);
   }
 
   public boolean isValid(Types.NestedField field, String name) {

ams/server/src/main/java/com/netease/arctic/server/utils/IcebergTableUtil.java

@@ -18,6 +18,9 @@
 
 package com.netease.arctic.server.utils;
 
+import com.google.common.base.Optional;
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterables;
 import com.netease.arctic.IcebergFileEntry;
 import com.netease.arctic.scan.TableEntriesScan;
 import com.netease.arctic.server.ArcticServiceConstants;
@@ -75,6 +78,11 @@ public static Snapshot getSnapshot(Table table, boolean refresh) {
     return table.currentSnapshot();
   }
 
+  public static Optional<Snapshot> findSnapshot(Table table, Predicate<Snapshot> predicate) {
+    Iterable<Snapshot> snapshots = table.snapshots();
+    return Iterables.tryFind(snapshots, predicate);
+  }
+
   public static Set<String> getAllContentFilePath(Table internalTable) {
     Set<String> validFilesPath = new HashSet<>();
 

core/src/main/java/com/netease/arctic/table/TableProperties.java

@@ -158,6 +158,8 @@ private TableProperties() {}
       "data-expire.datetime-number-format";
   public static final String DATA_EXPIRATION_DATE_NUMBER_FORMAT_DEFAULT = "TIMESTAMP_MS";
   public static final String DATA_EXPIRATION_RETENTION_TIME = "data-expire.retention-time";
+  public static final String DATA_EXPIRATION_SINCE = "data-expire.since";
+  public static final String DATA_EXPIRATION_SINCE_DEFAULT = "LATEST_SNAPSHOT";
 
   public static final String ENABLE_DANGLING_DELETE_FILES_CLEAN =
       "clean-dangling-delete-files.enabled";

docs/user-guides/configurations.md

@@ -47,20 +47,21 @@ Self-optimizing configurations are applicable to both Iceberg Format and Mixed s
 
 Data-cleaning configurations are applicable to both Iceberg Format and Mixed streaming Format.
 
-| Key                                         | Default       | Description                                                                                                 |
-|---------------------------------------------|---------------|-------------------------------------------------------------------------------------------------------------|
-| table-expire.enabled                        | true          | Enables periodically expire table                                                                           |
-| change.data.ttl.minutes                     | 10080(7 days) | Time to live in minutes for data of ChangeStore                                                             |
-| snapshot.base.keep.minutes                  | 720(12 hours) | Table-Expiration keeps the latest snapshots of BaseStore within a specified time in minutes                 |
-| clean-orphan-file.enabled                   | false         | Enables periodically clean orphan files                                                                     |
-| clean-orphan-file.min-existing-time-minutes | 2880(2 days)  | Cleaning orphan files keeps the files modified within a specified time in minutes                           |
-| clean-dangling-delete-files.enabled         | true          | Whether to enable cleaning of dangling delete files                                                         |
-| data-expire.enabled                         | false         | Whether to enable data expiration                                                                           |
-| data-expire.level                           | partition     | Level of data expiration. Including partition and file                                                      |
-| data-expire.field                           | NULL          | Field used to determine data expiration, supporting timestamp/timestampz/long type and string type field in date format |
-| data-expire.datetime-string-pattern         | yyyy-MM-dd    | Pattern used for matching string datetime |
-| data-expire.datetime-number-format          | TIMESTAMP_MS  | Timestamp unit for long field. Including TIMESTAMP_MS and TIMESTAMP_S |
-| data-expire.retention-time                  | NULL          | Retention period for data expiration. For example, 1d means retaining data for 1 day. Other supported units include h (hour), min (minute), s (second), ms (millisecond), etc. |
+| Key                                         | Default         | Description                                                                                                                                                                                                                                               |
+|---------------------------------------------|-----------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| table-expire.enabled                        | true            | Enables periodically expire table                                                                                                                                                                                                                         |
+| change.data.ttl.minutes                     | 10080(7 days)   | Time to live in minutes for data of ChangeStore                                                                                                                                                                                                           |
+| snapshot.base.keep.minutes                  | 720(12 hours)   | Table-Expiration keeps the latest snapshots of BaseStore within a specified time in minutes                                                                                                                                                               |
+| clean-orphan-file.enabled                   | false           | Enables periodically clean orphan files                                                                                                                                                                                                                   |
+| clean-orphan-file.min-existing-time-minutes | 2880(2 days)    | Cleaning orphan files keeps the files modified within a specified time in minutes                                                                                                                                                                         |
+| clean-dangling-delete-files.enabled         | true            | Whether to enable cleaning of dangling delete files                                                                                                                                                                                                       |
+| data-expire.enabled                         | false           | Whether to enable data expiration                                                                                                                                                                                                                         |
+| data-expire.level                           | partition       | Level of data expiration. Including partition and file                                                                                                                                                                                                    |
+| data-expire.field                           | NULL            | Field used to determine data expiration, supporting timestamp/timestampz/long type and string type field in date format                                                                                                                                   |
+| data-expire.datetime-string-pattern         | yyyy-MM-dd      | Pattern used for matching string datetime                                                                                                                                                                                                                 |
+| data-expire.datetime-number-format          | TIMESTAMP_MS    | Timestamp unit for long field. Including TIMESTAMP_MS and TIMESTAMP_S                                                                                                                                                                                     |
+| data-expire.retention-time                  | NULL            | Retention period for data expiration. For example, 1d means retaining data for 1 day. Other supported units include h (hour), min (minute), s (second), ms (millisecond), etc.                                                                            |
+| data-expire.since                           | LATEST_SNAPSHOT | A event to indicate when start expire data. Including LATEST_SNAPSHOT and CURRENT_TIMESTAMP. LATEST_SNAPSHOT uses the timestamp of latest **non-optimized** snapshot as the start of the expiration, which ensures that the table has retention-time data |
 
 ## Mixed Format configurations
 

docs/user-guides/using-tables.md

@@ -155,7 +155,7 @@ ALTER TABLE test_db.test_log_store set tblproperties (
     'data-expire.enabled' = 'true');
 ```
 
-### Set the data retention period
+### Set retention period
 
 The configuration for data retention duration consists of a number and a unit. For example, '90d' represents retaining data for 90 days, and '12h' indicates 12 hours.
 
@@ -164,7 +164,7 @@ ALTER TABLE test_db.test_log_store set tblproperties (
     'data-expire.retention-time' = '90d');
 ```
 
-### Select the event-time field
+### Select expiration field
 
 Data expiration requires users to specify a field for determining expiration. 
 In addition to supporting timestampz/timestamp field types for this purpose, it also supports string and long field type. 
@@ -186,7 +186,7 @@ ALTER TABLE test_db.test_log_store set tblproperties (
     'data-expire.datetime-number-format' = 'TIMESTAMP_MS');
 ```
 
-### Adjust the data expiration level
+### Adjust expiration level
 
 Data expiration supports two levels, including `PARTITION` and `FILE`. The default level is `PARTITION`, which means that AMS deletes files only when all the files within a partition have expired.
 
@@ -195,6 +195,15 @@ ALTER TABLE test_db.test_log_store set tblproperties (
     'data-expire.level' = 'partition');
 ```
 
+### Specify start time
+
+Amoro expire data since `CURRENT_SNAPSHOT` or `CURRENT_TIMESTAMP`. `CURRENT_SNAPSHOT` will follow the timestamp of the table's most recent snapshot as the start time of the expiration, which ensures that the table has `data-expire.retention-time` data; while `CURRENT_TIMESTAMP` will follow the current time of the service.
+
+```sql
+ALTER TABLE test_db.test_log_store set tblproperties (
+    'data-expire.since' = 'current_timestamp');
+```
+
 ## Delete table
 
 After logging into the AMS Dashboard. To modify a table, enter the modification statement in the `terminal` and execute it.