This is an automated email from the ASF dual-hosted git repository.
gustavodemorais pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new b3f4d60249a [FLINK-40968][table] Fix Table API map() op_mapping for TO
and FROM_CHANGELOG PTFs
b3f4d60249a is described below
commit b3f4d60249a60a6337030f3aac07a4dec8440537
Author: Timo Theusner <[email protected]>
AuthorDate: Wed Oct 21 16:41:41 2026 +0200
[FLINK-40968][table] Fix Table API map() op_mapping for TO and
FROM_CHANGELOG PTFs
Fixed in #29474.
---
.../resolver/rules/ResolveCallByArgumentsRule.java | 35 ++++-
.../strategies/FromChangelogTypeStrategy.java | 118 +++++----------
.../strategies/ToChangelogTypeStrategy.java | 52 ++-----
.../FromChangelogInputTypeStrategyTest.java | 32 +++-
.../ToChangelogInputTypeStrategyTest.java | 23 +++
.../exec/stream/FromChangelogSemanticTests.java | 4 +
.../exec/stream/FromChangelogTestPrograms.java | 162 +++++++++++++++++++++
.../exec/stream/ToChangelogSemanticTests.java | 6 +
.../nodes/exec/stream/ToChangelogTestPrograms.java | 154 ++++++++++++++++++++
9 files changed, 469 insertions(+), 117 deletions(-)
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
index 9f28484f585..07930a97875 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/expressions/resolver/rules/ResolveCallByArgumentsRule.java
@@ -60,6 +60,7 @@ import
org.apache.flink.table.types.inference.TypeInferenceUtil.Result;
import
org.apache.flink.table.types.inference.TypeInferenceUtil.SurroundingInfo;
import org.apache.flink.table.types.inference.TypeStrategies;
import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeFamily;
import org.apache.flink.table.types.utils.DataTypeUtils;
import javax.annotation.Nullable;
@@ -67,6 +68,7 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -607,7 +609,9 @@ final class ResolveCallByArgumentsRule implements
ResolverRule {
@Override
public boolean isArgumentLiteral(int pos) {
final ResolvedExpression arg = getArgument(pos);
- return arg instanceof ValueLiteralExpression || arg instanceof
TypeLiteralExpression;
+ return arg instanceof ValueLiteralExpression
+ || arg instanceof TypeLiteralExpression
+ || isLiteralMap(arg);
}
@Override
@@ -637,6 +641,9 @@ final class ResolveCallByArgumentsRule implements
ResolverRule {
final ValueLiteralExpression literal =
(ValueLiteralExpression) arg;
return literal.getValueAs(clazz);
}
+ if (clazz == Map.class && isLiteralMap(arg)) {
+ return Optional.of((T) toStringMap(arg));
+ }
return Optional.empty();
}
@@ -715,6 +722,32 @@ final class ResolveCallByArgumentsRule implements
ResolverRule {
return isGroupedAggregation;
}
+ private static boolean isLiteralMap(ResolvedExpression arg) {
+ return isFunction(arg, BuiltInFunctionDefinitions.MAP)
+ && arg.getResolvedChildren().stream()
+
.allMatch(TableApiCallContext::isNonNullStringLiteral);
+ }
+
+ private static boolean isNonNullStringLiteral(ResolvedExpression
expression) {
+ return expression instanceof ValueLiteralExpression
+ && ((ValueLiteralExpression)
expression).getValueAs(String.class).isPresent()
+ && expression
+ .getOutputDataType()
+ .getLogicalType()
+ .is(LogicalTypeFamily.CHARACTER_STRING);
+ }
+
+ private static Map<String, String> toStringMap(ResolvedExpression
literalMap) {
+ final List<ResolvedExpression> entries =
literalMap.getResolvedChildren();
+ final Map<String, String> map = new LinkedHashMap<>();
+ for (int i = 0; i < entries.size(); i += 2) {
+ map.put(
+ ExpressionUtils.stringValue(entries.get(i)),
+ ExpressionUtils.stringValue(entries.get(i + 1)));
+ }
+ return map;
+ }
+
private ResolvedExpression getArgument(int pos) {
if (pos >= resolvedArgs.size()) {
throw new IndexOutOfBoundsException(
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/FromChangelogTypeStrategy.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/FromChangelogTypeStrategy.java
index 13fa1245e62..14babdd82ea 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/FromChangelogTypeStrategy.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/FromChangelogTypeStrategy.java
@@ -157,68 +157,43 @@ public final class FromChangelogTypeStrategy {
private static Optional<List<DataType>> validateInputs(
final CallContext callContext, final boolean throwOnFailure) {
- Optional<List<DataType>> error;
-
- error = validateTableArg(callContext, throwOnFailure);
- if (error.isPresent()) {
- return error;
- }
-
- error = validateOpDescriptor(callContext, throwOnFailure);
- if (error.isPresent()) {
- return error;
- }
-
- error = validateOpColumn(callContext, throwOnFailure);
- if (error.isPresent()) {
- return error;
- }
-
- error = validateOpMapping(callContext, throwOnFailure);
- if (error.isPresent()) {
- return error;
- }
-
- error = validateErrorHandling(callContext, throwOnFailure);
+ final Optional<String> error =
+ validateTableArg(callContext)
+ .or(() -> validateOpDescriptor(callContext))
+ .or(() -> validateOpColumn(callContext))
+ .or(() -> validateOpMapping(callContext))
+ .or(() -> validateErrorHandling(callContext));
if (error.isPresent()) {
- return error;
+ return callContext.fail(throwOnFailure, error.get());
}
-
return Optional.of(callContext.getArgumentDataTypes());
}
- private static Optional<List<DataType>> validateTableArg(
- final CallContext callContext, final boolean throwOnFailure) {
+ private static Optional<String> validateTableArg(final CallContext
callContext) {
if (callContext.getTableSemantics(ARG_TABLE).isEmpty()) {
- return callContext.fail(
- throwOnFailure, "First argument must be a table for
FROM_CHANGELOG.");
+ return Optional.of("First argument must be a table for
FROM_CHANGELOG.");
}
return Optional.empty();
}
- private static Optional<List<DataType>> validateOpDescriptor(
- final CallContext callContext, final boolean throwOnFailure) {
+ private static Optional<String> validateOpDescriptor(final CallContext
callContext) {
final Optional<ColumnList> opDescriptor =
callContext.getArgumentValue(ARG_OP, ColumnList.class);
if (opDescriptor.isPresent() && opDescriptor.get().getNames().size()
!= 1) {
- return callContext.fail(
- throwOnFailure,
+ return Optional.of(
"The descriptor for argument 'op' must contain exactly one
column name.");
}
return Optional.empty();
}
/** Validates that the op column exists in the input schema and is of
STRING type. */
- private static Optional<List<DataType>> validateOpColumn(
- final CallContext callContext, final boolean throwOnFailure) {
-
+ private static Optional<String> validateOpColumn(final CallContext
callContext) {
final TableSemantics tableSemantics =
callContext.getTableSemantics(ARG_TABLE).get();
final String opColumnName =
ChangelogTypeStrategyUtils.resolveOpColumnName(callContext);
final OptionalInt opIndex =
ChangelogTypeStrategyUtils.resolveOpColumnIndex(tableSemantics, opColumnName);
if (opIndex.isEmpty()) {
- return callContext.fail(
- throwOnFailure,
+ return Optional.of(
String.format(
"The op column '%s' does not exist in the input
schema.",
opColumnName));
@@ -228,8 +203,7 @@ public final class FromChangelogTypeStrategy {
.get(opIndex.getAsInt())
.getLogicalType();
if (!opFieldType.is(LogicalTypeFamily.CHARACTER_STRING)) {
- return callContext.fail(
- throwOnFailure,
+ return Optional.of(
String.format(
"The op column '%s' must be of STRING type, but
was '%s'.",
opColumnName, opFieldType));
@@ -239,35 +213,32 @@ public final class FromChangelogTypeStrategy {
/** Validates op_mapping is a literal and its values are valid change
operation names. */
@SuppressWarnings("unchecked")
- private static Optional<List<DataType>> validateOpMapping(
- final CallContext callContext, final boolean throwOnFailure) {
+ private static Optional<String> validateOpMapping(final CallContext
callContext) {
final boolean hasMappingArgProvided =
!callContext.isArgumentNull(ARG_OP_MAPPING);
final boolean isMappingArgLiteral =
callContext.isArgumentLiteral(ARG_OP_MAPPING);
if (hasMappingArgProvided && !isMappingArgLiteral) {
- return callContext.fail(
- throwOnFailure, "The 'op_mapping' argument must be a
constant MAP literal.");
+ return Optional.of("The 'op_mapping' argument must be a constant
MAP literal.");
}
final Optional<Map> opMapping =
callContext.getArgumentValue(ARG_OP_MAPPING, Map.class);
- if (opMapping.isPresent()) {
- final Map<String, String> mapping = opMapping.get();
- final Optional<List<DataType>> validationError =
- validateOpMappingValues(callContext, mapping,
throwOnFailure);
- if (validationError.isPresent()) {
- return validationError;
- }
- // A mapping that produces UPDATE_AFTER without UPDATE_BEFORE
describes an upsert
- // changelog. Upsert mode requires a key, so PARTITION BY must be
present on the
- // table argument; otherwise the call would produce key-less
updates.
- if (describesUpsert(mapping) &&
!isPartitioned(callContext::getTableSemantics)) {
- return callContext.fail(
- throwOnFailure,
- "An 'op_mapping' that produces UPDATE_AFTER without
UPDATE_BEFORE "
- + "describes an upsert changelog and requires
a key. Add "
- + "PARTITION BY to the input table to define
the upsert key, "
- + "or include UPDATE_BEFORE in the mapping for
retract "
- + "semantics.");
- }
+ if (opMapping.isEmpty()) {
+ return Optional.empty();
+ }
+ final Map<String, String> mapping = opMapping.get();
+ final Optional<String> valuesError = validateOpMappingValues(mapping);
+ if (valuesError.isPresent()) {
+ return valuesError;
+ }
+ // A mapping that produces UPDATE_AFTER without UPDATE_BEFORE
describes an upsert
+ // changelog. Upsert mode requires a key, so PARTITION BY must be
present on the
+ // table argument; otherwise the call would produce key-less updates.
+ if (describesUpsert(mapping) &&
!isPartitioned(callContext::getTableSemantics)) {
+ return Optional.of(
+ "An 'op_mapping' that produces UPDATE_AFTER without
UPDATE_BEFORE "
+ + "describes an upsert changelog and requires a
key. Add "
+ + "PARTITION BY to the input table to define the
upsert key, "
+ + "or include UPDATE_BEFORE in the mapping for
retract "
+ + "semantics.");
}
return Optional.empty();
}
@@ -276,17 +247,13 @@ public final class FromChangelogTypeStrategy {
* Validates op_mapping values. Values must be valid Flink change
operation names. Each name
* must appear at most once across all entries.
*/
- private static Optional<List<DataType>> validateOpMappingValues(
- final CallContext callContext,
- final Map<String, String> opMapping,
- final boolean throwOnFailure) {
+ private static Optional<String> validateOpMappingValues(final Map<String,
String> opMapping) {
final Set<String> allRowKindsSeen = new HashSet<>();
for (final String value : opMapping.values()) {
final String rowKindName = value.trim();
if (!VALID_ROW_KIND_NAMES.contains(rowKindName)) {
- return callContext.fail(
- throwOnFailure,
+ return Optional.of(
String.format(
"Invalid target mapping for argument
'op_mapping'. "
+ "Unknown change operation: '%s'.
Valid values are: %s.",
@@ -294,8 +261,7 @@ public final class FromChangelogTypeStrategy {
}
final boolean isDuplicate = !allRowKindsSeen.add(rowKindName);
if (isDuplicate) {
- return callContext.fail(
- throwOnFailure,
+ return Optional.of(
String.format(
"Invalid target mapping for argument
'op_mapping'. "
+ "Duplicate change operation: '%s'. "
@@ -307,14 +273,11 @@ public final class FromChangelogTypeStrategy {
return Optional.empty();
}
- private static Optional<List<DataType>> validateErrorHandling(
- final CallContext callContext, final boolean throwOnFailure) {
+ private static Optional<String> validateErrorHandling(final CallContext
callContext) {
final boolean hasErrorHandlingArgProvided =
!callContext.isArgumentNull(ARG_ERROR_HANDLING);
final boolean isErrorHandlingArgLiteral =
callContext.isArgumentLiteral(ARG_ERROR_HANDLING);
if (hasErrorHandlingArgProvided && !isErrorHandlingArgLiteral) {
- return callContext.fail(
- throwOnFailure,
- "The 'error_handling' argument must be a constant STRING
literal.");
+ return Optional.of("The 'error_handling' argument must be a
constant STRING literal.");
}
final Optional<String> optionalErrorHandlingArg =
@@ -322,8 +285,7 @@ public final class FromChangelogTypeStrategy {
if (optionalErrorHandlingArg.isPresent()) {
final String errorHandlingMode = optionalErrorHandlingArg.get();
if (ErrorHandlingMode.fromName(errorHandlingMode).isEmpty()) {
- return callContext.fail(
- throwOnFailure,
+ return Optional.of(
String.format(
"Invalid value for argument 'error_handling':
'%s'. Valid values are: %s.",
errorHandlingMode,
ErrorHandlingMode.validNames()));
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java
index a2b8e13681b..194c5301329 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/strategies/ToChangelogTypeStrategy.java
@@ -102,42 +102,28 @@ public final class ToChangelogTypeStrategy {
private static Optional<List<DataType>> validateInputs(
final CallContext callContext, final boolean throwOnFailure) {
- Optional<List<DataType>> error;
-
- error = validateTableArg(callContext, throwOnFailure);
- if (error.isPresent()) {
- return error;
- }
-
- error = validateOpDescriptor(callContext, throwOnFailure);
+ final Optional<String> error =
+ validateTableArg(callContext)
+ .or(() -> validateOpDescriptor(callContext))
+ .or(() -> validateOpMapping(callContext));
if (error.isPresent()) {
- return error;
+ return callContext.fail(throwOnFailure, error.get());
}
-
- error = validateOpMapping(callContext, throwOnFailure);
- if (error.isPresent()) {
- return error;
- }
-
return Optional.of(callContext.getArgumentDataTypes());
}
- private static Optional<List<DataType>> validateTableArg(
- final CallContext callContext, final boolean throwOnFailure) {
+ private static Optional<String> validateTableArg(final CallContext
callContext) {
if (callContext.getTableSemantics(ARG_TABLE).isEmpty()) {
- return callContext.fail(
- throwOnFailure, "First argument must be a table for
TO_CHANGELOG.");
+ return Optional.of("First argument must be a table for
TO_CHANGELOG.");
}
return Optional.empty();
}
- private static Optional<List<DataType>> validateOpDescriptor(
- final CallContext callContext, final boolean throwOnFailure) {
+ private static Optional<String> validateOpDescriptor(final CallContext
callContext) {
final Optional<ColumnList> opDescriptor =
callContext.getArgumentValue(ARG_OP, ColumnList.class);
if (opDescriptor.isPresent() && opDescriptor.get().getNames().size()
!= 1) {
- return callContext.fail(
- throwOnFailure,
+ return Optional.of(
"The descriptor for argument 'op' must contain exactly one
column name.");
}
return Optional.empty();
@@ -145,19 +131,16 @@ public final class ToChangelogTypeStrategy {
/** Validates op_mapping is a constant literal and that its keys are
well-formed. */
@SuppressWarnings("rawtypes")
- private static Optional<List<DataType>> validateOpMapping(
- final CallContext callContext, final boolean throwOnFailure) {
+ private static Optional<String> validateOpMapping(final CallContext
callContext) {
final boolean hasMappingArgProvided =
!callContext.isArgumentNull(ARG_OP_MAPPING);
final boolean isMappingArgLiteral =
callContext.isArgumentLiteral(ARG_OP_MAPPING);
if (hasMappingArgProvided && !isMappingArgLiteral) {
- return callContext.fail(
- throwOnFailure, "The 'op_mapping' argument must be a
constant MAP literal.");
+ return Optional.of("The 'op_mapping' argument must be a constant
MAP literal.");
}
final Optional<Map> opMapping =
callContext.getArgumentValue(ARG_OP_MAPPING, Map.class);
if (opMapping.isPresent()) {
- return validateOpMappingKeys(
- callContext, (Map<String, String>) opMapping.get(),
throwOnFailure);
+ return validateOpMappingKeys((Map<String, String>)
opMapping.get());
}
return Optional.empty();
}
@@ -168,18 +151,14 @@ public final class ToChangelogTypeStrategy {
* trimmed. Names are case-sensitive and must match exactly (e.g., {@code
INSERT}, not {@code
* insert}). Each name must be valid and appear at most once across all
entries.
*/
- private static Optional<List<DataType>> validateOpMappingKeys(
- final CallContext callContext,
- final Map<String, String> opMapping,
- final boolean throwOnFailure) {
+ private static Optional<String> validateOpMappingKeys(final Map<String,
String> opMapping) {
final Set<String> allRowKindsSeen = new HashSet<>();
for (final String key : opMapping.keySet()) {
final String[] rowKindNames = key.split(",");
for (final String rawName : rowKindNames) {
final String rowKindName = rawName.trim();
if (!VALID_ROW_KIND_NAMES.contains(rowKindName)) {
- return callContext.fail(
- throwOnFailure,
+ return Optional.of(
String.format(
"Invalid target mapping for argument
'op_mapping'. "
+ "Unknown change operation: '%s'.
Operations are case-sensitive. Valid values are: %s.",
@@ -187,8 +166,7 @@ public final class ToChangelogTypeStrategy {
}
final boolean isDuplicate = !allRowKindsSeen.add(rowKindName);
if (isDuplicate) {
- return callContext.fail(
- throwOnFailure,
+ return Optional.of(
String.format(
"Invalid target mapping for argument
'op_mapping'. "
+ "Duplicate change operation:
'%s'.",
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/FromChangelogInputTypeStrategyTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/FromChangelogInputTypeStrategyTest.java
index 70595f1c597..41ff7e9941b 100644
---
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/FromChangelogInputTypeStrategyTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/FromChangelogInputTypeStrategyTest.java
@@ -20,14 +20,23 @@ package org.apache.flink.table.types.inference.strategies;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.InputTypeStrategies;
import org.apache.flink.table.types.inference.InputTypeStrategiesTestBase;
+import org.apache.flink.table.types.inference.utils.CallContextMock;
import org.apache.flink.table.types.inference.utils.TableSemanticsMock;
+import org.apache.flink.table.types.utils.DataTypeFactoryMock;
import org.apache.flink.types.ColumnList;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.stream.Stream;
import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.FROM_CHANGELOG_INPUT_TYPE_STRATEGY;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link FromChangelogTypeStrategy#INPUT_TYPE_STRATEGY}. */
class FromChangelogInputTypeStrategyTest extends InputTypeStrategiesTestBase {
@@ -233,6 +242,27 @@ class FromChangelogInputTypeStrategyTest extends
InputTypeStrategiesTestBase {
"d", "DELETE"))
.calledWithLiteralAt(3, "Fail")
.expectErrorMessage(
- "Invalid value for argument 'error_handling':
'Fail'. Valid values are: FAIL, SKIP."));
+ "Invalid value for argument 'error_handling':
'Fail'. Valid values are: FAIL, SKIP."),
+ TestSpec.forStrategy(
+ "Nested call is typed without the table
argument",
+ InputTypeStrategies.WILDCARD)
+
.surroundingStrategy(FROM_CHANGELOG_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(STRING_TYPE)
+ .expectArgumentTypes(STRING_TYPE));
+ }
+
+ @Test
+ void testReportsFailureWithoutThrowingWhenTableArgumentIsMissing() {
+ final CallContextMock callContext = new CallContextMock();
+ callContext.typeFactory = new DataTypeFactoryMock();
+ callContext.name = "FROM_CHANGELOG";
+ callContext.argumentDataTypes = List.of(TABLE_TYPE, DESCRIPTOR_TYPE,
MAP_TYPE, STRING_TYPE);
+ callContext.argumentLiterals = Collections.nCopies(4, false);
+ callContext.argumentNulls = Collections.nCopies(4, false);
+ callContext.argumentValues = Collections.nCopies(4, Optional.empty());
+ callContext.tableSemantics = Collections.emptyMap();
+
+
assertThat(FROM_CHANGELOG_INPUT_TYPE_STRATEGY.inferInputTypes(callContext,
false))
+ .isEmpty();
}
}
diff --git
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ToChangelogInputTypeStrategyTest.java
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ToChangelogInputTypeStrategyTest.java
index 36607885bd8..65991e64fa8 100644
---
a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ToChangelogInputTypeStrategyTest.java
+++
b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/types/inference/strategies/ToChangelogInputTypeStrategyTest.java
@@ -21,10 +21,17 @@ package org.apache.flink.table.types.inference.strategies;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.InputTypeStrategiesTestBase;
+import org.apache.flink.table.types.inference.utils.CallContextMock;
import org.apache.flink.table.types.inference.utils.TableSemanticsMock;
+import org.apache.flink.table.types.utils.DataTypeFactoryMock;
import org.apache.flink.types.ColumnList;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.stream.Stream;
import static
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies.TO_CHANGELOG_INPUT_TYPE_STRATEGY;
@@ -32,6 +39,7 @@ import static
org.apache.flink.table.types.inference.strategies.ToChangelogTypeS
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;
import static
org.apache.flink.table.types.inference.strategies.ToChangelogTypeStrategy.ARG_TABLE;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link ToChangelogTypeStrategy#INPUT_TYPE_STRATEGY}. */
class ToChangelogInputTypeStrategyTest extends InputTypeStrategiesTestBase {
@@ -130,4 +138,19 @@ class ToChangelogInputTypeStrategyTest extends
InputTypeStrategiesTestBase {
ARG_OP_MAPPING, Map.of("INSERT, DELETE", "A",
"DELETE", "B"))
.expectErrorMessage("Duplicate change operation:
'DELETE'"));
}
+
+ @Test
+ void testReportsFailureWithoutThrowingWhenTableArgumentIsMissing() {
+ final CallContextMock callContext = new CallContextMock();
+ callContext.typeFactory = new DataTypeFactoryMock();
+ callContext.name = "TO_CHANGELOG";
+ callContext.argumentDataTypes =
+ List.of(TABLE_TYPE, DESCRIPTOR_TYPE, MAP_TYPE, BOOLEAN_TYPE);
+ callContext.argumentLiterals = Collections.nCopies(4, false);
+ callContext.argumentNulls = Collections.nCopies(4, false);
+ callContext.argumentValues = Collections.nCopies(4, Optional.empty());
+ callContext.tableSemantics = Collections.emptyMap();
+
+
assertThat(TO_CHANGELOG_INPUT_TYPE_STRATEGY.inferInputTypes(callContext,
false)).isEmpty();
+ }
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/FromChangelogSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/FromChangelogSemanticTests.java
index ac0901853fb..3041b1e005d 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/FromChangelogSemanticTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/FromChangelogSemanticTests.java
@@ -39,6 +39,10 @@ public class FromChangelogSemanticTests extends
SemanticTestBase {
FromChangelogTestPrograms.SKIP_NULL_OP_CODE,
FromChangelogTestPrograms.TABLE_API_DEFAULT,
FromChangelogTestPrograms.TABLE_API_RETRACT_PARTITION_BY,
+ FromChangelogTestPrograms.TABLE_API_CUSTOM_OP_MAPPING,
+ FromChangelogTestPrograms.TABLE_API_UPSERT_PARTITION_BY,
+ FromChangelogTestPrograms.TABLE_API_SKIP_INVALID_OP_HANDLING,
+ FromChangelogTestPrograms.TABLE_API_ORDER_BY_CUSTOM_OP_MAPPING,
FromChangelogTestPrograms.ROUND_TRIP,
FromChangelogTestPrograms.INVALID_OP_CODE,
FromChangelogTestPrograms.NULL_OP_CODE);
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/FromChangelogTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/FromChangelogTestPrograms.java
index 5fa3956de3b..f29d2e23d95 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/FromChangelogTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/FromChangelogTestPrograms.java
@@ -26,7 +26,12 @@ import org.apache.flink.table.test.program.TableTestProgram;
import org.apache.flink.types.Row;
import org.apache.flink.types.RowKind;
+import java.time.Instant;
+
import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.descriptor;
+import static org.apache.flink.table.api.Expressions.lit;
+import static org.apache.flink.table.api.Expressions.map;
/** {@link TableTestProgram} definitions for testing the built-in
FROM_CHANGELOG PTF. */
public class FromChangelogTestPrograms {
@@ -313,6 +318,163 @@ public class FromChangelogTestPrograms {
"sink")
.build();
+ public static final TableTestProgram TABLE_API_CUSTOM_OP_MAPPING =
+ TableTestProgram.of(
+ "from-changelog-table-api-custom-op-mapping",
+ "Table.fromChangelog() with a map() op_mapping
using comma-separated keys")
+ .setupTableSource(
+ SourceTestStep.newBuilder("cdc_stream")
+ .addSchema(SIMPLE_CDC_SCHEMA)
+ .producedValues(
+ Row.of(1, "c", "Alice"),
+ Row.of(2, "r", "Bob"),
+ Row.of(1, "ub", "Alice"),
+ Row.of(1, "ua", "Alice2"),
+ Row.of(2, "d", "Bob"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("id INT", "name STRING")
+ .consumedValues(
+ 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, "Alice2"),
+ Row.ofKind(RowKind.DELETE, 2,
"Bob"))
+ .build())
+ .runTableApi(
+ env ->
+ env.from("cdc_stream")
+ .fromChangelog(
+
descriptor("op").asArgument("op"),
+ map(
+ "c, r",
+ "INSERT",
+ "ub",
+
"UPDATE_BEFORE",
+ "ua",
+
"UPDATE_AFTER",
+ "d",
+ "DELETE")
+
.asArgument("op_mapping")),
+ "sink")
+ .build();
+
+ public static final TableTestProgram TABLE_API_UPSERT_PARTITION_BY =
+ TableTestProgram.of(
+ "from-changelog-table-api-upsert-partition-by",
+ "PartitionedTable.fromChangelog() with a map()
op_mapping without "
+ + "UPDATE_BEFORE produces an upsert
changelog")
+ .setupTableSource(
+ SourceTestStep.newBuilder("cdc_stream")
+ .addSchema("name STRING", "id INT", "op
STRING")
+ .producedValues(
+ Row.of("Alice", 1, "INSERT"),
+ Row.of("Bob", 2, "INSERT"),
+ Row.of("Alice2", 1,
"UPDATE_AFTER"),
+ Row.of("Bob", 2, "DELETE"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("id INT PRIMARY KEY NOT
ENFORCED", "name STRING")
+ .consumedValues(
+ Row.ofKind(RowKind.INSERT, 1,
"Alice"),
+ Row.ofKind(RowKind.INSERT, 2,
"Bob"),
+ Row.ofKind(RowKind.UPDATE_AFTER,
1, "Alice2"),
+ Row.ofKind(RowKind.DELETE, 2,
"Bob"))
+ .build())
+ .runTableApi(
+ env ->
+ env.from("cdc_stream")
+ .partitionBy($("id"))
+ .fromChangelog(
+ map(
+ "INSERT",
+ "INSERT",
+
"UPDATE_AFTER",
+
"UPDATE_AFTER",
+ "DELETE",
+ "DELETE")
+
.asArgument("op_mapping")),
+ "sink")
+ .build();
+
+ public static final TableTestProgram TABLE_API_SKIP_INVALID_OP_HANDLING =
+ TableTestProgram.of(
+ "from-changelog-table-api-unmapped-codes-dropped",
+ "Table.fromChangelog() with error_handling='SKIP'
drops unmapped op codes")
+ .setupTableSource(
+ SourceTestStep.newBuilder("cdc_stream")
+ .addSchema(SIMPLE_CDC_SCHEMA)
+ .producedValues(
+ Row.of(1, "INSERT", "Alice"),
+ Row.of(2, "INSERT", "Bob"),
+ Row.of(1, "UNKNOWN", "Alice2"),
+ Row.of(2, "DELETE", "Bob"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("id INT", "name STRING")
+ .consumedValues(
+ Row.ofKind(RowKind.INSERT, 1,
"Alice"),
+ Row.ofKind(RowKind.INSERT, 2,
"Bob"),
+ Row.ofKind(RowKind.DELETE, 2,
"Bob"))
+ .build())
+ .runTableApi(
+ env ->
+ env.from("cdc_stream")
+ .fromChangelog(
+
lit("SKIP").asArgument("error_handling")),
+ "sink")
+ .build();
+
+ public static final TableTestProgram TABLE_API_ORDER_BY_CUSTOM_OP_MAPPING =
+ TableTestProgram.of(
+
"from-changelog-table-api-order-by-custom-op-mapping",
+ "PartitionedTable.orderBy().fromChangelog() with a
map() op_mapping "
+ + "applies out-of-order events per key in
event-time order")
+ .setupTableSource(
+ SourceTestStep.newBuilder("cdc_stream")
+ .addSchema(
+ "id INT",
+ "op STRING",
+ "name STRING",
+ "event_time TIMESTAMP_LTZ(3)",
+ "WATERMARK FOR event_time AS
event_time - INTERVAL '1' MINUTE")
+ // Each key's events arrive in reverse
event-time order. The
+ // watermark delay keeps them all on time.
+ .producedValues(
+ Row.of(1, "ua", "Alice2",
Instant.ofEpochMilli(2)),
+ Row.of(1, "c", "Alice",
Instant.ofEpochMilli(1)),
+ Row.of(2, "d", "Bob",
Instant.ofEpochMilli(4)),
+ Row.of(2, "c", "Bob",
Instant.ofEpochMilli(3)),
+ Row.of(3, "c", "Carol",
Instant.ofEpochMilli(5)))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("id INT PRIMARY KEY NOT
ENFORCED", "name STRING")
+ // Arrival order would leave [1, Alice]
and [2, Bob].
+ .consumedValues("+I[1, Alice2]", "+I[3,
Carol]")
+ .testMaterializedData()
+ .build())
+ .runTableApi(
+ env ->
+ env.from("cdc_stream")
+ .partitionBy($("id"))
+ .orderBy($("event_time").asc())
+ .fromChangelog(
+ map(
+ "c",
+ "INSERT",
+ "ua",
+
"UPDATE_AFTER",
+ "d",
+ "DELETE")
+
.asArgument("op_mapping"))
+ .select($("id"), $("name")),
+ "sink")
+ .build();
+
//
--------------------------------------------------------------------------------------------
// Round-trip test: FROM_CHANGELOG(TO_CHANGELOG(table))
//
--------------------------------------------------------------------------------------------
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogSemanticTests.java
index 24f248f01ce..35b91c54331 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogSemanticTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogSemanticTests.java
@@ -39,12 +39,18 @@ public class ToChangelogSemanticTests extends
SemanticTestBase {
ToChangelogTestPrograms.CUSTOM_OP_NAME,
ToChangelogTestPrograms.TABLE_API_DEFAULT,
ToChangelogTestPrograms.TABLE_API_RETRACT_PARTITION_BY,
+ ToChangelogTestPrograms.TABLE_API_CUSTOM_OP_MAPPING,
+ ToChangelogTestPrograms.TABLE_API_DELETION_FLAG,
+ ToChangelogTestPrograms.TABLE_API_PRODUCES_PARTIAL_DELETES,
ToChangelogTestPrograms.LAG_ON_UPSERT_VIA_CHANGELOG,
ToChangelogTestPrograms.LAG_ON_RETRACT_VIA_CHANGELOG,
ToChangelogTestPrograms.DELETION_FLAG,
ToChangelogTestPrograms.INVALID_DESCRIPTOR,
ToChangelogTestPrograms.INVALID_OP_MAPPING,
ToChangelogTestPrograms.DUPLICATE_ROW_KIND,
+ ToChangelogTestPrograms.TABLE_API_OP_MAPPING_WITH_NESTED_CALL,
+
ToChangelogTestPrograms.TABLE_API_OP_MAPPING_WITH_COLUMN_REFERENCE,
+ ToChangelogTestPrograms.TABLE_API_OP_MAPPING_WITH_NULL_ENTRY,
ToChangelogTestPrograms.RETRACT_PRODUCES_PARTIAL_DELETES,
ToChangelogTestPrograms.UPSERT_PRODUCES_FULL_DELETES,
ToChangelogTestPrograms.UPSERT_PRODUCES_PARTIAL_DELETES,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogTestPrograms.java
index a14b259b068..853e2e1d04a 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ToChangelogTestPrograms.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.planner.plan.nodes.exec.stream;
+import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.test.program.SinkTestStep;
@@ -29,6 +30,10 @@ import org.apache.flink.types.RowKind;
import java.time.Instant;
import static org.apache.flink.table.api.Expressions.$;
+import static org.apache.flink.table.api.Expressions.descriptor;
+import static org.apache.flink.table.api.Expressions.lit;
+import static org.apache.flink.table.api.Expressions.map;
+import static org.apache.flink.table.api.Expressions.nullOf;
/** {@link TableTestProgram} definitions for testing the built-in TO_CHANGELOG
PTF. */
public class ToChangelogTestPrograms {
@@ -481,6 +486,107 @@ public class ToChangelogTestPrograms {
.runTableApi(env ->
env.from("t").partitionBy($("id")).toChangelog(), "sink")
.build();
+ public static final TableTestProgram TABLE_API_CUSTOM_OP_MAPPING =
+ TableTestProgram.of(
+ "to-changelog-table-api-custom-op-mapping",
+ "Table.toChangelog() with a map() op_mapping maps
change operations to user-defined codes")
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema(
+ "name STRING PRIMARY KEY NOT
ENFORCED", "score BIGINT")
+ .addMode(ChangelogMode.all())
+ .producedValues(
+ Row.ofKind(RowKind.INSERT,
"Alice", 10L),
+ Row.ofKind(RowKind.INSERT, "Bob",
20L),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
"Alice", 10L),
+ Row.ofKind(RowKind.UPDATE_AFTER,
"Alice", 30L),
+ Row.ofKind(RowKind.DELETE, "Bob",
20L))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("op_code STRING", "name
STRING", "score BIGINT")
+ .consumedValues(
+ "+I[I, Alice, 10]",
+ "+I[I, Bob, 20]",
+ "+I[U, Alice, 30]")
+ .build())
+ .runTableApi(
+ env ->
+ env.from("t")
+ .toChangelog(
+
descriptor("op_code").asArgument("op"),
+ map("INSERT", "I",
"UPDATE_AFTER", "U")
+
.asArgument("op_mapping")),
+ "sink")
+ .build();
+
+ public static final TableTestProgram TABLE_API_DELETION_FLAG =
+ TableTestProgram.of(
+ "to-changelog-table-api-deletion-flag",
+ "Table.toChangelog() with a map() op_mapping
produces deletion flag output")
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema(
+ "name STRING PRIMARY KEY NOT
ENFORCED", "score BIGINT")
+ .addMode(ChangelogMode.all())
+ .producedValues(
+ Row.ofKind(RowKind.INSERT,
"Alice", 10L),
+ Row.ofKind(RowKind.INSERT, "Bob",
20L),
+ Row.ofKind(RowKind.UPDATE_BEFORE,
"Alice", 10L),
+ Row.ofKind(RowKind.UPDATE_AFTER,
"Alice", 30L),
+ Row.ofKind(RowKind.DELETE, "Bob",
20L))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("deleted STRING", "name
STRING", "score BIGINT")
+ .consumedValues(
+ "+I[false, Alice, 10]",
+ "+I[false, Bob, 20]",
+ "+I[false, Alice, 30]",
+ "+I[true, Bob, 20]")
+ .build())
+ .runTableApi(
+ env ->
+ env.from("t")
+ .toChangelog(
+
descriptor("deleted").asArgument("op"),
+ map(
+ "INSERT,
UPDATE_AFTER",
+ "false",
+ "DELETE",
+ "true")
+
.asArgument("op_mapping")),
+ "sink")
+ .build();
+
+ public static final TableTestProgram TABLE_API_PRODUCES_PARTIAL_DELETES =
+ TableTestProgram.of(
+ "to-changelog-table-api-produces-partial-deletes",
+ "Table.toChangelog() with
produces_full_deletes=false keeps only the "
+ + "upsert key on DELETE")
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema(
+ "name STRING PRIMARY KEY NOT
ENFORCED", "score BIGINT")
+ .addMode(ChangelogMode.upsert())
+ .producedValues(
+ Row.ofKind(RowKind.INSERT,
"Alice", 10L),
+ Row.ofKind(RowKind.DELETE,
"Alice", 10L))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("op STRING", "name STRING",
"score BIGINT")
+ .consumedValues(
+ "+I[INSERT, Alice, 10]",
"+I[DELETE, Alice, null]")
+ .build())
+ .runTableApi(
+ env ->
+ env.from("t")
+ .toChangelog(
+
lit(false).asArgument("produces_full_deletes")),
+ "sink")
+ .build();
+
//
--------------------------------------------------------------------------------------------
// Use case: LAG on updating streams via TO_CHANGELOG
//
--------------------------------------------------------------------------------------------
@@ -718,6 +824,54 @@ public class ToChangelogTestPrograms {
"Duplicate change operation: 'DELETE'")
.build();
+ public static final TableTestProgram TABLE_API_OP_MAPPING_WITH_NESTED_CALL
=
+ TableTestProgram.of(
+
"to-changelog-table-api-op-mapping-with-nested-call",
+ "fails when a map() op_mapping entry is a nested
call")
+ .setupTableSource(SIMPLE_SOURCE)
+ .runFailingTableApi(
+ env ->
+ env.from("t")
+ .toChangelog(
+ map("INSERT",
lit("i").upperCase())
+
.asArgument("op_mapping")),
+ "sink",
+ ValidationException.class,
+ "The 'op_mapping' argument must be a constant MAP
literal.")
+ .build();
+
+ public static final TableTestProgram
TABLE_API_OP_MAPPING_WITH_COLUMN_REFERENCE =
+ TableTestProgram.of(
+
"to-changelog-table-api-op-mapping-with-column-reference",
+ "fails when a map() op_mapping entry is a
non-literal column reference")
+ .setupTableSource(SIMPLE_SOURCE)
+ .runFailingTableApi(
+ env ->
+ env.from("t")
+ .toChangelog(
+ map("INSERT", $("name"))
+
.asArgument("op_mapping")),
+ "sink",
+ ValidationException.class,
+ "Cannot resolve field [name], input field
list:[].")
+ .build();
+
+ public static final TableTestProgram TABLE_API_OP_MAPPING_WITH_NULL_ENTRY =
+ TableTestProgram.of(
+
"to-changelog-table-api-op-mapping-with-null-entry",
+ "fails when a map() op_mapping entry is NULL")
+ .setupTableSource(SIMPLE_SOURCE)
+ .runFailingTableApi(
+ env ->
+ env.from("t")
+ .toChangelog(
+ map("INSERT",
nullOf(DataTypes.STRING()))
+
.asArgument("op_mapping")),
+ "sink",
+ ValidationException.class,
+ "The 'op_mapping' argument must be a constant MAP
literal.")
+ .build();
+
//
--------------------------------------------------------------------------------------------
// Full vs partial deletes matrix (input kind x PARTITION BY x
produces_full_deletes)
//
--------------------------------------------------------------------------------------------
