Skip to content

Commit 5af9dda

Browse files
authored
[tiering] Decouple common tiering class from flink module (#3554)
1 parent 233092a commit 5af9dda

39 files changed

Lines changed: 946 additions & 540 deletions

File tree

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

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
* limitations under the License.
1616
*/
1717

18-
package org.apache.fluss.flink.tiering.committer;
18+
package org.apache.fluss.client.tiering;
1919

2020
import org.apache.fluss.annotation.VisibleForTesting;
2121
import org.apache.fluss.client.metadata.MetadataUpdater;
@@ -229,7 +229,22 @@ public void commit(
229229
}
230230
}
231231

232-
void commit(
232+
/**
233+
* Registers a lake snapshot and its bucket offsets in Fluss.
234+
*
235+
* @param tableId the Fluss table ID
236+
* @param snapshotId the lake snapshot ID
237+
* @param lakeBucketTieredOffsetsPath the file containing tiered bucket offsets
238+
* @param readableLakeBucketTieredOffsetsPath the file containing readable bucket offsets, or
239+
* {@code null} if the snapshot is not readable
240+
* @param logEndOffsets the tiered log end offsets used for metrics reporting
241+
* @param logMaxTieredTimestamps the maximum tiered timestamps used for metrics reporting
242+
* @param earliestSnapshotIDToKeep the earliest snapshot ID to retain; {@code null} retains only
243+
* the latest snapshot, and {@link LakeCommitResult#KEEP_ALL_PREVIOUS} retains all previous
244+
* snapshots
245+
* @throws IOException if the snapshot cannot be committed to Fluss
246+
*/
247+
public void commit(
233248
long tableId,
234249
long snapshotId,
235250
String lakeBucketTieredOffsetsPath,
@@ -335,7 +350,7 @@ private CommitLakeTableSnapshotRequest toCommitLakeTableSnapshotRequest(
335350
commitLakeTableSnapshotRequest.addLakeTableSnapshotMetadata();
336351
pbLakeTableSnapshotMetadata.setSnapshotId(snapshotId);
337352
pbLakeTableSnapshotMetadata.setTableId(tableId);
338-
// tiered snapshot file path is equal to readable snapshot currently
353+
// Keep the tiered offsets path separate from the optional readable offsets path.
339354
pbLakeTableSnapshotMetadata.setTieredBucketOffsetsFilePath(tieredBucketOffsetsPath);
340355
if (readableBucketTieredOffsetsPath != null) {
341356
pbLakeTableSnapshotMetadata.setReadableBucketOffsetsFilePath(
@@ -360,7 +375,7 @@ private CommitLakeTableSnapshotRequest toCommitLakeTableSnapshotRequest(
360375
}
361376

362377
@VisibleForTesting
363-
protected CommitLakeTableSnapshotRequest addLogEndOffsets(
378+
CommitLakeTableSnapshotRequest addLogEndOffsets(
364379
CommitLakeTableSnapshotRequest commitLakeTableSnapshotRequest,
365380
long tableId,
366381
long snapshotId,

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
* limitations under the License.
1616
*/
1717

18-
package org.apache.fluss.flink.tiering.source;
18+
package org.apache.fluss.client.tiering;
1919

2020
import org.apache.fluss.lake.writer.LakeWriter;
2121
import org.apache.fluss.metadata.TableBucket;
Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.fluss.client.tiering;
19+
20+
import org.apache.fluss.lake.committer.TieringStats;
21+
22+
import javax.annotation.Nullable;
23+
24+
/**
25+
* The result of one table's commit round, holding the lake committable (nullable for empty commits
26+
* where no data was written) and the associated tiering statistics.
27+
*
28+
* @param <Committable> the type of the lake committable
29+
*/
30+
public class TieringCommitResult<Committable> {
31+
32+
/** The lake committable, or {@code null} if nothing was written in this round. */
33+
@Nullable private final Committable committable;
34+
35+
/** Per-table tiering statistics collected during this round. */
36+
@Nullable private final TieringStats stats;
37+
38+
public TieringCommitResult(@Nullable Committable committable, @Nullable TieringStats stats) {
39+
this.committable = committable;
40+
this.stats = stats;
41+
}
42+
43+
@Nullable
44+
public Committable getCommittable() {
45+
return committable;
46+
}
47+
48+
@Nullable
49+
public TieringStats getStats() {
50+
return stats;
51+
}
52+
}

0 commit comments

Comments
 (0)