[FLINK-39261][table] Throw ValidationException when invalid op mapping is received

The default invalid_op_handling behavior is FAIL. During, runtime if rowKind is not present in the op_map we should throw an ValidationException

Author: raminqaf

flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/FromChangelogInputTypeStrategyTest.java

@@ -61,20 +61,6 @@ protected Stream<TestSpec> testData() {
                                         "d", "DELETE"))
                         .expectArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE, MAP_TYPE),
 
-                // Valid: retract-style mapping with UPDATE_BEFORE
-                TestSpec.forStrategy("Valid with UPDATE_BEFORE", FROM_CHANGELOG_INPUT_TYPE_STRATEGY)
-                        .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE, MAP_TYPE)
-                        .calledWithTableSemanticsAt(0, new TableSemanticsMock(TABLE_TYPE))
-                        .calledWithLiteralAt(1, ColumnList.of(List.of("op")))
-                        .calledWithLiteralAt(
-                                2,
-                                Map.of(
-                                        "c", "INSERT",
-                                        "ub", "UPDATE_BEFORE",
-                                        "ua", "UPDATE_AFTER",
-                                        "d", "DELETE"))
-                        .expectArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE, MAP_TYPE),
-
                 // Error: op column not found
                 TestSpec.forStrategy(
                                 "Op column not found in schema", FROM_CHANGELOG_INPUT_TYPE_STRATEGY)

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

@@ -41,7 +41,6 @@ public List<TableTestProgram> programs() {
         return List.of(
                 FromChangelogTestPrograms.DEFAULT_OP_MAPPING,
                 FromChangelogTestPrograms.CUSTOM_OP_MAPPING,
-                FromChangelogTestPrograms.UNMAPPED_CODES_DROPPED,
                 FromChangelogTestPrograms.CUSTOM_OP_NAME,
                 FromChangelogTestPrograms.TABLE_API_DEFAULT,
                 FromChangelogTestPrograms.ROUND_TRIP);

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

@@ -93,32 +93,6 @@ public class FromChangelogTestPrograms {
                                     + "op_mapping => MAP['c, r', 'INSERT', 'ub', 'UPDATE_BEFORE', 'ua', 'UPDATE_AFTER', 'd', 'DELETE'])")
                     .build();
 
-    public static final TableTestProgram UNMAPPED_CODES_DROPPED =
-            TableTestProgram.of(
-                            "from-changelog-unmapped-codes-dropped",
-                            "unmapped op codes are silently dropped")
-                    .setupTableSource(
-                            SourceTestStep.newBuilder("cdc_stream")
-                                    .addSchema(SIMPLE_CDC_SCHEMA)
-                                    .producedValues(
-                                            Row.of(1, "INSERT", "Alice"),
-                                            Row.of(2, "INSERT", "Bob"),
-                                            Row.of(1, "UNKNOWN", "Alice2"),
-                                            Row.of(2, "DELETE", "Bob"))
-                                    .build())
-                    .setupTableSink(
-                            SinkTestStep.newBuilder("sink")
-                                    .addSchema("id INT", "name STRING")
-                                    .consumedValues(
-                                            Row.ofKind(RowKind.INSERT, 1, "Alice"),
-                                            Row.ofKind(RowKind.INSERT, 2, "Bob"),
-                                            Row.ofKind(RowKind.DELETE, 2, "Bob"))
-                                    .build())
-                    .runSql(
-                            "INSERT INTO sink SELECT * FROM FROM_CHANGELOG("
-                                    + "input => TABLE cdc_stream)")
-                    .build();
-
     /** Custom op column name via DESCRIPTOR. */
     public static final TableTestProgram CUSTOM_OP_NAME =
             TableTestProgram.of(

flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/FromChangelogFunction.java

@@ -19,6 +19,7 @@
 package org.apache.flink.table.runtime.functions.ptf;
 
 import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.data.MapData;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.data.StringData;
@@ -137,7 +138,10 @@ public void eval(
         final StringData opCode = input.getString(opColumnIndex);
         final RowKind rowKind = opMap.get(opCode);
         if (rowKind == null) {
-            return;
+            throw new ValidationException(
+                    String.format(
+                            "Received invalid op code '%s'. Defined op codes are: %s. Failing as configured.",
+                            opCode, opMap.keySet()));
         }
 
         projectedOutput.replaceRow(input);