[server] Lake tiering service report the max timestamp of fluss record

Author: LiebingYu

fluss-client/src/main/java/org/apache/fluss/client/metadata/LakeSnapshot.java

@@ -25,7 +25,7 @@
 
 /**
  * A class representing the lake snapshot information of a table. It contains:
- * <li>The snapshot id and the log offset for each bucket.
+ * <li>The snapshot id, the log offset and timestamp for each bucket.
  *
  * @since 0.3
  */
@@ -37,16 +37,21 @@ public class LakeSnapshot {
     // the specific log offset of the snapshot
     private final Map<TableBucket, Long> tableBucketsOffset;
 
+    // the specific timestamp of the snapshot
+    private final Map<TableBucket, Long> tableBucketsTimestamp;
+
     // the partition name by partition id of this lake snapshot if
     // is a partitioned table, empty if not a partitioned table
     private final Map<Long, String> partitionNameById;
 
     public LakeSnapshot(
             long snapshotId,
             Map<TableBucket, Long> tableBucketsOffset,
+            Map<TableBucket, Long> tableBucketsTimestamp,
             Map<Long, String> partitionNameById) {
         this.snapshotId = snapshotId;
         this.tableBucketsOffset = tableBucketsOffset;
+        this.tableBucketsTimestamp = tableBucketsTimestamp;
         this.partitionNameById = partitionNameById;
     }
 
@@ -58,6 +63,10 @@ public Map<TableBucket, Long> getTableBucketsOffset() {
         return Collections.unmodifiableMap(tableBucketsOffset);
     }
 
+    public Map<TableBucket, Long> getTableBucketsTimestamp() {
+        return Collections.unmodifiableMap(tableBucketsTimestamp);
+    }
+
     public Map<Long, String> getPartitionNameById() {
         return Collections.unmodifiableMap(partitionNameById);
     }
@@ -69,6 +78,8 @@ public String toString() {
                 + snapshotId
                 + ", tableBucketsOffset="
                 + tableBucketsOffset
+                + ", tableBucketsTimestamp="
+                + tableBucketsTimestamp
                 + ", partitionNameById="
                 + partitionNameById
                 + '}';

fluss-client/src/main/java/org/apache/fluss/client/utils/ClientRpcMessageUtils.java

@@ -202,6 +202,8 @@ public static LakeSnapshot toLakeTableSnapshotInfo(GetLatestLakeSnapshotResponse
         long snapshotId = response.getSnapshotId();
         Map<TableBucket, Long> tableBucketsOffset =
                 new HashMap<>(response.getBucketSnapshotsCount());
+        Map<TableBucket, Long> tableBucketsTimestamp =
+                new HashMap<>(response.getBucketSnapshotsCount());
         Map<Long, String> partitionNameById = new HashMap<>();
         for (PbLakeSnapshotForBucket pbLakeSnapshotForBucket : response.getBucketSnapshotsList()) {
             Long partitionId =
@@ -214,8 +216,12 @@ public static LakeSnapshot toLakeTableSnapshotInfo(GetLatestLakeSnapshotResponse
                 partitionNameById.put(partitionId, pbLakeSnapshotForBucket.getPartitionName());
             }
             tableBucketsOffset.put(tableBucket, pbLakeSnapshotForBucket.getLogOffset());
+            if (pbLakeSnapshotForBucket.hasTimestamp()) {
+                tableBucketsTimestamp.put(tableBucket, pbLakeSnapshotForBucket.getTimestamp());
+            }
         }
-        return new LakeSnapshot(snapshotId, tableBucketsOffset, partitionNameById);
+        return new LakeSnapshot(
+                snapshotId, tableBucketsOffset, tableBucketsTimestamp, partitionNameById);
     }
 
     public static List<FsPathAndFileName> toFsPathAndFileName(

fluss-common/src/main/java/org/apache/fluss/lake/committer/BucketOffset.java

@@ -22,23 +22,29 @@
 import java.io.Serializable;
 import java.util.Objects;
 
-/** The bucket offset information to be expected to be stored in Lake's snapshot property. */
+/**
+ * The bucket offset and timestamp information to be expected to be stored in Lake's snapshot
+ * property.
+ */
 public class BucketOffset implements Serializable {
 
     private static final long serialVersionUID = 1L;
     public static final String FLUSS_LAKE_SNAP_BUCKET_OFFSET_PROPERTY = "fluss-offsets";
 
     private final long logOffset;
+    private final long timestamp;
     private final int bucket;
     private final @Nullable Long partitionId;
     private final @Nullable String partitionQualifiedName;
 
     public BucketOffset(
             long logOffset,
+            long timestamp,
             int bucket,
             @Nullable Long partitionId,
             @Nullable String partitionQualifiedName) {
         this.logOffset = logOffset;
+        this.timestamp = timestamp;
         this.bucket = bucket;
         this.partitionId = partitionId;
         this.partitionQualifiedName = partitionQualifiedName;
@@ -48,6 +54,10 @@ public long getLogOffset() {
         return logOffset;
     }
 
+    public long getTimestamp() {
+        return timestamp;
+    }
+
     public int getBucket() {
         return bucket;
     }
@@ -62,6 +72,11 @@ public String getPartitionQualifiedName() {
         return partitionQualifiedName;
     }
 
+    @Override
+    public int hashCode() {
+        return Objects.hash(logOffset, timestamp, bucket, partitionId, partitionQualifiedName);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -73,6 +88,7 @@ public boolean equals(Object o) {
         BucketOffset that = (BucketOffset) o;
         return bucket == that.bucket
                 && logOffset == that.logOffset
+                && timestamp == that.timestamp
                 && Objects.equals(partitionId, that.partitionId)
                 && Objects.equals(partitionQualifiedName, that.partitionQualifiedName);
     }

fluss-common/src/main/java/org/apache/fluss/lake/committer/CommittedLakeSnapshot.java

@@ -22,17 +22,21 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 
 /**
- * The lake already committed snapshot, containing the lake snapshot id and the bucket end offsets
- * in this snapshot.
+ * The lake already committed snapshot, containing the lake snapshot id, the bucket end offsets and
+ * max timestamp in this snapshot.
  */
 public class CommittedLakeSnapshot {
 
     private final long lakeSnapshotId;
     // <partition_id, bucket> -> log offset, partition_id will be null if it's not a
     // partition bucket
     private final Map<Tuple2<Long, Integer>, Long> logEndOffsets = new HashMap<>();
+    // <partition_id, bucket> -> max timestamp, partition_id will be null if it's not a
+    // partition bucket
+    private final Map<Tuple2<Long, Integer>, Long> maxTimestamps = new HashMap<>();
 
     // partition id -> partition name, will be empty if is not a partitioned table
     // the partition name is a qualified name in the format: key1=value1/key2=value2
@@ -46,18 +50,32 @@ public long getLakeSnapshotId() {
         return lakeSnapshotId;
     }
 
-    public void addBucket(int bucketId, long offset) {
+    public void addBucket(int bucketId, long offset, long timestamp) {
         logEndOffsets.put(Tuple2.of(null, bucketId), offset);
+        maxTimestamps.put(Tuple2.of(null, bucketId), timestamp);
     }
 
     public void addPartitionBucket(
-            Long partitionId, String partitionQualifiedName, int bucketId, long offset) {
+            Long partitionId,
+            String partitionQualifiedName,
+            int bucketId,
+            long offset,
+            long timestamp) {
         logEndOffsets.put(Tuple2.of(partitionId, bucketId), offset);
+        maxTimestamps.put(Tuple2.of(partitionId, bucketId), timestamp);
         qualifiedPartitionNameById.put(partitionId, partitionQualifiedName);
     }
 
-    public Map<Tuple2<Long, Integer>, Long> getLogEndOffsets() {
-        return logEndOffsets;
+    public Set<Tuple2<Long, Integer>> tablePartitionBuckets() {
+        return logEndOffsets.keySet();
+    }
+
+    public Long getLogEndOffset(Tuple2<Long, Integer> partitionBucket) {
+        return logEndOffsets.get(partitionBucket);
+    }
+
+    public Long getMaxTimestamp(Tuple2<Long, Integer> partitionBucket) {
+        return maxTimestamps.get(partitionBucket);
     }
 
     public Map<Long, String> getQualifiedPartitionNameById() {
@@ -72,12 +90,14 @@ public boolean equals(Object o) {
         CommittedLakeSnapshot that = (CommittedLakeSnapshot) o;
         return lakeSnapshotId == that.lakeSnapshotId
                 && Objects.equals(logEndOffsets, that.logEndOffsets)
+                && Objects.equals(maxTimestamps, that.maxTimestamps)
                 && Objects.equals(qualifiedPartitionNameById, that.qualifiedPartitionNameById);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(lakeSnapshotId, logEndOffsets, qualifiedPartitionNameById);
+        return Objects.hash(
+                lakeSnapshotId, logEndOffsets, maxTimestamps, qualifiedPartitionNameById);
     }
 
     @Override
@@ -87,6 +107,8 @@ public String toString() {
                 + lakeSnapshotId
                 + ", logEndOffsets="
                 + logEndOffsets
+                + ", maxTimestamps="
+                + maxTimestamps
                 + ", partitionNameById="
                 + qualifiedPartitionNameById
                 + '}';

fluss-common/src/main/java/org/apache/fluss/utils/json/BucketOffsetJsonSerde.java

@@ -32,6 +32,7 @@ public class BucketOffsetJsonSerde
     private static final String BUCKET_ID = "bucket_id";
     private static final String PARTITION_NAME = "partition_name";
     private static final String LOG_OFFSET = "log_offset";
+    private static final String TIMESTAMP = "timestamp";
 
     @Override
     public BucketOffset deserialize(JsonNode node) {
@@ -46,7 +47,10 @@ public BucketOffset deserialize(JsonNode node) {
         // deserialize log offset
         long logOffset = node.get(LOG_OFFSET).asLong();
 
-        return new BucketOffset(logOffset, bucketId, partitionId, partitionName);
+        // deserialize timestamp
+        long timestamp = node.get(TIMESTAMP).asLong();
+
+        return new BucketOffset(logOffset, timestamp, bucketId, partitionId, partitionName);
     }
 
     @Override
@@ -67,6 +71,9 @@ public void serialize(BucketOffset bucketOffset, JsonGenerator generator) throws
         // serialize bucket offset
         generator.writeNumberField(LOG_OFFSET, bucketOffset.getLogOffset());
 
+        // serialize bucket timestamp
+        generator.writeNumberField(TIMESTAMP, bucketOffset.getTimestamp());
+
         generator.writeEndObject();
     }
 }

fluss-common/src/test/java/org/apache/fluss/utils/json/BucketOffsetJsonSerdeTest.java

@@ -29,16 +29,16 @@ public class BucketOffsetJsonSerdeTest extends JsonSerdeTestBase<BucketOffset> {
     @Override
     protected BucketOffset[] createObjects() {
         return new BucketOffset[] {
-            new BucketOffset(10, 1, 1L, "country=eu-central/year=2023/month=12"),
-            new BucketOffset(20, 2, null, null)
+            new BucketOffset(10, 100, 1, 1L, "country=eu-central/year=2023/month=12"),
+            new BucketOffset(20, 200, 2, null, null)
         };
     }
 
     @Override
     protected String[] expectedJsons() {
         return new String[] {
-            "{\"partition_id\":1,\"bucket_id\":1,\"partition_name\":\"country=eu-central/year=2023/month=12\",\"log_offset\":10}",
-            "{\"bucket_id\":2,\"log_offset\":20}"
+            "{\"partition_id\":1,\"bucket_id\":1,\"partition_name\":\"country=eu-central/year=2023/month=12\",\"log_offset\":10,\"timestamp\":100}",
+            "{\"bucket_id\":2,\"log_offset\":20,\"timestamp\":200}"
         };
     }
 }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshot.java

@@ -22,6 +22,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 
 /** A lake snapshot for a Fluss table. */
 class FlussTableLakeSnapshot {
@@ -34,10 +35,15 @@ class FlussTableLakeSnapshot {
     // if the bucket is not of a partition, the partition_name is null
     private final Map<Tuple2<TableBucket, String>, Long> logEndOffsets;
 
+    // <table_bucket, partition_name> -> max timestamps,
+    // if the bucket is not of a partition, the partition_name is null
+    private final Map<Tuple2<TableBucket, String>, Long> maxTimestamps;
+
     FlussTableLakeSnapshot(long tableId, long lakeSnapshotId) {
         this.tableId = tableId;
         this.lakeSnapshotId = lakeSnapshotId;
         this.logEndOffsets = new HashMap<>();
+        this.maxTimestamps = new HashMap<>();
     }
 
     public long tableId() {
@@ -48,16 +54,27 @@ public long lakeSnapshotId() {
         return lakeSnapshotId;
     }
 
-    public Map<Tuple2<TableBucket, String>, Long> logEndOffsets() {
-        return logEndOffsets;
+    public Set<Tuple2<TableBucket, String>> tablePartitionBuckets() {
+        return logEndOffsets.keySet();
     }
 
-    public void addBucketOffset(TableBucket bucket, long offset) {
+    public void addBucketOffsetAndTimestamp(TableBucket bucket, long offset, long timestamp) {
         logEndOffsets.put(Tuple2.of(bucket, null), offset);
+        maxTimestamps.put(Tuple2.of(bucket, null), timestamp);
     }
 
-    public void addPartitionBucketOffset(TableBucket bucket, String partitionName, long offset) {
+    public void addPartitionBucketOffsetAndTimestamp(
+            TableBucket bucket, String partitionName, long offset, long timestamp) {
         logEndOffsets.put(Tuple2.of(bucket, partitionName), offset);
+        maxTimestamps.put(Tuple2.of(bucket, partitionName), timestamp);
+    }
+
+    public long getLogEndOffset(Tuple2<TableBucket, String> bucketPartition) {
+        return logEndOffsets.get(bucketPartition);
+    }
+
+    public long getMaxTimestamp(Tuple2<TableBucket, String> bucketPartition) {
+        return maxTimestamps.get(bucketPartition);
     }
 
     @Override
@@ -69,6 +86,8 @@ public String toString() {
                 + lakeSnapshotId
                 + ", logEndOffsets="
                 + logEndOffsets
+                + ", maxTimestamps="
+                + maxTimestamps
                 + '}';
     }
 }