You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2025/02/28 17:31:13 UTC
(flink) branch master updated: [FLINK-37390] Push filter into ChangelogNormalize
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz 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 6cf885f3aa4 [FLINK-37390] Push filter into ChangelogNormalize
6cf885f3aa4 is described below
commit 6cf885f3aa477c80a7bc1ee5759613387062463f
Author: Dawid Wysakowicz <dw...@apache.org>
AuthorDate: Tue Feb 25 16:43:14 2025 +0100
[FLINK-37390] Push filter into ChangelogNormalize
---
.../nodes/exec/common/CommonExecLookupJoin.java | 5 +-
.../exec/stream/StreamExecChangelogNormalize.java | 38 +++--
.../nodes/exec/stream/StreamExecDeduplicate.java | 6 +-
.../stream/PushCalcPastChangelogNormalizeRule.java | 64 +++++---
...arkAssignerChangelogNormalizeTransposeRule.java | 41 ++++-
.../planner/codegen/FilterCodeGenerator.scala | 55 +++++++
.../planner/codegen/LookupJoinCodeGenerator.scala | 27 ----
.../stream/StreamPhysicalChangelogNormalize.scala | 36 ++++-
.../exec/stream/ChangelogNormalizeRestoreTest.java | 3 +-
.../stream/ChangelogNormalizeTestPrograms.java | 35 ++++
.../PushCalcPastChangelogNormalizeRuleTest.xml | 16 +-
.../planner/plan/stream/sql/TableScanTest.xml | 177 ++++++++++-----------
.../plan/changelog-normalize-upsert-filter.json | 155 ++++++++++++++++++
.../savepoint/_metadata | Bin 0 -> 13292 bytes
.../ProcTimeDeduplicateKeepLastRowFunction.java | 27 +++-
...imeMiniBatchDeduplicateKeepLastRowFunction.java | 26 ++-
.../utils/DeduplicateFunctionHelper.java | 68 ++++++++
.../ProcTimeDeduplicateFunctionTestBase.java | 39 +++++
...ProcTimeDeduplicateKeepLastRowFunctionTest.java | 52 +++++-
...iniBatchDeduplicateKeepLastRowFunctionTest.java | 76 ++++++++-
20 files changed, 773 insertions(+), 173 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java
index 41d658267d7..f30268510fb 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecLookupJoin.java
@@ -44,6 +44,7 @@ import org.apache.flink.table.legacy.sources.LookupableTableSource;
import org.apache.flink.table.legacy.sources.TableSource;
import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
import org.apache.flink.table.planner.codegen.CodeGeneratorContext;
+import org.apache.flink.table.planner.codegen.FilterCodeGenerator;
import org.apache.flink.table.planner.codegen.LookupJoinCodeGenerator;
import org.apache.flink.table.planner.delegation.PlannerBase;
import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
@@ -442,7 +443,7 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> {
JavaScalaConversionUtil.toScala(
Optional.ofNullable(remainingJoinCondition)));
GeneratedFilterCondition generatedPreFilterCondition =
- LookupJoinCodeGenerator.generatePreFilterCondition(
+ FilterCodeGenerator.generateFilterCondition(
config, classLoader, preFilterCondition, inputRowType);
DataStructureConverter<?, ?> fetcherConverter =
@@ -581,7 +582,7 @@ public abstract class CommonExecLookupJoin extends ExecNodeBase<RowData> {
true);
GeneratedFilterCondition generatedPreFilterCondition =
- LookupJoinCodeGenerator.generatePreFilterCondition(
+ FilterCodeGenerator.generateFilterCondition(
config, classLoader, preFilterCondition, inputRowType);
ProcessFunction<RowData, RowData> processFunc;
if (projectionOnTemporalTable != null) {
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecChangelogNormalize.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecChangelogNormalize.java
index 29171195e85..627b6ebad0d 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecChangelogNormalize.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecChangelogNormalize.java
@@ -27,6 +27,7 @@ import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.transformations.OneInputTransformation;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.planner.codegen.EqualiserCodeGenerator;
+import org.apache.flink.table.planner.codegen.FilterCodeGenerator;
import org.apache.flink.table.planner.delegation.PlannerBase;
import org.apache.flink.table.planner.plan.nodes.exec.ExecEdge;
import org.apache.flink.table.planner.plan.nodes.exec.ExecNode;
@@ -40,6 +41,7 @@ import org.apache.flink.table.planner.plan.nodes.exec.StateMetadata;
import org.apache.flink.table.planner.plan.nodes.exec.utils.ExecNodeUtil;
import org.apache.flink.table.planner.plan.utils.KeySelectorUtil;
import org.apache.flink.table.planner.plan.utils.MinibatchUtil;
+import org.apache.flink.table.runtime.generated.GeneratedFilterCondition;
import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser;
import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
import org.apache.flink.table.runtime.operators.bundle.KeyedMapBundleOperator;
@@ -53,6 +55,8 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCre
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonInclude;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.calcite.rex.RexNode;
+
import javax.annotation.Nullable;
import java.util.Collections;
@@ -82,6 +86,7 @@ public class StreamExecChangelogNormalize extends ExecNodeBase<RowData>
public static final String FIELD_NAME_UNIQUE_KEYS = "uniqueKeys";
public static final String FIELD_NAME_GENERATE_UPDATE_BEFORE = "generateUpdateBefore";
public static final String STATE_NAME = "changelogNormalizeState";
+ public static final String FIELD_NAME_FILTER_CONDITION = "filterCondition";
@JsonProperty(FIELD_NAME_UNIQUE_KEYS)
private final int[] uniqueKeys;
@@ -89,6 +94,11 @@ public class StreamExecChangelogNormalize extends ExecNodeBase<RowData>
@JsonProperty(FIELD_NAME_GENERATE_UPDATE_BEFORE)
private final boolean generateUpdateBefore;
+ @Nullable
+ @JsonProperty(FIELD_NAME_FILTER_CONDITION)
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ private final RexNode filterCondition;
+
@Nullable
@JsonProperty(FIELD_NAME_STATE)
@JsonInclude(JsonInclude.Include.NON_NULL)
@@ -98,6 +108,7 @@ public class StreamExecChangelogNormalize extends ExecNodeBase<RowData>
ReadableConfig tableConfig,
int[] uniqueKeys,
boolean generateUpdateBefore,
+ @Nullable RexNode filterCondition,
InputProperty inputProperty,
RowType outputType,
String description) {
@@ -110,7 +121,8 @@ public class StreamExecChangelogNormalize extends ExecNodeBase<RowData>
StateMetadata.getOneInputOperatorDefaultMeta(tableConfig, STATE_NAME),
Collections.singletonList(inputProperty),
outputType,
- description);
+ description,
+ filterCondition);
}
@JsonCreator
@@ -123,11 +135,13 @@ public class StreamExecChangelogNormalize extends ExecNodeBase<RowData>
@Nullable @JsonProperty(FIELD_NAME_STATE) List<StateMetadata> stateMetadataList,
@JsonProperty(FIELD_NAME_INPUT_PROPERTIES) List<InputProperty> inputProperties,
@JsonProperty(FIELD_NAME_OUTPUT_TYPE) RowType outputType,
- @JsonProperty(FIELD_NAME_DESCRIPTION) String description) {
+ @JsonProperty(FIELD_NAME_DESCRIPTION) String description,
+ @JsonProperty(FIELD_NAME_FILTER_CONDITION) @Nullable RexNode filterCondition) {
super(id, context, persistedConfig, inputProperties, outputType, description);
this.uniqueKeys = uniqueKeys;
this.generateUpdateBefore = generateUpdateBefore;
this.stateMetadataList = stateMetadataList;
+ this.filterCondition = filterCondition;
}
@SuppressWarnings("unchecked")
@@ -146,11 +160,16 @@ public class StreamExecChangelogNormalize extends ExecNodeBase<RowData>
StateMetadata.getStateTtlForOneInputOperator(config, stateMetadataList);
final boolean isMiniBatchEnabled = MinibatchUtil.isMiniBatchEnabled(config);
- GeneratedRecordEqualiser generatedEqualiser =
- new EqualiserCodeGenerator(
- rowTypeInfo.toRowType(), planner.getFlinkContext().getClassLoader())
+ final ClassLoader classLoader = planner.getFlinkContext().getClassLoader();
+ final RowType inputType = rowTypeInfo.toRowType();
+ final GeneratedRecordEqualiser generatedEqualiser =
+ new EqualiserCodeGenerator(inputType, classLoader)
.generateRecordEqualiser("DeduplicateRowEqualiser");
+ final GeneratedFilterCondition generatedFilterCondition =
+ FilterCodeGenerator.generateFilterCondition(
+ config, classLoader, filterCondition, inputType);
+
if (isMiniBatchEnabled) {
TypeSerializer<RowData> rowSerializer =
rowTypeInfo.createSerializer(
@@ -163,7 +182,8 @@ public class StreamExecChangelogNormalize extends ExecNodeBase<RowData>
generateUpdateBefore,
true, // generateInsert
false, // inputInsertOnly
- generatedEqualiser);
+ generatedEqualiser,
+ generatedFilterCondition);
CountBundleTrigger<RowData> trigger = MinibatchUtil.createMiniBatchTrigger(config);
operator = new KeyedMapBundleOperator<>(processFunction, trigger);
} else {
@@ -174,7 +194,8 @@ public class StreamExecChangelogNormalize extends ExecNodeBase<RowData>
generateUpdateBefore,
true, // generateInsert
false, // inputInsertOnly
- generatedEqualiser);
+ generatedEqualiser,
+ generatedFilterCondition);
operator = new KeyedProcessOperator<>(processFunction);
}
@@ -188,8 +209,7 @@ public class StreamExecChangelogNormalize extends ExecNodeBase<RowData>
false);
final RowDataKeySelector selector =
- KeySelectorUtil.getRowDataSelector(
- planner.getFlinkContext().getClassLoader(), uniqueKeys, rowTypeInfo);
+ KeySelectorUtil.getRowDataSelector(classLoader, uniqueKeys, rowTypeInfo);
transform.setStateKeySelector(selector);
transform.setStateKeyType(selector.getProducedType());
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeduplicate.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeduplicate.java
index e5ec2d3518a..bde1a186f6c 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeduplicate.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/stream/StreamExecDeduplicate.java
@@ -402,7 +402,8 @@ public class StreamExecDeduplicate extends ExecNodeBase<RowData>
generateUpdateBefore,
generateInsert(),
true,
- generatedEqualiser);
+ generatedEqualiser,
+ null);
return new KeyedMapBundleOperator<>(processFunction, trigger);
} else {
ProcTimeMiniBatchDeduplicateKeepFirstRowFunction processFunction =
@@ -419,7 +420,8 @@ public class StreamExecDeduplicate extends ExecNodeBase<RowData>
generateUpdateBefore,
generateInsert(),
true,
- generatedEqualiser);
+ generatedEqualiser,
+ null);
return new KeyedProcessOperator<>(processFunction);
} else {
ProcTimeDeduplicateKeepFirstRowFunction processFunction =
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRule.java
index 871deb72856..bcf3a5327b0 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRule.java
@@ -67,7 +67,9 @@ import static org.apache.flink.table.planner.plan.utils.RexNodeExtractor.extract
* size of its state.
*
* <p>Note that pushing primary key filters is safe to do, but pushing any other filters can lead to
- * incorrect results.
+ * incorrect results. Other filter conditions are pushed into the ChangelogNormalize operator. That
+ * way we can reduce the state of the operator if records don't match the filter. It's especially
+ * useful in case of highly selective filters.
*/
@Internal
@Value.Enclosing
@@ -103,22 +105,28 @@ public class PushCalcPastChangelogNormalizeRule
primaryKeyPredicates,
otherPredicates);
}
+ if (changelogNormalize.filterCondition() != null) {
+ otherPredicates.add(changelogNormalize.filterCondition());
+ }
// used input field indices
- int[] usedInputFields = extractUsedInputFields(calc, primaryKeyIndices);
+ int[] usedInputFields = extractUsedInputFields(calc, changelogNormalize, primaryKeyIndices);
// Construct a new ChangelogNormalize which has used fields project
// and primary key filters pushed into it
final StreamPhysicalChangelogNormalize newChangelogNormalize =
- pushCalcThroughChangelogNormalize(call, primaryKeyPredicates, usedInputFields);
+ pushCalcThroughChangelogNormalize(
+ call, primaryKeyPredicates, otherPredicates, usedInputFields);
// Retain only filters which haven't been pushed
- transformWithRemainingPredicates(
- call, newChangelogNormalize, otherPredicates, usedInputFields);
+ transformWithRemainingPredicates(call, newChangelogNormalize, usedInputFields);
}
/** Extracts input fields which are used in the Calc node and the ChangelogNormalize node. */
- private int[] extractUsedInputFields(StreamPhysicalCalc calc, Set<Integer> primaryKeyIndices) {
+ private int[] extractUsedInputFields(
+ StreamPhysicalCalc calc,
+ StreamPhysicalChangelogNormalize changelogNormalize,
+ Set<Integer> primaryKeyIndices) {
RexProgram program = calc.getProgram();
List<RexNode> projectsAndCondition =
program.getProjectList().stream()
@@ -127,6 +135,9 @@ public class PushCalcPastChangelogNormalizeRule
if (program.getCondition() != null) {
projectsAndCondition.add(program.expandLocalRef(program.getCondition()));
}
+ if (changelogNormalize.filterCondition() != null) {
+ projectsAndCondition.add(changelogNormalize.filterCondition());
+ }
Set<Integer> projectedFields =
Arrays.stream(extractRefInputFields(projectsAndCondition))
.boxed()
@@ -160,7 +171,10 @@ public class PushCalcPastChangelogNormalizeRule
* StreamPhysicalChangelogNormalize}.
*/
private StreamPhysicalChangelogNormalize pushCalcThroughChangelogNormalize(
- RelOptRuleCall call, List<RexNode> primaryKeyPredicates, int[] usedInputFields) {
+ RelOptRuleCall call,
+ List<RexNode> primaryKeyPredicates,
+ List<RexNode> otherPredicates,
+ int[] usedInputFields) {
final StreamPhysicalChangelogNormalize changelogNormalize = call.rel(1);
final StreamPhysicalExchange exchange = call.rel(2);
final Set<Integer> primaryKeyIndices =
@@ -168,9 +182,18 @@ public class PushCalcPastChangelogNormalizeRule
if (primaryKeyPredicates.isEmpty()
&& usedInputFields.length == changelogNormalize.getRowType().getFieldCount()) {
- // There are no filters and no project which can be pushed, so just return the existing
- // node.
- return changelogNormalize;
+ // Check if there is a condition that can be pushed into the ChangelogNormalize
+ if (otherPredicates.isEmpty()) {
+ return changelogNormalize;
+ } else {
+ final RexNode condition = call.builder().and(otherPredicates);
+ return (StreamPhysicalChangelogNormalize)
+ changelogNormalize.copy(
+ changelogNormalize.getTraitSet(),
+ exchange,
+ changelogNormalize.uniqueKeys(),
+ condition.isAlwaysTrue() ? null : condition);
+ }
}
final StreamPhysicalCalc pushedCalc =
@@ -182,6 +205,13 @@ public class PushCalcPastChangelogNormalizeRule
final List<Integer> newPrimaryKeyIndices =
primaryKeyIndices.stream().map(inputRefMapping::get).collect(Collectors.toList());
+ // add conditions
+ final List<RexNode> shiftedPredicates =
+ otherPredicates.stream()
+ .map(p -> adjustInputRef(p, inputRefMapping))
+ .collect(Collectors.toList());
+ final RexNode condition = call.builder().and(shiftedPredicates);
+
final FlinkRelDistribution newDistribution =
FlinkRelDistribution.hash(newPrimaryKeyIndices, true);
final RelTraitSet newTraitSet = exchange.getTraitSet().replace(newDistribution);
@@ -192,7 +222,8 @@ public class PushCalcPastChangelogNormalizeRule
changelogNormalize.copy(
changelogNormalize.getTraitSet(),
newExchange,
- newPrimaryKeyIndices.stream().mapToInt(Integer::intValue).toArray());
+ newPrimaryKeyIndices.stream().mapToInt(Integer::intValue).toArray(),
+ condition.isAlwaysTrue() ? null : condition);
}
/**
@@ -234,7 +265,6 @@ public class PushCalcPastChangelogNormalizeRule
private void transformWithRemainingPredicates(
RelOptRuleCall call,
StreamPhysicalChangelogNormalize changelogNormalize,
- List<RexNode> predicates,
int[] usedInputFields) {
final StreamPhysicalCalc calc = call.rel(0);
final RelBuilder relBuilder = call.builder();
@@ -250,16 +280,6 @@ public class PushCalcPastChangelogNormalizeRule
programBuilder.addProject(shiftedProject, ref.right);
}
- // add conditions
- final List<RexNode> shiftedPredicates =
- predicates.stream()
- .map(p -> adjustInputRef(p, inputRefMapping))
- .collect(Collectors.toList());
- final RexNode condition = relBuilder.and(shiftedPredicates);
- if (!condition.isAlwaysTrue()) {
- programBuilder.addCondition(condition);
- }
-
final RexProgram newProgram = programBuilder.getProgram();
if (newProgram.isTrivial()) {
call.transformTo(changelogNormalize);
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/WatermarkAssignerChangelogNormalizeTransposeRule.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/WatermarkAssignerChangelogNormalizeTransposeRule.java
index 51963d79eb3..c55c872b224 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/WatermarkAssignerChangelogNormalizeTransposeRule.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/WatermarkAssignerChangelogNormalizeTransposeRule.java
@@ -54,6 +54,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -368,13 +370,31 @@ public class WatermarkAssignerChangelogNormalizeTransposeRule
} else if (currentNode instanceof StreamPhysicalChangelogNormalize) {
final List<String> inputNodeFields = inputNode.getRowType().getFieldNames();
final List<String> currentNodeFields = currentNode.getRowType().getFieldNames();
+ final Map<Integer, Integer> indexMapping =
+ IntStream.range(0, currentNodeFields.size())
+ .boxed()
+ .collect(
+ Collectors.toMap(
+ Function.identity(),
+ i ->
+ inputNodeFields.indexOf(
+ currentNodeFields.get(i))));
+ final StreamPhysicalChangelogNormalize changelogNormalize =
+ (StreamPhysicalChangelogNormalize) currentNode;
int[] remappedUniqueKeys =
- Arrays.stream(((StreamPhysicalChangelogNormalize) currentNode).uniqueKeys())
- .map(ukIdx -> inputNodeFields.indexOf(currentNodeFields.get(ukIdx)))
+ Arrays.stream(changelogNormalize.uniqueKeys())
+ .map(indexMapping::get)
.toArray();
+
+ final RexNode remappedCondition =
+ changelogNormalize.filterCondition() == null
+ ? null
+ : adjustInputRef(
+ changelogNormalize.filterCondition(), indexMapping);
+
currentNode =
- ((StreamPhysicalChangelogNormalize) currentNode)
- .copy(nodeAndTrait.f1, inputNode, remappedUniqueKeys);
+ changelogNormalize.copy(
+ nodeAndTrait.f1, inputNode, remappedUniqueKeys, remappedCondition);
} else {
currentNode =
currentNode.copy(nodeAndTrait.f1, Collections.singletonList(inputNode));
@@ -384,6 +404,19 @@ public class WatermarkAssignerChangelogNormalizeTransposeRule
return currentNode;
}
+ /** Adjust the {@param expr} field indices according to the field index {@param mapping}. */
+ private RexNode adjustInputRef(RexNode expr, Map<Integer, Integer> mapping) {
+ return expr.accept(
+ new RexShuttle() {
+
+ @Override
+ public RexNode visitInputRef(RexInputRef inputRef) {
+ Integer newIndex = mapping.get(inputRef.getIndex());
+ return new RexInputRef(newIndex, inputRef.getType());
+ }
+ });
+ }
+
/** Rule configuration. */
@Value.Immutable(singleton = false)
public interface Config extends RelRule.Config {
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/FilterCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/FilterCodeGenerator.scala
new file mode 100644
index 00000000000..c84225f9b2f
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/FilterCodeGenerator.scala
@@ -0,0 +1,55 @@
+/*
+ * 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.codegen
+
+import org.apache.flink.configuration.ReadableConfig
+import org.apache.flink.table.runtime.generated.GeneratedFilterCondition
+import org.apache.flink.table.types.logical.LogicalType
+
+import org.apache.calcite.rex.RexNode
+
+/**
+ * A code generator for generating [[org.apache.flink.table.runtime.generated.FilterCondition]] that
+ * can be used in an operator code.
+ */
+object FilterCodeGenerator {
+
+ /** Generates filter condition runner. */
+ def generateFilterCondition(
+ tableConfig: ReadableConfig,
+ classLoader: ClassLoader,
+ filterCondition: RexNode,
+ inputType: LogicalType): GeneratedFilterCondition = {
+ val ctx = new CodeGeneratorContext(tableConfig, classLoader)
+ // should consider null fields
+ val exprGenerator =
+ new ExprCodeGenerator(ctx, false).bindInput(inputType, CodeGenUtils.DEFAULT_INPUT_TERM)
+
+ val bodyCode = if (filterCondition == null) {
+ "return true;"
+ } else {
+ val condition = exprGenerator.generateExpression(filterCondition)
+ s"""
+ |${condition.code}
+ |return ${condition.resultTerm};
+ |""".stripMargin
+ }
+
+ FunctionCodeGenerator.generateFilterCondition(ctx, "PreFilterCondition", bodyCode)
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/LookupJoinCodeGenerator.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/LookupJoinCodeGenerator.scala
index 5aa5b6c8d20..f18186ae3c1 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/LookupJoinCodeGenerator.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/LookupJoinCodeGenerator.scala
@@ -532,31 +532,4 @@ object LookupJoinCodeGenerator {
classLoader
)
}
-
- /**
- * Generates pre-filter condition for lookup join which can be applied before access the dimension
- * table.
- */
- def generatePreFilterCondition(
- tableConfig: ReadableConfig,
- classLoader: ClassLoader,
- preFilterCondition: RexNode,
- leftType: LogicalType): GeneratedFilterCondition = {
- val ctx = new CodeGeneratorContext(tableConfig, classLoader)
- // should consider null fields
- val exprGenerator =
- new ExprCodeGenerator(ctx, false).bindInput(leftType, CodeGenUtils.DEFAULT_INPUT_TERM)
-
- val bodyCode = if (preFilterCondition == null) {
- "return true;"
- } else {
- val condition = exprGenerator.generateExpression(preFilterCondition)
- s"""
- |${condition.code}
- |return ${condition.resultTerm};
- |""".stripMargin
- }
-
- FunctionCodeGenerator.generateFilterCondition(ctx, "PreFilterCondition", bodyCode)
- }
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalChangelogNormalize.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalChangelogNormalize.scala
index 54a7b26bf39..c15ab819672 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalChangelogNormalize.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamPhysicalChangelogNormalize.scala
@@ -17,15 +17,20 @@
*/
package org.apache.flink.table.planner.plan.nodes.physical.stream
+import org.apache.flink.table.planner.JList
import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.plan.nodes.exec.{ExecNode, InputProperty}
import org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecChangelogNormalize
import org.apache.flink.table.planner.plan.utils.ChangelogPlanUtils
+import org.apache.flink.table.planner.plan.utils.RelExplainUtil.{conditionsToString, conditionToString, preferExpressionFormat}
+import org.apache.flink.table.planner.utils.JavaScalaConversionUtil
import org.apache.flink.table.planner.utils.ShortcutUtils.unwrapTableConfig
+import org.apache.flink.table.runtime.generated.FilterCondition
import org.apache.calcite.plan.{RelOptCluster, RelTraitSet}
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.{RelNode, RelWriter, SingleRel}
+import org.apache.calcite.rex.RexNode
import java.util
@@ -39,7 +44,8 @@ class StreamPhysicalChangelogNormalize(
cluster: RelOptCluster,
traitSet: RelTraitSet,
input: RelNode,
- val uniqueKeys: Array[Int])
+ val uniqueKeys: Array[Int],
+ val filterCondition: RexNode = null)
extends SingleRel(cluster, traitSet, input)
with StreamPhysicalRel {
@@ -48,18 +54,39 @@ class StreamPhysicalChangelogNormalize(
override def deriveRowType(): RelDataType = getInput.getRowType
override def copy(traitSet: RelTraitSet, inputs: util.List[RelNode]): RelNode = {
- new StreamPhysicalChangelogNormalize(cluster, traitSet, inputs.get(0), uniqueKeys)
+ new StreamPhysicalChangelogNormalize(
+ cluster,
+ traitSet,
+ inputs.get(0),
+ uniqueKeys,
+ filterCondition)
}
- def copy(traitSet: RelTraitSet, input: RelNode, uniqueKeys: Array[Int]): RelNode = {
- new StreamPhysicalChangelogNormalize(cluster, traitSet, input, uniqueKeys)
+ def copy(
+ traitSet: RelTraitSet,
+ input: RelNode,
+ uniqueKeys: Array[Int],
+ filterCondition: RexNode): RelNode = {
+ new StreamPhysicalChangelogNormalize(cluster, traitSet, input, uniqueKeys, filterCondition)
}
override def explainTerms(pw: RelWriter): RelWriter = {
val fieldNames = getRowType.getFieldNames
+ val conditions = JavaScalaConversionUtil.toJava(List(filterCondition))
super
.explainTerms(pw)
.item("key", uniqueKeys.map(fieldNames.get).mkString(", "))
+ .itemIf(
+ "condition",
+ conditionsToString(
+ Option(conditions),
+ input.getRowType,
+ getExpressionString,
+ preferExpressionFormat(pw),
+ convertToExpressionDetail(pw.getDetailLevel)
+ ),
+ filterCondition != null
+ )
}
override def translateToExecNode(): ExecNode[_] = {
@@ -68,6 +95,7 @@ class StreamPhysicalChangelogNormalize(
unwrapTableConfig(this),
uniqueKeys,
generateUpdateBefore,
+ filterCondition,
InputProperty.DEFAULT,
FlinkTypeFactory.toLogicalRowType(getRowType),
getRelDetailedDescription)
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogNormalizeRestoreTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogNormalizeRestoreTest.java
index 08fe8bb9979..03fbd5db543 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogNormalizeRestoreTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogNormalizeRestoreTest.java
@@ -39,6 +39,7 @@ public class ChangelogNormalizeRestoreTest extends RestoreTestBase {
return Arrays.asList(
ChangelogNormalizeTestPrograms.CHANGELOG_SOURCE,
ChangelogNormalizeTestPrograms.CHANGELOG_SOURCE_MINI_BATCH,
- ChangelogNormalizeTestPrograms.UPSERT_SOURCE);
+ ChangelogNormalizeTestPrograms.UPSERT_SOURCE,
+ ChangelogNormalizeTestPrograms.UPSERT_SOURCE_WITH_FILTER);
}
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogNormalizeTestPrograms.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogNormalizeTestPrograms.java
index e4a3cd1825e..e92fe2839a4 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogNormalizeTestPrograms.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ChangelogNormalizeTestPrograms.java
@@ -164,4 +164,39 @@ public class ChangelogNormalizeTestPrograms {
.build())
.runSql("INSERT INTO sink_t SELECT a, b, c FROM source_t")
.build();
+
+ static final TableTestProgram UPSERT_SOURCE_WITH_FILTER =
+ TableTestProgram.of(
+ "changelog-normalize-upsert-filter",
+ "validates changelog normalize upsert with filter")
+ .setupConfig(
+ ExecutionConfigOptions.TABLE_EXEC_SOURCE_CDC_EVENTS_DUPLICATE, true)
+ .setupTableSource(
+ SourceTestStep.newBuilder("source_t")
+ .addOption("changelog-mode", "I,UA,D")
+ .addSchema(SOURCE_SCHEMA)
+ .producedBeforeRestore(
+ Row.ofKind(RowKind.UPDATE_AFTER, "one", 1, "a"),
+ Row.ofKind(RowKind.UPDATE_AFTER, "one", 2, "b"),
+ Row.ofKind(RowKind.UPDATE_AFTER, "one", 12, "b"),
+ Row.ofKind(RowKind.UPDATE_AFTER, "one", 13, "b"),
+ Row.ofKind(RowKind.UPDATE_AFTER, "three", 3, "cc"))
+ .producedAfterRestore(
+ Row.ofKind(RowKind.UPDATE_AFTER, "one", 15, "aa"),
+ Row.ofKind(RowKind.DELETE, "one", 15, "c"),
+ Row.ofKind(RowKind.DELETE, "three", 3, "cc"))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink_t")
+ .addSchema(SINK_SCHEMA)
+ .consumedBeforeRestore(
+ "+I[one, 1, a]",
+ "-U[one, 1, a]",
+ "+U[one, 2, b]",
+ "-D[one, 2, b]",
+ "+I[three, 3, cc]")
+ .consumedAfterRestore("-D[three, 3, cc]")
+ .build())
+ .runSql("INSERT INTO sink_t SELECT a, b, c FROM source_t WHERE b < 10")
+ .build();
}
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.xml
index 7b5f30bbd2f..a30c630e7ba 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.xml
@@ -29,8 +29,8 @@ LogicalProject(f1=[$1], f5=[$5])
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-Calc(select=[f1, f5], where=[IS NOT NULL(f3)])
-+- ChangelogNormalize(key=[f1, f2])
+Calc(select=[f1, f5])
++- ChangelogNormalize(key=[f1, f2], condition=[IS NOT NULL(f3)])
+- Exchange(distribution=[hash[f1, f2]])
+- Calc(select=[f1, f2, f3, f5], where=[OR(<(f1, 1), >(f2, 10))])
+- TableSourceScan(table=[[default_catalog, default_database, T]], fields=[f0, f1, f2, f3, f4, f5, f6, f7])
@@ -50,8 +50,8 @@ LogicalProject(f0=[$0], f1=[$1])
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-Calc(select=[f0, f1], where=[IS NOT NULL(f0)])
-+- ChangelogNormalize(key=[f1, f2])
+Calc(select=[f0, f1])
++- ChangelogNormalize(key=[f1, f2], condition=[IS NOT NULL(f0)])
+- Exchange(distribution=[hash[f1, f2]])
+- Calc(select=[f0, f1, f2], where=[OR(<(f1, 1), >(f2, 10))])
+- TableSourceScan(table=[[default_catalog, default_database, T]], fields=[f0, f1, f2])
@@ -90,8 +90,8 @@ LogicalProject(f1=[$1], f5=[$5])
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-Calc(select=[f1, f5], where=[IS NOT NULL(f3)])
-+- ChangelogNormalize(key=[f1, f2])
+Calc(select=[f1, f5])
++- ChangelogNormalize(key=[f1, f2], condition=[IS NOT NULL(f3)])
+- Exchange(distribution=[hash[f1, f2]])
+- Calc(select=[f1, f2, f3, f5], where=[<(f1, 1)])
+- TableSourceScan(table=[[default_catalog, default_database, T]], fields=[f0, f1, f2, f3, f4, f5, f6, f7])
@@ -132,8 +132,8 @@ LogicalAggregate(group=[{0}], sum=[SUM($0)])
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-Calc(select=[f1, f1 AS sum], where=[OR(>(f1, 3), IS NULL(f2))])
-+- ChangelogNormalize(key=[f1])
+Calc(select=[f1, f1 AS sum])
++- ChangelogNormalize(key=[f1], condition=[OR(>(f1, 3), IS NULL(f2))])
+- Exchange(distribution=[hash[f1]])
+- Calc(select=[f1, f2], where=[<(f1, 10)])
+- TableSourceScan(table=[[default_catalog, default_database, T]], fields=[f0, f1, f2])
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.xml
index 95ba887b90e..04775282ec1 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/TableScanTest.xml
@@ -95,8 +95,8 @@ LogicalProject(a=[$1], b=[$2], c=[$3])
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-Calc(select=[a, b, c], where=[>(a, 1)], changelogMode=[I,UB,UA,D])
-+- ChangelogNormalize(key=[id], changelogMode=[I,UB,UA,D])
+Calc(select=[a, b, c], changelogMode=[I,UB,UA,D])
++- ChangelogNormalize(key=[id], condition=[>(a, 1)], changelogMode=[I,UB,UA,D])
+- Exchange(distribution=[hash[id]], changelogMode=[I,UA,D])
+- Calc(select=[id, a, b, c], changelogMode=[I,UA,D])
+- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL SECOND)], changelogMode=[I,UA,D])
@@ -546,6 +546,90 @@ Calc(select=[id1, a, b], changelogMode=[I,UA,D])
]]>
</Resource>
</TestCase>
+ <TestCase name="testSetParallelismForSource">
+ <Resource name="explain">
+ <![CDATA[== Abstract Syntax Tree ==
+LogicalProject(id=[$0], b=[$1], c=[$2], id0=[$3], a=[$4])
++- LogicalFilter(condition=[>($2, 1)])
+ +- LogicalJoin(condition=[=($0, $3)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, src]])
+ +- LogicalTableScan(table=[[default_catalog, default_database, changelog_src]])
+
+== Optimized Physical Plan ==
+Join(joinType=[LeftOuterJoin], where=[=(id, id0)], select=[id, b, c, id0, a], leftInputSpec=[NoUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
+:- Exchange(distribution=[hash[id]])
+: +- Calc(select=[id, b, c], where=[>(c, 1)])
+: +- TableSourceScan(table=[[default_catalog, default_database, src, filter=[]]], fields=[id, b, c])
++- Exchange(distribution=[hash[id]])
+ +- ChangelogNormalize(key=[id])
+ +- Exchange(distribution=[hash[id]])
+ +- TableSourceScan(table=[[default_catalog, default_database, changelog_src]], fields=[id, a])
+
+== Optimized Execution Plan ==
+Join(joinType=[LeftOuterJoin], where=[(id = id0)], select=[id, b, c, id0, a], leftInputSpec=[NoUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
+:- Exchange(distribution=[hash[id]])
+: +- Calc(select=[id, b, c], where=[(c > 1)])
+: +- TableSourceScan(table=[[default_catalog, default_database, src, filter=[]]], fields=[id, b, c])
++- Exchange(distribution=[hash[id]])
+ +- ChangelogNormalize(key=[id])
+ +- Exchange(distribution=[hash[id]])
+ +- TableSourceScan(table=[[default_catalog, default_database, changelog_src]], fields=[id, a])
+
+== Physical Execution Plan ==
+{
+ "nodes" : [ {
+ "id" : ,
+ "type" : "Source: src[]",
+ "pact" : "Data Source",
+ "contents" : "[]:TableSourceScan(table=[[default_catalog, default_database, src, filter=[]]], fields=[id, b, c])",
+ "parallelism" : 3
+ }, {
+ "id" : ,
+ "type" : "Calc[]",
+ "pact" : "Operator",
+ "contents" : "[]:Calc(select=[id, b, c], where=[(c > 1)])",
+ "parallelism" : 10,
+ "predecessors" : [ {
+ "id" : ,
+ "ship_strategy" : "REBALANCE",
+ "side" : "second"
+ } ]
+ }, {
+ "id" : ,
+ "type" : "Source: changelog_src[]",
+ "pact" : "Data Source",
+ "contents" : "[]:TableSourceScan(table=[[default_catalog, default_database, changelog_src]], fields=[id, a])",
+ "parallelism" : 5
+ }, {
+ "id" : ,
+ "type" : "ChangelogNormalize[]",
+ "pact" : "Operator",
+ "contents" : "[]:ChangelogNormalize(key=[id])",
+ "parallelism" : 10,
+ "predecessors" : [ {
+ "id" : ,
+ "ship_strategy" : "HASH",
+ "side" : "second"
+ } ]
+ }, {
+ "id" : ,
+ "type" : "Join[]",
+ "pact" : "Operator",
+ "contents" : "[]:Join(joinType=[LeftOuterJoin], where=[(id = id0)], select=[id, b, c, id0, a], leftInputSpec=[NoUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])",
+ "parallelism" : 10,
+ "predecessors" : [ {
+ "id" : ,
+ "ship_strategy" : "HASH",
+ "side" : "second"
+ }, {
+ "id" : ,
+ "ship_strategy" : "HASH",
+ "side" : "second"
+ } ]
+ } ]
+}]]>
+ </Resource>
+ </TestCase>
<TestCase name="testUnionUpsertSourceAndAggregation">
<Resource name="sql">
<![CDATA[
@@ -634,8 +718,8 @@ LogicalProject(a=[$1], b=[$2], c=[$3])
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
-Calc(select=[a, b, c], where=[>(a, 1)], changelogMode=[I,UB,UA,D])
-+- ChangelogNormalize(key=[id], changelogMode=[I,UB,UA,D])
+Calc(select=[a, b, c], changelogMode=[I,UB,UA,D])
++- ChangelogNormalize(key=[id], condition=[>(a, 1)], changelogMode=[I,UB,UA,D])
+- Exchange(distribution=[hash[id]], changelogMode=[UA,D])
+- Calc(select=[id, a, b, c], changelogMode=[UA,D])
+- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL SECOND)], changelogMode=[UA,D])
@@ -711,89 +795,4 @@ Calc(select=[ts, a, b], where=[>(a, 1)], changelogMode=[I,UB,UA,D])
]]>
</Resource>
</TestCase>
-
- <TestCase name="testSetParallelismForSource">
- <Resource name="explain">
- <![CDATA[== Abstract Syntax Tree ==
-LogicalProject(id=[$0], b=[$1], c=[$2], id0=[$3], a=[$4])
-+- LogicalFilter(condition=[>($2, 1)])
- +- LogicalJoin(condition=[=($0, $3)], joinType=[left])
- :- LogicalTableScan(table=[[default_catalog, default_database, src]])
- +- LogicalTableScan(table=[[default_catalog, default_database, changelog_src]])
-
-== Optimized Physical Plan ==
-Join(joinType=[LeftOuterJoin], where=[=(id, id0)], select=[id, b, c, id0, a], leftInputSpec=[NoUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
-:- Exchange(distribution=[hash[id]])
-: +- Calc(select=[id, b, c], where=[>(c, 1)])
-: +- TableSourceScan(table=[[default_catalog, default_database, src, filter=[]]], fields=[id, b, c])
-+- Exchange(distribution=[hash[id]])
- +- ChangelogNormalize(key=[id])
- +- Exchange(distribution=[hash[id]])
- +- TableSourceScan(table=[[default_catalog, default_database, changelog_src]], fields=[id, a])
-
-== Optimized Execution Plan ==
-Join(joinType=[LeftOuterJoin], where=[(id = id0)], select=[id, b, c, id0, a], leftInputSpec=[NoUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])
-:- Exchange(distribution=[hash[id]])
-: +- Calc(select=[id, b, c], where=[(c > 1)])
-: +- TableSourceScan(table=[[default_catalog, default_database, src, filter=[]]], fields=[id, b, c])
-+- Exchange(distribution=[hash[id]])
- +- ChangelogNormalize(key=[id])
- +- Exchange(distribution=[hash[id]])
- +- TableSourceScan(table=[[default_catalog, default_database, changelog_src]], fields=[id, a])
-
-== Physical Execution Plan ==
-{
- "nodes" : [ {
- "id" : ,
- "type" : "Source: src[]",
- "pact" : "Data Source",
- "contents" : "[]:TableSourceScan(table=[[default_catalog, default_database, src, filter=[]]], fields=[id, b, c])",
- "parallelism" : 3
- }, {
- "id" : ,
- "type" : "Calc[]",
- "pact" : "Operator",
- "contents" : "[]:Calc(select=[id, b, c], where=[(c > 1)])",
- "parallelism" : 10,
- "predecessors" : [ {
- "id" : ,
- "ship_strategy" : "REBALANCE",
- "side" : "second"
- } ]
- }, {
- "id" : ,
- "type" : "Source: changelog_src[]",
- "pact" : "Data Source",
- "contents" : "[]:TableSourceScan(table=[[default_catalog, default_database, changelog_src]], fields=[id, a])",
- "parallelism" : 5
- }, {
- "id" : ,
- "type" : "ChangelogNormalize[]",
- "pact" : "Operator",
- "contents" : "[]:ChangelogNormalize(key=[id])",
- "parallelism" : 10,
- "predecessors" : [ {
- "id" : ,
- "ship_strategy" : "HASH",
- "side" : "second"
- } ]
- }, {
- "id" : ,
- "type" : "Join[]",
- "pact" : "Operator",
- "contents" : "[]:Join(joinType=[LeftOuterJoin], where=[(id = id0)], select=[id, b, c, id0, a], leftInputSpec=[NoUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey])",
- "parallelism" : 10,
- "predecessors" : [ {
- "id" : ,
- "ship_strategy" : "HASH",
- "side" : "second"
- }, {
- "id" : ,
- "ship_strategy" : "HASH",
- "side" : "second"
- } ]
- } ]
-}]]>
- </Resource>
- </TestCase>
</Root>
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-changelog-normalize_1/changelog-normalize-upsert-filter/plan/changelog-normalize-upsert-filter.json b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-changelog-normalize_1/changelog-normalize-upsert-filter/plan/changelog-normalize-upsert-filter.json
new file mode 100644
index 00000000000..6ed510d2b82
--- /dev/null
+++ b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-changelog-normalize_1/changelog-normalize-upsert-filter/plan/changelog-normalize-upsert-filter.json
@@ -0,0 +1,155 @@
+{
+ "flinkVersion" : "2.1",
+ "nodes" : [ {
+ "id" : 1,
+ "type" : "stream-exec-table-source-scan_1",
+ "scanTableSource" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`source_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "VARCHAR(2147483647) NOT NULL"
+ }, {
+ "name" : "b",
+ "dataType" : "INT NOT NULL"
+ }, {
+ "name" : "c",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ],
+ "primaryKey" : {
+ "name" : "PK_a",
+ "type" : "PRIMARY_KEY",
+ "columns" : [ "a" ]
+ }
+ },
+ "partitionKeys" : [ ]
+ }
+ },
+ "abilities" : [ {
+ "type" : "FilterPushDown",
+ "predicates" : [ ]
+ } ]
+ },
+ "outputType" : "ROW<`a` VARCHAR(2147483647) NOT NULL, `b` INT NOT NULL, `c` VARCHAR(2147483647)>",
+ "description" : "TableSourceScan(table=[[default_catalog, default_database, source_t, filter=[]]], fields=[a, b, c])",
+ "inputProperties" : [ ]
+ }, {
+ "id" : 2,
+ "type" : "stream-exec-exchange_1",
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "HASH",
+ "keys" : [ 0 ]
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` VARCHAR(2147483647) NOT NULL, `b` INT NOT NULL, `c` VARCHAR(2147483647)>",
+ "description" : "Exchange(distribution=[hash[a]])"
+ }, {
+ "id" : 3,
+ "type" : "stream-exec-changelog-normalize_1",
+ "configuration" : {
+ "table.exec.mini-batch.enabled" : "false",
+ "table.exec.mini-batch.size" : "-1"
+ },
+ "uniqueKeys" : [ 0 ],
+ "generateUpdateBefore" : true,
+ "state" : [ {
+ "index" : 0,
+ "ttl" : "0 ms",
+ "name" : "changelogNormalizeState"
+ } ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` VARCHAR(2147483647) NOT NULL, `b` INT NOT NULL, `c` VARCHAR(2147483647)>",
+ "description" : "ChangelogNormalize(key=[a], condition=[(b < 10)])",
+ "filterCondition" : {
+ "kind" : "CALL",
+ "syntax" : "BINARY",
+ "internalName" : "$<$1",
+ "operands" : [ {
+ "kind" : "INPUT_REF",
+ "inputIndex" : 1,
+ "type" : "INT NOT NULL"
+ }, {
+ "kind" : "LITERAL",
+ "value" : 10,
+ "type" : "INT NOT NULL"
+ } ],
+ "type" : "BOOLEAN NOT NULL"
+ }
+ }, {
+ "id" : 4,
+ "type" : "stream-exec-sink_1",
+ "configuration" : {
+ "table.exec.sink.keyed-shuffle" : "AUTO",
+ "table.exec.sink.not-null-enforcer" : "ERROR",
+ "table.exec.sink.rowtime-inserter" : "ENABLED",
+ "table.exec.sink.type-length-enforcer" : "IGNORE",
+ "table.exec.sink.upsert-materialize" : "AUTO"
+ },
+ "dynamicTableSink" : {
+ "table" : {
+ "identifier" : "`default_catalog`.`default_database`.`sink_t`",
+ "resolvedTable" : {
+ "schema" : {
+ "columns" : [ {
+ "name" : "a",
+ "dataType" : "VARCHAR(2147483647)"
+ }, {
+ "name" : "b",
+ "dataType" : "INT"
+ }, {
+ "name" : "c",
+ "dataType" : "VARCHAR(2147483647)"
+ } ],
+ "watermarkSpecs" : [ ]
+ },
+ "partitionKeys" : [ ]
+ }
+ }
+ },
+ "inputChangelogMode" : [ "INSERT", "UPDATE_BEFORE", "UPDATE_AFTER", "DELETE" ],
+ "inputUpsertKey" : [ 0 ],
+ "inputProperties" : [ {
+ "requiredDistribution" : {
+ "type" : "UNKNOWN"
+ },
+ "damBehavior" : "PIPELINED",
+ "priority" : 0
+ } ],
+ "outputType" : "ROW<`a` VARCHAR(2147483647) NOT NULL, `b` INT NOT NULL, `c` VARCHAR(2147483647)>",
+ "description" : "Sink(table=[default_catalog.default_database.sink_t], fields=[a, b, c])"
+ } ],
+ "edges" : [ {
+ "source" : 1,
+ "target" : 2,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 2,
+ "target" : 3,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ }, {
+ "source" : 3,
+ "target" : 4,
+ "shuffle" : {
+ "type" : "FORWARD"
+ },
+ "shuffleMode" : "PIPELINED"
+ } ]
+}
\ No newline at end of file
diff --git a/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-changelog-normalize_1/changelog-normalize-upsert-filter/savepoint/_metadata b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-changelog-normalize_1/changelog-normalize-upsert-filter/savepoint/_metadata
new file mode 100644
index 00000000000..2aed923053e
Binary files /dev/null and b/flink-table/flink-table-planner/src/test/resources/restore-tests/stream-exec-changelog-normalize_1/changelog-normalize-upsert-filter/savepoint/_metadata differ
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeDeduplicateKeepLastRowFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeDeduplicateKeepLastRowFunction.java
index f6eaf03e584..efac322c782 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeDeduplicateKeepLastRowFunction.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeDeduplicateKeepLastRowFunction.java
@@ -19,13 +19,17 @@
package org.apache.flink.table.runtime.operators.deduplicate;
import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.FilterCondition;
+import org.apache.flink.table.runtime.generated.GeneratedFilterCondition;
import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser;
import org.apache.flink.table.runtime.generated.RecordEqualiser;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.util.Collector;
import static org.apache.flink.table.runtime.operators.deduplicate.utils.DeduplicateFunctionHelper.processLastRowOnChangelog;
+import static org.apache.flink.table.runtime.operators.deduplicate.utils.DeduplicateFunctionHelper.processLastRowOnChangelogWithFilter;
import static org.apache.flink.table.runtime.operators.deduplicate.utils.DeduplicateFunctionHelper.processLastRowOnProcTime;
/** This function is used to deduplicate on keys and keeps only last row. */
@@ -41,21 +45,27 @@ public class ProcTimeDeduplicateKeepLastRowFunction
/** The code generated equaliser used to equal RowData. */
private final GeneratedRecordEqualiser genRecordEqualiser;
+ private final GeneratedFilterCondition genFilterCondition;
+
/** The record equaliser used to equal RowData. */
private transient RecordEqualiser equaliser;
+ private transient FilterCondition filterCondition;
+
public ProcTimeDeduplicateKeepLastRowFunction(
InternalTypeInfo<RowData> typeInfo,
long stateRetentionTime,
boolean generateUpdateBefore,
boolean generateInsert,
boolean inputInsertOnly,
- GeneratedRecordEqualiser genRecordEqualiser) {
+ GeneratedRecordEqualiser genRecordEqualiser,
+ GeneratedFilterCondition filterCondition) {
super(typeInfo, null, stateRetentionTime);
this.generateUpdateBefore = generateUpdateBefore;
this.generateInsert = generateInsert;
this.inputIsInsertOnly = inputInsertOnly;
this.genRecordEqualiser = genRecordEqualiser;
+ this.genFilterCondition = filterCondition;
this.isStateTtlEnabled = stateRetentionTime > 0;
}
@@ -63,6 +73,12 @@ public class ProcTimeDeduplicateKeepLastRowFunction
public void open(OpenContext openContext) throws Exception {
super.open(openContext);
equaliser = genRecordEqualiser.newInstance(getRuntimeContext().getUserCodeClassLoader());
+ if (genFilterCondition != null) {
+ filterCondition =
+ genFilterCondition.newInstance(getRuntimeContext().getUserCodeClassLoader());
+ FunctionUtils.setFunctionRuntimeContext(filterCondition, getRuntimeContext());
+ FunctionUtils.openFunction(filterCondition, openContext);
+ }
}
@Override
@@ -77,6 +93,15 @@ public class ProcTimeDeduplicateKeepLastRowFunction
out,
isStateTtlEnabled,
equaliser);
+ } else if (filterCondition != null) {
+ processLastRowOnChangelogWithFilter(
+ input,
+ generateUpdateBefore,
+ state,
+ out,
+ isStateTtlEnabled,
+ equaliser,
+ filterCondition);
} else {
processLastRowOnChangelog(
input, generateUpdateBefore, state, out, isStateTtlEnabled, equaliser);
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeMiniBatchDeduplicateKeepLastRowFunction.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeMiniBatchDeduplicateKeepLastRowFunction.java
index 3b71620aedb..f081ca0e0a3 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeMiniBatchDeduplicateKeepLastRowFunction.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeMiniBatchDeduplicateKeepLastRowFunction.java
@@ -18,9 +18,12 @@
package org.apache.flink.table.runtime.operators.deduplicate;
+import org.apache.flink.api.common.functions.util.FunctionUtils;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.runtime.context.ExecutionContext;
+import org.apache.flink.table.runtime.generated.FilterCondition;
+import org.apache.flink.table.runtime.generated.GeneratedFilterCondition;
import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser;
import org.apache.flink.table.runtime.generated.RecordEqualiser;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
@@ -31,6 +34,7 @@ import javax.annotation.Nullable;
import java.util.Map;
import static org.apache.flink.table.runtime.operators.deduplicate.utils.DeduplicateFunctionHelper.processLastRowOnChangelog;
+import static org.apache.flink.table.runtime.operators.deduplicate.utils.DeduplicateFunctionHelper.processLastRowOnChangelogWithFilter;
import static org.apache.flink.table.runtime.operators.deduplicate.utils.DeduplicateFunctionHelper.processLastRowOnProcTime;
/** This function is used to get the last row for every key partition in miniBatch mode. */
@@ -45,10 +49,13 @@ public class ProcTimeMiniBatchDeduplicateKeepLastRowFunction
private final boolean isStateTtlEnabled;
// The code generated equaliser used to equal RowData.
private final GeneratedRecordEqualiser genRecordEqualiser;
+ private final GeneratedFilterCondition genFilterCondition;
// The record equaliser used to equal RowData.
private transient RecordEqualiser equaliser;
+ private transient FilterCondition filterCondition;
+
public ProcTimeMiniBatchDeduplicateKeepLastRowFunction(
InternalTypeInfo<RowData> typeInfo,
TypeSerializer<RowData> serializer,
@@ -56,13 +63,15 @@ public class ProcTimeMiniBatchDeduplicateKeepLastRowFunction
boolean generateUpdateBefore,
boolean generateInsert,
boolean inputInsertOnly,
- GeneratedRecordEqualiser genRecordEqualiser) {
+ GeneratedRecordEqualiser genRecordEqualiser,
+ GeneratedFilterCondition genFilterCondition) {
super(typeInfo, stateRetentionTime);
this.serializer = serializer;
this.generateUpdateBefore = generateUpdateBefore;
this.generateInsert = generateInsert;
this.inputInsertOnly = inputInsertOnly;
this.genRecordEqualiser = genRecordEqualiser;
+ this.genFilterCondition = genFilterCondition;
this.isStateTtlEnabled = stateRetentionTime > 0;
}
@@ -71,6 +80,12 @@ public class ProcTimeMiniBatchDeduplicateKeepLastRowFunction
super.open(ctx);
equaliser =
genRecordEqualiser.newInstance(ctx.getRuntimeContext().getUserCodeClassLoader());
+ if (genFilterCondition != null) {
+ filterCondition =
+ genFilterCondition.newInstance(
+ ctx.getRuntimeContext().getUserCodeClassLoader());
+ FunctionUtils.setFunctionRuntimeContext(filterCondition, ctx.getRuntimeContext());
+ }
}
@Override
@@ -95,6 +110,15 @@ public class ProcTimeMiniBatchDeduplicateKeepLastRowFunction
out,
isStateTtlEnabled,
equaliser);
+ } else if (filterCondition != null) {
+ processLastRowOnChangelogWithFilter(
+ currentRow,
+ generateUpdateBefore,
+ state,
+ out,
+ isStateTtlEnabled,
+ equaliser,
+ filterCondition);
} else {
processLastRowOnChangelog(
currentRow, generateUpdateBefore, state, out, isStateTtlEnabled, equaliser);
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/utils/DeduplicateFunctionHelper.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/utils/DeduplicateFunctionHelper.java
index e9c7f81b51e..22d6f97c4a0 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/utils/DeduplicateFunctionHelper.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/deduplicate/utils/DeduplicateFunctionHelper.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.runtime.operators.deduplicate.utils;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.FilterCondition;
import org.apache.flink.table.runtime.generated.RecordEqualiser;
import org.apache.flink.types.RowKind;
import org.apache.flink.util.Collector;
@@ -152,6 +153,73 @@ public class DeduplicateFunctionHelper {
}
}
+ public static void processLastRowOnChangelogWithFilter(
+ RowData currentRow,
+ boolean generateUpdateBefore,
+ ValueState<RowData> state,
+ Collector<RowData> out,
+ boolean isStateTtlEnabled,
+ RecordEqualiser equaliser,
+ FilterCondition filterCondition)
+ throws Exception {
+ RowData preRow = state.value();
+ RowKind currentKind = currentRow.getRowKind();
+ if (currentKind == RowKind.INSERT || currentKind == RowKind.UPDATE_AFTER) {
+ if (preRow == null) {
+ if (filterCondition.apply(currentRow)) {
+ // the first row, send INSERT message
+ currentRow.setRowKind(RowKind.INSERT);
+ out.collect(currentRow);
+ } else {
+ // return, do not update the state
+ return;
+ }
+ } else {
+ if (!isStateTtlEnabled && equaliser.equals(preRow, currentRow)) {
+ // currentRow is the same as preRow and state cleaning is not enabled.
+ // We do not emit retraction and update message.
+ // If state cleaning is enabled, we have to emit messages to prevent too early
+ // state eviction of downstream operators.
+ return;
+ } else {
+ if (filterCondition.apply(currentRow)) {
+ if (generateUpdateBefore) {
+ preRow.setRowKind(RowKind.UPDATE_BEFORE);
+ out.collect(preRow);
+ }
+ currentRow.setRowKind(RowKind.UPDATE_AFTER);
+ out.collect(currentRow);
+ } else {
+ // generate retraction, because the row does not match any longer
+ preRow.setRowKind(RowKind.DELETE);
+ out.collect(preRow);
+ // clear the state, there is no row we will need to retract
+ state.clear();
+ return;
+ }
+ }
+ }
+ // normalize row kind
+ currentRow.setRowKind(RowKind.INSERT);
+ // save to state
+ state.update(currentRow);
+ } else {
+ // DELETE or UPDATER_BEFORE
+ if (preRow != null) {
+ // always set to DELETE because this row has been removed
+ // even the input is UPDATE_BEFORE, there may no UPDATE_AFTER after it.
+ preRow.setRowKind(RowKind.DELETE);
+ // output the preRow instead of currentRow,
+ // because preRow always contains the full content.
+ // currentRow may only contain key parts (e.g. Kafka tombstone records).
+ out.collect(preRow);
+ // clear state as the row has been removed
+ state.clear();
+ }
+ // nothing to do if removing a non-existed row
+ }
+ }
+
/**
* Processes element to deduplicate on keys with process time semantic, sends current element if
* it is first row.
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeDeduplicateFunctionTestBase.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeDeduplicateFunctionTestBase.java
index 3b8fb14d2c1..07f3aa1d585 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeDeduplicateFunctionTestBase.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeDeduplicateFunctionTestBase.java
@@ -18,7 +18,12 @@
package org.apache.flink.table.runtime.operators.deduplicate;
+import org.apache.flink.api.common.functions.IterationRuntimeContext;
+import org.apache.flink.api.common.functions.OpenContext;
+import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.generated.FilterCondition;
+import org.apache.flink.table.runtime.generated.GeneratedFilterCondition;
import org.apache.flink.table.runtime.generated.GeneratedRecordEqualiser;
import org.apache.flink.table.runtime.generated.RecordEqualiser;
import org.apache.flink.table.runtime.keyselector.RowDataKeySelector;
@@ -61,4 +66,38 @@ abstract class ProcTimeDeduplicateFunctionTestBase {
return new RowDataRecordEqualiser();
}
};
+
+ static GeneratedFilterCondition generatedFilterCondition =
+ new GeneratedFilterCondition("", "", new Object[0]) {
+ @Override
+ public FilterCondition newInstance(ClassLoader classLoader) {
+ return new TestingFilter();
+ }
+ };
+
+ private static class TestingFilter implements FilterCondition {
+ @Override
+ public boolean apply(RowData input) {
+ return input.getInt(2) > 10;
+ }
+
+ @Override
+ public void open(OpenContext openContext) throws Exception {}
+
+ @Override
+ public void close() throws Exception {}
+
+ @Override
+ public RuntimeContext getRuntimeContext() {
+ return null;
+ }
+
+ @Override
+ public IterationRuntimeContext getIterationRuntimeContext() {
+ return null;
+ }
+
+ @Override
+ public void setRuntimeContext(RuntimeContext t) {}
+ }
}
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeDeduplicateKeepLastRowFunctionTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeDeduplicateKeepLastRowFunctionTest.java
index 0dd4f9d4b1f..b228985400e 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeDeduplicateKeepLastRowFunctionTest.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeDeduplicateKeepLastRowFunctionTest.java
@@ -24,10 +24,13 @@ import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.table.data.RowData;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.ArrayList;
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.updateAfterRecord;
import static org.apache.flink.table.runtime.util.StreamRecordUtils.updateBeforeRecord;
@@ -37,7 +40,13 @@ class ProcTimeDeduplicateKeepLastRowFunctionTest extends ProcTimeDeduplicateFunc
private ProcTimeDeduplicateKeepLastRowFunction createFunctionWithoutStateTtl(
boolean generateUpdateBefore, boolean generateInsert) {
return new ProcTimeDeduplicateKeepLastRowFunction(
- inputRowType, 0, generateUpdateBefore, generateInsert, true, generatedEqualiser);
+ inputRowType,
+ 0,
+ generateUpdateBefore,
+ generateInsert,
+ true,
+ generatedEqualiser,
+ null);
}
private ProcTimeDeduplicateKeepLastRowFunction createFunction(
@@ -48,7 +57,20 @@ class ProcTimeDeduplicateKeepLastRowFunctionTest extends ProcTimeDeduplicateFunc
generateUpdateBefore,
generateInsert,
true,
- generatedEqualiser);
+ generatedEqualiser,
+ null);
+ }
+
+ private ProcTimeDeduplicateKeepLastRowFunction createFunctionWithFilter(
+ boolean generateUpdateBefore) {
+ return new ProcTimeDeduplicateKeepLastRowFunction(
+ inputRowType,
+ minTime.toMillis(),
+ generateUpdateBefore,
+ true,
+ false,
+ generatedEqualiser,
+ generatedFilterCondition);
}
private OneInputStreamOperatorTestHarness<RowData, RowData> createTestHarness(
@@ -158,4 +180,30 @@ class ProcTimeDeduplicateKeepLastRowFunctionTest extends ProcTimeDeduplicateFunc
assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
testHarness.close();
}
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testWithFilterCondition(boolean updateBefore) throws Exception {
+ ProcTimeDeduplicateKeepLastRowFunction func = createFunctionWithFilter(updateBefore);
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(func);
+ testHarness.open();
+ testHarness.processElement(updateAfterRecord("book", 1L, 12));
+ testHarness.processElement(updateAfterRecord("book", 1L, 15));
+ testHarness.processElement(updateAfterRecord("book", 1L, 8));
+ testHarness.processElement(updateAfterRecord("book", 1L, 9));
+ testHarness.processElement(updateAfterRecord("book", 1L, 13));
+ testHarness.processElement(deleteRecord("book", 1L, null));
+
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(insertRecord("book", 1L, 12));
+ if (updateBefore) {
+ expectedOutput.add(updateBeforeRecord("book", 1L, 12));
+ }
+ expectedOutput.add(updateAfterRecord("book", 1L, 15));
+ expectedOutput.add(deleteRecord("book", 1L, 15));
+ expectedOutput.add(insertRecord("book", 1L, 13));
+ expectedOutput.add(deleteRecord("book", 1L, 13));
+ assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
+ testHarness.close();
+ }
}
diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeMiniBatchDeduplicateKeepLastRowFunctionTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeMiniBatchDeduplicateKeepLastRowFunctionTest.java
index 7eeeb75edae..2700227fda6 100644
--- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeMiniBatchDeduplicateKeepLastRowFunctionTest.java
+++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/deduplicate/ProcTimeMiniBatchDeduplicateKeepLastRowFunctionTest.java
@@ -28,10 +28,13 @@ import org.apache.flink.table.runtime.operators.bundle.KeyedMapBundleOperator;
import org.apache.flink.table.runtime.operators.bundle.trigger.CountBundleTrigger;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.util.ArrayList;
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.updateAfterRecord;
import static org.apache.flink.table.runtime.util.StreamRecordUtils.updateBeforeRecord;
@@ -54,7 +57,22 @@ class ProcTimeMiniBatchDeduplicateKeepLastRowFunctionTest
generateUpdateBefore,
generateInsert,
true,
- generatedEqualiser);
+ generatedEqualiser,
+ null);
+ }
+
+ private ProcTimeMiniBatchDeduplicateKeepLastRowFunction createFunctionWithFilter(
+ boolean generateUpdateBefore) {
+
+ return new ProcTimeMiniBatchDeduplicateKeepLastRowFunction(
+ inputRowType,
+ typeSerializer,
+ minTime.toMillis(),
+ generateUpdateBefore,
+ false,
+ false,
+ generatedEqualiser,
+ generatedFilterCondition);
}
private OneInputStreamOperatorTestHarness<RowData, RowData> createTestHarness(
@@ -182,4 +200,60 @@ class ProcTimeMiniBatchDeduplicateKeepLastRowFunctionTest
expectedOutput.add(insertRecord("book", 2L, 18));
assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testWithFilterCondition(boolean updateBefore) throws Exception {
+ ProcTimeMiniBatchDeduplicateKeepLastRowFunction func =
+ createFunctionWithFilter(updateBefore);
+ OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(func);
+ testHarness.open();
+ testHarness.processElement(updateAfterRecord("book", 1L, 12));
+ testHarness.processElement(updateAfterRecord("book", 1L, 15));
+ // output is empty because bundle not trigger yet.
+ assertThat(testHarness.getOutput()).isEmpty();
+ testHarness.processElement(updateAfterRecord("book", 1L, 16));
+ // first bundle end
+
+ testHarness.processElement(updateAfterRecord("book", 1L, 16));
+ testHarness.processElement(updateAfterRecord("book", 1L, 18));
+ testHarness.processElement(updateAfterRecord("book", 1L, 19));
+ // second bundle end
+
+ testHarness.processElement(updateAfterRecord("book", 1L, 8));
+ testHarness.processElement(updateAfterRecord("book", 1L, 9));
+ testHarness.processElement(updateAfterRecord("book", 1L, 7));
+ // third bundle end
+
+ testHarness.processElement(updateAfterRecord("book", 1L, 21));
+ testHarness.processElement(updateAfterRecord("book", 1L, 22));
+ testHarness.processElement(updateAfterRecord("book", 1L, 23));
+ // fourth bundle end
+
+ testHarness.processElement(updateAfterRecord("book", 1L, 23));
+ testHarness.processElement(updateAfterRecord("book", 1L, 23));
+ testHarness.processElement(deleteRecord("book", 1L, null));
+ // fifth bundle end
+
+ List<Object> expectedOutput = new ArrayList<>();
+ // first bundle result
+ expectedOutput.add(insertRecord("book", 1L, 16));
+
+ // second bundle result
+ if (updateBefore) {
+ expectedOutput.add(updateBeforeRecord("book", 1L, 16));
+ }
+ expectedOutput.add(updateAfterRecord("book", 1L, 19));
+
+ // third bundle result
+ expectedOutput.add(deleteRecord("book", 1L, 19));
+
+ // fourth bundle result
+ expectedOutput.add(insertRecord("book", 1L, 23));
+
+ // fifth bundle result
+ expectedOutput.add(deleteRecord("book", 1L, 23));
+ assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
+ testHarness.close();
+ }
}