featzhang commented on code in PR #27847:
URL: https://github.com/apache/flink/pull/27847#discussion_r3008245904
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java:
##########
@@ -109,13 +109,22 @@ private static int[] buildNonPartitionIndices(
return IntStream.range(0, fieldCount).filter(i ->
!partitionKeySet.contains(i)).toArray();
}
+ /**
+ * Builds a RowKind-to-output-code map. Keys may be comma-separated (e.g.,
"INSERT,
+ * UPDATE_AFTER") to map multiple RowKinds to the same output code.
+ */
private static Map<RowKind, String> buildOpMap(@Nullable final Map<String,
String> opMapping) {
if (opMapping == null) {
return new EnumMap<>(DEFAULT_OP_MAPPING);
}
- final Map<RowKind, String> map = new EnumMap<>(RowKind.class);
- opMapping.forEach((name, code) -> map.put(RowKind.valueOf(name),
code));
- return map;
+ final Map<RowKind, String> result = new EnumMap<>(RowKind.class);
Review Comment:
Recommend adding logging for conflicts or invalid mappings to help users
debug configuration issues.
##########
flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java:
##########
@@ -140,21 +141,55 @@ private static Optional<List<DataType>> validateInputs(
final Optional<Map> opMapping = callContext.getArgumentValue(2,
Map.class);
if (opMapping.isPresent()) {
- final boolean hasInvalidMappingKey =
- opMapping.get().keySet().stream()
- .anyMatch(
- key ->
- !(key instanceof String)
- ||
!VALID_ROW_KIND_NAMES.contains(key));
- if (hasInvalidMappingKey) {
- return callContext.fail(
- throwOnFailure, "Invalid target mapping for argument
'op_mapping'.");
+ final Optional<List<DataType>> validationError =
+ validateOpMappingKeys(callContext, opMapping.get(),
throwOnFailure);
+ if (validationError.isPresent()) {
+ return validationError;
}
}
return Optional.of(callContext.getArgumentDataTypes());
}
+ /**
+ * Validates op_mapping keys. Keys may be comma-separated RowKind names
(e.g., "INSERT,
+ * UPDATE_AFTER") to map multiple RowKinds to the same output code. Each
individual RowKind must
+ * be valid and appear at most once across all entries.
+ */
+ @SuppressWarnings("rawtypes")
+ private static Optional<List<DataType>> validateOpMappingKeys(
Review Comment:
Recommend adding comments about whitespace and case handling for better user
understanding.
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogTestPrograms.java:
##########
@@ -353,13 +393,26 @@ public class ToChangelogTestPrograms {
public static final TableTestProgram INVALID_OP_MAPPING =
TableTestProgram.of(
"to-changelog-invalid-op-mapping",
- "fails when op_mapping has invalid RowKind name")
+ "fails when op_mapping has invalid change
operation name")
Review Comment:
Recommend adding notes on why UPDATE_BEFORE is dropped and practical use
cases of deletion flag pattern.
##########
flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java:
##########
@@ -159,6 +159,12 @@ public interface PartitionedTable {
* descriptor("op_code").asArgument("op"),
* map("INSERT", "I", "UPDATE_AFTER", "U").asArgument("op_mapping")
* );
+ *
+ * // Deletion flag pattern: comma-separated keys map multiple change
operations to the same code
Review Comment:
uggest enhancing Javadoc to clarify how multiple RowKinds mapping to the
same output may affect UPDATE_BEFORE behavior.
--
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]
