[flink] Add restore test for streaming union read log table (#1663)

caozhen1937 · web-flow · commit d863bb4f765f · 2025-09-10T21:12:22.000+08:00
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeSplitReaderGenerator.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeSplitReaderGenerator.java
@@ -68,7 +68,7 @@ public BoundedSplitReader getBoundedSplitScanner(SourceSplitBase split) {
             LakeSnapshotScanner lakeSnapshotScanner =
                     new LakeSnapshotScanner(lakeSource, lakeSnapshotSplit);
             return new BoundedSplitReader(
-                    lakeSnapshotScanner, lakeSnapshotSplit.getRecordsToSplit());
+                    lakeSnapshotScanner, lakeSnapshotSplit.getRecordsToSkip());
         } else if (split instanceof LakeSnapshotAndFlussLogSplit) {
             LakeSnapshotAndFlussLogSplit lakeSplit = (LakeSnapshotAndFlussLogSplit) split;
             return new BoundedSplitReader(getBatchScanner(lakeSplit), lakeSplit.getRecordsToSkip());
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/split/LakeSnapshotSplit.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/split/LakeSnapshotSplit.java
@@ -30,7 +30,7 @@ public class LakeSnapshotSplit extends SourceSplitBase {
 
     private final LakeSplit lakeSplit;
 
-    private final long recordsToSplit;
+    private final long recordsToSkip;
 
     private final int splitIndex;
 
@@ -47,19 +47,19 @@ public LakeSnapshotSplit(
             @Nullable String partitionName,
             LakeSplit lakeSplit,
             int splitIndex,
-            long recordsToSplit) {
+            long recordsToSkip) {
         super(tableBucket, partitionName);
         this.lakeSplit = lakeSplit;
         this.splitIndex = splitIndex;
-        this.recordsToSplit = recordsToSplit;
+        this.recordsToSkip = recordsToSkip;
     }
 
     public LakeSplit getLakeSplit() {
         return lakeSplit;
     }
 
-    public long getRecordsToSplit() {
-        return recordsToSplit;
+    public long getRecordsToSkip() {
+        return recordsToSkip;
     }
 
     public int getSplitIndex() {
@@ -93,8 +93,8 @@ public String toString() {
         return "LakeSnapshotSplit{"
                 + "lakeSplit="
                 + lakeSplit
-                + ", recordsToSplit="
-                + recordsToSplit
+                + ", recordsToSkip="
+                + recordsToSkip
                 + ", splitIndex="
                 + splitIndex
                 + ", tableBucket="
diff --git a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/state/LakeSnapshotSplitState.java b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/state/LakeSnapshotSplitState.java
@@ -25,16 +25,16 @@
 public class LakeSnapshotSplitState extends SourceSplitState {
 
     private final LakeSnapshotSplit split;
-    private long recordsToSplit;
+    private long recordsToSkip;
 
     public LakeSnapshotSplitState(LakeSnapshotSplit split) {
         super(split);
         this.split = split;
-        this.recordsToSplit = split.getRecordsToSplit();
+        this.recordsToSkip = split.getRecordsToSkip();
     }
 
     public void setRecordsToSkip(long recordsToSkip) {
-        this.recordsToSplit = recordsToSkip;
+        this.recordsToSkip = recordsToSkip;
     }
 
     @Override
@@ -44,6 +44,6 @@ public SourceSplitBase toSourceSplit() {
                 split.getPartitionName(),
                 split.getLakeSplit(),
                 split.getSplitIndex(),
-                recordsToSplit);
+                recordsToSkip);
     }
 }
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadLogTableITCase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadLogTableITCase.java
@@ -24,15 +24,20 @@
 import org.apache.fluss.row.TimestampNtz;
 
 import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.SavepointFormatType;
+import org.apache.flink.table.api.TableResult;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.CloseableIterator;
 import org.apache.flink.util.CollectionUtil;
 import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
 import javax.annotation.Nullable;
 
+import java.io.File;
 import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
@@ -42,13 +47,16 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
+import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertResultsExactOrder;
 import static org.apache.fluss.flink.source.testutils.FlinkRowAssertionsUtils.assertResultsIgnoreOrder;
 import static org.apache.fluss.testutils.DataTestUtils.row;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** The IT case for Flink union data in lake and fluss for log table. */
 class FlinkUnionReadLogTableITCase extends FlinkUnionReadTestBase {
 
+    @TempDir public static File savepointDir;
+
     @BeforeAll
     protected static void beforeAll() {
         FlinkUnionReadTestBase.beforeAll();
@@ -169,6 +177,63 @@ void testReadLogTableInStreamMode(boolean isPartitioned) throws Exception {
                 actual, writtenRows.stream().map(Row::toString).collect(Collectors.toList()), true);
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
+    void testUnionReadLogTableFailover(boolean isPartitioned) throws Exception {
+        // first of all, start tiering
+        JobClient jobClient = buildTieringJob(execEnv);
+
+        String tableName1 =
+                "restore_logTable_" + (isPartitioned ? "partitioned" : "non_partitioned");
+        String resultTableName =
+                "result_table" + (isPartitioned ? "partitioned" : "non_partitioned");
+
+        TablePath table1 = TablePath.of(DEFAULT_DB, tableName1);
+        TablePath resultTable = TablePath.of(DEFAULT_DB, resultTableName);
+        List<Row> writtenRows = new LinkedList<>();
+        long tableId = prepareLogTable(table1, DEFAULT_BUCKET_NUM, isPartitioned, writtenRows);
+        // wait until records has been synced
+        waitUntilBucketSynced(table1, tableId, DEFAULT_BUCKET_NUM, isPartitioned);
+
+        StreamTableEnvironment streamTEnv = buildSteamTEnv(null);
+        // now, start to read the log table to write to a fluss result table
+        // may read fluss or not, depends on the log offset of paimon snapshot
+        createFullTypeLogTable(resultTable, DEFAULT_BUCKET_NUM, isPartitioned, false);
+        TableResult insertResult =
+                streamTEnv.executeSql(
+                        "insert into " + resultTableName + " select * from " + tableName1);
+
+        CloseableIterator<Row> actual =
+                streamTEnv.executeSql("select * from " + resultTableName).collect();
+        assertResultsExactOrder(actual, writtenRows, false);
+
+        // now, stop the job with save point
+        String savepointPath =
+                insertResult
+                        .getJobClient()
+                        .get()
+                        .stopWithSavepoint(
+                                false,
+                                savepointDir.getAbsolutePath(),
+                                SavepointFormatType.CANONICAL)
+                        .get();
+
+        // re buildSteamTEnv
+        streamTEnv = buildSteamTEnv(savepointPath);
+        insertResult =
+                streamTEnv.executeSql(
+                        "insert into " + resultTableName + " select * from " + tableName1);
+
+        // write some log data again
+        List<Row> rows = writeRows(table1, 3, isPartitioned);
+
+        assertResultsExactOrder(actual, rows, true);
+
+        // cancel jobs
+        insertResult.getJobClient().get().cancel().get();
+        jobClient.cancel().get();
+    }
+
     private long prepareLogTable(
             TablePath tablePath, int bucketNum, boolean isPartitioned, List<Row> flinkRows)
             throws Exception {
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadTestBase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/flink/FlinkUnionReadTestBase.java
@@ -21,12 +21,15 @@
 import org.apache.fluss.lake.paimon.testutils.FlinkPaimonTieringTestBase;
 import org.apache.fluss.server.testutils.FlussClusterExtension;
 
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
+import javax.annotation.Nullable;
+
 import static org.apache.fluss.flink.FlinkConnectorOptions.BOOTSTRAP_SERVERS;
 
 /** Base class for Flink union read test. */
@@ -60,7 +63,12 @@ protected FlussClusterExtension getFlussClusterExtension() {
         return FLUSS_CLUSTER_EXTENSION;
     }
 
-    private void buildStreamTEnv() {
+    protected StreamTableEnvironment buildSteamTEnv(@Nullable String savepointPath) {
+        Configuration conf = new Configuration();
+        if (savepointPath != null) {
+            conf.setString("execution.savepoint.path", savepointPath);
+            execEnv.configure(conf);
+        }
         String bootstrapServers = String.join(",", clientConf.get(ConfigOptions.BOOTSTRAP_SERVERS));
         // create table environment
         streamTEnv = StreamTableEnvironment.create(execEnv, EnvironmentSettings.inStreamingMode());
@@ -71,6 +79,11 @@ private void buildStreamTEnv() {
                         CATALOG_NAME, BOOTSTRAP_SERVERS.key(), bootstrapServers));
         streamTEnv.executeSql("use catalog " + CATALOG_NAME);
         streamTEnv.executeSql("use " + DEFAULT_DB);
+        return streamTEnv;
+    }
+
+    private void buildStreamTEnv() {
+        buildSteamTEnv(null);
     }
 
     public void buildBatchTEnv() {
diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/FlinkPaimonTieringTestBase.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/testutils/FlinkPaimonTieringTestBase.java
@@ -282,6 +282,12 @@ protected long createLogTable(TablePath tablePath, int bucketNum, boolean isPart
 
     protected long createFullTypeLogTable(TablePath tablePath, int bucketNum, boolean isPartitioned)
             throws Exception {
+        return createFullTypeLogTable(tablePath, bucketNum, isPartitioned, true);
+    }
+
+    protected long createFullTypeLogTable(
+            TablePath tablePath, int bucketNum, boolean isPartitioned, boolean lakeEnabled)
+            throws Exception {
         Schema.Builder schemaBuilder =
                 Schema.newBuilder()
                         .column("f_boolean", DataTypes.BOOLEAN())
@@ -301,10 +307,12 @@ protected long createFullTypeLogTable(TablePath tablePath, int bucketNum, boolea
                         .column("f_binary", DataTypes.BINARY(4));
 
         TableDescriptor.Builder tableBuilder =
-                TableDescriptor.builder()
-                        .distributedBy(bucketNum, "f_int")
-                        .property(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")
-                        .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, Duration.ofMillis(500));
+                TableDescriptor.builder().distributedBy(bucketNum, "f_int");
+        if (lakeEnabled) {
+            tableBuilder
+                    .property(ConfigOptions.TABLE_DATALAKE_ENABLED.key(), "true")
+                    .property(ConfigOptions.TABLE_DATALAKE_FRESHNESS, Duration.ofMillis(500));
+        }
 
         if (isPartitioned) {
             schemaBuilder.column("p", DataTypes.STRING());

Original file line number	Diff line number	Diff line change
`@@ -25,16 +25,16 @@`
`25`	`25`	`public class LakeSnapshotSplitState extends SourceSplitState {`
`26`	`26`
`27`	`27`	`private final LakeSnapshotSplit split;`
`28`		`- private long recordsToSplit;`
	`28`	`+ private long recordsToSkip;`
`29`	`29`
`30`	`30`	`public LakeSnapshotSplitState(LakeSnapshotSplit split) {`
`31`	`31`	`super(split);`
`32`	`32`	`this.split = split;`
`33`		`- this.recordsToSplit = split.getRecordsToSplit();`
	`33`	`+ this.recordsToSkip = split.getRecordsToSkip();`
`34`	`34`	`}`
`35`	`35`
`36`	`36`	`public void setRecordsToSkip(long recordsToSkip) {`
`37`		`- this.recordsToSplit = recordsToSkip;`
	`37`	`+ this.recordsToSkip = recordsToSkip;`
`38`	`38`	`}`
`39`	`39`
`40`	`40`	`@Override`
`@@ -44,6 +44,6 @@ public SourceSplitBase toSourceSplit() {`
`44`	`44`	`split.getPartitionName(),`
`45`	`45`	`split.getLakeSplit(),`
`46`	`46`	`split.getSplitIndex(),`
`47`		`- recordsToSplit);`
	`47`	`+ recordsToSkip);`
`48`	`48`	`}`
`49`	`49`	`}`