You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "fhan688 (via GitHub)" <gi...@apache.org> on 2026/06/21 03:18:01 UTC

[PR] [lake/hudi] Introduce tag field for tiering splits [fluss]

fhan688 opened a new pull request, #3503:
URL: https://github.com/apache/fluss/pull/3503

   ### Purpose
   
     Linked issue: https://github.com/apache/fluss/issues/3279
   
     This PR introduces split tags for the tiering source so lake writers can coordinate all splits that belong to the same tiering round. Each generated tiering round has one first-split tag and a shared round tag for the  remaining splits.
   
   ### Brief change log
   
     - Add `WriterInitContext#tag()` as a default method.
     - Add immutable tag support to `TieringSplit`, `TieringLogSplit`, and `TieringSnapshotSplit`.
     - Populate tiering-round tags when generating splits.
     - Preserve split tags through copy, split state conversion, and split serialization.
     - Pass split tags from `TieringSplitReader` to `TieringWriterInitContext`.
     - Add tests for tag serialization, state conversion, and tag assignment.
   
   ### Tests
   
     - `git diff --check`
     - `mvn -pl fluss-flink/fluss-flink-common -am -DskipITs -Dcheckstyle.skip=true -DfailIfNoTests=false -Dtest=TieringSplitSerializerTest,TieringSplitStateTest test`
   
   ### API and Format
   
     - Adds a source-compatible default method `WriterInitContext#tag()`.
     - Updates internal `TieringSplitSerializer` version to include split tags.
     - Does not change Fluss table data format or lake storage format.
   
   ### Documentation
   
     No user-facing documentation changes.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [lake/hudi] Introduce tag field for tiering splits [fluss]

Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
fhan688 commented on code in PR #3503:
URL: https://github.com/apache/fluss/pull/3503#discussion_r3448554621


##########
fluss-test-coverage/src/test/java/org/apache/fluss/lake/writer/WriterInitContextCoverageTest.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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 WriterInitContextCoverageTest {

Review Comment:
   > seems the test doesn't have much meaning, we can remove this test.
   
   Thanks for pointing this out. I agree this standalone test is not very meaningful from a behavior perspective.
   
     I added it only because the new default method in `WriterInitContext` triggered the JaCoCo per-class coverage check. I’ll remove this coverage-only test and move the coverage to a more natural place, or avoid introducing
     a default method that needs such artificial coverage if the API is updated based on the split metadata redesign.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [lake/hudi] Introduce tag field for tiering splits [fluss]

Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #3503:
URL: https://github.com/apache/fluss/pull/3503#issuecomment-4762198710

   @fhan688 Thanks for the PR! I spent quite a while trying to understand the purpose of the `tag` field, and I have some concerns about this approach.
   
   The current `tag` feels a bit hacky and hard to follow. It's a free-form `String` that overloads two meanings via the `"firstSplit-"` prefix plus a timestamp, and the only way to interpret it is by string-parsing the prefix. I can imagine the Hudi writer will then have to parse this `tag` string as well to figure out the first split, which ends up being fairly hard-coded.
   
   I'm not very familiar with Hudi, so please correct me if I'm wrong. But based on FIP-24, my understanding of what we actually need is:
   
   > the Hudi writer needs to know which split is the first split, so that only the `LakeWriter` handling the first split generates the global `instantTime`, and the other writers retrieve it via `CkpMetadata`.
   
   If that's the case, instead of encoding this into a string `tag`, what if we introduce a typed field `splitIndex` in `TieringSplit` representing the index of the current split within this tiering round? It pairs naturally with the existing `numberOfSplits` (`splitIndex` / `numberOfSplits`, similar to Flink's `getIndexOfThisSubtask()` / `getNumberOfParallelSubtasks()`), so it reads much less weird. Then the coordinator is simply the writer whose split has `splitIndex == 0`, and that writer generates the global `instantTime`.
   
   This avoids the string parsing / hard-coding, makes the split immutable (no `setTag()`), and the "exactly one first split" property becomes structural (contiguous indices ⇒ exactly one `0`).
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [lake/hudi] Introduce tag field for tiering splits [fluss]

Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
fhan688 commented on PR #3503:
URL: https://github.com/apache/fluss/pull/3503#issuecomment-4762246552

   > @fhan688 Thanks for the PR! I spent quite a while trying to understand the purpose of the `tag` field, and I have some concerns about this approach.
   > 
   > The current `tag` feels a bit hacky and hard to follow. It's a free-form `String` that overloads two meanings via the `"firstSplit-"` prefix plus a timestamp, and the only way to interpret it is by string-parsing the prefix. I can imagine the Hudi writer will then have to parse this `tag` string as well to figure out the first split, which ends up being fairly hard-coded.
   > 
   > I'm not very familiar with Hudi, so please correct me if I'm wrong. But based on FIP-24, my understanding of what we actually need is:
   > 
   > > the Hudi writer needs to know which split is the first split, so that only the `LakeWriter` handling the first split generates the global `instantTime`, and the other writers retrieve it via `CkpMetadata`.
   > 
   > If that's the case, instead of encoding this into a string `tag`, what if we introduce a typed field `splitIndex` in `TieringSplit` representing the index of the current split within this tiering round? It pairs naturally with the existing `numberOfSplits` (`splitIndex` / `numberOfSplits`, similar to Flink's `getIndexOfThisSubtask()` / `getNumberOfParallelSubtasks()`), so it reads much less weird. Then the coordinator is simply the writer whose split has `splitIndex == 0`, and that writer generates the global `instantTime`.
   > 
   > This avoids the string parsing / hard-coding, makes the split immutable (no `setTag()`), and the "exactly one first split" property becomes structural (contiguous indices ⇒ exactly one `0`).
   
   Thanks for the detailed suggestion. I agree that overloading `tag` with both the first-split marker and a timestamp makes the protocol hard to follow.
   
     I double-checked the Hudi tiering writer side. Besides identifying the writer that creates the global Hudi instant, the writer also needs a round-level timestamp to avoid consuming a stale pending instant from
     `CkpMetadata`. In the current Hudi writer flow, non-first writers wait for `CkpMetadata.lastPendingInstant()`, and compare it with the split-generated timestamp to make sure the pending instant belongs to the current
     tiering round. So `splitIndex` alone can cover the first-writer election, but it cannot replace the round timestamp semantics.
   
     I’ll update the design to make these two meanings explicit instead of encoding them into a free-form string:
   
     - introduce a typed `splitIndex` in `TieringSplit`, where `splitIndex == 0` identifies the writer responsible for creating the Hudi instant;
     - introduce an explicit round timestamp field, for example `tieringRoundTimestamp`, to let Hudi writers validate the pending instant from `CkpMetadata`;
     - remove the `"firstSplit-"` string parsing and avoid mutable `setTag()`.
   
     This should keep the first-split property structural while preserving the timestamp requirement needed by the Hudi writer.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [lake/hudi] Introduce tag field for tiering splits [fluss]

Posted by "fhan688 (via GitHub)" <gi...@apache.org>.
fhan688 commented on code in PR #3503:
URL: https://github.com/apache/fluss/pull/3503#discussion_r3448553735


##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializer.java:
##########
@@ -39,15 +39,15 @@ public class TieringSplitSerializer implements SimpleVersionedSerializer<Tiering
 
     public static final TieringSplitSerializer INSTANCE = new TieringSplitSerializer();
 
-    private static final int VERSION_0 = 0;
+    private static final int VERSION_1 = 1;

Review Comment:
   > we can avoid bump the version number since the tiering job is statelsss and the serializer is just for network transmission.
   
   Thanks for the explanation. That makes sense.
   
     Since the tiering source is stateless and this serializer is only used for split transmission, I’ll avoid bumping the serializer version here and keep it as version 0 while carrying the new fields in the current
     serialization format.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [lake/hudi] Introduce tag field for tiering splits [fluss]

Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #3503:
URL: https://github.com/apache/fluss/pull/3503


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [lake/hudi] Introduce tag field for tiering splits [fluss]

Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #3503:
URL: https://github.com/apache/fluss/pull/3503#discussion_r3448563773


##########
fluss-test-coverage/src/test/java/org/apache/fluss/lake/writer/WriterInitContextCoverageTest.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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 WriterInitContextCoverageTest {

Review Comment:
   we can just ignore the test coverage for WriterInitContext.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [lake/hudi] Introduce tag field for tiering splits [fluss]

Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3503:
URL: https://github.com/apache/fluss/pull/3503#discussion_r3447939674


##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/state/TieringSplitState.java:
##########
@@ -54,7 +55,8 @@ public TieringSplit toSourceSplit() {
                     split.getPartitionName(),
                     split.getStartingOffset(),
                     split.getStoppingOffset(),
-                    split.getNumberOfSplits());
+                    split.getNumberOfSplits(),
+                    split.getTag());

Review Comment:
   `TieringSplitState#toSourceSplit()` drops the `skipCurrentRound` flag for log splits as well. This can change behavior after converting split state (the reader treats skipped splits as empty). Preserve the flag when reconstructing the `TieringLogSplit`.



##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java:
##########
@@ -494,6 +495,24 @@ private List<TieringSplit> populateNumberOfTieringSplits(List<TieringSplit> tier
                 .collect(Collectors.toList());
     }
 
+    private List<TieringSplit> populateTagsForTieringSplits(List<TieringSplit> tieringSplits) {
+        if (tieringSplits.isEmpty()) {
+            return tieringSplits;
+        }
+        String tieringRoundId = String.valueOf(System.currentTimeMillis());

Review Comment:
   Using `System.currentTimeMillis()` as the tiering-round id can produce duplicate tags if two rounds are generated within the same millisecond (e.g., under fast retries/tests). Since the tag is meant to coordinate a tiering round, it should be unique; using a UUID avoids collisions.



##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/state/TieringSplitState.java:
##########
@@ -45,7 +45,8 @@ public TieringSplit toSourceSplit() {
                     split.getPartitionName(),
                     split.getSnapshotId(),
                     split.getLogOffsetOfSnapshot(),
-                    split.getNumberOfSplits());
+                    split.getNumberOfSplits(),
+                    split.getTag());

Review Comment:
   `TieringSplitState#toSourceSplit()` drops the `skipCurrentRound` flag. That flag is part of split semantics (used by `TieringSplitReader#handleSplitsChanges`), so reconstructing a split from state should preserve it to avoid accidentally processing a split that was marked to be skipped.



##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java:
##########
@@ -31,12 +31,15 @@ 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 String EMPTY_TAG = "";
+    public static final String FIRST_SPLIT_TAG_PREFIX = "firstSplit-";
 
     protected static final int UNKNOWN_NUMBER_OF_SPLITS = -1;
 
     protected final TablePath tablePath;
     protected final TableBucket tableBucket;
     @Nullable protected final String partitionName;
+    protected final String tag;
 

Review Comment:
   A new `tag` field is introduced on `TieringSplit`, but the class’ `equals()` / `hashCode()` implementations (further down) do not incorporate it. If tags are intended to be part of split state (they're serialized/copied and passed into writer init), ignoring them can cause different tagged splits to compare equal and be deduplicated in sets/maps, and can hide tag regressions in tests. Consider including `tag` in `equals()` / `hashCode()`, or documenting explicitly why it is intentionally excluded.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [lake/hudi] Introduce tag field for tiering splits [fluss]

Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #3503:
URL: https://github.com/apache/fluss/pull/3503#issuecomment-4762269561

   > > @fhan688 Thanks for the PR! I spent quite a while trying to understand the purpose of the `tag` field, and I have some concerns about this approach.
   > > The current `tag` feels a bit hacky and hard to follow. It's a free-form `String` that overloads two meanings via the `"firstSplit-"` prefix plus a timestamp, and the only way to interpret it is by string-parsing the prefix. I can imagine the Hudi writer will then have to parse this `tag` string as well to figure out the first split, which ends up being fairly hard-coded.
   > > I'm not very familiar with Hudi, so please correct me if I'm wrong. But based on FIP-24, my understanding of what we actually need is:
   > > > the Hudi writer needs to know which split is the first split, so that only the `LakeWriter` handling the first split generates the global `instantTime`, and the other writers retrieve it via `CkpMetadata`.
   > > 
   > > 
   > > If that's the case, instead of encoding this into a string `tag`, what if we introduce a typed field `splitIndex` in `TieringSplit` representing the index of the current split within this tiering round? It pairs naturally with the existing `numberOfSplits` (`splitIndex` / `numberOfSplits`, similar to Flink's `getIndexOfThisSubtask()` / `getNumberOfParallelSubtasks()`), so it reads much less weird. Then the coordinator is simply the writer whose split has `splitIndex == 0`, and that writer generates the global `instantTime`.
   > > This avoids the string parsing / hard-coding, makes the split immutable (no `setTag()`), and the "exactly one first split" property becomes structural (contiguous indices ⇒ exactly one `0`).
   > 
   > Thanks for the detailed suggestion. I agree that overloading `tag` with both the first-split marker and a timestamp makes the protocol hard to follow.
   > 
   > I double-checked the Hudi tiering writer side. Besides identifying the writer that creates the global Hudi instant, the writer also needs a round-level timestamp to avoid consuming a stale pending instant from `CkpMetadata`. In the current Hudi writer flow, non-first writers wait for `CkpMetadata.lastPendingInstant()`, and compare it with the split-generated timestamp to make sure the pending instant belongs to the current tiering round. So `splitIndex` alone can cover the first-writer election, but it cannot replace the round timestamp semantics.
   > 
   > I’ll update the design to make these two meanings explicit instead of encoding them into a free-form string:
   > 
   > * introduce a typed `splitIndex` in `TieringSplit`, where `splitIndex == 0` identifies the writer responsible for creating the Hudi instant;
   > * introduce an explicit round timestamp field, for example `tieringRoundTimestamp`, to let Hudi writers validate the pending instant from `CkpMetadata`;
   > * remove the `"firstSplit-"` string parsing and avoid mutable `setTag()`.
   > 
   > This should keep the first-split property structural while preserving the timestamp requirement needed by the Hudi writer.
   
   Thanks for digging into the Hudi writer side — that makes sense


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [lake/hudi] Introduce tag field for tiering splits [fluss]

Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #3503:
URL: https://github.com/apache/fluss/pull/3503#discussion_r3448508221


##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializer.java:
##########
@@ -39,15 +39,15 @@ public class TieringSplitSerializer implements SimpleVersionedSerializer<Tiering
 
     public static final TieringSplitSerializer INSTANCE = new TieringSplitSerializer();
 
-    private static final int VERSION_0 = 0;
+    private static final int VERSION_1 = 1;

Review Comment:
   we can avoid bump the version number since the tiering job is statelsss and the serializer is just for network transmission.



##########
fluss-test-coverage/src/test/java/org/apache/fluss/lake/writer/WriterInitContextCoverageTest.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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 WriterInitContextCoverageTest {

Review Comment:
   seems the test doesn't have much meaning, we can remove this test.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org