[lake] Lake tiering service report the max timestamp of fluss record (#1700)

Author: LiebingYu

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
                 + '}';
     }
 }

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

@@ -88,7 +88,12 @@ public void commit(long tableId, CommittedLakeSnapshot committedLakeSnapshot)
             Long partitionId = partitionBucket.f0;
             if (partitionId == null) {
                 tableBucket = new TableBucket(tableId, partitionBucket.f1);
-                flussTableLakeSnapshot.addBucketOffset(tableBucket, entry.getValue());
+                // we use -1 since we don't store timestamp in lake snapshot property for
+                // simplicity, it may cause the timestamp to be -1 during constructing lake
+                // snapshot to commit to Fluss.
+                // But it should happen rarely and should be a normal value after next tiering.
+                flussTableLakeSnapshot.addBucketOffsetAndTimestamp(
+                        tableBucket, entry.getValue(), -1);
             } else {
                 tableBucket = new TableBucket(tableId, partitionId, partitionBucket.f1);
                 // the partition name is qualified partition name in format:
@@ -98,8 +103,11 @@ public void commit(long tableId, CommittedLakeSnapshot committedLakeSnapshot)
                         committedLakeSnapshot.getQualifiedPartitionNameById().get(partitionId);
                 ResolvedPartitionSpec resolvedPartitionSpec =
                         ResolvedPartitionSpec.fromPartitionQualifiedName(qualifiedPartitionName);
-                flussTableLakeSnapshot.addPartitionBucketOffset(
-                        tableBucket, resolvedPartitionSpec.getPartitionName(), entry.getValue());
+                flussTableLakeSnapshot.addPartitionBucketOffsetAndTimestamp(
+                        tableBucket,
+                        resolvedPartitionSpec.getPartitionName(),
+                        entry.getValue(),
+                        -1);
             }
         }
         commit(flussTableLakeSnapshot);
@@ -114,13 +122,14 @@ private CommitLakeTableSnapshotRequest toCommitLakeTableSnapshotRequest(
 
         pbLakeTableSnapshotInfo.setTableId(flussTableLakeSnapshot.tableId());
         pbLakeTableSnapshotInfo.setSnapshotId(flussTableLakeSnapshot.lakeSnapshotId());
-        for (Map.Entry<Tuple2<TableBucket, String>, Long> bucketEndOffsetEntry :
-                flussTableLakeSnapshot.logEndOffsets().entrySet()) {
+        for (Tuple2<TableBucket, String> bucketPartition :
+                flussTableLakeSnapshot.tablePartitionBuckets()) {
             PbLakeTableOffsetForBucket pbLakeTableOffsetForBucket =
                     pbLakeTableSnapshotInfo.addBucketsReq();
-            TableBucket tableBucket = bucketEndOffsetEntry.getKey().f0;
-            String partitionName = bucketEndOffsetEntry.getKey().f1;
-            long endOffset = bucketEndOffsetEntry.getValue();
+            TableBucket tableBucket = bucketPartition.f0;
+            String partitionName = bucketPartition.f1;
+            long endOffset = flussTableLakeSnapshot.getLogEndOffset(bucketPartition);
+            long maxTimestamp = flussTableLakeSnapshot.getMaxTimestamp(bucketPartition);
             if (tableBucket.getPartitionId() != null) {
                 pbLakeTableOffsetForBucket.setPartitionId(tableBucket.getPartitionId());
             }
@@ -129,6 +138,7 @@ private CommitLakeTableSnapshotRequest toCommitLakeTableSnapshotRequest(
             }
             pbLakeTableOffsetForBucket.setBucketId(tableBucket.getBucket());
             pbLakeTableOffsetForBucket.setLogEndOffset(endOffset);
+            pbLakeTableOffsetForBucket.setMaxTimestamp(maxTimestamp);
         }
         return commitLakeTableSnapshotRequest;
     }

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

@@ -229,10 +229,14 @@ private Committable commitWriteResults(
             for (TableBucketWriteResult<WriteResult> writeResult : committableWriteResults) {
                 TableBucket tableBucket = writeResult.tableBucket();
                 if (writeResult.tableBucket().getPartitionId() == null) {
-                    flussTableLakeSnapshot.addBucketOffset(tableBucket, writeResult.logEndOffset());
+                    flussTableLakeSnapshot.addBucketOffsetAndTimestamp(
+                            tableBucket, writeResult.logEndOffset(), writeResult.maxTimestamp());
                 } else {
-                    flussTableLakeSnapshot.addPartitionBucketOffset(
-                            tableBucket, writeResult.partitionName(), writeResult.logEndOffset());
+                    flussTableLakeSnapshot.addPartitionBucketOffsetAndTimestamp(
+                            tableBucket,
+                            writeResult.partitionName(),
+                            writeResult.logEndOffset(),
+                            writeResult.maxTimestamp());
                 }
             }
             flussTableLakeSnapshotCommitter.commit(flussTableLakeSnapshot);

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TableBucketWriteResult.java

@@ -42,12 +42,16 @@ public class TableBucketWriteResult<WriteResult> implements Serializable {
     // null when the bucket is not for a partition
     @Nullable private final String partitionName;
 
-    // will be null when no any data write, such as for tiering a empty log split
+    // will be null when no any data write, such as for tiering an empty log split
     @Nullable private final WriteResult writeResult;
 
     // the end offset of tiering, should be the last tiered record's offset + 1
     private final long logEndOffset;
 
+    // the max timestamp of tiering, should be the last tiered record's timestamp,
+    // will be -1 for empty log splits or snapshot splits
+    private final long maxTimestamp;
+
     // the total number of write results in one round of tiering,
     // used for downstream commiter operator to determine when all write results
     // for the round of tiering is finished
@@ -59,12 +63,14 @@ public TableBucketWriteResult(
             @Nullable String partitionName,
             @Nullable WriteResult writeResult,
             long logEndOffset,
+            long maxTimestamp,
             int numberOfWriteResults) {
         this.tablePath = tablePath;
         this.tableBucket = tableBucket;
         this.partitionName = partitionName;
         this.writeResult = writeResult;
         this.logEndOffset = logEndOffset;
+        this.maxTimestamp = maxTimestamp;
         this.numberOfWriteResults = numberOfWriteResults;
     }
 
@@ -93,4 +99,8 @@ public int numberOfWriteResults() {
     public long logEndOffset() {
         return logEndOffset;
     }
+
+    public long maxTimestamp() {
+        return maxTimestamp;
+    }
 }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TableBucketWriteResultSerializer.java

@@ -85,6 +85,9 @@ public byte[] serialize(TableBucketWriteResult<WriteResult> tableBucketWriteResu
         // serialize log end offset
         out.writeLong(tableBucketWriteResult.logEndOffset());
 
+        // serialize max timestamp
+        out.writeLong(tableBucketWriteResult.maxTimestamp());
+
         // serialize number of write results
         out.writeInt(tableBucketWriteResult.numberOfWriteResults());
 
@@ -129,6 +132,8 @@ public TableBucketWriteResult<WriteResult> deserialize(int version, byte[] seria
 
         // deserialize log end offset
         long logEndOffset = in.readLong();
+        // deserialize max timestamp
+        long maxTimestamp = in.readLong();
         // deserialize number of write results
         int numberOfWriteResults = in.readInt();
         return new TableBucketWriteResult<>(
@@ -137,6 +142,7 @@ public TableBucketWriteResult<WriteResult> deserialize(int version, byte[] seria
                 partitionName,
                 writeResult,
                 logEndOffset,
+                maxTimestamp,
                 numberOfWriteResults);
     }
 }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java

@@ -68,6 +68,9 @@ public class TieringSplitReader<WriteResult>
 
     private static final Duration POLL_TIMEOUT = Duration.ofMillis(10000L);
 
+    // unknown bucket timestamp for empty split or snapshot split
+    private static final long UNKNOWN_BUCKET_TIMESTAMP = -1;
+
     private final LakeTieringFactory<WriteResult, ?> lakeTieringFactory;
 
     // the id for the pending tables to be tiered
@@ -285,7 +288,10 @@ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> forLogRecords(
                 writeResults.put(
                         bucket,
                         completeLakeWriter(
-                                bucket, currentTieringSplit.getPartitionName(), stoppingOffset));
+                                bucket,
+                                currentTieringSplit.getPartitionName(),
+                                stoppingOffset,
+                                lastRecord.timestamp()));
                 // put split of the bucket
                 finishedSplitIds.put(bucket, currentSplitId);
                 LOG.info("Split {} has been finished.", currentSplitId);
@@ -316,7 +322,10 @@ private LakeWriter<WriteResult> getOrCreateLakeWriter(
     }
 
     private TableBucketWriteResult<WriteResult> completeLakeWriter(
-            TableBucket bucket, @Nullable String partitionName, long logEndOffset)
+            TableBucket bucket,
+            @Nullable String partitionName,
+            long logEndOffset,
+            long maxTimestamp)
             throws IOException {
         LakeWriter<WriteResult> lakeWriter = lakeWriters.remove(bucket);
         WriteResult writeResult = lakeWriter.complete();
@@ -327,6 +336,7 @@ private TableBucketWriteResult<WriteResult> completeLakeWriter(
                 partitionName,
                 writeResult,
                 logEndOffset,
+                maxTimestamp,
                 checkNotNull(currentTableNumberOfSplits));
     }
 
@@ -344,6 +354,7 @@ private TableBucketWriteResultWithSplitIds forEmptySplits(Set<TieringLogSplit> e
                             logSplit.getPartitionName(),
                             null,
                             logSplit.getStoppingOffset(),
+                            UNKNOWN_BUCKET_TIMESTAMP,
                             logSplit.getNumberOfSplits()));
         }
         return new TableBucketWriteResultWithSplitIds(writeResults, finishedSplitIds);
@@ -363,7 +374,10 @@ private TableBucketWriteResultWithSplitIds finishCurrentSnapshotSplit() throws I
         String splitId = currentTableSplitsByBucket.remove(tableBucket).splitId();
         TableBucketWriteResult<WriteResult> writeResult =
                 completeLakeWriter(
-                        tableBucket, currentSnapshotSplit.getPartitionName(), logEndOffset);
+                        tableBucket,
+                        currentSnapshotSplit.getPartitionName(),
+                        logEndOffset,
+                        UNKNOWN_BUCKET_TIMESTAMP);
         closeCurrentSnapshotSplit();
         mayFinishCurrentTable();
         return new TableBucketWriteResultWithSplitIds(
@@ -483,9 +497,16 @@ private TableBucketWriteResult<WriteResult> toTableBucketWriteResult(
             @Nullable String partitionName,
             @Nullable WriteResult writeResult,
             long endLogOffset,
+            long maxTimestamp,
             int numberOfSplits) {
         return new TableBucketWriteResult<>(
-                tablePath, tableBucket, partitionName, writeResult, endLogOffset, numberOfSplits);
+                tablePath,
+                tableBucket,
+                partitionName,
+                writeResult,
+                endLogOffset,
+                maxTimestamp,
+                numberOfSplits);
     }
 
     private class TableBucketWriteResultWithSplitIds