[flink] Flink source support detect new buckets for existing tables or partitions

Author: LiebingYu

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

@@ -107,6 +107,21 @@ public class FlinkConnectorOptions {
                                     + "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.");
 
+    public static final ConfigOption<Duration> SCAN_BUCKET_DISCOVERY_INTERVAL =
+            ConfigOptions.key("scan.bucket.discovery.interval")
+                    .durationType()
+                    .defaultValue(Duration.ofMinutes(1))
+                    .withDescription(
+                            "The time interval for the Fluss source to discover "
+                                    + "the new buckets for table while scanning."
+                                    + " A non-positive value disables the bucket discovery. "
+                                    + "For partitioned tables, if "
+                                    + SCAN_PARTITION_DISCOVERY_INTERVAL.key()
+                                    + " is also set, then the minimum value of "
+                                    + SCAN_PARTITION_DISCOVERY_INTERVAL.key()
+                                    + " and the current value will be used as the time interval for "
+                                    + "discovering partitions and buckets.");
+
     public static final ConfigOption<Boolean> SINK_IGNORE_DELETE =
             ConfigOptions.key("sink.ignore-delete")
                     .booleanType()

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

@@ -137,6 +137,8 @@ public DynamicTableSource createDynamicTableSource(Context context) {
                 tableOptions
                         .get(FlinkConnectorOptions.SCAN_PARTITION_DISCOVERY_INTERVAL)
                         .toMillis();
+        long bucketDiscoveryIntervalMs =
+                tableOptions.get(FlinkConnectorOptions.SCAN_BUCKET_DISCOVERY_INTERVAL).toMillis();
 
         return new FlinkTableSource(
                 toFlussTablePath(context.getObjectIdentifier()),
@@ -151,6 +153,7 @@ public DynamicTableSource createDynamicTableSource(Context context) {
                 tableOptions.get(FlinkConnectorOptions.LOOKUP_ASYNC),
                 cache,
                 partitionDiscoveryIntervalMs,
+                bucketDiscoveryIntervalMs,
                 tableOptions.get(toFlinkOption(ConfigOptions.TABLE_DATALAKE_ENABLED)),
                 tableOptions.get(toFlinkOption(ConfigOptions.TABLE_MERGE_ENGINE)),
                 context.getCatalogTable().getOptions());

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

@@ -63,6 +63,7 @@ public class FlinkSource<OUT>
     @Nullable private final int[] projectedFields;
     protected final OffsetsInitializer offsetsInitializer;
     protected final long scanPartitionDiscoveryIntervalMs;
+    protected final long scanBucketDiscoveryIntervalMs;
     private final boolean streaming;
     private final FlussDeserializationSchema<OUT> deserializationSchema;
     @Nullable private final Predicate partitionFilters;
@@ -77,6 +78,7 @@ public FlinkSource(
             @Nullable int[] projectedFields,
             OffsetsInitializer offsetsInitializer,
             long scanPartitionDiscoveryIntervalMs,
+            long scanBucketDiscoveryIntervalMs,
             FlussDeserializationSchema<OUT> deserializationSchema,
             boolean streaming,
             @Nullable Predicate partitionFilters) {
@@ -89,6 +91,7 @@ public FlinkSource(
                 projectedFields,
                 offsetsInitializer,
                 scanPartitionDiscoveryIntervalMs,
+                scanBucketDiscoveryIntervalMs,
                 deserializationSchema,
                 streaming,
                 partitionFilters,
@@ -104,6 +107,7 @@ public FlinkSource(
             @Nullable int[] projectedFields,
             OffsetsInitializer offsetsInitializer,
             long scanPartitionDiscoveryIntervalMs,
+            long scanBucketDiscoveryIntervalMs,
             FlussDeserializationSchema<OUT> deserializationSchema,
             boolean streaming,
             @Nullable Predicate partitionFilters,
@@ -116,6 +120,7 @@ public FlinkSource(
         this.projectedFields = projectedFields;
         this.offsetsInitializer = offsetsInitializer;
         this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs;
+        this.scanBucketDiscoveryIntervalMs = scanBucketDiscoveryIntervalMs;
         this.deserializationSchema = deserializationSchema;
         this.streaming = streaming;
         this.partitionFilters = partitionFilters;
@@ -138,6 +143,7 @@ public SplitEnumerator<SourceSplitBase, SourceEnumeratorState> createEnumerator(
                 splitEnumeratorContext,
                 offsetsInitializer,
                 scanPartitionDiscoveryIntervalMs,
+                scanBucketDiscoveryIntervalMs,
                 streaming,
                 partitionFilters,
                 lakeSource);
@@ -158,6 +164,7 @@ public SplitEnumerator<SourceSplitBase, SourceEnumeratorState> restoreEnumerator
                 sourceEnumeratorState.getRemainingHybridLakeFlussSplits(),
                 offsetsInitializer,
                 scanPartitionDiscoveryIntervalMs,
+                scanBucketDiscoveryIntervalMs,
                 streaming,
                 partitionFilters,
                 lakeSource);

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

@@ -131,6 +131,7 @@ public class FlinkTableSource
     @Nullable private final LookupCache cache;
 
     private final long scanPartitionDiscoveryIntervalMs;
+    private final long scanBucketDiscoveryIntervalMs;
     private final boolean isDataLakeEnabled;
     @Nullable private final MergeEngineType mergeEngineType;
 
@@ -168,6 +169,7 @@ public FlinkTableSource(
             boolean lookupAsync,
             @Nullable LookupCache cache,
             long scanPartitionDiscoveryIntervalMs,
+            long scanBucketDiscoveryIntervalMs,
             boolean isDataLakeEnabled,
             @Nullable MergeEngineType mergeEngineType,
             Map<String, String> tableOptions) {
@@ -185,6 +187,7 @@ public FlinkTableSource(
         this.cache = cache;
 
         this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs;
+        this.scanBucketDiscoveryIntervalMs = scanBucketDiscoveryIntervalMs;
         this.isDataLakeEnabled = isDataLakeEnabled;
         this.mergeEngineType = mergeEngineType;
         this.tableOptions = tableOptions;
@@ -328,6 +331,7 @@ public boolean isBounded() {
                         projectedFields,
                         offsetsInitializer,
                         scanPartitionDiscoveryIntervalMs,
+                        scanBucketDiscoveryIntervalMs,
                         new RowDataDeserializationSchema(),
                         streaming,
                         partitionFilters,
@@ -446,6 +450,7 @@ public DynamicTableSource copy() {
                         lookupAsync,
                         cache,
                         scanPartitionDiscoveryIntervalMs,
+                        scanBucketDiscoveryIntervalMs,
                         isDataLakeEnabled,
                         mergeEngineType,
                         tableOptions);

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

@@ -42,6 +42,7 @@
  *     .setProjectedFields("orderId", "amount")
  *     .setStartingOffsets(OffsetsInitializer.earliest())
  *     .setScanPartitionDiscoveryIntervalMs(1000L)
+ *     .setScanBucketDiscoveryIntervalMs(1000L)
  *     .setDeserializationSchema(new OrderDeserializationSchema())
  *     .build();
  *
@@ -66,6 +67,7 @@ public class FlussSource<OUT> extends FlinkSource<OUT> {
             @Nullable int[] projectedFields,
             OffsetsInitializer offsetsInitializer,
             long scanPartitionDiscoveryIntervalMs,
+            long scanBucketDiscoveryIntervalMs,
             FlussDeserializationSchema<OUT> deserializationSchema,
             boolean streaming) {
         // TODO: Support partition pushDown in datastream
@@ -78,6 +80,7 @@ public class FlussSource<OUT> extends FlinkSource<OUT> {
                 projectedFields,
                 offsetsInitializer,
                 scanPartitionDiscoveryIntervalMs,
+                scanBucketDiscoveryIntervalMs,
                 deserializationSchema,
                 streaming,
                 null);

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

@@ -69,6 +69,7 @@ public class FlussSourceBuilder<OUT> {
     private int[] projectedFields;
     private String[] projectedFieldNames;
     private Long scanPartitionDiscoveryIntervalMs;
+    private Long scanBucketDiscoveryIntervalMs;
     private OffsetsInitializer offsetsInitializer;
     private FlussDeserializationSchema<OUT> deserializationSchema;
 
@@ -131,6 +132,21 @@ public FlussSourceBuilder<OUT> setScanPartitionDiscoveryIntervalMs(
         return this;
     }
 
+    /**
+     * Sets the scan bucket discovery interval in milliseconds.
+     *
+     * <p>If not specified, the default value from {@link
+     * FlinkConnectorOptions#SCAN_BUCKET_DISCOVERY_INTERVAL} is used.
+     *
+     * @param scanBucketDiscoveryIntervalMs interval in milliseconds
+     * @return this builder
+     */
+    public FlussSourceBuilder<OUT> setScanBucketDiscoveryIntervalMs(
+            long scanBucketDiscoveryIntervalMs) {
+        this.scanBucketDiscoveryIntervalMs = scanBucketDiscoveryIntervalMs;
+        return this;
+    }
+
     /**
      * Sets the starting offsets strategy for the Fluss source.
      *
@@ -225,6 +241,12 @@ public FlussSource<OUT> build() {
                             .toMillis();
         }
 
+        // if null use the default value:
+        if (scanBucketDiscoveryIntervalMs == null) {
+            scanBucketDiscoveryIntervalMs =
+                    FlinkConnectorOptions.SCAN_BUCKET_DISCOVERY_INTERVAL.defaultValue().toMillis();
+        }
+
         if (this.flussConf == null) {
             this.flussConf = new Configuration();
         }
@@ -299,6 +321,7 @@ public FlussSource<OUT> build() {
                 projectedFields,
                 offsetsInitializer,
                 scanPartitionDiscoveryIntervalMs,
+                scanBucketDiscoveryIntervalMs,
                 deserializationSchema,
                 true);
     }