[lake/hudi] Introduce tag field for tiering splits

[fhan688]

Purpose

Linked issue: #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.

[Copilot]

Pull request overview

This PR adds a “tag” concept to tiering splits and exposes it via WriterInitContext#tag() so lake writers can coordinate work across all splits in the same tiering round.

Changes:

• Add WriterInitContext#tag() (default empty) and pass tiering split tags through Flink tiering writer initialization.
• Extend tiering split types/state/serialization to carry an immutable tag end-to-end.
• Generate per-round tags when enumerating splits and add tests for tag serialization/state conversion/assignment.

Reviewed changes

Copilot reviewed 13 out of 13 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
fluss-common/src/main/java/org/apache/fluss/lake/writer/WriterInitContext.java	Adds default tag() API for lake writer initialization.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringWriterInitContext.java	Implements WriterInitContext#tag() and stores tag in init context.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java	Passes split tag into TieringWriterInitContext when creating lake writers.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/state/TieringSplitState.java	Preserves tag when converting split state back to source split.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplit.java	Introduces tag field and tag-aware copy API on base split.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSnapshotSplit.java	Adds tag-aware constructors/copy/toString for snapshot splits.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringLogSplit.java	Adds tag-aware constructors/copy/toString for log splits.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializer.java	Bumps serializer version and includes tag in split (de)serialization.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java	Assigns per-round tags to generated splits.
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/state/TieringSplitStateTest.java	Verifies tag is preserved through split state conversion.
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializerTest.java	Adds tag serialization/deserialization test coverage.
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java	Asserts tiering-round tag assignment behavior in enumerator tests.
fluss-test-coverage/src/test/java/org/apache/fluss/lake/writer/WriterInitContextCoverageTest.java	Coverage test to ensure default WriterInitContext#tag() is empty.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[luoyuxia]

@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).

[luoyuxia]

@fhan688 Thanks for the pr.

[fhan688]

@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.

[luoyuxia]

@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

[luoyuxia]

@fhan688 Thanks. +1