[flink] Support bounded streaming read for Fluss source

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java

@@ -137,6 +137,42 @@ public class FlinkConnectorOptions {
                                     + "The format is 'timestamp' or 'yyyy-MM-dd HH:mm:ss'. "
                                     + "Like '1678883047356' or '2023-12-09 23:09:12'.");
 
+    public static final ConfigOption<ScanBoundedMode> SCAN_BOUNDED_MODE =
+            ConfigOptions.key("scan.bounded.mode")
+                    .enumType(ScanBoundedMode.class)
+                    .defaultValue(ScanBoundedMode.UNBOUNDED)
+                    .withDescription(
+                            String.format(
+                                    "Bounded mode for the Fluss source. Default is '%s'. In batch "
+                                            + "execution mode, '%s' behaves the same as '%s': the "
+                                            + "source reads up to the latest log offsets captured "
+                                            + "at startup. In streaming execution mode, a bounded "
+                                            + "mode other than '%s' makes the source stop at the "
+                                            + "given stopping offsets and then the job finishes "
+                                            + "(a bounded streaming read). Bounded modes other "
+                                            + "than '%s' are supported for log tables and the "
+                                            + "changelog of primary key tables (earliest/latest/"
+                                            + "timestamp startup mode), but not for the full "
+                                            + "startup mode of primary key tables or the datalake "
+                                            + "union read.",
+                                    ScanBoundedMode.UNBOUNDED.value,
+                                    ScanBoundedMode.UNBOUNDED.value,
+                                    ScanBoundedMode.LATEST_OFFSET.value,
+                                    ScanBoundedMode.UNBOUNDED.value,
+                                    ScanBoundedMode.UNBOUNDED.value));
+
+    public static final ConfigOption<String> SCAN_BOUNDED_TIMESTAMP =
+            ConfigOptions.key("scan.bounded.timestamp")
+                    .stringType()
+                    .noDefaultValue()
+                    .withDescription(
+                            "Optional timestamp for Fluss source in case of bounded mode is timestamp. "
+                                    + "The source stops before the first record batch whose commit "
+                                    + "timestamp is greater than or equal to the given timestamp, i.e. "
+                                    + "only records with a commit timestamp smaller than the given "
+                                    + "timestamp are read. The format is 'timestamp' or "
+                                    + "'yyyy-MM-dd HH:mm:ss'. Like '1678883047356' or '2023-12-09 23:09:12'.");
+
     public static final ConfigOption<Duration> SCAN_PARTITION_DISCOVERY_INTERVAL =
             ConfigOptions.key("scan.partition.discovery.interval")
                     .durationType()
@@ -333,4 +369,41 @@ public InlineElement getDescription() {
             return description;
         }
     }
+
+    /** Bounded mode for the fluss scanner, see {@link #SCAN_BOUNDED_MODE}. */
+    public enum ScanBoundedMode implements DescribedEnum {
+        UNBOUNDED(
+                "unbounded",
+                text(
+                        "In streaming execution mode, the source never stops. In batch execution "
+                                + "mode, the source reads up to the latest log offsets captured "
+                                + "at startup.")),
+        LATEST_OFFSET(
+                "latest-offset",
+                text("Bounded by the latest log offsets captured when the source starts.")),
+        TIMESTAMP(
+                "timestamp",
+                text(
+                        "Bounded by a user-supplied timestamp. The source stops before the first "
+                                + "record batch whose commit timestamp is greater than or equal "
+                                + "to the given timestamp."));
+
+        private final String value;
+        private final InlineElement description;
+
+        ScanBoundedMode(String value, InlineElement description) {
+            this.value = value;
+            this.description = description;
+        }
+
+        @Override
+        public String toString() {
+            return value;
+        }
+
+        @Override
+        public InlineElement getDescription() {
+            return description;
+        }
+    }
 }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java

@@ -119,6 +119,8 @@ public DynamicTableSource createDynamicTableSource(Context context) {
                         context.getConfiguration().get(TableConfigOptions.LOCAL_TIME_ZONE));
         final FlinkConnectorOptionsUtils.StartupOptions startupOptions =
                 FlinkConnectorOptionsUtils.getStartupOptions(tableOptions, timeZone);
+        final FlinkConnectorOptionsUtils.BoundedOptions boundedOptions =
+                FlinkConnectorOptionsUtils.getBoundedOptions(tableOptions, timeZone);
 
         ResolvedSchema resolvedSchema = context.getCatalogTable().getResolvedSchema();
         ResolvedCatalogTable resolvedCatalogTable = context.getCatalogTable();
@@ -161,6 +163,7 @@ public DynamicTableSource createDynamicTableSource(Context context) {
                 partitionKeyIndexes,
                 isStreamingMode,
                 startupOptions,
+                boundedOptions,
                 tableOptions.get(FlinkConnectorOptions.LOOKUP_ASYNC),
                 tableOptions.get(FlinkConnectorOptions.LOOKUP_INSERT_IF_NOT_EXISTS),
                 cache,
@@ -236,6 +239,8 @@ public Set<ConfigOption<?>> optionalOptions() {
                                 FlinkConnectorOptions.BUCKET_NUMBER,
                                 FlinkConnectorOptions.SCAN_STARTUP_MODE,
                                 FlinkConnectorOptions.SCAN_STARTUP_TIMESTAMP,
+                                FlinkConnectorOptions.SCAN_BOUNDED_MODE,
+                                FlinkConnectorOptions.SCAN_BOUNDED_TIMESTAMP,
                                 FlinkConnectorOptions.SCAN_PARTITION_DISCOVERY_INTERVAL,
                                 FlinkConnectorOptions.SCAN_SPLIT_ASSIGNMENT_BATCH_SIZE,
                                 FlinkConnectorOptions.SCAN_KV_SNAPSHOT_LEASE_ID,
@@ -356,6 +361,8 @@ private DynamicTableSource createChangelogTableSource(
                         context.getConfiguration().get(TableConfigOptions.LOCAL_TIME_ZONE));
         final FlinkConnectorOptionsUtils.StartupOptions startupOptions =
                 FlinkConnectorOptionsUtils.getStartupOptions(tableOptions, timeZone);
+        final FlinkConnectorOptionsUtils.BoundedOptions boundedOptions =
+                FlinkConnectorOptionsUtils.getBoundedOptions(tableOptions, timeZone);
 
         ResolvedCatalogTable resolvedCatalogTable = context.getCatalogTable();
 
@@ -379,6 +386,7 @@ private DynamicTableSource createChangelogTableSource(
                 partitionKeyIndexes,
                 isStreamingMode,
                 startupOptions,
+                boundedOptions,
                 partitionDiscoveryIntervalMs,
                 splitAssignmentBatchSize,
                 catalogTableOptions);
@@ -410,6 +418,8 @@ private DynamicTableSource createBinlogTableSource(
                         context.getConfiguration().get(TableConfigOptions.LOCAL_TIME_ZONE));
         final FlinkConnectorOptionsUtils.StartupOptions startupOptions =
                 FlinkConnectorOptionsUtils.getStartupOptions(tableOptions, timeZone);
+        final FlinkConnectorOptionsUtils.BoundedOptions boundedOptions =
+                FlinkConnectorOptionsUtils.getBoundedOptions(tableOptions, timeZone);
 
         // Check if the table is partitioned from the internal option
         boolean isPartitioned =
@@ -429,6 +439,7 @@ private DynamicTableSource createBinlogTableSource(
                 isPartitioned,
                 isStreamingMode,
                 startupOptions,
+                boundedOptions,
                 partitionDiscoveryIntervalMs,
                 splitAssignmentBatchSize,
                 catalogTableOptions);

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java

@@ -55,6 +55,7 @@ public class BinlogFlinkTableSource
     private final boolean isPartitioned;
     private final boolean streaming;
     private final FlinkConnectorOptionsUtils.StartupOptions startupOptions;
+    private final FlinkConnectorOptionsUtils.BoundedOptions boundedOptions;
     private final long scanPartitionDiscoveryIntervalMs;
     private final int splitPerAssignmentBatchSize;
     private final Map<String, String> tableOptions;
@@ -93,12 +94,37 @@ public BinlogFlinkTableSource(
             long scanPartitionDiscoveryIntervalMs,
             int splitPerAssignmentBatchSize,
             Map<String, String> tableOptions) {
+        this(
+                tablePath,
+                flussConfig,
+                binlogOutputType,
+                isPartitioned,
+                streaming,
+                startupOptions,
+                FlinkConnectorOptionsUtils.BoundedOptions.unbounded(),
+                scanPartitionDiscoveryIntervalMs,
+                splitPerAssignmentBatchSize,
+                tableOptions);
+    }
+
+    public BinlogFlinkTableSource(
+            TablePath tablePath,
+            Configuration flussConfig,
+            org.apache.flink.table.types.logical.RowType binlogOutputType,
+            boolean isPartitioned,
+            boolean streaming,
+            FlinkConnectorOptionsUtils.StartupOptions startupOptions,
+            FlinkConnectorOptionsUtils.BoundedOptions boundedOptions,
+            long scanPartitionDiscoveryIntervalMs,
+            int splitPerAssignmentBatchSize,
+            Map<String, String> tableOptions) {
         this.tablePath = tablePath;
         this.flussConfig = flussConfig;
         this.binlogOutputType = binlogOutputType;
         this.isPartitioned = isPartitioned;
         this.streaming = streaming;
         this.startupOptions = startupOptions;
+        this.boundedOptions = boundedOptions;
         this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs;
         this.splitPerAssignmentBatchSize = splitPerAssignmentBatchSize;
         this.tableOptions = tableOptions;
@@ -142,6 +168,8 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
         }
 
         // Create the source with the binlog deserialization schema
+        OffsetsInitializer stoppingOffsetsInitializer =
+                FlinkConnectorOptionsUtils.toStoppingOffsetsInitializer(boundedOptions);
         FlinkSource<RowData> source =
                 new FlinkSource<>(
                         flussConfig,
@@ -152,6 +180,7 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
                         null,
                         null,
                         offsetsInitializer,
+                        stoppingOffsetsInitializer,
                         scanPartitionDiscoveryIntervalMs,
                         splitPerAssignmentBatchSize,
                         new BinlogDeserializationSchema(),
@@ -160,6 +189,7 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
                         // $binlog data/partition columns are nested inside before/after ROWs, so no
                         // top-level partition filter is pushable; always scan without one.
                         null,
+                        null,
                         LeaseContext.DEFAULT);
 
         return SourceProvider.of(source);
@@ -175,6 +205,7 @@ public DynamicTableSource copy() {
                         isPartitioned,
                         streaming,
                         startupOptions,
+                        boundedOptions,
                         scanPartitionDiscoveryIntervalMs,
                         splitPerAssignmentBatchSize,
                         tableOptions);

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java

@@ -74,6 +74,7 @@ public class ChangelogFlinkTableSource
     private final int[] partitionKeyIndexes;
     private final boolean streaming;
     private final FlinkConnectorOptionsUtils.StartupOptions startupOptions;
+    private final FlinkConnectorOptionsUtils.BoundedOptions boundedOptions;
     private final long scanPartitionDiscoveryIntervalMs;
     private final int splitPerAssignmentBatchSize;
     private final Map<String, String> tableOptions;
@@ -129,13 +130,38 @@ public ChangelogFlinkTableSource(
             long scanPartitionDiscoveryIntervalMs,
             int splitPerAssignmentBatchSize,
             Map<String, String> tableOptions) {
+        this(
+                tablePath,
+                flussConfig,
+                changelogOutputType,
+                partitionKeyIndexes,
+                streaming,
+                startupOptions,
+                FlinkConnectorOptionsUtils.BoundedOptions.unbounded(),
+                scanPartitionDiscoveryIntervalMs,
+                splitPerAssignmentBatchSize,
+                tableOptions);
+    }
+
+    public ChangelogFlinkTableSource(
+            TablePath tablePath,
+            Configuration flussConfig,
+            org.apache.flink.table.types.logical.RowType changelogOutputType,
+            int[] partitionKeyIndexes,
+            boolean streaming,
+            FlinkConnectorOptionsUtils.StartupOptions startupOptions,
+            FlinkConnectorOptionsUtils.BoundedOptions boundedOptions,
+            long scanPartitionDiscoveryIntervalMs,
+            int splitPerAssignmentBatchSize,
+            Map<String, String> tableOptions) {
         this.tablePath = tablePath;
         this.flussConfig = flussConfig;
         // The changelogOutputType already includes metadata columns from FlinkCatalog
         this.changelogOutputType = changelogOutputType;
         this.partitionKeyIndexes = partitionKeyIndexes;
         this.streaming = streaming;
         this.startupOptions = startupOptions;
+        this.boundedOptions = boundedOptions;
         this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs;
         this.splitPerAssignmentBatchSize = splitPerAssignmentBatchSize;
         this.tableOptions = tableOptions;
@@ -198,6 +224,8 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
         }
 
         // Create the source with the changelog deserialization schema
+        OffsetsInitializer stoppingOffsetsInitializer =
+                FlinkConnectorOptionsUtils.toStoppingOffsetsInitializer(boundedOptions);
         FlinkSource<RowData> source =
                 new FlinkSource<>(
                         flussConfig,
@@ -212,12 +240,14 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
                         dataProjection,
                         logRecordBatchFilter,
                         offsetsInitializer,
+                        stoppingOffsetsInitializer,
                         scanPartitionDiscoveryIntervalMs,
                         splitPerAssignmentBatchSize,
                         new ChangelogDeserializationSchema(),
                         FlinkConversions.toFlussRowType(producedDataType),
                         streaming,
                         partitionFilters,
+                        null,
                         LeaseContext.DEFAULT); // Lake source not supported
 
         return SourceProvider.of(source);
@@ -233,6 +263,7 @@ public DynamicTableSource copy() {
                         partitionKeyIndexes,
                         streaming,
                         startupOptions,
+                        boundedOptions,
                         scanPartitionDiscoveryIntervalMs,
                         splitPerAssignmentBatchSize,
                         tableOptions);

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java

@@ -75,6 +75,7 @@ public class FlinkSource<OUT>
     @Nullable private final FlinkRecordEmitter.OutputProjection<OUT> outputProjection;
     @Nullable private final int[] projectedFields;
     protected final OffsetsInitializer offsetsInitializer;
+    @Nullable protected final OffsetsInitializer stoppingOffsetsInitializer;
     protected final long scanPartitionDiscoveryIntervalMs;
     protected final int splitPerAssignmentBatchSize;
     private final boolean streaming;
@@ -208,6 +209,7 @@ public FlinkSource(
                 projectedFields,
                 logRecordBatchFilter,
                 offsetsInitializer,
+                null,
                 scanPartitionDiscoveryIntervalMs,
                 splitPerAssignmentBatchSize,
                 deserializationSchema,
@@ -243,6 +245,7 @@ public FlinkSource(
                 projectedFields,
                 logRecordBatchFilter,
                 offsetsInitializer,
+                null,
                 scanPartitionDiscoveryIntervalMs,
                 splitPerAssignmentBatchSize,
                 deserializationSchema,
@@ -262,6 +265,7 @@ public FlinkSource(
             @Nullable int[] projectedFields,
             @Nullable Predicate logRecordBatchFilter,
             OffsetsInitializer offsetsInitializer,
+            @Nullable OffsetsInitializer stoppingOffsetsInitializer,
             long scanPartitionDiscoveryIntervalMs,
             int splitPerAssignmentBatchSize,
             FlussDeserializationSchema<OUT> deserializationSchema,
@@ -278,6 +282,7 @@ public FlinkSource(
         this.projectedFields = projectedFields;
         this.logRecordBatchFilter = logRecordBatchFilter;
         this.offsetsInitializer = offsetsInitializer;
+        this.stoppingOffsetsInitializer = stoppingOffsetsInitializer;
         this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs;
         this.splitPerAssignmentBatchSize = splitPerAssignmentBatchSize;
         this.deserializationSchema = deserializationSchema;
@@ -291,7 +296,12 @@ public FlinkSource(
 
     @Override
     public Boundedness getBoundedness() {
-        return streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED;
+        // User-supplied stopping offsets make the source bounded even in streaming execution
+        // mode (bounded streaming read), so that the job finishes once all splits reach their
+        // stopping offsets.
+        return (streaming && stoppingOffsetsInitializer == null)
+                ? Boundedness.CONTINUOUS_UNBOUNDED
+                : Boundedness.BOUNDED;
     }
 
     @Override
@@ -304,6 +314,7 @@ public SplitEnumerator<SourceSplitBase, SourceEnumeratorState> createEnumerator(
                 isPartitioned,
                 splitEnumeratorContext,
                 offsetsInitializer,
+                stoppingOffsetsInitializer,
                 scanPartitionDiscoveryIntervalMs,
                 splitPerAssignmentBatchSize,
                 streaming,
@@ -335,6 +346,7 @@ public SplitEnumerator<SourceSplitBase, SourceEnumeratorState> restoreEnumerator
                 sourceEnumeratorState.getAssignedPartitions(),
                 remainingHybridLakeFlussSplits,
                 offsetsInitializer,
+                stoppingOffsetsInitializer,
                 scanPartitionDiscoveryIntervalMs,
                 splitPerAssignmentBatchSize,
                 streaming,