mateczagany commented on code in PR #29102:
URL: https://github.com/apache/flink/pull/29102#discussion_r3950854571
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogTestPrograms.java:
##########
@@ -62,6 +62,28 @@ 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")
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema("id INT", "name STRING")
+ .addMode(ChangelogMode.insertOnly())
+ .producedValues(
+ Row.ofKind(RowKind.INSERT, 1,
"Alice"),
+ Row.ofKind(RowKind.INSERT, 2,
"Bob"))
Review Comment:
`UPDATE_BEFORE` / `UPDATE_AFTER` / `DELETE` kind would fail this test
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -192,25 +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);
final RowData payload;
if (input.getRowKind() == RowKind.DELETE && !producesFullDelete) {
payload = buildPartialDeletePayload(input);
} else {
payload = projectedOutput.replaceRow(input);
}
- collect(output.replace(opRow, payload));
+
+ if (this.includeOpColumn) {
+ opRow.setField(0, opCode);
+ collect(output.replace(opRow, payload));
+ } else {
+ collect(payload);
Review Comment:
Updates and full deletes fail the insert-only collector when
`include_op_column` is `false`. Could we always emit through the existing
`JoinedRowData` wrapper, using an empty `opRow` when the column is omitted?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
