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 4644cabd8bf [FLINK-39538][table] Support upsert output mode for
FROM_CHANGELOG (#28164)
4644cabd8bf is described below
commit 4644cabd8bfed1104a6a1dbea854fe45a90877cd
Author: Ramin Gharib <[email protected]>
AuthorDate: Wed May 20 10:54:05 2026 +0200
[FLINK-39538][table] Support upsert output mode for FROM_CHANGELOG (#28164)
FROM_CHANGELOG now emits an upsert changelog (INSERT, UPDATE_AFTER, full
DELETE) when the input table is partitioned (set semantics via PARTITION BY)
and the active op_mapping maps to UPDATE_AFTER without UPDATE_BEFORE. The
partition key acts as the upsert key. In all other cases the output remains a
retract changelog.
Submitting an op_mapping with UPDATE_AFTER but no UPDATE_BEFORE without
PARTITION BY is rejected at validation time, because upsert mode requires a key.
To enable the strategy to inspect the resolved op_mapping and the input
table's partition keys, ChangelogFunction.ChangelogContext is extended with two
default methods: getArgumentValue(int, Class) and getTableSemantics(int).
Defaults return Optional.empty() to preserve source compatibility for existing
implementations.
The planner-side wrapper in FlinkChangelogModeInferenceProgram delegates
the two new methods to the underlying CallContext.
Upsert mode uses full deletes (ChangelogMode.upsert(false)) because the
runtime forwards each input delete row with all fields populated; only the
RowKind is rewritten. This matches the runtime's behavior and avoids forcing
downstream operators to reconstruct rows from state.
The upsert key derivation in FlinkRelMdUniqueKeys.getPtfUniqueKeys already
returns the partition columns when a PTF emits upsert, so no metadata changes
are needed.
---
.../docs/sql/reference/queries/changelog.md | 35 ++++++++++
flink-python/pyflink/table/table.py | 6 ++
.../apache/flink/table/api/PartitionedTable.java | 12 ++++
.../java/org/apache/flink/table/api/Table.java | 5 ++
.../functions/BuiltInFunctionDefinitions.java | 4 +-
.../flink/table/functions/ChangelogFunction.java | 26 ++++++++
.../strategies/FromChangelogTypeStrategy.java | 74 +++++++++++++++++++---
.../FromChangelogInputTypeStrategyTest.java | 36 +++++++++--
.../FlinkChangelogModeInferenceProgram.scala | 14 +++-
.../exec/stream/FromChangelogSemanticTests.java | 1 +
.../exec/stream/FromChangelogTestPrograms.java | 32 ++++++++++
.../planner/plan/stream/sql/FromChangelogTest.java | 26 ++++++++
.../planner/plan/stream/sql/FromChangelogTest.xml | 20 ++++++
.../flink/table/planner/utils/TableTestBase.scala | 35 ++++++++--
14 files changed, 302 insertions(+), 24 deletions(-)
diff --git a/docs/content/docs/sql/reference/queries/changelog.md
b/docs/content/docs/sql/reference/queries/changelog.md
index b02215129e5..3a7a97534c3 100644
--- a/docs/content/docs/sql/reference/queries/changelog.md
+++ b/docs/content/docs/sql/reference/queries/changelog.md
@@ -149,6 +149,41 @@ Prefer row semantics, when possible. `PARTITION BY` is
only necessary when downs
If you are producing an upsert table — that is, you are emitting
`UPDATE_AFTER` but no `UPDATE_BEFORE` from your CDC input stream — the
partition key you select here will be considered both the primary key and the
upsert key by the engine. Make sure the `PARTITION BY` key matches your primary
key exactly.
+#### Upsert table
+
+To generate an upsert table, two requirements must be met:
+
+* **Key partitioning**: use `PARTITION BY <key>`, where the partition key
corresponds to the unique/primary key of the dataset.
+* **Op mapping configuration**: the `op_mapping` must include `UPDATE_AFTER`
and must NOT include `UPDATE_BEFORE`.
+
+An `op_mapping` that produces `UPDATE_AFTER` without `UPDATE_BEFORE` describes
an upsert changelog and requires a key. The engine assumes that the keys
provided in the `PARTITION BY` clause function as the unique upsert keys. The
resulting output changelog becomes an upsert table keyed on these partition
columns. Each incoming row is evaluated and produces `INSERT`, `UPDATE_AFTER`,
or `DELETE` events, using the partition key as the explicit upsert key.
Therefore, if the incoming changelo [...]
+
+```sql
+-- Upsert input: INSERT / UPDATE_AFTER / DELETE only
+-- +I[id:1, op:'INSERT', name:'Alice']
+-- +I[id:2, op:'INSERT', name:'Bob']
+-- +I[id:1, op:'UPDATE_AFTER', name:'Alice2']
+-- +I[id:2, op:'DELETE', name:'Bob']
+
+SELECT * FROM FROM_CHANGELOG(
+ input => TABLE cdc_stream PARTITION BY id,
+ op_mapping => MAP[
+ 'INSERT', 'INSERT',
+ 'UPDATE_AFTER', 'UPDATE_AFTER',
+ 'DELETE', 'DELETE']
+)
+
+-- Output (upsert changelog, upsert key = id):
+-- +I[id:1, name:'Alice']
+-- +I[id:2, name:'Bob']
+-- +U[id:1, name:'Alice2']
+-- -D[id:2, name:'Bob']
+```
+
+The `FROM_CHANGELOG` PTF assumes events arrive in order. If the source itself
does not guarantee ordering for events sharing the same `PARTITION BY` key, use
[`ORDER BY`]({{< ref "docs/dev/table/functions/ptfs" >}}#ordering) to reorder
them.
+
+By default, without `PARTITION BY`, or when the active `op_mapping` includes
`UPDATE_BEFORE`, the output remains a retract changelog.
+
#### Ordering CDC events with ORDER BY
CDC streams can deliver events out of order. For example, a key's
`UPDATE_AFTER` may arrive before its matching `UPDATE_BEFORE` when events are
partitioned across upstream brokers. If the source itself does not guarantee
ordering, applying such a changelog directly produces incorrect state.
diff --git a/flink-python/pyflink/table/table.py
b/flink-python/pyflink/table/table.py
index bebf1cafa29..583167fa478 100644
--- a/flink-python/pyflink/table/table.py
+++ b/flink-python/pyflink/table/table.py
@@ -1234,6 +1234,12 @@ class Table(object):
in the mapping; pass ``error_handling => 'SKIP'`` to silently drop
those
rows instead.
+ The output is a retract changelog. To emit an upsert changelog
instead, combine
+ ``PARTITION BY`` (set semantics on the table argument) with an
``op_mapping`` that
+ maps to ``UPDATE_AFTER`` without ``UPDATE_BEFORE``. The partition key
becomes the
+ upsert key. An upsert mapping without ``PARTITION BY`` is rejected at
validation
+ time, since upsert mode requires a key.
+
Example:
::
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java
index 3f65a22075b..51086f1edfe 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/PartitionedTable.java
@@ -226,6 +226,18 @@ public interface PartitionedTable {
* <p>For row semantics (each row processed independently), use {@link
Table#fromChangelog} on
* the unpartitioned table.
*
+ * <p>Output changelog mode:
+ *
+ * <ul>
+ * <li><b>Retract</b> (default): the active {@code op_mapping} includes
{@code UPDATE_BEFORE}
+ * or no updates at all. The output possibly emits {@code INSERT},
{@code UPDATE_BEFORE},
+ * {@code UPDATE_AFTER}, and {@code DELETE}.
+ * <li><b>Upsert</b>: the {@code op_mapping} maps {@code UPDATE_AFTER}
without {@code
+ * UPDATE_BEFORE}. The output emits {@code INSERT}, {@code
UPDATE_AFTER}, and {@code
+ * DELETE}, keyed on the partition columns. An upsert mapping
without {@code PARTITION BY}
+ * is rejected at validation time, since upsert mode requires a key.
+ * </ul>
+ *
* <p>Examples:
*
* <pre>{@code
diff --git
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java
index e7f690263a4..0ffb90c5fcc 100644
---
a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java
+++
b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/Table.java
@@ -1489,6 +1489,11 @@ public interface Table extends Explainable<Table>,
Executable {
* .fromChangelog();
* }</pre>
*
+ * <p>The output is a retract changelog. To emit an upsert changelog
instead, combine {@code
+ * PARTITION BY} with an {@code op_mapping} that maps to {@code
UPDATE_AFTER} without {@code
+ * UPDATE_BEFORE}. The partition key becomes the upsert key. See {@link
+ * PartitionedTable#fromChangelog(Expression...)} for details.
+ *
* <p>Optional arguments can be passed using named expressions:
*
* <pre>{@code
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 7a25ae6058a..4bf07dc4337 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
@@ -27,7 +27,6 @@ import org.apache.flink.table.api.JsonQueryWrapper;
import org.apache.flink.table.api.JsonType;
import org.apache.flink.table.api.JsonValueOnEmptyOrError;
import org.apache.flink.table.api.TableException;
-import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.expressions.TimeIntervalUnit;
import org.apache.flink.table.expressions.TimePointUnit;
import org.apache.flink.table.expressions.ValueLiteralExpression;
@@ -39,6 +38,7 @@ import
org.apache.flink.table.types.inference.StaticArgumentTrait;
import org.apache.flink.table.types.inference.TraitCondition;
import org.apache.flink.table.types.inference.TypeStrategies;
import
org.apache.flink.table.types.inference.strategies.ArrayOfStringArgumentTypeStrategy;
+import
org.apache.flink.table.types.inference.strategies.FromChangelogTypeStrategy;
import
org.apache.flink.table.types.inference.strategies.SpecificInputTypeStrategies;
import
org.apache.flink.table.types.inference.strategies.SpecificTypeStrategies;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
@@ -854,7 +854,7 @@ public final class BuiltInFunctionDefinitions {
DataTypes.MAP(DataTypes.STRING(),
DataTypes.STRING()),
true),
StaticArgument.scalar("error_handling",
DataTypes.STRING(), true))
- .changelogModeStrategy(ctx -> ChangelogMode.all())
+
.changelogModeStrategy(FromChangelogTypeStrategy.CHANGELOG_MODE_STRATEGY)
.inputTypeStrategy(FROM_CHANGELOG_INPUT_TYPE_STRATEGY)
.outputTypeStrategy(FROM_CHANGELOG_OUTPUT_TYPE_STRATEGY)
.runtimeClass(
diff --git
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ChangelogFunction.java
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ChangelogFunction.java
index ad6e56d09d9..e9b16bff6cf 100644
---
a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ChangelogFunction.java
+++
b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/functions/ChangelogFunction.java
@@ -21,8 +21,11 @@ package org.apache.flink.table.functions;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.table.annotation.ArgumentTrait;
import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.types.RowKind;
+import java.util.Optional;
+
/**
* An extension that allows a process table function (PTF) to emit results
with changelog semantics.
*
@@ -105,5 +108,28 @@ public interface ChangelogFunction extends
FunctionDefinition {
* are required and {@link ChangelogMode#keyOnlyDeletes()} are
supported.
*/
ChangelogMode getRequiredChangelogMode();
+
+ /**
+ * Returns the resolved literal value of the scalar argument at the
given position.
+ *
+ * <p>Returns empty if the argument is absent, NULL, not a literal, or
cannot be expressed
+ * as an instance of the provided class. Conversions follow the
default conversion classes
+ * of {@link LogicalType LogicalTypes}.
+ */
+ default <T> Optional<T> getArgumentValue(int pos, Class<T> clazz) {
+ return Optional.empty();
+ }
+
+ /**
+ * Returns information about the table that has been passed to a table
argument at the given
+ * position.
+ *
+ * <p>Semantics are only available for table arguments. They expose,
for example, the
+ * partition-by columns when the table argument uses {@link
ArgumentTrait#SET_SEMANTIC_TABLE
+ * set semantics}.
+ */
+ default Optional<TableSemantics> getTableSemantics(int pos) {
+ return Optional.empty();
+ }
}
}
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 f4d54f276d7..13fa1245e62 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
@@ -22,6 +22,9 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.DataTypes.Field;
import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.functions.ChangelogFunction.ChangelogContext;
+import org.apache.flink.table.functions.ChangelogModeStrategy;
import org.apache.flink.table.functions.TableSemantics;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.inference.CallContext;
@@ -36,9 +39,11 @@ import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.Set;
+import java.util.function.IntFunction;
import java.util.stream.Collectors;
/** Type strategies for the {@code FROM_CHANGELOG} process table function. */
@@ -95,6 +100,57 @@ public final class FromChangelogTypeStrategy {
return Optional.of(DataTypes.ROW(outputFields).notNull());
};
+ //
--------------------------------------------------------------------------------------------
+ // Changelog mode inference
+ //
--------------------------------------------------------------------------------------------
+
+ /**
+ * Emits an upsert changelog when the input is partitioned (set semantics)
and the resolved
+ * {@code op_mapping} maps to {@code UPDATE_AFTER} without {@code
UPDATE_BEFORE}. In all other
+ * cases the output is a retract changelog. When upsert mode is selected,
the partition key acts
+ * as the upsert key.
+ */
+ public static final ChangelogModeStrategy CHANGELOG_MODE_STRATEGY =
+ ctx -> isUpsertConfig(ctx) ? ChangelogMode.upsert(false) :
ChangelogMode.all();
+
+ /**
+ * Returns {@code true} when the FROM_CHANGELOG call should emit an upsert
changelog: the input
+ * table is partitioned AND the resolved {@code op_mapping} contains
{@code UPDATE_AFTER}
+ * without {@code UPDATE_BEFORE}. Falls back to {@code false} when the
mapping is absent or
+ * cannot be resolved as a literal. The default mapping maps to a retract
table.
+ */
+ private static boolean isUpsertConfig(final ChangelogContext ctx) {
+ if (!isPartitioned(ctx::getTableSemantics)) {
+ return false;
+ }
+ return ctx.getArgumentValue(ARG_OP_MAPPING, Map.class)
+ .map(FromChangelogTypeStrategy::describesUpsert)
+ .orElse(false);
+ }
+
+ /**
+ * Returns {@code true} if the mapping maps {@code UPDATE_AFTER} without
{@code UPDATE_BEFORE},
+ * i.e. it describes an upsert changelog.
+ */
+ private static boolean describesUpsert(final Map<String, String> mapping) {
+ return mappingContains(mapping, UPDATE_AFTER) &&
!mappingContains(mapping, UPDATE_BEFORE);
+ }
+
+ private static boolean mappingContains(
+ final Map<String, String> mapping, final String rowKindName) {
+ return mapping.values().stream()
+ .filter(Objects::nonNull)
+ .anyMatch(v -> rowKindName.equals(v.trim()));
+ }
+
+ private static boolean isPartitioned(
+ final IntFunction<Optional<TableSemantics>> tableSemantics) {
+ return tableSemantics
+ .apply(ARG_TABLE)
+ .map(ts -> ts.partitionByColumns().length > 0)
+ .orElse(false);
+ }
+
//
--------------------------------------------------------------------------------------------
// Helpers
//
--------------------------------------------------------------------------------------------
@@ -200,17 +256,17 @@ public final class FromChangelogTypeStrategy {
if (validationError.isPresent()) {
return validationError;
}
-
- final boolean hasUpdateBefore =
- mapping.values().stream().anyMatch(v ->
UPDATE_BEFORE.equals(v.trim()));
- final boolean hasUpdateAfter =
- mapping.values().stream().anyMatch(v ->
UPDATE_AFTER.equals(v.trim()));
- if (hasUpdateAfter && !hasUpdateBefore) {
+ // 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,
- "The 'op_mapping' must include UPDATE_BEFORE for
retract mode. "
- + "Upsert mode (without UPDATE_BEFORE) is not
supported "
- + "in this version.");
+ "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();
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 322cf94a49e..70595f1c597 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
@@ -139,11 +139,33 @@ class FromChangelogInputTypeStrategyTest extends
InputTypeStrategiesTestBase {
.calledWithLiteralAt(3, "FAIL")
.expectArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE,
MAP_TYPE, STRING_TYPE),
- // Error: UPDATE_AFTER without UPDATE_BEFORE not supported
+ // Valid: upsert mapping (INSERT + UPDATE_AFTER + DELETE, no
UPDATE_BEFORE) with
+ // PARTITION BY. FROM_CHANGELOG emits an upsert changelog
keyed on the partition
+ // column.
TestSpec.forStrategy(
- "UPDATE_AFTER requires UPDATE_BEFORE",
+ "Valid upsert mapping with PARTITION BY",
FROM_CHANGELOG_INPUT_TYPE_STRATEGY)
- .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE,
MAP_TYPE)
+ .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE,
MAP_TYPE, STRING_TYPE)
+ .calledWithTableSemanticsAt(
+ 0,
+ new TableSemanticsMock(
+ TABLE_TYPE, new int[] {0}, new int[0],
-1, null))
+ .calledWithLiteralAt(1, ColumnList.of("op"))
+ .calledWithLiteralAt(
+ 2,
+ Map.of(
+ "c", "INSERT",
+ "u", "UPDATE_AFTER",
+ "d", "DELETE"))
+ .calledWithLiteralAt(3, "FAIL")
+ .expectArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE,
MAP_TYPE, STRING_TYPE),
+
+ // Error: upsert mapping (INSERT + UPDATE_AFTER + DELETE)
without PARTITION BY.
+ // Upsert mode requires a key, so the call must use set
semantics.
+ TestSpec.forStrategy(
+ "Upsert mapping without PARTITION BY rejected",
+ FROM_CHANGELOG_INPUT_TYPE_STRATEGY)
+ .calledWithArgumentTypes(TABLE_TYPE, DESCRIPTOR_TYPE,
MAP_TYPE, STRING_TYPE)
.calledWithTableSemanticsAt(0, new
TableSemanticsMock(TABLE_TYPE))
.calledWithLiteralAt(1, ColumnList.of("op"))
.calledWithLiteralAt(
@@ -152,7 +174,13 @@ class FromChangelogInputTypeStrategyTest extends
InputTypeStrategiesTestBase {
"c", "INSERT",
"u", "UPDATE_AFTER",
"d", "DELETE"))
- .expectErrorMessage("must include UPDATE_BEFORE"),
+ .calledWithLiteralAt(3, "FAIL")
+ .expectErrorMessage(
+ "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."),
// Error: Invalid error_handling mode
TestSpec.forStrategy(
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
index 6839cc1c28c..fd646a34148 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkChangelogModeInferenceProgram.scala
@@ -23,7 +23,7 @@ import
org.apache.flink.table.api.InsertConflictStrategy.ConflictBehavior
import org.apache.flink.table.api.config.ExecutionConfigOptions
import
org.apache.flink.table.api.config.ExecutionConfigOptions.UpsertMaterialize
import org.apache.flink.table.connector.ChangelogMode
-import org.apache.flink.table.functions.{BuiltInFunctionDefinition,
ChangelogFunction}
+import org.apache.flink.table.functions.{BuiltInFunctionDefinition,
ChangelogFunction, TableSemantics}
import org.apache.flink.table.functions.ChangelogFunction.ChangelogContext
import org.apache.flink.table.planner.calcite.{FlinkTypeFactory,
RexTableArgCall}
import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction
@@ -1678,8 +1678,8 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
val callContext =
function.toCallContext(udfCall, inputTimeColumns, inputChangelogModes,
outputChangelogMode)
- // Expose a simplified context to let users focus on important
characteristics.
- // If necessary, we can expose the full CallContext in the future.
+ // Expose a simplified context focused on changelog-relevant inputs:
changelog modes,
+ // resolved literal arguments, and table semantics (e.g., partition-by
columns).
new ChangelogContext {
override def getTableChangelogMode(pos: Int): ChangelogMode = {
val tableSemantics = callContext.getTableSemantics(pos).orElse(null)
@@ -1692,6 +1692,14 @@ class FlinkChangelogModeInferenceProgram extends
FlinkOptimizeProgram[StreamOpti
override def getRequiredChangelogMode: ChangelogMode = {
callContext.getOutputChangelogMode.orElse(null)
}
+
+ override def getArgumentValue[T](pos: Int, clazz: Class[T]):
java.util.Optional[T] = {
+ callContext.getArgumentValue(pos, clazz)
+ }
+
+ override def getTableSemantics(pos: Int):
java.util.Optional[TableSemantics] = {
+ callContext.getTableSemantics(pos)
+ }
}
}
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 c3922fb28ad..2892d22eda9 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
@@ -43,6 +43,7 @@ public class FromChangelogSemanticTests extends
SemanticTestBase {
FromChangelogTestPrograms.CUSTOM_OP_MAPPING,
FromChangelogTestPrograms.CUSTOM_OP_NAME,
FromChangelogTestPrograms.RETRACT_PARTITION_BY,
+ FromChangelogTestPrograms.UPSERT_PARTITION_BY,
FromChangelogTestPrograms.DELETION_FLAG_PARTITION_BY,
FromChangelogTestPrograms.SKIP_INVALID_OP_HANDLING,
FromChangelogTestPrograms.SKIP_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 a964a57bdbe..2ab92f052d7 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
@@ -200,6 +200,38 @@ public class FromChangelogTestPrograms {
+ "input => TABLE cdc_stream PARTITION BY
id)")
.build();
+ public static final TableTestProgram UPSERT_PARTITION_BY =
+ TableTestProgram.of(
+ "from-changelog-upsert-partition-by",
+ "PARTITION BY + op_mapping without UPDATE_BEFORE
produces an "
+ + "upsert changelog keyed on the partition
columns")
+ .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())
+ .runSql(
+ "INSERT INTO sink SELECT * FROM FROM_CHANGELOG("
+ + "input => TABLE cdc_stream PARTITION BY
id, "
+ + "op_mapping => MAP["
+ + "'INSERT', 'INSERT', "
+ + "'UPDATE_AFTER', 'UPDATE_AFTER', "
+ + "'DELETE', 'DELETE'])")
+ .build();
+
public static final TableTestProgram DELETION_FLAG_PARTITION_BY =
TableTestProgram.of(
"from-changelog-deletion-flag-partition-by",
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/FromChangelogTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/FromChangelogTest.java
index 17d7f64fa48..100d7b3e505 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/FromChangelogTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/FromChangelogTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.planner.plan.stream.sql;
import org.apache.flink.table.api.ExplainDetail;
import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.planner.utils.PlanKind;
import org.apache.flink.table.planner.utils.TableTestBase;
import org.apache.flink.table.planner.utils.TableTestUtil;
@@ -29,6 +30,8 @@ import org.junit.jupiter.api.Test;
import java.util.Collections;
import java.util.List;
+import scala.Enumeration;
+
/**
* Plan tests for the FROM_CHANGELOG built-in process table function. Uses
{@link
* ExplainDetail#CHANGELOG_MODE} to verify changelog mode propagation through
the plan.
@@ -71,4 +74,27 @@ public class FromChangelogTest extends TableTestBase {
"SELECT * FROM FROM_CHANGELOG(input => TABLE cdc_stream
PARTITION BY id)",
CHANGELOG_MODE);
}
+
+ @Test
+ void testUpsertPartitionBy() {
+ util.tableEnv()
+ .executeSql(
+ "CREATE TABLE cdc_stream ("
+ + " id INT,"
+ + " op STRING,"
+ + " name STRING"
+ + ") WITH ('connector' = 'values')");
+ util.doVerifyPlan(
+ "SELECT * FROM FROM_CHANGELOG("
+ + "input => TABLE cdc_stream PARTITION BY id, "
+ + "op_mapping => MAP["
+ + "'INSERT', 'INSERT', "
+ + "'UPDATE_AFTER', 'UPDATE_AFTER', "
+ + "'DELETE', 'DELETE'])",
+ new ExplainDetail[] {ExplainDetail.CHANGELOG_MODE},
+ false,
+ new Enumeration.Value[] {PlanKind.AST(), PlanKind.OPT_REL()},
+ false,
+ true);
+ }
}
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/FromChangelogTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/FromChangelogTest.xml
index 2addfe9f5ff..5e632be5788 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/FromChangelogTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/FromChangelogTest.xml
@@ -50,6 +50,26 @@ LogicalProject(id=[$0], name=[$1])
<Resource name="optimized rel plan">
<![CDATA[
ProcessTableFunction(invocation=[FROM_CHANGELOG(TABLE(#0) PARTITION BY($0),
DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT(), DEFAULT())], uid=[FROM_CHANGELOG],
select=[id,name], rowType=[RecordType(INTEGER id, VARCHAR(2147483647) name)],
changelogMode=[I,UB,UA,D])
++- Exchange(distribution=[hash[id]], changelogMode=[I])
+ +- TableSourceScan(table=[[default_catalog, default_database, cdc_stream]],
fields=[id, op, name], changelogMode=[I])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testUpsertPartitionBy">
+ <Resource name="sql">
+ <![CDATA[SELECT * FROM FROM_CHANGELOG(input => TABLE cdc_stream
PARTITION BY id, op_mapping => MAP['INSERT', 'INSERT', 'UPDATE_AFTER',
'UPDATE_AFTER', 'DELETE', 'DELETE'])]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(id=[$0], name=[$1])
++- LogicalTableFunctionScan(invocation=[FROM_CHANGELOG(TABLE(#0) PARTITION
BY($0), DEFAULT(), MAP(_UTF-16LE'INSERT':VARCHAR(12) CHARACTER SET "UTF-16LE",
_UTF-16LE'INSERT':VARCHAR(12) CHARACTER SET "UTF-16LE",
_UTF-16LE'UPDATE_AFTER':VARCHAR(12) CHARACTER SET "UTF-16LE",
_UTF-16LE'UPDATE_AFTER':VARCHAR(12) CHARACTER SET "UTF-16LE",
_UTF-16LE'DELETE':VARCHAR(12) CHARACTER SET "UTF-16LE",
_UTF-16LE'DELETE':VARCHAR(12) CHARACTER SET "UTF-16LE"), DEFAULT(), DEFAULT(),
DEFAULT())], rowType=[R [...]
+ +- LogicalProject(id=[$0], op=[$1], name=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
cdc_stream]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+ProcessTableFunction(invocation=[FROM_CHANGELOG(TABLE(#0) PARTITION BY($0),
DEFAULT(), MAP(_UTF-16LE'INSERT':VARCHAR(12) CHARACTER SET "UTF-16LE",
_UTF-16LE'INSERT':VARCHAR(12) CHARACTER SET "UTF-16LE",
_UTF-16LE'UPDATE_AFTER':VARCHAR(12) CHARACTER SET "UTF-16LE",
_UTF-16LE'UPDATE_AFTER':VARCHAR(12) CHARACTER SET "UTF-16LE",
_UTF-16LE'DELETE':VARCHAR(12) CHARACTER SET "UTF-16LE",
_UTF-16LE'DELETE':VARCHAR(12) CHARACTER SET "UTF-16LE"), DEFAULT(), DEFAULT(),
DEFAULT())], uid=[FROM_CHANGEL [...]
+- Exchange(distribution=[hash[id]], changelogMode=[I])
+- TableSourceScan(table=[[default_catalog, default_database, cdc_stream]],
fields=[id, op, name], changelogMode=[I])
]]>
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
index 3e8d5729ee7..70ad0d426d7 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/utils/TableTestBase.scala
@@ -614,7 +614,7 @@ abstract class TableTestUtilBase(test: TableTestBase,
isStreamingMode: Boolean)
/** Java-friendly overload that accepts a list of [[ExplainDetail]]s. */
def verifyRelPlan(query: String, extraDetails:
java.util.List[ExplainDetail]): Unit = {
- verifyRelPlan(query, extraDetails.asScala.toSeq: _*)
+ verifyRelPlan(query, extraDetails.asScala: _*)
}
/**
@@ -1048,6 +1048,22 @@ abstract class TableTestUtilBase(test: TableTestBase,
isStreamingMode: Boolean)
withRowType: Boolean,
expectedPlans: Array[PlanKind],
withQueryBlockAlias: Boolean): Unit = {
+ doVerifyPlan(
+ query,
+ extraDetails,
+ withRowType,
+ expectedPlans,
+ withQueryBlockAlias,
+ withUpsertKey = false)
+ }
+
+ def doVerifyPlan(
+ query: String,
+ extraDetails: Array[ExplainDetail],
+ withRowType: Boolean,
+ expectedPlans: Array[PlanKind],
+ withQueryBlockAlias: Boolean,
+ withUpsertKey: Boolean): Unit = {
val table = getTableEnv.sqlQuery(query)
val relNode = TableTestUtil.toRelNode(table)
@@ -1057,7 +1073,9 @@ abstract class TableTestUtilBase(test: TableTestBase,
isStreamingMode: Boolean)
withRowType,
expectedPlans,
() => assertEqualsOrExpand("sql", query),
- withQueryBlockAlias)
+ withQueryBlockAlias = withQueryBlockAlias,
+ withUpsertKey = withUpsertKey
+ )
}
/**
@@ -1192,7 +1210,8 @@ abstract class TableTestUtilBase(test: TableTestBase,
isStreamingMode: Boolean)
expectedPlans: Array[PlanKind],
assertSqlEqualsOrExpandFunc: () => Unit,
withQueryBlockAlias: Boolean = false,
- withDuplicateChanges: Boolean = false): Unit = {
+ withDuplicateChanges: Boolean = false,
+ withUpsertKey: Boolean = false): Unit = {
val expectedPlanKinds = new
util.HashSet[PlanKind](expectedPlans.toSeq.asJava)
@@ -1219,7 +1238,8 @@ abstract class TableTestUtilBase(test: TableTestBase,
isStreamingMode: Boolean)
optimizedRels.toArray,
extraDetails,
withRowType = withRowType,
- withDuplicateChanges = withDuplicateChanges)
+ withDuplicateChanges = withDuplicateChanges,
+ withUpsertKey = withUpsertKey)
// build optimized exec plan if `expectedPlanKinds` contains OPT_EXEC
val optimizedExecPlan = if (expectedPlanKinds.contains(PlanKind.OPT_EXEC))
{
@@ -1276,7 +1296,8 @@ abstract class TableTestUtilBase(test: TableTestBase,
isStreamingMode: Boolean)
optimizedRels: Array[RelNode],
extraDetails: Array[ExplainDetail],
withRowType: Boolean,
- withDuplicateChanges: Boolean): String = {
+ withDuplicateChanges: Boolean,
+ withUpsertKey: Boolean = false): String = {
require(optimizedRels.nonEmpty)
val explainLevel = if
(extraDetails.contains(ExplainDetail.ESTIMATED_COST)) {
SqlExplainLevel.ALL_ATTRIBUTES
@@ -1303,7 +1324,9 @@ abstract class TableTestUtilBase(test: TableTestBase,
isStreamingMode: Boolean)
detailLevel = explainLevel,
withChangelogTraits = withChangelogTraits,
withRowType = withRowType,
- withDuplicateChangesTrait = withDuplicateChanges)
+ withUpsertKey = withUpsertKey,
+ withDuplicateChangesTrait = withDuplicateChanges
+ )
}
.mkString("\n")
}
