diff --git a/docs/content/docs/sql/reference/queries/changelog.md b/docs/content/docs/sql/reference/queries/changelog.md index 71539a9b8cb03..1bcb021526fe0 100644 --- a/docs/content/docs/sql/reference/queries/changelog.md +++ b/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` 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 diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java index ecf58fbe9b391..eb0d21a92ae7c 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java +++ b/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 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. * *

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. * *

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")); * } * * @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); diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java index 9d8183d7e4d4e..2e58f917a1f04 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java +++ b/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 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. * *

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.). * *

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") + * ); * } * - * @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); diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java index fdfa00390d6f4..d8573d321038a 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java +++ b/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( diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java index a2b8e13681b08..06b953ce17e1a 100644 --- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java +++ b/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 VALID_ROW_KIND_NAMES = Set.of("INSERT", "UPDATE_BEFORE", "UPDATE_AFTER", "DELETE"); @@ -83,15 +84,16 @@ public Optional> 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 outputFields = - buildOutputFields(semantics, opColumnName, producesFullDeletes); + buildOutputFields( + semantics, producesFullDeletes, includeOpColumn, callContext); return Optional.of(DataTypes.ROW(outputFields).notNull()); }; @@ -100,6 +102,17 @@ public Optional> inferInputTypes( // Helpers // -------------------------------------------------------------------------------------------- + /** + * Returns whether {@code TO_CHANGELOG} should include the operation column in its output. + * + *

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> validateInputs( final CallContext callContext, final boolean throwOnFailure) { Optional> error; @@ -227,12 +240,16 @@ private static boolean mapsDelete(final Map opMapping) { */ private static List buildOutputFields( final TableSemantics semantics, - final String opColumnName, - final boolean producesFullDeletes) { + final boolean producesFullDeletes, + final boolean includeOpColumn, + final CallContext callContext) { final List inputFields = DataType.getFields(semantics.dataType()); final int[] outputIndices = ChangelogTypeStrategyUtils.computeOutputIndices(semantics); final List 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 preserved = producesFullDeletes ? Collections.emptySet() diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ToChangelogOutputTypeStrategyTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ToChangelogOutputTypeStrategyTest.java index 8c06c26d7bb75..1d94916f3f3b8 100644 --- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ToChangelogOutputTypeStrategyTest.java +++ b/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 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 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( diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogSemanticTests.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogSemanticTests.java index c35394c7694ee..63e1e11f021a2 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogSemanticTests.java +++ b/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 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, diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogTestPrograms.java index a14b259b068b0..73d0f5d1a0fa4 100644 --- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogTestPrograms.java +++ b/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", diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.xml index 174addbf41890..552c319ecbad9 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.xml @@ -23,14 +23,14 @@ limitations under the License. @@ -42,14 +42,14 @@ ProcessTableFunction(invocation=[TO_CHANGELOG(TABLE(#0), DEFAULT(), DEFAULT(), D @@ -62,14 +62,14 @@ ProcessTableFunction(invocation=[TO_CHANGELOG(TABLE(#0) PARTITION BY($0), DEFAUL @@ -81,14 +81,14 @@ ProcessTableFunction(invocation=[TO_CHANGELOG(TABLE(#0), DEFAULT(), DEFAULT(), D @@ -101,14 +101,14 @@ ProcessTableFunction(invocation=[TO_CHANGELOG(TABLE(#0) PARTITION BY($0), DESCRI @@ -121,14 +121,14 @@ ProcessTableFunction(invocation=[TO_CHANGELOG(TABLE(#0) PARTITION BY($0), DESCRI Converts each input row into an INSERT-only output row with an operation code column. Output - * schema is {@code [op_column, ...projected_input_columns...]}. Partition columns are prepended by - * the framework outside this function and are not part of the projection. + *

Converts each input row into an INSERT-only output row. When requested, an operation code + * column is prepended to the projected input columns. Partition columns are prepended by the + * framework outside this function and are not part of the projection. * *

Uses {@link JoinedRowData} to combine the op column with the full input row. */ @@ -75,6 +76,7 @@ public class ToChangelogFunction extends BuiltInProcessTableFunction { private final int[] outputIndices; private final RowType inputRowType; private final boolean producesFullDelete; + private final boolean includeOpColumn; private final boolean[] upsertKeyColumn; private transient Map opMap; @@ -94,9 +96,10 @@ public ToChangelogFunction(final SpecializedContext context) { final Map opMapping = callContext.getArgumentValue(ARG_OP_MAPPING, Map.class).orElse(null); this.rawOpMap = buildOpMap(opMapping); - this.producesFullDelete = callContext.getArgumentValue(ARG_PRODUCES_FULL_DELETES, Boolean.class).orElse(true); + this.includeOpColumn = shouldIncludeOpColumn(callContext); + final boolean isExplicit = !callContext.isArgumentNull(ARG_PRODUCES_FULL_DELETES); validateProducesPartialDeletes(producesFullDelete, isExplicit, tableSemantics); @@ -126,7 +129,7 @@ public void open(final FunctionContext context) throws Exception { super.open(context); opMap = new EnumMap<>(RowKind.class); rawOpMap.forEach((kind, code) -> opMap.put(kind, StringData.fromString(code))); - opRow = new GenericRowData(1); + opRow = new GenericRowData(includeOpColumn ? 1 : 0); output = new JoinedRowData(); projectedOutput = ProjectedRowData.from(outputIndices); partialDeletePayload = new GenericRowData(outputIndices.length); @@ -192,24 +195,41 @@ private static void validateProducesPartialDeletes( } } + /** Maintains compatibility with compiled plans created before {@code include_op_column}. */ public void eval( final Context ctx, final RowData input, @Nullable final ColumnList op, @Nullable final MapData opMapping, @Nullable final Boolean producesFullDeletes) { + + eval(ctx, input, op, opMapping, producesFullDeletes, null); + } + + public void eval( + final Context ctx, + final RowData input, + @Nullable final ColumnList op, + @Nullable final MapData opMapping, + @Nullable final Boolean producesFullDeletes, + @Nullable final Boolean includeOpColumn) { + final StringData opCode = opMap.get(input.getRowKind()); if (opCode == null) { return; } - opRow.setField(0, opCode); + if (this.includeOpColumn) { + opRow.setField(0, opCode); + } + final RowData payload; if (input.getRowKind() == RowKind.DELETE && !producesFullDelete) { payload = buildPartialDeletePayload(input); } else { payload = projectedOutput.replaceRow(input); } + collect(output.replace(opRow, payload)); }