[WIP] Flink support detecting bucket change

Author: LiebingYu

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

@@ -100,6 +100,20 @@ public class FlinkConnectorOptions {
                                     + "the new partitions for partitioned table while scanning."
                                     + " A non-positive value disables the partition discovery.");
 
+    public static final ConfigOption<Duration> SCAN_BUCKET_DISCOVERY_INTERVAL =
+            ConfigOptions.key("scan.bucket.discovery.interval")
+                    .durationType()
+                    .defaultValue(Duration.ofSeconds(10))
+                    .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

@@ -127,6 +127,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()),
@@ -142,6 +144,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

@@ -67,6 +67,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;
 
@@ -83,6 +84,7 @@ public FlinkSource(
             @Nullable int[] projectedFields,
             OffsetsInitializer offsetsInitializer,
             long scanPartitionDiscoveryIntervalMs,
+            long scanBucketDiscoveryIntervalMs,
             FlussDeserializationSchema<OUT> deserializationSchema,
             boolean streaming,
             List<FieldEqual> partitionFilters) {
@@ -95,6 +97,7 @@ public FlinkSource(
                 projectedFields,
                 offsetsInitializer,
                 scanPartitionDiscoveryIntervalMs,
+                scanBucketDiscoveryIntervalMs,
                 deserializationSchema,
                 streaming,
                 partitionFilters,
@@ -110,6 +113,7 @@ public FlinkSource(
             @Nullable int[] projectedFields,
             OffsetsInitializer offsetsInitializer,
             long scanPartitionDiscoveryIntervalMs,
+            long scanBucketDiscoveryIntervalMs,
             FlussDeserializationSchema<OUT> deserializationSchema,
             boolean streaming,
             List<FieldEqual> partitionFilters,
@@ -122,6 +126,7 @@ public FlinkSource(
         this.projectedFields = projectedFields;
         this.offsetsInitializer = offsetsInitializer;
         this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs;
+        this.scanBucketDiscoveryIntervalMs = scanBucketDiscoveryIntervalMs;
         this.deserializationSchema = deserializationSchema;
         this.streaming = streaming;
         this.partitionFilters = checkNotNull(partitionFilters);
@@ -144,6 +149,7 @@ public SplitEnumerator<SourceSplitBase, SourceEnumeratorState> createEnumerator(
                 splitEnumeratorContext,
                 offsetsInitializer,
                 scanPartitionDiscoveryIntervalMs,
+                scanBucketDiscoveryIntervalMs,
                 streaming,
                 partitionFilters,
                 lakeSource);
@@ -164,6 +170,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

@@ -125,6 +125,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;
 
@@ -163,6 +164,7 @@ public FlinkTableSource(
             boolean lookupAsync,
             @Nullable LookupCache cache,
             long scanPartitionDiscoveryIntervalMs,
+            long scanBucketDiscoveryIntervalMs,
             boolean isDataLakeEnabled,
             @Nullable MergeEngineType mergeEngineType,
             Map<String, String> tableOptions) {
@@ -181,6 +183,7 @@ public FlinkTableSource(
         this.cache = cache;
 
         this.scanPartitionDiscoveryIntervalMs = scanPartitionDiscoveryIntervalMs;
+        this.scanBucketDiscoveryIntervalMs = scanBucketDiscoveryIntervalMs;
         this.isDataLakeEnabled = isDataLakeEnabled;
         this.mergeEngineType = mergeEngineType;
         this.tableOptions = tableOptions;
@@ -311,6 +314,7 @@ public boolean isBounded() {
                         projectedFields,
                         offsetsInitializer,
                         scanPartitionDiscoveryIntervalMs,
+                        scanBucketDiscoveryIntervalMs,
                         new RowDataDeserializationSchema(),
                         streaming,
                         partitionFilters,
@@ -432,6 +436,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

@@ -44,6 +44,7 @@
  *     .setProjectedFields("orderId", "amount")
  *     .setStartingOffsets(OffsetsInitializer.earliest())
  *     .setScanPartitionDiscoveryIntervalMs(1000L)
+ *     .setScanBucketDiscoveryIntervalMs(1000L)
  *     .setDeserializationSchema(new OrderDeserializationSchema())
  *     .build();
  *
@@ -68,6 +69,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
@@ -80,6 +82,7 @@ public class FlussSource<OUT> extends FlinkSource<OUT> {
                 projectedFields,
                 offsetsInitializer,
                 scanPartitionDiscoveryIntervalMs,
+                scanBucketDiscoveryIntervalMs,
                 deserializationSchema,
                 streaming,
                 Collections.emptyList());

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

@@ -54,6 +54,7 @@
  *     .setTable("orders")
  *     .setProjectedFields("orderId", "amount")
  *     .setScanPartitionDiscoveryIntervalMs(1000L)
+ *     .setScanBucketDiscoveryIntervalMs(1000L)
  *     .setStartingOffsets(OffsetsInitializer.earliest())
  *     .setDeserializationSchema(new OrderDeserializationSchema())
  *     .build();
@@ -69,6 +70,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 +133,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 +242,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 +322,7 @@ public FlussSource<OUT> build() {
                 projectedFields,
                 offsetsInitializer,
                 scanPartitionDiscoveryIntervalMs,
+                scanBucketDiscoveryIntervalMs,
                 deserializationSchema,
                 true);
     }