[fluss-lake] Introduce TieringKvSplit and TieringLogSplit for Flink lake tiering service

Author: leonardBang

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/source/split/TieringKvSplit.java

@@ -0,0 +1,107 @@
+/*
+ *  Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package com.alibaba.fluss.flink.tiering.source.split;
+
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+
+import javax.annotation.Nullable;
+
+import java.util.Objects;
+
+/**
+ * The table split for tiering service. It's used to describe the snapshot data of a KV table
+ * bucket.
+ */
+public class TieringKvSplit extends TieringSplit {
+
+    private static final String TIERING_KV_SPLIT_PREFIX = "tiering-kv-split-";
+
+    /** The snapshot id. It's used to identify the snapshot for a kv bucket. */
+    private final long snapshotId;
+
+    /** The log offset corresponding to the KV table bucket snapshot finished. */
+    private final long logOffsetOfSnapshot;
+
+    /** The records to skip when reading the snapshot. */
+    private final long recordsToSkip;
+
+    public TieringKvSplit(
+            TablePath tablePath,
+            TableBucket tableBucket,
+            @Nullable String partitionName,
+            long snapshotId,
+            long logOffsetOfSnapshot,
+            long recordsToSkip) {
+        super(tablePath, tableBucket, partitionName);
+        this.snapshotId = snapshotId;
+        this.logOffsetOfSnapshot = logOffsetOfSnapshot;
+        this.recordsToSkip = recordsToSkip;
+    }
+
+    @Override
+    public String splitId() {
+        return toSplitId(TIERING_KV_SPLIT_PREFIX, this.tableBucket);
+    }
+
+    public long getSnapshotId() {
+        return snapshotId;
+    }
+
+    public long getLogOffsetOfSnapshot() {
+        return logOffsetOfSnapshot;
+    }
+
+    public long getRecordsToSkip() {
+        return recordsToSkip;
+    }
+
+    @Override
+    public String toString() {
+        return "TieringKvSplit{"
+                + "tablePath="
+                + tablePath
+                + ", tableBucket="
+                + tableBucket
+                + ", partitionName='"
+                + partitionName
+                + '\''
+                + ", snapshotId="
+                + snapshotId
+                + ", logOffsetOfSnapshot="
+                + logOffsetOfSnapshot
+                + ", recordsToSkip="
+                + recordsToSkip
+                + '}';
+    }
+
+    @Override
+    public boolean equals(Object object) {
+        if (!(object instanceof TieringKvSplit)) {
+            return false;
+        }
+        TieringKvSplit that = (TieringKvSplit) object;
+        return snapshotId == that.snapshotId
+                && logOffsetOfSnapshot == that.logOffsetOfSnapshot
+                && recordsToSkip == that.recordsToSkip;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(snapshotId, logOffsetOfSnapshot, recordsToSkip);
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/source/split/TieringLogSplit.java

@@ -0,0 +1,88 @@
+/*
+ *  Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package com.alibaba.fluss.flink.tiering.source.split;
+
+import com.alibaba.fluss.metadata.TableBucket;
+import com.alibaba.fluss.metadata.TablePath;
+
+import javax.annotation.Nullable;
+
+import java.util.Objects;
+
+/** The table split for tiering service. It's used to describe the log data of a table bucket. */
+public class TieringLogSplit extends TieringSplit {
+
+    private static final String TIERING_LOG_SPLIT_PREFIX = "tiering-log-split-";
+
+    private final long startingOffset;
+    private final long stoppingOffset;
+
+    public TieringLogSplit(
+            TablePath tablePath,
+            TableBucket tableBucket,
+            @Nullable String partitionName,
+            long startingOffset,
+            long stoppingOffset) {
+        super(tablePath, tableBucket, partitionName);
+        this.startingOffset = startingOffset;
+        this.stoppingOffset = stoppingOffset;
+    }
+
+    @Override
+    public String splitId() {
+        return toSplitId(TIERING_LOG_SPLIT_PREFIX, this.tableBucket);
+    }
+
+    public long getStartingOffset() {
+        return startingOffset;
+    }
+
+    public long getStoppingOffset() {
+        return stoppingOffset;
+    }
+
+    @Override
+    public String toString() {
+        return "TieringLogSplit{"
+                + "tablePath="
+                + tablePath
+                + ", tableBucket="
+                + tableBucket
+                + ", partitionName='"
+                + partitionName
+                + '\''
+                + ", startingOffset="
+                + startingOffset
+                + ", stoppingOffset="
+                + stoppingOffset
+                + '}';
+    }
+
+    @Override
+    public boolean equals(Object object) {
+        if (!(object instanceof TieringLogSplit)) {
+            return false;
+        }
+        TieringLogSplit that = (TieringLogSplit) object;
+        return startingOffset == that.startingOffset && stoppingOffset == that.stoppingOffset;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(startingOffset, stoppingOffset);
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/tiering/source/split/TieringSplit.java

@@ -23,103 +23,81 @@
 
 import javax.annotation.Nullable;
 
-import java.util.Objects;
+/** The base table split for tiering service. */
+public abstract class TieringSplit implements SourceSplit {
 
-/** The table split for tiering service. It's used to describe the log data of a table bucket. */
-public class TieringSplit implements SourceSplit {
+    private static final String TIERING_SPLIT_PREFIX = "tiering-";
+    public static final byte TIERING_KV_SPLIT_FLAG = 1;
+    public static final byte TIERING_LOG_SPLIT_FLAG = 2;
 
-    private static final String TIERING_SPLIT_PREFIX = "tiering-split-";
-
-    private final TablePath tablePath;
+    protected final TablePath tablePath;
     protected final TableBucket tableBucket;
     @Nullable protected final String partitionName;
-    private final long startingOffset;
-    private final long stoppingOffset;
 
     public TieringSplit(
-            TablePath tablePath,
-            TableBucket tableBucket,
-            @Nullable String partitionName,
-            long startingOffset,
-            long stoppingOffset) {
+            TablePath tablePath, TableBucket tableBucket, @Nullable String partitionName) {
         this.tablePath = tablePath;
         this.tableBucket = tableBucket;
         this.partitionName = partitionName;
-        this.startingOffset = startingOffset;
-        this.stoppingOffset = stoppingOffset;
         if ((tableBucket.getPartitionId() == null && partitionName != null)
                 || (tableBucket.getPartitionId() != null && partitionName == null)) {
             throw new IllegalArgumentException(
                     "Partition name and partition id must be both null or both not null.");
         }
     }
 
-    public long getStartingOffset() {
-        return startingOffset;
+    /** Checks whether this split is a kv split to tier. */
+    public final boolean isTieringKvSplit() {
+        return getClass() == TieringKvSplit.class;
     }
 
-    public long getStoppingOffset() {
-        return stoppingOffset;
+    /** Casts this split into a {@link TieringKvSplit}. */
+    public TieringKvSplit asTieringKvSplit() {
+        return (TieringKvSplit) this;
     }
-
-    public TablePath getTablePath() {
-        return tablePath;
+    /** Checks whether this split is a log split to tier. */
+    public final boolean isTieringLogSplit() {
+        return getClass() == TieringLogSplit.class;
     }
 
-    public TableBucket getTableBucket() {
-        return tableBucket;
+    /** Casts this split into a {@link TieringLogSplit}. */
+    public TieringLogSplit asTieringLogSplit() {
+        return (TieringLogSplit) this;
     }
 
-    @Nullable
-    public String getPartitionName() {
-        return partitionName;
+    protected byte splitKind() {
+        if (isTieringKvSplit()) {
+            return TIERING_KV_SPLIT_FLAG;
+        } else if (isTieringLogSplit()) {
+            return TIERING_LOG_SPLIT_FLAG;
+        } else {
+            throw new IllegalArgumentException("Unsupported split kind for " + getClass());
+        }
     }
 
-    public String splitId() {
+    protected static String toSplitId(String splitPrefix, TableBucket tableBucket) {
         if (tableBucket.getPartitionId() != null) {
-            return TIERING_SPLIT_PREFIX
+            return splitPrefix
                     + tableBucket.getTableId()
                     + "-p"
                     + tableBucket.getPartitionId()
                     + "-"
                     + tableBucket.getBucket();
         } else {
-            return TIERING_SPLIT_PREFIX + tableBucket.getTableId() + "-" + tableBucket.getBucket();
+            return splitPrefix + tableBucket.getTableId() + "-" + tableBucket.getBucket();
         }
     }
 
-    @Override
-    public String toString() {
-        return "TieringSplit{"
-                + "tablePath="
-                + tablePath
-                + ", tableBucket="
-                + tableBucket
-                + ", partitionName='"
-                + partitionName
-                + '\''
-                + ", startingOffset="
-                + startingOffset
-                + ", stoppingOffset="
-                + stoppingOffset
-                + '}';
+    public TablePath getTablePath() {
+        return tablePath;
     }
 
-    @Override
-    public boolean equals(Object object) {
-        if (!(object instanceof TieringSplit)) {
-            return false;
-        }
-        TieringSplit that = (TieringSplit) object;
-        return startingOffset == that.startingOffset
-                && stoppingOffset == that.stoppingOffset
-                && Objects.equals(tablePath, that.tablePath)
-                && Objects.equals(tableBucket, that.tableBucket)
-                && Objects.equals(partitionName, that.partitionName);
+    public TableBucket getTableBucket() {
+        return tableBucket;
     }
 
-    @Override
-    public int hashCode() {
-        return Objects.hash(tablePath, tableBucket, partitionName, startingOffset, stoppingOffset);
+    @Nullable
+    public String getPartitionName() {
+        return partitionName;
     }
 }