[flink] Add union read support to datastream (#3432)

* [flink] add union read support to datastream

* [flink] support bounded mode in ds

* [flink] address comments

Author: polyzos

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

@@ -23,6 +23,8 @@
 import org.apache.fluss.flink.FlinkConnectorOptions;
 import org.apache.fluss.flink.source.deserializer.FlussDeserializationSchema;
 import org.apache.fluss.flink.source.reader.LeaseContext;
+import org.apache.fluss.lake.source.LakeSource;
+import org.apache.fluss.lake.source.LakeSplit;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.predicate.Predicate;
 import org.apache.fluss.types.RowType;
@@ -71,7 +73,8 @@ public class FlussSource<OUT> extends FlinkSource<OUT> {
             OffsetsInitializer offsetsInitializer,
             long scanPartitionDiscoveryIntervalMs,
             FlussDeserializationSchema<OUT> deserializationSchema,
-            boolean streaming) {
+            boolean streaming,
+            @Nullable LakeSource<LakeSplit> lakeSource) {
         this(
                 flussConf,
                 tablePath,
@@ -84,7 +87,8 @@ public class FlussSource<OUT> extends FlinkSource<OUT> {
                 scanPartitionDiscoveryIntervalMs,
                 FlinkConnectorOptions.SCAN_SPLIT_ASSIGNMENT_BATCH_SIZE.defaultValue(),
                 deserializationSchema,
-                streaming);
+                streaming,
+                lakeSource);
     }
 
     FlussSource(
@@ -99,7 +103,8 @@ public class FlussSource<OUT> extends FlinkSource<OUT> {
             long scanPartitionDiscoveryIntervalMs,
             int splitPerAssignmentBatchSize,
             FlussDeserializationSchema<OUT> deserializationSchema,
-            boolean streaming) {
+            boolean streaming,
+            @Nullable LakeSource<LakeSplit> lakeSource) {
         // TODO: Support partition pushDown in datastream
         super(
                 flussConf,
@@ -115,6 +120,7 @@ public class FlussSource<OUT> extends FlinkSource<OUT> {
                 deserializationSchema,
                 streaming,
                 null,
+                lakeSource,
                 LeaseContext.DEFAULT);
     }
 

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

@@ -21,10 +21,14 @@
 import org.apache.fluss.client.ConnectionFactory;
 import org.apache.fluss.client.admin.Admin;
 import org.apache.fluss.client.initializer.OffsetsInitializer;
+import org.apache.fluss.client.initializer.SnapshotOffsetsInitializer;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.flink.FlinkConnectorOptions;
 import org.apache.fluss.flink.source.deserializer.FlussDeserializationSchema;
+import org.apache.fluss.flink.utils.LakeSourceUtils;
+import org.apache.fluss.lake.source.LakeSource;
+import org.apache.fluss.lake.source.LakeSplit;
 import org.apache.fluss.metadata.TableInfo;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.predicate.Predicate;
@@ -33,6 +37,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -60,6 +65,11 @@
  *     .build();
  * }</pre>
  *
+ * <p>When the target table has datalake enabled and the source starts in full mode (the default,
+ * {@link OffsetsInitializer#full()}), the built source performs a union read: it reads the
+ * historical data tiered to the lake (e.g. Iceberg, Paimon) together with the real-time data still
+ * in Fluss. Other startup modes (earliest/latest/timestamp) read data from Fluss only.
+ *
  * @param <OUT> The type of records produced by the source being built
  */
 public class FlussSourceBuilder<OUT> {
@@ -73,6 +83,7 @@ public class FlussSourceBuilder<OUT> {
     private Long scanPartitionDiscoveryIntervalMs;
     private Integer splitPerAssignmentBatchSize;
     private OffsetsInitializer offsetsInitializer;
+    private boolean bounded;
     private FlussDeserializationSchema<OUT> deserializationSchema;
 
     private String bootstrapServers;
@@ -161,6 +172,19 @@ public FlussSourceBuilder<OUT> setStartingOffsets(OffsetsInitializer offsetsInit
         return this;
     }
 
+    /**
+     * 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).
+     *
+     * @return this builder
+     */
+    public FlussSourceBuilder<OUT> setBounded() {
+        this.bounded = true;
+        return this;
+    }
+
     /**
      * Sets the deserialization schema for converting Fluss records to output records.
      *
@@ -324,6 +348,40 @@ public FlussSource<OUT> build() {
                         ? tableInfo.getRowType().project(projectedFields)
                         : tableInfo.getRowType();
 
+        // union read (lake historical + Fluss) only applies to full startup mode, like the SQL
+        // connector; other startup modes read Fluss only.
+        boolean lakeEnabled = tableInfo.getTableConfig().isDataLakeEnabled();
+        boolean fullStartup = offsetsInitializer instanceof SnapshotOffsetsInitializer;
+
+        if (bounded && !(lakeEnabled && fullStartup)) {
+            throw new IllegalArgumentException(
+                    String.format(
+                            "Bounded (batch) read requires a datalake-enabled table started in "
+                                    + "full mode (OffsetsInitializer.full()), but table '%s' has "
+                                    + "datalake enabled=%s and full startup mode=%s.",
+                            tablePath, lakeEnabled, fullStartup));
+        }
+
+        LakeSource<LakeSplit> lakeSource = null;
+        if (lakeEnabled && fullStartup) {
+            lakeSource =
+                    LakeSourceUtils.createLakeSource(tablePath, tableInfo.getProperties().toMap());
+            if (lakeSource != null) {
+                if (projectedFields != null) {
+                    int[][] nestedProjectedFields = new int[projectedFields.length][];
+                    for (int i = 0; i < projectedFields.length; i++) {
+                        nestedProjectedFields[i] = new int[] {projectedFields[i]};
+                    }
+                    lakeSource.withProject(nestedProjectedFields);
+                }
+                // push the record-batch filter to the lake side as well,
+                // so the historical lake scan is filtered consistently with Fluss.
+                if (logRecordBatchFilter != null) {
+                    lakeSource.withFilters(Collections.singletonList(logRecordBatchFilter));
+                }
+            }
+        }
+
         LOG.info("Creating Fluss Source with Configuration: {}", flussConf);
 
         return new FlussSource<>(
@@ -338,6 +396,7 @@ public FlussSource<OUT> build() {
                 scanPartitionDiscoveryIntervalMs,
                 splitPerAssignmentBatchSize,
                 deserializationSchema,
-                true);
+                !bounded,
+                lakeSource);
     }
 }