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 0b73413b87a9..4857d40c13a0 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 @@ private ReadBuilder createReadBuilder(@Nullable org.apache.paimon.types.RowType return readBuilder.dropStats(); } - private DataStream 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 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 buildPostponeMergeSource() { @@ -330,16 +378,6 @@ private boolean inferScanParallelism() { return conf.get(FlinkConnectorOptions.INFER_SCAN_PARALLELISM); } - private DataStream buildContinuousFileSource() { - return toDataStream( - new ContinuousFileStoreSource( - createReadBuilder(projectedRowType()), - table.options(), - limit, - unordered, - outerProject())); - } - private DataStream buildAlignedContinuousFileSource() { assertStreamingConfigurationForAlignMode(env); return toDataStream( @@ -418,18 +456,75 @@ public DataStream 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)}. + * + *

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. + * + *

Example: + * + * 

{@code
+     * Source bounded =
+     *         new FlinkSourceBuilder(table).sourceBounded(true).buildSource();
+     * HybridSource 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())));
+     * }
+ * + *

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. + * + *

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. + * + *

{@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. + * + *

{@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 buildSource() { + return new PaimonDataStreamSource<>(createSource(), table); + } + /** Build source {@link DataStream} with {@link RowData}. */ public DataStream 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 mergeSource = buildPostponeMergeSource(); @@ -442,7 +537,7 @@ public DataStream build() { if (conf.get(FlinkConnectorOptions.SCAN_DEDICATED_SPLIT_GENERATION)) { return buildDedicatedSplitGenSource(true); } - return buildStaticFileSource(); + return toDataStream(createSource()); } TableScanUtils.streamingReadingValidate(table); @@ -459,7 +554,7 @@ public DataStream build() { == 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 9e891b12df41..5ade069c3616 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.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 @@ 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"); @@ -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 bounded = + new FlinkSourceBuilder(table).sourceBounded(true).buildSource(); + assertThat(bounded).isInstanceOf(PaimonDataStreamSource.class); + assertThat(bounded.getBoundedness()).isEqualTo(Boundedness.BOUNDED); + + Source 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 bounded = new FlinkSourceBuilder(table).env(batchEnv).buildSource(); + assertThat(bounded.getBoundedness()).isEqualTo(Boundedness.BOUNDED); + + StreamExecutionEnvironment streamingEnv = + StreamExecutionEnvironment.getExecutionEnvironment(); + streamingEnv.setRuntimeMode(RuntimeExecutionMode.STREAMING); + Source 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 source = + new FlinkSourceBuilder(table).sourceBounded(true).buildSource(); + assertThat(source).isInstanceOf(PaimonDataStreamSource.class); + assertThat(source.getBoundedness()).isEqualTo(Boundedness.BOUNDED); + + StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); + DataStream 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 000000000000..0598e64a0f67 --- /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 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 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 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 rows = + env.fromSource( + hybridSource, + WatermarkStrategy.noWatermarks(), + "HybridSource", + InternalTypeInfo.of(toLogicalType(table.rowType()))) + .map((MapFunction) 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 + } + } + } +}