[flink] Support consuming historical partition changelog

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeSplitGenerator.java

@@ -47,6 +47,7 @@
 import static org.apache.fluss.client.table.scanner.log.LogScanner.EARLIEST_OFFSET;
 import static org.apache.fluss.flink.source.split.LogSplit.NO_STOPPING_OFFSET;
 import static org.apache.fluss.metadata.ResolvedPartitionSpec.PARTITION_SPEC_SEPARATOR;
+import static org.apache.fluss.utils.PartitionUtils.HISTORICAL_PARTITION_VALUE;
 
 /** A generator for lake splits. */
 public class LakeSplitGenerator {
@@ -148,6 +149,9 @@ private List<SourceSplitBase> generatePartitionTableSplit(
                                         (existing, replacement) -> existing,
                                         LinkedHashMap::new));
         long lakeSplitPartitionId = -1L;
+        boolean hasHistoricalPartition =
+                !isLogTable && flussPartitionByName.containsKey(HISTORICAL_PARTITION_VALUE);
+        Map<Integer, List<LakeSplit>> historicalLakeSplits = new HashMap<>();
 
         // iterate lake splits
         for (Map.Entry<String, Map<Integer, List<LakeSplit>>> lakeSplitEntry :
@@ -175,6 +179,16 @@ private List<SourceSplitBase> generatePartitionTableSplit(
                                 tableBucketSnapshotLogOffset,
                                 bucketEndOffset));
 
+            } else if (hasHistoricalPartition) {
+                // Tiering preserves the Fluss bucket for each business partition. Group its lake
+                // splits with the corresponding historical log so the existing hybrid reader
+                // emits the complete baseline before any later changes for those keys.
+                lakeSplitsOfPartition.forEach(
+                        (bucket, bucketSplits) -> {
+                            historicalLakeSplits
+                                    .computeIfAbsent(bucket, ignored -> new ArrayList<>())
+                                    .addAll(bucketSplits);
+                        });
             } else {
                 // only lake data
                 splits.addAll(
@@ -193,6 +207,8 @@ private List<SourceSplitBase> generatePartitionTableSplit(
         // iterate remain fluss splits
         for (PartitionInfo flussPartition : flussPartitionByName.values()) {
             String partitionName = flussPartition.getPartitionName();
+            boolean historical =
+                    hasHistoricalPartition && HISTORICAL_PARTITION_VALUE.equals(partitionName);
             int partitionBucketCount = flussPartition.getBucketCount();
             Map<Integer, Long> bucketEndOffset =
                     stoppingOffsetInitializer.getBucketOffsets(
@@ -203,7 +219,7 @@ private List<SourceSplitBase> generatePartitionTableSplit(
                             bucketOffsetsRetriever);
             splits.addAll(
                     generateSplit(
-                            null,
+                            historical ? historicalLakeSplits : null,
                             flussPartition.getPartitionId(),
                             partitionName,
                             partitionBucketCount,
@@ -321,11 +337,13 @@ private SourceSplitBase generateSplitForPrimaryKeyTableBucket(
             @Nullable String partitionName,
             @Nullable Long snapshotLogOffset,
             long stoppingOffset) {
-        // no snapshot data for this bucket or no a corresponding log offset in this bucket,
-        // can only scan from change log
         if (snapshotLogOffset == null || snapshotLogOffset < 0) {
-            return new LakeSnapshotAndFlussLogSplit(
-                    tableBucket, partitionName, null, EARLIEST_OFFSET, stoppingOffset);
+            snapshotLogOffset = EARLIEST_OFFSET;
+            // Historical lake splits are baselines from expired normal partitions. Keep them even
+            // when the historical changelog has no tiered offset yet.
+            if (!HISTORICAL_PARTITION_VALUE.equals(partitionName)) {
+                lakeSplits = null;
+            }
         }
 
         return new LakeSnapshotAndFlussLogSplit(

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/reader/SeekableLakeSnapshotSplitScanner.java

@@ -71,14 +71,12 @@ public CloseableIterator<InternalRow> pollBatch(Duration timeout) throws IOExcep
             updateCurrentIterator();
         }
 
-        // has no next record in currentIterator, update currentIterator
-        if (currentLakeRecordIterator != null && !currentLakeRecordIterator.hasNext()) {
+        // A lake split can be empty after filtering. Only finish after checking every split.
+        while (currentLakeRecordIterator != null && !currentLakeRecordIterator.hasNext()) {
             updateCurrentIterator();
         }
 
-        return currentLakeRecordIterator != null && currentLakeRecordIterator.hasNext()
-                ? currentLakeRecordIterator
-                : null;
+        return currentLakeRecordIterator;
     }
 
     private void updateCurrentIterator() throws IOException {

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

@@ -87,6 +87,7 @@
 import java.util.TreeMap;
 import java.util.stream.Collectors;
 
+import static org.apache.fluss.utils.PartitionUtils.HISTORICAL_PARTITION_VALUE;
 import static org.apache.fluss.utils.Preconditions.checkArgument;
 import static org.apache.fluss.utils.Preconditions.checkNotNull;
 import static org.apache.fluss.utils.Preconditions.checkState;
@@ -846,9 +847,33 @@ private Set<PartitionInfo> listPartitions() {
             return Collections.emptySet();
         }
         try {
-            List<PartitionInfo> partitionInfos = flussAdmin.listPartitionInfos(tablePath).get();
+            boolean includeHistoricalPartition =
+                    streaming
+                            && hasPrimaryKey
+                            && tableInfo.getTableConfig().isHistoricalPartitionEnabled();
+            List<PartitionInfo> partitionInfos =
+                    flussAdmin.listPartitionInfos(tablePath, includeHistoricalPartition).get();
+            List<PartitionInfo> historicalPartitions = Collections.emptyList();
+            if (includeHistoricalPartition) {
+                historicalPartitions =
+                        partitionInfos.stream()
+                                .filter(
+                                        partition ->
+                                                HISTORICAL_PARTITION_VALUE.equals(
+                                                        partition.getPartitionName()))
+                                .collect(Collectors.toList());
+                partitionInfos =
+                        partitionInfos.stream()
+                                .filter(
+                                        partition ->
+                                                !HISTORICAL_PARTITION_VALUE.equals(
+                                                        partition.getPartitionName()))
+                                .collect(Collectors.toList());
+            }
             partitionInfos = applyPartitionFilter(partitionInfos);
-            return new LinkedHashSet<>(partitionInfos);
+            Set<PartitionInfo> partitions = new LinkedHashSet<>(partitionInfos);
+            partitions.addAll(historicalPartitions);
+            return partitions;
         } catch (Exception e) {
             throw new FlinkRuntimeException(
                     String.format("Failed to list partitions for %s", tablePath),
@@ -1077,6 +1102,17 @@ private List<SourceSplitBase> initPrimaryKeyTablePartitionSplits(
         List<SourceSplitBase> splits = new ArrayList<>();
         for (Partition partition : newPartitions) {
             String partitionName = partition.getPartitionName();
+            if (HISTORICAL_PARTITION_VALUE.equals(partitionName)) {
+                // Historical partitions have no KV snapshot. Without a lake snapshot, consume
+                // their retained changelog using the existing full-mode fallback.
+                splits.addAll(
+                        getLogSplit(
+                                partition.getPartitionId(),
+                                partitionName,
+                                OffsetsInitializer.earliest(),
+                                partition.getBucketCount()));
+                continue;
+            }
             splits.addAll(
                     getSnapshotAndLogSplits(
                             getLatestKvSnapshotsAndRegister(partitionName), partitionName));

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/lake/LakeSplitGeneratorTest.java

@@ -17,6 +17,8 @@
 
 package org.apache.fluss.flink.lake;
 
+import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.client.initializer.NoStoppingOffsetsInitializer;
 import org.apache.fluss.client.initializer.OffsetsInitializer;
 import org.apache.fluss.client.metadata.LakeSnapshot;
 import org.apache.fluss.flink.lake.split.LakeSnapshotAndFlussLogSplit;
@@ -39,23 +41,32 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
 import static org.apache.fluss.client.table.scanner.log.LogScanner.EARLIEST_OFFSET;
+import static org.apache.fluss.utils.PartitionUtils.HISTORICAL_PARTITION_VALUE;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 /** Tests lake and log split planning for partitioned tables. */
 class LakeSplitGeneratorTest {
 
     /** Table-level bucket count, kept different from the per-partition counts used below. */
     private static final int TABLE_LEVEL_BUCKET_COUNT = 3;
 
+    private static final long TABLE_ID = 1L;
+    private static final PartitionInfo ACTIVE = partition(10L, "20260916");
+    private static final PartitionInfo HISTORICAL = partition(20L, HISTORICAL_PARTITION_VALUE);
+
     /**
      * Builds a {@link LakeSplitGenerator} for a partitioned primary-key table (schema: a INT, b
      * STRING, c STRING; PK a+c) whose single partition "p" has {@code partitionBucketCount}
@@ -248,4 +259,114 @@ public Map<Integer, Long> latestOffsets(
                                         : split.asLogSplit().getStoppingOffset().get())
                 .containsExactly(110L, 20L);
     }
+
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    void testHistoricalBaselineGroupedByBucket(boolean hasHistoricalOffset) throws Exception {
+        Map<TableBucket, Long> offsets = new HashMap<>();
+        for (int bucket = 0; bucket < 2; bucket++) {
+            offsets.put(new TableBucket(TABLE_ID, ACTIVE.getPartitionId(), bucket), 100L);
+            if (hasHistoricalOffset) {
+                offsets.put(new TableBucket(TABLE_ID, HISTORICAL.getPartitionId(), bucket), 50L);
+            }
+        }
+        List<SourceSplitBase> splits =
+                generateSplits(
+                        Arrays.asList(ACTIVE, partition(1L, "20240101"), partition(2L, "20240102")),
+                        offsets);
+
+        assertThat(splits).hasSize(4);
+        for (SourceSplitBase split : splits) {
+            assertThat(split).isInstanceOf(LakeSnapshotAndFlussLogSplit.class);
+            LakeSnapshotAndFlussLogSplit hybrid = (LakeSnapshotAndFlussLogSplit) split;
+            assertThat(hybrid.isLakeSplitFinished()).isFalse();
+            assertThat(hybrid.getLakeSplits())
+                    .allSatisfy(
+                            lakeSplit ->
+                                    assertThat(lakeSplit.bucket())
+                                            .isEqualTo(split.getTableBucket().getBucket()));
+            if (HISTORICAL_PARTITION_VALUE.equals(split.getPartitionName())) {
+                assertThat(split.getTableBucket().getPartitionId())
+                        .isEqualTo(HISTORICAL.getPartitionId());
+                assertThat(hybrid.getStartingOffset())
+                        .isEqualTo(hasHistoricalOffset ? 50L : EARLIEST_OFFSET);
+                assertThat(hybrid.getLakeSplits())
+                        .extracting(LakeSplit::partition)
+                        .containsExactlyInAnyOrder(
+                                Collections.singletonList("20240101"),
+                                Collections.singletonList("20240102"));
+            } else {
+                assertThat(hybrid.getStartingOffset()).isEqualTo(100L);
+                assertThat(hybrid.getLakeSplits())
+                        .extracting(LakeSplit::partition)
+                        .containsExactly(Collections.singletonList(ACTIVE.getPartitionName()));
+            }
+        }
+    }
+
+    @Test
+    void testHistoricalOffsetWithoutMatchingLakeSplits() throws Exception {
+        TableBucket historicalBucket = new TableBucket(TABLE_ID, HISTORICAL.getPartitionId(), 0);
+        // Partition predicates can prune every lake split without changing the shared log offset.
+        List<SourceSplitBase> splits =
+                generateSplits(
+                        Collections.emptyList(), Collections.singletonMap(historicalBucket, 50L));
+        assertThat(splits)
+                .filteredOn(split -> HISTORICAL_PARTITION_VALUE.equals(split.getPartitionName()))
+                .hasSize(2)
+                .allSatisfy(
+                        split -> {
+                            assertThat(split).isInstanceOf(LakeSnapshotAndFlussLogSplit.class);
+                            LakeSnapshotAndFlussLogSplit hybrid =
+                                    (LakeSnapshotAndFlussLogSplit) split;
+                            assertThat(hybrid.getLakeSplits()).isNull();
+                            assertThat(hybrid.isLakeSplitFinished()).isTrue();
+                            assertThat(hybrid.getStartingOffset())
+                                    .isEqualTo(
+                                            split.getTableBucket().equals(historicalBucket)
+                                                    ? 50L
+                                                    : EARLIEST_OFFSET);
+                        });
+    }
+
+    private List<SourceSplitBase> generateSplits(
+            List<PartitionInfo> lakePartitions, Map<TableBucket, Long> offsets) throws Exception {
+        TablePath tablePath = TablePath.of("db", "historical");
+        TableInfo tableInfo =
+                TableInfo.of(
+                        tablePath,
+                        TABLE_ID,
+                        0,
+                        TableDescriptor.builder()
+                                .schema(
+                                        Schema.newBuilder()
+                                                .column("id", DataTypes.INT())
+                                                .column("dt", DataTypes.STRING())
+                                                .primaryKey("id", "dt")
+                                                .build())
+                                .distributedBy(2, "id")
+                                .partitionedBy("dt")
+                                .build(),
+                        null,
+                        0,
+                        0);
+        Admin admin = mock(Admin.class);
+        when(admin.getReadableLakeSnapshot(tablePath))
+                .thenReturn(CompletableFuture.completedFuture(new LakeSnapshot(1L, offsets)));
+        LakeSplitGenerator generator =
+                new LakeSplitGenerator(
+                        tableInfo,
+                        admin,
+                        new TestingLakeSource(2, lakePartitions),
+                        mock(OffsetsInitializer.BucketOffsetsRetriever.class),
+                        new NoStoppingOffsetsInitializer(),
+                        2,
+                        () -> new HashSet<>(Arrays.asList(ACTIVE, HISTORICAL)));
+        return generator.generateHybridLakeFlussSplits();
+    }
+
+    private static PartitionInfo partition(long id, String value) {
+        return new PartitionInfo(
+                id, ResolvedPartitionSpec.fromPartitionValue("dt", value), null, 2);
+    }
 }

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/lake/reader/SeekableLakeSnapshotSplitScannerTest.java

@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.fluss.flink.lake.reader;
+
+import org.apache.fluss.lake.source.LakeSplit;
+import org.apache.fluss.lake.source.RecordReader;
+import org.apache.fluss.lake.source.TestingLakeSource;
+import org.apache.fluss.lake.source.TestingLakeSplit;
+import org.apache.fluss.record.LogRecord;
+import org.apache.fluss.row.InternalRow;
+import org.apache.fluss.utils.CloseableIterator;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+
+import static org.apache.fluss.testutils.DataTestUtils.row;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests for reading all lake splits, including splits made empty by filtering. */
+class SeekableLakeSnapshotSplitScannerTest {
+
+    @Test
+    void testEmptySplitsDoNotFinishSnapshotEarly() throws Exception {
+        TestingLakeSource source =
+                new TestingLakeSource() {
+                    @Override
+                    public RecordReader createRecordReader(ReaderContext<LakeSplit> context) {
+                        String partition = context.lakeSplit().partition().get(0);
+                        if (partition.equals("empty")) {
+                            return CloseableIterator::emptyIterator;
+                        }
+                        LogRecord record = mock(LogRecord.class);
+                        when(record.getRow()).thenReturn(row(partition));
+                        return () ->
+                                CloseableIterator.wrap(Collections.singleton(record).iterator());
+                    }
+                };
+        try (SeekableLakeSnapshotSplitScanner scanner =
+                new SeekableLakeSnapshotSplitScanner(
+                        source,
+                        Arrays.asList(
+                                split("empty"),
+                                split("empty"),
+                                split("a"),
+                                split("empty"),
+                                split("b"),
+                                split("empty")),
+                        0)) {
+            try (CloseableIterator<InternalRow> first = scanner.pollBatch(Duration.ZERO)) {
+                assertThat(first).isNotNull();
+                assertThat(first.next().getString(0).toString()).isEqualTo("a");
+                assertThat(first.hasNext()).isFalse();
+            }
+            try (CloseableIterator<InternalRow> second = scanner.pollBatch(Duration.ZERO)) {
+                assertThat(second).isNotNull();
+                assertThat(second.next().getString(0).toString()).isEqualTo("b");
+                assertThat(second.hasNext()).isFalse();
+            }
+            assertThat(scanner.pollBatch(Duration.ZERO)).isNull();
+        }
+    }
+
+    private static LakeSplit split(String partition) {
+        return new TestingLakeSplit(0, Collections.singletonList(partition));
+    }
+}