You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by go...@apache.org on 2022/09/15 10:42:50 UTC
[flink] branch release-1.16 updated (66f996aecaf -> 390612320fa)
This is an automated email from the ASF dual-hosted git repository.
godfrey pushed a change to branch release-1.16
in repository https://gitbox.apache.org/repos/asf/flink.git
from 66f996aecaf [FLINK-29287][tests] Rewrite sql-client packaging tests
new 2f89d63d57b [FLINK-28569][table-planner] Add projectRowType to RowTypeUtils and deprecate AggCodeGenHelper#projectRowType
new bab5c93ca1f [FLINK-28569][table-planner] Move non-deterministic test functions to userDefinedScalarFunctions
new 390612320fa [FLINK-28569][table-planner] Fix SinkUpsertMaterializer that should be aware of the input upsertKey if it is not empty to prevent wrong results
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../plan/nodes/exec/batch/BatchExecSink.java | 3 +-
.../plan/nodes/exec/common/CommonExecSink.java | 25 +++-
.../plan/nodes/exec/stream/StreamExecSink.java | 12 +-
.../table/planner/plan/utils/UpsertKeyUtil.java | 66 +++++++++
.../table/planner/typeutils/RowTypeUtils.java | 35 +++++
.../codegen/agg/batch/AggCodeGenHelper.scala | 4 -
.../codegen/agg/batch/HashAggCodeGenerator.scala | 3 +-
.../codegen/agg/batch/SortAggCodeGenerator.scala | 3 +-
.../codegen/agg/batch/WindowCodeGenerator.scala | 3 +-
.../nodes/physical/stream/StreamPhysicalSink.scala | 10 +-
.../nodes/exec/stream/TableSinkJsonPlanTest.java | 36 +++++
.../planner/plan/utils/UpsertKeyUtilTest.java | 52 ++++++++
.../utils/JavaUserDefinedScalarFunctions.java | 4 +
.../table/planner/typeutils/RowTypeUtilsTest.java | 46 +++++++
.../testChangelogSource.out | 1 +
.../testUpsertSource.out | 1 +
.../testDeduplication.out | 1 +
.../ExpandJsonPlanTest_jsonplan/testExpand.out | 1 +
...tDistinctAggCalls[isMiniBatchEnabled=false].out | 1 +
...stDistinctAggCalls[isMiniBatchEnabled=true].out | 1 +
...gCallsWithGroupBy[isMiniBatchEnabled=false].out | 1 +
...ggCallsWithGroupBy[isMiniBatchEnabled=true].out | 1 +
...erDefinedAggCalls[isMiniBatchEnabled=false].out | 1 +
...serDefinedAggCalls[isMiniBatchEnabled=true].out | 1 +
.../testEventTimeTumbleWindow.out | 1 +
.../testProcTimeTumbleWindow.out | 1 +
.../testIncrementalAggregate.out | 1 +
...lAggregateWithSumCountDistinctAndRetraction.out | 1 +
.../testInnerJoinWithEqualPk.out | 1 +
...WithNonDeterministicFuncSinkWithDifferentPk.out | 147 +++++++++++++++++++++
.../utils/userDefinedScalarFunctions.scala | 53 +++++++-
.../plan/stream/sql/NonDeterministicDagTest.scala | 64 +--------
.../runtime/stream/sql/TableSinkITCase.scala | 62 +++++++++
.../operators/sink/SinkUpsertMaterializer.java | 125 +++++++++++++-----
.../operators/sink/SinkUpsertMaterializerTest.java | 102 +++++++++++---
35 files changed, 746 insertions(+), 124 deletions(-)
create mode 100644 flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/UpsertKeyUtil.java
create mode 100644 flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/UpsertKeyUtilTest.java
create mode 100644 flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSinkJsonPlanTest_jsonplan/testCdcWithNonDeterministicFuncSinkWithDifferentPk.out
[flink] 02/03: [FLINK-28569][table-planner] Move non-deterministic test functions to userDefinedScalarFunctions
Posted by go...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
godfrey pushed a commit to branch release-1.16
in repository https://gitbox.apache.org/repos/asf/flink.git
commit bab5c93ca1f839a8e3a12f0149bb9b4d0ad7f83f
Author: lincoln.lil <li...@gmail.com>
AuthorDate: Thu Sep 8 18:11:15 2022 +0800
[FLINK-28569][table-planner] Move non-deterministic test functions to userDefinedScalarFunctions
This closes #20791
(cherry picked from commit 7df115c016cab326627e4c12b2a4c449c1794e95)
---
.../utils/userDefinedScalarFunctions.scala | 53 +++++++++++++++++-
.../plan/stream/sql/NonDeterministicDagTest.scala | 64 ++--------------------
2 files changed, 58 insertions(+), 59 deletions(-)
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/utils/userDefinedScalarFunctions.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/utils/userDefinedScalarFunctions.scala
index d45a71ba40c..e1468bead66 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/utils/userDefinedScalarFunctions.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/expressions/utils/userDefinedScalarFunctions.scala
@@ -20,7 +20,8 @@ package org.apache.flink.table.planner.expressions.utils
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.table.annotation.DataTypeHint
import org.apache.flink.table.api.Types
-import org.apache.flink.table.functions.{FunctionContext, ScalarFunction}
+import org.apache.flink.table.functions.{AggregateFunction, FunctionContext, ScalarFunction, TableFunction}
+import org.apache.flink.table.planner.utils.CountAccumulator
import org.apache.flink.types.Row
import org.apache.commons.lang3.StringUtils
@@ -232,3 +233,53 @@ class SplitUDF(deterministic: Boolean) extends ScalarFunction {
override def isDeterministic: Boolean = deterministic
}
+
+@SerialVersionUID(1L)
+class TestNonDeterministicUdf extends ScalarFunction {
+ val random = new Random()
+
+ def eval(id: JLong): JLong = {
+ id + random.nextInt()
+ }
+
+ def eval(id: Int): Int = {
+ id + random.nextInt()
+ }
+
+ def eval(id: String): String = {
+ s"$id-${random.nextInt()}"
+ }
+
+ override def isDeterministic: Boolean = false
+}
+
+@SerialVersionUID(1L)
+class TestNonDeterministicUdtf extends TableFunction[String] {
+
+ val random = new Random()
+
+ def eval(id: Int): Unit = {
+ collect(s"${id + random.nextInt()}")
+ }
+
+ def eval(id: String): Unit = {
+ id.split(",").foreach(str => collect(s"$str#${random.nextInt()}"))
+ }
+
+ override def isDeterministic: Boolean = false
+}
+
+class TestNonDeterministicUdaf extends AggregateFunction[JLong, CountAccumulator] {
+
+ val random = new Random()
+
+ def accumulate(acc: CountAccumulator, in: JLong): Unit = {
+ acc.f0 += (in + random.nextInt())
+ }
+
+ override def getValue(acc: CountAccumulator): JLong = acc.f0
+
+ override def createAccumulator(): CountAccumulator = new CountAccumulator
+
+ override def isDeterministic: Boolean = false
+}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.scala
index c86bb1c07e4..dbd44247dd0 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.scala
@@ -26,10 +26,10 @@ import org.apache.flink.table.api.config.{ExecutionConfigOptions, OptimizerConfi
import org.apache.flink.table.api.config.OptimizerConfigOptions.NonDeterministicUpdateStrategy
import org.apache.flink.table.api.internal.TableEnvironmentInternal
import org.apache.flink.table.data.RowData
-import org.apache.flink.table.functions.{AggregateFunction, ScalarFunction, TableFunction}
-import org.apache.flink.table.planner.{JBoolean, JLong}
+import org.apache.flink.table.planner.JBoolean
+import org.apache.flink.table.planner.expressions.utils.{TestNonDeterministicUdaf, TestNonDeterministicUdf, TestNonDeterministicUdtf}
import org.apache.flink.table.planner.runtime.utils.JavaUserDefinedTableFunctions.StringSplit
-import org.apache.flink.table.planner.utils.{CountAccumulator, StreamTableTestUtil, TableTestBase}
+import org.apache.flink.table.planner.utils.{StreamTableTestUtil, TableTestBase}
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo
import org.apache.flink.table.sinks.UpsertStreamTableSink
import org.apache.flink.table.types.DataType
@@ -42,8 +42,6 @@ import org.junit.runners.Parameterized
import java.util
-import scala.util.Random
-
@RunWith(classOf[Parameterized])
class NonDeterministicDagTest(nonDeterministicUpdateStrategy: NonDeterministicUpdateStrategy)
extends TableTestBase {
@@ -212,9 +210,9 @@ class NonDeterministicDagTest(nonDeterministicUpdateStrategy: NonDeterministicUp
|)""".stripMargin)
// custom ND function
- util.tableEnv.createTemporaryFunction("ndFunc", TestNonDeterministicUdf)
- util.tableEnv.createTemporaryFunction("ndTableFunc", TestNonDeterministicUdtf)
- util.tableEnv.createTemporaryFunction("ndAggFunc", TestTestNonDeterministicUdaf)
+ util.tableEnv.createTemporaryFunction("ndFunc", new TestNonDeterministicUdf)
+ util.tableEnv.createTemporaryFunction("ndTableFunc", new TestNonDeterministicUdtf)
+ util.tableEnv.createTemporaryFunction("ndAggFunc", new TestNonDeterministicUdaf)
// deterministic table function
util.tableEnv.createTemporaryFunction("str_split", new StringSplit())
}
@@ -1561,56 +1559,6 @@ class NonDeterministicDagTest(nonDeterministicUpdateStrategy: NonDeterministicUp
util.verifyExecPlan(stmtSet)
}
- @SerialVersionUID(1L)
- object TestNonDeterministicUdf extends ScalarFunction {
- val random = new Random()
-
- def eval(id: JLong): JLong = {
- id + random.nextInt()
- }
-
- def eval(id: Int): Int = {
- id + random.nextInt()
- }
-
- def eval(id: String): String = {
- s"$id-${random.nextInt()}"
- }
-
- override def isDeterministic: Boolean = false
- }
-
- @SerialVersionUID(1L)
- object TestNonDeterministicUdtf extends TableFunction[String] {
-
- val random = new Random()
-
- def eval(id: Int): Unit = {
- collect(s"${id + random.nextInt()}")
- }
-
- def eval(id: String): Unit = {
- id.split(",").foreach(str => collect(s"$str#${random.nextInt()}"))
- }
-
- override def isDeterministic: Boolean = false
- }
-
- object TestTestNonDeterministicUdaf extends AggregateFunction[JLong, CountAccumulator] {
-
- val random = new Random()
-
- def accumulate(acc: CountAccumulator, in: JLong): Unit = {
- acc.f0 += (in + random.nextInt())
- }
-
- override def getValue(acc: CountAccumulator): JLong = acc.f0
-
- override def createAccumulator(): CountAccumulator = new CountAccumulator
-
- override def isDeterministic: Boolean = false
- }
-
/**
* This upsert test sink does support getting primary key from table schema. We defined a similar
* test sink here not using existing {@link TestingUpsertTableSink} in {@link StreamTestSink}
[flink] 03/03: [FLINK-28569][table-planner] Fix SinkUpsertMaterializer that should be aware of the input upsertKey if it is not empty to prevent wrong results
Posted by go...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
godfrey pushed a commit to branch release-1.16
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 390612320fa9be297d9eed1f4b75f8ba2ec83c40
Author: lincoln.lil <li...@gmail.com>
AuthorDate: Thu Sep 8 21:50:10 2022 +0800
[FLINK-28569][table-planner] Fix SinkUpsertMaterializer that should be aware of the input upsertKey if it is not empty to prevent wrong results
This closes #20791
(cherry picked from commit bff1fc2782e04598275ab05177b89df38f891c94)
---
.../plan/nodes/exec/batch/BatchExecSink.java | 3 +-
.../plan/nodes/exec/common/CommonExecSink.java | 25 +++-
.../plan/nodes/exec/stream/StreamExecSink.java | 12 +-
.../table/planner/plan/utils/UpsertKeyUtil.java | 66 +++++++++
.../nodes/physical/stream/StreamPhysicalSink.scala | 10 +-
.../nodes/exec/stream/TableSinkJsonPlanTest.java | 36 +++++
.../planner/plan/utils/UpsertKeyUtilTest.java | 52 ++++++++
.../utils/JavaUserDefinedScalarFunctions.java | 4 +
.../testChangelogSource.out | 1 +
.../testUpsertSource.out | 1 +
.../testDeduplication.out | 1 +
.../ExpandJsonPlanTest_jsonplan/testExpand.out | 1 +
...tDistinctAggCalls[isMiniBatchEnabled=false].out | 1 +
...stDistinctAggCalls[isMiniBatchEnabled=true].out | 1 +
...gCallsWithGroupBy[isMiniBatchEnabled=false].out | 1 +
...ggCallsWithGroupBy[isMiniBatchEnabled=true].out | 1 +
...erDefinedAggCalls[isMiniBatchEnabled=false].out | 1 +
...serDefinedAggCalls[isMiniBatchEnabled=true].out | 1 +
.../testEventTimeTumbleWindow.out | 1 +
.../testProcTimeTumbleWindow.out | 1 +
.../testIncrementalAggregate.out | 1 +
...lAggregateWithSumCountDistinctAndRetraction.out | 1 +
.../testInnerJoinWithEqualPk.out | 1 +
...WithNonDeterministicFuncSinkWithDifferentPk.out | 147 +++++++++++++++++++++
.../runtime/stream/sql/TableSinkITCase.scala | 62 +++++++++
.../operators/sink/SinkUpsertMaterializer.java | 125 +++++++++++++-----
.../operators/sink/SinkUpsertMaterializerTest.java | 102 +++++++++++---
27 files changed, 601 insertions(+), 58 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSink.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSink.java
index a6202871816..c9068656a51 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSink.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecSink.java
@@ -71,6 +71,7 @@ public class BatchExecSink extends CommonExecSink implements BatchExecNode<Objec
inputTransform,
tableSink,
-1,
- false);
+ false,
+ null);
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
index 71f8184df2c..7f978565060 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
@@ -61,6 +61,7 @@ import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecNode;
import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
import org.apache.flink.table.planner.plan.nodes.exec.utils.TransformationMetadata;
import org.apache.flink.table.planner.plan.utils.KeySelectorUtil;
+import org.apache.flink.table.planner.typeutils.RowTypeUtils;
import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext;
import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser;
import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
@@ -141,7 +142,8 @@ public abstract class CommonExecSink extends ExecNodeBase<Object>
Transformation<RowData> inputTransform,
DynamicTableSink tableSink,
int rowtimeFieldIndex,
- boolean upsertMaterialize) {
+ boolean upsertMaterialize,
+ int[] inputUpsertKey) {
final ResolvedSchema schema = tableSinkSpec.getContextResolvedTable().getResolvedSchema();
final SinkRuntimeProvider runtimeProvider =
tableSink.getSinkRuntimeProvider(new SinkRuntimeProviderContext(isBounded));
@@ -193,7 +195,8 @@ public abstract class CommonExecSink extends ExecNodeBase<Object>
sinkParallelism,
config,
classLoader,
- physicalRowType);
+ physicalRowType,
+ inputUpsertKey);
}
return (Transformation<Object>)
@@ -402,16 +405,28 @@ public abstract class CommonExecSink extends ExecNodeBase<Object>
int sinkParallelism,
ExecNodeConfig config,
ClassLoader classLoader,
- RowType physicalRowType) {
- GeneratedRecordEqualiser equaliser =
+ RowType physicalRowType,
+ int[] inputUpsertKey) {
+ final GeneratedRecordEqualiser rowEqualiser =
new EqualiserCodeGenerator(physicalRowType, classLoader)
.generateRecordEqualiser("SinkMaterializeEqualiser");
+ final GeneratedRecordEqualiser upsertKeyEqualiser =
+ inputUpsertKey == null
+ ? null
+ : new EqualiserCodeGenerator(
+ RowTypeUtils.projectRowType(
+ physicalRowType, inputUpsertKey),
+ classLoader)
+ .generateRecordEqualiser("SinkMaterializeUpsertKeyEqualiser");
+
SinkUpsertMaterializer operator =
new SinkUpsertMaterializer(
StateConfigUtil.createTtlConfig(
config.get(ExecutionConfigOptions.IDLE_STATE_RETENTION).toMillis()),
InternalSerializers.create(physicalRowType),
- equaliser);
+ rowEqualiser,
+ upsertKeyEqualiser,
+ inputUpsertKey);
final String[] fieldNames = physicalRowType.getFieldNames().toArray(new String[0]);
final List<String> pkFieldNames =
Arrays.stream(primaryKeys)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java
index 1bd7ad0b22b..6932c1a951b 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecSink.java
@@ -74,6 +74,7 @@ public class StreamExecSink extends CommonExecSink implements StreamExecNode<Obj
public static final String FIELD_NAME_INPUT_CHANGELOG_MODE = "inputChangelogMode";
public static final String FIELD_NAME_REQUIRE_UPSERT_MATERIALIZE = "requireUpsertMaterialize";
+ public static final String FIELD_NAME_INPUT_UPSERT_KEY = "inputUpsertKey";
@JsonProperty(FIELD_NAME_INPUT_CHANGELOG_MODE)
private final ChangelogMode inputChangelogMode;
@@ -82,6 +83,10 @@ public class StreamExecSink extends CommonExecSink implements StreamExecNode<Obj
@JsonInclude(JsonInclude.Include.NON_DEFAULT)
private final boolean upsertMaterialize;
+ @JsonProperty(FIELD_NAME_INPUT_UPSERT_KEY)
+ @JsonInclude(JsonInclude.Include.NON_DEFAULT)
+ private final int[] inputUpsertKey;
+
public StreamExecSink(
ReadableConfig tableConfig,
DynamicTableSinkSpec tableSinkSpec,
@@ -89,6 +94,7 @@ public class StreamExecSink extends CommonExecSink implements StreamExecNode<Obj
InputProperty inputProperty,
LogicalType outputType,
boolean upsertMaterialize,
+ int[] inputUpsertKey,
String description) {
this(
ExecNodeContext.newNodeId(),
@@ -99,6 +105,7 @@ public class StreamExecSink extends CommonExecSink implements StreamExecNode<Obj
Collections.singletonList(inputProperty),
outputType,
upsertMaterialize,
+ inputUpsertKey,
description);
}
@@ -112,6 +119,7 @@ public class StreamExecSink extends CommonExecSink implements StreamExecNode<Obj
@JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties,
@JsonProperty(FIELD_NAME_OUTPUT_TYPE) LogicalType outputType,
@JsonProperty(FIELD_NAME_REQUIRE_UPSERT_MATERIALIZE) boolean upsertMaterialize,
+ @JsonProperty(FIELD_NAME_INPUT_UPSERT_KEY) int[] inputUpsertKey,
@JsonProperty(FIELD_NAME_DESCRIPTION) String description) {
super(
id,
@@ -125,6 +133,7 @@ public class StreamExecSink extends CommonExecSink implements StreamExecNode<Obj
description);
this.inputChangelogMode = inputChangelogMode;
this.upsertMaterialize = upsertMaterialize;
+ this.inputUpsertKey = inputUpsertKey;
}
@SuppressWarnings("unchecked")
@@ -171,6 +180,7 @@ public class StreamExecSink extends CommonExecSink implements StreamExecNode<Obj
inputTransform,
tableSink,
rowtimeFieldIndex,
- upsertMaterialize);
+ upsertMaterialize,
+ inputUpsertKey);
}
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/UpsertKeyUtil.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/UpsertKeyUtil.java
new file mode 100644
index 00000000000..3e054f6793f
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/UpsertKeyUtil.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.utils;
+
+import org.apache.calcite.util.ImmutableBitSet;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import java.util.Set;
+
+/**
+ * Utility for upsertKey which represented as a Set of {@link
+ * org.apache.calcite.util.ImmutableBitSet}.
+ */
+public class UpsertKeyUtil {
+
+ /**
+ * Returns the smallest key of given upsert keys. The rule of 'small' is an upsert key
+ * represented by {@link ImmutableBitSet} has smaller cardinality or has a smaller leading
+ * element when the same cardinality. E.g., '{0,1}' is smaller than '{0,1,2}' and '{0,1}' is
+ * smaller than '{0,2}'.
+ *
+ * @param upsertKeys input upsert keys
+ * @return the smallest key
+ */
+ @Nonnull
+ public static int[] getSmallestKey(@Nullable Set<ImmutableBitSet> upsertKeys) {
+ if (null == upsertKeys || upsertKeys.isEmpty()) {
+ return new int[0];
+ }
+ return upsertKeys.stream()
+ .map(ImmutableBitSet::toArray)
+ .reduce(
+ (k1, k2) -> {
+ if (k1.length < k2.length) {
+ return k1;
+ }
+ if (k1.length == k2.length) {
+ for (int index = 0; index < k1.length; index++) {
+ if (k1[index] < k2[index]) {
+ return k1;
+ }
+ }
+ }
+ return k2;
+ })
+ .get();
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalSink.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalSink.scala
index c64c41721ca..bffb097d6d7 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalSink.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalSink.scala
@@ -21,11 +21,12 @@ import org.apache.flink.table.catalog.ContextResolvedTable
import org.apache.flink.table.connector.sink.DynamicTableSink
import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.plan.abilities.sink.SinkAbilitySpec
+import org.apache.flink.table.planner.plan.metadata.FlinkRelMetadataQuery
import org.apache.flink.table.planner.plan.nodes.calcite.Sink
import org.apache.flink.table.planner.plan.nodes.exec.{ExecNode, InputProperty}
import org.apache.flink.table.planner.plan.nodes.exec.spec.DynamicTableSinkSpec
import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink
-import org.apache.flink.table.planner.plan.utils.ChangelogPlanUtils
+import org.apache.flink.table.planner.plan.utils.{ChangelogPlanUtils, UpsertKeyUtil}
import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
@@ -81,6 +82,12 @@ class StreamPhysicalSink(
val tableSinkSpec =
new DynamicTableSinkSpec(contextResolvedTable, util.Arrays.asList(abilitySpecs: _*))
tableSinkSpec.setTableSink(tableSink)
+ // no need to call getUpsertKeysInKeyGroupRange here because there's no exchange before sink,
+ // and only add exchange in exec sink node.
+ val inputUpsertKeys = FlinkRelMetadataQuery
+ .reuseOrCreate(cluster.getMetadataQuery)
+ .getUpsertKeys(inputRel)
+
new StreamExecSink(
unwrapTableConfig(this),
tableSinkSpec,
@@ -88,6 +95,7 @@ class StreamPhysicalSink(
InputProperty.DEFAULT,
FlinkTypeFactory.toLogicalRowType(getRowType),
upsertMaterialize,
+ UpsertKeyUtil.getSmallestKey(inputUpsertKeys),
getRelDetailedDescription)
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSinkJsonPlanTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSinkJsonPlanTest.java
index e161bddd8f6..a467493499d 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSinkJsonPlanTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSinkJsonPlanTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.planner.plan.nodes.exec.stream;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.table.planner.runtime.utils.JavaUserDefinedScalarFunctions;
import org.apache.flink.table.planner.utils.StreamTableTestUtil;
import org.apache.flink.table.planner.utils.TableTestBase;
@@ -91,4 +92,39 @@ public class TableSinkJsonPlanTest extends TableTestBase {
tEnv.executeSql(sinkTableDdl);
util.verifyJsonPlan("insert into MySink select * from MyTable");
}
+
+ @Test
+ public void testCdcWithNonDeterministicFuncSinkWithDifferentPk() {
+ tEnv.createTemporaryFunction(
+ "ndFunc", new JavaUserDefinedScalarFunctions.NonDeterministicUdf());
+
+ String cdcDdl =
+ "CREATE TABLE users (\n"
+ + " user_id STRING,\n"
+ + " user_name STRING,\n"
+ + " email STRING,\n"
+ + " balance DECIMAL(18,2),\n"
+ + " primary key (user_id) not enforced\n"
+ + ") WITH (\n"
+ + " 'connector' = 'values',\n"
+ + " 'changelog-mode' = 'I,UA,UB,D'\n"
+ + ")";
+
+ String sinkTableDdl =
+ "CREATE TABLE sink (\n"
+ + " user_id STRING,\n"
+ + " user_name STRING,\n"
+ + " email STRING,\n"
+ + " balance DECIMAL(18,2),\n"
+ + " PRIMARY KEY(email) NOT ENFORCED\n"
+ + ") WITH(\n"
+ + " 'connector' = 'values',\n"
+ + " 'sink-insert-only' = 'false'\n"
+ + ")";
+ tEnv.executeSql(cdcDdl);
+ tEnv.executeSql(sinkTableDdl);
+
+ util.verifyJsonPlan(
+ "insert into sink select user_id, ndFunc(user_name), email, balance from users");
+ }
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/UpsertKeyUtilTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/UpsertKeyUtilTest.java
new file mode 100644
index 00000000000..4ddfadfc1f9
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/utils/UpsertKeyUtilTest.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.plan.utils;
+
+import org.apache.calcite.util.ImmutableBitSet;
+import org.junit.Test;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link UpsertKeyUtil}. */
+public class UpsertKeyUtilTest {
+ private final int[] emptyKey = new int[0];
+
+ @Test
+ public void testSmallestKey() {
+ assertThat(UpsertKeyUtil.getSmallestKey(null)).isEqualTo(emptyKey);
+ assertThat(UpsertKeyUtil.getSmallestKey(new HashSet<>())).isEqualTo(emptyKey);
+
+ ImmutableBitSet smallestKey = ImmutableBitSet.of(0, 1);
+ ImmutableBitSet middleKey = ImmutableBitSet.of(0, 2);
+ ImmutableBitSet longKey = ImmutableBitSet.of(0, 1, 2);
+
+ Set<ImmutableBitSet> upsertKeys = new HashSet<>();
+ upsertKeys.add(smallestKey);
+ upsertKeys.add(middleKey);
+ assertThat(UpsertKeyUtil.getSmallestKey(upsertKeys)).isEqualTo(smallestKey.toArray());
+
+ upsertKeys.clear();
+ upsertKeys.add(smallestKey);
+ upsertKeys.add(longKey);
+ assertThat(UpsertKeyUtil.getSmallestKey(upsertKeys)).isEqualTo(smallestKey.toArray());
+ }
+}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedScalarFunctions.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedScalarFunctions.java
index 73c52d16e4d..5db0abf9cc8 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedScalarFunctions.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/runtime/utils/JavaUserDefinedScalarFunctions.java
@@ -131,6 +131,10 @@ public class JavaUserDefinedScalarFunctions {
return v + random.nextInt();
}
+ public String eval(String v) {
+ return v + "-" + random.nextInt();
+ }
+
@Override
public boolean isDeterministic() {
return false;
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogSourceJsonPlanTest_jsonplan/testChangelogSource.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogSourceJsonPlanTest_jsonplan/testChangelogSource.out
index 1f4ca9d8dd7..46c9b032843 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogSourceJsonPlanTest_jsonplan/testChangelogSource.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogSourceJsonPlanTest_jsonplan/testChangelogSource.out
@@ -133,6 +133,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`a` BIGINT NOT NULL, `b` INT NOT NULL>",
+ "inputUpsertKey" : [ 0, 1 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a, b])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogSourceJsonPlanTest_jsonplan/testUpsertSource.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogSourceJsonPlanTest_jsonplan/testUpsertSource.out
index 81bd0cc7972..e29a2507619 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogSourceJsonPlanTest_jsonplan/testUpsertSource.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogSourceJsonPlanTest_jsonplan/testUpsertSource.out
@@ -121,6 +121,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`a` BIGINT NOT NULL, `b` INT NOT NULL>",
+ "inputUpsertKey" : [ 0, 1 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a, b])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicationJsonPlanTest_jsonplan/testDeduplication.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicationJsonPlanTest_jsonplan/testDeduplication.out
index f2c4a58820d..0c219aae929 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicationJsonPlanTest_jsonplan/testDeduplication.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/DeduplicationJsonPlanTest_jsonplan/testDeduplication.out
@@ -272,6 +272,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`order_id` BIGINT, `user` VARCHAR(2147483647), `product` VARCHAR(2147483647), `order_time` TIMESTAMP(3)>",
+ "inputUpsertKey" : [ 2 ],
"description" : "Sink(table=[default_catalog.default_database.sink], fields=[order_id, user, product, order_time])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ExpandJsonPlanTest_jsonplan/testExpand.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ExpandJsonPlanTest_jsonplan/testExpand.out
index 3982472c711..8262813cced 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ExpandJsonPlanTest_jsonplan/testExpand.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/ExpandJsonPlanTest_jsonplan/testExpand.out
@@ -365,6 +365,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`a` BIGINT, `$f1` BIGINT NOT NULL, `$f2` VARCHAR(2147483647)>",
+ "inputUpsertKey" : [ 0 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a, $f1, $f2])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=false].out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=false].out
index 52aad805789..1af9ee2c41c 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=false].out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=false].out
@@ -297,6 +297,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`d` BIGINT, `cnt_a1` BIGINT, `cnt_a2` BIGINT, `sum_a` BIGINT, `sum_b` INT, `avg_b` DOUBLE, `cnt_c` BIGINT>",
+ "inputUpsertKey" : [ 0 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[d, cnt_a1, cnt_a2, sum_a, sum_b, avg_b, cnt_c])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=true].out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=true].out
index bfef96f23e1..12b2452fbef 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=true].out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testDistinctAggCalls[isMiniBatchEnabled=true].out
@@ -545,6 +545,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`d` BIGINT, `cnt_a1` BIGINT, `cnt_a2` BIGINT, `sum_a` BIGINT, `sum_b` INT, `avg_b` DOUBLE, `cnt_c` BIGINT>",
+ "inputUpsertKey" : [ 0 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[d, cnt_a1, cnt_a2, sum_a, sum_b, avg_b, cnt_c])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testSimpleAggCallsWithGroupBy[isMiniBatchEnabled=false].out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testSimpleAggCallsWithGroupBy[isMiniBatchEnabled=false].out
index 4ca7313f0a2..f564a72dbad 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testSimpleAggCallsWithGroupBy[isMiniBatchEnabled=false].out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testSimpleAggCallsWithGroupBy[isMiniBatchEnabled=false].out
@@ -240,6 +240,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`b` BIGINT, `cnt_a` BIGINT, `max_b` BIGINT, `min_c` VARCHAR(2147483647)>",
+ "inputUpsertKey" : [ 0 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[b, cnt_a, max_b, min_c])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testSimpleAggCallsWithGroupBy[isMiniBatchEnabled=true].out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testSimpleAggCallsWithGroupBy[isMiniBatchEnabled=true].out
index b31d031ffd0..bfbee492d41 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testSimpleAggCallsWithGroupBy[isMiniBatchEnabled=true].out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testSimpleAggCallsWithGroupBy[isMiniBatchEnabled=true].out
@@ -305,6 +305,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`b` BIGINT, `cnt_a` BIGINT, `max_b` BIGINT, `min_c` VARCHAR(2147483647)>",
+ "inputUpsertKey" : [ 0 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[b, cnt_a, max_b, min_c])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testUserDefinedAggCalls[isMiniBatchEnabled=false].out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testUserDefinedAggCalls[isMiniBatchEnabled=false].out
index 1ce190c4f7c..41573678da6 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testUserDefinedAggCalls[isMiniBatchEnabled=false].out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testUserDefinedAggCalls[isMiniBatchEnabled=false].out
@@ -233,6 +233,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`b` BIGINT, `a1` BIGINT, `a2` BIGINT, `a3` BIGINT, `c1` BIGINT>",
+ "inputUpsertKey" : [ 0 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[b, a1, a2, a3, c1])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testUserDefinedAggCalls[isMiniBatchEnabled=true].out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testUserDefinedAggCalls[isMiniBatchEnabled=true].out
index 201e9547f11..36478bc1dd5 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testUserDefinedAggCalls[isMiniBatchEnabled=true].out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupAggregateJsonPlanTest_jsonplan/testUserDefinedAggCalls[isMiniBatchEnabled=true].out
@@ -249,6 +249,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`b` BIGINT, `a1` BIGINT, `a2` BIGINT, `a3` BIGINT, `c1` BIGINT>",
+ "inputUpsertKey" : [ 0 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[b, a1, a2, a3, c1])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
index d30a69a2d5f..ab47bb954c1 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testEventTimeTumbleWindow.out
@@ -466,6 +466,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`b` BIGINT, `window_start` TIMESTAMP(3) NOT NULL, `window_end` TIMESTAMP(3) NOT NULL, `EXPR$3` BIGINT NOT NULL, `EXPR$4` INT, `EXPR$5` BIGINT NOT NULL, `EXPR$6` VARCHAR(2147483647)>",
+ "inputUpsertKey" : [ 0, 1 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[b, window_start, window_end, EXPR$3, EXPR$4, EXPR$5, EXPR$6])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testProcTimeTumbleWindow.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testProcTimeTumbleWindow.out
index bc5fd3fa7c0..52af443ff67 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testProcTimeTumbleWindow.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/GroupWindowAggregateJsonPlanTest_jsonplan/testProcTimeTumbleWindow.out
@@ -451,6 +451,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`b` BIGINT, `window_end` TIMESTAMP(3) NOT NULL, `EXPR$2` BIGINT NOT NULL>",
+ "inputUpsertKey" : [ 0, 1 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[b, window_end, EXPR$2])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/IncrementalAggregateJsonPlanTest_jsonplan/testIncrementalAggregate.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/IncrementalAggregateJsonPlanTest_jsonplan/testIncrementalAggregate.out
index a2bcf705ab2..2f0585073c6 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/IncrementalAggregateJsonPlanTest_jsonplan/testIncrementalAggregate.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/IncrementalAggregateJsonPlanTest_jsonplan/testIncrementalAggregate.out
@@ -327,6 +327,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`a` BIGINT, `$f1` BIGINT NOT NULL>",
+ "inputUpsertKey" : [ 0 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a, $f1])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/IncrementalAggregateJsonPlanTest_jsonplan/testIncrementalAggregateWithSumCountDistinctAndRetraction.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/IncrementalAggregateJsonPlanTest_jsonplan/testIncrementalAggregateWithSumCountDistinctAndRetraction.out
index c4a0f219d8e..517dc8bc130 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/IncrementalAggregateJsonPlanTest_jsonplan/testIncrementalAggregateWithSumCountDistinctAndRetraction.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/IncrementalAggregateJsonPlanTest_jsonplan/testIncrementalAggregateWithSumCountDistinctAndRetraction.out
@@ -485,6 +485,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`b` BIGINT NOT NULL, `$f1` INT NOT NULL, `$f2` BIGINT NOT NULL, `$f3` BIGINT NOT NULL>",
+ "inputUpsertKey" : [ 0 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[b, $f1, $f2, $f3])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/JoinJsonPlanTest_jsonplan/testInnerJoinWithEqualPk.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/JoinJsonPlanTest_jsonplan/testInnerJoinWithEqualPk.out
index 141b22ea160..a230c7d8695 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/JoinJsonPlanTest_jsonplan/testInnerJoinWithEqualPk.out
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/JoinJsonPlanTest_jsonplan/testInnerJoinWithEqualPk.out
@@ -242,6 +242,7 @@
"priority" : 0
} ],
"outputType" : "ROW<`a1` INT, `b1` INT>",
+ "inputUpsertKey" : [ 0 ],
"description" : "Sink(table=[default_catalog.default_database.MySink], fields=[a1, b1])"
} ],
"edges" : [ {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSinkJsonPlanTest_jsonplan/testCdcWithNonDeterministicFuncSinkWithDifferentPk.out b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSinkJsonPlanTest_jsonplan/testCdcWithNonDeterministicFuncSinkWithDifferentPk.out
new file mode 100644
index 00000000000..40bedb2ea92
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/nodes/exec/stream/TableSinkJsonPlanTest_jsonplan/testCdcWithNonDeterministicFuncSinkWithDifferentPk.out
@@ -0,0 +1,147 @@
+{
+ "flinkVersion" : "",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`users`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_id",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "user_name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "email",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "balance",
+ "dataType" : "DECIMAL(18, 2)"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_-147132882",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "user_id" ]
+ }
+ },
+ "partitionKeys" : [ ],
+ "options" : {
+ "connector" : "values",
+ "changelog-mode" : "I,UA,UB,D"
+ }
+ }
+ }
+ },
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647) NOT NULL, `user_name` VARCHAR(2147483647), `email` VARCHAR(2147483647), `balance` DECIMAL(18, 2)>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, users]], fields=[user_id, user_name, email, balance])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 2,
+ "type" : "stream-exec-calc_1",
+ "projection" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 0,
+ "type" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "kind" : "CALL",
+ "catalogName" : "`default_catalog`.`default_database`.`ndFunc`",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "VARCHAR(2147483647)"
+ } ],
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 2,
+ "type" : "VARCHAR(2147483647)"
+ }, {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 3,
+ "type" : "DECIMAL(18, 2)"
+ } ],
+ "condition" : null,
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647) NOT NULL, `EXPR$1` VARCHAR(2147483647), `email` VARCHAR(2147483647), `balance` DECIMAL(18, 2)>",
+ "description" : "Calc(select=[user_id, ndFunc(user_name) AS EXPR$1, email, balance])"
+ }, {
+ "id" : 3,
+ "type" : "stream-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "user_id",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "user_name",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "email",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "balance",
+ "dataType" : "DECIMAL(18, 2)"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_96619451",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "email" ]
+ }
+ },
+ "partitionKeys" : [ ],
+ "options" : {
+ "sink-insert-only" : "false",
+ "connector" : "values"
+ }
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER", "DELETE" ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`user_id` VARCHAR(2147483647) NOT NULL, `EXPR$1` VARCHAR(2147483647), `email` VARCHAR(2147483647), `balance` DECIMAL(18, 2)>",
+ "requireUpsertMaterialize" : true,
+ "inputUpsertKey" : [ 0 ],
+ "description" : "Sink(table=[default_catalog.default_database.sink], fields=[user_id, EXPR$1, email, balance], upsertMaterialize=[true])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 3,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/TableSinkITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/TableSinkITCase.scala
index 75b0d43706e..5dc7bc70e53 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/TableSinkITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/TableSinkITCase.scala
@@ -17,6 +17,7 @@
*/
package org.apache.flink.table.planner.runtime.stream.sql
+import org.apache.flink.table.planner.expressions.utils.TestNonDeterministicUdf
import org.apache.flink.table.planner.factories.TestValuesTableFactory
import org.apache.flink.table.planner.runtime.utils._
import org.apache.flink.table.planner.runtime.utils.BatchTestBase.row
@@ -73,6 +74,21 @@ class TableSinkITCase(mode: StateBackendMode) extends StreamingWithStateTestBase
| 'data-id' = '$peopleDataId'
|)
|""".stripMargin)
+
+ val userDataId = TestValuesTableFactory.registerData(TestData.userChangelog)
+ tEnv.executeSql(s"""
+ |CREATE TABLE users (
+ | user_id STRING,
+ | user_name STRING,
+ | email STRING,
+ | balance DECIMAL(18,2),
+ | primary key (user_id) not enforced
+ |) WITH (
+ | 'connector' = 'values',
+ | 'data-id' = '$userDataId',
+ | 'changelog-mode' = 'I,UA,UB,D'
+ |)
+ |""".stripMargin)
}
@Test
@@ -157,6 +173,52 @@ class TableSinkITCase(mode: StateBackendMode) extends StreamingWithStateTestBase
assertEquals(expected.sorted, result.sorted)
}
+ @Test
+ def testChangelogSourceWithNonDeterministicFuncSinkWithDifferentPk(): Unit = {
+ tEnv.createTemporaryFunction("ndFunc", new TestNonDeterministicUdf)
+ tEnv.executeSql("""
+ |CREATE TABLE sink_with_pk (
+ | user_id STRING,
+ | user_name STRING,
+ | email STRING,
+ | balance DECIMAL(18,2),
+ | PRIMARY KEY(email) NOT ENFORCED
+ |) WITH(
+ | 'connector' = 'values',
+ | 'sink-insert-only' = 'false'
+ |)
+ |""".stripMargin)
+
+ tEnv
+ .executeSql(s"""
+ |insert into sink_with_pk
+ |select user_id, SPLIT_INDEX(ndFunc(user_name), '-', 0), email, balance
+ |from users
+ |""".stripMargin)
+ .await()
+
+ val result = TestValuesTableFactory.getResults("sink_with_pk")
+ val expected = List(
+ "+I[user1, Tom, tom123@gmail.com, 8.10]",
+ "+I[user3, Bailey, bailey@qq.com, 9.99]",
+ "+I[user4, Tina, tina@gmail.com, 11.30]")
+ assertEquals(expected.sorted, result.sorted)
+
+ val rawResult = TestValuesTableFactory.getRawResults("sink_with_pk")
+ val expectedRaw = List(
+ "+I[user1, Tom, tom@gmail.com, 10.02]",
+ "+I[user2, Jack, jack@hotmail.com, 71.20]",
+ "-D[user1, Tom, tom@gmail.com, 10.02]",
+ "+I[user1, Tom, tom123@gmail.com, 8.10]",
+ "+I[user3, Bailey, bailey@gmail.com, 9.99]",
+ "-D[user2, Jack, jack@hotmail.com, 71.20]",
+ "+I[user4, Tina, tina@gmail.com, 11.30]",
+ "-D[user3, Bailey, bailey@gmail.com, 9.99]",
+ "+I[user3, Bailey, bailey@qq.com, 9.99]"
+ )
+ assertEquals(expectedRaw, rawResult.toList)
+ }
+
@Test
def testInsertPartColumn(): Unit = {
tEnv.executeSql("""
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/SinkUpsertMaterializer.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/SinkUpsertMaterializer.java
index 0d18cb4c540..8ba4b792e2b 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/SinkUpsertMaterializer.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/sink/SinkUpsertMaterializer.java
@@ -27,14 +27,19 @@ import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.utils.ProjectedRowData;
import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser;
import org.apache.flink.table.runtime.generated.RecordEqualiser;
import org.apache.flink.table.runtime.operators.TableStreamOperator;
import org.apache.flink.types.RowKind;
+import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
@@ -66,29 +71,59 @@ public class SinkUpsertMaterializer extends TableStreamOperator<RowData>
+ "You can increase the state ttl to avoid this.";
private final StateTtlConfig ttlConfig;
+ private final GeneratedRecordEqualiser generatedRecordEqualiser;
+ private final GeneratedRecordEqualiser generatedUpsertKeyEqualiser;
private final TypeSerializer<RowData> serializer;
- private final GeneratedRecordEqualiser generatedEqualiser;
+ private final int[] inputUpsertKey;
+ private final boolean hasUpsertKey;
+ // The equaliser here may behaviors differently due to hasUpsertKey:
+ // if true: the equaliser only compares the upsertKey (a projected row data)
+ // if false: the equaliser compares the complete row
private transient RecordEqualiser equaliser;
+
// Buffer of emitted insertions on which deletions will be applied first.
// The row kind might be +I or +U and will be ignored when applying the deletion.
private transient ValueState<List<RowData>> state;
private transient TimestampedCollector<RowData> collector;
+ // Reused ProjectedRowData for comparing upsertKey if hasUpsertKey.
+ private transient ProjectedRowData upsertKeyProjectedRow1;
+ private transient ProjectedRowData upsertKeyProjectedRow2;
+
public SinkUpsertMaterializer(
StateTtlConfig ttlConfig,
TypeSerializer<RowData> serializer,
- GeneratedRecordEqualiser generatedEqualiser) {
+ GeneratedRecordEqualiser generatedRecordEqualiser,
+ @Nullable GeneratedRecordEqualiser generatedUpsertKeyEqualiser,
+ @Nullable int[] inputUpsertKey) {
this.ttlConfig = ttlConfig;
this.serializer = serializer;
- this.generatedEqualiser = generatedEqualiser;
+ this.generatedRecordEqualiser = generatedRecordEqualiser;
+ this.generatedUpsertKeyEqualiser = generatedUpsertKeyEqualiser;
+ this.inputUpsertKey = inputUpsertKey;
+ this.hasUpsertKey = null != inputUpsertKey && inputUpsertKey.length > 0;
+ if (hasUpsertKey) {
+ Preconditions.checkNotNull(
+ generatedUpsertKeyEqualiser,
+ "GeneratedUpsertKeyEqualiser cannot be null when inputUpsertKey is not empty!");
+ }
}
@Override
public void open() throws Exception {
super.open();
- this.equaliser =
- generatedEqualiser.newInstance(getRuntimeContext().getUserCodeClassLoader());
+ if (hasUpsertKey) {
+ this.equaliser =
+ generatedUpsertKeyEqualiser.newInstance(
+ getRuntimeContext().getUserCodeClassLoader());
+ upsertKeyProjectedRow1 = ProjectedRowData.from(inputUpsertKey);
+ upsertKeyProjectedRow2 = ProjectedRowData.from(inputUpsertKey);
+ } else {
+ this.equaliser =
+ generatedRecordEqualiser.newInstance(
+ getRuntimeContext().getUserCodeClassLoader());
+ }
ValueStateDescriptor<List<RowData>> descriptor =
new ValueStateDescriptor<>("values", new ListSerializer<>(serializer));
if (ttlConfig.isEnabled()) {
@@ -109,31 +144,55 @@ public class SinkUpsertMaterializer extends TableStreamOperator<RowData>
switch (row.getRowKind()) {
case INSERT:
case UPDATE_AFTER:
- row.setRowKind(values.isEmpty() ? INSERT : UPDATE_AFTER);
- values.add(row);
- collector.collect(row);
+ addRow(values, row);
break;
case UPDATE_BEFORE:
case DELETE:
- final int lastIndex = values.size() - 1;
- final int index = removeFirst(values, row);
- if (index == -1) {
- LOG.info(STATE_CLEARED_WARN_MSG);
- return;
- }
- if (values.isEmpty()) {
- // Delete this row
- row.setRowKind(DELETE);
- collector.collect(row);
- } else if (index == lastIndex) {
- // Last row has been removed, update to the second last one
- final RowData latestRow = values.get(values.size() - 1);
- latestRow.setRowKind(UPDATE_AFTER);
- collector.collect(latestRow);
- }
+ retractRow(values, row);
break;
}
+ }
+
+ private void addRow(List<RowData> values, RowData add) throws IOException {
+ RowKind outRowKind = values.isEmpty() ? INSERT : UPDATE_AFTER;
+ if (hasUpsertKey) {
+ int index = findFirst(values, add);
+ if (index == -1) {
+ values.add(add);
+ } else {
+ values.set(index, add);
+ }
+ } else {
+ values.add(add);
+ }
+ add.setRowKind(outRowKind);
+ collector.collect(add);
+
+ // Always need to sync with state
+ state.update(values);
+ }
+
+ private void retractRow(List<RowData> values, RowData retract) throws IOException {
+ final int lastIndex = values.size() - 1;
+ final int index = findFirst(values, retract);
+ if (index == -1) {
+ LOG.info(STATE_CLEARED_WARN_MSG);
+ return;
+ } else {
+ // Remove first found row
+ values.remove(index);
+ }
+ if (values.isEmpty()) {
+ // Delete this row
+ retract.setRowKind(DELETE);
+ collector.collect(retract);
+ } else if (index == lastIndex) {
+ // Last row has been removed, update to the second last one
+ final RowData latestRow = values.get(values.size() - 1);
+ latestRow.setRowKind(UPDATE_AFTER);
+ collector.collect(latestRow);
+ }
if (values.isEmpty()) {
state.clear();
@@ -142,19 +201,25 @@ public class SinkUpsertMaterializer extends TableStreamOperator<RowData>
}
}
- private int removeFirst(List<RowData> values, RowData remove) {
+ private int findFirst(List<RowData> values, RowData target) {
final Iterator<RowData> iterator = values.iterator();
int i = 0;
while (iterator.hasNext()) {
- final RowData row = iterator.next();
- // Ignore kind during comparison
- remove.setRowKind(row.getRowKind());
- if (equaliser.equals(row, remove)) {
- iterator.remove();
+ if (equalsIgnoreRowKind(target, iterator.next())) {
return i;
}
i++;
}
return -1;
}
+
+ private boolean equalsIgnoreRowKind(RowData newRow, RowData oldRow) {
+ newRow.setRowKind(oldRow.getRowKind());
+ if (hasUpsertKey) {
+ return equaliser.equals(
+ upsertKeyProjectedRow1.replaceRow(newRow),
+ upsertKeyProjectedRow2.replaceRow(oldRow));
+ }
+ return equaliser.equals(newRow, oldRow);
+ }
}
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/sink/SinkUpsertMaterializerTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/sink/SinkUpsertMaterializerTest.java
index cdc6299a839..671b2ec07d6 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/sink/SinkUpsertMaterializerTest.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/sink/SinkUpsertMaterializerTest.java
@@ -29,6 +29,7 @@ import org.apache.flink.table.runtime.generated.RecordEqualiser;
import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
import org.apache.flink.table.runtime.util.StateConfigUtil;
+import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.VarCharType;
@@ -43,16 +44,19 @@ import java.util.List;
import static org.apache.flink.table.runtime.util.StreamRecordUtils.deleteRecord;
import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord;
import static org.apache.flink.table.runtime.util.StreamRecordUtils.rowOfKind;
+import static org.apache.flink.table.runtime.util.StreamRecordUtils.updateAfterRecord;
import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link SinkUpsertMaterializer}. */
public class SinkUpsertMaterializerTest {
private final StateTtlConfig ttlConfig = StateConfigUtil.createTtlConfig(1000);
- private final LogicalType[] types = new LogicalType[] {new IntType(), new VarCharType()};
+ private final LogicalType[] types =
+ new LogicalType[] {new BigIntType(), new IntType(), new VarCharType()};
private final RowDataSerializer serializer = new RowDataSerializer(types);
private final RowDataKeySelector keySelector =
- HandwrittenSelectorUtil.getRowDataSelector(new int[0], types);
+ HandwrittenSelectorUtil.getRowDataSelector(new int[] {1}, types);
+
private final GeneratedRecordEqualiser equaliser =
new GeneratedRecordEqualiser("", "", new Object[0]) {
@@ -62,10 +66,20 @@ public class SinkUpsertMaterializerTest {
}
};
+ private final GeneratedRecordEqualiser upsertKeyEqualiser =
+ new GeneratedRecordEqualiser("", "", new Object[0]) {
+
+ @Override
+ public RecordEqualiser newInstance(ClassLoader classLoader) {
+ return new TestUpsertKeyEqualiser();
+ }
+ };
+
@Test
public void test() throws Exception {
SinkUpsertMaterializer materializer =
- new SinkUpsertMaterializer(ttlConfig, serializer, equaliser);
+ new SinkUpsertMaterializer(
+ ttlConfig, serializer, equaliser, upsertKeyEqualiser, null);
KeyedOneInputStreamOperatorTestHarness<RowData, RowData, RowData> testHarness =
new KeyedOneInputStreamOperatorTestHarness<>(
materializer, keySelector, keySelector.getProducedType());
@@ -74,30 +88,69 @@ public class SinkUpsertMaterializerTest {
testHarness.setStateTtlProcessingTime(1);
- testHarness.processElement(insertRecord(1, "a1"));
- shouldEmit(testHarness, rowOfKind(RowKind.INSERT, 1, "a1"));
+ testHarness.processElement(insertRecord(1L, 1, "a1"));
+ shouldEmit(testHarness, rowOfKind(RowKind.INSERT, 1L, 1, "a1"));
- testHarness.processElement(insertRecord(1, "a2"));
- shouldEmit(testHarness, rowOfKind(RowKind.UPDATE_AFTER, 1, "a2"));
+ testHarness.processElement(insertRecord(2L, 1, "a2"));
+ shouldEmit(testHarness, rowOfKind(RowKind.UPDATE_AFTER, 2L, 1, "a2"));
- testHarness.processElement(insertRecord(1, "a3"));
- shouldEmit(testHarness, rowOfKind(RowKind.UPDATE_AFTER, 1, "a3"));
+ testHarness.processElement(insertRecord(3L, 1, "a3"));
+ shouldEmit(testHarness, rowOfKind(RowKind.UPDATE_AFTER, 3L, 1, "a3"));
- testHarness.processElement(deleteRecord(1, "a2"));
+ testHarness.processElement(deleteRecord(2L, 1, "a2"));
shouldEmitNothing(testHarness);
- testHarness.processElement(deleteRecord(1, "a3"));
- shouldEmit(testHarness, rowOfKind(RowKind.UPDATE_AFTER, 1, "a1"));
+ testHarness.processElement(deleteRecord(3L, 1, "a3"));
+ shouldEmit(testHarness, rowOfKind(RowKind.UPDATE_AFTER, 1L, 1, "a1"));
- testHarness.processElement(deleteRecord(1, "a1"));
- shouldEmit(testHarness, rowOfKind(RowKind.DELETE, 1, "a1"));
+ testHarness.processElement(deleteRecord(1L, 1, "a1"));
+ shouldEmit(testHarness, rowOfKind(RowKind.DELETE, 1L, 1, "a1"));
- testHarness.processElement(insertRecord(1, "a4"));
- shouldEmit(testHarness, rowOfKind(RowKind.INSERT, 1, "a4"));
+ testHarness.processElement(insertRecord(4L, 1, "a4"));
+ shouldEmit(testHarness, rowOfKind(RowKind.INSERT, 4L, 1, "a4"));
testHarness.setStateTtlProcessingTime(1002);
- testHarness.processElement(deleteRecord(1, "a4"));
+ testHarness.processElement(deleteRecord(4L, 1, "a4"));
+ shouldEmitNothing(testHarness);
+
+ testHarness.close();
+ }
+
+ @Test
+ public void testInputHasUpsertKeyWithNonDeterministicColumn() throws Exception {
+ SinkUpsertMaterializer materializer =
+ new SinkUpsertMaterializer(
+ ttlConfig, serializer, equaliser, upsertKeyEqualiser, new int[] {0});
+ KeyedOneInputStreamOperatorTestHarness<RowData, RowData, RowData> testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(
+ materializer, keySelector, keySelector.getProducedType());
+
+ testHarness.open();
+
+ testHarness.setStateTtlProcessingTime(1);
+
+ testHarness.processElement(insertRecord(1L, 1, "a1"));
+ shouldEmit(testHarness, rowOfKind(RowKind.INSERT, 1L, 1, "a1"));
+
+ testHarness.processElement(updateAfterRecord(1L, 1, "a11"));
+ shouldEmit(testHarness, rowOfKind(RowKind.UPDATE_AFTER, 1L, 1, "a11"));
+
+ testHarness.processElement(insertRecord(3L, 1, "a3"));
+ shouldEmit(testHarness, rowOfKind(RowKind.UPDATE_AFTER, 3L, 1, "a3"));
+
+ testHarness.processElement(deleteRecord(1L, 1, "a111"));
+ shouldEmitNothing(testHarness);
+
+ testHarness.processElement(deleteRecord(3L, 1, "a33"));
+ shouldEmit(testHarness, rowOfKind(RowKind.DELETE, 3L, 1, "a33"));
+
+ testHarness.processElement(insertRecord(4L, 1, "a4"));
+ shouldEmit(testHarness, rowOfKind(RowKind.INSERT, 4L, 1, "a4"));
+
+ testHarness.setStateTtlProcessingTime(1002);
+
+ testHarness.processElement(deleteRecord(4L, 1, "a4"));
shouldEmitNothing(testHarness);
testHarness.close();
@@ -118,7 +171,8 @@ public class SinkUpsertMaterializerTest {
Object o;
while ((o = harness.getOutput().poll()) != null) {
RowData value = (RowData) ((StreamRecord<?>) o).getValue();
- GenericRowData newRow = GenericRowData.of(value.getInt(0), value.getString(1));
+ GenericRowData newRow =
+ GenericRowData.of(value.getLong(0), value.getInt(1), value.getString(2));
newRow.setRowKind(value.getRowKind());
rows.add(newRow);
}
@@ -129,8 +183,16 @@ public class SinkUpsertMaterializerTest {
@Override
public boolean equals(RowData row1, RowData row2) {
return row1.getRowKind() == row2.getRowKind()
- && row1.getInt(0) == row2.getInt(0)
- && row1.getString(1).equals(row2.getString(1));
+ && row1.getLong(0) == row2.getLong(0)
+ && row1.getInt(1) == row2.getInt(1)
+ && row1.getString(2).equals(row2.getString(2));
+ }
+ }
+
+ private static class TestUpsertKeyEqualiser implements RecordEqualiser {
+ @Override
+ public boolean equals(RowData row1, RowData row2) {
+ return row1.getRowKind() == row2.getRowKind() && row1.getLong(0) == row2.getLong(0);
}
}
}
[flink] 01/03: [FLINK-28569][table-planner] Add projectRowType to RowTypeUtils and deprecate AggCodeGenHelper#projectRowType
Posted by go...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
godfrey pushed a commit to branch release-1.16
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2f89d63d57b19d16b4fbfd0bb5b70939a511244e
Author: lincoln.lil <li...@gmail.com>
AuthorDate: Thu Sep 8 18:08:58 2022 +0800
[FLINK-28569][table-planner] Add projectRowType to RowTypeUtils and deprecate AggCodeGenHelper#projectRowType
This closes #20791
(cherry picked from commit dbcd2d7b86fcb7fa7a26e181f1719ea4c6dad828)
---
.../table/planner/typeutils/RowTypeUtils.java | 35 ++++++++++++++++
.../codegen/agg/batch/AggCodeGenHelper.scala | 4 --
.../codegen/agg/batch/HashAggCodeGenerator.scala | 3 +-
.../codegen/agg/batch/SortAggCodeGenerator.scala | 3 +-
.../codegen/agg/batch/WindowCodeGenerator.scala | 3 +-
.../table/planner/typeutils/RowTypeUtilsTest.java | 46 ++++++++++++++++++++++
6 files changed, 87 insertions(+), 7 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/RowTypeUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/RowTypeUtils.java
index ffb9a68a131..4d9879d7b99 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/RowTypeUtils.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/typeutils/RowTypeUtils.java
@@ -18,7 +18,13 @@
package org.apache.flink.table.planner.typeutils;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+
+import javax.annotation.Nonnull;
+
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -46,4 +52,33 @@ public class RowTypeUtils {
}
return result;
}
+
+ /**
+ * Returns projected {@link RowType} by given projection indexes over original {@link RowType}.
+ * Will raise an error when projection index beyond the field count of original rowType.
+ *
+ * @param rowType source row type
+ * @param projection indexes array
+ * @return projected {@link RowType}
+ */
+ public static RowType projectRowType(@Nonnull RowType rowType, @Nonnull int[] projection)
+ throws IllegalArgumentException {
+ final int fieldCnt = rowType.getFieldCount();
+ return RowType.of(
+ Arrays.stream(projection)
+ .mapToObj(
+ index -> {
+ if (index >= fieldCnt) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Invalid projection index: %d of source rowType size: %d",
+ index, fieldCnt));
+ }
+ return rowType.getTypeAt(index);
+ })
+ .toArray(LogicalType[]::new),
+ Arrays.stream(projection)
+ .mapToObj(index -> rowType.getFieldNames().get(index))
+ .toArray(String[]::new));
+ }
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/AggCodeGenHelper.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/AggCodeGenHelper.scala
index bb1135fc236..c401e50ea24 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/AggCodeGenHelper.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/AggCodeGenHelper.scala
@@ -93,10 +93,6 @@ object AggCodeGenHelper {
.asInstanceOf[Map[AggregateFunction[_, _], String]]
}
- def projectRowType(rowType: RowType, mapping: Array[Int]): RowType = {
- RowType.of(mapping.map(rowType.getTypeAt), mapping.map(rowType.getFieldNames.get(_)))
- }
-
/** Add agg handler to class member and open it. */
private[flink] def addAggsHandler(
aggsHandler: GeneratedAggsHandleFunction,
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala
index 550a93df3fb..c768ffd5f70 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/HashAggCodeGenerator.scala
@@ -25,6 +25,7 @@ import org.apache.flink.table.functions.AggregateFunction
import org.apache.flink.table.planner.codegen.{CodeGeneratorContext, CodeGenUtils, ProjectionCodeGenerator}
import org.apache.flink.table.planner.functions.aggfunctions.DeclarativeAggregateFunction
import org.apache.flink.table.planner.plan.utils.{AggregateInfo, AggregateInfoList}
+import org.apache.flink.table.planner.typeutils.RowTypeUtils
import org.apache.flink.table.runtime.generated.GeneratedOperator
import org.apache.flink.table.runtime.operators.TableStreamOperator
import org.apache.flink.table.runtime.operators.aggregate.BytesHashMapSpillMemorySegmentPool
@@ -60,7 +61,7 @@ class HashAggCodeGenerator(
private lazy val aggBufferTypes: Array[Array[LogicalType]] =
AggCodeGenHelper.getAggBufferTypes(inputType, auxGrouping, aggInfos)
- private lazy val groupKeyRowType = AggCodeGenHelper.projectRowType(inputType, grouping)
+ private lazy val groupKeyRowType = RowTypeUtils.projectRowType(inputType, grouping)
private lazy val aggBufferRowType = RowType.of(aggBufferTypes.flatten, aggBufferNames.flatten)
def genWithKeys(): GeneratedOperator[OneInputStreamOperator[RowData, RowData]] = {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/SortAggCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/SortAggCodeGenerator.scala
index 3a183dc4183..02d44d733ed 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/SortAggCodeGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/SortAggCodeGenerator.scala
@@ -25,6 +25,7 @@ import org.apache.flink.table.functions.AggregateFunction
import org.apache.flink.table.planner.codegen.{CodeGeneratorContext, CodeGenUtils, ProjectionCodeGenerator}
import org.apache.flink.table.planner.codegen.OperatorCodeGenerator.generateCollect
import org.apache.flink.table.planner.plan.utils.AggregateInfoList
+import org.apache.flink.table.planner.typeutils.RowTypeUtils
import org.apache.flink.table.runtime.generated.GeneratedOperator
import org.apache.flink.table.runtime.operators.TableStreamOperator
import org.apache.flink.table.types.logical.RowType
@@ -63,7 +64,7 @@ object SortAggCodeGenerator {
val currentKeyTerm = "currentKey"
val currentKeyWriterTerm = "currentKeyWriter"
- val groupKeyRowType = AggCodeGenHelper.projectRowType(inputType, grouping)
+ val groupKeyRowType = RowTypeUtils.projectRowType(inputType, grouping)
val keyProjectionCode = ProjectionCodeGenerator
.generateProjectionExpression(
ctx,
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/WindowCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/WindowCodeGenerator.scala
index ab4a4221cbb..2df6ae64b14 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/WindowCodeGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/agg/batch/WindowCodeGenerator.scala
@@ -38,6 +38,7 @@ import org.apache.flink.table.planner.expressions.ExpressionBuilder._
import org.apache.flink.table.planner.expressions.converter.ExpressionConverter
import org.apache.flink.table.planner.plan.logical.{LogicalWindow, SlidingGroupWindow, TumblingGroupWindow}
import org.apache.flink.table.planner.plan.utils.{AggregateInfo, AggregateInfoList, AggregateUtil}
+import org.apache.flink.table.planner.typeutils.RowTypeUtils
import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTypeFactory
import org.apache.flink.table.runtime.groupwindow.NamedWindowProperty
import org.apache.flink.table.runtime.operators.window.TimeWindow
@@ -82,7 +83,7 @@ abstract class WindowCodeGenerator(
AggCodeGenHelper.getAggBufferTypes(inputRowType, auxGrouping, aggInfos)
protected lazy val groupKeyRowType: RowType =
- AggCodeGenHelper.projectRowType(inputRowType, grouping)
+ RowTypeUtils.projectRowType(inputRowType, grouping)
protected lazy val timestampInternalType: LogicalType =
if (inputTimeIsDate) new IntType() else new BigIntType()
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/typeutils/RowTypeUtilsTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/typeutils/RowTypeUtilsTest.java
index b0e754a037b..7449d83a903 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/typeutils/RowTypeUtilsTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/typeutils/RowTypeUtilsTest.java
@@ -18,7 +18,16 @@
package org.apache.flink.table.planner.typeutils;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.ExpectedException;
import java.util.Arrays;
@@ -27,6 +36,13 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link RowTypeUtils}. */
public class RowTypeUtilsTest {
+ @Rule public ExpectedException expectedException = ExpectedException.none();
+
+ private final RowType srcType =
+ RowType.of(
+ new LogicalType[] {new IntType(), new VarCharType(), new BigIntType()},
+ new String[] {"f0", "f1", "f2"});
+
@Test
public void testGetUniqueName() {
assertThat(
@@ -39,4 +55,34 @@ public class RowTypeUtilsTest {
Arrays.asList("Alice", "Bob")))
.isEqualTo(Arrays.asList("Bob_0", "Bob_1", "Dave", "Alice_0"));
}
+
+ @Test
+ public void testProjectRowType() {
+ assertThat(RowTypeUtils.projectRowType(srcType, new int[] {0}))
+ .isEqualTo(RowType.of(new LogicalType[] {new IntType()}, new String[] {"f0"}));
+
+ assertThat(RowTypeUtils.projectRowType(srcType, new int[] {0, 2}))
+ .isEqualTo(
+ RowType.of(
+ new LogicalType[] {new IntType(), new BigIntType()},
+ new String[] {"f0", "f2"}));
+
+ assertThat(RowTypeUtils.projectRowType(srcType, new int[] {0, 1, 2})).isEqualTo(srcType);
+ }
+
+ @Test
+ public void testInvalidProjectRowType() {
+
+ expectedException.expect(IllegalArgumentException.class);
+ expectedException.expectMessage("Invalid projection index: 3");
+ RowTypeUtils.projectRowType(srcType, new int[] {0, 1, 2, 3});
+
+ expectedException.expect(IllegalArgumentException.class);
+ expectedException.expectMessage("Invalid projection index: 3");
+ RowTypeUtils.projectRowType(srcType, new int[] {0, 1, 3});
+
+ expectedException.expect(ValidationException.class);
+ expectedException.expectMessage("Field names must be unique. Found duplicates");
+ RowTypeUtils.projectRowType(srcType, new int[] {0, 0, 0, 0});
+ }
}