diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java index 23d8f0b2e9c..ceffa72a333 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/FlinkConnectorOptions.java +++ b/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 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 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 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; + } + } } diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java index 1b082855e6e..43051038b97 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkTableFactory.java +++ b/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> 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); diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java index d442dee27ab..8c7c7141121 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/BinlogFlinkTableSource.java +++ b/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 tableOptions; @@ -93,12 +94,37 @@ public BinlogFlinkTableSource( long scanPartitionDiscoveryIntervalMs, int splitPerAssignmentBatchSize, Map 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 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(streaming, boundedOptions); FlinkSource source = new FlinkSource<>( flussConfig, @@ -152,6 +180,8 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { null, null, offsetsInitializer, + stoppingOffsetsInitializer, + FlinkConnectorOptionsUtils.toBoundedness(streaming, boundedOptions), scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, new BinlogDeserializationSchema(), @@ -160,6 +190,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 +206,7 @@ public DynamicTableSource copy() { isPartitioned, streaming, startupOptions, + boundedOptions, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, tableOptions); diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java index fef9f40ebd2..195dd73765b 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/ChangelogFlinkTableSource.java +++ b/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 tableOptions; @@ -129,6 +130,30 @@ public ChangelogFlinkTableSource( long scanPartitionDiscoveryIntervalMs, int splitPerAssignmentBatchSize, Map 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 tableOptions) { this.tablePath = tablePath; this.flussConfig = flussConfig; // The changelogOutputType already includes metadata columns from FlinkCatalog @@ -136,6 +161,7 @@ public ChangelogFlinkTableSource( 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(streaming, boundedOptions); FlinkSource source = new FlinkSource<>( flussConfig, @@ -212,12 +240,15 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { dataProjection, logRecordBatchFilter, offsetsInitializer, + stoppingOffsetsInitializer, + FlinkConnectorOptionsUtils.toBoundedness(streaming, boundedOptions), scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, new ChangelogDeserializationSchema(), FlinkConversions.toFlussRowType(producedDataType), streaming, partitionFilters, + null, LeaseContext.DEFAULT); // Lake source not supported return SourceProvider.of(source); @@ -233,6 +264,7 @@ public DynamicTableSource copy() { partitionKeyIndexes, streaming, startupOptions, + boundedOptions, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, tableOptions); diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java index 85ace290409..e2041814f59 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkSource.java @@ -17,6 +17,7 @@ package org.apache.fluss.flink.source; +import org.apache.fluss.client.initializer.NoStoppingOffsetsInitializer; import org.apache.fluss.client.initializer.OffsetsInitializer; import org.apache.fluss.config.Configuration; import org.apache.fluss.flink.FlinkConnectorOptions; @@ -75,9 +76,11 @@ public class FlinkSource @Nullable private final FlinkRecordEmitter.OutputProjection outputProjection; @Nullable private final int[] projectedFields; protected final OffsetsInitializer offsetsInitializer; + protected final OffsetsInitializer stoppingOffsetsInitializer; protected final long scanPartitionDiscoveryIntervalMs; protected final int splitPerAssignmentBatchSize; private final boolean streaming; + private final Boundedness boundedness; private final FlussDeserializationSchema deserializationSchema; @Nullable private final Predicate partitionFilters; @Nullable private final LakeSource lakeSource; @@ -208,6 +211,8 @@ public FlinkSource( projectedFields, logRecordBatchFilter, offsetsInitializer, + streaming ? new NoStoppingOffsetsInitializer() : OffsetsInitializer.latest(), + streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, deserializationSchema, @@ -243,6 +248,8 @@ public FlinkSource( projectedFields, logRecordBatchFilter, offsetsInitializer, + streaming ? new NoStoppingOffsetsInitializer() : OffsetsInitializer.latest(), + streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, deserializationSchema, @@ -262,6 +269,8 @@ public FlinkSource( @Nullable int[] projectedFields, @Nullable Predicate logRecordBatchFilter, OffsetsInitializer offsetsInitializer, + OffsetsInitializer stoppingOffsetsInitializer, + Boundedness boundedness, long scanPartitionDiscoveryIntervalMs, int splitPerAssignmentBatchSize, FlussDeserializationSchema deserializationSchema, @@ -278,6 +287,8 @@ public FlinkSource( this.projectedFields = projectedFields; this.logRecordBatchFilter = logRecordBatchFilter; this.offsetsInitializer = offsetsInitializer; + this.stoppingOffsetsInitializer = stoppingOffsetsInitializer; + this.boundedness = boundedness; this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs; this.splitPerAssignmentBatchSize = splitPerAssignmentBatchSize; this.deserializationSchema = deserializationSchema; @@ -291,7 +302,7 @@ public FlinkSource( @Override public Boundedness getBoundedness() { - return streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED; + return boundedness; } @Override @@ -304,6 +315,8 @@ public SplitEnumerator createEnumerator( isPartitioned, splitEnumeratorContext, offsetsInitializer, + stoppingOffsetsInitializer, + boundedness, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, streaming, @@ -335,6 +348,8 @@ public SplitEnumerator restoreEnumerator sourceEnumeratorState.getAssignedPartitions(), remainingHybridLakeFlussSplits, offsetsInitializer, + stoppingOffsetsInitializer, + boundedness, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, streaming, diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java index d2d3c7eaeec..714ef921b51 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlinkTableSource.java @@ -130,6 +130,7 @@ public class FlinkTableSource private final int[] partitionKeyIndexes; private final boolean streaming; private final FlinkConnectorOptionsUtils.StartupOptions startupOptions; + private final FlinkConnectorOptionsUtils.BoundedOptions boundedOptions; // options for lookup source private final boolean lookupAsync; @@ -221,6 +222,7 @@ public FlinkTableSource( int[] partitionKeyIndexes, boolean streaming, FlinkConnectorOptionsUtils.StartupOptions startupOptions, + FlinkConnectorOptionsUtils.BoundedOptions boundedOptions, boolean lookupAsync, boolean insertIfNotExists, @Nullable LookupCache cache, @@ -239,6 +241,7 @@ public FlinkTableSource( this.partitionKeyIndexes = partitionKeyIndexes; this.streaming = streaming; this.startupOptions = checkNotNull(startupOptions, "startupOptions must not be null"); + this.boundedOptions = checkNotNull(boundedOptions, "boundedOptions must not be null"); this.lookupAsync = lookupAsync; this.insertIfNotExists = insertIfNotExists; @@ -264,6 +267,47 @@ public FlinkTableSource( PushdownUtils.computeAvailableStatsColumns(flussRowType, tableConfig); } + public FlinkTableSource( + TablePath tablePath, + Configuration flussConfig, + TableConfig tableConfig, + org.apache.flink.table.types.logical.RowType tableOutputType, + int[] primaryKeyIndexes, + int[] bucketKeyIndexes, + int[] partitionKeyIndexes, + boolean streaming, + FlinkConnectorOptionsUtils.StartupOptions startupOptions, + boolean lookupAsync, + boolean insertIfNotExists, + @Nullable LookupCache cache, + long scanPartitionDiscoveryIntervalMs, + int splitPerAssignmentBatchSize, + boolean isDataLakeEnabled, + @Nullable MergeEngineType mergeEngineType, + Map tableOptions, + LeaseContext leaseContext) { + this( + tablePath, + flussConfig, + tableConfig, + tableOutputType, + primaryKeyIndexes, + bucketKeyIndexes, + partitionKeyIndexes, + streaming, + startupOptions, + FlinkConnectorOptionsUtils.BoundedOptions.unbounded(), + lookupAsync, + insertIfNotExists, + cache, + scanPartitionDiscoveryIntervalMs, + splitPerAssignmentBatchSize, + isDataLakeEnabled, + mergeEngineType, + tableOptions, + leaseContext); + } + @Override public ChangelogMode getChangelogMode() { if (!streaming) { @@ -397,6 +441,8 @@ public boolean isBounded() { "Unsupported startup mode: " + startupOptions.startupMode); } + OffsetsInitializer stoppingOffsetsInitializer = createStoppingOffsetsInitializer(); + FlinkSource source = new FlinkSource<>( flussConfig, @@ -407,6 +453,8 @@ public boolean isBounded() { projectedFields, logRecordBatchFilter, offsetsInitializer, + stoppingOffsetsInitializer, + FlinkConnectorOptionsUtils.toBoundedness(streaming, boundedOptions), scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, new RowDataDeserializationSchema(), @@ -458,6 +506,46 @@ public boolean isBounded() { } } + /** Creates the stopping offsets initializer from the configured bounded options. */ + private OffsetsInitializer createStoppingOffsetsInitializer() { + if (boundedOptions.boundedMode != FlinkConnectorOptions.ScanBoundedMode.UNBOUNDED) { + validateBoundedModeSupported(); + } + return FlinkConnectorOptionsUtils.toStoppingOffsetsInitializer(streaming, boundedOptions); + } + + private void validateBoundedModeSupported() { + if (hasPrimaryKey()) { + if (!streaming) { + throw new UnsupportedOperationException( + String.format( + "'%s' is not supported for primary key tables in batch execution mode.", + FlinkConnectorOptions.SCAN_BOUNDED_MODE.key())); + } + if (startupOptions.startupMode == FlinkConnectorOptions.ScanStartupMode.FULL) { + throw new UnsupportedOperationException( + String.format( + "'%s' is not supported for primary key tables in '%s' startup mode, " + + "because the snapshot reading phase has no bounded end. " + + "Use 'earliest', 'latest' or 'timestamp' startup mode to " + + "read the changelog of a primary key table with a bounded end.", + FlinkConnectorOptions.SCAN_BOUNDED_MODE.key(), + FlinkConnectorOptions.ScanStartupMode.FULL)); + } + } + if (isDataLakeEnabled + && startupOptions.startupMode == FlinkConnectorOptions.ScanStartupMode.FULL) { + throw new UnsupportedOperationException( + String.format( + "'%s' is not supported for the datalake union read, i.e. '%s' startup " + + "mode on a datalake-enabled table. Use 'earliest', 'latest' " + + "or 'timestamp' startup mode to read only the Fluss log with " + + "a bounded end.", + FlinkConnectorOptions.SCAN_BOUNDED_MODE.key(), + FlinkConnectorOptions.ScanStartupMode.FULL)); + } + } + @Override public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) { LookupNormalizer lookupNormalizer = @@ -512,6 +600,7 @@ public DynamicTableSource copy() { partitionKeyIndexes, streaming, startupOptions, + boundedOptions, lookupAsync, insertIfNotExists, cache, diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSource.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSource.java index 392d77708fb..17f6f16bb99 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSource.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSource.java @@ -18,6 +18,7 @@ package org.apache.fluss.flink.source; import org.apache.fluss.annotation.VisibleForTesting; +import org.apache.fluss.client.initializer.NoStoppingOffsetsInitializer; import org.apache.fluss.client.initializer.OffsetsInitializer; import org.apache.fluss.client.initializer.SnapshotOffsetsInitializer; import org.apache.fluss.config.Configuration; @@ -30,6 +31,8 @@ import org.apache.fluss.predicate.Predicate; import org.apache.fluss.types.RowType; +import org.apache.flink.api.connector.source.Boundedness; + import javax.annotation.Nullable; /** @@ -106,6 +109,40 @@ public class FlussSource extends FlinkSource { FlussDeserializationSchema deserializationSchema, boolean streaming, @Nullable LakeSource lakeSource) { + this( + flussConf, + tablePath, + hasPrimaryKey, + isPartitioned, + sourceOutputType, + projectedFields, + logRecordBatchFilter, + offsetsInitializer, + streaming ? new NoStoppingOffsetsInitializer() : OffsetsInitializer.latest(), + streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED, + scanPartitionDiscoveryIntervalMs, + splitPerAssignmentBatchSize, + deserializationSchema, + streaming, + lakeSource); + } + + FlussSource( + Configuration flussConf, + TablePath tablePath, + boolean hasPrimaryKey, + boolean isPartitioned, + RowType sourceOutputType, + @Nullable int[] projectedFields, + @Nullable Predicate logRecordBatchFilter, + OffsetsInitializer offsetsInitializer, + OffsetsInitializer stoppingOffsetsInitializer, + Boundedness boundedness, + long scanPartitionDiscoveryIntervalMs, + int splitPerAssignmentBatchSize, + FlussDeserializationSchema deserializationSchema, + boolean streaming, + @Nullable LakeSource lakeSource) { // TODO: Support partition pushDown in datastream super( flussConf, @@ -116,9 +153,12 @@ public class FlussSource extends FlinkSource { projectedFields, logRecordBatchFilter, validateBatchStartupMode(offsetsInitializer, hasPrimaryKey, streaming, tablePath), + stoppingOffsetsInitializer, + boundedness, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, deserializationSchema, + null, streaming, null, lakeSource, diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java index ca691453392..08945afeb7c 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/FlussSourceBuilder.java @@ -20,8 +20,11 @@ import org.apache.fluss.client.Connection; import org.apache.fluss.client.ConnectionFactory; import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.client.initializer.LatestOffsetsInitializer; +import org.apache.fluss.client.initializer.NoStoppingOffsetsInitializer; import org.apache.fluss.client.initializer.OffsetsInitializer; import org.apache.fluss.client.initializer.SnapshotOffsetsInitializer; +import org.apache.fluss.client.initializer.TimestampOffsetsInitializer; import org.apache.fluss.config.ConfigOptions; import org.apache.fluss.config.Configuration; import org.apache.fluss.flink.FlinkConnectorOptions; @@ -34,6 +37,7 @@ import org.apache.fluss.predicate.Predicate; import org.apache.fluss.types.RowType; +import org.apache.flink.api.connector.source.Boundedness; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -44,6 +48,7 @@ import java.util.concurrent.ExecutionException; import static org.apache.flink.util.Preconditions.checkNotNull; +import static org.apache.fluss.utils.Preconditions.checkArgument; /** * Builder class for creating {@link FlussSource} instances. @@ -83,7 +88,13 @@ public class FlussSourceBuilder { private Long scanPartitionDiscoveryIntervalMs; private Integer splitPerAssignmentBatchSize; private OffsetsInitializer offsetsInitializer; + private OffsetsInitializer stoppingOffsetsInitializer = new NoStoppingOffsetsInitializer(); + + // This flag preserves the execution semantics of the legacy no-argument setBounded(), which + // switches the source to batch mode. The parameterized setBounded(OffsetsInitializer) keeps + // streaming execution semantics and only makes the source bounded. private boolean bounded; + private Boundedness boundedness = Boundedness.CONTINUOUS_UNBOUNDED; private FlussDeserializationSchema deserializationSchema; private String bootstrapServers; @@ -176,12 +187,43 @@ public FlussSourceBuilder setStartingOffsets(OffsetsInitializer offsetsInit * Builds a bounded source for batch execution. The source reads up to the latest offsets at job * startup and then finishes; combined with the default {@link OffsetsInitializer#full()} on a * datalake-enabled table this performs a bounded union read of the lake snapshot and the Fluss - * log. If not called, the source is unbounded (streaming). + * log. + * + *

This overload is retained for compatibility and switches the source to batch execution + * semantics. Use {@link #setBounded(OffsetsInitializer)} for a bounded streaming read. * * @return this builder */ public FlussSourceBuilder setBounded() { this.bounded = true; + this.boundedness = Boundedness.BOUNDED; + this.stoppingOffsetsInitializer = OffsetsInitializer.latest(); + return this; + } + + /** + * Builds a bounded source that stops once it reaches the given stopping offsets and then + * finishes, even in streaming execution mode (a bounded streaming read). Typical usages are + * replaying a bounded time range of the log, backfilling and archiving. + * + *

Supported stopping offsets initializers are {@link OffsetsInitializer#latest()} and {@link + * OffsetsInitializer#timestamp(long)}. + * + * @param stoppingOffsetsInitializer the strategy for determining the stopping offsets + * @return this builder + */ + public FlussSourceBuilder setBounded(OffsetsInitializer stoppingOffsetsInitializer) { + OffsetsInitializer checkedStoppingOffsetsInitializer = + checkNotNull( + stoppingOffsetsInitializer, "stoppingOffsetsInitializer must not be null"); + checkArgument( + checkedStoppingOffsetsInitializer instanceof LatestOffsetsInitializer + || checkedStoppingOffsetsInitializer instanceof TimestampOffsetsInitializer, + "Only OffsetsInitializer.latest() and OffsetsInitializer.timestamp(...) are " + + "supported as stopping offsets, but was %s.", + checkedStoppingOffsetsInitializer.getClass().getName()); + this.stoppingOffsetsInitializer = checkedStoppingOffsetsInitializer; + this.boundedness = Boundedness.BOUNDED; return this; } @@ -362,6 +404,35 @@ public FlussSource build() { tablePath, fullStartup)); } + // Bounded streaming read support (user-supplied stopping offsets): + // - Log tables and the changelog of primary key tables (earliest/latest/timestamp + // startup mode) are supported. + // - The full startup mode of primary key tables is not supported, because the snapshot + // reading phase has no bounded end. + // - The datalake union read (full startup mode on a datalake-enabled table) is not + // supported, because lake splits have no bounded end. + if (!bounded && boundedness == Boundedness.BOUNDED) { + if (hasPrimaryKey && fullStartup) { + throw new IllegalArgumentException( + String.format( + "Bounded read with stopping offsets on primary key table '%s' is " + + "not supported in full startup mode, because the " + + "snapshot reading phase has no bounded end. Use " + + "earliest/latest/timestamp starting offsets to read the " + + "changelog with a bounded end.", + tablePath)); + } + if (lakeEnabled && fullStartup) { + throw new IllegalArgumentException( + String.format( + "Bounded read with stopping offsets on datalake-enabled table '%s' " + + "is not supported in full startup mode (datalake union " + + "read). Use earliest/latest/timestamp starting offsets " + + "to read only the Fluss log with a bounded end.", + tablePath)); + } + } + LakeSource lakeSource = null; if (lakeEnabled && fullStartup) { lakeSource = @@ -393,6 +464,8 @@ public FlussSource build() { projectedFields, logRecordBatchFilter, offsetsInitializer, + stoppingOffsetsInitializer, + boundedness, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, deserializationSchema, diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java index f774986ef22..c0552e2f2d8 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumerator.java @@ -56,6 +56,7 @@ import org.apache.flink.annotation.Internal; import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.api.connector.source.Boundedness; import org.apache.flink.api.connector.source.SourceEvent; import org.apache.flink.api.connector.source.SplitEnumerator; import org.apache.flink.api.connector.source.SplitEnumeratorContext; @@ -151,6 +152,7 @@ public class FlinkSourceEnumerator private final long scanPartitionDiscoveryIntervalMs; private final boolean streaming; + private final Boundedness boundedness; private final OffsetsInitializer startingOffsetsInitializer; private final OffsetsInitializer stoppingOffsetsInitializer; @@ -268,6 +270,8 @@ public FlinkSourceEnumerator( isPartitioned, context, startingOffsetsInitializer, + streaming ? new NoStoppingOffsetsInitializer() : OffsetsInitializer.latest(), + streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED, scanPartitionDiscoveryIntervalMs, FlinkConnectorOptions.SCAN_SPLIT_ASSIGNMENT_BATCH_SIZE.defaultValue(), streaming, @@ -291,6 +295,40 @@ public FlinkSourceEnumerator( @Nullable LakeSource lakeSource, LeaseContext leaseContext, boolean checkpointTriggeredBefore) { + this( + tablePath, + flussConf, + hasPrimaryKey, + isPartitioned, + context, + startingOffsetsInitializer, + streaming ? new NoStoppingOffsetsInitializer() : OffsetsInitializer.latest(), + streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED, + scanPartitionDiscoveryIntervalMs, + splitPerAssignmentBatchSize, + streaming, + partitionFilters, + lakeSource, + leaseContext, + checkpointTriggeredBefore); + } + + public FlinkSourceEnumerator( + TablePath tablePath, + Configuration flussConf, + boolean hasPrimaryKey, + boolean isPartitioned, + SplitEnumeratorContext context, + OffsetsInitializer startingOffsetsInitializer, + OffsetsInitializer stoppingOffsetsInitializer, + Boundedness boundedness, + long scanPartitionDiscoveryIntervalMs, + int splitPerAssignmentBatchSize, + boolean streaming, + @Nullable Predicate partitionFilters, + @Nullable LakeSource lakeSource, + LeaseContext leaseContext, + boolean checkpointTriggeredBefore) { this( tablePath, flussConf, @@ -301,11 +339,14 @@ public FlinkSourceEnumerator( Collections.emptyMap(), null, startingOffsetsInitializer, + stoppingOffsetsInitializer, + boundedness, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, streaming, partitionFilters, lakeSource, + new WorkerExecutor(context), leaseContext, checkpointTriggeredBefore, false, @@ -380,6 +421,53 @@ public FlinkSourceEnumerator( assignedPartitions, pendingHybridLakeFlussSplits, startingOffsetsInitializer, + streaming ? new NoStoppingOffsetsInitializer() : OffsetsInitializer.latest(), + streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED, + scanPartitionDiscoveryIntervalMs, + splitPerAssignmentBatchSize, + streaming, + partitionFilters, + lakeSource, + new WorkerExecutor(context), + leaseContext, + checkpointTriggeredBefore, + initialDiscoveryFinished, + unassignedSplits); + } + + public FlinkSourceEnumerator( + TablePath tablePath, + Configuration flussConf, + boolean hasPrimaryKey, + boolean isPartitioned, + SplitEnumeratorContext context, + Set assignedTableBuckets, + Map assignedPartitions, + List pendingHybridLakeFlussSplits, + OffsetsInitializer startingOffsetsInitializer, + OffsetsInitializer stoppingOffsetsInitializer, + Boundedness boundedness, + long scanPartitionDiscoveryIntervalMs, + int splitPerAssignmentBatchSize, + boolean streaming, + @Nullable Predicate partitionFilters, + @Nullable LakeSource lakeSource, + LeaseContext leaseContext, + boolean checkpointTriggeredBefore, + boolean initialDiscoveryFinished, + Collection unassignedSplits) { + this( + tablePath, + flussConf, + hasPrimaryKey, + isPartitioned, + context, + assignedTableBuckets, + assignedPartitions, + pendingHybridLakeFlussSplits, + startingOffsetsInitializer, + stoppingOffsetsInitializer, + boundedness, scanPartitionDiscoveryIntervalMs, splitPerAssignmentBatchSize, streaming, @@ -419,6 +507,8 @@ public FlinkSourceEnumerator( assignedPartitions, pendingHybridLakeFlussSplits, startingOffsetsInitializer, + streaming ? new NoStoppingOffsetsInitializer() : OffsetsInitializer.latest(), + streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED, scanPartitionDiscoveryIntervalMs, FlinkConnectorOptions.SCAN_SPLIT_ASSIGNMENT_BATCH_SIZE.defaultValue(), streaming, @@ -451,6 +541,52 @@ public FlinkSourceEnumerator( boolean checkpointTriggeredBefore, boolean initialDiscoveryFinished, Collection unassignedSplits) { + this( + tablePath, + flussConf, + hasPrimaryKey, + isPartitioned, + context, + assignedTableBuckets, + assignedPartitions, + pendingHybridLakeFlussSplits, + startingOffsetsInitializer, + streaming ? new NoStoppingOffsetsInitializer() : OffsetsInitializer.latest(), + streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED, + scanPartitionDiscoveryIntervalMs, + splitPerAssignmentBatchSize, + streaming, + partitionFilters, + lakeSource, + workerExecutor, + leaseContext, + checkpointTriggeredBefore, + initialDiscoveryFinished, + unassignedSplits); + } + + FlinkSourceEnumerator( + TablePath tablePath, + Configuration flussConf, + boolean hasPrimaryKey, + boolean isPartitioned, + SplitEnumeratorContext context, + Set assignedTableBuckets, + Map assignedPartitions, + List pendingHybridLakeFlussSplits, + OffsetsInitializer startingOffsetsInitializer, + OffsetsInitializer stoppingOffsetsInitializer, + Boundedness boundedness, + long scanPartitionDiscoveryIntervalMs, + int splitPerAssignmentBatchSize, + boolean streaming, + @Nullable Predicate partitionFilters, + @Nullable LakeSource lakeSource, + WorkerExecutor workerExecutor, + LeaseContext leaseContext, + boolean checkpointTriggeredBefore, + boolean initialDiscoveryFinished, + Collection unassignedSplits) { checkArgument( splitPerAssignmentBatchSize > 0, "Split assignment batch size must be positive, but was %s.", @@ -471,9 +607,9 @@ public FlinkSourceEnumerator( this.newDiscoveryOffsetsInitializer = OffsetsInitializer.earliest(); this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs; this.streaming = streaming; + this.boundedness = checkNotNull(boundedness); this.partitionFilters = partitionFilters; - this.stoppingOffsetsInitializer = - streaming ? new NoStoppingOffsetsInitializer() : OffsetsInitializer.latest(); + this.stoppingOffsetsInitializer = checkNotNull(stoppingOffsetsInitializer); this.lakeSource = lakeSource; this.workerExecutor = workerExecutor; this.leaseContext = leaseContext; @@ -541,7 +677,7 @@ public void start() { } } - if (scanPartitionDiscoveryIntervalMs > 0) { + if (isPeriodicPartitionDiscoveryEnabled()) { // should do partition discovery LOG.info( "Starting the FlussSourceEnumerator for table {} " @@ -555,7 +691,7 @@ public void start() { 0, scanPartitionDiscoveryIntervalMs); } else { - // just call once + // Call once for a bounded read or when partition discovery is disabled. LOG.info( "Starting the FlussSourceEnumerator for table {} without partition discovery.", tablePath); @@ -626,6 +762,9 @@ private void startInStreamModeForNonPartitionedTable() { tablePath, pendingSplitAssignment.values().stream().mapToInt(List::size).sum()); initialDiscoveryFinished = true; + if (!isPeriodicPartitionDiscoveryEnabled()) { + noMoreNewSplits = true; + } return; } @@ -733,6 +872,10 @@ private void checkPartitionChanges(Set partitionInfos, Throwable if (!initialDiscoveryFinished) { initialDiscoveryFinished = true; } + if (!isPeriodicPartitionDiscoveryEnabled()) { + noMoreNewSplits = true; + assignPendingSplits(context.registeredReaders().keySet()); + } LOG.debug("No partition changes detected for table {}", tablePath); return; } @@ -1041,15 +1184,28 @@ private List getLogSplit( Map stoppingOffsets = stoppingOffsetsInitializer.getBucketOffsets( partitionName, bucketsNeedInitOffset, bucketOffsetsRetriever); - startingOffsets.forEach( - (bucketId, startingOffset) -> - splits.add( - new LogSplit( - new TableBucket( - tableInfo.getTableId(), partitionId, bucketId), - partitionName, - startingOffset, - stoppingOffsets.get(bucketId)))); + for (Integer bucketId : bucketsNeedInitOffset) { + Long startingOffset = startingOffsets.get(bucketId); + Long stoppingOffset = stoppingOffsets.get(bucketId); + checkState( + startingOffset != null, + "Starting offset should be present for bucket %s.", + bucketId); + checkState( + stoppingOffset != null + && (stoppingOffset == LogSplit.NO_STOPPING_OFFSET + || stoppingOffset >= 0), + "Stopping offset for bucket %s must be non-negative or the no-stopping " + + "sentinel, but was %s.", + bucketId, + stoppingOffset); + splits.add( + new LogSplit( + new TableBucket(tableInfo.getTableId(), partitionId, bucketId), + partitionName, + startingOffset, + stoppingOffset)); + } } return splits; } @@ -1159,9 +1315,16 @@ private static boolean shouldRemoveForDroppedPartition( return removedPartitionsMap.containsKey(split.getTableBucket().getPartitionId()); } + private boolean isPeriodicPartitionDiscoveryEnabled() { + return isPartitioned + && streaming + && boundedness == Boundedness.CONTINUOUS_UNBOUNDED + && scanPartitionDiscoveryIntervalMs > 0; + } + private void handleSplitsAdd(List splits, Throwable t) { if (t != null) { - if (isPartitioned && streaming && scanPartitionDiscoveryIntervalMs > 0) { + if (isPeriodicPartitionDiscoveryEnabled()) { // it means continuously read new partition splits, not throw exception, temporally // warn it to avoid job fail. TODO: fix me in #288 LOG.warn("Failed to list splits for {}.", tablePath, t); @@ -1192,15 +1355,7 @@ private void handleSplitsAdd(List splits, Throwable t) { ? "null" : pendingHybridLakeFlussSplits.size()); - if (isPartitioned) { - if (!streaming || scanPartitionDiscoveryIntervalMs <= 0) { - // if not streaming or partition discovery is disabled - // should only add splits only once, no more new splits - noMoreNewSplits = true; - } - } else { - // if not partitioned, only will add splits only once, - // so, noMoreNewPartitionSplits should be set to true + if (!isPeriodicPartitionDiscoveryEnabled()) { noMoreNewSplits = true; } doHandleSplitsAdd(splits); diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java index f589203b7ad..a90f240ec8f 100644 --- a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java +++ b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtils.java @@ -17,12 +17,16 @@ package org.apache.fluss.flink.utils; +import org.apache.fluss.client.initializer.NoStoppingOffsetsInitializer; +import org.apache.fluss.client.initializer.OffsetsInitializer; import org.apache.fluss.config.Configuration; import org.apache.fluss.flink.FlinkConnectorOptions; +import org.apache.fluss.flink.FlinkConnectorOptions.ScanBoundedMode; import org.apache.fluss.flink.FlinkConnectorOptions.ScanStartupMode; import org.apache.fluss.flink.sink.shuffle.DistributionMode; import org.apache.fluss.metadata.MergeEngineType; +import org.apache.flink.api.connector.source.Boundedness; import org.apache.flink.configuration.ConfigurationUtils; import org.apache.flink.configuration.ReadableConfig; import org.apache.flink.table.api.ValidationException; @@ -43,6 +47,8 @@ import static org.apache.fluss.config.ConfigOptions.CLIENT_SCANNER_IO_TMP_DIR; import static org.apache.fluss.config.ConfigOptions.LAKE_TIERING_IO_TMP_DIRS; +import static org.apache.fluss.flink.FlinkConnectorOptions.SCAN_BOUNDED_MODE; +import static org.apache.fluss.flink.FlinkConnectorOptions.SCAN_BOUNDED_TIMESTAMP; import static org.apache.fluss.flink.FlinkConnectorOptions.SCAN_SPLIT_ASSIGNMENT_BATCH_SIZE; import static org.apache.fluss.flink.FlinkConnectorOptions.SCAN_STARTUP_MODE; import static org.apache.fluss.flink.FlinkConnectorOptions.SCAN_STARTUP_TIMESTAMP; @@ -62,6 +68,7 @@ public static ZoneId getLocalTimeZone(String timeZone) { public static void validateTableSourceOptions(ReadableConfig tableOptions) { validateScanStartupMode(tableOptions); + validateScanBoundedMode(tableOptions); validateScanSplitAssignmentBatchSize(tableOptions); } @@ -109,6 +116,56 @@ public static StartupOptions getStartupOptions(ReadableConfig tableOptions, Zone return options; } + public static BoundedOptions getBoundedOptions(ReadableConfig tableOptions, ZoneId timeZone) { + ScanBoundedMode scanBoundedMode = tableOptions.get(SCAN_BOUNDED_MODE); + final BoundedOptions options = new BoundedOptions(); + options.boundedMode = scanBoundedMode; + if (scanBoundedMode == ScanBoundedMode.TIMESTAMP) { + options.boundedTimestampMs = + parseTimestamp( + tableOptions.get(SCAN_BOUNDED_TIMESTAMP), + SCAN_BOUNDED_TIMESTAMP.key(), + timeZone); + } + return options; + } + + /** Creates the stopping offsets initializer from the given bounded options. */ + public static OffsetsInitializer toStoppingOffsetsInitializer(BoundedOptions boundedOptions) { + switch (boundedOptions.boundedMode) { + case UNBOUNDED: + return new NoStoppingOffsetsInitializer(); + case LATEST_OFFSET: + return OffsetsInitializer.latest(); + case TIMESTAMP: + return OffsetsInitializer.timestamp(boundedOptions.boundedTimestampMs); + default: + throw new IllegalArgumentException( + "Unsupported bounded mode: " + boundedOptions.boundedMode); + } + } + + /** + * Creates the stopping offsets initializer for the execution mode and bounded options. + * + *

Batch execution remains bounded by the latest offsets when no explicit bounded mode is + * configured. + */ + public static OffsetsInitializer toStoppingOffsetsInitializer( + boolean streaming, BoundedOptions boundedOptions) { + if (!streaming && boundedOptions.boundedMode == ScanBoundedMode.UNBOUNDED) { + return OffsetsInitializer.latest(); + } + return toStoppingOffsetsInitializer(boundedOptions); + } + + /** Returns the Flink source boundedness for the execution mode and bounded options. */ + public static Boundedness toBoundedness(boolean streaming, BoundedOptions boundedOptions) { + return streaming && boundedOptions.boundedMode == ScanBoundedMode.UNBOUNDED + ? Boundedness.CONTINUOUS_UNBOUNDED + : Boundedness.BOUNDED; + } + public static List getBucketKeys(ReadableConfig tableOptions) { Optional bucketKey = tableOptions.getOptional(FlinkConnectorOptions.BUCKET_KEY); if (!bucketKey.isPresent()) { @@ -156,6 +213,18 @@ private static void validateScanStartupMode(ReadableConfig tableOptions) { } } + private static void validateScanBoundedMode(ReadableConfig tableOptions) { + ScanBoundedMode scanBoundedMode = tableOptions.get(SCAN_BOUNDED_MODE); + if (scanBoundedMode == ScanBoundedMode.TIMESTAMP) { + if (!tableOptions.getOptional(SCAN_BOUNDED_TIMESTAMP).isPresent()) { + throw new ValidationException( + String.format( + "'%s' is required in '%s' bounded mode but missing.", + SCAN_BOUNDED_TIMESTAMP.key(), ScanBoundedMode.TIMESTAMP)); + } + } + } + private static void validateScanSplitAssignmentBatchSize(ReadableConfig tableOptions) { int batchSize = tableOptions.get(SCAN_SPLIT_ASSIGNMENT_BATCH_SIZE); if (batchSize <= 0) { @@ -238,4 +307,17 @@ public static class StartupOptions { public ScanStartupMode startupMode; public long startupTimestampMs; } + + /** Fluss bounded options. */ + public static class BoundedOptions { + public ScanBoundedMode boundedMode; + public long boundedTimestampMs; + + /** Returns the default bounded options, i.e. no user-specified stopping offsets. */ + public static BoundedOptions unbounded() { + BoundedOptions options = new BoundedOptions(); + options.boundedMode = ScanBoundedMode.UNBOUNDED; + return options; + } + } } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java index bbf5c29aa8e..532333fca76 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkTableFactoryTest.java @@ -105,6 +105,22 @@ void testTableSourceOptions() { FlinkConnectorOptions.SCAN_STARTUP_TIMESTAMP.key(), "2023-12-09 23:09:12"); createTableSource(schema, scanModeProperties); + // test scan bounded mode options + Map boundedModeProperties = getBasicOptions(); + boundedModeProperties.put(FlinkConnectorOptions.SCAN_BOUNDED_MODE.key(), "timestamp"); + assertThatThrownBy(() -> createTableSource(schema, boundedModeProperties)) + .isInstanceOf(ValidationException.class) + .hasMessageContaining( + "'scan.bounded.timestamp' is required in 'timestamp' bounded mode but missing."); + boundedModeProperties.put( + FlinkConnectorOptions.SCAN_BOUNDED_TIMESTAMP.key(), "1678883047356"); + createTableSource(schema, boundedModeProperties); + boundedModeProperties.put( + FlinkConnectorOptions.SCAN_BOUNDED_TIMESTAMP.key(), "2023-12-09 23:09:12"); + createTableSource(schema, boundedModeProperties); + boundedModeProperties.put(FlinkConnectorOptions.SCAN_BOUNDED_MODE.key(), "latest-offset"); + createTableSource(schema, boundedModeProperties); + // test split assignment batch size Map splitAssignmentBatchProperties = getBasicOptions(); splitAssignmentBatchProperties.put( diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java index bb4c21330e8..75627809206 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/BinlogVirtualTableITCase.java @@ -712,4 +712,43 @@ protected static void waitForCheckpoint(JobID jobId) { Duration.ofSeconds(60), "Timeout waiting for checkpoint for job " + jobIdStr); } + + @Test + public void testBinlogBoundedRead() throws Exception { + tEnv.executeSql( + "CREATE TABLE bounded_binlog_test (" + + " id INT NOT NULL," + + " name STRING," + + " PRIMARY KEY (id) NOT ENFORCED" + + ") WITH ('bucket.num' = '1')"); + TablePath tablePath = TablePath.of(DEFAULT_DB, "bounded_binlog_test"); + + CLOCK.advanceTime(Duration.ofMillis(1000)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Item-1"), row(2, "Item-2")), false); + // the update produces an update_before/update_after pair in the log + writeRows(conn, tablePath, Arrays.asList(row(1, "Item-1-Updated")), false); + CLOCK.advanceTime(Duration.ofMillis(1000)); + long boundedTimestamp = CLOCK.milliseconds(); + // records written at or after the bounded timestamp are not read + writeRows(conn, tablePath, Arrays.asList(row(2, "Item-2-Updated")), false); + + // The stopping offsets are aligned to record batch boundaries, and the update_before/ + // update_after pair of a single update is always written in one record batch, so the + // pair is never split apart by the stopping offset: the last update is either fully + // included (merged into one binlog row) or fully excluded. + String query = + "SELECT _change_type, before.id, before.name, after.id, after.name " + + "FROM bounded_binlog_test$binlog " + + String.format( + "/*+ OPTIONS('scan.bounded.mode' = 'timestamp', " + + "'scan.bounded.timestamp' = '%d') */", + boundedTimestamp); + try (CloseableIterator rowIter = tEnv.executeSql(query).collect()) { + assertThat(collectBatchRows(rowIter)) + .containsExactly( + "+I[insert, null, null, 1, Item-1]", + "+I[insert, null, null, 2, Item-2]", + "+I[update, 1, Item-1, 1, Item-1-Updated]"); + } + } } diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java index 71974955ece..a3fe6b27b9f 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/ChangelogVirtualTableITCase.java @@ -443,6 +443,35 @@ public void testChangelogVirtualTableWithLogTable() throws Exception { .isEqualTo("+I[insert, 2, 1970-01-01T00:00:02Z, 3, purchase]"); } + @Test + public void testChangelogBoundedRead() throws Exception { + tEnv.executeSql( + "CREATE TABLE bounded_changelog_test (" + + " id INT NOT NULL," + + " name STRING," + + " PRIMARY KEY (id) NOT ENFORCED" + + ") WITH ('bucket.num' = '1')"); + TablePath tablePath = TablePath.of(DEFAULT_DB, "bounded_changelog_test"); + + CLOCK.advanceTime(Duration.ofMillis(1000)); + writeRows(conn, tablePath, Arrays.asList(row(1, "Alice"), row(2, "Bob")), false); + writeRows(conn, tablePath, Arrays.asList(row(1, "Alice-2")), false); + + // the bounded changelog read stops at the latest offsets captured at startup and then + // the job finishes + String query = + "SELECT _change_type, id, name FROM bounded_changelog_test$changelog " + + "/*+ OPTIONS('scan.bounded.mode' = 'latest-offset') */"; + try (CloseableIterator rowIter = tEnv.executeSql(query).collect()) { + assertThat(collectBatchRows(rowIter)) + .containsExactly( + "+I[insert, 1, Alice]", + "+I[insert, 2, Bob]", + "+I[update_before, 1, Alice]", + "+I[update_after, 1, Alice-2]"); + } + } + @Test public void testProjectionOnChangelogTable() throws Exception { // Create a primary key table with 1 bucket and extra columns to test projection diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceITCase.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceITCase.java index f19f20163e8..25c052c7e16 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceITCase.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlinkTableSourceITCase.java @@ -75,6 +75,7 @@ import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertQueryResultExactOrder; import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertResultsIgnoreOrder; +import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.collectBatchRows; import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.collectRowsWithTimeout; import static org.apache.fluss.flink.utils.FlinkTestBase.waitUntilPartitions; import static org.apache.fluss.flink.utils.FlinkTestBase.writeRows; @@ -695,6 +696,88 @@ void testReadPrimaryKeyPartitionedTable(boolean isAutoPartition) throws Exceptio assertResultsIgnoreOrder(rowIter, expectedRowValues, true); } + @Test + void testLogTableBoundedReadLatestOffset() throws Exception { + tEnv.executeSql("create table bounded_latest_log_table (a int, b varchar)"); + TablePath tablePath = TablePath.of(DEFAULT_DB, "bounded_latest_log_table"); + writeRows(conn, tablePath, Arrays.asList(row(1, "v1"), row(2, "v2"), row(3, "v3")), true); + + // in streaming execution mode, the bounded source stops at the latest offsets captured + // at startup and then the job finishes + try (CloseableIterator rowIter = + tEnv.executeSql( + "select * from bounded_latest_log_table " + + "/*+ OPTIONS('scan.bounded.mode' = 'latest-offset') */") + .collect()) { + assertThat(collectBatchRows(rowIter)) + .containsExactlyInAnyOrder("+I[1, v1]", "+I[2, v2]", "+I[3, v3]"); + } + } + + @Test + void testLogTableBoundedReadTimestampRange() throws Exception { + tEnv.executeSql("create table bounded_ts_range_log_table (a int, b varchar)"); + TablePath tablePath = TablePath.of(DEFAULT_DB, "bounded_ts_range_log_table"); + + writeRows(conn, tablePath, Arrays.asList(row(1, "v1"), row(2, "v2")), true); + CLOCK.advanceTime(Duration.ofMillis(100)); + long startTimestamp = CLOCK.milliseconds(); + writeRows(conn, tablePath, Arrays.asList(row(3, "v3"), row(4, "v4")), true); + CLOCK.advanceTime(Duration.ofMillis(100)); + long stopTimestamp = CLOCK.milliseconds(); + writeRows(conn, tablePath, Arrays.asList(row(5, "v5"), row(6, "v6")), true); + + // replay only the records committed in the time range [startTimestamp, stopTimestamp) + String options = + String.format( + " /*+ OPTIONS('scan.startup.mode' = 'timestamp', " + + "'scan.startup.timestamp' = '%d', " + + "'scan.bounded.mode' = 'timestamp', " + + "'scan.bounded.timestamp' = '%d') */", + startTimestamp, stopTimestamp); + try (CloseableIterator rowIter = + tEnv.executeSql("select * from bounded_ts_range_log_table" + options).collect()) { + assertThat(collectBatchRows(rowIter)) + .containsExactlyInAnyOrder("+I[3, v3]", "+I[4, v4]"); + } + } + + @Test + void testPrimaryKeyTableChangelogBoundedRead() throws Exception { + tEnv.executeSql( + "create table bounded_pk_changelog_table (a int not null primary key not enforced," + + " b varchar) with ('bucket.num' = '1')"); + TablePath tablePath = TablePath.of(DEFAULT_DB, "bounded_pk_changelog_table"); + writeRows(conn, tablePath, Arrays.asList(row(1, "v1"), row(2, "v2")), false); + writeRows(conn, tablePath, Arrays.asList(row(1, "v11")), false); + + // read the changelog of the primary key table from the earliest offset up to the latest + // offsets captured at startup, then the job finishes + try (CloseableIterator rowIter = + tEnv.executeSql( + "select * from bounded_pk_changelog_table " + + "/*+ OPTIONS('scan.startup.mode' = 'earliest', " + + "'scan.bounded.mode' = 'latest-offset') */") + .collect()) { + assertThat(collectBatchRows(rowIter)) + .containsExactly("+I[1, v1]", "+I[2, v2]", "-U[1, v1]", "+U[1, v11]"); + } + } + + @Test + void testPrimaryKeyTableFullStartupBoundedReadThrows() { + tEnv.executeSql( + "create table bounded_pk_full_table " + + "(a int not null primary key not enforced, b varchar)"); + assertThatThrownBy( + () -> + tEnv.executeSql( + "select * from bounded_pk_full_table " + + "/*+ OPTIONS('scan.bounded.mode' = 'latest-offset') */")) + .hasStackTraceContaining( + "'scan.bounded.mode' is not supported for primary key tables in 'full' startup mode"); + } + @Test void testReadTimestampGreaterThanMaxTimestamp() throws Exception { tEnv.executeSql("create table timestamp_table (a int, b varchar) "); diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlussSourceBuilderTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlussSourceBuilderTest.java index 0f54fc04b4d..550ce59052c 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlussSourceBuilderTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/FlussSourceBuilderTest.java @@ -26,6 +26,7 @@ import org.apache.fluss.types.RowType; import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.api.connector.source.Boundedness; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.function.Executable; @@ -60,6 +61,54 @@ public void testBuildWithValidConfiguration() { // Then assertThat(source).isNotNull(); + assertThat(source.getBoundedness()).isEqualTo(Boundedness.CONTINUOUS_UNBOUNDED); + } + + @Test + public void testRejectUnsupportedStoppingOffsetsInitializer() { + FlussSourceBuilder builder = FlussSource.builder(); + + assertThatThrownBy(() -> builder.setBounded(OffsetsInitializer.earliest())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "Only OffsetsInitializer.latest() and " + + "OffsetsInitializer.timestamp(...) are supported"); + assertThatThrownBy(() -> builder.setBounded(OffsetsInitializer.full())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining( + "Only OffsetsInitializer.latest() and " + + "OffsetsInitializer.timestamp(...) are supported"); + + assertThat(builder.setBounded(OffsetsInitializer.timestamp(1L))).isSameAs(builder); + } + + @Test + public void testBuildBoundedStreamingSource() { + FlussSource source = + FlussSource.builder() + .setBootstrapServers(bootstrapServers) + .setDatabase(DEFAULT_DB) + .setTable(DEFAULT_TABLE_PATH.getTableName()) + .setStartingOffsets(OffsetsInitializer.earliest()) + .setBounded(OffsetsInitializer.latest()) + .setDeserializationSchema(new TestDeserializationSchema()) + .build(); + + assertThat(source.getBoundedness()).isEqualTo(Boundedness.BOUNDED); + } + + @Test + public void testBuildLegacyBoundedSource() { + FlussSource source = + FlussSource.builder() + .setBootstrapServers(bootstrapServers) + .setDatabase(DEFAULT_DB) + .setTable(DEFAULT_TABLE_PATH.getTableName()) + .setBounded() + .setDeserializationSchema(new TestDeserializationSchema()) + .build(); + + assertThat(source.getBoundedness()).isEqualTo(Boundedness.BOUNDED); } @Test diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java index a49dbebc7d8..97a11ae08b5 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/source/enumerator/FlinkSourceEnumeratorTest.java @@ -17,6 +17,7 @@ package org.apache.fluss.flink.source.enumerator; +import org.apache.fluss.client.admin.OffsetSpec; import org.apache.fluss.client.initializer.OffsetsInitializer; import org.apache.fluss.client.table.Table; import org.apache.fluss.client.table.writer.UpsertWriter; @@ -57,6 +58,7 @@ import org.apache.fluss.types.DataTypes; import org.apache.fluss.types.RowType; +import org.apache.flink.api.connector.source.Boundedness; import org.apache.flink.api.connector.source.ReaderInfo; import org.apache.flink.api.connector.source.SourceEvent; import org.apache.flink.api.connector.source.SplitEnumerator; @@ -148,6 +150,9 @@ void testPkTableNoSnapshotSplits() throws Throwable { Map> actualAssignment = getReadersAssignments(context); assertThat(actualAssignment).isEqualTo(expectedAssignment); + for (int i = 0; i < numSubtasks; i++) { + assertThat(context.hasNoMoreSplits(i)).isTrue(); + } } } @@ -1063,6 +1068,203 @@ void testDiscoverPartitionsPeriodically(boolean isPrimaryKeyTable) throws Throwa } } + @Test + void testBatchModeWithTimestampStoppingOffsets() throws Throwable { + int numSubtasks = DEFAULT_BUCKET_NUM; + createTable(DEFAULT_TABLE_PATH, DEFAULT_LOG_TABLE_DESCRIPTOR); + List rows = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + rows.add(row(i, "v" + i)); + } + writeRows(conn, DEFAULT_TABLE_PATH, rows, true); + + // A stopping timestamp not earlier than all commit timestamps resolves to the latest + // offsets, so the generated splits cover all written records. + List bucketIds = new ArrayList<>(); + for (int bucket = 0; bucket < DEFAULT_BUCKET_NUM; bucket++) { + bucketIds.add(bucket); + } + Map latestOffsets = + admin.listOffsets(DEFAULT_TABLE_PATH, bucketIds, new OffsetSpec.LatestSpec()) + .all() + .get(); + + // wait until the clock strictly advances past the write acknowledgement, so that the + // stopping timestamp is strictly greater than all commit timestamps + long writeAckTime = System.currentTimeMillis(); + long stoppingTimestamp; + do { + stoppingTimestamp = System.currentTimeMillis(); + } while (stoppingTimestamp <= writeAckTime); + + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(numSubtasks)) { + FlinkSourceEnumerator enumerator = + new FlinkSourceEnumerator( + DEFAULT_TABLE_PATH, + flussConf, + false, + false, + context, + OffsetsInitializer.earliest(), + OffsetsInitializer.timestamp(stoppingTimestamp), + Boundedness.BOUNDED, + DEFAULT_SCAN_PARTITION_DISCOVERY_INTERVAL_MS, + FlinkConnectorOptions.SCAN_SPLIT_ASSIGNMENT_BATCH_SIZE.defaultValue(), + false, + null, + null, + LeaseContext.DEFAULT, + false); + + enumerator.start(); + for (int i = 0; i < numSubtasks; i++) { + registerReader(context, enumerator, i); + } + context.runNextOneTimeCallable(); + + List assignedSplits = + getReadersAssignments(context).values().stream() + .flatMap(List::stream) + .collect(Collectors.toList()); + assertThat(assignedSplits).hasSize(DEFAULT_BUCKET_NUM); + assertThat(assignedSplits) + .allSatisfy( + split -> { + LogSplit logSplit = split.asLogSplit(); + assertThat(logSplit.getStartingOffset()).isEqualTo(EARLIEST_OFFSET); + assertThat(logSplit.getStoppingOffset()) + .contains( + latestOffsets.get( + logSplit.getTableBucket().getBucket())); + }); + } + } + + @Test + void testBoundedStreamingReadSignalsNoMoreSplits() throws Throwable { + int numSubtasks = 3; + createTable(DEFAULT_TABLE_PATH, DEFAULT_AUTO_PARTITIONED_LOG_TABLE_DESCRIPTOR); + ZooKeeperClient zooKeeperClient = FLUSS_CLUSTER_EXTENSION.getZooKeeperClient(); + Map partitionNameByIds = + waitUntilPartitions(zooKeeperClient, DEFAULT_TABLE_PATH); + List rows = new ArrayList<>(); + for (String partitionName : partitionNameByIds.values()) { + for (int i = 0; i < 5; i++) { + rows.add(row(i, partitionName)); + } + } + writeRows(conn, DEFAULT_TABLE_PATH, rows, true); + + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(numSubtasks)) { + // a streaming read with user-supplied stopping offsets is a bounded read + FlinkSourceEnumerator enumerator = + new FlinkSourceEnumerator( + DEFAULT_TABLE_PATH, + flussConf, + false, + true, + context, + OffsetsInitializer.earliest(), + OffsetsInitializer.latest(), + Boundedness.BOUNDED, + DEFAULT_SCAN_PARTITION_DISCOVERY_INTERVAL_MS, + FlinkConnectorOptions.SCAN_SPLIT_ASSIGNMENT_BATCH_SIZE.defaultValue(), + true, + null, + null, + LeaseContext.DEFAULT, + false); + + enumerator.start(); + for (int i = 0; i < numSubtasks; i++) { + registerReader(context, enumerator, i); + } + + // a bounded streaming read only performs a one-time partition discovery, even though + // the partition discovery interval is positive + assertThat(context.getPeriodicCallables()).isEmpty(); + // discover the partitions and then initialize the splits + context.runNextOneTimeCallable(); + context.runNextOneTimeCallable(); + + List assignedSplits = + getReadersAssignments(context).values().stream() + .flatMap(List::stream) + .collect(Collectors.toList()); + assertThat(assignedSplits) + .hasSize(partitionNameByIds.size() * DEFAULT_BUCKET_NUM) + .allSatisfy( + split -> { + LogSplit logSplit = split.asLogSplit(); + assertThat(logSplit.getStartingOffset()).isEqualTo(EARLIEST_OFFSET); + assertThat(logSplit.getStoppingOffset()).isPresent(); + }); + // the stopping offsets are the latest offsets captured at startup, which sum up to + // the total number of written records + long totalStoppingOffset = + assignedSplits.stream() + .mapToLong(split -> split.asLogSplit().getStoppingOffset().get()) + .sum(); + assertThat(totalStoppingOffset).isEqualTo(rows.size()); + + // all splits are added at once for a bounded read, so all readers have been signaled + // that no more splits will come, which lets the job finish eventually + for (int i = 0; i < numSubtasks; i++) { + assertThat(context.hasNoMoreSplits(i)).isTrue(); + } + } + } + + @Test + void testBoundedStreamingReadWithNoPartitionsSignalsNoMoreSplits() throws Throwable { + TablePath tablePath = TablePath.of(DEFAULT_DB, "bounded-empty-partition-table"); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("partition_col", DataTypes.STRING()) + .build(); + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema(schema) + .partitionedBy("partition_col") + .distributedBy(DEFAULT_BUCKET_NUM, "id") + .build(); + createTable(tablePath, tableDescriptor); + + try (MockSplitEnumeratorContext context = + new MockSplitEnumeratorContext<>(2); + FlinkSourceEnumerator enumerator = + new FlinkSourceEnumerator( + tablePath, + flussConf, + false, + true, + context, + OffsetsInitializer.earliest(), + OffsetsInitializer.latest(), + Boundedness.BOUNDED, + DEFAULT_SCAN_PARTITION_DISCOVERY_INTERVAL_MS, + FlinkConnectorOptions.SCAN_SPLIT_ASSIGNMENT_BATCH_SIZE + .defaultValue(), + true, + null, + null, + LeaseContext.DEFAULT, + false)) { + enumerator.start(); + registerReader(context, enumerator, 0); + registerReader(context, enumerator, 1); + + context.runNextOneTimeCallable(); + + assertThat(context.getSplitsAssignmentSequence()).isEmpty(); + assertThat(context.hasNoMoreSplits(0)).isTrue(); + assertThat(context.hasNoMoreSplits(1)).isTrue(); + } + } + @Test void testGetSplitOwner() throws Exception { int numSubtasks = 3; diff --git a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtilTest.java b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtilTest.java index 2fe355c25a6..cee97c59abd 100644 --- a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtilTest.java +++ b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/utils/FlinkConnectorOptionsUtilTest.java @@ -17,8 +17,12 @@ package org.apache.fluss.flink.utils; +import org.apache.fluss.client.initializer.LatestOffsetsInitializer; +import org.apache.fluss.client.initializer.NoStoppingOffsetsInitializer; import org.apache.fluss.config.Configuration; +import org.apache.fluss.flink.FlinkConnectorOptions.ScanBoundedMode; +import org.apache.flink.api.connector.source.Boundedness; import org.apache.flink.table.api.ValidationException; import org.junit.jupiter.api.Test; @@ -83,6 +87,27 @@ void testValidateSplitAssignmentBatchSize() { .hasMessage("'scan.split.assignment.batch-size' must be positive, but was 0."); } + @Test + void testStoppingOffsetsAndBoundedness() { + FlinkConnectorOptionsUtils.BoundedOptions boundedOptions = + FlinkConnectorOptionsUtils.BoundedOptions.unbounded(); + + assertThat(FlinkConnectorOptionsUtils.toStoppingOffsetsInitializer(boundedOptions)) + .isInstanceOf(NoStoppingOffsetsInitializer.class); + assertThat(FlinkConnectorOptionsUtils.toStoppingOffsetsInitializer(false, boundedOptions)) + .isInstanceOf(LatestOffsetsInitializer.class); + assertThat(FlinkConnectorOptionsUtils.toBoundedness(true, boundedOptions)) + .isEqualTo(Boundedness.CONTINUOUS_UNBOUNDED); + assertThat(FlinkConnectorOptionsUtils.toBoundedness(false, boundedOptions)) + .isEqualTo(Boundedness.BOUNDED); + + boundedOptions.boundedMode = ScanBoundedMode.LATEST_OFFSET; + assertThat(FlinkConnectorOptionsUtils.toStoppingOffsetsInitializer(true, boundedOptions)) + .isInstanceOf(LatestOffsetsInitializer.class); + assertThat(FlinkConnectorOptionsUtils.toBoundedness(true, boundedOptions)) + .isEqualTo(Boundedness.BOUNDED); + } + @Test void testGetClientScannerIoTmpDir() { Configuration flussConfig = diff --git a/website/docs/engine-flink/datastream.mdx b/website/docs/engine-flink/datastream.mdx index 483f4b73037..48db2e1af54 100644 --- a/website/docs/engine-flink/datastream.mdx +++ b/website/docs/engine-flink/datastream.mdx @@ -95,6 +95,7 @@ The `FlussSourceBuilder` provides several methods for configuring the source con * **setProjectedFields(String... projectedFieldNames):** Sets the fields to project from the table (if not specified, all fields are included) * **setScanPartitionDiscoveryIntervalMs(long intervalMs):** Sets the interval for discovering new partitions (default: from configuration) * **setStartingOffsets(OffsetsInitializer initializer):** Sets the strategy for determining starting offsets (default: `OffsetsInitializer.full()`) +* **setBounded(OffsetsInitializer stoppingOffsetsInitializer):** Makes the source bounded: it stops once it reaches the given stopping offsets and then finishes, even in streaming execution mode (see [Bounded Reads](#bounded-reads)) * **setFlussConfig(Configuration flussConf):** Sets custom Fluss configuration properties ### Offset Initializers @@ -128,6 +129,28 @@ FlussSource source = FlussSource.builder() .build(); ``` +### Bounded Reads +By default, the source is unbounded and keeps reading new records. Calling `setBounded(OffsetsInitializer)` makes the source stop once it reaches the given stopping offsets and then finish, even in streaming execution mode (a bounded streaming read). Typical usages are replaying a bounded time range of the log, backfilling and archiving. Supported stopping offsets initializers are `OffsetsInitializer.latest()` and `OffsetsInitializer.timestamp(long)`. + +Bounded reads are supported for log tables and the changelog of primary key tables (earliest/latest/timestamp starting offsets), but not for the full startup mode of primary key tables or the datalake union read. + +**Example:** +```java +// Replay a bounded time range of the log: from one hour ago up to now +FlussSource timeRangeSource = FlussSource.builder() + .setStartingOffsets(OffsetsInitializer.timestamp(System.currentTimeMillis() - 3600 * 1000)) + .setBounded(OffsetsInitializer.timestamp(System.currentTimeMillis())) + // other configuration... + .build(); + +// Read up to the latest offsets captured at startup and then finish +FlussSource latestSource = FlussSource.builder() + .setStartingOffsets(OffsetsInitializer.earliest()) + .setBounded(OffsetsInitializer.latest()) + // other configuration... + .build(); +``` + ### Deserialization Schemas The `FlussDeserializationSchema` interface is used to convert Fluss records to your desired output type. Fluss provides some built-in implementations: diff --git a/website/docs/engine-flink/options.md b/website/docs/engine-flink/options.md index d04285ee942..7a0c5a5dd6c 100644 --- a/website/docs/engine-flink/options.md +++ b/website/docs/engine-flink/options.md @@ -102,6 +102,8 @@ See more details about [ALTER TABLE ... SET](engine-flink/ddl.md#set-properties) |-----------------------------------------------|------------|-------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | scan.startup.mode | Enum | full | The scan startup mode enables you to specify the starting point for data consumption. Fluss currently supports the following `scan.startup.mode` options: `full` (default), earliest, latest, timestamp. See the [Start Reading Position](engine-flink/reads.md#start-reading-position) for more details. | | scan.startup.timestamp | Long | (None) | The timestamp to start reading the data from. This option is only valid when `scan.startup.mode` is set to `timestamp`. The format is 'milli-second-since-epoch' or `yyyy-MM-dd HH:mm:ss`, like `1678883047356` or `2023-12-09 23:09:12`. | +| scan.bounded.mode | Enum | unbounded | The scan bounded mode enables you to specify where the source stops reading. Fluss currently supports the following `scan.bounded.mode` options: `unbounded` (default), `latest-offset`, `timestamp`. Bounded modes other than `unbounded` make the source bounded even in streaming execution mode (a bounded streaming read), and are supported for Log Tables and the changelog of Primary Key Tables (`earliest`/`latest`/`timestamp` startup mode). See the [Stop Reading Position](engine-flink/reads.md#stop-reading-position) for more details. | +| scan.bounded.timestamp | Long | (None) | The timestamp to stop reading the data at. This option is only valid when `scan.bounded.mode` is set to `timestamp`. Only records with a commit timestamp smaller than the given timestamp are read. The timestamp must not be in the future. The format is 'milli-second-since-epoch' or `yyyy-MM-dd HH:mm:ss`, like `1678883047356` or `2023-12-09 23:09:12`. | | scan.partition.discovery.interval | Duration | 1min | The time interval for the Fluss source to discover the new partitions for partitioned table while scanning. A non-positive value disables the partition discovery. The default value is 1 minute. Currently, since Fluss Admin#listPartitions(TablePath tablePath) requires a large number of requests to ZooKeeper in server, this option cannot be set too small, as a small value would cause frequent requests and increase server load. In the future, once list partitions is optimized, the default value of this parameter can be reduced. | | scan.kv.snapshot.lease.id | String | UUID | The lease ID used to protect acquired KV snapshots from deletion. If specified, the snapshots will be retained until either the consumer finishes processing all of them or the lease duration expires. By default, this value is set to a randomly generated UUID string if not explicitly provided. | | scan.kv.snapshot.lease.duration | Duration | 1day | The time period how long to wait before expiring the kv snapshot lease to avoid kv snapshot blocking to delete. | diff --git a/website/docs/engine-flink/reads.md b/website/docs/engine-flink/reads.md index 4e4d82e5bc8..cee06ee237a 100644 --- a/website/docs/engine-flink/reads.md +++ b/website/docs/engine-flink/reads.md @@ -396,6 +396,40 @@ SELECT * FROM log_table 'scan.startup.timestamp' = '2023-12-09 23:09:12') */; ``` +### Stop Reading Position + +The config option `scan.bounded.mode` enables you to specify where the source stops reading. A bounded mode other than `unbounded` makes the source bounded even in streaming execution mode: the job finishes once the source reaches the stopping position (a bounded streaming read). Typical usages are replaying a bounded time range of the log, backfilling and archiving. + +It is supported for Log Tables, the changelog of Primary Key Tables (`earliest`, `latest` or `timestamp` startup mode) and the `$changelog`/`$binlog` virtual tables, but not for the `full` startup mode of Primary Key Tables (the snapshot reading phase has no bounded end) or the datalake union read. + +Fluss supports the following `scan.bounded.mode` options: +- `unbounded` (default): In streaming execution mode, the source never stops. In batch execution mode, the source reads up to the latest log offsets captured when the source starts. +- `latest-offset`: The source stops at the latest log offsets captured when the source starts. In batch execution mode, this behaves the same as `unbounded`. +- `timestamp`: The source stops before the first record batch whose commit timestamp is greater than or equal to the specified time (defined by the configuration item `scan.bounded.timestamp`), i.e. only records with a commit timestamp smaller than the specified time are read. The specified time must not be in the future. + +The following SQL statement reads the `log_table` table up to a specified time. +```sql title="Flink SQL" +SELECT * FROM log_table +/*+ OPTIONS('scan.bounded.mode' = 'timestamp', +'scan.bounded.timestamp' = '2023-12-09 23:09:12') */; +``` + +The start and stop reading positions can be combined to replay a time range of the log, even in streaming execution mode. +```sql title="Flink SQL" +SELECT * FROM log_table +/*+ OPTIONS('scan.startup.mode' = 'timestamp', +'scan.startup.timestamp' = '2023-12-09 00:00:00', +'scan.bounded.mode' = 'timestamp', +'scan.bounded.timestamp' = '2023-12-10 00:00:00') */; +``` + +The following SQL statement reads the changelog of a primary key table up to the latest log offsets captured when the source starts, and then finishes. +```sql title="Flink SQL" +SELECT * FROM pk_table +/*+ OPTIONS('scan.startup.mode' = 'earliest', +'scan.bounded.mode' = 'latest-offset') */; +``` +