add restore test for calc on source

Author: xuyangzhong

flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/test/program/SourceTestStep.java

@@ -57,6 +57,19 @@ public static Builder newBuilder(String name) {
         return new Builder(name);
     }
 
+    public SourceTestStep withNewOptions(Map<String, String> newOptions) {
+        return new SourceTestStep(
+                name,
+                schemaComponents,
+                distribution,
+                partitionKeys,
+                newOptions,
+                indexes,
+                dataBeforeRestore,
+                dataAfterRestore,
+                treatDataBeforeRestoreAsFullStageData);
+    }
+
     @Override
     public TestKind getKind() {
         return dataBeforeRestore.isEmpty()

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeltaJoinRestoreTest.java

@@ -37,8 +37,13 @@ public List<TableTestProgram> programs() {
                 DeltaJoinTestPrograms.DELTA_JOIN_WITH_JOIN_KEY_EQUALS_INDEX,
                 DeltaJoinTestPrograms.DELTA_JOIN_WITH_JOIN_KEY_CONTAINS_INDEX,
                 DeltaJoinTestPrograms.DELTA_JOIN_WITH_NON_EQUIV_CONDITION,
+                DeltaJoinTestPrograms.DELTA_JOIN_WITH_CALC_ON_SOURCE,
+                DeltaJoinTestPrograms.DELTA_JOIN_WITH_CALC_ON_SOURCE_AND_FILTER_PUSHED_DOWN,
                 DeltaJoinTestPrograms.DELTA_JOIN_WITH_CACHE,
+                DeltaJoinTestPrograms.DELTA_JOIN_WITH_CACHE_AND_CALC_ON_SOURCE,
                 DeltaJoinTestPrograms.DELTA_JOIN_WITH_CDC_SOURCE_WITHOUT_DELETE,
-                DeltaJoinTestPrograms.DELTA_JOIN_WITH_CACHE_AND_CDC_SOURCE_WITHOUT_DELETE);
+                DeltaJoinTestPrograms.DELTA_JOIN_WITH_CALC_ON_CDC_SOURCE_WITHOUT_DELETE,
+                DeltaJoinTestPrograms.DELTA_JOIN_WITH_CACHE_AND_CDC_SOURCE_WITHOUT_DELETE,
+                DeltaJoinTestPrograms.DELTA_JOIN_WITH_CACHE_AND_CALC_ON_CDC_SOURCE_WITHOUT_DELETE);
     }
 }

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/DeltaJoinTestPrograms.java

@@ -27,6 +27,7 @@
 import org.apache.flink.types.RowKind;
 
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -168,6 +169,70 @@ public class DeltaJoinTestPrograms {
                                     + "on a1 = b1 and a2 <> b2")
                     .build();
 
+    public static final TableTestProgram DELTA_JOIN_WITH_CALC_ON_SOURCE =
+            TableTestProgram.of(
+                            "delta-join-with-calc-on-source",
+                            "validates delta join with calc on source")
+                    .setupConfig(
+                            OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY,
+                            OptimizerConfigOptions.DeltaJoinStrategy.FORCE)
+                    .setupTableSources(
+                            DELTA_JOIN_WITH_JOIN_KEY_EQUALS_INDEX.getSetupSourceTestSteps())
+                    .setupTableSink(
+                            SinkTestStep.newBuilder("snk")
+                                    .addSchema(addPk2Schema(SINK_TABLE_BASE_SCHEMA, "a0", "b0"))
+                                    .addOptions(TABLE_BASE_OPTIONS)
+                                    .testMaterializedData()
+                                    // deduplicate data by pk
+                                    .deduplicatedFieldIndices(new int[] {1, 3})
+                                    .consumedBeforeRestore(Row.of(6, 5.0, "l-5-1", 5.0, "r-5-1", 6))
+                                    .consumedAfterRestore(Row.of(6, 5.0, "l-5-2", 5.0, "r-5-1", 6))
+                                    .build())
+                    .runSql(
+                            "insert into snk "
+                                    + "select new_a1, a0, a2, b0, b2, new_b1 from ( "
+                                    + "   select a0, a1, a2, a1 + 1 as new_a1 from leftSrc "
+                                    + "       where a1 = 1 or a1 = 5 "
+                                    + ") join ("
+                                    + "   select b0, b1, b1 + 1 as new_b1, b2 from rightSrc "
+                                    + "       where b0 = cast(3.0 as double) or b0 = cast(5.0 as double) "
+                                    + ") "
+                                    + "on a1 = b1 and a0 = b0")
+                    .build();
+
+    public static final TableTestProgram DELTA_JOIN_WITH_CALC_ON_SOURCE_AND_FILTER_PUSHED_DOWN =
+            TableTestProgram.of(
+                            "delta-join-with-calc-on-source-and-filter-pushed-down",
+                            "validates delta join with calc on source and filter pushed down")
+                    .setupConfig(
+                            OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY,
+                            OptimizerConfigOptions.DeltaJoinStrategy.FORCE)
+                    .setupConfig(ExecutionConfigOptions.TABLE_EXEC_DELTA_JOIN_CACHE_ENABLED, true)
+                    .setupTableSources(
+                            DELTA_JOIN_WITH_CALC_ON_SOURCE.getSetupSourceTestSteps().stream()
+                                    .map(
+                                            sourceTestStep -> {
+                                                String filterableFields;
+                                                if (sourceTestStep.name.equals("leftSrc")) {
+                                                    filterableFields = "a1";
+                                                } else if (sourceTestStep.name.equals("rightSrc")) {
+                                                    filterableFields = "b0";
+                                                } else {
+                                                    throw new IllegalStateException(
+                                                            "Unknown test table name: "
+                                                                    + sourceTestStep.name);
+                                                }
+                                                Map<String, String> oldOptions =
+                                                        new HashMap<>(sourceTestStep.options);
+                                                oldOptions.put(
+                                                        "filterable-fields", filterableFields);
+                                                return sourceTestStep.withNewOptions(oldOptions);
+                                            })
+                                    .collect(Collectors.toList()))
+                    .setupTableSinks(DELTA_JOIN_WITH_CALC_ON_SOURCE.getSetupSinkTestSteps())
+                    .runSql(DELTA_JOIN_WITH_CALC_ON_SOURCE.getRunSqlTestStep().sql)
+                    .build();
+
     public static final TableTestProgram DELTA_JOIN_WITH_CACHE =
             TableTestProgram.of("delta-join-with-cache", "validates delta join with cache")
                     .setupConfig(
@@ -180,6 +245,19 @@ public class DeltaJoinTestPrograms {
                     .runSql(DELTA_JOIN_WITH_NON_EQUIV_CONDITION.getRunSqlTestStep().sql)
                     .build();
 
+    public static final TableTestProgram DELTA_JOIN_WITH_CACHE_AND_CALC_ON_SOURCE =
+            TableTestProgram.of(
+                            "delta-join-with-cache-and-calc-on-source",
+                            "validates delta join with cache and calc on source")
+                    .setupConfig(
+                            OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY,
+                            OptimizerConfigOptions.DeltaJoinStrategy.FORCE)
+                    .setupConfig(ExecutionConfigOptions.TABLE_EXEC_DELTA_JOIN_CACHE_ENABLED, true)
+                    .setupTableSources(DELTA_JOIN_WITH_CALC_ON_SOURCE.getSetupSourceTestSteps())
+                    .setupTableSinks(DELTA_JOIN_WITH_CALC_ON_SOURCE.getSetupSinkTestSteps())
+                    .runSql(DELTA_JOIN_WITH_CALC_ON_SOURCE.getRunSqlTestStep().sql)
+                    .build();
+
     public static final TableTestProgram DELTA_JOIN_WITH_CDC_SOURCE_WITHOUT_DELETE =
             TableTestProgram.of(
                             "delta-join-with-cdc-source-without-delete",
@@ -259,6 +337,44 @@ public class DeltaJoinTestPrograms {
                                     + "on a1 = b1")
                     .build();
 
+    public static final TableTestProgram DELTA_JOIN_WITH_CALC_ON_CDC_SOURCE_WITHOUT_DELETE =
+            TableTestProgram.of(
+                            "delta-join-with-calc-on-cdc-source-without-delete",
+                            "validates delta join with calc on cdc source without delete ")
+                    .setupConfig(
+                            OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY,
+                            OptimizerConfigOptions.DeltaJoinStrategy.FORCE)
+                    .setupConfig(ExecutionConfigOptions.TABLE_EXEC_DELTA_JOIN_CACHE_ENABLED, true)
+                    .setupTableSources(
+                            DELTA_JOIN_WITH_CDC_SOURCE_WITHOUT_DELETE.getSetupSourceTestSteps())
+                    .setupTableSink(
+                            SinkTestStep.newBuilder("snk")
+                                    .addSchema(
+                                            addPk2Schema(
+                                                    SINK_TABLE_BASE_SCHEMA, "a0", "b0", "a1", "b1"))
+                                    .addOptions(TABLE_BASE_OPTIONS)
+                                    .testMaterializedData()
+                                    // deduplicate data by pk
+                                    .deduplicatedFieldIndices(new int[] {0, 1, 3, 5})
+                                    .consumedBeforeRestore(
+                                            Row.of(1, 1.0, "left-pk1-2-s", 2.0, "right-pk1-1-s", 1),
+                                            Row.of(1, 2.0, "left-pk2-1-s", 2.0, "right-pk1-1-s", 1))
+                                    .consumedAfterRestore(
+                                            Row.of(1, 1.0, "left-pk1-2-s", 2.0, "right-pk1-2-s", 1),
+                                            Row.of(1, 2.0, "left-pk2-2-s", 2.0, "right-pk1-2-s", 1))
+                                    .build())
+                    .runSql(
+                            "insert into snk "
+                                    + "select a1, a0, new_a2, b0, new_b2, b1 from ( "
+                                    + "   select a0, a1, a2, concat(a2, '-s') as new_a2 from leftSrc "
+                                    + "       where a0 = cast(1.0 as double) or a0 = cast(2.0 as double) "
+                                    + ") join ("
+                                    + "   select b0, b1, concat(b2, '-s') as new_b2, b2 from rightSrc "
+                                    + "       where b0 = cast(2.0 as double) or b0 = cast(3.0 as double) "
+                                    + ") "
+                                    + "on a1 = b1")
+                    .build();
+
     public static final TableTestProgram DELTA_JOIN_WITH_CACHE_AND_CDC_SOURCE_WITHOUT_DELETE =
             TableTestProgram.of(
                             "delta-join-with-cache-and-cdc-source-without-delete",
@@ -274,6 +390,29 @@ public class DeltaJoinTestPrograms {
                     .runSql(DELTA_JOIN_WITH_CDC_SOURCE_WITHOUT_DELETE.getRunSqlTestStep().sql)
                     .build();
 
+    public static final TableTestProgram
+            DELTA_JOIN_WITH_CACHE_AND_CALC_ON_CDC_SOURCE_WITHOUT_DELETE =
+                    TableTestProgram.of(
+                                    "delta-join-with-cache-and-calc-on-cdc-source-without-delete",
+                                    "validates delta join with cache and calc on cdc source without delete")
+                            .setupConfig(
+                                    OptimizerConfigOptions.TABLE_OPTIMIZER_DELTA_JOIN_STRATEGY,
+                                    OptimizerConfigOptions.DeltaJoinStrategy.FORCE)
+                            .setupConfig(
+                                    ExecutionConfigOptions.TABLE_EXEC_DELTA_JOIN_CACHE_ENABLED,
+                                    true)
+                            .setupTableSources(
+                                    DELTA_JOIN_WITH_CALC_ON_CDC_SOURCE_WITHOUT_DELETE
+                                            .getSetupSourceTestSteps())
+                            .setupTableSinks(
+                                    DELTA_JOIN_WITH_CALC_ON_CDC_SOURCE_WITHOUT_DELETE
+                                            .getSetupSinkTestSteps())
+                            .runSql(
+                                    DELTA_JOIN_WITH_CALC_ON_CDC_SOURCE_WITHOUT_DELETE
+                                            .getRunSqlTestStep()
+                                            .sql)
+                            .build();
+
     private static String[] addPk2Schema(String[] originalSchema, String... pkCols) {
         return Stream.concat(
                         Arrays.stream(originalSchema),

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/testutils/RestoreTestBase.java

@@ -326,7 +326,6 @@ public void generateTestSetupFiles(TableTestProgram program) throws Exception {
         Files.move(savepointPath, savepointDirPath, StandardCopyOption.ATOMIC_MOVE);
     }
 
-    //    @Disabled
     @ParameterizedTest
     @MethodSource("createSpecs")
     @Order(1)