This is an automated email from the ASF dual-hosted git repository.
fhueske 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 cd34cf96b61 [FLINK-39708][table] Support TO_CHANGELOG retract/upsert
stream -> upsert stream with set semantics (#28199)
cd34cf96b61 is described below
commit cd34cf96b61d0417e25629c4fc82608f95af8ce8
Author: Gustavo de Morais <[email protected]>
AuthorDate: Thu May 21 17:11:43 2026 +0200
[FLINK-39708][table] Support TO_CHANGELOG retract/upsert stream -> upsert
stream with set semantics (#28199)
---
.../docs/sql/reference/queries/changelog.md | 12 ++++
.../functions/BuiltInFunctionDefinitions.java | 62 ++++++++++++++++--
.../table/types/inference/BuiltInCondition.java | 5 +-
.../table/types/inference/TraitCondition.java | 39 +++++++++++
.../flink/table/types/inference/TraitContext.java | 17 +++++
.../functions/bridging/BridgingSqlFunction.java | 11 ++++
.../exec/stream/ToChangelogSemanticTests.java | 3 +
.../nodes/exec/stream/ToChangelogTestPrograms.java | 75 ++++++++++++++++++++++
.../planner/plan/stream/sql/ToChangelogTest.java | 40 ++++++++++++
.../planner/plan/stream/sql/ToChangelogTest.xml | 40 ++++++++++++
.../runtime/functions/ptf/ToChangelogFunction.java | 34 ++++++++++
11 files changed, 330 insertions(+), 8 deletions(-)
diff --git a/docs/content/docs/sql/reference/queries/changelog.md
b/docs/content/docs/sql/reference/queries/changelog.md
index 3a7a97534c3..78f61e1ce10 100644
--- a/docs/content/docs/sql/reference/queries/changelog.md
+++ b/docs/content/docs/sql/reference/queries/changelog.md
@@ -372,6 +372,18 @@ SELECT * FROM TO_CHANGELOG(
-- op_code values are 'I' and 'U' instead of full names
```
+#### Upsert stream
+
+```sql
+SELECT * FROM TO_CHANGELOG(
+ input => TABLE upsert_source PARTITION BY id,
+ op_mapping => MAP['INSERT, UPDATE_AFTER', 'u', 'DELETE', 'd']
+)
+-- INSERT and UPDATE_AFTER produce op='u' (upsert)
+-- DELETE produces op='d'
+-- UPDATE_BEFORE is omitted
+```
+
#### Deletion flag pattern
```sql
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
index 4bf07dc4337..e3d30a5c8da 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/BuiltInFunctionDefinitions.java
@@ -60,6 +60,7 @@ import java.util.Collections;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
@@ -806,9 +807,10 @@ public final class BuiltInFunctionDefinitions {
.name("TO_CHANGELOG")
.kind(PROCESS_TABLE)
.staticArguments(
- // Row semantics (no PARTITION BY).
- // With PARTITION BY, switches to set
- // semantics for co-located parallel execution.
+ // Row semantics by default; PARTITION BY switches
to set semantics.
+ // REQUIRE_UPDATE_BEFORE / REQUIRE_FULL_DELETE are
conditional so the
+ // planner can skip ChangelogNormalize for upsert
sources whose
+ // op_mapping does not emit UB or full deletes.
StaticArgument.table(
"input",
Row.class,
@@ -816,12 +818,42 @@ public final class BuiltInFunctionDefinitions {
EnumSet.of(
StaticArgumentTrait.TABLE,
StaticArgumentTrait.ROW_SEMANTIC_TABLE,
-
StaticArgumentTrait.SUPPORT_UPDATES,
-
StaticArgumentTrait.REQUIRE_UPDATE_BEFORE,
-
StaticArgumentTrait.REQUIRE_FULL_DELETE))
+
StaticArgumentTrait.SUPPORT_UPDATES))
.withConditionalTrait(
StaticArgumentTrait.SET_SEMANTIC_TABLE,
- TraitCondition.hasPartitionBy()),
+ TraitCondition.hasPartitionBy())
+ .withConditionalTrait(
+
StaticArgumentTrait.REQUIRE_UPDATE_BEFORE,
+ TraitCondition.or(
+ // op_mapping omitted:
default mapping includes
+ // UPDATE_BEFORE.
+ TraitCondition.not(
+
TraitCondition.argIsPresent(
+
"op_mapping")),
+ TraitCondition.argMatches(
+ "op_mapping",
+ Map.class,
+ mapping ->
+
opMappingContainsKey(
+
(Map<String, String>)
+
mapping,
+
"UPDATE_BEFORE"))))
+ .withConditionalTrait(
+
StaticArgumentTrait.REQUIRE_FULL_DELETE,
+ TraitCondition.or(
+ // op_mapping omitted:
default mapping includes
+ // DELETE.
+ TraitCondition.not(
+
TraitCondition.argIsPresent(
+
"op_mapping")),
+ TraitCondition.argMatches(
+ "op_mapping",
+ Map.class,
+ mapping ->
+
opMappingContainsKey(
+
(Map<String, String>)
+
mapping,
+
"DELETE")))),
StaticArgument.scalar("op",
DataTypes.DESCRIPTOR(), true),
StaticArgument.scalar(
"op_mapping",
@@ -3373,6 +3405,22 @@ public final class BuiltInFunctionDefinitions {
public static final List<FunctionDefinition> ORDERING =
Arrays.asList(ORDER_ASC, ORDER_DESC);
+ /**
+ * True when {@code key} appears among the {@code op_mapping} keys. Each
map key may itself be a
+ * comma-separated list (e.g. {@code "INSERT,UPDATE_AFTER"}) - each part
is trimmed and compared
+ * against {@code key}, so one entry can cover multiple change kinds.
+ *
+ * <p>Used by {@code TO_CHANGELOG} conditional traits to inspect the
user-provided {@code
+ * op_mapping} argument.
+ */
+ private static boolean opMappingContainsKey(
+ final Map<String, String> opMapping, final String key) {
+ return opMapping.keySet().stream()
+ .flatMap(k -> Arrays.stream(k.split(",")))
+ .map(String::trim)
+ .anyMatch(key::equals);
+ }
+
@Internal
public static List<BuiltInFunctionDefinition> getDefinitions() {
final Field[] fields = BuiltInFunctionDefinitions.class.getFields();
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/BuiltInCondition.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/BuiltInCondition.java
index 2a06191ad13..ba87cc0559d 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/BuiltInCondition.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/BuiltInCondition.java
@@ -37,7 +37,10 @@ final class BuiltInCondition implements TraitCondition {
enum Kind {
HAS_PARTITION_BY,
ARG_IS_EQUAL_TO,
- NOT
+ ARG_IS_PRESENT,
+ NOT,
+ OR,
+ ARG_MATCHES
}
private final Kind kind;
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TraitCondition.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TraitCondition.java
index 93bdc5c8ce4..14734db11f4 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TraitCondition.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TraitCondition.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.types.inference;
import org.apache.flink.annotation.PublicEvolving;
import java.util.List;
+import java.util.function.Predicate;
/**
* A condition that determines whether a conditional trait on a {@link
StaticArgument} should be
@@ -32,6 +33,9 @@ import java.util.List;
* <p>Implementations must implement {@code hashCode} and {@code equals} for
{@link
* StaticArgument#equals}/{@link StaticArgument#hashCode} to work correctly.
The built-in factories
* below return value-comparable instances; user-supplied lambdas do not -
prefer the factories.
+ * {@link #argMatches} accepts a caller-supplied {@code Predicate} and is
therefore only value-equal
+ * when the same {@code Predicate} reference is reused; build the predicate
once and cache it if
+ * equality matters.
*
* <pre>{@code
* import static org.apache.flink.table.types.inference.TraitCondition.*;
@@ -68,4 +72,39 @@ public interface TraitCondition {
return new BuiltInCondition(
BuiltInCondition.Kind.NOT, List.of(condition), ctx ->
!condition.test(ctx));
}
+
+ /**
+ * True when either the {@code left} or the {@code right} {@link
TraitCondition} evaluates to
+ * true.
+ */
+ static TraitCondition or(final TraitCondition left, final TraitCondition
right) {
+ return new BuiltInCondition(
+ BuiltInCondition.Kind.OR,
+ List.of(left, right),
+ ctx -> left.test(ctx) || right.test(ctx));
+ }
+
+ /** True when the named scalar argument was provided by the caller. */
+ static TraitCondition argIsPresent(final String argName) {
+ return new BuiltInCondition(
+ BuiltInCondition.Kind.ARG_IS_PRESENT,
+ List.of(argName),
+ ctx -> ctx.hasScalarArgument(argName));
+ }
+
+ /**
+ * True when the named scalar argument is present and its value matches
{@code predicate}. False
+ * when the argument is absent, cannot be resolved as a literal of {@code
argClass}, or {@code
+ * predicate} evaluates to {@code false}.
+ *
+ * <p>Use this for ad-hoc conditions on scalar literals. Prefer the named
factories above when
+ * one fits.
+ */
+ static <T> TraitCondition argMatches(
+ final String argName, final Class<T> argClass, final Predicate<T>
predicate) {
+ return new BuiltInCondition(
+ BuiltInCondition.Kind.ARG_MATCHES,
+ List.of(argName, argClass, predicate),
+ ctx -> ctx.getScalarArgument(argName,
argClass).stream().anyMatch(predicate));
+ }
}
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TraitContext.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TraitContext.java
index e58a4b36dce..1c8d1a9b8bf 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TraitContext.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/types/inference/TraitContext.java
@@ -46,6 +46,12 @@ public interface TraitContext {
*/
<T> Optional<T> getScalarArgument(String name, Class<T> clazz);
+ /**
+ * Whether the named scalar argument was provided by the caller. Returns
false when the argument
+ * was omitted, or when an argument with the given name is not declared by
the function.
+ */
+ boolean hasScalarArgument(String name);
+
/**
* Builds a {@link TraitContext} from validation-time inputs.
*
@@ -76,6 +82,17 @@ public interface TraitContext {
}
return Optional.empty();
}
+
+ @Override
+ public boolean hasScalarArgument(final String name) {
+ for (int i = 0; i < staticArgs.size(); i++) {
+ final StaticArgument arg = staticArgs.get(i);
+ if (arg.is(StaticArgumentTrait.SCALAR) &&
arg.getName().equals(name)) {
+ return !callContext.isArgumentNull(i);
+ }
+ }
+ return false;
+ }
};
}
}
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlFunction.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlFunction.java
index 059fcd6184d..2c393adf875 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlFunction.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/bridging/BridgingSqlFunction.java
@@ -315,6 +315,17 @@ public class BridgingSqlFunction extends SqlFunction {
}
return Optional.empty();
}
+
+ @Override
+ public boolean hasScalarArgument(String name) {
+ for (int i = 0; i < declared.size(); i++) {
+ final StaticArgument arg = declared.get(i);
+ if (arg.is(StaticArgumentTrait.SCALAR) &&
arg.getName().equals(name)) {
+ return !callContext.isArgumentNull(i);
+ }
+ }
+ return false;
+ }
};
}
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 e038850fb8d..bfffa8bd67d 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
@@ -42,6 +42,8 @@ public class ToChangelogSemanticTests extends
SemanticTestBase {
ToChangelogTestPrograms.INSERT,
ToChangelogTestPrograms.RETRACT,
ToChangelogTestPrograms.UPSERT,
+ ToChangelogTestPrograms.UPSERT_PARTITION_BY,
+ ToChangelogTestPrograms.UPSERT_PARTITION_BY_KEY_ONLY_DELETES,
ToChangelogTestPrograms.RETRACT_PARTITION_BY,
ToChangelogTestPrograms.CUSTOM_OP_MAPPING,
ToChangelogTestPrograms.CUSTOM_OP_NAME,
@@ -52,6 +54,7 @@ public class ToChangelogSemanticTests extends
SemanticTestBase {
ToChangelogTestPrograms.DELETION_FLAG,
ToChangelogTestPrograms.INVALID_DESCRIPTOR,
ToChangelogTestPrograms.INVALID_OP_MAPPING,
+ ToChangelogTestPrograms.OP_MAPPING_REFERENCES_UNSUPPORTED_KIND,
ToChangelogTestPrograms.DUPLICATE_ROW_KIND);
}
}
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 7f33cfc1419..250ad1f1d8f 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
@@ -184,6 +184,68 @@ public class ToChangelogTestPrograms {
.runSql("INSERT INTO sink SELECT * FROM TO_CHANGELOG(input
=> TABLE t)")
.build();
+ public static final TableTestProgram UPSERT_PARTITION_BY =
+ TableTestProgram.of(
+ "to-changelog-upsert-partition-by",
+ "PARTITION BY upsert key + mapping without UB
skips ChangelogNormalize")
+ .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.INSERT, "Bob",
20L),
+ Row.ofKind(RowKind.UPDATE_AFTER,
"Alice", 30L),
+ Row.ofKind(RowKind.DELETE, "Bob",
20L))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("name STRING", "op STRING",
"score BIGINT")
+ .consumedValues(
+ "+I[Alice, C, 10]",
+ "+I[Bob, C, 20]",
+ "+I[Alice, C, 30]",
+ "+I[Bob, D, 20]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM TO_CHANGELOG("
+ + "input => TABLE t PARTITION BY name, "
+ + "op => DESCRIPTOR(op), "
+ + "op_mapping =>
MAP['INSERT,UPDATE_AFTER', 'C', 'DELETE', 'D'])")
+ .build();
+
+ public static final TableTestProgram UPSERT_PARTITION_BY_KEY_ONLY_DELETES =
+ TableTestProgram.of(
+
"to-changelog-upsert-partition-by-key-only-deletes",
+ "PARTITION BY upsert key + mapping without
UB/DELETE handles key-only deletes")
+ .setupTableSource(
+ SourceTestStep.newBuilder("t")
+ .addSchema(
+ "name STRING PRIMARY KEY NOT
ENFORCED", "score BIGINT")
+ .addMode(ChangelogMode.upsert(true))
+ .producedValues(
+ Row.ofKind(RowKind.INSERT,
"Alice", 10L),
+ Row.ofKind(RowKind.INSERT, "Bob",
20L),
+ Row.ofKind(RowKind.UPDATE_AFTER,
"Alice", 30L),
+ // Key-only delete: source only
knows the key.
+ Row.ofKind(RowKind.DELETE, "Bob",
null))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("name STRING", "op STRING",
"score BIGINT")
+ .consumedValues(
+ "+I[Alice, U, 10]",
+ "+I[Bob, U, 20]",
+ "+I[Alice, U, 30]")
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM TO_CHANGELOG("
+ + "input => TABLE t PARTITION BY name, "
+ + "op => DESCRIPTOR(op), "
+ + "op_mapping =>
MAP['INSERT,UPDATE_AFTER', 'U'])")
+ .build();
+
public static final TableTestProgram CUSTOM_OP_MAPPING =
TableTestProgram.of(
"to-changelog-custom-op-mapping",
@@ -511,6 +573,19 @@ public class ToChangelogTestPrograms {
"Unknown change operation: 'INVALID_KIND'")
.build();
+ public static final TableTestProgram
OP_MAPPING_REFERENCES_UNSUPPORTED_KIND =
+ TableTestProgram.of(
+
"to-changelog-op-mapping-references-unsupported-kind",
+ "fails when op_mapping references a change
operation the input cannot produce")
+ .setupTableSource(SIMPLE_SOURCE)
+ .runFailingSql(
+ "SELECT * FROM TO_CHANGELOG("
+ + "input => TABLE t, "
+ + "op_mapping => MAP['INSERT', 'I',
'DELETE', 'D'])",
+ ValidationException.class,
+ "the input table only produces [INSERT] and does
not produce [DELETE]")
+ .build();
+
public static final TableTestProgram DUPLICATE_ROW_KIND =
TableTestProgram.of(
"to-changelog-duplicate-rowkind",
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.java
index 12790318688..7f6f93849c2 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.java
@@ -104,4 +104,44 @@ public class ToChangelogTest extends TableTestBase {
"SELECT * FROM TO_CHANGELOG(input => TABLE retract_source
PARTITION BY id)",
CHANGELOG_MODE);
}
+
+ @Test
+ void testUpsertPartitionBy() {
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE upsert_source ("
+ + " id INT,"
+ + " name STRING,"
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'changelog-mode' = 'I,UA,D'"
+ + ")");
+ util.verifyRelPlan(
+ "SELECT * FROM TO_CHANGELOG("
+ + "input => TABLE upsert_source PARTITION BY id, "
+ + "op => DESCRIPTOR(op), "
+ + "op_mapping => MAP['INSERT,UPDATE_AFTER', 'C',
'DELETE', 'D'])",
+ CHANGELOG_MODE);
+ }
+
+ @Test
+ void testUpsertPartitionByNoUpdateBeforeAndDelete() {
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE upsert_source ("
+ + " id INT,"
+ + " name STRING,"
+ + " PRIMARY KEY (id) NOT ENFORCED"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'changelog-mode' = 'I,UA,D'"
+ + ")");
+ util.verifyRelPlan(
+ "SELECT * FROM TO_CHANGELOG("
+ + "input => TABLE upsert_source PARTITION BY id, "
+ + "op => DESCRIPTOR(op), "
+ + "op_mapping => MAP['INSERT,UPDATE_AFTER', 'C'])",
+ CHANGELOG_MODE);
+ }
}
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.xml
index b477039f2da..7cea1058e5c 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ToChangelogTest.xml
@@ -71,6 +71,46 @@ LogicalProject(op=[$0], id=[$1], name=[$2])
<![CDATA[
ProcessTableFunction(invocation=[TO_CHANGELOG(TABLE(#0), DEFAULT(), DEFAULT(),
DEFAULT(), DEFAULT())], uid=[null], select=[op,id,name],
rowType=[RecordType(VARCHAR(2147483647) op, INTEGER id, VARCHAR(2147483647)
name)], changelogMode=[I])
+- TableSourceScan(table=[[default_catalog, default_database,
retract_source]], fields=[id, name], changelogMode=[I,UB,UA,D])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUpsertPartitionBy">
+ <Resource name="sql">
+ <![CDATA[SELECT * FROM TO_CHANGELOG(input => TABLE upsert_source
PARTITION BY id, op => DESCRIPTOR(op), op_mapping => MAP['INSERT,UPDATE_AFTER',
'C', 'DELETE', 'D'])]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(id=[$0], op=[$1], name=[$2])
++- LogicalTableFunctionScan(invocation=[TO_CHANGELOG(TABLE(#0) PARTITION
BY($0), DESCRIPTOR(_UTF-16LE'op'),
MAP(_UTF-16LE'INSERT,UPDATE_AFTER':VARCHAR(19) CHARACTER SET "UTF-16LE",
_UTF-16LE'C', _UTF-16LE'DELETE':VARCHAR(19) CHARACTER SET "UTF-16LE",
_UTF-16LE'D'), DEFAULT(), DEFAULT())], rowType=[RecordType(INTEGER id,
VARCHAR(2147483647) op, VARCHAR(2147483647) name)])
+ +- LogicalProject(id=[$0], name=[$1])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
upsert_source]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+ProcessTableFunction(invocation=[TO_CHANGELOG(TABLE(#0) PARTITION BY($0),
DESCRIPTOR(_UTF-16LE'op'), MAP(_UTF-16LE'INSERT,UPDATE_AFTER':VARCHAR(19)
CHARACTER SET "UTF-16LE", _UTF-16LE'C', _UTF-16LE'DELETE':VARCHAR(19) CHARACTER
SET "UTF-16LE", _UTF-16LE'D'), DEFAULT(), DEFAULT())], uid=[TO_CHANGELOG],
select=[id,op,name], rowType=[RecordType(INTEGER id, VARCHAR(2147483647) op,
VARCHAR(2147483647) name)], changelogMode=[I])
++- Exchange(distribution=[hash[id]], changelogMode=[I,UA,D])
+ +- TableSourceScan(table=[[default_catalog, default_database,
upsert_source]], fields=[id, name], changelogMode=[I,UA,D])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUpsertPartitionByNoUpdateBeforeAndDelete">
+ <Resource name="sql">
+ <![CDATA[SELECT * FROM TO_CHANGELOG(input => TABLE upsert_source
PARTITION BY id, op => DESCRIPTOR(op), op_mapping => MAP['INSERT,UPDATE_AFTER',
'C'])]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(id=[$0], op=[$1], name=[$2])
++- LogicalTableFunctionScan(invocation=[TO_CHANGELOG(TABLE(#0) PARTITION
BY($0), DESCRIPTOR(_UTF-16LE'op'), MAP(_UTF-16LE'INSERT,UPDATE_AFTER',
_UTF-16LE'C'), DEFAULT(), DEFAULT())], rowType=[RecordType(INTEGER id,
VARCHAR(2147483647) op, VARCHAR(2147483647) name)])
+ +- LogicalProject(id=[$0], name=[$1])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
upsert_source]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+ProcessTableFunction(invocation=[TO_CHANGELOG(TABLE(#0) PARTITION BY($0),
DESCRIPTOR(_UTF-16LE'op'), MAP(_UTF-16LE'INSERT,UPDATE_AFTER', _UTF-16LE'C'),
DEFAULT(), DEFAULT())], uid=[TO_CHANGELOG], select=[id,op,name],
rowType=[RecordType(INTEGER id, VARCHAR(2147483647) op, VARCHAR(2147483647)
name)], changelogMode=[I])
++- Exchange(distribution=[hash[id]], changelogMode=[I,UA,D])
+ +- TableSourceScan(table=[[default_catalog, default_database,
upsert_source]], fields=[id, name], changelogMode=[I,UA,D])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java
index 36ed1a615bb..cf0547dbab0 100644
---
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java
+++
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/functions/ptf/ToChangelogFunction.java
@@ -19,6 +19,8 @@
package org.apache.flink.table.runtime.functions.ptf;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.MapData;
import org.apache.flink.table.data.RowData;
@@ -37,6 +39,7 @@ import org.apache.flink.types.RowKind;
import javax.annotation.Nullable;
import java.util.EnumMap;
+import java.util.List;
import java.util.Map;
/**
@@ -77,6 +80,9 @@ public class ToChangelogFunction extends
BuiltInProcessTableFunction<RowData> {
final Map<String, String> opMapping =
callContext.getArgumentValue(2, Map.class).orElse(null);
this.rawOpMap = buildOpMap(opMapping);
+ if (opMapping != null) {
+ validateOpMap(this.rawOpMap, tableSemantics);
+ }
this.outputIndices =
ChangelogTypeStrategyUtils.computeOutputIndices(tableSemantics);
}
@@ -108,6 +114,34 @@ public class ToChangelogFunction extends
BuiltInProcessTableFunction<RowData> {
return result;
}
+ /**
+ * Rejects user-provided mappings that reference change operations the
input changelog cannot
+ * produce. Without this check the extra entries are dead code: the
corresponding rows never
+ * arrive. E.g., if the input is INSERT-only, then UPDATE_BEFORE and
DELETE mappings are
+ * ignored, which is likely a user mistake.
+ *
+ * <p>Lives here rather than in the input type strategy because {@link
+ * TableSemantics#changelogMode()} returns empty during type inference and
is only populated at
+ * specialization time, which is when this constructor runs.
+ */
+ private static void validateOpMap(
+ final Map<RowKind, String> mapping, final TableSemantics
tableSemantics) {
+ final ChangelogMode inputMode =
tableSemantics.changelogMode().orElse(null);
+ if (inputMode == null) {
+ return;
+ }
+ final List<RowKind> unsupported =
+ mapping.keySet().stream().filter(kind ->
!inputMode.contains(kind)).toList();
+ if (!unsupported.isEmpty()) {
+ throw new ValidationException(
+ String.format(
+ "Invalid 'op_mapping' for TO_CHANGELOG: the input
table only produces "
+ + "%s and does not produce %s. Remove
those entries from the "
+ + "mapping.",
+ inputMode.getContainedKinds(), unsupported));
+ }
+ }
+
public void eval(
final Context ctx,
final RowData input,
