Skip to content

[flink] Expose Flink Source in FlinkSourceBuilder API #10713

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
eye-gu wants to merge 1 commit into
base: master
Choose a base branch
from
+409 −37
Open

[flink] Expose Flink Source in FlinkSourceBuilder API #10713

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
161 changes: 128 additions & 33 deletions .../paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -217,20 +217,68 @@ private ReadBuilder createReadBuilder(@Nullable org.apache.paimon.types.RowType
return readBuilder.dropStats();
}

private DataStream<RowData> buildStaticFileSource() {
Options options = Options.fromMap(table.options());
return toDataStream(
new StaticFileStoreSource(
createReadBuilder(projectedRowType()),
limit,
options.get(FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_BATCH_SIZE),
options.get(FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_ASSIGN_MODE),
dynamicPartitionFilteringInfo,
outerProject(),
SplitWeightUtils.splitWeightFunc(options),
null,
options.get(CoreOptions.BLOB_AS_DESCRIPTOR),
skipPreloadTargetSnapshot));
private Source<RowData, ?, ?> createSource() {
checkConsumerId();
if (sourceBounded == null) {
throw new IllegalArgumentException(
"Cannot determine boundedness of source: please provide the "
+ "StreamExecutionEnvironment via env() or specify sourceBounded().");
}

if (sourceBounded) {
if (PostponeMergeOnRead.usesCustomSource(table)) {
throw new UnsupportedOperationException(
"Postpone merge-on-read relies on a custom multi-operator topology, "
+ "please use build() instead.");
}
if (conf.get(FlinkConnectorOptions.SCAN_DEDICATED_SPLIT_GENERATION)) {
throw new UnsupportedOperationException(
"scan.dedicated-split-generation relies on a monitor topology, "
+ "please use build() instead.");
}
Options options = Options.fromMap(table.options());
return new StaticFileStoreSource(
createReadBuilder(projectedRowType()),
limit,
options.get(FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_BATCH_SIZE),
options.get(FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_ASSIGN_MODE),
dynamicPartitionFilteringInfo,
outerProject(),
SplitWeightUtils.splitWeightFunc(options),
null,
options.get(CoreOptions.BLOB_AS_DESCRIPTOR),
skipPreloadTargetSnapshot);
}

TableScanUtils.streamingReadingValidate(table);
if (conf.get(FlinkConnectorOptions.SOURCE_CHECKPOINT_ALIGN_ENABLED)) {
throw new UnsupportedOperationException(
"Checkpoint align mode requires validation against the environment, "
+ "please use build() instead.");
}
if (conf.contains(CoreOptions.CONSUMER_ID)
&& conf.get(CoreOptions.CONSUMER_CONSISTENCY_MODE)
== CoreOptions.ConsumerMode.EXACTLY_ONCE) {
throw new UnsupportedOperationException(
"Exactly-once consumer mode relies on a monitor topology, "
+ "please use build() instead.");
}
return new ContinuousFileStoreSource(
createReadBuilder(projectedRowType()),
table.options(),
limit,
unordered,
outerProject());
}

private void checkConsumerId() {
if (conf.contains(CoreOptions.CONSUMER_ID)
&& !conf.contains(CoreOptions.CONSUMER_EXPIRATION_TIME)) {
throw new IllegalArgumentException(
"You need to configure 'consumer.expiration-time' (ALTER TABLE) and restart your write job for it"
+ " to take effect, when you need consumer-id feature. This is to prevent consumers from leaving"
+ " too many snapshots that could pose a risk to the file system.");
}
}

private @Nullable DataStream<RowData> buildPostponeMergeSource() {
Expand Down Expand Up @@ -330,16 +378,6 @@ private boolean inferScanParallelism() {
return conf.get(FlinkConnectorOptions.INFER_SCAN_PARALLELISM);
}

private DataStream<RowData> buildContinuousFileSource() {
return toDataStream(
new ContinuousFileStoreSource(
createReadBuilder(projectedRowType()),
table.options(),
limit,
unordered,
outerProject()));
}

private DataStream<RowData> buildAlignedContinuousFileSource() {
assertStreamingConfigurationForAlignMode(env);
return toDataStream(
Expand Down Expand Up @@ -418,18 +456,75 @@ public DataStream<Row> buildForRow() {
return result;
}

/**
* Build a {@link Source} for APIs requiring a plain source instance, such as Flink's {@code
* HybridSource}. Different from {@link #build()}, no {@link StreamExecutionEnvironment} is
* required: boundedness is either inferred from the environment set via {@link
* #env(StreamExecutionEnvironment)} or must be specified with {@link #sourceBounded(boolean)}.
*
* <p>When a Paimon bounded source is the first (historical) segment of a {@code HybridSource}
* in a streaming job, {@code sourceBounded(true)} must be set explicitly, because {@link
* #env(StreamExecutionEnvironment)} infers unboundedness from a streaming environment.
*
* <p>Example:
*
* <pre>{@code
* Source<RowData, ?, ?> bounded =
* new FlinkSourceBuilder(table).sourceBounded(true).buildSource();
* HybridSource<RowData> hybrid =
* HybridSource.builder(bounded)
* .addSource(
* context ->
* new FlinkSourceBuilder(table)
* .sourceBounded(false)
* .buildSource(),
* Boundedness.CONTINUOUS_UNBOUNDED)
* .build();
*
* // env is the StreamExecutionEnvironment the job runs on
* env.fromSource(
* hybrid,
* WatermarkStrategy.noWatermarks(),
* "hybrid",
* InternalTypeInfo.of(LogicalTypeConversion.toLogicalType(table.rowType())));
* }</pre>
*
* <p>The second (streaming) segment is created when {@code HybridSource} switches to it, and
* its start position is resolved at that moment. There is no hand-off with the bounded segment:
* {@code scan.mode=latest} reads only newer changes, so changes committed after the bounded
* segment planned its splits but before the switch are skipped; the default {@code latest-full}
* re-reads the whole table first, so nothing is lost but the bounded segment's data is
* duplicated.
*
* <p>Only the default static and continuous scan paths are supported; postpone merge-on-read,
* {@code scan.dedicated-split-generation}, {@code source.checkpoint-align.enabled} and
* exactly-once consumer mode rely on other topologies and require {@link #build()}. The
* exception is postpone merge-on-read with {@code scan.mode=compacted-full}, which reads
* through the static path.
*
* <p>{@link #sourceName(String)}, {@link #watermarkStrategy(WatermarkStrategy)}, {@link
* #sourceParallelism(Integer)} and the uid suffix option are ignored by this method; apply the
* source name and the watermark strategy in your {@code env.fromSource(...)} call, and the
* parallelism and uid on the resulting {@code DataStreamSource}. A job that relies on the uid
* suffix for state compatibility will not restore from state when switching to this method.
*
* <p>{@code HybridSource} does not implement {@code ResultTypeQueryable}, so the type
* information must be passed to {@code env.fromSource(...)} explicitly, for example {@code
* InternalTypeInfo.of(LogicalTypeConversion.toLogicalType(rowType))} where {@code rowType} is
* {@code table.rowType()}, or the projected row type if {@code projection(...)} is used. Note
* that {@code InternalTypeInfo} is a Flink internal API with no public alternative for {@code
* RowData} type information.
*/
public Source<RowData, ?, ?> buildSource() {
return new PaimonDataStreamSource<>(createSource(), table);
}

/** Build source {@link DataStream} with {@link RowData}. */
public DataStream<RowData> build() {
if (env == null) {
throw new IllegalArgumentException("StreamExecutionEnvironment should not be null.");
}
if (conf.contains(CoreOptions.CONSUMER_ID)
&& !conf.contains(CoreOptions.CONSUMER_EXPIRATION_TIME)) {
throw new IllegalArgumentException(
"You need to configure 'consumer.expiration-time' (ALTER TABLE) and restart your write job for it"
+ " to take effect, when you need consumer-id feature. This is to prevent consumers from leaving"
+ " too many snapshots that could pose a risk to the file system.");
}
checkConsumerId();

if (sourceBounded && PostponeMergeOnRead.configured(table)) {
DataStream<RowData> mergeSource = buildPostponeMergeSource();
Expand All @@ -442,7 +537,7 @@ public DataStream<RowData> build() {
if (conf.get(FlinkConnectorOptions.SCAN_DEDICATED_SPLIT_GENERATION)) {
return buildDedicatedSplitGenSource(true);
}
return buildStaticFileSource();
return toDataStream(createSource());
}
TableScanUtils.streamingReadingValidate(table);

Expand All @@ -459,7 +554,7 @@ public DataStream<RowData> build() {
== CoreOptions.ConsumerMode.EXACTLY_ONCE) {
return buildDedicatedSplitGenSource(false);
} else {
return buildContinuousFileSource();
return toDataStream(createSource());
}
}

Expand Down
150 changes: 146 additions & 4 deletions ...mon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkSourceBuilderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,9 @@
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.runtime.plugable.SerializationDelegate;
import org.apache.flink.streaming.api.datastream.DataStream;
Expand Down Expand Up @@ -99,11 +101,23 @@ private void initCatalog() throws Exception {
private Table createTable(
String tableName, boolean hasPrimaryKey, int bucketNum, boolean bucketAppendOrdered)
throws Exception {
return createTableWithOptions(
tableName,
hasPrimaryKey,
bucketNum,
"bucket-append-ordered",
String.valueOf(bucketAppendOrdered));
}

private Table createTableWithOptions(
String tableName, boolean hasPrimaryKey, int bucketNum, String... kvOptions)
throws Exception {
Schema.Builder schemaBuilder =
Schema.newBuilder()
.column("a", DataTypes.INT())
.option("bucket", bucketNum + "")
.option("bucket-append-ordered", String.valueOf(bucketAppendOrdered));
Schema.newBuilder().column("a", DataTypes.INT()).option("bucket", bucketNum + "");

for (int i = 0; i < kvOptions.length; i += 2) {
schemaBuilder.option(kvOptions[i], kvOptions[i + 1]);
}

if (hasPrimaryKey) {
schemaBuilder.primaryKey("a");
Expand Down Expand Up @@ -304,6 +318,134 @@ public void testBuildWrapsContinuousSourceWithPaimonDataStreamSource() throws Ex
assertThat(transformation.getSource()).isInstanceOf(PaimonDataStreamSource.class);
}

@Test
public void testBuildSourceWithoutEnv() throws Exception {
Table table = createTable("build_source", false, -1, true);

Source<RowData, ?, ?> bounded =
new FlinkSourceBuilder(table).sourceBounded(true).buildSource();
assertThat(bounded).isInstanceOf(PaimonDataStreamSource.class);
assertThat(bounded.getBoundedness()).isEqualTo(Boundedness.BOUNDED);

Source<RowData, ?, ?> unbounded =
new FlinkSourceBuilder(table).sourceBounded(false).buildSource();
assertThat(unbounded).isInstanceOf(PaimonDataStreamSource.class);
assertThat(unbounded.getBoundedness()).isEqualTo(Boundedness.CONTINUOUS_UNBOUNDED);
}

@Test
public void testBuildSourceInfersBoundednessFromEnv() throws Exception {
Table table = createTable("build_source_env", false, -1, true);

StreamExecutionEnvironment batchEnv = StreamExecutionEnvironment.getExecutionEnvironment();
batchEnv.setRuntimeMode(RuntimeExecutionMode.BATCH);
Source<RowData, ?, ?> bounded = new FlinkSourceBuilder(table).env(batchEnv).buildSource();
assertThat(bounded.getBoundedness()).isEqualTo(Boundedness.BOUNDED);

StreamExecutionEnvironment streamingEnv =
StreamExecutionEnvironment.getExecutionEnvironment();
streamingEnv.setRuntimeMode(RuntimeExecutionMode.STREAMING);
Source<RowData, ?, ?> unbounded =
new FlinkSourceBuilder(table).env(streamingEnv).buildSource();
assertThat(unbounded.getBoundedness()).isEqualTo(Boundedness.CONTINUOUS_UNBOUNDED);
}

@Test
public void testBuildSourceWithoutBoundedness() throws Exception {
Table table = createTable("build_source_no_boundedness", false, -1, true);

assertThatThrownBy(() -> new FlinkSourceBuilder(table).buildSource())
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("sourceBounded");
}

@Test
public void testBuildSourceRejectsPostponeMerge() throws Exception {
Table table =
createTableWithOptions(
"build_source_postpone_merge", true, -2, "postpone.merge-on-read", "true");

assertThatThrownBy(() -> new FlinkSourceBuilder(table).sourceBounded(true).buildSource())
.isInstanceOf(UnsupportedOperationException.class)
.hasMessageContaining("build()");
}

@Test
public void testBuildSourceSupportsPostponeMergeCompactedFull() throws Exception {
// COMPACTED_FULL intentionally falls back to the static source in build(),
// so buildSource() must not reject it either
Table table =
createTableWithOptions(
"build_source_postpone_merge_compacted_full",
true,
-2,
"postpone.merge-on-read",
"true",
"scan.mode",
"compacted-full");

Source<RowData, ?, ?> source =
new FlinkSourceBuilder(table).sourceBounded(true).buildSource();
assertThat(source).isInstanceOf(PaimonDataStreamSource.class);
assertThat(source.getBoundedness()).isEqualTo(Boundedness.BOUNDED);

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<RowData> dataStream =
new FlinkSourceBuilder(table).env(env).sourceBounded(true).build();
assertThat(dataStream.getTransformation()).isInstanceOf(SourceTransformation.class);
SourceTransformation<?, ?, ?> transformation =
(SourceTransformation<?, ?, ?>) dataStream.getTransformation();
assertThat(transformation.getSource()).isInstanceOf(PaimonDataStreamSource.class);
assertThat(transformation.getSource().getBoundedness()).isEqualTo(Boundedness.BOUNDED);
}

@Test
public void testBuildSourceRejectsDedicatedSplitGeneration() throws Exception {
Table table =
createTableWithOptions(
"build_source_dedicated_split",
false,
-1,
"scan.dedicated-split-generation",
"true");

assertThatThrownBy(() -> new FlinkSourceBuilder(table).sourceBounded(true).buildSource())
.isInstanceOf(UnsupportedOperationException.class)
.hasMessageContaining("build()");
}

@Test
public void testBuildSourceRejectsCheckpointAlign() throws Exception {
Table table =
createTableWithOptions(
"build_source_checkpoint_align",
false,
-1,
"source.checkpoint-align.enabled",
"true");

assertThatThrownBy(() -> new FlinkSourceBuilder(table).sourceBounded(false).buildSource())
.isInstanceOf(UnsupportedOperationException.class)
.hasMessageContaining("build()");
}

@Test
public void testBuildSourceRejectsExactlyOnceConsumer() throws Exception {
Table table =
createTableWithOptions(
"build_source_exactly_once_consumer",
false,
-1,
"consumer-id",
"my-consumer",
"consumer.expiration-time",
"1 h");

assertThatThrownBy(() -> new FlinkSourceBuilder(table).sourceBounded(false).buildSource())
.isInstanceOf(UnsupportedOperationException.class)
.hasMessageContaining("build()");
}

@Test
public void testPostponeMergeOnReadIgnoresContinuousSource() throws Exception {
Identifier identifier = Identifier.create("default", "postpone_merge_on_read");
Expand Down
Loading
Loading