Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ferenc-csaky wants to merge 1 commit into
base: master
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions docs/content/docs/sql/reference/queries/changelog.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand All @@ -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]
)
```

Expand All @@ -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

Expand All @@ -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
Expand Down
21 changes: 13 additions & 8 deletions ...table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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);
Expand Down
20 changes: 13 additions & 7 deletions flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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);

Expand Down
3 changes: 2 additions & 1 deletion ...ble-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
29 changes: 23 additions & 6 deletions .../main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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");
Expand Down Expand Up @@ -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());
};
Expand All @@ -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;
Expand Down Expand Up @@ -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()
Expand Down
46 changes: 42 additions & 4 deletions .../org/apache/flink/table/types/inference/strategies/ToChangelogOutputTypeStrategyTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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(
Expand All @@ -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"))
Expand All @@ -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(
Expand All @@ -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(
Expand Down
1 change: 1 addition & 0 deletions .../java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogSemanticTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
30 changes: 30 additions & 0 deletions ...t/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogTestPrograms.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
Loading