[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,12 @@ private List<SourceSplitBase> generatePartitionTableSplit(
                                         (existing, replacement) -> existing,
                                         LinkedHashMap::new));
         long lakeSplitPartitionId = -1L;
+        // Primary-key tables must emit a key's snapshot row before its later updates/deletes.
+        // Group their lake splits with the historical log to preserve this order.
+        // Log tables only append records, so their lake splits can be read independently.
+        boolean groupHistoricalLakeSplits =
+                !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 +182,17 @@ private List<SourceSplitBase> generatePartitionTableSplit(
                                 tableBucketSnapshotLogOffset,
                                 bucketEndOffset));
 
+            } else if (groupHistoricalLakeSplits) {
+                // Expired business partitions still have data in the lake, but share the
+                // historical partition for later writes. Lake data and historical writes use the
+                // same bucket IDs, so collect all expired partitions' lake splits for each bucket.
+                // A single hybrid split will read these snapshots before that bucket's log.
+                lakeSplitsOfPartition.forEach(
+                        (bucket, bucketSplits) -> {
+                            historicalLakeSplits
+                                    .computeIfAbsent(bucket, ignored -> new ArrayList<>())
+                                    .addAll(bucketSplits);
+                        });
             } else {
                 // only lake data
                 splits.addAll(
@@ -193,6 +211,8 @@ private List<SourceSplitBase> generatePartitionTableSplit(
         // iterate remain fluss splits
         for (PartitionInfo flussPartition : flussPartitionByName.values()) {
             String partitionName = flussPartition.getPartitionName();
+            boolean historical =
+                    groupHistoricalLakeSplits && HISTORICAL_PARTITION_VALUE.equals(partitionName);
             int partitionBucketCount = flussPartition.getBucketCount();
             Map<Integer, Long> bucketEndOffset =
                     stoppingOffsetInitializer.getBucketOffsets(
@@ -203,7 +223,7 @@ private List<SourceSplitBase> generatePartitionTableSplit(
                             bucketOffsetsRetriever);
             splits.addAll(
                     generateSplit(
-                            null,
+                            historical ? historicalLakeSplits : null,
                             flussPartition.getPartitionId(),
                             partitionName,
                             partitionBucketCount,
@@ -321,15 +341,24 @@ 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) {
+        if (snapshotLogOffset != null && snapshotLogOffset >= 0) {
+            // Read the lake snapshot, then continue from its corresponding log offset.
             return new LakeSnapshotAndFlussLogSplit(
-                    tableBucket, partitionName, null, EARLIEST_OFFSET, stoppingOffset);
+                    tableBucket, partitionName, lakeSplits, snapshotLogOffset, stoppingOffset);
         }
 
+        if (HISTORICAL_PARTITION_VALUE.equals(partitionName)) {
+            // Rows may have been tiered before their business partition expired, while no
+            // historical writes have been tiered yet. Keep those lake rows and read the
+            // historical log from earliest; the historical log alone cannot recover those rows.
+            return new LakeSnapshotAndFlussLogSplit(
+                    tableBucket, partitionName, lakeSplits, EARLIEST_OFFSET, stoppingOffset);
+        }
+
+        // Without a snapshot offset, we cannot tell where an ordinary partition's lake snapshot
+        // ends in its log. Fall back to reading only the retained log to avoid overlapping reads.
         return new LakeSnapshotAndFlussLogSplit(
-                tableBucket, partitionName, lakeSplits, snapshotLogOffset, stoppingOffset);
+                tableBucket, partitionName, null, EARLIEST_OFFSET, stoppingOffset);
     }
 
     private List<SourceSplitBase> generateNoPartitionedTableSplit(

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,48 +847,54 @@ private Set<PartitionInfo> listPartitions() {
             return Collections.emptySet();
         }
         try {
-            List<PartitionInfo> partitionInfos = flussAdmin.listPartitionInfos(tablePath).get();
-            partitionInfos = applyPartitionFilter(partitionInfos);
-            return new LinkedHashSet<>(partitionInfos);
+            boolean includeHistoricalPartition =
+                    streaming && tableInfo.getTableConfig().isHistoricalPartitionEnabled();
+            List<PartitionInfo> partitionInfos =
+                    flussAdmin.listPartitionInfos(tablePath, includeHistoricalPartition).get();
+            return new LinkedHashSet<>(applyPartitionFilter(partitionInfos));
         } catch (Exception e) {
             throw new FlinkRuntimeException(
                     String.format("Failed to list partitions for %s", tablePath),
                     ExceptionUtils.stripCompletionException(e));
         }
     }
 
-    /** Apply partition filter. */
+    /** Apply partition filter to ordinary partitions, retaining the shared historical partition. */
     private List<PartitionInfo> applyPartitionFilter(List<PartitionInfo> partitionInfos) {
         if (partitionFilters == null) {
             return partitionInfos;
         } else {
             int originalSize = partitionInfos.size();
             RowType partitionRowType = PartitionUtils.partitionRowType(tableInfo);
+            // Historical logs contain multiple business partitions. Keep their subscription and
+            // let Flink filter each record using its original business partition values.
             List<PartitionInfo> filteredPartitionInfos =
                     partitionInfos.stream()
                             .filter(
                                     partition ->
-                                            partitionFilters.test(
-                                                    PartitionUtils.toPartitionRow(
-                                                            partition
-                                                                    .getResolvedPartitionSpec()
-                                                                    .getPartitionValues(),
-                                                            partitionRowType)))
+                                            HISTORICAL_PARTITION_VALUE.equals(
+                                                            partition.getPartitionName())
+                                                    || partitionFilters.test(
+                                                            PartitionUtils.toPartitionRow(
+                                                                    partition
+                                                                            .getResolvedPartitionSpec()
+                                                                            .getPartitionValues(),
+                                                                    partitionRowType)))
                             .collect(Collectors.toList());
 
             int filteredSize = filteredPartitionInfos.size();
             if (originalSize != filteredSize) {
                 LOG.debug(
                         "Applied partition filter for table {}: {} partitions filtered down to {} "
-                                + "matching partitions with predicate: {}. Matching partitions after filtering: {}",
+                                + "retained partitions with predicate: {}. Partitions after filtering: {}",
                         tablePath,
                         originalSize,
                         filteredSize,
                         partitionFilters,
                         filteredPartitionInfos);
             } else {
                 LOG.debug(
-                        "Partition filter applied for table {}, but all {} partitions matched the predicate",
+                        "Partition filter applied for table {}, but all {} partitions were retained",
                         tablePath,
                         originalSize);
             }
@@ -1077,6 +1084,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,9 +17,11 @@
 
 package org.apache.fluss.flink.lake;
 
+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;
+import org.apache.fluss.flink.lake.split.LakeSnapshotSplit;
 import org.apache.fluss.flink.sink.testutils.TestAdminAdapter;
 import org.apache.fluss.flink.source.split.SourceSplitBase;
 import org.apache.fluss.lake.source.LakeSource;
@@ -37,16 +39,22 @@
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 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 java.util.stream.Stream;
 
 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;
 
@@ -56,6 +64,10 @@ 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 +260,142 @@ public Map<Integer, Long> latestOffsets(
                                         : split.asLogSplit().getStoppingOffset().get())
                 .containsExactly(110L, 20L);
     }
+
+    private static Stream<Arguments> historicalPartitionCases() {
+        return Stream.of(
+                Arguments.of("primary-key table with lake splits", true, true),
+                Arguments.of("primary-key table with all lake splits pruned", true, false),
+                Arguments.of("log table with lake splits", false, true),
+                Arguments.of("log table with all lake splits pruned", false, false));
+    }
+
+    @ParameterizedTest(name = "{0}")
+    @MethodSource("historicalPartitionCases")
+    void testHistoricalPartitionSplits(
+            String description, boolean primaryKey, boolean hasLakeSplits) 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);
+        }
+        // Historical bucket 0 resumes at its tiered offset; bucket 1 starts from earliest.
+        offsets.put(new TableBucket(TABLE_ID, HISTORICAL.getPartitionId(), 0), 50L);
+        // Partition predicates may prune every lake split without changing the log offsets.
+        List<SourceSplitBase> splits =
+                generateSplits(
+                        primaryKey,
+                        hasLakeSplits
+                                ? Arrays.asList(
+                                        ACTIVE,
+                                        partition(1L, "20240101"),
+                                        partition(2L, "20240102"))
+                                : Collections.emptyList(),
+                        offsets);
+
+        // Both table types retain four log subscriptions. Log tables also have six independent
+        // lake splits; primary-key tables group the lake splits with the corresponding logs.
+        boolean hasIndependentLakeSplits = !primaryKey && hasLakeSplits;
+        assertThat(splits).hasSize(hasIndependentLakeSplits ? 10 : 4);
+        assertThat(splits)
+                .filteredOn(split -> !(split instanceof LakeSnapshotSplit))
+                .extracting(SourceSplitBase::getTableBucket)
+                .containsExactlyInAnyOrder(
+                        new TableBucket(TABLE_ID, ACTIVE.getPartitionId(), 0),
+                        new TableBucket(TABLE_ID, ACTIVE.getPartitionId(), 1),
+                        new TableBucket(TABLE_ID, HISTORICAL.getPartitionId(), 0),
+                        new TableBucket(TABLE_ID, HISTORICAL.getPartitionId(), 1));
+        if (hasIndependentLakeSplits) {
+            assertThat(splits)
+                    .filteredOn(split -> split instanceof LakeSnapshotSplit)
+                    .extracting(SourceSplitBase::getPartitionName)
+                    .containsExactlyInAnyOrder(
+                            ACTIVE.getPartitionName(),
+                            ACTIVE.getPartitionName(),
+                            "20240101",
+                            "20240101",
+                            "20240102",
+                            "20240102");
+        }
+        for (SourceSplitBase split : splits) {
+            if (split instanceof LakeSnapshotSplit) {
+                continue;
+            }
+            long expectedOffset = offsets.getOrDefault(split.getTableBucket(), EARLIEST_OFFSET);
+            if (!primaryKey) {
+                assertThat(split.isLogSplit()).isTrue();
+                assertThat(split.asLogSplit().getStartingOffset()).isEqualTo(expectedOffset);
+                continue;
+            }
+            assertThat(split).isInstanceOf(LakeSnapshotAndFlussLogSplit.class);
+            LakeSnapshotAndFlussLogSplit hybrid = (LakeSnapshotAndFlussLogSplit) split;
+            assertThat(hybrid.getStartingOffset()).isEqualTo(expectedOffset);
+            assertThat(hybrid.isLakeSplitFinished()).isEqualTo(!hasLakeSplits);
+            if (!hasLakeSplits) {
+                assertThat(hybrid.getLakeSplits()).isNull();
+                continue;
+            }
+            assertThat(hybrid.getLakeSplits())
+                    .allSatisfy(
+                            lakeSplit ->
+                                    assertThat(lakeSplit.bucket())
+                                            .isEqualTo(split.getTableBucket().getBucket()));
+            if (HISTORICAL_PARTITION_VALUE.equals(split.getPartitionName())) {
+                assertThat(hybrid.getLakeSplits())
+                        .extracting(LakeSplit::partition)
+                        .containsExactlyInAnyOrder(
+                                Collections.singletonList("20240101"),
+                                Collections.singletonList("20240102"));
+            } else {
+                assertThat(hybrid.getLakeSplits())
+                        .extracting(LakeSplit::partition)
+                        .containsExactly(Collections.singletonList(ACTIVE.getPartitionName()));
+            }
+        }
+    }
+
+    private List<SourceSplitBase> generateSplits(
+            boolean primaryKey, List<PartitionInfo> lakePartitions, Map<TableBucket, Long> offsets)
+            throws Exception {
+        TablePath tablePath = TablePath.of("db", "historical");
+        Schema.Builder schemaBuilder =
+                Schema.newBuilder().column("id", DataTypes.INT()).column("dt", DataTypes.STRING());
+        if (primaryKey) {
+            schemaBuilder.primaryKey("id", "dt");
+        }
+        TableInfo tableInfo =
+                TableInfo.of(
+                        tablePath,
+                        TABLE_ID,
+                        0,
+                        TableDescriptor.builder()
+                                .schema(schemaBuilder.build())
+                                .distributedBy(2, "id")
+                                .partitionedBy("dt")
+                                .build(),
+                        null,
+                        0,
+                        0);
+        TestAdminAdapter admin =
+                new TestAdminAdapter() {
+                    @Override
+                    public CompletableFuture<LakeSnapshot> getReadableLakeSnapshot(
+                            TablePath ignored) {
+                        return CompletableFuture.completedFuture(new LakeSnapshot(1L, offsets));
+                    }
+                };
+        LakeSplitGenerator generator =
+                new LakeSplitGenerator(
+                        tableInfo,
+                        admin,
+                        new TestingLakeSource(2, lakePartitions),
+                        new ZeroOffsetsRetriever(),
+                        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);
+    }
 }