You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2021/04/22 09:16:31 UTC
[flink] 01/03: [FLINK-19980][table-common] Add a
ChangelogMode.upsert() shortcut
This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5efca9ec1e4a035a0826e3beca2e93dc69b20c2b
Author: Timo Walther <tw...@apache.org>
AuthorDate: Mon Apr 19 08:30:39 2021 +0200
[FLINK-19980][table-common] Add a ChangelogMode.upsert() shortcut
---
.../org/apache/flink/table/connector/ChangelogMode.java | 15 +++++++++++++++
.../table/planner/factories/TestValuesTableFactory.java | 8 +-------
2 files changed, 16 insertions(+), 7 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/ChangelogMode.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/ChangelogMode.java
index c8d59a7..d8f7e2e 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/ChangelogMode.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/connector/ChangelogMode.java
@@ -38,6 +38,13 @@ public final class ChangelogMode {
private static final ChangelogMode INSERT_ONLY =
ChangelogMode.newBuilder().addContainedKind(RowKind.INSERT).build();
+ private static final ChangelogMode UPSERT =
+ ChangelogMode.newBuilder()
+ .addContainedKind(RowKind.INSERT)
+ .addContainedKind(RowKind.UPDATE_AFTER)
+ .addContainedKind(RowKind.DELETE)
+ .build();
+
private final Set<RowKind> kinds;
private ChangelogMode(Set<RowKind> kinds) {
@@ -51,6 +58,14 @@ public final class ChangelogMode {
return INSERT_ONLY;
}
+ /**
+ * Shortcut for an upsert changelog that describes idempotent updates on a key and thus does not
+ * contain {@link RowKind#UPDATE_BEFORE} rows.
+ */
+ public static ChangelogMode upsert() {
+ return UPSERT;
+ }
+
/** Builder for configuring and creating instances of {@link ChangelogMode}. */
public static Builder newBuilder() {
return new Builder();
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java
index ea90ba6..da5fc9a 100644
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java
@@ -1229,15 +1229,9 @@ public final class TestValuesTableFactory
if (isInsertOnly) {
return ChangelogMode.insertOnly();
} else {
- ChangelogMode.Builder builder = ChangelogMode.newBuilder();
if (primaryKeyIndices.length > 0) {
// can update on key, ignore UPDATE_BEFORE
- for (RowKind kind : requestedMode.getContainedKinds()) {
- if (kind != RowKind.UPDATE_BEFORE) {
- builder.addContainedKind(kind);
- }
- }
- return builder.build();
+ return ChangelogMode.upsert();
} else {
// don't have key, works in retract mode
return requestedMode;