This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new c108feae4 [lake/hudi] Introduce split index field for tiering splits
(#3503)
c108feae4 is described below
commit c108feae452ddfb4da376c97fa17936b35205f74
Author: fhan <[email protected]>
AuthorDate: Mon Jun 22 10:37:30 2026 +0800
[lake/hudi] Introduce split index field for tiering splits (#3503)
---
.../fluss/lake/writer/WriterInitContext.java | 32 +++++++++
.../fluss/lake/writer/WriterInitContextTest.java | 63 +++++++++++++++++
.../flink/tiering/source/TieringSplitReader.java | 30 ++++++---
.../tiering/source/TieringWriterInitContext.java | 30 +++++++++
.../source/enumerator/TieringSourceEnumerator.java | 21 ++++--
.../tiering/source/split/TieringLogSplit.java | 62 ++++++++++++++++-
.../tiering/source/split/TieringSnapshotSplit.java | 63 ++++++++++++++++-
.../flink/tiering/source/split/TieringSplit.java | 78 +++++++++++++++++++++-
.../source/split/TieringSplitSerializer.java | 13 +++-
.../tiering/source/state/TieringSplitState.java | 10 ++-
.../enumerator/TieringSourceEnumeratorTest.java | 44 ++++++++++++
.../source/split/TieringSplitSerializerTest.java | 53 +++++++++++++--
.../source/state/TieringSplitStateTest.java | 20 ++++--
13 files changed, 483 insertions(+), 36 deletions(-)
diff --git
a/fluss-common/src/main/java/org/apache/fluss/lake/writer/WriterInitContext.java
b/fluss-common/src/main/java/org/apache/fluss/lake/writer/WriterInitContext.java
index 6c9c2f408..318f53f54 100644
---
a/fluss-common/src/main/java/org/apache/fluss/lake/writer/WriterInitContext.java
+++
b/fluss-common/src/main/java/org/apache/fluss/lake/writer/WriterInitContext.java
@@ -34,6 +34,12 @@ import javax.annotation.Nullable;
@PublicEvolving
public interface WriterInitContext {
+ /** Unknown split index for lake writers that do not need split-level
coordination. */
+ int UNKNOWN_SPLIT_INDEX = -1;
+
+ /** Unknown tiering round timestamp for lake writers that do not need
round coordination. */
+ long UNKNOWN_TIERING_ROUND_TIMESTAMP = -1L;
+
/**
* Returns the table path.
*
@@ -62,4 +68,30 @@ public interface WriterInitContext {
* @return the Fluss table info
*/
TableInfo tableInfo();
+
+ /**
+ * Returns the split index in the current tiering round.
+ *
+ * <p>The split whose index is {@code 0} is the first split in the current
tiering round. Lake
+ * writers that need one writer to coordinate round-level metadata can use
the first split as
+ * the coordinator. The value is {@link #UNKNOWN_SPLIT_INDEX} when the
lake writer does not need
+ * this information.
+ *
+ * @return the split index in the current tiering round
+ */
+ default int splitIndex() {
+ return UNKNOWN_SPLIT_INDEX;
+ }
+
+ /**
+ * Returns the timestamp when the current tiering round was generated.
+ *
+ * <p>The value is {@link #UNKNOWN_TIERING_ROUND_TIMESTAMP} when the lake
writer does not need
+ * round-level timestamp coordination.
+ *
+ * @return the tiering round timestamp in milliseconds
+ */
+ default long tieringRoundTimestamp() {
+ return UNKNOWN_TIERING_ROUND_TIMESTAMP;
+ }
}
diff --git
a/fluss-common/src/test/java/org/apache/fluss/lake/writer/WriterInitContextTest.java
b/fluss-common/src/test/java/org/apache/fluss/lake/writer/WriterInitContextTest.java
new file mode 100644
index 000000000..bf0198bd8
--- /dev/null
+++
b/fluss-common/src/test/java/org/apache/fluss/lake/writer/WriterInitContextTest.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.fluss.lake.writer;
+
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TableInfo;
+import org.apache.fluss.metadata.TablePath;
+
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link WriterInitContext}. */
+class WriterInitContextTest {
+
+ @Test
+ void testDefaultTieringRoundMetadata() {
+ WriterInitContext context =
+ new WriterInitContext() {
+ @Override
+ public TablePath tablePath() {
+ return null;
+ }
+
+ @Override
+ public TableBucket tableBucket() {
+ return null;
+ }
+
+ @Nullable
+ @Override
+ public String partition() {
+ return null;
+ }
+
+ @Override
+ public TableInfo tableInfo() {
+ return null;
+ }
+ };
+
+
assertThat(context.splitIndex()).isEqualTo(WriterInitContext.UNKNOWN_SPLIT_INDEX);
+ assertThat(context.tieringRoundTimestamp())
+ .isEqualTo(WriterInitContext.UNKNOWN_TIERING_ROUND_TIMESTAMP);
+ }
+}
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java
index 324ce8a9e..960c4db7e 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java
@@ -447,12 +447,15 @@ public class TieringSplitReader<WriteResult>
long lastWrittenTimestamp =
handler.handleRecords(
records,
- () ->
- getOrCreateLakeWriter(
- bucket,
- currentTableSplitsByBucket
- .get(bucket)
- .getPartitionName()),
+ () -> {
+ TieringSplit split =
+
checkNotNull(currentTableSplitsByBucket.get(bucket));
+ return getOrCreateLakeWriter(
+ bucket,
+ split.getPartitionName(),
+ split.getSplitIndex(),
+ split.getTieringRoundTimestamp());
+ },
stoppingOffset);
// The split owns offsets before stoppingOffset only. If the
scanner consumed past
@@ -589,7 +592,11 @@ public class TieringSplitReader<WriteResult>
}
private LakeWriter<WriteResult> getOrCreateLakeWriter(
- TableBucket bucket, @Nullable String partitionName) throws
IOException {
+ TableBucket bucket,
+ @Nullable String partitionName,
+ int splitIndex,
+ long tieringRoundTimestamp)
+ throws IOException {
LakeWriter<WriteResult> lakeWriter = lakeWriters.get(bucket);
if (lakeWriter == null) {
lakeWriter =
@@ -598,7 +605,9 @@ public class TieringSplitReader<WriteResult>
currentTablePath,
bucket,
partitionName,
- currentTable.getTableInfo()));
+ currentTable.getTableInfo(),
+ splitIndex,
+ tieringRoundTimestamp));
lakeWriters.put(bucket, lakeWriter);
}
return lakeWriter;
@@ -683,7 +692,10 @@ public class TieringSplitReader<WriteResult>
if (lakeWriter == null) {
lakeWriter =
getOrCreateLakeWriter(
- bucket,
checkNotNull(currentSnapshotSplit).getPartitionName());
+ bucket,
+
checkNotNull(currentSnapshotSplit).getPartitionName(),
+ currentSnapshotSplit.getSplitIndex(),
+
currentSnapshotSplit.getTieringRoundTimestamp());
}
lakeWriter.write(scanRecord);
if (scanRecord.getSizeInBytes() > 0) {
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringWriterInitContext.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringWriterInitContext.java
index 121c4fb9c..7dbb6e41b 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringWriterInitContext.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringWriterInitContext.java
@@ -31,16 +31,36 @@ public class TieringWriterInitContext implements
WriterInitContext {
private final TableBucket tableBucket;
@Nullable private final String partition;
private final TableInfo tableInfo;
+ private final int splitIndex;
+ private final long tieringRoundTimestamp;
public TieringWriterInitContext(
TablePath tablePath,
TableBucket tableBucket,
@Nullable String partition,
TableInfo tableInfo) {
+ this(
+ tablePath,
+ tableBucket,
+ partition,
+ tableInfo,
+ UNKNOWN_SPLIT_INDEX,
+ UNKNOWN_TIERING_ROUND_TIMESTAMP);
+ }
+
+ public TieringWriterInitContext(
+ TablePath tablePath,
+ TableBucket tableBucket,
+ @Nullable String partition,
+ TableInfo tableInfo,
+ int splitIndex,
+ long tieringRoundTimestamp) {
this.tablePath = tablePath;
this.tableBucket = tableBucket;
this.partition = partition;
this.tableInfo = tableInfo;
+ this.splitIndex = splitIndex;
+ this.tieringRoundTimestamp = tieringRoundTimestamp;
}
@Override
@@ -63,4 +83,14 @@ public class TieringWriterInitContext implements
WriterInitContext {
public TableInfo tableInfo() {
return tableInfo;
}
+
+ @Override
+ public int splitIndex() {
+ return splitIndex;
+ }
+
+ @Override
+ public long tieringRoundTimestamp() {
+ return tieringRoundTimestamp;
+ }
}
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java
index 7c79596e8..5545e9a50 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java
@@ -449,11 +449,11 @@ public class TieringSourceEnumerator
try {
TablePath tablePath = tieringTable.f2;
final TableInfo tableInfo =
flussAdmin.getTableInfo(tablePath).get();
- List<TieringSplit> tieringSplits =
-
populateNumberOfTieringSplits(splitGenerator.generateTableSplits(tableInfo));
+ List<TieringSplit> tieringSplits =
splitGenerator.generateTableSplits(tableInfo);
// shuffle tiering split to avoid splits tiering skew
// after introduce tiering max duration
Collections.shuffle(tieringSplits);
+ tieringSplits = populateTieringRoundMetadata(tieringSplits);
LOG.info(
"Generate Tiering {} splits for table {} with cost {}ms.",
tieringSplits.size(),
@@ -487,11 +487,20 @@ public class TieringSourceEnumerator
}
}
- private List<TieringSplit>
populateNumberOfTieringSplits(List<TieringSplit> tieringSplits) {
+ private List<TieringSplit> populateTieringRoundMetadata(List<TieringSplit>
tieringSplits) {
int numberOfSplits = tieringSplits.size();
- return tieringSplits.stream()
- .map(split -> split.copy(numberOfSplits))
- .collect(Collectors.toList());
+ if (numberOfSplits == 0) {
+ return Collections.emptyList();
+ }
+ long tieringRoundTimestamp = System.currentTimeMillis();
+ List<TieringSplit> splitsWithMetadata = new
ArrayList<>(numberOfSplits);
+ for (int splitIndex = 0; splitIndex < numberOfSplits; splitIndex++) {
+ splitsWithMetadata.add(
+ tieringSplits
+ .get(splitIndex)
+ .copy(numberOfSplits, splitIndex,
tieringRoundTimestamp));
+ }
+ return splitsWithMetadata;
}
@Override
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringLogSplit.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringLogSplit.java
index afd8d75f8..a1e983b50 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringLogSplit.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringLogSplit.java
@@ -64,6 +64,27 @@ public class TieringLogSplit extends TieringSplit {
false);
}
+ public TieringLogSplit(
+ TablePath tablePath,
+ TableBucket tableBucket,
+ @Nullable String partitionName,
+ long startingOffset,
+ long stoppingOffset,
+ int numberOfSplits,
+ int splitIndex,
+ long tieringRoundTimestamp) {
+ this(
+ tablePath,
+ tableBucket,
+ partitionName,
+ startingOffset,
+ stoppingOffset,
+ numberOfSplits,
+ false,
+ splitIndex,
+ tieringRoundTimestamp);
+ }
+
public TieringLogSplit(
TablePath tablePath,
TableBucket tableBucket,
@@ -72,7 +93,36 @@ public class TieringLogSplit extends TieringSplit {
long stoppingOffset,
int numberOfSplits,
boolean skipCurrentRound) {
- super(tablePath, tableBucket, partitionName, numberOfSplits,
skipCurrentRound);
+ this(
+ tablePath,
+ tableBucket,
+ partitionName,
+ startingOffset,
+ stoppingOffset,
+ numberOfSplits,
+ skipCurrentRound,
+ UNKNOWN_SPLIT_INDEX,
+ UNKNOWN_TIERING_ROUND_TIMESTAMP);
+ }
+
+ public TieringLogSplit(
+ TablePath tablePath,
+ TableBucket tableBucket,
+ @Nullable String partitionName,
+ long startingOffset,
+ long stoppingOffset,
+ int numberOfSplits,
+ boolean skipCurrentRound,
+ int splitIndex,
+ long tieringRoundTimestamp) {
+ super(
+ tablePath,
+ tableBucket,
+ partitionName,
+ numberOfSplits,
+ skipCurrentRound,
+ splitIndex,
+ tieringRoundTimestamp);
this.startingOffset = startingOffset;
this.stoppingOffset = stoppingOffset;
}
@@ -108,11 +158,15 @@ public class TieringLogSplit extends TieringSplit {
+ startingOffset
+ ", stoppingOffset="
+ stoppingOffset
+ + ", splitIndex="
+ + splitIndex
+ + ", tieringRoundTimestamp="
+ + tieringRoundTimestamp
+ '}';
}
@Override
- public TieringLogSplit copy(int numberOfSplits) {
+ public TieringLogSplit copy(int numberOfSplits, int splitIndex, long
tieringRoundTimestamp) {
return new TieringLogSplit(
tablePath,
tableBucket,
@@ -120,7 +174,9 @@ public class TieringLogSplit extends TieringSplit {
startingOffset,
stoppingOffset,
numberOfSplits,
- skipCurrentRound);
+ skipCurrentRound,
+ splitIndex,
+ tieringRoundTimestamp);
}
@Override
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSnapshotSplit.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSnapshotSplit.java
index e6b37a925..8c3bad364 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSnapshotSplit.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSnapshotSplit.java
@@ -55,6 +55,27 @@ public class TieringSnapshotSplit extends TieringSplit {
false);
}
+ public TieringSnapshotSplit(
+ TablePath tablePath,
+ TableBucket tableBucket,
+ @Nullable String partitionName,
+ long snapshotId,
+ long logOffsetOfSnapshot,
+ int numberOfSplits,
+ int splitIndex,
+ long tieringRoundTimestamp) {
+ this(
+ tablePath,
+ tableBucket,
+ partitionName,
+ snapshotId,
+ logOffsetOfSnapshot,
+ numberOfSplits,
+ false,
+ splitIndex,
+ tieringRoundTimestamp);
+ }
+
public TieringSnapshotSplit(
TablePath tablePath,
TableBucket tableBucket,
@@ -63,7 +84,36 @@ public class TieringSnapshotSplit extends TieringSplit {
long logOffsetOfSnapshot,
int numberOfSplits,
boolean skipCurrentRound) {
- super(tablePath, tableBucket, partitionName, numberOfSplits,
skipCurrentRound);
+ this(
+ tablePath,
+ tableBucket,
+ partitionName,
+ snapshotId,
+ logOffsetOfSnapshot,
+ numberOfSplits,
+ skipCurrentRound,
+ UNKNOWN_SPLIT_INDEX,
+ UNKNOWN_TIERING_ROUND_TIMESTAMP);
+ }
+
+ public TieringSnapshotSplit(
+ TablePath tablePath,
+ TableBucket tableBucket,
+ @Nullable String partitionName,
+ long snapshotId,
+ long logOffsetOfSnapshot,
+ int numberOfSplits,
+ boolean skipCurrentRound,
+ int splitIndex,
+ long tieringRoundTimestamp) {
+ super(
+ tablePath,
+ tableBucket,
+ partitionName,
+ numberOfSplits,
+ skipCurrentRound,
+ splitIndex,
+ tieringRoundTimestamp);
this.snapshotId = snapshotId;
this.logOffsetOfSnapshot = logOffsetOfSnapshot;
}
@@ -99,11 +149,16 @@ public class TieringSnapshotSplit extends TieringSplit {
+ snapshotId
+ ", logOffsetOfSnapshot="
+ logOffsetOfSnapshot
+ + ", splitIndex="
+ + splitIndex
+ + ", tieringRoundTimestamp="
+ + tieringRoundTimestamp
+ '}';
}
@Override
- public TieringSnapshotSplit copy(int numberOfSplits) {
+ public TieringSnapshotSplit copy(
+ int numberOfSplits, int splitIndex, long tieringRoundTimestamp) {
return new TieringSnapshotSplit(
tablePath,
tableBucket,
@@ -111,7 +166,9 @@ public class TieringSnapshotSplit extends TieringSplit {
snapshotId,
logOffsetOfSnapshot,
numberOfSplits,
- skipCurrentRound);
+ skipCurrentRound,
+ splitIndex,
+ tieringRoundTimestamp);
}
@Override
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java
index d6827a62c..b0ea25693 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java
@@ -31,6 +31,8 @@ public abstract class TieringSplit implements SourceSplit {
public static final byte TIERING_SNAPSHOT_SPLIT_FLAG = 1;
public static final byte TIERING_LOG_SPLIT_FLAG = 2;
+ public static final int UNKNOWN_SPLIT_INDEX = -1;
+ public static final long UNKNOWN_TIERING_ROUND_TIMESTAMP = -1L;
protected static final int UNKNOWN_NUMBER_OF_SPLITS = -1;
@@ -40,6 +42,10 @@ public abstract class TieringSplit implements SourceSplit {
// the total number of splits in one round of tiering
protected final int numberOfSplits;
+ // the split index in one round of tiering
+ protected final int splitIndex;
+ // the timestamp when one round of tiering was generated
+ protected final long tieringRoundTimestamp;
/**
* Indicates whether to skip tiering data for this split in the current
round of tiering. When
@@ -53,6 +59,24 @@ public abstract class TieringSplit implements SourceSplit {
@Nullable String partitionName,
int numberOfSplits,
boolean skipCurrentRound) {
+ this(
+ tablePath,
+ tableBucket,
+ partitionName,
+ numberOfSplits,
+ skipCurrentRound,
+ UNKNOWN_SPLIT_INDEX,
+ UNKNOWN_TIERING_ROUND_TIMESTAMP);
+ }
+
+ public TieringSplit(
+ TablePath tablePath,
+ TableBucket tableBucket,
+ @Nullable String partitionName,
+ int numberOfSplits,
+ boolean skipCurrentRound,
+ int splitIndex,
+ long tieringRoundTimestamp) {
this.tablePath = tablePath;
this.tableBucket = tableBucket;
this.partitionName = partitionName;
@@ -63,6 +87,10 @@ public abstract class TieringSplit implements SourceSplit {
}
this.numberOfSplits = numberOfSplits;
this.skipCurrentRound = skipCurrentRound;
+ validateSplitIndex(numberOfSplits, splitIndex);
+ validateTieringRoundTimestamp(tieringRoundTimestamp);
+ this.splitIndex = splitIndex;
+ this.tieringRoundTimestamp = tieringRoundTimestamp;
}
/** Checks whether this split is a primary key table split to tier. */
@@ -116,6 +144,18 @@ public abstract class TieringSplit implements SourceSplit {
return numberOfSplits;
}
+ public int getSplitIndex() {
+ return splitIndex;
+ }
+
+ public boolean isFirstSplit() {
+ return splitIndex == 0;
+ }
+
+ public long getTieringRoundTimestamp() {
+ return tieringRoundTimestamp;
+ }
+
protected static String toSplitId(String splitPrefix, TableBucket
tableBucket) {
if (tableBucket.getPartitionId() != null) {
return splitPrefix
@@ -142,7 +182,12 @@ public abstract class TieringSplit implements SourceSplit {
return partitionName;
}
- public abstract TieringSplit copy(int numberOfSplits);
+ public TieringSplit copy(int numberOfSplits) {
+ return copy(numberOfSplits, splitIndex, tieringRoundTimestamp);
+ }
+
+ public abstract TieringSplit copy(
+ int numberOfSplits, int splitIndex, long tieringRoundTimestamp);
@Override
public boolean equals(Object object) {
@@ -154,12 +199,39 @@ public abstract class TieringSplit implements SourceSplit
{
&& Objects.equals(tableBucket, that.tableBucket)
&& Objects.equals(partitionName, that.partitionName)
&& numberOfSplits == that.numberOfSplits
- && skipCurrentRound == that.skipCurrentRound;
+ && skipCurrentRound == that.skipCurrentRound
+ && splitIndex == that.splitIndex
+ && tieringRoundTimestamp == that.tieringRoundTimestamp;
}
@Override
public int hashCode() {
return Objects.hash(
- tablePath, tableBucket, partitionName, numberOfSplits,
skipCurrentRound);
+ tablePath,
+ tableBucket,
+ partitionName,
+ numberOfSplits,
+ skipCurrentRound,
+ splitIndex,
+ tieringRoundTimestamp);
+ }
+
+ private static void validateSplitIndex(int numberOfSplits, int splitIndex)
{
+ if (splitIndex < UNKNOWN_SPLIT_INDEX) {
+ throw new IllegalArgumentException("Split index must be -1 or
non-negative.");
+ }
+ if (splitIndex != UNKNOWN_SPLIT_INDEX
+ && numberOfSplits != UNKNOWN_NUMBER_OF_SPLITS
+ && (numberOfSplits <= 0 || splitIndex >= numberOfSplits)) {
+ throw new IllegalArgumentException(
+ "Split index must be smaller than the number of splits.");
+ }
+ }
+
+ private static void validateTieringRoundTimestamp(long
tieringRoundTimestamp) {
+ if (tieringRoundTimestamp != UNKNOWN_TIERING_ROUND_TIMESTAMP
+ && tieringRoundTimestamp <= 0) {
+ throw new IllegalArgumentException("Tiering round timestamp must
be -1 or positive.");
+ }
}
}
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializer.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializer.java
index 62c4f7886..e336ee467 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializer.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializer.java
@@ -86,6 +86,9 @@ public class TieringSplitSerializer implements
SimpleVersionedSerializer<Tiering
out.writeInt(split.getNumberOfSplits());
// write skipCurrentRound
out.writeBoolean(split.shouldSkipCurrentRound());
+ // write split metadata for this tiering round
+ out.writeInt(split.getSplitIndex());
+ out.writeLong(split.getTieringRoundTimestamp());
if (split.isTieringSnapshotSplit()) {
// Snapshot split
TieringSnapshotSplit tieringSnapshotSplit =
split.asTieringSnapshotSplit();
@@ -139,6 +142,8 @@ public class TieringSplitSerializer implements
SimpleVersionedSerializer<Tiering
// deserialize number of splits
int numberOfSplits = in.readInt();
boolean skipCurrentRound = in.readBoolean();
+ int splitIndex = in.readInt();
+ long tieringRoundTimestamp = in.readLong();
if (splitKind == TIERING_SNAPSHOT_SPLIT_FLAG) {
// deserialize snapshot id
@@ -152,7 +157,9 @@ public class TieringSplitSerializer implements
SimpleVersionedSerializer<Tiering
snapshotId,
logOffsetOfSnapshot,
numberOfSplits,
- skipCurrentRound);
+ skipCurrentRound,
+ splitIndex,
+ tieringRoundTimestamp);
} else {
// deserialize starting offset
long startingOffset = in.readLong();
@@ -165,7 +172,9 @@ public class TieringSplitSerializer implements
SimpleVersionedSerializer<Tiering
startingOffset,
stoppingOffset,
numberOfSplits,
- skipCurrentRound);
+ skipCurrentRound,
+ splitIndex,
+ tieringRoundTimestamp);
}
}
}
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/state/TieringSplitState.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/state/TieringSplitState.java
index 1b5be62b7..0690da628 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/state/TieringSplitState.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/state/TieringSplitState.java
@@ -45,7 +45,10 @@ public class TieringSplitState {
split.getPartitionName(),
split.getSnapshotId(),
split.getLogOffsetOfSnapshot(),
- split.getNumberOfSplits());
+ split.getNumberOfSplits(),
+ split.shouldSkipCurrentRound(),
+ split.getSplitIndex(),
+ split.getTieringRoundTimestamp());
} else {
final TieringLogSplit split = (TieringLogSplit) tieringSplit;
return new TieringLogSplit(
@@ -54,7 +57,10 @@ public class TieringSplitState {
split.getPartitionName(),
split.getStartingOffset(),
split.getStoppingOffset(),
- split.getNumberOfSplits());
+ split.getNumberOfSplits(),
+ split.shouldSkipCurrentRound(),
+ split.getSplitIndex(),
+ split.getTieringRoundTimestamp());
}
}
}
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java
index 446534688..9e7975d0c 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java
@@ -142,6 +142,8 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
List<TieringSplit> actualLogAssignment = new ArrayList<>();
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualLogAssignment::addAll));
+ assertTieringRoundMetadata(actualLogAssignment);
+ clearTieringSplitMetadata(actualLogAssignment);
assertThat(actualLogAssignment)
.containsExactlyInAnyOrderElementsOf(expectedLogAssignment);
}
@@ -185,6 +187,8 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
List<TieringSplit> actualAssignment = new ArrayList<>();
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualAssignment::addAll));
+ assertTieringRoundMetadata(actualAssignment);
+ clearTieringSplitMetadata(actualAssignment);
assertThat(actualAssignment)
.containsExactlyInAnyOrderElementsOf(expectedSnapshotAssignment);
@@ -230,6 +234,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
List<TieringSplit> actualLogAssignment = new ArrayList<>();
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualLogAssignment::addAll));
+ clearTieringSplitMetadata(actualLogAssignment);
assertThat(actualLogAssignment)
.containsExactlyInAnyOrderElementsOf(expectedLogAssignment);
}
@@ -273,6 +278,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualAssignment::addAll));
+ clearTieringSplitMetadata(actualAssignment);
assertThat(actualAssignment).containsExactlyInAnyOrderElementsOf(expectedAssignment);
// mock finished tiered this round, check second round
@@ -316,6 +322,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
List<TieringSplit> actualLogAssignment = new ArrayList<>();
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualLogAssignment::addAll));
+ clearTieringSplitMetadata(actualLogAssignment);
assertThat(actualLogAssignment)
.containsExactlyInAnyOrderElementsOf(expectedLogAssignment);
}
@@ -416,6 +423,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
context.getSplitsAssignmentSequence()) {
splitsAssignment.assignment().values().forEach(actualLogAssignment::addAll);
}
+ clearTieringSplitMetadata(actualLogAssignment);
assertThat(actualLogAssignment)
.containsExactlyInAnyOrderElementsOf(expectedLogAssignment);
}
@@ -472,6 +480,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
context.getSplitsAssignmentSequence()) {
splitsAssignment.assignment().values().forEach(actualAssignment::addAll);
}
+ clearTieringSplitMetadata(actualAssignment);
assertThat(actualAssignment).containsExactlyInAnyOrderElementsOf(expectedAssignment);
// mock finished tiered this round, check second round
@@ -539,6 +548,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
context.getSplitsAssignmentSequence()) {
splitsAssignment.assignment().values().forEach(actualLogAssignment::addAll);
}
+ clearTieringSplitMetadata(actualLogAssignment);
assertThat(actualLogAssignment)
.containsExactlyInAnyOrderElementsOf(expectedLogAssignment);
}
@@ -579,6 +589,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualAssignment::addAll));
+ clearTieringSplitMetadata(actualAssignment);
assertThat(actualAssignment).containsExactlyInAnyOrderElementsOf(expectedAssignment);
// mock tiering fail by send tiering fail event
@@ -593,6 +604,7 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
List<TieringSplit> actualAssignment1 = new ArrayList<>();
context.getSplitsAssignmentSequence()
.forEach(a ->
a.assignment().values().forEach(actualAssignment1::addAll));
+ clearTieringSplitMetadata(actualAssignment1);
assertThat(actualAssignment1).containsExactlyInAnyOrderElementsOf(expectedAssignment);
}
}
@@ -673,6 +685,38 @@ class TieringSourceEnumeratorTest extends TieringTestBase {
}
// --------------------- Test Utils ---------------------
+ private static void assertTieringRoundMetadata(List<TieringSplit>
tieringSplits) {
+ assertThat(tieringSplits).isNotEmpty();
+
assertThat(tieringSplits).filteredOn(TieringSplit::isFirstSplit).hasSize(1);
+ assertThat(tieringSplits)
+ .extracting(TieringSplit::getSplitIndex)
+ .containsExactlyInAnyOrderElementsOf(
+ java.util.stream.IntStream.range(0,
tieringSplits.size())
+ .boxed()
+ .collect(Collectors.toList()));
+ long tieringRoundTimestamp =
tieringSplits.get(0).getTieringRoundTimestamp();
+ assertThat(tieringRoundTimestamp).isPositive();
+ assertThat(tieringSplits)
+ .allSatisfy(
+ split -> {
+
assertThat(split.getNumberOfSplits()).isEqualTo(tieringSplits.size());
+ assertThat(split.getTieringRoundTimestamp())
+ .isEqualTo(tieringRoundTimestamp);
+ });
+ }
+
+ private static void clearTieringSplitMetadata(List<TieringSplit>
tieringSplits) {
+ for (int i = 0; i < tieringSplits.size(); i++) {
+ TieringSplit tieringSplit = tieringSplits.get(i);
+ tieringSplits.set(
+ i,
+ tieringSplit.copy(
+ tieringSplit.getNumberOfSplits(),
+ TieringSplit.UNKNOWN_SPLIT_INDEX,
+ TieringSplit.UNKNOWN_TIERING_ROUND_TIMESTAMP));
+ }
+ }
+
private void registerReaderAndHandleSplitRequests(
FlussMockSplitEnumeratorContext<TieringSplit> context,
TieringSourceEnumerator enumerator,
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializerTest.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializerTest.java
index 45cbd76a8..ef7de7f08 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializerTest.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializerTest.java
@@ -69,8 +69,8 @@ class TieringSplitSerializerTest {
String expectedSplitString =
isPartitionedTable
- ?
"TieringSnapshotSplit{tablePath=test_db.test_partitioned_table,
tableBucket=TableBucket{tableId=1, partitionId=100, bucket=2},
partitionName='1024', numberOfSplits=30, skipCurrentRound=false, snapshotId=0,
logOffsetOfSnapshot=200}"
- : "TieringSnapshotSplit{tablePath=test_db.test_table,
tableBucket=TableBucket{tableId=1, bucket=2}, partitionName='null',
numberOfSplits=30, skipCurrentRound=false, snapshotId=0,
logOffsetOfSnapshot=200}";
+ ?
"TieringSnapshotSplit{tablePath=test_db.test_partitioned_table,
tableBucket=TableBucket{tableId=1, partitionId=100, bucket=2},
partitionName='1024', numberOfSplits=30, skipCurrentRound=false, snapshotId=0,
logOffsetOfSnapshot=200, splitIndex=-1, tieringRoundTimestamp=-1}"
+ : "TieringSnapshotSplit{tablePath=test_db.test_table,
tableBucket=TableBucket{tableId=1, bucket=2}, partitionName='null',
numberOfSplits=30, skipCurrentRound=false, snapshotId=0,
logOffsetOfSnapshot=200, splitIndex=-1, tieringRoundTimestamp=-1}";
assertThat(new TieringSnapshotSplit(path, bucket, partitionName, 0L,
200L, 30).toString())
.isEqualTo(expectedSplitString);
}
@@ -103,8 +103,8 @@ class TieringSplitSerializerTest {
String expectedSplitString =
isPartitionedTable
- ?
"TieringLogSplit{tablePath=test_db.test_partitioned_table,
tableBucket=TableBucket{tableId=1, partitionId=100, bucket=2},
partitionName='1024', numberOfSplits=2, skipCurrentRound=false,
startingOffset=100, stoppingOffset=200}"
- : "TieringLogSplit{tablePath=test_db.test_table,
tableBucket=TableBucket{tableId=1, bucket=2}, partitionName='null',
numberOfSplits=2, skipCurrentRound=false, startingOffset=100,
stoppingOffset=200}";
+ ?
"TieringLogSplit{tablePath=test_db.test_partitioned_table,
tableBucket=TableBucket{tableId=1, partitionId=100, bucket=2},
partitionName='1024', numberOfSplits=2, skipCurrentRound=false,
startingOffset=100, stoppingOffset=200, splitIndex=-1,
tieringRoundTimestamp=-1}"
+ : "TieringLogSplit{tablePath=test_db.test_table,
tableBucket=TableBucket{tableId=1, bucket=2}, partitionName='null',
numberOfSplits=2, skipCurrentRound=false, startingOffset=100,
stoppingOffset=200, splitIndex=-1, tieringRoundTimestamp=-1}";
assertThat(new TieringLogSplit(path, bucket, partitionName, 100, 200,
2).toString())
.isEqualTo(expectedSplitString);
}
@@ -151,4 +151,49 @@ class TieringSplitSerializerTest {
(TieringLogSplit)
serializer.deserialize(serializer.getVersion(), serialized);
assertThat(deserializedLogSplit).isEqualTo(logSplit);
}
+
+ @Test
+ void testTieringRoundMetadataSerde() throws Exception {
+ TieringSnapshotSplit snapshotSplit =
+ new TieringSnapshotSplit(tablePath, tableBucket, null, 0L,
200L, 10, 0, 1000L);
+ byte[] serialized = serializer.serialize(snapshotSplit);
+ TieringSnapshotSplit deserializedSnapshotSplit =
+ (TieringSnapshotSplit)
serializer.deserialize(serializer.getVersion(), serialized);
+ assertThat(deserializedSnapshotSplit.getSplitIndex()).isZero();
+ assertThat(deserializedSnapshotSplit.isFirstSplit()).isTrue();
+
assertThat(deserializedSnapshotSplit.getTieringRoundTimestamp()).isEqualTo(1000L);
+
+ TieringLogSplit logSplit =
+ new TieringLogSplit(tablePath, tableBucket, null, 100, 200,
40, 2, 2000L);
+ serialized = serializer.serialize(logSplit);
+ TieringLogSplit deserializedLogSplit =
+ (TieringLogSplit)
serializer.deserialize(serializer.getVersion(), serialized);
+ assertThat(deserializedLogSplit.getSplitIndex()).isEqualTo(2);
+ assertThat(deserializedLogSplit.isFirstSplit()).isFalse();
+
assertThat(deserializedLogSplit.getTieringRoundTimestamp()).isEqualTo(2000L);
+ }
+
+ @Test
+ void testTieringRoundMetadataParticipatesInSplitIdentity() {
+ TieringSnapshotSplit snapshotSplit =
+ new TieringSnapshotSplit(tablePath, tableBucket, null, 0L,
200L, 10, 0, 1000L);
+ TieringSnapshotSplit snapshotSplitWithDifferentMetadata =
+ new TieringSnapshotSplit(tablePath, tableBucket, null, 0L,
200L, 10, 1, 2000L);
+
assertThat(snapshotSplit).isNotEqualTo(snapshotSplitWithDifferentMetadata);
+
+ TieringLogSplit logSplit =
+ new TieringLogSplit(tablePath, tableBucket, null, 100, 200,
40, 0, 1000L);
+ TieringLogSplit logSplitWithDifferentMetadata =
+ new TieringLogSplit(tablePath, tableBucket, null, 100, 200,
40, 1, 2000L);
+ assertThat(logSplit).isNotEqualTo(logSplitWithDifferentMetadata);
+ }
+
+ @Test
+ void testLogSplitDefaultTieringRoundMetadataIsUnknown() {
+ TieringLogSplit logSplit = new TieringLogSplit(tablePath, tableBucket,
null, 100, 200);
+
+
assertThat(logSplit.getSplitIndex()).isEqualTo(TieringSplit.UNKNOWN_SPLIT_INDEX);
+ assertThat(logSplit.getTieringRoundTimestamp())
+ .isEqualTo(TieringSplit.UNKNOWN_TIERING_ROUND_TIMESTAMP);
+ }
}
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/state/TieringSplitStateTest.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/state/TieringSplitStateTest.java
index e51072872..ebc08553e 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/state/TieringSplitStateTest.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/state/TieringSplitStateTest.java
@@ -36,9 +36,16 @@ class TieringSplitStateTest {
// verify conversion between TieringSnapshotSplitState and
TieringSnapshotSplit
TieringSnapshotSplit tieringSnapshotSplit =
- new TieringSnapshotSplit(tablePath, tableBucket, "partition1",
0L, 200L, 10);
+ new TieringSnapshotSplit(
+ tablePath, tableBucket, "partition1", 0L, 200L, 10, 0,
1000L);
+ tieringSnapshotSplit.skipCurrentRound();
TieringSplitState tieringSnapshotSplitState = new
TieringSplitState(tieringSnapshotSplit);
-
assertThat(tieringSnapshotSplitState.toSourceSplit()).isEqualTo(tieringSnapshotSplit);
+ TieringSnapshotSplit restoredSplit =
+ (TieringSnapshotSplit)
tieringSnapshotSplitState.toSourceSplit();
+ assertThat(restoredSplit).isEqualTo(tieringSnapshotSplit);
+ assertThat(restoredSplit.shouldSkipCurrentRound()).isTrue();
+ assertThat(restoredSplit.getSplitIndex()).isZero();
+ assertThat(restoredSplit.getTieringRoundTimestamp()).isEqualTo(1000L);
}
@Test
@@ -48,8 +55,13 @@ class TieringSplitStateTest {
// verify conversion between TieringLogSplitState and TieringLogSplit
TieringLogSplit tieringLogSplit =
- new TieringLogSplit(tablePath, tableBucket, "partition1",
100L, 200L, 20);
+ new TieringLogSplit(tablePath, tableBucket, "partition1",
100L, 200L, 20, 1, 2000L);
+ tieringLogSplit.skipCurrentRound();
TieringSplitState tieringLogSplitState = new
TieringSplitState(tieringLogSplit);
-
assertThat(tieringLogSplitState.toSourceSplit()).isEqualTo(tieringLogSplit);
+ TieringLogSplit restoredSplit = (TieringLogSplit)
tieringLogSplitState.toSourceSplit();
+ assertThat(restoredSplit).isEqualTo(tieringLogSplit);
+ assertThat(restoredSplit.shouldSkipCurrentRound()).isTrue();
+ assertThat(restoredSplit.getSplitIndex()).isEqualTo(1);
+ assertThat(restoredSplit.getTieringRoundTimestamp()).isEqualTo(2000L);
}
}
