[FLINK-39261][table] Add more semantic tests

Author: raminqaf

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

@@ -42,7 +42,11 @@ public List<TableTestProgram> programs() {
                 FromChangelogTestPrograms.DEFAULT_OP_MAPPING,
                 FromChangelogTestPrograms.DEBEZIUM_MAPPING,
                 FromChangelogTestPrograms.UNMAPPED_CODES_DROPPED,
+                FromChangelogTestPrograms.CUSTOM_OP_NAME,
                 FromChangelogTestPrograms.TABLE_API_DEFAULT,
-                FromChangelogTestPrograms.MISSING_PARTITION_BY);
+                FromChangelogTestPrograms.MISSING_PARTITION_BY
+                // TODO: enable after TO_CHANGELOG switches to row semantics (PR #27911)
+                // FromChangelogTestPrograms.ROUND_TRIP
+                );
     }
 }

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

@@ -19,6 +19,7 @@
 package org.apache.flink.table.planner.plan.nodes.exec.stream;
 
 import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.connector.ChangelogMode;
 import org.apache.flink.table.test.program.SinkTestStep;
 import org.apache.flink.table.test.program.SourceTestStep;
 import org.apache.flink.table.test.program.TableTestProgram;
@@ -133,6 +134,35 @@ public class FromChangelogTestPrograms {
                                     + "input => TABLE cdc_stream PARTITION BY id)")
                     .build();
 
+    /** Custom op column name via DESCRIPTOR. */
+    public static final TableTestProgram CUSTOM_OP_NAME =
+            TableTestProgram.of(
+                            "from-changelog-custom-op-name", "custom op column name via DESCRIPTOR")
+                    .setupTableSource(
+                            SourceTestStep.newBuilder("cdc_stream")
+                                    .addSchema("id INT", "operation STRING", "name STRING")
+                                    .producedValues(
+                                            Row.of(1, "INSERT", "Alice"),
+                                            Row.of(1, "UPDATE_BEFORE", "Alice"),
+                                            Row.of(1, "UPDATE_AFTER", "Alice2"))
+                                    .build())
+                    .setupTableSink(
+                            SinkTestStep.newBuilder("sink")
+                                    .addSchema(
+                                            "id INT NOT NULL",
+                                            "name STRING",
+                                            "PRIMARY KEY (id) NOT ENFORCED")
+                                    .consumedValues(
+                                            Row.ofKind(RowKind.INSERT, 1, "Alice"),
+                                            Row.ofKind(RowKind.UPDATE_BEFORE, 1, "Alice"),
+                                            Row.ofKind(RowKind.UPDATE_AFTER, 1, "Alice2"))
+                                    .build())
+                    .runSql(
+                            "INSERT INTO sink SELECT * FROM FROM_CHANGELOG("
+                                    + "input => TABLE cdc_stream PARTITION BY id, "
+                                    + "op => DESCRIPTOR(operation))")
+                    .build();
+
     // --------------------------------------------------------------------------------------------
     // Table API test
     // --------------------------------------------------------------------------------------------
@@ -163,6 +193,53 @@ public class FromChangelogTestPrograms {
                             "sink")
                     .build();
 
+    // --------------------------------------------------------------------------------------------
+    // Round-trip test: FROM_CHANGELOG(TO_CHANGELOG(table))
+    // --------------------------------------------------------------------------------------------
+
+    /**
+     * Verifies that FROM_CHANGELOG(TO_CHANGELOG(table)) recovers the original dynamic table.
+     * Requires TO_CHANGELOG with row semantics (PR #27911).
+     */
+    public static final TableTestProgram ROUND_TRIP =
+            TableTestProgram.of(
+                            "from-changelog-round-trip",
+                            "FROM_CHANGELOG(TO_CHANGELOG(table)) recovers original table")
+                    .setupTableSource(
+                            SourceTestStep.newBuilder("orders")
+                                    .addSchema(
+                                            "id INT NOT NULL",
+                                            "name STRING",
+                                            "PRIMARY KEY (id) NOT ENFORCED")
+                                    .addMode(ChangelogMode.all())
+                                    .producedValues(
+                                            Row.ofKind(RowKind.INSERT, 1, "Alice"),
+                                            Row.ofKind(RowKind.INSERT, 2, "Bob"),
+                                            Row.ofKind(RowKind.UPDATE_BEFORE, 1, "Alice"),
+                                            Row.ofKind(RowKind.UPDATE_AFTER, 1, "Alice2"),
+                                            Row.ofKind(RowKind.DELETE, 2, "Bob"))
+                                    .build())
+                    .setupTableSink(
+                            SinkTestStep.newBuilder("sink")
+                                    .addSchema(
+                                            "id INT NOT NULL",
+                                            "name STRING",
+                                            "PRIMARY KEY (id) NOT ENFORCED")
+                                    .consumedValues(
+                                            Row.ofKind(RowKind.INSERT, 1, "Alice"),
+                                            Row.ofKind(RowKind.INSERT, 2, "Bob"),
+                                            Row.ofKind(RowKind.UPDATE_BEFORE, 1, "Alice"),
+                                            Row.ofKind(RowKind.UPDATE_AFTER, 1, "Alice2"),
+                                            Row.ofKind(RowKind.DELETE, 2, "Bob"))
+                                    .build())
+                    .setupSql(
+                            "CREATE VIEW changelog_view AS "
+                                    + "SELECT * FROM TO_CHANGELOG(input => TABLE orders)")
+                    .runSql(
+                            "INSERT INTO sink SELECT * FROM FROM_CHANGELOG("
+                                    + "input => TABLE changelog_view PARTITION BY id)")
+                    .build();
+
     // --------------------------------------------------------------------------------------------
     // Error validation tests
     // --------------------------------------------------------------------------------------------