eye-gu opened a new pull request, #10713:
URL: https://github.com/apache/paimon/pull/10713
Closes #7395.
### Purpose
`FlinkSourceBuilder` can only produce a `DataStream` through `build()` and
`buildForRow()`. Users who need a plain `Source` instance — for example to
bootstrap a pipeline with historical data from a bounded Paimon source and
continue streaming within one Flink `HybridSource` — currently have no way to
obtain one.
This PR adds `FlinkSourceBuilder.buildSource()`, returning a
`Source<RowData, ?, ?>` wrapped in `PaimonDataStreamSource` to preserve lineage
(#8001). Unlike `build()`, no `StreamExecutionEnvironment` is required:
boundedness is either inferred from the environment set via `env()` or must be
given explicitly with `sourceBounded(...)`.
`build()` and `buildSource()` share a single private construction path
(`createSource()`), and the consumer-id validation moved into
`checkConsumerId()` so both entry points keep it. For `build()` this is a pure
refactor: the extra checks inside `createSource()` 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.enabled` and exactly-once consumer mode rely on other
topologies and throw `UnsupportedOperationException` pointing to `build()`. The
exception is postpone merge-on-read with `scan.mode=compacted-full`, which
reads through the static path, mirroring the existing fall-through in `build()`
(#10521).
The javadoc shows a `HybridSource` example and documents the pitfalls: an
explicit `sourceBounded(true)` is needed when the bounded segment runs in a
streaming job; there is no snapshot hand-off between segments, so
`scan.mode=latest` may skip changes committed around the switch while the
default `latest-full` re-reads and duplicates the bounded segment; and
`env.fromSource(...)` requires an explicit `TypeInformation` because
`HybridSource` is not `ResultTypeQueryable`.
### Tests
`FlinkSourceBuilderTest` (26 tests): boundedness via `sourceBounded(...)`,
via env inference, missing boundedness → error; the four unsupported modes
rejected; postpone merge-on-read with `scan.mode=compacted-full` still builds
through the static path in both `buildSource()` and `build()` — locking the new
checks to `build()`'s existing fall-through behavior.
`HybridSourceITCase`: a streaming job whose `HybridSource` embeds a bounded
`buildSource()` segment followed by a continuous one. The continuous segment
filters `a <= 2`, so rows 1–2 can only come from the bounded segment;
`DISTINCT` tolerates the `latest-full` re-read overlap. All tests passed
locally:
```shell
mvn -o -pl paimon-flink/paimon-flink-common -am -Pflink1 \
-DfailIfNoTests=false -DwildcardSuites=none \
'-Dtest=FlinkSourceBuilderTest,HybridSourceITCase' 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
