This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1747cc2347 [flink] Expose Flink Source in FlinkSourceBuilder API
(#10713)
1747cc2347 is described below
commit 1747cc2347dfa84984af4f77bc19c59a624ca3df
Author: eye-gu <[email protected]>
AuthorDate: Mon Oct 26 12:18:15 2026 +0800
[flink] Expose Flink Source in FlinkSourceBuilder API (#10713)
---
.../paimon/flink/source/FlinkSourceBuilder.java | 161 ++++++++++++++++-----
.../flink/source/FlinkSourceBuilderTest.java | 150 ++++++++++++++++++-
.../paimon/flink/source/HybridSourceITCase.java | 135 +++++++++++++++++
3 files changed, 409 insertions(+), 37 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
index 0b73413b87..4857d40c13 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
@@ -217,20 +217,68 @@ public class FlinkSourceBuilder {
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() {
@@ -330,16 +378,6 @@ public class FlinkSourceBuilder {
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(
@@ -418,18 +456,75 @@ public class FlinkSourceBuilder {
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();
@@ -442,7 +537,7 @@ public class FlinkSourceBuilder {
if
(conf.get(FlinkConnectorOptions.SCAN_DEDICATED_SPLIT_GENERATION)) {
return buildDedicatedSplitGenSource(true);
}
- return buildStaticFileSource();
+ return toDataStream(createSource());
}
TableScanUtils.streamingReadingValidate(table);
@@ -459,7 +554,7 @@ public class FlinkSourceBuilder {
== CoreOptions.ConsumerMode.EXACTLY_ONCE) {
return buildDedicatedSplitGenSource(false);
} else {
- return buildContinuousFileSource();
+ return toDataStream(createSource());
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkSourceBuilderTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkSourceBuilderTest.java
index 9e891b12df..5ade069c36 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkSourceBuilderTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkSourceBuilderTest.java
@@ -38,7 +38,9 @@ import org.apache.paimon.table.source.Split;
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;
@@ -99,11 +101,23 @@ public class FlinkSourceBuilderTest {
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");
@@ -304,6 +318,134 @@ public class FlinkSourceBuilderTest {
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");
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/HybridSourceITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/HybridSourceITCase.java
new file mode 100644
index 0000000000..0598e64a0f
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/HybridSourceITCase.java
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.source;
+
+import org.apache.paimon.flink.CatalogITCaseBase;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.table.FileStoreTable;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.connector.base.source.hybrid.HybridSource;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.util.DataFormatConverters;
+import org.apache.flink.table.runtime.typeutils.ExternalTypeInfo;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.types.Row;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import static
org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType;
+import static org.apache.paimon.flink.LogicalTypeConversion.toLogicalType;
+
+/** End-to-end test of {@link FlinkSourceBuilder#buildSource()} embedded in a
HybridSource. */
+public class HybridSourceITCase extends CatalogITCaseBase {
+
+ @Override
+ protected List<String> ddl() {
+ return Arrays.asList(
+ "CREATE TABLE hybrid_source (a INT, b STRING)",
+ "CREATE TABLE hybrid_sink (a INT, b STRING)");
+ }
+
+ @Timeout(300)
+ @Test
+ public void testBoundedThenContinuousRead() throws Exception {
+ batchSql("INSERT INTO hybrid_source VALUES (1, 'first'), (2,
'second')");
+
+ FileStoreTable table = paimonTable("hybrid_source");
+
+ StreamExecutionEnvironment env =
+ streamExecutionEnvironmentBuilder()
+ .streamingMode()
+ .parallelism(2)
+ .checkpointIntervalMs(500)
+ .build();
+ StreamTableEnvironment streamTableEnv =
StreamTableEnvironment.create(env);
+ streamTableEnv.registerCatalog("PAIMON_STREAM",
sEnv.getCatalog("PAIMON").get());
+ streamTableEnv.executeSql("USE CATALOG PAIMON_STREAM");
+
+ Source<RowData, ?, ?> bounded =
+ new
FlinkSourceBuilder(table).sourceBounded(true).buildSource();
+ // The continuous segment (latest-full) would re-read the bounded
segment's data, so it
+ // filters a <= 2: rows 1-2 can then only be delivered by the bounded
segment.
+ Predicate batch2Only = new
PredicateBuilder(table.rowType()).greaterThan(0, 2);
+ HybridSource<RowData> hybridSource =
+ HybridSource.builder(bounded)
+ .addSource(
+ context ->
+ new FlinkSourceBuilder(table)
+ .predicate(batch2Only)
+ .sourceBounded(false)
+ .buildSource(),
+ Boundedness.CONTINUOUS_UNBOUNDED)
+ .build();
+
+ DataType dataType =
fromLogicalToDataType(toLogicalType(table.rowType()));
+ DataFormatConverters.RowConverter converter =
+ new DataFormatConverters.RowConverter(
+ dataType.getChildren().toArray(new DataType[0]));
+ DataStream<Row> rows =
+ env.fromSource(
+ hybridSource,
+ WatermarkStrategy.noWatermarks(),
+ "HybridSource",
+
InternalTypeInfo.of(toLogicalType(table.rowType())))
+ .map((MapFunction<RowData, Row>) converter::toExternal)
+ .returns(ExternalTypeInfo.of(dataType));
+
+ streamTableEnv.createTemporaryView("S",
streamTableEnv.fromDataStream(rows).as("a", "b"));
+ TableResult result = streamTableEnv.executeSql("INSERT INTO
hybrid_sink SELECT * FROM S");
+ JobClient client = result.getJobClient().get();
+
+ try {
+ // written after the job is submitted; the continuous segment sees
it either in its
+ // starting snapshot or incrementally, so both segments are still
verified
+ batchSql("INSERT INTO hybrid_source VALUES (3, 'third'), (4,
'fourth')");
+
+ // DISTINCT tolerates batch2 being read by both segments, but rows
1-2 are filtered
+ // out of the continuous segment and thus prove the bounded
segment ran
+ sqlAssertWithRetry(
+ "SELECT DISTINCT * FROM hybrid_sink",
+ assertion ->
+ assertion.containsExactlyInAnyOrder(
+ Row.of(1, "first"),
+ Row.of(2, "second"),
+ Row.of(3, "third"),
+ Row.of(4, "fourth")));
+ } finally {
+ try {
+ client.cancel().get(30, TimeUnit.SECONDS);
+ } catch (Exception ignored) {
+ // best-effort cleanup: the job may already be terminated
+ }
+ }
+ }
+}
