raminqaf commented on code in PR #29102:
URL: https://github.com/apache/flink/pull/29102#discussion_r3990478018
##########
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,
Review Comment:
Could you please add a `WITHOUT_OP_COLUMN_RESTORE` restore test too
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -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);
+ }
Review Comment:
If `include_op_column => false` we ignore the `op` column but I am still
capable of using `op_mapping`. correct? We should maybe just explain what is
the use case of that.
Or if we should also ignore the `op_mapping` we should do this:
```suggestion
if (this.includeOpColumn) {
final StringData opCode = opMap.get(input.getRowKind());
if (opCode == null) {
return;
}
opRow.setField(0, opCode);
}
```
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java:
##########
@@ -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()));
+ }
Review Comment:
If `include_op_column => false` we ignore the `op` column but I am still
capable of using `op_mapping`. correct? We should maybe just explain what is
the use case of that.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -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);
}
+
Review Comment:
remove
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -192,24 +195,41 @@ private static void validateProducesPartialDeletes(
}
}
+ /** Maintains compatibility with compiled plans created before {@code
include_op_column}. */
Review Comment:
Not such a helpful javaDocs
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -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);
Review Comment:
Then you can make `includeOpColumn` a local field here and lower in the
function just check if this is `null`. (Alternative you can use `Optional`)
```suggestion
opRow = includeOpColumn ? new GenericRowData(1) : null;
```
--
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]
