Repository navigation
Conversation
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.
Closes #7395.
Purpose
FlinkSourceBuildercan only produce aDataStreamthroughbuild()andbuildForRow(). Users who need a plainSourceinstance — for example to bootstrap a pipeline with historical data from a bounded Paimon source and continue streaming within one FlinkHybridSource— currently have no way to obtain one.This PR adds
FlinkSourceBuilder.buildSource(), returning aSource<RowData, ?, ?>wrapped inPaimonDataStreamSourceto preserve lineage (#8001). Unlikebuild(), noStreamExecutionEnvironmentis required: boundedness is either inferred from the environment set viaenv()or must be given explicitly withsourceBounded(...).build()andbuildSource()share a single private construction path (createSource()), and the consumer-id validation moved intocheckConsumerId()so both entry points keep it. Forbuild()this is a pure refactor: the extra checks insidecreateSource()are unreachable from that path.Only the default static and continuous scan paths are supported by
buildSource(); postpone merge-on-read,scan.dedicated-split-generation,source.checkpoint-align.enabledand exactly-once consumer mode rely on other topologies and throwUnsupportedOperationExceptionpointing tobuild(). The exception is postpone merge-on-read withscan.mode=compacted-full, which reads through the static path, mirroring the existing fall-through inbuild()(#10521).The javadoc shows a
HybridSourceexample and documents the pitfalls: an explicitsourceBounded(true)is needed when the bounded segment runs in a streaming job; there is no snapshot hand-off between segments, soscan.mode=latestmay skip changes committed around the switch while the defaultlatest-fullre-reads and duplicates the bounded segment; andenv.fromSource(...)requires an explicitTypeInformationbecauseHybridSourceis notResultTypeQueryable.Tests
FlinkSourceBuilderTest(26 tests): boundedness viasourceBounded(...), via env inference, missing boundedness → error; the four unsupported modes rejected; postpone merge-on-read withscan.mode=compacted-fullstill builds through the static path in bothbuildSource()andbuild()— locking the new checks tobuild()'s existing fall-through behavior.HybridSourceITCase: a streaming job whoseHybridSourceembeds a boundedbuildSource()segment followed by a continuous one. The continuous segment filtersa <= 2, so rows 1–2 can only come from the bounded segment;DISTINCTtolerates thelatest-fullre-read overlap. All tests passed locally: