twalthr commented on code in PR #27847:
URL: https://github.com/apache/flink/pull/27847#discussion_r3008024233
##########
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(
+ final CallContext callContext, final Map opMapping, final boolean
throwOnFailure) {
+ final Set<String> allRowKindsSeen = new HashSet<>();
+ for (final Object key : opMapping.keySet()) {
+ if (!(key instanceof String)) {
+ return callContext.fail(
+ throwOnFailure, "Invalid target mapping for argument
'op_mapping'.");
+ }
+ final String[] rowKindNames = ((String) key).split(",");
+ for (final String rawName : rowKindNames) {
+ final String rowKindName = rawName.trim();
+ if (!VALID_ROW_KIND_NAMES.contains(rowKindName)) {
+ return callContext.fail(
+ throwOnFailure,
+ String.format(
+ "Invalid target mapping for argument
'op_mapping'. "
+ + "Unknown RowKind: '%s'. Valid
values are: %s.",
Review Comment:
RowKind is a Java class. And is confusing to SQL persona. This is why it
called `op`:
```suggestion
+ "Unknown change operation:
'%s'. Valid values are: %s.",
```
##########
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(
+ final CallContext callContext, final Map opMapping, final boolean
throwOnFailure) {
+ final Set<String> allRowKindsSeen = new HashSet<>();
+ for (final Object key : opMapping.keySet()) {
+ if (!(key instanceof String)) {
+ return callContext.fail(
+ throwOnFailure, "Invalid target mapping for argument
'op_mapping'.");
+ }
+ final String[] rowKindNames = ((String) key).split(",");
+ for (final String rawName : rowKindNames) {
+ final String rowKindName = rawName.trim();
+ if (!VALID_ROW_KIND_NAMES.contains(rowKindName)) {
+ return callContext.fail(
+ throwOnFailure,
+ String.format(
+ "Invalid target mapping for argument
'op_mapping'. "
+ + "Unknown RowKind: '%s'. Valid
values are: %s.",
+ rowKindName, VALID_ROW_KIND_NAMES));
+ }
+ final boolean isDuplicate = !allRowKindsSeen.add(rowKindName);
+ if (isDuplicate) {
+ return callContext.fail(
+ throwOnFailure,
+ String.format(
+ "Invalid target mapping for argument
'op_mapping'. "
+ + "Duplicate RowKind: '%s'.",
Review Comment:
please update all other locations
##########
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(
+ final CallContext callContext, final Map opMapping, final boolean
throwOnFailure) {
+ final Set<String> allRowKindsSeen = new HashSet<>();
+ for (final Object key : opMapping.keySet()) {
+ if (!(key instanceof String)) {
+ return callContext.fail(
+ throwOnFailure, "Invalid target mapping for argument
'op_mapping'.");
+ }
+ final String[] rowKindNames = ((String) key).split(",");
+ for (final String rawName : rowKindNames) {
+ final String rowKindName = rawName.trim();
+ if (!VALID_ROW_KIND_NAMES.contains(rowKindName)) {
+ return callContext.fail(
+ throwOnFailure,
+ String.format(
+ "Invalid target mapping for argument
'op_mapping'. "
+ + "Unknown RowKind: '%s'. Valid
values are: %s.",
+ rowKindName, VALID_ROW_KIND_NAMES));
+ }
+ final boolean isDuplicate = !allRowKindsSeen.add(rowKindName);
+ if (isDuplicate) {
+ return callContext.fail(
+ throwOnFailure,
+ String.format(
+ "Invalid target mapping for argument
'op_mapping'. "
+ + "Duplicate RowKind: '%s'.",
Review Comment:
```suggestion
+ "Duplicate change operation:
'%s'.",
```
--
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]
