[tiering] Decouple common tiering class from flink module - #3554
Open
Yohahaha wants to merge 5 commits into
Open
[tiering] Decouple common tiering class from flink module#3554Yohahaha wants to merge 5 commits into
Yohahaha wants to merge 5 commits into
Conversation
Yohahaha
force-pushed
the
spark-tiering-phase1
branch
2 times, most recently
from
July 30, 2026 07:38
4b9f5ff to
693b4c0
Compare
Contributor
Author
Contributor
There was a problem hiding this comment.
Pull request overview
This PR moves tiering “common” building blocks (splits, write result container, snapshot committer, and commit logic) out of the Flink connector layer into fluss-client, and adapts the Flink tiering source to use these engine-agnostic types via a Flink-specific wrapper (FlinkTieringSplit). This supports the longer-term goal of enabling additional compute backends (e.g., Spark tiering) without duplicating core tiering protocol code.
Changes:
- Relocate tiering split types and tiering init/commit helpers into
fluss-clientand update downstream imports/usages. - Introduce
TieringCommitterandTieringCommitResultinfluss-clientand wire Flink’sTieringCommitOperatorto use it. - Add
FlinkTieringSplitadapter and update Flink tiering source/enumerator/serde/state + tests to use it.
Reviewed changes
Copilot reviewed 33 out of 33 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/utils/DvTableReadableSnapshotRetrieverTest.java | Switch test to use snapshot committer moved to fluss-client. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringSplitReaderTest.java | Update tests to use FlinkTieringSplit wrapper and client tiering split types. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringSourceReaderTest.java | Update reader tests to wrap client TieringLogSplit in FlinkTieringSplit. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TableBucketWriteResultSerializerTest.java | Update imports to use TableBucketWriteResult from fluss-client. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/state/TieringSplitStateTest.java | Update split state tests for FlinkTieringSplit + client split types. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializerTest.java | Adapt serde tests to serialize/deserialize FlinkTieringSplit and validate wrapper behavior. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumeratorTest.java | Update enumerator tests to use client split types + Flink wrapper. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorTest.java | Update imports to client FlussTableLakeSnapshotCommitter + TableBucketWriteResult. |
| fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/committer/FlussTableLakeSnapshotCommitterTest.java | Update test to use committer from fluss-client and keep compatibility coverage. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java | Switch SplitReader to Flink wrapper split type and unwrap to engine-agnostic split for processing. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceReader.java | Switch source reader split type to FlinkTieringSplit and update state initialization. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSourceFetcherManager.java | Propagate split type change to FlinkTieringSplit. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSource.java | Update Flink Source generics to use FlinkTieringSplit. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TableBucketWriteResultTypeInfo.java | Update to client TableBucketWriteResult. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TableBucketWriteResultSerializer.java | Update to client TableBucketWriteResult. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TableBucketWriteResultEmitter.java | Update to client TableBucketWriteResult. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/state/TieringSplitState.java | Update split state to store/restore FlinkTieringSplit while preserving underlying split metadata. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitSerializer.java | Update split serializer to operate on FlinkTieringSplit while encoding engine-agnostic split data. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/TieringSplitGenerator.java | Update imports to use client TieringSplit types. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/split/FlinkTieringSplit.java | New adapter implementing Flink SourceSplit around engine-agnostic TieringSplit. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/enumerator/TieringSourceEnumerator.java | Switch enumerator split type to FlinkTieringSplit while generating engine-agnostic splits. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperatorFactory.java | Update to client TableBucketWriteResult. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/TieringCommitOperator.java | Replace embedded commit protocol with reusable TieringCommitter from fluss-client. |
| fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/committer/CommittableMessageTypeInfo.java | Update to client TableBucketWriteResult. |
| fluss-client/src/main/java/org/apache/fluss/client/tiering/TieringWriterInitContext.java | Move writer init context into fluss-client tiering package. |
| fluss-client/src/main/java/org/apache/fluss/client/tiering/TieringSplit.java | Move engine-agnostic split base into fluss-client and remove Flink SourceSplit dependency. |
| fluss-client/src/main/java/org/apache/fluss/client/tiering/TieringSnapshotSplit.java | Move snapshot split into fluss-client tiering package. |
| fluss-client/src/main/java/org/apache/fluss/client/tiering/TieringLogSplit.java | Move log split into fluss-client tiering package. |
| fluss-client/src/main/java/org/apache/fluss/client/tiering/TieringCommitterInitContext.java | Move committer init context into fluss-client tiering package. |
| fluss-client/src/main/java/org/apache/fluss/client/tiering/TieringCommitter.java | New engine-agnostic committer extracting the shared lake+Fluss commit protocol. |
| fluss-client/src/main/java/org/apache/fluss/client/tiering/TieringCommitResult.java | New DTO for returning committable + stats from engine-agnostic commit logic. |
| fluss-client/src/main/java/org/apache/fluss/client/tiering/TableBucketWriteResult.java | Move write-result container into fluss-client tiering package. |
| fluss-client/src/main/java/org/apache/fluss/client/tiering/FlussTableLakeSnapshotCommitter.java | Move snapshot committer into fluss-client and adjust visibility for cross-module usage/testing. |
Comments suppressed due to low confidence (2)
fluss-client/src/main/java/org/apache/fluss/client/tiering/TieringSplit.java:32
- TieringSplit is Serializable but doesn’t declare a serialVersionUID, while other Serializable classes in this package/module do (e.g., TableBucketWriteResult). This can cause unintended InvalidClassException if the class evolves and serialized splits are persisted or transferred across versions (relevant for engine-agnostic tiering splits).
fluss-client/src/main/java/org/apache/fluss/client/tiering/FlussTableLakeSnapshotCommitter.java:202 - The low-level commit(tableId, snapshotId, …) overload is public but appears intended for internal wiring / tests (similar to prepareLakeSnapshot(), addLogEndOffsets(), getCoordinatorGateway()). Marking it
@VisibleForTestingclarifies it’s not meant as a general public API entry point.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
169
to
173
| } else { | ||
| // deserialize starting offset | ||
| long startingOffset = in.readLong(); | ||
| // deserialize starting offset | ||
| long stoppingOffset = in.readLong(); |
Contributor
Author
There was a problem hiding this comment.
Good catch, though note this is pre-existing behaviour rather than something introduced by this refactor. On main the deserialize method already had the same if (splitKind == TIERING_SNAPSHOT_SPLIT_FLAG) { ... } else { ... } shape; this PR only reshaped the two return new ... statements into an assignment plus a single new FlinkTieringSplit(...) wrap at the end, which is why the diff surfaced these lines.
That said, the underlying concern is valid and worth fixing here, so I've addressed it in d92f2a3:
deserializenow matchesTIERING_LOG_SPLIT_FLAGexplicitly and throwsIOException("Unknown split kind " + splitKind)for anything else, instead of falling through to the log-split branch.- Removed the serializer's private copies of the two flag constants in favour of the public ones on
TieringSplit. PreviouslyTIERING_LOG_SPLIT_FLAGwas declared but never compared against, which was exactly the symptom of the missing branch; now both flags have a single definition and both are actually used. - Added
testDeserializeUnknownSplitKindFailsFast, which tampers with the split-kind byte and asserts theIOException. I verified it fails against the previouselsefallback and passes with the fix.
One note on the motivation: the "corrupted splitKind" scenario isn't really the risk here, since these bytes only travel from TieringSourceEnumerator to TieringSource within a single job over Flink's own transport (the class javadoc explicitly states it does not need to consider compatibility). The real value is guarding against a future third split kind being added to splitKind()/serialize() while deserialize() is forgotten - that would otherwise silently misread the field layout. Failing fast is strictly better, so the change is worth it either way.
Extract engine-agnostic split generation from fluss-flink so that other engines (e.g. Spark tiering) can reuse it. Replace FlinkRuntimeException with FlussRuntimeException to drop the Flink dependency.
TieringSplitSerializer#deserialize treated every non-snapshot splitKind as a log split, so an unknown kind was silently deserialized with the wrong field layout instead of failing fast. Match explicitly against TIERING_LOG_SPLIT_FLAG and throw an IOException otherwise. Also drop the serializer's private copies of the split kind flags in favour of the public constants on TieringSplit, so both flags now have exactly one definition and are both actually used.
Yohahaha
force-pushed
the
spark-tiering-phase1
branch
from
August 18, 2026 03:10
d92f2a3 to
5339daf
Compare
Contributor
Author
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
Linked issue: close #xxx
Phase 1 of #2985
Brief change log
Tests
API and Format
Documentation