[FLINK-40557][table] Make operation column optional in TO_CHANGELOG…

docs/content/docs/sql/reference/queries/changelog.md

@@ -286,7 +286,7 @@ Table result = cdcStream
 
 ## TO_CHANGELOG
 
-The `TO_CHANGELOG` PTF converts a dynamic table (i.e. an updating table) into an append-only table with an explicit operation code column. Each input row - regardless of its original change operation (INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE) - is emitted as an INSERT-only row with a string column indicating the original operation.
+The `TO_CHANGELOG` PTF converts a dynamic table (i.e. an updating table) into an append-only table. By default, each input row - regardless of its original change operation (INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE) - is emitted as an INSERT-only row with a string column indicating the original operation. Set `include_op_column` to `false` to omit that column.
 
 This is useful when you need to materialize changelog events into a downstream system that only supports appends (e.g., a message queue, log store, or append-only file sink). It is also useful to filter out certain types of updates, for example DELETEs.
 
@@ -297,7 +297,8 @@ SELECT * FROM TO_CHANGELOG(
   input => TABLE source_table [PARTITION BY key_col],
   [op => DESCRIPTOR(op_column_name),]
   [op_mapping => MAP['INSERT', 'I', 'DELETE', 'D', ...],]
-  [produces_full_deletes => BOOLEAN]
+  [produces_full_deletes => BOOLEAN,]
+  [include_op_column => BOOLEAN]
 )
 ```
 
@@ -309,6 +310,7 @@ SELECT * FROM TO_CHANGELOG(
 | `op`                    | No       | A `DESCRIPTOR` with a single column name for the operation code column. Defaults to `op`.                                                                                                                                                                                                                                                                |
 | `op_mapping`            | No       | A `MAP<STRING, STRING>` mapping change operation names to custom output codes. Keys can contain comma-separated names to map multiple operations to the same code (e.g., `'INSERT, UPDATE_AFTER'`). When provided, only mapped operations are forwarded - unmapped events are dropped. Each change operation may appear at most once across all entries. |
 | `produces_full_deletes` | No       | A `BOOLEAN` literal that controls how DELETE rows are emitted. When `true` (default), DELETE rows carry all columns, the full image. When `false`, only the identifying key columns are preserved and the rest are nulled. See [Full vs partial deletes](#full-vs-partial-deletes) for more details.                                                     |
+| `include_op_column`     | No       | A `BOOLEAN` literal that controls whether the operation code column is included in the output. When `true` (default), the operation code column is prepended to the output. When `false`, the output schema contains only the input columns.                                                                                                             |
 
 #### Default op_mapping
 
@@ -323,12 +325,14 @@ When `op_mapping` is omitted, all four change operations are mapped to their sta
 
 ### Output Schema
 
-The output schema is:
+By default, the output schema is:
 
 ```
 [op_column, all_input_columns]
 ```
 
+With `include_op_column => false`, the output schema contains only the input columns.
+
 All output rows have `INSERT` - the table is always append-only.
 
 ### Examples

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java

@@ -171,13 +171,12 @@ public interface PartitionedTable {
     Table process(Class<? extends UserDefinedFunction> function, Object... arguments);
 
     /**
-     * Converts this partitioned dynamic table into an append-only table with an explicit operation
-     * code column using the built-in {@code TO_CHANGELOG} process table function with set
-     * semantics.
+     * Converts this partitioned dynamic table into an append-only table using the built-in {@code
+     * TO_CHANGELOG} process table function with set semantics.
      *
      * <p>Each input row - regardless of its original change operation - is emitted as an
-     * INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
-     * UPDATE_AFTER, DELETE, etc.). With set semantics, rows for the same partition key are
+     * INSERT-only row. By default, a string {@code "op"} column indicates the original operation
+     * (INSERT, UPDATE_AFTER, DELETE, etc.). With set semantics, rows for the same partition key are
      * co-located in the same parallel operator instance.
      *
      * <p>For row semantics (each row processed independently), use {@link Table#toChangelog} on the
@@ -211,12 +210,18 @@ public interface PartitionedTable {
      * Table result = table
      *     .partitionBy($("id"))
      *     .toChangelog(lit(false).asArgument("produces_full_deletes"));
+     *
+     * // Omit the operation code column from the output schema.
+     * Table result = table
+     *     .partitionBy($("id"))
+     *     .toChangelog(lit(false).asArgument("include_op_column"));
      * }</pre>
      *
      * @param arguments optional named arguments for {@code op}, {@code op_mapping}, and {@code
-     *     produces_full_deletes}
-     * @return an append-only {@link Table} with output schema {@code [partition_keys, op,
-     *     non_partition_input_columns]}
+     *     produces_full_deletes}, and {@code include_op_column}
+     * @return an append-only {@link Table} with output schema {@code [partition_keys,
+     *     non_partition_input_columns]}, optionally with {@code op} before the non-partition input
+     *     columns
      * @see Table#toChangelog(Expression...)
      */
     Table toChangelog(Expression... arguments);

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java

@@ -1424,12 +1424,12 @@ default TableResult executeInsert(
     Table process(Class<? extends UserDefinedFunction> function, Object... arguments);
 
     /**
-     * Converts this dynamic table into an append-only table with an explicit operation code column
-     * using the built-in {@code TO_CHANGELOG} process table function.
+     * Converts this dynamic table into an append-only table using the built-in {@code TO_CHANGELOG}
+     * process table function.
      *
      * <p>Each input row - regardless of its original change operation - is emitted as an
-     * INSERT-only row with a string {@code "op"} column indicating the original operation (INSERT,
-     * UPDATE_AFTER, DELETE, etc.).
+     * INSERT-only row. By default, a string {@code "op"} column indicates the original operation
+     * (INSERT, UPDATE_AFTER, DELETE, etc.).
      *
      * <p>By default, the input is processed with row semantics (each row independently). To
      * co-locate rows with the same key in the same parallel operator instance, partition the input
@@ -1466,11 +1466,17 @@ default TableResult executeInsert(
      * Table result = table.toChangelog(
      *     lit(false).asArgument("produces_full_deletes")
      * );
+     *
+     * // Omit the operation code column from the output schema.
+     * Table result = table.toChangelog(
+     *     lit(false).asArgument("include_op_column")
+     * );
      * }</pre>
      *
-     * @param arguments optional named arguments for {@code op}, {@code op_mapping}, and {@code
-     *     produces_full_deletes}
-     * @return an append-only {@link Table} with an {@code op} column prepended to the input columns
+     * @param arguments optional named arguments for {@code op}, {@code op_mapping}, {@code
+     *     produces_full_deletes}, and {@code include_op_column}
+     * @return an append-only {@link Table}, optionally with an {@code op} column prepended to the
+     *     input columns
      */
     Table toChangelog(Expression... arguments);
 

flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java

@@ -888,7 +888,8 @@ ANY, and(logical(LogicalTypeRoot.BOOLEAN), LITERAL)
                                     DataTypes.MAP(DataTypes.STRING(), DataTypes.STRING()),
                                     true),
                             StaticArgument.scalar(
-                                    "produces_full_deletes", DataTypes.BOOLEAN(), true))
+                                    "produces_full_deletes", DataTypes.BOOLEAN(), true),
+                            StaticArgument.scalar("include_op_column", DataTypes.BOOLEAN(), true))
                     .inputTypeStrategy(TO_CHANGELOG_INPUT_TYPE_STRATEGY)
                     .outputTypeStrategy(TO_CHANGELOG_OUTPUT_TYPE_STRATEGY)
                     .runtimeClass(

flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java

@@ -50,6 +50,7 @@ public final class ToChangelogTypeStrategy {
     public static final int ARG_OP = 1;
     public static final int ARG_OP_MAPPING = 2;
     public static final int ARG_PRODUCES_FULL_DELETES = 3;
+    public static final int ARG_INCLUDE_OP_COLUMN = 4;
 
     private static final Set<String> VALID_ROW_KIND_NAMES =
             Set.of("INSERT", "UPDATE_BEFORE", "UPDATE_AFTER", "DELETE");
@@ -83,15 +84,16 @@ public Optional<List<DataType>> inferInputTypes(
                                                 new ValidationException(
                                                         "First argument must be a table for TO_CHANGELOG."));
 
-                final String opColumnName =
-                        ChangelogTypeStrategyUtils.resolveOpColumnName(callContext);
                 final boolean producesFullDeletes =
                         callContext
                                 .getArgumentValue(ARG_PRODUCES_FULL_DELETES, Boolean.class)
                                 .orElse(true);
 
+                final boolean includeOpColumn = shouldIncludeOpColumn(callContext);
+
                 final List<Field> outputFields =
-                        buildOutputFields(semantics, opColumnName, producesFullDeletes);
+                        buildOutputFields(
+                                semantics, producesFullDeletes, includeOpColumn, callContext);
 
                 return Optional.of(DataTypes.ROW(outputFields).notNull());
             };
@@ -100,6 +102,17 @@ public Optional<List<DataType>> inferInputTypes(
     // Helpers
     // --------------------------------------------------------------------------------------------
 
+    /**
+     * Returns whether {@code TO_CHANGELOG} should include the operation column in its output.
+     *
+     * <p>Compiled plans created before this argument was introduced have only four arguments and
+     * retain the default value of {@code true}.
+     */
+    public static boolean shouldIncludeOpColumn(final CallContext callContext) {
+        return callContext.getArgumentDataTypes().size() <= ARG_INCLUDE_OP_COLUMN
+                || callContext.getArgumentValue(ARG_INCLUDE_OP_COLUMN, Boolean.class).orElse(true);
+    }
+
     private static Optional<List<DataType>> validateInputs(
             final CallContext callContext, final boolean throwOnFailure) {
         Optional<List<DataType>> error;
@@ -227,12 +240,16 @@ private static boolean mapsDelete(final Map<String, String> opMapping) {
      */
     private static List<Field> buildOutputFields(
             final TableSemantics semantics,
-            final String opColumnName,
-            final boolean producesFullDeletes) {
+            final boolean producesFullDeletes,
+            final boolean includeOpColumn,
+            final CallContext callContext) {
         final List<Field> inputFields = DataType.getFields(semantics.dataType());
         final int[] outputIndices = ChangelogTypeStrategyUtils.computeOutputIndices(semantics);
         final List<Field> outputFields = new ArrayList<>();
-        outputFields.add(DataTypes.FIELD(opColumnName, DataTypes.STRING()));
+        if (includeOpColumn) {
+            final String opColumnName = ChangelogTypeStrategyUtils.resolveOpColumnName(callContext);
+            outputFields.add(DataTypes.FIELD(opColumnName, DataTypes.STRING()));
+        }
         final Set<Integer> preserved =
                 producesFullDeletes
                         ? Collections.emptySet()

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

@@ -28,6 +28,7 @@
 import java.util.stream.Stream;
 
 import static org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies.TO_CHANGELOG_OUTPUT_TYPE_STRATEGY;
+import static org.apache.flink.table.types.inference.strategies.ToChangelogTypeStrategy.ARG_INCLUDE_OP_COLUMN;
 import static org.apache.flink.table.types.inference.strategies.ToChangelogTypeStrategy.ARG_OP;
 import static org.apache.flink.table.types.inference.strategies.ToChangelogTypeStrategy.ARG_OP_MAPPING;
 import static org.apache.flink.table.types.inference.strategies.ToChangelogTypeStrategy.ARG_PRODUCES_FULL_DELETES;
@@ -75,11 +76,36 @@ ARG_TABLE, new TableSemanticsMock(TABLE_TYPE_NOT_NULL_SCORE))
                                                 DataTypes.FIELD(
                                                         "score", DataTypes.BIGINT().notNull()))
                                         .notNull()),
+                TestSpec.forStrategy(
+                                "include_op_column=false omits the operation column",
+                                TO_CHANGELOG_OUTPUT_TYPE_STRATEGY)
+                        .inputTypes(
+                                TABLE_TYPE_NOT_NULL_SCORE,
+                                DESCRIPTOR_TYPE,
+                                MAP_TYPE,
+                                BOOLEAN_TYPE,
+                                BOOLEAN_TYPE)
+                        .calledWithTableSemanticsAt(
+                                ARG_TABLE, new TableSemanticsMock(TABLE_TYPE_NOT_NULL_SCORE))
+                        .calledWithLiteralAt(ARG_OP, ColumnList.of("op"))
+                        .calledWithLiteralAt(ARG_OP_MAPPING, null)
+                        .calledWithLiteralAt(ARG_INCLUDE_OP_COLUMN, false)
+                        .expectDataType(
+                                DataTypes.ROW(
+                                                DataTypes.FIELD(
+                                                        "name", DataTypes.STRING().notNull()),
+                                                DataTypes.FIELD(
+                                                        "score", DataTypes.BIGINT().notNull()))
+                                        .notNull()),
                 TestSpec.forStrategy(
                                 "produces_full_deletes=false widens non-upsert-key columns to nullable",
                                 TO_CHANGELOG_OUTPUT_TYPE_STRATEGY)
                         .inputTypes(
-                                TABLE_TYPE_NOT_NULL_SCORE, DESCRIPTOR_TYPE, MAP_TYPE, BOOLEAN_TYPE)
+                                TABLE_TYPE_NOT_NULL_SCORE,
+                                DESCRIPTOR_TYPE,
+                                MAP_TYPE,
+                                BOOLEAN_TYPE,
+                                BOOLEAN_TYPE)
                         .calledWithTableSemanticsAt(
                                 ARG_TABLE,
                                 new TableSemanticsMock(
@@ -103,7 +129,11 @@ ARG_TABLE, new TableSemanticsMock(TABLE_TYPE_NOT_NULL_SCORE))
                                 "produces_full_deletes=false without upsert key widens all columns",
                                 TO_CHANGELOG_OUTPUT_TYPE_STRATEGY)
                         .inputTypes(
-                                TABLE_TYPE_NOT_NULL_SCORE, DESCRIPTOR_TYPE, MAP_TYPE, BOOLEAN_TYPE)
+                                TABLE_TYPE_NOT_NULL_SCORE,
+                                DESCRIPTOR_TYPE,
+                                MAP_TYPE,
+                                BOOLEAN_TYPE,
+                                BOOLEAN_TYPE)
                         .calledWithTableSemanticsAt(
                                 ARG_TABLE, new TableSemanticsMock(TABLE_TYPE_NOT_NULL_SCORE))
                         .calledWithLiteralAt(ARG_OP, ColumnList.of("op"))
@@ -123,7 +153,11 @@ private static Stream<TestSpec> setSemantics() {
                                 "produces_full_deletes=true in set semantics preserves NOT NULL on non-partition columns",
                                 TO_CHANGELOG_OUTPUT_TYPE_STRATEGY)
                         .inputTypes(
-                                TABLE_TYPE_NOT_NULL_SCORE, DESCRIPTOR_TYPE, MAP_TYPE, BOOLEAN_TYPE)
+                                TABLE_TYPE_NOT_NULL_SCORE,
+                                DESCRIPTOR_TYPE,
+                                MAP_TYPE,
+                                BOOLEAN_TYPE,
+                                BOOLEAN_TYPE)
                         .calledWithTableSemanticsAt(
                                 ARG_TABLE,
                                 new TableSemanticsMock(
@@ -146,7 +180,11 @@ private static Stream<TestSpec> setSemantics() {
                                 "produces_full_deletes=false in set semantics widens non-partition-key columns",
                                 TO_CHANGELOG_OUTPUT_TYPE_STRATEGY)
                         .inputTypes(
-                                TABLE_TYPE_NOT_NULL_SCORE, DESCRIPTOR_TYPE, MAP_TYPE, BOOLEAN_TYPE)
+                                TABLE_TYPE_NOT_NULL_SCORE,
+                                DESCRIPTOR_TYPE,
+                                MAP_TYPE,
+                                BOOLEAN_TYPE,
+                                BOOLEAN_TYPE)
                         .calledWithTableSemanticsAt(
                                 ARG_TABLE,
                                 new TableSemanticsMock(

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

@@ -47,6 +47,7 @@ public List<TableTestProgram> programs() {
                 ToChangelogTestPrograms.RETRACT_PARTITION_BY,
                 ToChangelogTestPrograms.CUSTOM_OP_MAPPING,
                 ToChangelogTestPrograms.CUSTOM_OP_NAME,
+                ToChangelogTestPrograms.WITHOUT_OP_COLUMN,
                 ToChangelogTestPrograms.TABLE_API_DEFAULT,
                 ToChangelogTestPrograms.TABLE_API_RETRACT_PARTITION_BY,
                 ToChangelogTestPrograms.LAG_ON_UPSERT_VIA_CHANGELOG,

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

@@ -62,6 +62,36 @@ public class ToChangelogTestPrograms {
                     .runSql("INSERT INTO sink SELECT * FROM TO_CHANGELOG(input => TABLE t)")
                     .build();
 
+    public static final TableTestProgram WITHOUT_OP_COLUMN =
+            TableTestProgram.of(
+                            "to-changelog-without-op-column",
+                            "include_op_column=false preserves the input schema and emits insert-only rows")
+                    .setupTableSource(
+                            SourceTestStep.newBuilder("t")
+                                    .addSchema("id INT", "name STRING")
+                                    .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, "Alicia"),
+                                            Row.ofKind(RowKind.DELETE, 2, "Bob"))
+                                    .build())
+                    .setupTableSink(
+                            SinkTestStep.newBuilder("sink")
+                                    .addSchema("id INT", "name STRING")
+                                    .consumedValues(
+                                            "+I[1, Alice]",
+                                            "+I[2, Bob]",
+                                            "+I[1, Alice]",
+                                            "+I[1, Alicia]",
+                                            "+I[2, Bob]")
+                                    .build())
+                    .runSql(
+                            "INSERT INTO sink SELECT * FROM TO_CHANGELOG("
+                                    + "input => TABLE t, include_op_column => false)")
+                    .build();
+
     public static final TableTestProgram RETRACT =
             TableTestProgram.of(
                             "to-changelog-updating-input",