You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ku...@apache.org on 2019/04/28 13:04:27 UTC
[flink] branch master updated: [FLINK-12307][table-planner-blink]
Support translation from StreamExecWindowJoin to StreamTransformation.
This is an automated email from the ASF dual-hosted git repository.
kurt 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 177e7ff [FLINK-12307][table-planner-blink] Support translation from StreamExecWindowJoin to StreamTransformation.
177e7ff is described below
commit 177e7ff1291f397294f7b7dab1d791a07adc6a5a
Author: Jing Zhang <be...@126.com>
AuthorDate: Sun Apr 28 21:04:16 2019 +0800
[FLINK-12307][table-planner-blink] Support translation from StreamExecWindowJoin to StreamTransformation.
This closes #8261
---
.../JoinTypeUtil.java} | 75 +-
.../metadata/FlinkRelMdModifiedMonotonicity.scala | 9 +-
.../plan/nodes/common/CommonPhysicalJoin.scala | 13 +-
.../physical/batch/BatchExecSortMergeJoin.scala | 26 +-
.../nodes/physical/stream/StreamExecJoinBase.scala | 8 +-
.../physical/stream/StreamExecWindowJoin.scala | 260 +++++-
.../physical/batch/BatchExecHashJoinRule.scala | 18 +-
.../physical/batch/BatchExecJoinRuleBase.scala | 10 +-
.../batch/BatchExecSingleRowJoinRule.scala | 14 +-
.../flink/table/plan/util/RelExplainUtil.scala | 15 -
.../flink/table/plan/util/WindowJoinUtil.scala | 80 +-
.../runtime/stream/sql/DeduplicateITCase.scala | 1 -
.../runtime/stream/sql/WindowJoinITCase.scala | 950 +++++++++++++++++++++
.../table/runtime/join/EmitAwareCollector.java | 56 ++
.../flink/table/runtime/join/FlinkJoinType.java} | 56 +-
.../KeyedCoProcessOperatorWithWatermarkDelay.java | 60 ++
.../table/runtime/join/OuterJoinPaddingUtil.java | 91 ++
.../runtime/join/ProcTimeBoundedStreamJoin.java | 63 ++
.../runtime/join/RowTimeBoundedStreamJoin.java | 83 ++
.../table/runtime/join/SortMergeJoinOperator.java | 24 +-
.../table/runtime/join/TimeBoundedStreamJoin.java | 477 +++++++++++
.../runtime/rank/RetractableTopNFunction.java | 2 +-
.../DeduplicateKeepFirstRowFunctionTest.java | 8 +-
.../DeduplicateKeepLastRowFunctionTest.java | 22 +-
...niBatchDeduplicateKeepFirstRowFunctionTest.java | 8 +-
...iniBatchDeduplicateKeepLastRowFunctionTest.java | 36 +-
.../join/Int2SortMergeJoinOperatorTest.java | 19 +-
.../join/ProcTimeBoundedStreamJoinTest.java | 178 ++++
.../runtime/join/RandomSortMergeInnerJoinTest.java | 3 +-
.../runtime/join/RandomSortMergeOuterJoinTest.java | 17 +-
.../runtime/join/RowTimeBoundedStreamJoinTest.java | 387 +++++++++
.../join/String2SortMergeJoinOperatorTest.java | 11 +-
.../join/TimeBoundedStreamJoinTestBase.java | 63 ++
.../runtime/rank/AppendOnlyTopNFunctionTest.java | 18 +-
.../runtime/rank/RetractableTopNFunctionTest.java | 144 ++--
.../table/runtime/rank/TopNFunctionTestBase.java | 144 ++--
.../runtime/rank/UpdatableTopNFunctionTest.java | 150 ++--
.../runtime/sort/ProcTimeSortOperatorTest.java | 34 +-
.../runtime/sort/RowTimeSortOperatorTest.java | 82 +-
.../table/runtime/sort/StreamSortOperatorTest.java | 28 +-
.../table/runtime/window/WindowOperatorTest.java | 336 ++++----
41 files changed, 3358 insertions(+), 721 deletions(-)
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/plan/FlinkJoinRelType.java b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/plan/util/JoinTypeUtil.java
similarity index 54%
copy from flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/plan/FlinkJoinRelType.java
copy to flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/plan/util/JoinTypeUtil.java
index 99cd7ea..53c8d41 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/plan/FlinkJoinRelType.java
+++ b/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/plan/util/JoinTypeUtil.java
@@ -16,76 +16,53 @@
* limitations under the License.
*/
-package org.apache.flink.table.plan;
+package org.apache.flink.table.plan.util;
+
+import org.apache.flink.table.runtime.join.FlinkJoinType;
import org.apache.calcite.rel.core.Join;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.core.SemiJoin;
/**
- * Enumeration of join types.
+ * Utility for {@link FlinkJoinType}.
*/
-public enum FlinkJoinRelType {
- INNER, LEFT, RIGHT, FULL, SEMI, ANTI;
-
- public boolean isOuter() {
- switch (this) {
- case LEFT:
- case RIGHT:
- case FULL:
- return true;
- default:
- return false;
- }
- }
-
- public boolean isLeftOuter() {
- switch (this) {
- case LEFT:
- case FULL:
- return true;
- default:
- return false;
- }
- }
-
- public boolean isRightOuter() {
- switch (this) {
- case RIGHT:
- case FULL:
- return true;
- default:
- return false;
- }
- }
+public class JoinTypeUtil {
- /** Convert JoinRelType to FlinkJoinRelType. */
- public static FlinkJoinRelType toFlinkJoinRelType(JoinRelType joinType) {
- switch (joinType) {
+ /**
+ * Converts {@link JoinRelType} to {@link FlinkJoinType}.
+ */
+ public static FlinkJoinType toFlinkJoinType(JoinRelType joinRelType) {
+ switch (joinRelType) {
case INNER:
- return FlinkJoinRelType.INNER;
+ return FlinkJoinType.INNER;
case LEFT:
- return FlinkJoinRelType.LEFT;
+ return FlinkJoinType.LEFT;
case RIGHT:
- return FlinkJoinRelType.RIGHT;
+ return FlinkJoinType.RIGHT;
case FULL:
- return FlinkJoinRelType.FULL;
+ return FlinkJoinType.FULL;
default:
- throw new IllegalArgumentException("invalid: " + joinType);
+ throw new IllegalArgumentException("invalid: " + joinRelType);
}
}
- public static FlinkJoinRelType getFlinkJoinRelType(Join join) {
+ /**
+ * Gets {@link FlinkJoinType} of the input Join RelNode.
+ */
+ public static FlinkJoinType getFlinkJoinType(Join join) {
if (join instanceof SemiJoin) {
// TODO supports ANTI
- return SEMI;
+ return FlinkJoinType.SEMI;
} else {
- return toFlinkJoinRelType(join.getJoinType());
+ return toFlinkJoinType(join.getJoinType());
}
}
- /** Convert FlinkJoinRelType to JoinRelType. */
- public static JoinRelType toJoinRelType(FlinkJoinRelType joinType) {
+ /**
+ * Converts {@link FlinkJoinType} to {@link JoinRelType}.
+ */
+ public static JoinRelType toJoinRelType(FlinkJoinType joinType) {
switch (joinType) {
case INNER:
return JoinRelType.INNER;
@@ -99,5 +76,5 @@ public enum FlinkJoinRelType {
throw new IllegalArgumentException("invalid: " + joinType);
}
}
-}
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/metadata/FlinkRelMdModifiedMonotonicity.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/metadata/FlinkRelMdModifiedMonotonicity.scala
index 374cd8b..75a5c0f 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/metadata/FlinkRelMdModifiedMonotonicity.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/metadata/FlinkRelMdModifiedMonotonicity.scala
@@ -21,7 +21,6 @@ package org.apache.flink.table.plan.metadata
import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.functions.sql.SqlIncrSumAggFunction
import org.apache.flink.table.functions.utils.ScalarSqlFunction
-import org.apache.flink.table.plan.FlinkJoinRelType
import org.apache.flink.table.plan.`trait`.RelModifiedMonotonicity
import org.apache.flink.table.plan.metadata.FlinkMetadata.ModifiedMonotonicity
import org.apache.flink.table.plan.nodes.calcite.{Expand, Rank}
@@ -30,6 +29,8 @@ import org.apache.flink.table.plan.nodes.physical.batch.{BatchExecCorrelate, Bat
import org.apache.flink.table.plan.nodes.physical.stream._
import org.apache.flink.table.plan.schema.DataStreamTable
import org.apache.flink.table.plan.stats.{WithLower, WithUpper}
+import org.apache.flink.table.plan.util.JoinTypeUtil
+import org.apache.flink.table.runtime.join.FlinkJoinType
import org.apache.flink.table.{JByte, JDouble, JFloat, JList, JLong, JShort}
import org.apache.calcite.plan.hep.HepRelVertex
@@ -363,7 +364,7 @@ class FlinkRelMdModifiedMonotonicity private extends MetadataHandler[ModifiedMon
val joinInfo = rel.analyzeCondition
val leftKeys = joinInfo.leftKeys
val rightKeys = joinInfo.rightKeys
- val joinType = FlinkJoinRelType.toFlinkJoinRelType(rel.getJoinType)
+ val joinType = JoinTypeUtil.toFlinkJoinType(rel.getJoinType)
val fmq = FlinkRelMetadataQuery.reuseOrCreate(mq)
// if group set contains update return null
@@ -378,11 +379,11 @@ class FlinkRelMdModifiedMonotonicity private extends MetadataHandler[ModifiedMon
val isKeyAllAppend = isAllConstantOnKeys(left, leftKeys.toIntArray) &&
isAllConstantOnKeys(right, rightKeys.toIntArray)
- if (!containDelete && !joinType.equals(FlinkJoinRelType.ANTI) && isKeyAllAppend &&
+ if (!containDelete && !joinType.equals(FlinkJoinType.ANTI) && isKeyAllAppend &&
(containUpdate && joinInfo.isEqui || !containUpdate)) {
// output rowtype of semi equals to the rowtype of left child
- if (joinType.equals(FlinkJoinRelType.SEMI)) {
+ if (joinType.equals(FlinkJoinType.SEMI)) {
fmq.getRelModifiedMonotonicity(left)
} else {
val leftFieldMonotonicities = fmq.getRelModifiedMonotonicity(left).fieldMonotonicities
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/common/CommonPhysicalJoin.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/common/CommonPhysicalJoin.scala
index fd41e79..12f08c9 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/common/CommonPhysicalJoin.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/common/CommonPhysicalJoin.scala
@@ -18,9 +18,9 @@
package org.apache.flink.table.plan.nodes.common
-import org.apache.flink.table.plan.FlinkJoinRelType
import org.apache.flink.table.plan.nodes.physical.FlinkPhysicalRel
-import org.apache.flink.table.plan.util.{JoinUtil, RelExplainUtil}
+import org.apache.flink.table.plan.util.{JoinTypeUtil, JoinUtil, RelExplainUtil}
+import org.apache.flink.table.runtime.join.FlinkJoinType
import org.apache.calcite.rel.RelWriter
import org.apache.calcite.rel.`type`.{RelDataType, RelDataTypeField}
@@ -46,12 +46,7 @@ trait CommonPhysicalJoin extends Join with FlinkPhysicalRel {
lazy val keyPairs: List[IntPair] = joinInfo.pairs.toList
- // TODO supports FlinkJoinRelType.ANTI
- lazy val flinkJoinType: FlinkJoinRelType = this match {
- case sj: SemiJoin => FlinkJoinRelType.SEMI
- case j: Join => FlinkJoinRelType.toFlinkJoinRelType(getJoinType)
- case _ => throw new IllegalArgumentException(s"Illegal join node: ${this.getRelTypeName}")
- }
+ lazy val flinkJoinType: FlinkJoinType = JoinTypeUtil.getFlinkJoinType(this)
lazy val inputRowType: RelDataType = this match {
case sj: SemiJoin =>
@@ -70,7 +65,7 @@ trait CommonPhysicalJoin extends Join with FlinkPhysicalRel {
override def explainTerms(pw: RelWriter): RelWriter = {
pw.input("left", getLeft).input("right", getRight)
- .item("joinType", RelExplainUtil.joinTypeToString(flinkJoinType))
+ .item("joinType", flinkJoinType.toString)
.item("where",
RelExplainUtil.expressionToString(getCondition, inputRowType, getExpressionString))
.item("select", getRowType.getFieldNames.mkString(", "))
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSortMergeJoin.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSortMergeJoin.scala
index 4db56da..19fefe2 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSortMergeJoin.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSortMergeJoin.scala
@@ -26,12 +26,11 @@ import org.apache.flink.table.codegen.CodeGeneratorContext
import org.apache.flink.table.codegen.ProjectionCodeGenerator.generateProjection
import org.apache.flink.table.codegen.sort.SortCodeGenerator
import org.apache.flink.table.dataformat.BaseRow
-import org.apache.flink.table.plan.FlinkJoinRelType
import org.apache.flink.table.plan.cost.{FlinkCost, FlinkCostFactory}
import org.apache.flink.table.plan.nodes.ExpressionFormat
import org.apache.flink.table.plan.nodes.exec.{BatchExecNode, ExecNode}
import org.apache.flink.table.plan.util.{FlinkRelMdUtil, JoinUtil, SortUtil}
-import org.apache.flink.table.runtime.join.SortMergeJoinOperator
+import org.apache.flink.table.runtime.join.{FlinkJoinType, SortMergeJoinOperator}
import org.apache.calcite.plan._
import org.apache.calcite.rel.core._
@@ -56,11 +55,11 @@ trait BatchExecSortMergeJoinBase extends BatchExecJoinBase with BatchExecNode[Ba
protected lazy val (leftAllKey, rightAllKey) =
JoinUtil.checkAndGetJoinKeys(keyPairs, getLeft, getRight)
- protected def isMergeJoinSupportedType(joinRelType: FlinkJoinRelType): Boolean = {
- joinRelType == FlinkJoinRelType.INNER ||
- joinRelType == FlinkJoinRelType.LEFT ||
- joinRelType == FlinkJoinRelType.RIGHT ||
- joinRelType == FlinkJoinRelType.FULL
+ protected def isMergeJoinSupportedType(joinRelType: FlinkJoinType): Boolean = {
+ joinRelType == FlinkJoinType.INNER ||
+ joinRelType == FlinkJoinType.LEFT ||
+ joinRelType == FlinkJoinType.RIGHT ||
+ joinRelType == FlinkJoinType.FULL
}
protected lazy val smjType: SortMergeJoinType.Value = {
@@ -68,10 +67,10 @@ trait BatchExecSortMergeJoinBase extends BatchExecJoinBase with BatchExecNode[Ba
case (true, true) if isMergeJoinSupportedType(flinkJoinType) =>
SortMergeJoinType.MergeJoin
case (false, true) //TODO support more
- if flinkJoinType == FlinkJoinRelType.INNER || flinkJoinType == FlinkJoinRelType.RIGHT =>
+ if flinkJoinType == FlinkJoinType.INNER || flinkJoinType == FlinkJoinType.RIGHT =>
SortMergeJoinType.SortLeftJoin
case (true, false) //TODO support more
- if flinkJoinType == FlinkJoinRelType.INNER || flinkJoinType == FlinkJoinRelType.LEFT =>
+ if flinkJoinType == FlinkJoinType.INNER || flinkJoinType == FlinkJoinType.LEFT =>
SortMergeJoinType.SortRightJoin
case _ => SortMergeJoinType.SortMergeJoin
}
@@ -178,14 +177,7 @@ trait BatchExecSortMergeJoinBase extends BatchExecJoinBase with BatchExecNode[Ba
sortMemory,
sortMemory,
externalBufferMemory,
- flinkJoinType match {
- case FlinkJoinRelType.INNER => SortMergeJoinOperator.SortMergeJoinType.INNER
- case FlinkJoinRelType.LEFT => SortMergeJoinOperator.SortMergeJoinType.LEFT
- case FlinkJoinRelType.RIGHT => SortMergeJoinOperator.SortMergeJoinType.RIGHT
- case FlinkJoinRelType.FULL => SortMergeJoinOperator.SortMergeJoinType.FULL
- case FlinkJoinRelType.SEMI => SortMergeJoinOperator.SortMergeJoinType.SEMI
- case FlinkJoinRelType.ANTI => SortMergeJoinOperator.SortMergeJoinType.ANTI
- },
+ flinkJoinType,
estimateOutputSize(getLeft) < estimateOutputSize(getRight),
condFunc,
generateProjection(
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecJoinBase.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecJoinBase.scala
index 164f141..e9fb5a7 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecJoinBase.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecJoinBase.scala
@@ -21,9 +21,9 @@ package org.apache.flink.table.plan.nodes.physical.stream
import org.apache.flink.streaming.api.transformations.StreamTransformation
import org.apache.flink.table.api.{StreamTableEnvironment, TableException}
import org.apache.flink.table.dataformat.BaseRow
-import org.apache.flink.table.plan.FlinkJoinRelType
import org.apache.flink.table.plan.nodes.common.CommonPhysicalJoin
import org.apache.flink.table.plan.nodes.exec.{ExecNode, StreamExecNode}
+import org.apache.flink.table.runtime.join.FlinkJoinType
import org.apache.calcite.plan._
import org.apache.calcite.plan.hep.HepRelVertex
@@ -48,7 +48,7 @@ trait StreamExecJoinBase
with StreamExecNode[BaseRow] {
override def producesUpdates: Boolean = {
- flinkJoinType != FlinkJoinRelType.INNER && flinkJoinType != FlinkJoinRelType.SEMI
+ flinkJoinType != FlinkJoinType.INNER && flinkJoinType != FlinkJoinType.SEMI
}
override def needsUpdatesAsRetraction(input: RelNode): Boolean = {
@@ -80,8 +80,8 @@ trait StreamExecJoinBase
override def producesRetractions: Boolean = {
flinkJoinType match {
- case FlinkJoinRelType.FULL | FlinkJoinRelType.RIGHT | FlinkJoinRelType.LEFT => true
- case FlinkJoinRelType.ANTI => true
+ case FlinkJoinType.FULL | FlinkJoinType.RIGHT | FlinkJoinType.LEFT => true
+ case FlinkJoinType.ANTI => true
case _ => false
}
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecWindowJoin.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecWindowJoin.scala
index 2138c2f..abc3c51 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecWindowJoin.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/stream/StreamExecWindowJoin.scala
@@ -18,16 +18,25 @@
package org.apache.flink.table.plan.nodes.physical.stream
-import org.apache.flink.streaming.api.transformations.StreamTransformation
+import org.apache.flink.api.common.functions.{FlatJoinFunction, FlatMapFunction, MapFunction}
+import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator
+import org.apache.flink.streaming.api.operators.{StreamFlatMap, StreamMap, TwoInputStreamOperator}
+import org.apache.flink.streaming.api.transformations.{OneInputTransformation, StreamTransformation, TwoInputTransformation, UnionTransformation}
import org.apache.flink.table.api.{StreamTableEnvironment, TableException}
+import org.apache.flink.table.calcite.FlinkTypeFactory
import org.apache.flink.table.dataformat.BaseRow
-import org.apache.flink.table.plan.FlinkJoinRelType
+import org.apache.flink.table.generated.GeneratedFunction
import org.apache.flink.table.plan.nodes.exec.{ExecNode, StreamExecNode}
-import org.apache.flink.table.plan.util.RelExplainUtil
+import org.apache.flink.table.plan.util.{JoinTypeUtil, KeySelectorUtil, RelExplainUtil, UpdatingPlanChecker, WindowJoinUtil}
+import org.apache.flink.table.runtime.join.{FlinkJoinType, KeyedCoProcessOperatorWithWatermarkDelay, OuterJoinPaddingUtil, ProcTimeBoundedStreamJoin, RowTimeBoundedStreamJoin}
+import org.apache.flink.table.typeutils.BaseRowTypeInfo
+import org.apache.flink.util.Collector
import org.apache.calcite.plan._
import org.apache.calcite.rel.`type`.RelDataType
-import org.apache.calcite.rel.core.JoinRelType
+import org.apache.calcite.rel.core.{JoinInfo, JoinRelType}
import org.apache.calcite.rel.{BiRel, RelNode, RelWriter}
import org.apache.calcite.rex.RexNode
@@ -56,6 +65,8 @@ class StreamExecWindowJoin(
with StreamPhysicalRel
with StreamExecNode[BaseRow] {
+ private lazy val flinkJoinType: FlinkJoinType = JoinTypeUtil.toFlinkJoinType(joinType)
+
override def producesUpdates: Boolean = false
override def needsUpdatesAsRetraction(input: RelNode): Boolean = false
@@ -90,8 +101,7 @@ class StreamExecWindowJoin(
s"leftUpperBound=$leftUpperBound, leftTimeIndex=$leftTimeIndex, " +
s"rightTimeIndex=$rightTimeIndex"
super.explainTerms(pw)
- .item("joinType",
- RelExplainUtil.joinTypeToString(FlinkJoinRelType.toFlinkJoinRelType(joinType)))
+ .item("joinType", flinkJoinType.toString)
.item("windowBounds", windowBounds)
.item("where",
RelExplainUtil.expressionToString(joinCondition, outputRowType, getExpressionString))
@@ -103,16 +113,246 @@ class StreamExecWindowJoin(
override def getInputNodes: util.List[ExecNode[StreamTableEnvironment, _]] = {
getInputs.map(_.asInstanceOf[ExecNode[StreamTableEnvironment, _]])
}
-
+
override def replaceInputNode(
- ordinalInParent: Int,
- newInputNode: ExecNode[StreamTableEnvironment, _]): Unit = {
+ ordinalInParent: Int, newInputNode: ExecNode[StreamTableEnvironment, _]): Unit = {
replaceInput(ordinalInParent, newInputNode.asInstanceOf[RelNode])
}
override protected def translateToPlanInternal(
tableEnv: StreamTableEnvironment): StreamTransformation[BaseRow] = {
- throw new TableException("Implements this")
+ val isLeftAppendOnly = UpdatingPlanChecker.isAppendOnly(left)
+ val isRightAppendOnly = UpdatingPlanChecker.isAppendOnly(right)
+ if (!isLeftAppendOnly || !isRightAppendOnly) {
+ throw new TableException(
+ "Window Join: Windowed stream join does not support updates.\n" +
+ "please re-check window join statement according to description above.")
+ }
+
+ val leftPlan = getInputNodes.get(0).translateToPlan(tableEnv)
+ .asInstanceOf[StreamTransformation[BaseRow]]
+ val rightPlan = getInputNodes.get(1).translateToPlan(tableEnv)
+ .asInstanceOf[StreamTransformation[BaseRow]]
+
+ flinkJoinType match {
+ case FlinkJoinType.INNER |
+ FlinkJoinType.LEFT |
+ FlinkJoinType.RIGHT |
+ FlinkJoinType.FULL =>
+ val leftRowType = FlinkTypeFactory.toInternalRowType(getLeft.getRowType)
+ val rightRowType = FlinkTypeFactory.toInternalRowType(getRight.getRowType)
+ val returnType = FlinkTypeFactory.toInternalRowType(getRowType).toTypeInfo
+
+ val relativeWindowSize = leftUpperBound - leftLowerBound
+ if (relativeWindowSize < 0) {
+ LOG.warn(s"The relative window size $relativeWindowSize is negative," +
+ " please check the join conditions.")
+ createNegativeWindowSizeJoin(
+ leftPlan,
+ rightPlan,
+ leftRowType.getArity,
+ rightRowType.getArity,
+ returnType)
+ } else {
+ // get the equi-keys and other conditions
+ val joinInfo = JoinInfo.of(left, right, joinCondition)
+ val leftKeys = joinInfo.leftKeys.toIntArray
+ val rightKeys = joinInfo.rightKeys.toIntArray
+
+ // generate join function
+ val joinFunction = WindowJoinUtil.generateJoinFunction(
+ tableEnv.getConfig,
+ joinType,
+ leftRowType,
+ rightRowType,
+ getRowType,
+ remainCondition,
+ "WindowJoinFunction")
+
+ if (isRowTime) {
+ createRowTimeJoin(
+ leftPlan,
+ rightPlan,
+ returnType,
+ joinFunction,
+ leftKeys,
+ rightKeys)
+ } else {
+ createProcTimeJoin(
+ leftPlan,
+ rightPlan,
+ returnType,
+ joinFunction,
+ leftKeys,
+ rightKeys)
+ }
+ }
+ case FlinkJoinType.ANTI =>
+ throw new TableException(
+ "Window Join: {Anti Join} between stream and stream is not supported yet.\n" +
+ "please re-check window join statement according to description above.")
+ case FlinkJoinType.SEMI =>
+ throw new TableException(
+ "Window Join: {Semi Join} between stream and stream is not supported yet.\n" +
+ "please re-check window join statement according to description above.")
+ }
+ }
+
+ private def createNegativeWindowSizeJoin(
+ leftPlan: StreamTransformation[BaseRow],
+ rightPlan: StreamTransformation[BaseRow],
+ leftArity: Int,
+ rightArity: Int,
+ returnTypeInfo: BaseRowTypeInfo): StreamTransformation[BaseRow] = {
+ // We filter all records instead of adding an empty source to preserve the watermarks.
+ val allFilter = new FlatMapFunction[BaseRow, BaseRow] with ResultTypeQueryable[BaseRow] {
+ override def flatMap(value: BaseRow, out: Collector[BaseRow]): Unit = {}
+
+ override def getProducedType: TypeInformation[BaseRow] = returnTypeInfo
+ }
+
+ val leftPadder = new MapFunction[BaseRow, BaseRow] with ResultTypeQueryable[BaseRow] {
+ val paddingUtil = new OuterJoinPaddingUtil(leftArity, rightArity)
+
+ override def map(value: BaseRow): BaseRow = paddingUtil.padLeft(value)
+
+ override def getProducedType: TypeInformation[BaseRow] = returnTypeInfo
+ }
+
+ val rightPadder = new MapFunction[BaseRow, BaseRow] with ResultTypeQueryable[BaseRow] {
+ val paddingUtil = new OuterJoinPaddingUtil(leftArity, rightArity)
+
+ override def map(value: BaseRow): BaseRow = paddingUtil.padRight(value)
+
+ override def getProducedType: TypeInformation[BaseRow] = returnTypeInfo
+ }
+
+ val leftParallelism = leftPlan.getParallelism
+ val rightParallelism = rightPlan.getParallelism
+
+ val filterAllLeftStream = new OneInputTransformation[BaseRow, BaseRow](
+ leftPlan,
+ "filter all left input transformation",
+ new StreamFlatMap[BaseRow, BaseRow](allFilter),
+ returnTypeInfo,
+ leftParallelism)
+
+ val filterAllRightStream = new OneInputTransformation[BaseRow, BaseRow](
+ rightPlan,
+ "filter all right input transformation",
+ new StreamFlatMap[BaseRow, BaseRow](allFilter),
+ returnTypeInfo,
+ rightParallelism)
+
+ val padLeftStream = new OneInputTransformation[BaseRow, BaseRow](
+ leftPlan,
+ "pad left input transformation",
+ new StreamMap[BaseRow, BaseRow](leftPadder),
+ returnTypeInfo,
+ leftParallelism
+ )
+
+ val padRightStream = new OneInputTransformation[BaseRow, BaseRow](
+ rightPlan,
+ "pad right input transformation",
+ new StreamMap[BaseRow, BaseRow](rightPadder),
+ returnTypeInfo,
+ rightParallelism
+ )
+ flinkJoinType match {
+ case FlinkJoinType.INNER =>
+ new UnionTransformation(List(filterAllLeftStream, filterAllRightStream))
+ case FlinkJoinType.LEFT =>
+ new UnionTransformation(List(padLeftStream, filterAllRightStream))
+ case FlinkJoinType.RIGHT =>
+ new UnionTransformation(List(filterAllLeftStream, padRightStream))
+ case FlinkJoinType.FULL =>
+ new UnionTransformation(List(padLeftStream, padRightStream))
+ }
+ }
+
+ private def createProcTimeJoin(
+ leftPlan: StreamTransformation[BaseRow],
+ rightPlan: StreamTransformation[BaseRow],
+ returnTypeInfo: BaseRowTypeInfo,
+ joinFunction: GeneratedFunction[FlatJoinFunction[BaseRow, BaseRow, BaseRow]],
+ leftKeys: Array[Int],
+ rightKeys: Array[Int]): StreamTransformation[BaseRow] = {
+ val leftTypeInfo = leftPlan.getOutputType.asInstanceOf[BaseRowTypeInfo]
+ val rightTypeInfo = rightPlan.getOutputType.asInstanceOf[BaseRowTypeInfo]
+ val procJoinFunc = new ProcTimeBoundedStreamJoin(
+ flinkJoinType,
+ leftLowerBound,
+ leftUpperBound,
+ leftTypeInfo,
+ rightTypeInfo,
+ joinFunction)
+
+ val ret = new TwoInputTransformation[BaseRow, BaseRow, BaseRow](
+ leftPlan,
+ rightPlan,
+ "Co-Process",
+ new KeyedCoProcessOperator(procJoinFunc).
+ asInstanceOf[TwoInputStreamOperator[BaseRow,BaseRow,BaseRow]],
+ returnTypeInfo,
+ leftPlan.getParallelism
+ )
+
+ if (leftKeys.isEmpty) {
+ ret.setParallelism(1)
+ ret.setMaxParallelism(1)
+ }
+
+ // set KeyType and Selector for state
+ val leftSelect = KeySelectorUtil.getBaseRowSelector(leftKeys, leftTypeInfo)
+ val rightSelect = KeySelectorUtil.getBaseRowSelector(rightKeys, rightTypeInfo)
+ ret.setStateKeySelectors(leftSelect, rightSelect)
+ ret.setStateKeyType(leftSelect.getProducedType)
+ ret
+ }
+
+ private def createRowTimeJoin(
+ leftPlan: StreamTransformation[BaseRow],
+ rightPlan: StreamTransformation[BaseRow],
+ returnTypeInfo: BaseRowTypeInfo,
+ joinFunction: GeneratedFunction[FlatJoinFunction[BaseRow, BaseRow, BaseRow]],
+ leftKeys: Array[Int],
+ rightKeys: Array[Int]
+ ): StreamTransformation[BaseRow] = {
+ val leftTypeInfo = leftPlan.getOutputType.asInstanceOf[BaseRowTypeInfo]
+ val rightTypeInfo = rightPlan.getOutputType.asInstanceOf[BaseRowTypeInfo]
+ val rowJoinFunc = new RowTimeBoundedStreamJoin(
+ flinkJoinType,
+ leftLowerBound,
+ leftUpperBound,
+ 0L,
+ leftTypeInfo,
+ rightTypeInfo,
+ joinFunction,
+ leftTimeIndex,
+ rightTimeIndex)
+
+ val ret = new TwoInputTransformation[BaseRow, BaseRow, BaseRow](
+ leftPlan,
+ rightPlan,
+ "Co-Process",
+ new KeyedCoProcessOperatorWithWatermarkDelay(rowJoinFunc, rowJoinFunc.getMaxOutputDelay)
+ .asInstanceOf[TwoInputStreamOperator[BaseRow,BaseRow,BaseRow]],
+ returnTypeInfo,
+ leftPlan.getParallelism
+ )
+
+ if (leftKeys.isEmpty) {
+ ret.setParallelism(1)
+ ret.setMaxParallelism(1)
+ }
+
+ // set KeyType and Selector for state
+ val leftSelector = KeySelectorUtil.getBaseRowSelector(leftKeys, leftTypeInfo)
+ val rightSelector = KeySelectorUtil.getBaseRowSelector(rightKeys, rightTypeInfo)
+ ret.setStateKeySelectors(leftSelector, rightSelector)
+ ret.setStateKeyType(leftSelector.getProducedType)
+ ret
}
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecHashJoinRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecHashJoinRule.scala
index 5b9ad57..a4a6bc2 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecHashJoinRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecHashJoinRule.scala
@@ -21,11 +21,11 @@ package org.apache.flink.table.plan.rules.physical.batch
import org.apache.flink.table.JDouble
import org.apache.flink.table.api.{OperatorType, PlannerConfigOptions, TableConfig}
import org.apache.flink.table.calcite.FlinkContext
-import org.apache.flink.table.plan.FlinkJoinRelType
import org.apache.flink.table.plan.`trait`.FlinkRelDistribution
import org.apache.flink.table.plan.nodes.FlinkConventions
import org.apache.flink.table.plan.nodes.logical.FlinkLogicalJoin
import org.apache.flink.table.plan.nodes.physical.batch.BatchExecHashJoin
+import org.apache.flink.table.runtime.join.FlinkJoinType
import org.apache.calcite.plan.RelOptRule.{any, operand}
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall, RelTraitSet}
@@ -61,7 +61,7 @@ class BatchExecHashJoinRule(joinClass: Class[_ <: Join])
val isShuffleHashJoinEnabled = tableConfig.isOperatorEnabled(OperatorType.ShuffleHashJoin)
val isBroadcastHashJoinEnabled = tableConfig.isOperatorEnabled(OperatorType.BroadcastHashJoin)
- val joinType = getFlinkJoinRelType(join)
+ val joinType = getFlinkJoinType(join)
val leftSize = binaryRowRelNodeSize(join.getLeft)
val rightSize = binaryRowRelNodeSize(join.getRight)
val (isBroadcast, _) = canBroadcast(joinType, leftSize, rightSize, tableConfig)
@@ -74,7 +74,7 @@ class BatchExecHashJoinRule(joinClass: Class[_ <: Join])
val tableConfig = call.getPlanner.getContext.asInstanceOf[FlinkContext].getTableConfig
val join: Join = call.rel(0)
val joinInfo = join.analyzeCondition
- val joinType = getFlinkJoinRelType(join)
+ val joinType = getFlinkJoinType(join)
val left = join.getLeft
val right = join.getRight
@@ -155,7 +155,7 @@ class BatchExecHashJoinRule(joinClass: Class[_ <: Join])
* as broadcast side, false else.
*/
private def canBroadcast(
- joinType: FlinkJoinRelType,
+ joinType: FlinkJoinType,
leftSize: JDouble,
rightSize: JDouble,
tableConfig: TableConfig): (Boolean, Boolean) = {
@@ -166,13 +166,13 @@ class BatchExecHashJoinRule(joinClass: Class[_ <: Join])
val threshold = tableConfig.getConf.getLong(
PlannerConfigOptions.SQL_OPTIMIZER_HASH_JOIN_BROADCAST_THRESHOLD)
joinType match {
- case FlinkJoinRelType.LEFT => (rightSize <= threshold, false)
- case FlinkJoinRelType.RIGHT => (leftSize <= threshold, true)
- case FlinkJoinRelType.FULL => (false, false)
- case FlinkJoinRelType.INNER =>
+ case FlinkJoinType.LEFT => (rightSize <= threshold, false)
+ case FlinkJoinType.RIGHT => (leftSize <= threshold, true)
+ case FlinkJoinType.FULL => (false, false)
+ case FlinkJoinType.INNER =>
(leftSize <= threshold || rightSize <= threshold, leftSize < rightSize)
// left side cannot be used as build side in SEMI/ANTI join.
- case FlinkJoinRelType.SEMI | FlinkJoinRelType.ANTI => (rightSize <= threshold, false)
+ case FlinkJoinType.SEMI | FlinkJoinType.ANTI => (rightSize <= threshold, false)
}
}
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecJoinRuleBase.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecJoinRuleBase.scala
index 5a118af..87b0567 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecJoinRuleBase.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecJoinRuleBase.scala
@@ -19,15 +19,14 @@
package org.apache.flink.table.plan.rules.physical.batch
import org.apache.flink.table.JDouble
-import org.apache.flink.table.plan.FlinkJoinRelType
import org.apache.flink.table.plan.nodes.FlinkConventions
import org.apache.flink.table.plan.nodes.logical.FlinkLogicalJoin
import org.apache.flink.table.plan.nodes.physical.batch.BatchExecLocalHashAggregate
-import org.apache.flink.table.plan.util.FlinkRelMdUtil
+import org.apache.flink.table.plan.util.{FlinkRelMdUtil, JoinTypeUtil}
+import org.apache.flink.table.runtime.join.FlinkJoinType
import org.apache.calcite.plan.RelOptRule
import org.apache.calcite.rel.RelNode
-import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.Join
import org.apache.calcite.tools.RelBuilder
@@ -53,9 +52,8 @@ trait BatchExecJoinRuleBase {
Seq())
}
- def getFlinkJoinRelType(join: Join): FlinkJoinRelType = join match {
- case j: FlinkLogicalJoin =>
- FlinkJoinRelType.toFlinkJoinRelType(j.getJoinType)
+ def getFlinkJoinType(join: Join): FlinkJoinType = join match {
+ case j: FlinkLogicalJoin => JoinTypeUtil.getFlinkJoinType(j)
case _ => throw new IllegalArgumentException(s"Illegal join node: ${join.getRelTypeName}")
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecSingleRowJoinRule.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecSingleRowJoinRule.scala
index fc52268..d44254a 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecSingleRowJoinRule.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/rules/physical/batch/BatchExecSingleRowJoinRule.scala
@@ -18,10 +18,10 @@
package org.apache.flink.table.plan.rules.physical.batch
-import org.apache.flink.table.plan.FlinkJoinRelType
import org.apache.flink.table.plan.nodes.FlinkConventions
import org.apache.flink.table.plan.nodes.logical.FlinkLogicalJoin
import org.apache.flink.table.plan.nodes.physical.batch.BatchExecNestedLoopJoin
+import org.apache.flink.table.runtime.join.FlinkJoinType
import org.apache.calcite.plan.volcano.RelSubset
import org.apache.calcite.plan.{RelOptRule, RelOptRuleCall}
@@ -44,14 +44,14 @@ class BatchExecSingleRowJoinRule(joinClass: Class[_ <: Join])
override def matches(call: RelOptRuleCall): Boolean = {
val join: Join = call.rel(0)
- val joinType = getFlinkJoinRelType(join)
+ val joinType = getFlinkJoinType(join)
joinType match {
- case FlinkJoinRelType.INNER | FlinkJoinRelType.FULL =>
+ case FlinkJoinType.INNER | FlinkJoinType.FULL =>
isSingleRow(join.getLeft) || isSingleRow(join.getRight)
- case FlinkJoinRelType.LEFT if isSingleRow(join.getRight) => true
- case FlinkJoinRelType.RIGHT if isSingleRow(join.getLeft) => true
- case FlinkJoinRelType.SEMI if isSingleRow(join.getRight) => true
- case FlinkJoinRelType.ANTI if isSingleRow(join.getRight) => true
+ case FlinkJoinType.LEFT if isSingleRow(join.getRight) => true
+ case FlinkJoinType.RIGHT if isSingleRow(join.getLeft) => true
+ case FlinkJoinType.SEMI if isSingleRow(join.getRight) => true
+ case FlinkJoinType.ANTI if isSingleRow(join.getRight) => true
case _ => false
}
}
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/RelExplainUtil.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/RelExplainUtil.scala
index 87257d0..f8127ea 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/RelExplainUtil.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/RelExplainUtil.scala
@@ -22,7 +22,6 @@ import org.apache.flink.table.api.TableException
import org.apache.flink.table.functions.aggfunctions.DeclarativeAggregateFunction
import org.apache.flink.table.functions.utils.TableSqlFunction
import org.apache.flink.table.functions.{AggregateFunction, UserDefinedFunction}
-import org.apache.flink.table.plan.FlinkJoinRelType
import org.apache.flink.table.plan.nodes.ExpressionFormat
import org.apache.flink.table.plan.nodes.ExpressionFormat.ExpressionFormat
@@ -91,20 +90,6 @@ object RelExplainUtil {
}
/**
- * Converts [[FlinkJoinRelType]] to String.
- */
- def joinTypeToString(joinType: FlinkJoinRelType): String = {
- joinType match {
- case FlinkJoinRelType.INNER => "InnerJoin"
- case FlinkJoinRelType.LEFT => "LeftOuterJoin"
- case FlinkJoinRelType.RIGHT => "RightOuterJoin"
- case FlinkJoinRelType.FULL => "FullOuterJoin"
- case FlinkJoinRelType.SEMI => "LeftSemiJoin"
- case FlinkJoinRelType.ANTI => "LeftAntiJoin"
- }
- }
-
- /**
* Converts sort fetch to String.
*/
def fetchToString(fetch: RexNode): String = {
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/WindowJoinUtil.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/WindowJoinUtil.scala
index 545abff..1e2a41b 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/WindowJoinUtil.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/util/WindowJoinUtil.scala
@@ -18,13 +18,19 @@
package org.apache.flink.table.plan.util
+import org.apache.flink.api.common.functions.FlatJoinFunction
import org.apache.flink.table.api.{PlannerConfigOptions, TableConfig}
import org.apache.flink.table.calcite.FlinkTypeFactory
-import org.apache.flink.table.codegen.ExpressionReducer
+import org.apache.flink.table.codegen.{CodeGenUtils, CodeGeneratorContext, ExprCodeGenerator,
+ExpressionReducer, FunctionCodeGenerator}
+import org.apache.flink.table.dataformat.{BaseRow, JoinedRow}
+import org.apache.flink.table.generated.GeneratedFunction
import org.apache.flink.table.plan.schema.TimeIndicatorRelDataType
+import org.apache.flink.table.`type`.RowType
import org.apache.calcite.plan.RelOptUtil
import org.apache.calcite.rel.`type`.RelDataType
+import org.apache.calcite.rel.core.JoinRelType
import org.apache.calcite.rex._
import org.apache.calcite.sql.SqlKind
@@ -74,6 +80,78 @@ object WindowJoinUtil {
}
/**
+ * Generates a JoinFunction that applies additional join predicates and projects the result.
+ *
+ * @param config table env config
+ * @param joinType join type to determine whether input can be null
+ * @param leftType left stream type
+ * @param rightType right stream type
+ * @param returnType return type
+ * @param otherCondition non-equi condition
+ * @param funcName name of generated function
+ */
+ private[flink] def generateJoinFunction(
+ config: TableConfig,
+ joinType: JoinRelType,
+ leftType: RowType,
+ rightType: RowType,
+ returnType: RelDataType,
+ otherCondition: Option[RexNode],
+ funcName: String): GeneratedFunction[FlatJoinFunction[BaseRow, BaseRow, BaseRow]] = {
+
+ // whether input can be null
+ val nullCheck = joinType match {
+ case JoinRelType.INNER => false
+ case JoinRelType.LEFT => true
+ case JoinRelType.RIGHT => true
+ case JoinRelType.FULL => true
+ }
+
+ val ctx = CodeGeneratorContext(config)
+ val collectorTerm = CodeGenUtils.DEFAULT_COLLECTOR_TERM
+
+ val returnTypeInfo = FlinkTypeFactory.toInternalRowType(returnType)
+ val joinedRow = "joinedRow"
+ ctx.addReusableOutputRecord(returnTypeInfo, classOf[JoinedRow], joinedRow)
+
+ val exprGenerator = new ExprCodeGenerator(ctx, nullCheck)
+ .bindInput(leftType)
+ .bindSecondInput(rightType)
+
+ val leftRow = CodeGenUtils.DEFAULT_INPUT1_TERM
+ val rightRow = CodeGenUtils.DEFAULT_INPUT2_TERM
+ val buildJoinedRow =
+ s"""
+ |$joinedRow.replace($leftRow,$rightRow);""".stripMargin
+
+ val body = otherCondition match {
+ case None =>
+ s"""
+ |$buildJoinedRow
+ |$collectorTerm.collect($joinedRow)
+ |""".stripMargin
+ case Some(remainCondition) =>
+ val genCond = exprGenerator.generateExpression(remainCondition)
+ s"""
+ |${genCond.code}
+ |if (${genCond.resultTerm}) {
+ | $buildJoinedRow
+ | $collectorTerm.collect($joinedRow);
+ |}""".stripMargin
+ }
+
+ FunctionCodeGenerator.generateFunction(
+ ctx,
+ funcName,
+ classOf[FlatJoinFunction[BaseRow, BaseRow, BaseRow]],
+ body,
+ returnTypeInfo,
+ leftType,
+ input2Type = Some(rightType),
+ collectorTerm = collectorTerm)
+ }
+
+ /**
* Extracts the window bounds from a join predicate.
*
* @param predicate join predicate
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/DeduplicateITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/DeduplicateITCase.scala
index 1793940..8abe855 100644
--- a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/DeduplicateITCase.scala
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/DeduplicateITCase.scala
@@ -23,7 +23,6 @@ import org.apache.flink.table.api.scala._
import org.apache.flink.table.runtime.utils.StreamingWithMiniBatchTestBase.MiniBatchMode
import org.apache.flink.table.runtime.utils._
import org.apache.flink.table.runtime.utils.StreamingWithStateTestBase.StateBackendMode
-import org.apache.flink.table.runtime.utils.TimeTestUtil.TimestampAndWatermarkWithOffset
import org.apache.flink.types.Row
import org.junit.Assert._
diff --git a/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/WindowJoinITCase.scala b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/WindowJoinITCase.scala
new file mode 100644
index 0000000..509df85
--- /dev/null
+++ b/flink-table/flink-table-planner-blink/src/test/scala/org/apache/flink/table/runtime/stream/sql/WindowJoinITCase.scala
@@ -0,0 +1,950 @@
+/*
+ * 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.runtime.stream.sql
+
+import org.apache.flink.api.scala._
+import org.apache.flink.streaming.api.TimeCharacteristic
+import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
+import org.apache.flink.streaming.api.watermark.Watermark
+import org.apache.flink.table.api.scala._
+import org.apache.flink.table.runtime.utils._
+import org.apache.flink.table.runtime.utils.StreamingWithStateTestBase.StateBackendMode
+import org.apache.flink.types.Row
+
+import org.junit.Assert._
+import org.junit.{Ignore, Test}
+import org.junit.runner.RunWith
+import org.junit.runners.Parameterized
+
+import scala.collection.mutable
+
+@RunWith(classOf[Parameterized])
+class WindowJoinITCase(mode: StateBackendMode) extends StreamingWithStateTestBase(mode) {
+
+ // Tests for inner join.
+ /** test proctime inner join **/
+ @Test
+ def testProcessTimeInnerJoin(): Unit = {
+ env.setParallelism(1)
+
+ val sqlQuery =
+ """
+ |SELECT t2.a, t2.c, t1.c
+ |FROM T1 as t1 join T2 as t2 ON
+ | t1.a = t2.a AND
+ | t1.proctime BETWEEN t2.proctime - INTERVAL '5' SECOND AND
+ | t2.proctime + INTERVAL '5' SECOND
+ |""".stripMargin
+
+ val data1 = new mutable.MutableList[(Int, Long, String)]
+ data1.+=((1, 1L, "Hi1"))
+ data1.+=((1, 2L, "Hi2"))
+ data1.+=((1, 5L, "Hi3"))
+ data1.+=((2, 7L, "Hi5"))
+ data1.+=((1, 9L, "Hi6"))
+ data1.+=((1, 8L, "Hi8"))
+
+ val data2 = new mutable.MutableList[(Int, Long, String)]
+ data2.+=((1, 1L, "HiHi"))
+ data2.+=((2, 2L, "HeHe"))
+
+ val tmp1 = env.fromCollection(data1).toTable(tEnv, 'a, 'b, 'c, 'proctime)
+ tEnv.registerTable("TmpT1", tmp1)
+ val subquery1 = "SELECT IF(a = 1, CAST(NULL AS INT), a) as a, b, c, proctime FROM TmpT1"
+ val t1 = tEnv.sqlQuery(subquery1)
+ tEnv.registerTable("T1", t1)
+
+ val tmp2 = env.fromCollection(data2).toTable(tEnv, 'a, 'b, 'c, 'proctime)
+ tEnv.registerTable("TmpT2", tmp2)
+ val subquery2 = "SELECT IF(a = 1, CAST(NULL AS INT), a) as a, b, c, proctime FROM TmpT2"
+ val t2 = tEnv.sqlQuery(subquery2)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+ }
+
+ /** test proctime inner join with other condition **/
+ @Test
+ def testProcessTimeInnerJoinWithOtherConditions(): Unit = {
+ env.setParallelism(2)
+
+ val sqlQuery =
+ """
+ |SELECT t2.a, t2.c, t1.c
+ |FROM T1 as t1 JOIN T2 as t2 ON
+ | t1.a = t2.a AND
+ | t1.proctime BETWEEN t2.proctime - interval '5' SECOND AND
+ | t2.proctime + interval '5' second AND
+ | t1.b = t2.b
+ |""".stripMargin
+
+ val data1 = new mutable.MutableList[(String, Long, String)]
+ data1.+=(("1", 1L, "Hi1"))
+ data1.+=(("1", 2L, "Hi2"))
+ data1.+=(("1", 5L, "Hi3"))
+ data1.+=(("2", 7L, "Hi5"))
+ data1.+=(("1", 9L, "Hi6"))
+ data1.+=(("1", 8L, "Hi8"))
+
+ val data2 = new mutable.MutableList[(String, Long, String)]
+ data2.+=(("1", 5L, "HiHi"))
+ data2.+=(("2", 2L, "HeHe"))
+
+ // For null key test
+ data1.+=((null.asInstanceOf[String], 20L, "leftNull"))
+ data2.+=((null.asInstanceOf[String], 20L, "rightNull"))
+
+ val t1 = env.fromCollection(data1).toTable(tEnv, 'a, 'b, 'c, 'proctime)
+ val t2 = env.fromCollection(data2).toTable(tEnv, 'a, 'b, 'c, 'proctime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+
+ // Assert there is no result with null keys.
+ assertFalse(sink.getAppendResults.contains("null"))
+ }
+
+ /** test rowtime inner join **/
+ @Test
+ def testRowTimeInnerJoin(): Unit = {
+ val sqlQuery =
+ """
+ |SELECT t2.key, t2.id, t1.id
+ |FROM T1 as t1 join T2 as t2 ON
+ | t1.key = t2.key AND
+ | t1.rowtime BETWEEN t2.rowtime - INTERVAL '5' SECOND AND
+ | t2.rowtime + INTERVAL '6' SECOND
+ |""".stripMargin
+
+ val data1 = new mutable.MutableList[(String, String, Long)]
+ // for boundary test
+ data1.+=(("A", "LEFT0.999", 999L))
+ data1.+=(("A", "LEFT1", 1000L))
+ data1.+=(("A", "LEFT2", 2000L))
+ data1.+=(("A", "LEFT3", 3000L))
+ data1.+=(("B", "LEFT4", 4000L))
+ data1.+=(("A", "LEFT5", 5000L))
+ data1.+=(("A", "LEFT6", 6000L))
+ // test null key
+ data1.+=((null.asInstanceOf[String], "LEFT8", 8000L))
+
+ val data2 = new mutable.MutableList[(String, String, Long)]
+ data2.+=(("A", "RIGHT6", 6000L))
+ data2.+=(("B", "RIGHT7", 7000L))
+ // test null key
+ data2.+=((null.asInstanceOf[String], "RIGHT10", 10000L))
+
+ val t1 = env.fromCollection(data1)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+ val t2 = env.fromCollection(data2)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+ val expected = mutable.MutableList("A,RIGHT6,LEFT1", "A,RIGHT6,LEFT2", "A,RIGHT6,LEFT3",
+ "A,RIGHT6,LEFT5",
+ "A,RIGHT6,LEFT6",
+ "B,RIGHT7,LEFT4")
+ assertEquals(expected, sink.getAppendResults.sorted)
+ }
+
+ @Test
+ def testUnboundedAggAfterRowtimeInnerJoin(): Unit = {
+ val innerSql=
+ """
+ |SELECT t2.key as key, t2.id as id1, t1.id as id2
+ |FROM T1 as t1 join T2 as t2 ON
+ | t1.key = t2.key AND
+ | t1.rowtime BETWEEN t2.rowtime - INTERVAL '5' SECOND AND
+ | t2.rowtime + INTERVAL '6' SECOND
+ |""".stripMargin
+
+ val sqlQuery = "SELECT key, COUNT(DISTINCT id1), COUNT(DISTINCT id2) FROM (" +
+ innerSql + ") GROUP BY key"
+
+ val data1 = new mutable.MutableList[(String, String, Long)]
+ // for boundary test
+ data1.+=(("A", "LEFT0.999", 999L))
+ data1.+=(("A", "LEFT1", 1000L))
+ data1.+=(("A", "LEFT2", 2000L))
+ data1.+=(("A", "LEFT3", 3000L))
+ data1.+=(("B", "LEFT4", 4000L))
+ data1.+=(("A", "LEFT5", 5000L))
+ data1.+=(("A", "LEFT6", 6000L))
+ // test null key
+ data1.+=((null.asInstanceOf[String], "LEFT8", 8000L))
+
+ val data2 = new mutable.MutableList[(String, String, Long)]
+ data2.+=(("A", "RIGHT6", 6000L))
+ data2.+=(("B", "RIGHT7", 7000L))
+ // test null key
+ data2.+=((null.asInstanceOf[String], "RIGHT10", 10000L))
+
+ val t1 = env.fromCollection(data1)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+ val t2 = env.fromCollection(data2)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+ val sink = new TestingRetractSink
+ val result = tEnv.sqlQuery(sqlQuery).toRetractStream[Row]
+ result.addSink(sink)
+ env.execute()
+ val expected = mutable.MutableList("A,1,5", "B,1,1")
+ assertEquals(expected, sink.getRetractResults.sorted)
+ }
+
+ /** test row time inner join with equi-times **/
+ @Test
+ def testRowTimeInnerJoinWithEquiTimeAttrs(): Unit = {
+ env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+
+ val sqlQuery =
+ """
+ |SELECT t2.key, t2.id, t1.id
+ |FROM T1 AS t1 JOIN T2 AS t2 ON
+ |t1.key = t2.key AND
+ |t2.rowtime = t1.rowtime
+ """.stripMargin
+
+ val data1 = new mutable.MutableList[(Int, Long, String, Long)]
+
+ data1.+=((4, 4000L, "A", 4000L))
+ data1.+=((5, 5000L, "A", 5000L))
+ data1.+=((6, 6000L, "A", 6000L))
+ data1.+=((6, 6000L, "B", 6000L))
+
+ val data2 = new mutable.MutableList[(String, String, Long)]
+ data2.+=(("A", "R-5", 5000L))
+ data2.+=(("B", "R-6", 6000L))
+
+ val t1 = env.fromCollection(data1)
+ .assignTimestampsAndWatermarks(new Row4WatermarkExtractor)
+ .toTable(tEnv, 'id, 'tm, 'key, 'rowtime)
+ val t2 = env.fromCollection(data2)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+
+ val expected = mutable.MutableList[String](
+ "A,R-5,5",
+ "B,R-6,6"
+ )
+
+ assertEquals(expected.toList.sorted, sink.getAppendResults.sorted)
+ }
+
+ /** test rowtime inner join with other conditions **/
+ @Test
+ def testRowTimeInnerJoinWithOtherConditions(): Unit = {
+ val sqlQuery =
+ """
+ |SELECT t2.a, t1.c, t2.c
+ |FROM T1 as t1 JOIN T2 as t2 ON
+ | t1.a = t2.a AND
+ | t1.rowtime > t2.rowtime - INTERVAL '5' SECOND AND
+ | t1.rowtime < t2.rowtime - INTERVAL '1' SECOND AND
+ | t1.b < t2.b AND
+ | t1.b > 2
+ |""".stripMargin
+
+ val data1 = new mutable.MutableList[(Int, Long, String, Long)]
+ data1.+=((1, 4L, "LEFT1", 1000L))
+ // for boundary test
+ data1.+=((1, 8L, "LEFT1.1", 1001L))
+ // predicate (t1.b > 2) push down
+ data1.+=((1, 2L, "LEFT2", 2000L))
+ data1.+=((1, 7L, "LEFT3", 3000L))
+ data1.+=((2, 5L, "LEFT4", 4000L))
+ // for boundary test
+ data1.+=((1, 4L, "LEFT4.9", 4999L))
+ data1.+=((1, 4L, "LEFT5", 5000L))
+ data1.+=((1, 10L, "LEFT6", 6000L))
+
+ val data2 = new mutable.MutableList[(Int, Long, String, Long)]
+ // just for watermark
+ data2.+=((1, 1L, "RIGHT1", 1000L))
+ data2.+=((1, 9L, "RIGHT6", 6000L))
+ data2.+=((2, 14L, "RIGHT7", 7000L))
+ data2.+=((1, 4L, "RIGHT8", 8000L))
+
+ val t1 = env.fromCollection(data1)
+ .assignTimestampsAndWatermarks(new Row4WatermarkExtractor)
+ .toTable(tEnv, 'a, 'b, 'c, 'rowtime)
+ val t2 = env.fromCollection(data2)
+ .assignTimestampsAndWatermarks(new Row4WatermarkExtractor)
+ .toTable(tEnv, 'a, 'b, 'c, 'rowtime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+
+ // There may be two expected results according to the process order.
+ val expected = mutable.MutableList[String]("1,LEFT3,RIGHT6",
+ "1,LEFT1.1,RIGHT6",
+ "2,LEFT4,RIGHT7",
+ "1,LEFT4.9,RIGHT6")
+ assertEquals(expected.toList.sorted, sink.getAppendResults.sorted)
+ }
+
+ /** test rowtime inner join with another time condition **/
+ @Test
+ def testRowTimeInnerJoinWithOtherTimeCondition(): Unit = {
+ val sqlQuery =
+ """
+ |SELECT t2.a, t1.c, t2.c
+ |FROM T1 as t1 JOIN T2 as t2 ON
+ | t1.a = t2.a AND
+ | t1.rowtime > t2.rowtime - INTERVAL '4' SECOND AND
+ | t1.rowtime < t2.rowtime AND
+ | QUARTER(t1.rowtime) = t2.a
+ |""".stripMargin
+
+ val data1 = new mutable.MutableList[(Int, Long, String, Long)]
+ data1.+=((1, 4L, "LEFT1", 1000L))
+ data1.+=((1, 2L, "LEFT2", 2000L))
+ data1.+=((1, 7L, "LEFT3", 3000L))
+ data1.+=((2, 5L, "LEFT4", 4000L))
+ data1.+=((1, 4L, "LEFT5", 5000L))
+ data1.+=((1, 10L, "LEFT6", 6000L))
+
+ val data2 = new mutable.MutableList[(Int, Long, String, Long)]
+ data2.+=((1, 1L, "RIGHT1", 1000L))
+ data2.+=((1, 9L, "RIGHT6", 6000L))
+ data2.+=((2, 8, "RIGHT7", 7000L))
+ data2.+=((1, 4L, "RIGHT8", 8000L))
+
+ val t1 = env.fromCollection(data1)
+ .assignTimestampsAndWatermarks(new Row4WatermarkExtractor)
+ .toTable(tEnv, 'a, 'b, 'c, 'rowtime)
+ val t2 = env.fromCollection(data2)
+ .assignTimestampsAndWatermarks(new Row4WatermarkExtractor)
+ .toTable(tEnv, 'a, 'b, 'c, 'rowtime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+
+ val expected = mutable.MutableList[String](
+ "1,LEFT3,RIGHT6",
+ "1,LEFT5,RIGHT6",
+ "1,LEFT5,RIGHT8",
+ "1,LEFT6,RIGHT8")
+
+ assertEquals(expected.toList.sorted, sink.getAppendResults.sorted)
+ }
+
+ /** test rowtime inner join with window aggregation **/
+ @Ignore("Enable after StreamExecWindowAggregate is merged")
+ @Test
+ def testRowTimeInnerJoinWithWindowAggregateOnFirstTime(): Unit = {
+ val sqlQuery =
+ """
+ |SELECT t1.key, TUMBLE_END(t1.rowtime, INTERVAL '4' SECOND), COUNT(t2.key)
+ |FROM T1 AS t1 join T2 AS t2 ON
+ | t1.key = t2.key AND
+ | t1.rowtime BETWEEN t2.rowtime - INTERVAL '5' SECOND AND
+ | t2.rowtime + INTERVAL '5' SECOND
+ |GROUP BY TUMBLE(t1.rowtime, INTERVAL '4' SECOND), t1.key
+ |""".stripMargin
+
+ val data1 = new mutable.MutableList[(String, String, Long)]
+ data1.+=(("A", "L-1", 1000L)) // no joining record
+ data1.+=(("A", "L-2", 2000L)) // 1 joining record
+ data1.+=(("A", "L-3", 3000L)) // 2 joining records
+ //data1.+=(("B", "L-8", 2000L)) // 1 joining records
+ data1.+=(("B", "L-4", 4000L)) // 1 joining record
+ data1.+=(("C", "L-5", 4000L)) // no joining record
+ data1.+=(("A", "L-6", 10000L)) // 2 joining records
+ data1.+=(("A", "L-7", 13000L)) // 1 joining record
+
+ val data2 = new mutable.MutableList[(String, String, Long)]
+ data2.+=(("A", "R-1", 7000L)) // 3 joining records
+ data2.+=(("B", "R-4", 7000L)) // 1 joining records
+ data2.+=(("A", "R-3", 8000L)) // 3 joining records
+ data2.+=(("D", "R-2", 8000L)) // no joining record
+
+ val t1 = env.fromCollection(data1)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+ val t2 = env.fromCollection(data2)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val t_r = tEnv.sqlQuery(sqlQuery)
+ val result = t_r.toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+ val expected = mutable.MutableList[String](
+ "A,1970-01-01 00:00:04.0,3",
+ "A,1970-01-01 00:00:12.0,2",
+ "A,1970-01-01 00:00:16.0,1",
+ //"B,1970-01-01 00:00:04.0,1",
+ "B,1970-01-01 00:00:08.0,1")
+ assertEquals(expected.toList.sorted, sink.getAppendResults.sorted)
+ }
+
+ /** test row time inner join with window aggregation **/
+ @Ignore("Enable after StreamExecWindowAggregate is merged")
+ @Test
+ def testRowTimeInnerJoinWithWindowAggregateOnSecondTime(): Unit = {
+ val sqlQuery =
+ """
+ |SELECT t2.key, TUMBLE_END(t2.rowtime, INTERVAL '4' SECOND), COUNT(t1.key)
+ |FROM T1 AS t1 join T2 AS t2 ON
+ | t1.key = t2.key AND
+ | t1.rowtime BETWEEN t2.rowtime - INTERVAL '5' SECOND AND
+ | t2.rowtime + INTERVAL '5' SECOND
+ | GROUP BY TUMBLE(t2.rowtime, INTERVAL '4' SECOND), t2.key
+ """.stripMargin
+
+ val data1 = new mutable.MutableList[(String, String, Long)]
+ data1.+=(("A", "L-1", 1000L)) // no joining record
+ data1.+=(("A", "L-2", 2000L)) // 1 joining record
+ data1.+=(("A", "L-3", 3000L)) // 2 joining records
+ data1.+=(("B", "L-4", 4000L)) // 1 joining record
+ data1.+=(("C", "L-5", 4000L)) // no joining record
+ data1.+=(("A", "L-6", 10000L)) // 2 joining records
+ data1.+=(("A", "L-7", 13000L)) // 1 joining record
+
+ val data2 = new mutable.MutableList[(String, String, Long)]
+ data2.+=(("A", "R-1", 7000L)) // 3 joining records
+ data2.+=(("B", "R-4", 7000L)) // 1 joining records
+ data2.+=(("A", "R-3", 8000L)) // 3 joining records
+ data2.+=(("D", "R-2", 8000L)) // no joining record
+
+ val t1 = env.fromCollection(data1)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+ val t2 = env.fromCollection(data2)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+ val expected = mutable.MutableList[String](
+ "A,1970-01-01 00:00:08.0,3",
+ "A,1970-01-01 00:00:12.0,3",
+ "B,1970-01-01 00:00:08.0,1")
+ assertEquals(expected.toList.sorted, sink.getAppendResults.sorted)
+ }
+
+ /** Tests for left outer join **/
+ @Test
+ def testProcTimeLeftOuterJoin(): Unit = {
+ env.setParallelism(1)
+
+ val sqlQuery =
+ """
+ |SELECT t2.a, t2.c, t1.c
+ |FROM T1 AS t1 LEFT OUTER JOIN T2 AS t2 ON
+ | t1.a = t2.a AND
+ | t1.proctime BETWEEN t2.proctime - INTERVAL '5' SECOND AND
+ | t2.proctime + INTERVAL '3' SECOND
+ """.stripMargin
+
+ val data1 = new mutable.MutableList[(Int, Long, String)]
+ data1.+=((1, 1L, "Hi1"))
+ data1.+=((1, 2L, "Hi2"))
+ data1.+=((1, 5L, "Hi3"))
+ data1.+=((2, 7L, "Hi5"))
+
+ val data2 = new mutable.MutableList[(Int, Long, String)]
+ data2.+=((1, 1L, "HiHi"))
+ data2.+=((2, 2L, "HeHe"))
+
+ val t1 = env.fromCollection(data1)
+ .toTable(tEnv, 'a, 'b, 'c, 'proctime)
+ val t2 = env.fromCollection(data2)
+ .toTable(tEnv, 'a, 'b, 'c, 'proctime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+
+ }
+
+ /** Tests row time left outer join **/
+ @Test
+ def testRowTimeLeftOuterJoin(): Unit = {
+ val sqlQuery =
+ """
+ |SELECT t1.key, t2.id, t1.id
+ |FROM T1 AS t1 LEFT OUTER JOIN T2 AS t2 ON
+ | t1.key = t2.key AND
+ | t1.rowtime BETWEEN t2.rowtime - INTERVAL '5' SECOND AND
+ | t2.rowtime + INTERVAL '6' SECOND AND
+ | t1.id <> 'L-5'
+ """.stripMargin
+
+ val data1 = new mutable.MutableList[(String, String, Long)]
+ // for boundary test
+ data1.+=(("A", "L-1", 1000L))
+ data1.+=(("A", "L-2", 2000L))
+ data1.+=(("B", "L-4", 4000L))
+ data1.+=(("B", "L-5", 5000L))
+ data1.+=(("A", "L-6", 6000L))
+ data1.+=(("C", "L-7", 7000L))
+ data1.+=(("A", "L-10", 10000L))
+ data1.+=(("A", "L-12", 12000L))
+ data1.+=(("A", "L-20", 20000L))
+
+ val data2 = new mutable.MutableList[(String, String, Long)]
+ data2.+=(("A", "R-6", 6000L))
+ data2.+=(("B", "R-7", 7000L))
+ data2.+=(("D", "R-8", 8000L))
+ data2.+=(("A", "R-11", 11000L))
+
+ val t1 = env.fromCollection(data1)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+
+ val t2 = env.fromCollection(data2)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+ val expected = mutable.MutableList[String](
+ "A,R-6,L-1",
+ "A,R-6,L-2",
+ "A,R-6,L-6",
+ "A,R-6,L-10",
+ "A,R-6,L-12",
+ "B,R-7,L-4",
+ "A,R-11,L-6",
+ "A,R-11,L-10",
+ "A,R-11,L-12",
+ "B,null,L-5",
+ "C,null,L-7",
+ "A,null,L-20")
+
+ assertEquals(expected.toList.sorted, sink.getAppendResults.sorted)
+ }
+
+ @Test
+ def testRowTimeLeftOuterJoinNegativeWindowSize(): Unit = {
+ val sqlQuery =
+ """
+ |SELECT t2.key, t2.id, t1.id
+ |FROM T1 AS t1 LEFT OUTER JOIN T2 AS t2 ON
+ | t1.key = t2.key AND
+ | t1.rowtime BETWEEN t2.rowtime + INTERVAL '3' SECOND AND
+ | t2.rowtime + INTERVAL '1' SECOND
+ """.stripMargin
+
+ val data1 = new mutable.MutableList[(String, String, Long)]
+ // for boundary test
+ data1.+=(("A", "L-1", 1000L))
+ data1.+=(("B", "L-4", 4000L))
+ data1.+=(("C", "L-7", 7000L))
+
+ val data2 = new mutable.MutableList[(String, String, Long)]
+ data2.+=(("A", "R-6", 6000L))
+ data2.+=(("B", "R-7", 7000L))
+ data2.+=(("D", "R-8", 8000L))
+
+ val t1 = env.fromCollection(data1)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+ val t2 = env.fromCollection(data2)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+
+ val expected = mutable.MutableList[String](
+ "null,null,L-1",
+ "null,null,L-4",
+ "null,null,L-7"
+ )
+ assertEquals(expected.toList.sorted, sink.getAppendResults.sorted)
+ }
+
+ //Test for right outer join
+ @Test
+ def testProcTimeRightOuterJoin(): Unit = {
+ env.setParallelism(1)
+
+ val sqlQuery =
+ """
+ |SELECT t2.a, t2.c, t1.c
+ |FROM T1 as t1 RIGHT OUTER JOIN T2 as t2 ON
+ | t1.a = t2.a AND
+ | t1.proctime BETWEEN t2.proctime - INTERVAL '5' SECOND AND
+ | t2.proctime + INTERVAL '3' SECOND
+ """.stripMargin
+
+ val data1 = new mutable.MutableList[(Int, Long, String)]
+ data1.+=((1, 1L, "Hi1"))
+ data1.+=((1, 2L, "Hi2"))
+ data1.+=((1, 5L, "Hi3"))
+ data1.+=((2, 7L, "Hi5"))
+
+ val data2 = new mutable.MutableList[(Int, Long, String)]
+ data2.+=((1, 1L, "HiHi"))
+ data2.+=((2, 2L, "HeHe"))
+
+ val t1 = env.fromCollection(data1).toTable(tEnv, 'a, 'b, 'c, 'proctime)
+ val t2 = env.fromCollection(data2).toTable(tEnv, 'a, 'b, 'c, 'proctime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+
+ }
+
+ @Test
+ def testRowTimeRightOuterJoin(): Unit = {
+ val sqlQuery =
+ """
+ |SELECT t2.key, t2.id, t1.id
+ |FROM T1 AS t1 RIGHT OUTER JOIN T2 AS t2 ON
+ | t1.key = t2.key AND
+ | t1.rowtime BETWEEN t2.rowtime - INTERVAL '5' SECOND AND
+ | t2.rowtime + INTERVAL '6' SECOND AND
+ | t2.id <> 'R-5'
+ """.stripMargin
+
+ val data1 = new mutable.MutableList[(String, String, Long)]
+ // for boundary test
+ data1.+=(("A", "L-1", 1000L))
+ data1.+=(("A", "L-2", 2000L))
+ data1.+=(("B", "L-4", 4000L))
+ data1.+=(("A", "L-6", 6000L))
+ data1.+=(("C", "L-7", 7000L))
+ data1.+=(("A", "L-10", 10000L))
+ data1.+=(("A", "L-12", 12000L))
+
+ val data2 = new mutable.MutableList[(String, String, Long)]
+ data2.+=(("A", "R-5", 5000L))
+ data2.+=(("A", "R-6", 6000L))
+ data2.+=(("B", "R-7", 7000L))
+ data2.+=(("D", "R-8", 8000L))
+ data2.+=(("A", "R-20", 20000L))
+
+ val t1 = env.fromCollection(data1)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+ val t2 = env.fromCollection(data2)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+ val expected = mutable.MutableList[String](
+ "A,R-5,null",
+ "A,R-6,L-1",
+ "A,R-6,L-2",
+ "A,R-6,L-6",
+ "A,R-6,L-10",
+ "A,R-6,L-12",
+ "A,R-20,null",
+ "B,R-7,L-4",
+ "D,R-8,null"
+ )
+
+ assertEquals(expected.toList.sorted, sink.getAppendResults.sorted)
+ }
+
+ @Test
+ def testRowTimeRightOuterJoinNegativeWindowSize(): Unit = {
+ val sqlQuery =
+ """
+ |SELECT t2.key, t2.id, t1.id
+ |FROM T1 AS t1 RIGHT OUTER JOIN T2 AS t2 ON
+ |t1.key = t2.key AND
+ |t1.rowtime BETWEEN t2.rowtime + INTERVAL '5' SECOND AND
+ |t2.rowtime + INTERVAL '1' SECOND
+ """.stripMargin
+
+ val data1 = new mutable.MutableList[(String, String, Long)]
+ // for boundary test
+ data1.+=(("A", "L-1", 1000L))
+ data1.+=(("B", "L-4", 4000L))
+ data1.+=(("C", "L-7", 7000L))
+
+ val data2 = new mutable.MutableList[(String, String, Long)]
+ data2.+=(("A", "R-6", 6000L))
+ data2.+=(("B", "R-7", 7000L))
+ data2.+=(("D", "R-8", 8000L))
+
+ val t1 = env.fromCollection(data1)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+ val t2 = env.fromCollection(data2)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+
+ val expected = mutable.MutableList[String](
+ "A,R-6,null",
+ "B,R-7,null",
+ "D,R-8,null"
+ )
+ assertEquals(expected.toList.sorted, sink.getAppendResults.sorted)
+ }
+
+ //Tests for full outer join
+ @Test
+ def testProcTimeFullOuterJoin(): Unit = {
+ env.setParallelism(1)
+
+ val sqlQuery =
+ """
+ |SELECT t2.a, t2.c, t1.c
+ |FROM T1 as t1 FULL OUTER JOIN T2 as t2 ON
+ |t1.a = t2.a AND
+ |t1.proctime BETWEEN t2.proctime - INTERVAL '5' SECOND AND
+ |t2.proctime
+ """.stripMargin
+
+ val data1 = new mutable.MutableList[(Int, Long, String)]
+ data1.+=((1, 1L, "Hi1"))
+ data1.+=((1, 2L, "Hi2"))
+ data1.+=((1, 5L, "Hi3"))
+ data1.+=((2, 7L, "Hi5"))
+
+ val data2 = new mutable.MutableList[(Int, Long, String)]
+ data2.+=((1, 1L, "HiHi"))
+ data2.+=((2, 2L, "HeHe"))
+
+ val t1 = env.fromCollection(data1).toTable(tEnv, 'a, 'b, 'c, 'proctime)
+ val t2 = env.fromCollection(data2).toTable(tEnv, 'a, 'b, 'c, 'proctime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+ }
+
+ @Test
+ def testRowTimeFullOuterJoin(): Unit = {
+ val sqlQuery =
+ """
+ |SELECT t2.key, t2.id, t1.id
+ |FROM T1 AS t1 FULL OUTER JOIN T2 AS t2 ON
+ |t1.key = t2.key AND
+ |t1.rowtime BETWEEN t2.rowtime - INTERVAL '5' SECOND AND
+ |t2.rowtime + INTERVAL '6' SECOND AND
+ |NOT (t1.id = 'L-5' OR t2.id = 'R-5')
+ """.stripMargin
+
+ val data1 = new mutable.MutableList[(String, String, Long)]
+ // for boundary test
+ data1.+=(("A", "L-1", 1000L))
+ data1.+=(("A", "L-2", 2000L))
+ data1.+=(("B", "L-4", 4000L))
+ data1.+=(("B", "L-5", 5000L))
+ data1.+=(("A", "L-6", 6000L))
+ data1.+=(("C", "L-7", 7000L))
+ data1.+=(("A", "L-10", 10000L))
+ data1.+=(("A", "L-12", 12000L))
+ data1.+=(("A", "L-20", 20000L))
+
+ val data2 = new mutable.MutableList[(String, String, Long)]
+ data2.+=(("A", "R-5", 5000L))
+ data2.+=(("A", "R-6", 6000L))
+ data2.+=(("B", "R-7", 7000L))
+ data2.+=(("D", "R-8", 8000L))
+
+ val t1 = env.fromCollection(data1)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+ val t2 = env.fromCollection(data2)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+
+ val expected = mutable.MutableList[String](
+ "A,R-6,L-1",
+ "A,R-6,L-2",
+ "A,R-6,L-6",
+ "A,R-6,L-10",
+ "A,R-6,L-12",
+ "B,R-7,L-4",
+ "A,R-5,null",
+ "D,R-8,null",
+ "null,null,L-5",
+ "null,null,L-7",
+ "null,null,L-20"
+ )
+ assertEquals(expected.toList.sorted, sink.getAppendResults.sorted)
+ }
+
+ @Test
+ def testRowTimeFullOuterJoinNegativeWindowSize(): Unit = {
+ val sqlQuery =
+ """
+ |SELECT t2.key, t2.id, t1.id
+ |FROM T1 AS t1 FULL OUTER JOIN T2 AS t2 ON
+ |t1.key = t2.key AND
+ |t1.rowtime BETWEEN t2.rowtime + INTERVAL '5' SECOND AND
+ |t2.rowtime + INTERVAL '4' SECOND
+ """.stripMargin
+
+ val data1 = new mutable.MutableList[(String, String, Long)]
+ // for boundary test
+ data1.+=(("A", "L-1", 1000L))
+ data1.+=(("B", "L-4", 4000L))
+ data1.+=(("C", "L-7", 7000L))
+
+ val data2 = new mutable.MutableList[(String, String, Long)]
+ data2.+=(("A", "R-6", 6000L))
+ data2.+=(("B", "R-7", 7000L))
+ data2.+=(("D", "R-8", 8000L))
+
+ val t1 = env.fromCollection(data1)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+ val t2 = env.fromCollection(data2)
+ .assignTimestampsAndWatermarks(new Row3WatermarkExtractor2)
+ .toTable(tEnv, 'key, 'id, 'rowtime)
+
+ tEnv.registerTable("T1", t1)
+ tEnv.registerTable("T2", t2)
+
+ val sink = new TestingAppendSink
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(sink)
+ env.execute()
+
+ val expected = mutable.MutableList[String](
+ "null,null,L-1",
+ "null,null,L-4",
+ "null,null,L-7",
+ "A,R-6,null",
+ "B,R-7,null",
+ "D,R-8,null"
+ )
+ assertEquals(expected.toList.sorted, sink.getAppendResults.sorted)
+ }
+}
+
+private class Row4WatermarkExtractor
+ extends AssignerWithPunctuatedWatermarks[(Int, Long, String, Long)] {
+
+ override def checkAndGetNextWatermark(
+ lastElement: (Int, Long, String, Long),
+ extractedTimestamp: Long): Watermark = {
+ new Watermark(extractedTimestamp - 1)
+ }
+
+ override def extractTimestamp(
+ element: (Int, Long, String, Long),
+ previousElementTimestamp: Long): Long = {
+ element._4
+ }
+}
+
+private class Row3WatermarkExtractor2
+ extends AssignerWithPunctuatedWatermarks[(String, String, Long)] {
+
+ override def checkAndGetNextWatermark(
+ lastElement: (String, String, Long),
+ extractedTimestamp: Long): Watermark = {
+ new Watermark(extractedTimestamp - 1)
+ }
+
+ override def extractTimestamp(
+ element: (String, String, Long),
+ previousElementTimestamp: Long): Long = {
+ element._3
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/EmitAwareCollector.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/EmitAwareCollector.java
new file mode 100644
index 0000000..f9097bb
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/EmitAwareCollector.java
@@ -0,0 +1,56 @@
+/*
+ * 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.runtime.join;
+
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.util.Collector;
+
+/**
+ * Collector to wrap a [[org.apache.flink.table.dataformat.BaseRow]] and to track whether a row has been
+ * emitted by the inner collector.
+ */
+class EmitAwareCollector implements Collector<BaseRow> {
+
+ private boolean emitted = false;
+ private Collector<BaseRow> innerCollector;
+
+ void reset() {
+ emitted = false;
+ }
+
+ boolean isEmitted() {
+ return emitted;
+ }
+
+ void setInnerCollector(Collector<BaseRow> innerCollector) {
+ this.innerCollector = innerCollector;
+ }
+
+ @Override
+ public void collect(BaseRow record) {
+ emitted = true;
+ innerCollector.collect(record);
+ }
+
+ @Override
+ public void close() {
+ innerCollector.close();
+ }
+
+}
diff --git a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/plan/FlinkJoinRelType.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/FlinkJoinType.java
similarity index 53%
rename from flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/plan/FlinkJoinRelType.java
rename to flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/FlinkJoinType.java
index 99cd7ea..cbf6a35 100644
--- a/flink-table/flink-table-planner-blink/src/main/java/org/apache/flink/table/plan/FlinkJoinRelType.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/FlinkJoinType.java
@@ -16,16 +16,12 @@
* limitations under the License.
*/
-package org.apache.flink.table.plan;
-
-import org.apache.calcite.rel.core.Join;
-import org.apache.calcite.rel.core.JoinRelType;
-import org.apache.calcite.rel.core.SemiJoin;
+package org.apache.flink.table.runtime.join;
/**
- * Enumeration of join types.
+ * Join type for join.
*/
-public enum FlinkJoinRelType {
+public enum FlinkJoinType {
INNER, LEFT, RIGHT, FULL, SEMI, ANTI;
public boolean isOuter() {
@@ -59,45 +55,23 @@ public enum FlinkJoinRelType {
}
}
- /** Convert JoinRelType to FlinkJoinRelType. */
- public static FlinkJoinRelType toFlinkJoinRelType(JoinRelType joinType) {
- switch (joinType) {
- case INNER:
- return FlinkJoinRelType.INNER;
- case LEFT:
- return FlinkJoinRelType.LEFT;
- case RIGHT:
- return FlinkJoinRelType.RIGHT;
- case FULL:
- return FlinkJoinRelType.FULL;
- default:
- throw new IllegalArgumentException("invalid: " + joinType);
- }
- }
-
- public static FlinkJoinRelType getFlinkJoinRelType(Join join) {
- if (join instanceof SemiJoin) {
- // TODO supports ANTI
- return SEMI;
- } else {
- return toFlinkJoinRelType(join.getJoinType());
- }
- }
-
- /** Convert FlinkJoinRelType to JoinRelType. */
- public static JoinRelType toJoinRelType(FlinkJoinRelType joinType) {
- switch (joinType) {
+ @Override
+ public String toString() {
+ switch (this) {
case INNER:
- return JoinRelType.INNER;
+ return "InnerJoin";
case LEFT:
- return JoinRelType.LEFT;
+ return "LeftOuterJoin";
case RIGHT:
- return JoinRelType.RIGHT;
+ return "RightOuterJoin";
case FULL:
- return JoinRelType.FULL;
+ return "FullOuterJoin";
+ case SEMI:
+ return "LeftSemiJoin";
+ case ANTI:
+ return "LeftAntiJoin";
default:
- throw new IllegalArgumentException("invalid: " + joinType);
+ throw new IllegalArgumentException("Invalid join type: " + name());
}
}
}
-
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/KeyedCoProcessOperatorWithWatermarkDelay.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/KeyedCoProcessOperatorWithWatermarkDelay.java
new file mode 100644
index 0000000..f76c27a
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/KeyedCoProcessOperatorWithWatermarkDelay.java
@@ -0,0 +1,60 @@
+/*
+ * 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.runtime.join;
+
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Serializable;
+import java.util.function.Consumer;
+
+/**
+ * A {@link KeyedCoProcessOperator} that supports holding back watermarks with a static delay.
+ */
+public class KeyedCoProcessOperatorWithWatermarkDelay<K, IN1, IN2, OUT>
+ extends KeyedCoProcessOperator<K, IN1, IN2, OUT> {
+
+ private static final long serialVersionUID = -7435774708099223442L;
+
+ private final Consumer<Watermark> emitter;
+
+ public KeyedCoProcessOperatorWithWatermarkDelay(CoProcessFunction<IN1, IN2, OUT> flatMapper, long watermarkDelay) {
+ super(flatMapper);
+ Preconditions.checkArgument(watermarkDelay >= 0, "The watermark delay should be non-negative.");
+ if (watermarkDelay == 0) {
+ // emits watermark without delay
+ emitter = (Consumer<Watermark> & Serializable) (Watermark mark) -> output.emitWatermark(mark);
+ } else {
+ // emits watermark with delay
+ emitter = (Consumer<Watermark> & Serializable) (Watermark mark) -> output
+ .emitWatermark(new Watermark(mark.getTimestamp() - watermarkDelay));
+ }
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ if (timeServiceManager != null) {
+ timeServiceManager.advanceWatermark(mark);
+ }
+ emitter.accept(mark);
+ }
+
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/OuterJoinPaddingUtil.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/OuterJoinPaddingUtil.java
new file mode 100644
index 0000000..4242730
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/OuterJoinPaddingUtil.java
@@ -0,0 +1,91 @@
+/*
+ * 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.runtime.join;
+
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.dataformat.GenericRow;
+import org.apache.flink.table.dataformat.JoinedRow;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.Serializable;
+
+/**
+ * An utility to generate reusable padding results for outer joins.
+ */
+public class OuterJoinPaddingUtil implements Serializable {
+
+ private static final long serialVersionUID = -2295909099427806938L;
+
+ private final int leftArity;
+ private final int rightArity;
+ private transient JoinedRow joinedRow = new JoinedRow();
+ private transient GenericRow leftNullPaddingRow;
+ private transient GenericRow rightNullPaddingRow;
+
+ public OuterJoinPaddingUtil(int leftArity, int rightArity) {
+ this.leftArity = leftArity;
+ this.rightArity = rightArity;
+ initLeftNullPaddingRow();
+ initRightNullPaddingRow();
+ }
+
+ private void initLeftNullPaddingRow() {
+ //Initialize the two reusable padding results
+ leftNullPaddingRow = new GenericRow(leftArity);
+ for (int idx = 0; idx < leftArity; idx++) {
+ leftNullPaddingRow.setNullAt(idx);
+ }
+ }
+
+ private void initRightNullPaddingRow() {
+ rightNullPaddingRow = new GenericRow(rightArity);
+ for (int idx = 0; idx < rightArity; idx++) {
+ rightNullPaddingRow.setNullAt(idx);
+ }
+ }
+
+ /**
+ * Returns a padding result with the given right row.
+ *
+ * @param rightRow the right row to pad
+ * @return the reusable null padding result
+ */
+ public final BaseRow padRight(BaseRow rightRow) {
+ return joinedRow.replace(leftNullPaddingRow, rightRow);
+ }
+
+ /**
+ * Returns a padding result with the given left row.
+ *
+ * @param leftRow the left row to pad
+ * @return the reusable null padding result
+ */
+ public final BaseRow padLeft(BaseRow leftRow) {
+ return joinedRow.replace(leftRow, rightNullPaddingRow);
+ }
+
+ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
+ in.defaultReadObject();
+
+ joinedRow = new JoinedRow();
+ initLeftNullPaddingRow();
+ initRightNullPaddingRow();
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/ProcTimeBoundedStreamJoin.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/ProcTimeBoundedStreamJoin.java
new file mode 100644
index 0000000..08c41fd
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/ProcTimeBoundedStreamJoin.java
@@ -0,0 +1,63 @@
+/*
+ * 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.runtime.join;
+
+import org.apache.flink.api.common.functions.FlatJoinFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.generated.GeneratedFunction;
+
+/**
+ * The function to execute processing time bounded stream inner-join.
+ */
+public final class ProcTimeBoundedStreamJoin extends TimeBoundedStreamJoin {
+
+ private static final long serialVersionUID = 9204647938032023101L;
+
+ public ProcTimeBoundedStreamJoin(
+ FlinkJoinType joinType,
+ long leftLowerBound,
+ long leftUpperBound,
+ TypeInformation<BaseRow> leftType,
+ TypeInformation<BaseRow> rightType,
+ GeneratedFunction<FlatJoinFunction<BaseRow, BaseRow, BaseRow>> genJoinFunc) {
+ super(joinType, leftLowerBound, leftUpperBound, 0L, leftType, rightType, genJoinFunc);
+ }
+
+ @Override
+ void updateOperatorTime(Context ctx) {
+ leftOperatorTime = ctx.timerService().currentProcessingTime();
+ rightOperatorTime = leftOperatorTime;
+ }
+
+ @Override
+ long getTimeForLeftStream(Context ctx, BaseRow row) {
+ return leftOperatorTime;
+ }
+
+ @Override
+ long getTimeForRightStream(Context ctx, BaseRow row) {
+ return rightOperatorTime;
+ }
+
+ @Override
+ void registerTimer(Context ctx, long cleanupTime) {
+ ctx.timerService().registerProcessingTimeTimer(cleanupTime);
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/RowTimeBoundedStreamJoin.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/RowTimeBoundedStreamJoin.java
new file mode 100644
index 0000000..ed73834
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/RowTimeBoundedStreamJoin.java
@@ -0,0 +1,83 @@
+/*
+ * 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.runtime.join;
+
+import org.apache.flink.api.common.functions.FlatJoinFunction;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.generated.GeneratedFunction;
+
+/**
+ * The function to execute row(event) time bounded stream inner-join.
+ */
+public final class RowTimeBoundedStreamJoin extends TimeBoundedStreamJoin {
+
+ private static final long serialVersionUID = -2923709329817468698L;
+
+ private final int leftTimeIdx;
+ private final int rightTimeIdx;
+
+ public RowTimeBoundedStreamJoin(
+ FlinkJoinType joinType,
+ long leftLowerBound,
+ long leftUpperBound,
+ long allowedLateness,
+ TypeInformation<BaseRow> leftType,
+ TypeInformation<BaseRow> rightType,
+ GeneratedFunction<FlatJoinFunction<BaseRow, BaseRow, BaseRow>> genJoinFunc,
+ int leftTimeIdx,
+ int rightTimeIdx) {
+ super(joinType, leftLowerBound, leftUpperBound, allowedLateness, leftType, rightType, genJoinFunc);
+ this.leftTimeIdx = leftTimeIdx;
+ this.rightTimeIdx = rightTimeIdx;
+ }
+
+ /**
+ * Get the maximum interval between receiving a row and emitting it (as part of a joined result).
+ * This is the time interval by which watermarks need to be held back.
+ *
+ * @return the maximum delay for the outputs
+ */
+ public long getMaxOutputDelay() {
+ return Math.max(leftRelativeSize, rightRelativeSize) + allowedLateness;
+ }
+
+ @Override
+ void updateOperatorTime(Context ctx) {
+ leftOperatorTime = ctx.timerService().currentWatermark() > 0 ? ctx.timerService().currentWatermark() : 0L;
+ // We may set different operator times in the future.
+ rightOperatorTime = leftOperatorTime;
+ }
+
+ @Override
+ long getTimeForLeftStream(Context ctx, BaseRow row) {
+ return row.getLong(leftTimeIdx);
+ }
+
+ @Override
+ long getTimeForRightStream(Context ctx, BaseRow row) {
+ return row.getLong(rightTimeIdx);
+ }
+
+ @Override
+ void registerTimer(Context ctx, long cleanupTime) {
+ // Maybe we can register timers for different streams in the future.
+ ctx.timerService().registerEventTimeTimer(cleanupTime);
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/SortMergeJoinOperator.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/SortMergeJoinOperator.java
index 2390111..16415d2 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/SortMergeJoinOperator.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/SortMergeJoinOperator.java
@@ -66,7 +66,7 @@ public class SortMergeJoinOperator extends TableStreamOperator<BaseRow>
private final long reservedSortMemory1;
private final long reservedSortMemory2;
private final long externalBufferMemory;
- private final SortMergeJoinType type;
+ private final FlinkJoinType type;
private final boolean leftIsSmaller;
private final boolean[] filterNulls;
@@ -99,7 +99,7 @@ public class SortMergeJoinOperator extends TableStreamOperator<BaseRow>
@VisibleForTesting
public SortMergeJoinOperator(
- long reservedSortMemory, long externalBufferMemory, SortMergeJoinType type, boolean leftIsSmaller,
+ long reservedSortMemory, long externalBufferMemory, FlinkJoinType type, boolean leftIsSmaller,
GeneratedJoinCondition condFuncCode,
GeneratedProjection projectionCode1, GeneratedProjection projectionCode2,
GeneratedNormalizedKeyComputer computer1, GeneratedRecordComparator comparator1,
@@ -114,7 +114,7 @@ public class SortMergeJoinOperator extends TableStreamOperator<BaseRow>
public SortMergeJoinOperator(
long reservedSortMemory1,
long reservedSortMemory2,
- long externalBufferMemory, SortMergeJoinType type, boolean leftIsSmaller,
+ long externalBufferMemory, FlinkJoinType type, boolean leftIsSmaller,
GeneratedJoinCondition condFuncCode,
GeneratedProjection projectionCode1, GeneratedProjection projectionCode2,
GeneratedNormalizedKeyComputer computer1, GeneratedRecordComparator comparator1,
@@ -227,7 +227,7 @@ public class SortMergeJoinOperator extends TableStreamOperator<BaseRow>
MutableObjectIterator iterator1 = sorter1.getIterator();
MutableObjectIterator iterator2 = sorter2.getIterator();
- if (type.equals(SortMergeJoinType.INNER)) {
+ if (type.equals(FlinkJoinType.INNER)) {
if (!leftIsSmaller) {
try (SortMergeInnerJoinIterator joinIterator = new SortMergeInnerJoinIterator(
serializer1, serializer2, projection1, projection2,
@@ -241,26 +241,26 @@ public class SortMergeJoinOperator extends TableStreamOperator<BaseRow>
innerJoin(joinIterator, true);
}
}
- } else if (type.equals(SortMergeJoinType.LEFT)) {
+ } else if (type.equals(FlinkJoinType.LEFT)) {
try (SortMergeOneSideOuterJoinIterator joinIterator = new SortMergeOneSideOuterJoinIterator(
serializer1, serializer2, projection1, projection2,
keyComparator, iterator1, iterator2, newBuffer(serializer2), filterNulls)) {
oneSideOuterJoin(joinIterator, false, rightNullRow);
}
- } else if (type.equals(SortMergeJoinType.RIGHT)) {
+ } else if (type.equals(FlinkJoinType.RIGHT)) {
try (SortMergeOneSideOuterJoinIterator joinIterator = new SortMergeOneSideOuterJoinIterator(
serializer2, serializer1, projection2, projection1,
keyComparator, iterator2, iterator1, newBuffer(serializer1), filterNulls)) {
oneSideOuterJoin(joinIterator, true, leftNullRow);
}
- } else if (type.equals(SortMergeJoinType.FULL)) {
+ } else if (type.equals(FlinkJoinType.FULL)) {
try (SortMergeFullOuterJoinIterator fullOuterJoinIterator = new SortMergeFullOuterJoinIterator(
serializer1, serializer2, projection1, projection2,
keyComparator, iterator1, iterator2,
newBuffer(serializer1), newBuffer(serializer2), filterNulls)) {
fullOuterJoin(fullOuterJoinIterator);
}
- } else if (type.equals(SortMergeJoinType.SEMI)) {
+ } else if (type.equals(FlinkJoinType.SEMI)) {
try (SortMergeInnerJoinIterator joinIterator = new SortMergeInnerJoinIterator(
serializer1, serializer2, projection1, projection2,
keyComparator, iterator1, iterator2, newBuffer(serializer2), filterNulls)) {
@@ -281,7 +281,7 @@ public class SortMergeJoinOperator extends TableStreamOperator<BaseRow>
}
}
}
- } else if (type.equals(SortMergeJoinType.ANTI)) {
+ } else if (type.equals(FlinkJoinType.ANTI)) {
try (SortMergeOneSideOuterJoinIterator joinIterator = new SortMergeOneSideOuterJoinIterator(
serializer1, serializer2, projection1, projection2,
keyComparator, iterator1, iterator2, newBuffer(serializer2), filterNulls)) {
@@ -458,10 +458,4 @@ public class SortMergeJoinOperator extends TableStreamOperator<BaseRow>
}
}
- /**
- * Join type for sort merge join.
- */
- public enum SortMergeJoinType {
- INNER, LEFT, RIGHT, FULL, SEMI, ANTI
- }
}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/TimeBoundedStreamJoin.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/TimeBoundedStreamJoin.java
new file mode 100644
index 0000000..d3d31d0
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/join/TimeBoundedStreamJoin.java
@@ -0,0 +1,477 @@
+/*
+ * 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.runtime.join;
+
+import org.apache.flink.api.common.functions.FlatJoinFunction;
+import org.apache.flink.api.common.state.MapState;
+import org.apache.flink.api.common.state.MapStateDescriptor;
+import org.apache.flink.api.common.state.ValueState;
+import org.apache.flink.api.common.state.ValueStateDescriptor;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ListTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.generated.GeneratedFunction;
+import org.apache.flink.util.Collector;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A CoProcessFunction to execute time-bounded stream inner-join.
+ * Two kinds of time criteria:
+ * "L.time between R.time + X and R.time + Y" or "R.time between L.time - Y and L.time - X"
+ * X and Y might be negative or positive and X <= Y.
+ */
+abstract class TimeBoundedStreamJoin extends CoProcessFunction<BaseRow, BaseRow, BaseRow> {
+ private static final Logger LOGGER = LoggerFactory.getLogger(TimeBoundedStreamJoin.class);
+ private final FlinkJoinType joinType;
+ protected final long leftRelativeSize;
+ protected final long rightRelativeSize;
+
+ // Minimum interval by which state is cleaned up
+ private final long minCleanUpInterval;
+ protected final long allowedLateness;
+ private final TypeInformation<BaseRow> leftType;
+ private final TypeInformation<BaseRow> rightType;
+ private GeneratedFunction<FlatJoinFunction<BaseRow, BaseRow, BaseRow>> genJoinFunc;
+ private transient OuterJoinPaddingUtil paddingUtil;
+
+ private transient EmitAwareCollector joinCollector;
+
+ // the join function for other conditions
+ private transient FlatJoinFunction<BaseRow, BaseRow, BaseRow> joinFunction;
+
+ // cache to store rows form the left stream
+ private transient MapState<Long, List<Tuple2<BaseRow, Boolean>>> leftCache;
+ // cache to store rows from the right stream
+ private transient MapState<Long, List<Tuple2<BaseRow, Boolean>>> rightCache;
+
+ // state to record the timer on the left stream. 0 means no timer set
+ private transient ValueState<Long> leftTimerState;
+ // state to record the timer on the right stream. 0 means no timer set
+ private transient ValueState<Long> rightTimerState;
+
+ // Points in time until which the respective cache has been cleaned.
+ private long leftExpirationTime = 0L;
+ private long rightExpirationTime = 0L;
+
+ // Current time on the respective input stream.
+ protected long leftOperatorTime = 0L;
+ protected long rightOperatorTime = 0L;
+
+ TimeBoundedStreamJoin(
+ FlinkJoinType joinType,
+ long leftLowerBound,
+ long leftUpperBound,
+ long allowedLateness,
+ TypeInformation<BaseRow> leftType,
+ TypeInformation<BaseRow> rightType,
+ GeneratedFunction<FlatJoinFunction<BaseRow, BaseRow, BaseRow>> genJoinFunc) {
+ this.joinType = joinType;
+ this.leftRelativeSize = -leftLowerBound;
+ this.rightRelativeSize = leftUpperBound;
+ minCleanUpInterval = (leftRelativeSize + rightRelativeSize) / 2;
+ if (allowedLateness < 0) {
+ throw new IllegalArgumentException("The allowed lateness must be non-negative.");
+ }
+ this.allowedLateness = allowedLateness;
+ this.leftType = leftType;
+ this.rightType = rightType;
+ this.genJoinFunc = genJoinFunc;
+ }
+
+ @Override
+ public void open(Configuration parameters) throws Exception {
+ LOGGER.debug("Instantiating JoinFunction: {} \n\n Code:\n{}", genJoinFunc.getClassName(),
+ genJoinFunc.getCode());
+ joinFunction = genJoinFunc.newInstance(getRuntimeContext().getUserCodeClassLoader());
+ genJoinFunc = null;
+
+ joinCollector = new EmitAwareCollector();
+
+ // Initialize the data caches.
+ ListTypeInfo<Tuple2<BaseRow, Boolean>> leftRowListTypeInfo = new ListTypeInfo<>(
+ new TupleTypeInfo<>(leftType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<BaseRow, Boolean>>> leftMapStateDescriptor = new MapStateDescriptor<>(
+ "WindowJoinLeftCache",
+ BasicTypeInfo.LONG_TYPE_INFO,
+ leftRowListTypeInfo);
+ leftCache = getRuntimeContext().getMapState(leftMapStateDescriptor);
+
+ ListTypeInfo<Tuple2<BaseRow, Boolean>> rightRowListTypeInfo = new ListTypeInfo<>(
+ new TupleTypeInfo<>(rightType, BasicTypeInfo.BOOLEAN_TYPE_INFO));
+ MapStateDescriptor<Long, List<Tuple2<BaseRow, Boolean>>> rightMapStateDescriptor = new MapStateDescriptor<>(
+ "WindowJoinRightCache",
+ BasicTypeInfo.LONG_TYPE_INFO,
+ rightRowListTypeInfo);
+ rightCache = getRuntimeContext().getMapState(rightMapStateDescriptor);
+
+ // Initialize the timer states.
+ ValueStateDescriptor<Long> leftValueStateDescriptor = new ValueStateDescriptor<>(
+ "WindowJoinLeftTimerState",
+ Long.class);
+ leftTimerState = getRuntimeContext().getState(leftValueStateDescriptor);
+
+ ValueStateDescriptor<Long> rightValueStateDescriptor = new ValueStateDescriptor<>(
+ "WindowJoinRightTimerState",
+ Long.class);
+ rightTimerState = getRuntimeContext().getState(rightValueStateDescriptor);
+
+ paddingUtil = new OuterJoinPaddingUtil(leftType.getArity(), rightType.getArity());
+ }
+
+ @Override
+ public void processElement1(BaseRow leftRow, Context ctx, Collector<BaseRow> out) throws Exception {
+ joinCollector.setInnerCollector(out);
+ updateOperatorTime(ctx);
+
+ long timeForLeftRow = getTimeForLeftStream(ctx, leftRow);
+ long rightQualifiedLowerBound = timeForLeftRow - rightRelativeSize;
+ long rightQualifiedUpperBound = timeForLeftRow + leftRelativeSize;
+ boolean emitted = false;
+
+ // Check if we need to join the current row against cached rows of the right input.
+ // The condition here should be rightMinimumTime < rightQualifiedUpperBound.
+ // We use rightExpirationTime as an approximation of the rightMinimumTime here,
+ // since rightExpirationTime <= rightMinimumTime is always true.
+ if (rightExpirationTime < rightQualifiedUpperBound) {
+ // Upper bound of current join window has not passed the cache expiration time yet.
+ // There might be qualifying rows in the cache that the current row needs to be joined with.
+ rightExpirationTime = calExpirationTime(leftOperatorTime, rightRelativeSize);
+ // Join the leftRow with rows from the right cache.
+ Iterator<Map.Entry<Long, List<Tuple2<BaseRow, Boolean>>>> rightIterator = rightCache.iterator();
+ while (rightIterator.hasNext()) {
+ Map.Entry<Long, List<Tuple2<BaseRow, Boolean>>> rightEntry = rightIterator.next();
+ Long rightTime = rightEntry.getKey();
+ if (rightTime >= rightQualifiedLowerBound && rightTime <= rightQualifiedUpperBound) {
+ List<Tuple2<BaseRow, Boolean>> rightRows = rightEntry.getValue();
+ boolean entryUpdated = false;
+ for (Tuple2<BaseRow, Boolean> tuple : rightRows) {
+ joinCollector.reset();
+ joinFunction.join(leftRow, tuple.f0, joinCollector);
+ emitted = emitted || joinCollector.isEmitted();
+ if (joinType.isRightOuter()) {
+ if (!tuple.f1 && joinCollector.isEmitted()) {
+ // Mark the right row as being successfully joined and emitted.
+ tuple.f1 = true;
+ entryUpdated = true;
+ }
+ }
+ }
+ if (entryUpdated) {
+ // Write back the edited entry (mark emitted) for the right cache.
+ rightEntry.setValue(rightRows);
+ }
+ }
+ // Clean up the expired right cache row, clean the cache while join
+ if (rightTime <= rightExpirationTime) {
+ if (joinType.isRightOuter()) {
+ List<Tuple2<BaseRow, Boolean>> rightRows = rightEntry.getValue();
+ rightRows.forEach((Tuple2<BaseRow, Boolean> tuple) -> {
+ if (!tuple.f1) {
+ // Emit a null padding result if the right row has never been successfully joined.
+ joinCollector.collect(paddingUtil.padRight(tuple.f0));
+ }
+ });
+ }
+ // eager remove
+ rightIterator.remove();
+ } // We could do the short-cutting optimization here once we get a state with ordered keys.
+ }
+ }
+ // Check if we need to cache the current row.
+ if (rightOperatorTime < rightQualifiedUpperBound) {
+ // Operator time of right stream has not exceeded the upper window bound of the current
+ // row. Put it into the left cache, since later coming records from the right stream are
+ // expected to be joined with it.
+ List<Tuple2<BaseRow, Boolean>> leftRowList = leftCache.get(timeForLeftRow);
+ if (leftRowList == null) {
+ leftRowList = new ArrayList<>(1);
+ }
+ leftRowList.add(Tuple2.of(leftRow, emitted));
+ leftCache.put(timeForLeftRow, leftRowList);
+ if (rightTimerState.value() == null) {
+ // Register a timer on the RIGHT stream to remove rows.
+ registerCleanUpTimer(ctx, timeForLeftRow, true);
+ }
+ } else if (!emitted && joinType.isLeftOuter()) {
+ // Emit a null padding result if the left row is not cached and successfully joined.
+ joinCollector.collect(paddingUtil.padLeft(leftRow));
+ }
+ }
+
+ @Override
+ public void processElement2(BaseRow rightRow, Context ctx, Collector<BaseRow> out) throws Exception {
+ joinCollector.setInnerCollector(out);
+ updateOperatorTime(ctx);
+ long timeForRightRow = getTimeForRightStream(ctx, rightRow);
+ long leftQualifiedLowerBound = timeForRightRow - leftRelativeSize;
+ long leftQualifiedUpperBound = timeForRightRow + rightRelativeSize;
+ boolean emitted = false;
+
+ // Check if we need to join the current row against cached rows of the left input.
+ // The condition here should be leftMinimumTime < leftQualifiedUpperBound.
+ // We use leftExpirationTime as an approximation of the leftMinimumTime here,
+ // since leftExpirationTime <= leftMinimumTime is always true.
+ if (leftExpirationTime < leftQualifiedUpperBound) {
+ leftExpirationTime = calExpirationTime(rightOperatorTime, leftRelativeSize);
+ // Join the rightRow with rows from the left cache.
+ Iterator<Map.Entry<Long, List<Tuple2<BaseRow, Boolean>>>> leftIterator = leftCache.iterator();
+ while (leftIterator.hasNext()) {
+ Map.Entry<Long, List<Tuple2<BaseRow, Boolean>>> leftEntry = leftIterator.next();
+ Long leftTime = leftEntry.getKey();
+ if (leftTime >= leftQualifiedLowerBound && leftTime <= leftQualifiedUpperBound) {
+ List<Tuple2<BaseRow, Boolean>> leftRows = leftEntry.getValue();
+ boolean entryUpdated = false;
+ for (Tuple2<BaseRow, Boolean> tuple : leftRows) {
+ joinCollector.reset();
+ joinFunction.join(tuple.f0, rightRow, joinCollector);
+ emitted = emitted || joinCollector.isEmitted();
+ if (joinType.isLeftOuter()) {
+ if (!tuple.f1 && joinCollector.isEmitted()) {
+ // Mark the left row as being successfully joined and emitted.
+ tuple.f1 = true;
+ entryUpdated = true;
+ }
+ }
+ }
+ if (entryUpdated) {
+ // Write back the edited entry (mark emitted) for the right cache.
+ leftEntry.setValue(leftRows);
+ }
+ }
+
+ if (leftTime <= leftExpirationTime) {
+ if (joinType.isLeftOuter()) {
+ List<Tuple2<BaseRow, Boolean>> leftRows = leftEntry.getValue();
+ leftRows.forEach((Tuple2<BaseRow, Boolean> tuple) -> {
+ if (!tuple.f1) {
+ // Emit a null padding result if the left row has never been successfully joined.
+ joinCollector.collect(paddingUtil.padLeft(tuple.f0));
+ }
+ });
+ }
+ // eager remove
+ leftIterator.remove();
+ } // We could do the short-cutting optimization here once we get a state with ordered keys.
+ }
+ }
+ // Check if we need to cache the current row.
+ if (leftOperatorTime < leftQualifiedUpperBound) {
+ // Operator time of left stream has not exceeded the upper window bound of the current
+ // row. Put it into the right cache, since later coming records from the left stream are
+ // expected to be joined with it.
+ List<Tuple2<BaseRow, Boolean>> rightRowList = rightCache.get(timeForRightRow);
+ if (null == rightRowList) {
+ rightRowList = new ArrayList<>(1);
+ }
+ rightRowList.add(Tuple2.of(rightRow, emitted));
+ rightCache.put(timeForRightRow, rightRowList);
+ if (leftTimerState.value() == null) {
+ // Register a timer on the LEFT stream to remove rows.
+ registerCleanUpTimer(ctx, timeForRightRow, false);
+ }
+ } else if (!emitted && joinType.isRightOuter()) {
+ // Emit a null padding result if the right row is not cached and successfully joined.
+ joinCollector.collect(paddingUtil.padRight(rightRow));
+ }
+ }
+
+ @Override
+ public void onTimer(long timestamp, OnTimerContext ctx, Collector<BaseRow> out) throws Exception {
+ joinCollector.setInnerCollector(out);
+ updateOperatorTime(ctx);
+ // In the future, we should separate the left and right watermarks. Otherwise, the
+ // registered timer of the faster stream will be delayed, even if the watermarks have
+ // already been emitted by the source.
+ Long leftCleanUpTime = leftTimerState.value();
+ if (leftCleanUpTime != null && timestamp == leftCleanUpTime) {
+ rightExpirationTime = calExpirationTime(leftOperatorTime, rightRelativeSize);
+ removeExpiredRows(
+ joinCollector,
+ rightExpirationTime,
+ rightCache,
+ leftTimerState,
+ ctx,
+ false
+ );
+ }
+
+ Long rightCleanUpTime = rightTimerState.value();
+ if (rightCleanUpTime != null && timestamp == rightCleanUpTime) {
+ leftExpirationTime = calExpirationTime(rightOperatorTime, leftRelativeSize);
+ removeExpiredRows(
+ joinCollector,
+ leftExpirationTime,
+ leftCache,
+ rightTimerState,
+ ctx,
+ true
+ );
+ }
+ }
+
+ /**
+ * Calculate the expiration time with the given operator time and relative window size.
+ *
+ * @param operatorTime the operator time
+ * @param relativeSize the relative window size
+ * @return the expiration time for cached rows
+ */
+ private long calExpirationTime(long operatorTime, long relativeSize) {
+ if (operatorTime < Long.MAX_VALUE) {
+ return operatorTime - relativeSize - allowedLateness - 1;
+ } else {
+ // When operatorTime = Long.MaxValue, it means the stream has reached the end.
+ return Long.MAX_VALUE;
+ }
+ }
+
+ /**
+ * Register a timer for cleaning up rows in a specified time.
+ *
+ * @param ctx the context to register timer
+ * @param rowTime time for the input row
+ * @param leftRow whether this row comes from the left stream
+ */
+ private void registerCleanUpTimer(Context ctx, long rowTime, boolean leftRow) throws IOException {
+ if (leftRow) {
+ long cleanUpTime = rowTime + leftRelativeSize + minCleanUpInterval + allowedLateness + 1;
+ registerTimer(ctx, cleanUpTime);
+ rightTimerState.update(cleanUpTime);
+ } else {
+ long cleanUpTime = rowTime + rightRelativeSize + minCleanUpInterval + allowedLateness + 1;
+ registerTimer(ctx, cleanUpTime);
+ leftTimerState.update(cleanUpTime);
+ }
+ }
+
+ /**
+ * Remove the expired rows. Register a new timer if the cache still holds valid rows
+ * after the cleaning up.
+ *
+ * @param collector the collector to emit results
+ * @param expirationTime the expiration time for this cache
+ * @param rowCache the row cache
+ * @param timerState timer state for the opposite stream
+ * @param ctx the context to register the cleanup timer
+ * @param removeLeft whether to remove the left rows
+ */
+ private void removeExpiredRows(
+ Collector<BaseRow> collector,
+ long expirationTime,
+ MapState<Long, List<Tuple2<BaseRow, Boolean>>> rowCache,
+ ValueState<Long> timerState,
+ OnTimerContext ctx,
+ boolean removeLeft) throws Exception {
+ Iterator<Map.Entry<Long, List<Tuple2<BaseRow, Boolean>>>> iterator = rowCache.iterator();
+
+ long earliestTimestamp = -1L;
+
+ // We remove all expired keys and do not leave the loop early.
+ // Hence, we do a full pass over the state.
+ while (iterator.hasNext()) {
+ Map.Entry<Long, List<Tuple2<BaseRow, Boolean>>> entry = iterator.next();
+ Long rowTime = entry.getKey();
+ if (rowTime <= expirationTime) {
+ if (removeLeft && joinType.isLeftOuter()) {
+ List<Tuple2<BaseRow, Boolean>> rows = entry.getValue();
+ rows.forEach((Tuple2<BaseRow, Boolean> tuple) -> {
+ if (!tuple.f1) {
+ // Emit a null padding result if the row has never been successfully joined.
+ collector.collect(paddingUtil.padLeft(tuple.f0));
+ }
+ });
+ } else if (!removeLeft && joinType.isRightOuter()) {
+ List<Tuple2<BaseRow, Boolean>> rows = entry.getValue();
+ rows.forEach((Tuple2<BaseRow, Boolean> tuple) -> {
+ if (!tuple.f1) {
+ // Emit a null padding result if the row has never been successfully joined.
+ collector.collect(paddingUtil.padRight(tuple.f0));
+ }
+ });
+ }
+ iterator.remove();
+ } else {
+ // We find the earliest timestamp that is still valid.
+ if (rowTime < earliestTimestamp || earliestTimestamp < 0) {
+ earliestTimestamp = rowTime;
+ }
+ }
+ }
+
+ if (earliestTimestamp > 0) {
+ // There are rows left in the cache. Register a timer to expire them later.
+ registerCleanUpTimer(ctx, earliestTimestamp, removeLeft);
+ } else {
+ // No rows left in the cache. Clear the states and the timerState will be 0.
+ timerState.clear();
+ rowCache.clear();
+ }
+
+ }
+
+ /**
+ * Update the operator time of the two streams.
+ * Must be the first call in all processing methods (i.e., processElement(), onTimer()).
+ *
+ * @param ctx the context to acquire watermarks
+ */
+ abstract void updateOperatorTime(Context ctx);
+
+ /**
+ * Return the time for the target row from the left stream.
+ * Requires that [[updateOperatorTime()]] has been called before.
+ *
+ * @param ctx the runtime context
+ * @param row the target row
+ * @return time for the target row
+ */
+ abstract long getTimeForLeftStream(Context ctx, BaseRow row);
+
+ /**
+ * Return the time for the target row from the right stream.
+ * Requires that [[updateOperatorTime()]] has been called before.
+ *
+ * @param ctx the runtime context
+ * @param row the target row
+ * @return time for the target row
+ */
+ abstract long getTimeForRightStream(Context ctx, BaseRow row);
+
+ /**
+ * Register a proctime or rowtime timer.
+ *
+ * @param ctx the context to register the timer
+ * @param cleanupTime timestamp for the timer
+ */
+ abstract void registerTimer(Context ctx, long cleanupTime);
+
+}
diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/rank/RetractableTopNFunction.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/rank/RetractableTopNFunction.java
index eb1abef..20e77ec 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/rank/RetractableTopNFunction.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/rank/RetractableTopNFunction.java
@@ -49,7 +49,7 @@ import java.util.SortedMap;
import java.util.TreeMap;
/**
- * The function could handle update input stream, input stream could contain acc, delete, retract or update record .
+ * The function could handle retract stream. Input stream could only contain acc, delete or retract record.
*/
public class RetractableTopNFunction extends AbstractTopNFunction {
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/DeduplicateKeepFirstRowFunctionTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/DeduplicateKeepFirstRowFunctionTest.java
index 1451397..cbfc2be 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/DeduplicateKeepFirstRowFunctionTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/DeduplicateKeepFirstRowFunctionTest.java
@@ -54,10 +54,10 @@ public class DeduplicateKeepFirstRowFunctionTest extends DeduplicateFunctionTest
testHarness.close();
// Keep FirstRow in deduplicate will not send retraction
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 1L, 12));
- expectedOutputOutput.add(record("book", 2L, 11));
- assertor.assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 1L, 12));
+ expectedOutput.add(record("book", 2L, 11));
+ assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/DeduplicateKeepLastRowFunctionTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/DeduplicateKeepLastRowFunctionTest.java
index 02a00de..c60b22c 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/DeduplicateKeepLastRowFunctionTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/DeduplicateKeepLastRowFunctionTest.java
@@ -58,11 +58,11 @@ public class DeduplicateKeepLastRowFunctionTest extends DeduplicateFunctionTestB
testHarness.processElement(record("book", 1L, 13));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 1L, 12));
- expectedOutputOutput.add(record("book", 2L, 11));
- expectedOutputOutput.add(record("book", 1L, 13));
- assertor.assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 1L, 12));
+ expectedOutput.add(record("book", 2L, 11));
+ expectedOutput.add(record("book", 1L, 13));
+ assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
@Test
@@ -76,11 +76,11 @@ public class DeduplicateKeepLastRowFunctionTest extends DeduplicateFunctionTestB
testHarness.close();
// Keep LastRow in deduplicate may send retraction
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 1L, 12));
- expectedOutputOutput.add(retractRecord("book", 1L, 12));
- expectedOutputOutput.add(record("book", 1L, 13));
- expectedOutputOutput.add(record("book", 2L, 11));
- assertor.assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 1L, 12));
+ expectedOutput.add(retractRecord("book", 1L, 12));
+ expectedOutput.add(record("book", 1L, 13));
+ expectedOutput.add(record("book", 2L, 11));
+ assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/MiniBatchDeduplicateKeepFirstRowFunctionTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/MiniBatchDeduplicateKeepFirstRowFunctionTest.java
index c7e4be1..3d79fff 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/MiniBatchDeduplicateKeepFirstRowFunctionTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/MiniBatchDeduplicateKeepFirstRowFunctionTest.java
@@ -64,10 +64,10 @@ public class MiniBatchDeduplicateKeepFirstRowFunctionTest extends DeduplicateFun
testHarness.processElement(record("book", 1L, 13));
// Keep FirstRow in deduplicate will not send retraction
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 1L, 12));
- expectedOutputOutput.add(record("book", 2L, 11));
- assertor.assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 1L, 12));
+ expectedOutput.add(record("book", 2L, 11));
+ assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
testHarness.close();
}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/MiniBatchDeduplicateKeepLastRowFunctionTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/MiniBatchDeduplicateKeepLastRowFunctionTest.java
index f9865f4..1779786 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/MiniBatchDeduplicateKeepLastRowFunctionTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/deduplicate/MiniBatchDeduplicateKeepLastRowFunctionTest.java
@@ -67,20 +67,20 @@ public class MiniBatchDeduplicateKeepLastRowFunctionTest extends DeduplicateFunc
testHarness.processElement(record("book", 1L, 13));
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 2L, 11));
- expectedOutputOutput.add(record("book", 1L, 13));
- assertor.assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 2L, 11));
+ expectedOutput.add(record("book", 1L, 13));
+ assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("book", 1L, 12));
testHarness.processElement(record("book", 2L, 11));
testHarness.processElement(record("book", 3L, 11));
- expectedOutputOutput.add(record("book", 1L, 12));
- expectedOutputOutput.add(record("book", 2L, 11));
- expectedOutputOutput.add(record("book", 3L, 11));
+ expectedOutput.add(record("book", 1L, 12));
+ expectedOutput.add(record("book", 2L, 11));
+ expectedOutput.add(record("book", 3L, 11));
testHarness.close();
- assertor.assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
@Test
@@ -95,22 +95,22 @@ public class MiniBatchDeduplicateKeepLastRowFunctionTest extends DeduplicateFunc
testHarness.processElement(record("book", 1L, 13));
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 2L, 11));
- expectedOutputOutput.add(record("book", 1L, 13));
- assertor.assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 2L, 11));
+ expectedOutput.add(record("book", 1L, 13));
+ assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("book", 1L, 12));
testHarness.processElement(record("book", 2L, 11));
testHarness.processElement(record("book", 3L, 11));
// this will send retract message to downstream
- expectedOutputOutput.add(retractRecord("book", 1L, 13));
- expectedOutputOutput.add(record("book", 1L, 12));
- expectedOutputOutput.add(retractRecord("book", 2L, 11));
- expectedOutputOutput.add(record("book", 2L, 11));
- expectedOutputOutput.add(record("book", 3L, 11));
+ expectedOutput.add(retractRecord("book", 1L, 13));
+ expectedOutput.add(record("book", 1L, 12));
+ expectedOutput.add(retractRecord("book", 2L, 11));
+ expectedOutput.add(record("book", 2L, 11));
+ expectedOutput.add(record("book", 3L, 11));
testHarness.close();
- assertor.assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/Int2SortMergeJoinOperatorTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/Int2SortMergeJoinOperatorTest.java
index e46b346..a4d33aa 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/Int2SortMergeJoinOperatorTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/Int2SortMergeJoinOperatorTest.java
@@ -32,7 +32,6 @@ import org.apache.flink.table.generated.NormalizedKeyComputer;
import org.apache.flink.table.generated.Projection;
import org.apache.flink.table.generated.RecordComparator;
import org.apache.flink.table.runtime.join.Int2HashJoinOperatorTest.MyProjection;
-import org.apache.flink.table.runtime.join.SortMergeJoinOperator.SortMergeJoinType;
import org.apache.flink.table.runtime.sort.IntNormalizedKeyComputer;
import org.apache.flink.table.runtime.sort.IntRecordComparator;
import org.apache.flink.table.runtime.util.UniformBinaryRowGenerator;
@@ -97,7 +96,7 @@ public class Int2SortMergeJoinOperatorTest {
MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(numKeys, buildValsPerKey, false);
MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys, probeValsPerKey, true);
- buildJoin(buildInput, probeInput, SortMergeJoinType.INNER, numKeys * buildValsPerKey * probeValsPerKey,
+ buildJoin(buildInput, probeInput, FlinkJoinType.INNER, numKeys * buildValsPerKey * probeValsPerKey,
numKeys, 165);
}
@@ -112,7 +111,7 @@ public class Int2SortMergeJoinOperatorTest {
MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(numKeys1, buildValsPerKey, true);
MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys2, probeValsPerKey, true);
- buildJoin(buildInput, probeInput, SortMergeJoinType.LEFT, numKeys1 * buildValsPerKey * probeValsPerKey,
+ buildJoin(buildInput, probeInput, FlinkJoinType.LEFT, numKeys1 * buildValsPerKey * probeValsPerKey,
numKeys1, 165);
}
@@ -126,7 +125,7 @@ public class Int2SortMergeJoinOperatorTest {
MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(numKeys1, buildValsPerKey, true);
MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys2, probeValsPerKey, true);
- buildJoin(buildInput, probeInput, SortMergeJoinType.RIGHT, 280, numKeys2, -1);
+ buildJoin(buildInput, probeInput, FlinkJoinType.RIGHT, 280, numKeys2, -1);
}
@Test
@@ -139,7 +138,7 @@ public class Int2SortMergeJoinOperatorTest {
MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(numKeys1, buildValsPerKey, true);
MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys2, probeValsPerKey, true);
- buildJoin(buildInput, probeInput, SortMergeJoinType.FULL, 280, numKeys2, -1);
+ buildJoin(buildInput, probeInput, FlinkJoinType.FULL, 280, numKeys2, -1);
}
@Test
@@ -152,7 +151,7 @@ public class Int2SortMergeJoinOperatorTest {
MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(numKeys1, buildValsPerKey, true);
MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys2, probeValsPerKey, true);
- StreamOperator operator = newOperator(SortMergeJoinType.SEMI, false);
+ StreamOperator operator = newOperator(FlinkJoinType.SEMI, false);
joinAndAssert(operator, buildInput, probeInput, 90, 9, 45, true, false);
}
@@ -166,14 +165,14 @@ public class Int2SortMergeJoinOperatorTest {
MutableObjectIterator<BinaryRow> buildInput = new UniformBinaryRowGenerator(numKeys1, buildValsPerKey, true);
MutableObjectIterator<BinaryRow> probeInput = new UniformBinaryRowGenerator(numKeys2, probeValsPerKey, true);
- StreamOperator operator = newOperator(SortMergeJoinType.ANTI, false);
+ StreamOperator operator = newOperator(FlinkJoinType.ANTI, false);
joinAndAssert(operator, buildInput, probeInput, 10, 1, 45, true, false);
}
private void buildJoin(
MutableObjectIterator<BinaryRow> input1,
MutableObjectIterator<BinaryRow> input2,
- SortMergeJoinType type,
+ FlinkJoinType type,
int expertOutSize, int expertOutKeySize, int expertOutVal) throws Exception {
joinAndAssert(
@@ -181,11 +180,11 @@ public class Int2SortMergeJoinOperatorTest {
input1, input2, expertOutSize, expertOutKeySize, expertOutVal, false, false);
}
- private StreamOperator getOperator(SortMergeJoinType type) {
+ private StreamOperator getOperator(FlinkJoinType type) {
return newOperator(type, leftIsSmaller);
}
- static StreamOperator newOperator(SortMergeJoinType type, boolean leftIsSmaller) {
+ static StreamOperator newOperator(FlinkJoinType type, boolean leftIsSmaller) {
return new SortMergeJoinOperator(
32 * 32 * 1024, 1024 * 1024, type, leftIsSmaller,
new GeneratedJoinCondition("", "", new Object[0]) {
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/ProcTimeBoundedStreamJoinTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/ProcTimeBoundedStreamJoinTest.java
new file mode 100644
index 0000000..e854be2
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/ProcTimeBoundedStreamJoinTest.java
@@ -0,0 +1,178 @@
+/*
+ * 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.runtime.join;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator;
+import org.apache.flink.streaming.util.KeyedTwoInputStreamOperatorTestHarness;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.runtime.util.BinaryRowKeySelector;
+import org.apache.flink.table.typeutils.BaseRowTypeInfo;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.runtime.util.StreamRecordUtils.record;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for {@link ProcTimeBoundedStreamJoin}.
+ */
+public class ProcTimeBoundedStreamJoinTest extends TimeBoundedStreamJoinTestBase {
+
+ private int keyIdx = 0;
+ private BinaryRowKeySelector keySelector = new BinaryRowKeySelector(new int[] { keyIdx },
+ rowType.getInternalTypes());
+ private TypeInformation<BaseRow> keyType = new BaseRowTypeInfo();
+
+
+ /** a.proctime >= b.proctime - 10 and a.proctime <= b.proctime + 20. **/
+ @Test
+ public void testProcTimeInnerJoinWithCommonBounds() throws Exception {
+ ProcTimeBoundedStreamJoin joinProcessFunc = new ProcTimeBoundedStreamJoin(
+ FlinkJoinType.INNER, -10, 20, rowType, rowType, generatedFunction);
+ KeyedTwoInputStreamOperatorTestHarness<BaseRow, BaseRow, BaseRow, BaseRow> testHarness = createTestHarness(
+ joinProcessFunc);
+ testHarness.open();
+ testHarness.setProcessingTime(1);
+ testHarness.processElement1(record(1L, "1a1"));
+ assertEquals(1, testHarness.numProcessingTimeTimers());
+
+ testHarness.setProcessingTime(2);
+ testHarness.processElement1(record(2L, "2a2"));
+ // timers for key = 1 and key = 2
+ assertEquals(2, testHarness.numProcessingTimeTimers());
+
+ testHarness.setProcessingTime(3);
+ testHarness.processElement1(record(1L, "1a3"));
+ assertEquals(4, testHarness.numKeyedStateEntries());
+ // The number of timers won't increase.
+ assertEquals(2, testHarness.numProcessingTimeTimers());
+
+ testHarness.processElement2(record(1L, "1b3"));
+
+ testHarness.setProcessingTime(4);
+ testHarness.processElement2(record(2L, "2b4"));
+ // The number of states should be doubled.
+ assertEquals(8, testHarness.numKeyedStateEntries());
+ assertEquals(4, testHarness.numProcessingTimeTimers());
+
+ // Test for -10 boundary (13 - 10 = 3).
+ // The left row (key = 1) with timestamp = 1 will be eagerly removed here.
+ testHarness.setProcessingTime(13);
+ testHarness.processElement2(record(1L, "1b13"));
+
+ // Test for +20 boundary (13 + 20 = 33).
+ testHarness.setProcessingTime(33);
+ assertEquals(4, testHarness.numKeyedStateEntries());
+ assertEquals(2, testHarness.numProcessingTimeTimers());
+
+ testHarness.processElement1(record(1L, "1a33"));
+ testHarness.processElement1(record(2L, "2a33"));
+ // The left row (key = 2) with timestamp = 2 will be eagerly removed here.
+ testHarness.processElement2(record(2L, "2b33"));
+
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record(1L, "1a1", 1L, "1b3"));
+ expectedOutput.add(record(1L, "1a3", 1L, "1b3"));
+ expectedOutput.add(record(2L, "2a2", 2L, "2b4"));
+ expectedOutput.add(record(1L, "1a3", 1L, "1b13"));
+ expectedOutput.add(record(1L, "1a33", 1L, "1b13"));
+ expectedOutput.add(record(2L, "2a33", 2L, "2b33"));
+
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
+ testHarness.close();
+ }
+
+ /** a.proctime >= b.proctime - 10 and a.proctime <= b.proctime - 5. **/
+ @Test
+ public void testProcTimeInnerJoinWithNegativeBounds() throws Exception {
+ ProcTimeBoundedStreamJoin joinProcessFunc = new ProcTimeBoundedStreamJoin(
+ FlinkJoinType.INNER, -10, -5, rowType, rowType, generatedFunction);
+
+ KeyedTwoInputStreamOperatorTestHarness<BaseRow, BaseRow, BaseRow, BaseRow> testHarness = createTestHarness(
+ joinProcessFunc);
+ testHarness.open();
+
+ testHarness.setProcessingTime(1);
+ testHarness.processElement1(record(1L, "1a1"));
+
+ testHarness.setProcessingTime(2);
+ testHarness.processElement1(record(2L, "2a2"));
+
+ testHarness.setProcessingTime(3);
+ testHarness.processElement1(record(1L, "1a3"));
+ assertEquals(4, testHarness.numKeyedStateEntries());
+ assertEquals(2, testHarness.numProcessingTimeTimers());
+
+ // All the right rows will not be cached.
+ testHarness.processElement2(record(1L, "1b3"));
+ assertEquals(4, testHarness.numKeyedStateEntries());
+ assertEquals(2, testHarness.numProcessingTimeTimers());
+
+ testHarness.setProcessingTime(7);
+
+ // Meets a.proctime <= b.proctime - 5.
+ // This row will only be joined without being cached (7 >= 7 - 5).
+ testHarness.processElement2(record(2L, "2b7"));
+ assertEquals(4, testHarness.numKeyedStateEntries());
+ assertEquals(2, testHarness.numProcessingTimeTimers());
+
+ testHarness.setProcessingTime(12);
+ // The left row (key = 1) with timestamp = 1 will be eagerly removed here.
+ testHarness.processElement2(record(1L, "1b12"));
+
+ // We add a delay (relativeWindowSize / 2) for cleaning up state.
+ // No timers will be triggered here.
+ testHarness.setProcessingTime(13);
+ assertEquals(4, testHarness.numKeyedStateEntries());
+ assertEquals(2, testHarness.numProcessingTimeTimers());
+
+ // Trigger the timer registered by the left row (key = 1) with timestamp = 1
+ // (1 + 10 + 2 + 0 + 1 = 14).
+ // The left row (key = 1) with timestamp = 3 will removed here.
+ testHarness.setProcessingTime(14);
+ assertEquals(2, testHarness.numKeyedStateEntries());
+ assertEquals(1, testHarness.numProcessingTimeTimers());
+
+ // Clean up the left row (key = 2) with timestamp = 2.
+ testHarness.setProcessingTime(16);
+ assertEquals(0, testHarness.numKeyedStateEntries());
+ assertEquals(0, testHarness.numProcessingTimeTimers());
+
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record(2L, "2a2", 2L, "2b7"));
+ expectedOutput.add(record(1L, "1a3", 1L, "1b12"));
+
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
+ testHarness.close();
+ }
+
+ private KeyedTwoInputStreamOperatorTestHarness<BaseRow, BaseRow, BaseRow, BaseRow> createTestHarness(
+ ProcTimeBoundedStreamJoin windowJoinFunc)
+ throws Exception {
+ KeyedCoProcessOperator<BaseRow, BaseRow, BaseRow, BaseRow> operator = new KeyedCoProcessOperator<>(
+ windowJoinFunc);
+ KeyedTwoInputStreamOperatorTestHarness<BaseRow, BaseRow, BaseRow, BaseRow> testHarness =
+ new KeyedTwoInputStreamOperatorTestHarness<>(operator, keySelector, keySelector, keyType);
+ return testHarness;
+ }
+}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/RandomSortMergeInnerJoinTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/RandomSortMergeInnerJoinTest.java
index c5b7794..01e0fd5 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/RandomSortMergeInnerJoinTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/RandomSortMergeInnerJoinTest.java
@@ -41,7 +41,6 @@ import org.apache.flink.table.dataformat.BinaryRow;
import org.apache.flink.table.dataformat.BinaryRowWriter;
import org.apache.flink.table.dataformat.BinaryString;
import org.apache.flink.table.dataformat.JoinedRow;
-import org.apache.flink.table.runtime.join.SortMergeJoinOperator.SortMergeJoinType;
import org.apache.flink.table.type.InternalTypes;
import org.apache.flink.table.typeutils.BaseRowTypeInfo;
import org.apache.flink.util.MutableObjectIterator;
@@ -350,7 +349,7 @@ public class RandomSortMergeInnerJoinTest {
}
private StreamOperator getOperator() {
- return Int2SortMergeJoinOperatorTest.newOperator(SortMergeJoinType.INNER, leftIsSmall);
+ return Int2SortMergeJoinOperatorTest.newOperator(FlinkJoinType.INNER, leftIsSmall);
}
public static LinkedBlockingQueue<Object> transformToBinary(LinkedBlockingQueue<Object> output) {
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/RandomSortMergeOuterJoinTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/RandomSortMergeOuterJoinTest.java
index b4761a4..46c3ce8 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/RandomSortMergeOuterJoinTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/RandomSortMergeOuterJoinTest.java
@@ -32,7 +32,6 @@ import org.apache.flink.runtime.operators.testutils.TestData.TupleGenerator.KeyM
import org.apache.flink.runtime.operators.testutils.TestData.TupleGenerator.ValueMode;
import org.apache.flink.runtime.operators.testutils.TestData.TupleGeneratorIterator;
import org.apache.flink.streaming.api.operators.StreamOperator;
-import org.apache.flink.table.runtime.join.SortMergeJoinOperator.SortMergeJoinType;
import org.apache.flink.util.MutableObjectIterator;
import org.junit.Assert;
@@ -58,22 +57,22 @@ public class RandomSortMergeOuterJoinTest {
@Test
public void testFullOuterJoinWithHighNumberOfCommonKeys() {
- testOuterJoinWithHighNumberOfCommonKeys(SortMergeJoinType.FULL, 200, 500, 2048, 0.02f, 200, 500, 2048, 0.02f);
+ testOuterJoinWithHighNumberOfCommonKeys(FlinkJoinType.FULL, 200, 500, 2048, 0.02f, 200, 500, 2048, 0.02f);
}
@Test
public void testLeftOuterJoinWithHighNumberOfCommonKeys() {
- testOuterJoinWithHighNumberOfCommonKeys(SortMergeJoinType.LEFT, 200, 10, 4096, 0.02f, 100, 4000, 2048, 0.02f);
+ testOuterJoinWithHighNumberOfCommonKeys(FlinkJoinType.LEFT, 200, 10, 4096, 0.02f, 100, 4000, 2048, 0.02f);
}
@Test
public void testRightOuterJoinWithHighNumberOfCommonKeys() {
- testOuterJoinWithHighNumberOfCommonKeys(SortMergeJoinType.RIGHT, 100, 10, 2048, 0.02f, 200, 4000, 4096, 0.02f);
+ testOuterJoinWithHighNumberOfCommonKeys(FlinkJoinType.RIGHT, 100, 10, 2048, 0.02f, 200, 4000, 4096, 0.02f);
}
@SuppressWarnings("unchecked, rawtypes")
protected void testOuterJoinWithHighNumberOfCommonKeys(
- SortMergeJoinType outerJoinType, int input1Size, int input1Duplicates, int input1ValueLength,
+ FlinkJoinType outerJoinType, int input1Size, int input1Duplicates, int input1ValueLength,
float input1KeyDensity, int input2Size, int input2Duplicates, int input2ValueLength,
float input2KeyDensity) {
TypeComparator<Tuple2<Integer, String>> comparator1 = new TupleComparator<>(
@@ -167,14 +166,14 @@ public class RandomSortMergeOuterJoinTest {
private Map<Integer, Collection<Match>> joinValues(
Map<Integer, Collection<String>> leftMap,
Map<Integer, Collection<String>> rightMap,
- SortMergeJoinType outerJoinType) {
+ FlinkJoinType outerJoinType) {
Map<Integer, Collection<Match>> map = new HashMap<>();
for (Integer key : leftMap.keySet()) {
Collection<String> leftValues = leftMap.get(key);
Collection<String> rightValues = rightMap.get(key);
- if (outerJoinType == SortMergeJoinType.RIGHT && rightValues == null) {
+ if (outerJoinType == FlinkJoinType.RIGHT && rightValues == null) {
continue;
}
@@ -195,7 +194,7 @@ public class RandomSortMergeOuterJoinTest {
}
}
- if (outerJoinType == SortMergeJoinType.RIGHT || outerJoinType == SortMergeJoinType.FULL) {
+ if (outerJoinType == FlinkJoinType.RIGHT || outerJoinType == FlinkJoinType.FULL) {
for (Integer key : rightMap.keySet()) {
Collection<String> leftValues = leftMap.get(key);
Collection<String> rightValues = rightMap.get(key);
@@ -219,7 +218,7 @@ public class RandomSortMergeOuterJoinTest {
return map;
}
- protected StreamOperator getOperator(SortMergeJoinType outerJoinType) {
+ protected StreamOperator getOperator(FlinkJoinType outerJoinType) {
return Int2SortMergeJoinOperatorTest.newOperator(outerJoinType, false);
}
}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/RowTimeBoundedStreamJoinTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/RowTimeBoundedStreamJoinTest.java
new file mode 100644
index 0000000..ef248e9
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/RowTimeBoundedStreamJoinTest.java
@@ -0,0 +1,387 @@
+/*
+ * 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.runtime.join;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.util.KeyedTwoInputStreamOperatorTestHarness;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.runtime.util.BinaryRowKeySelector;
+import org.apache.flink.table.typeutils.BaseRowTypeInfo;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.table.runtime.util.StreamRecordUtils.record;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Test for {@link RowTimeBoundedStreamJoin}.
+ */
+public class RowTimeBoundedStreamJoinTest extends TimeBoundedStreamJoinTestBase {
+
+ private int keyIdx = 1;
+ private BinaryRowKeySelector keySelector = new BinaryRowKeySelector(new int[] { keyIdx },
+ rowType.getInternalTypes());
+ private TypeInformation<BaseRow> keyType = new BaseRowTypeInfo();
+
+ /** a.rowtime >= b.rowtime - 10 and a.rowtime <= b.rowtime + 20. **/
+ @Test
+ public void testRowTimeInnerJoinWithCommonBounds() throws Exception {
+ RowTimeBoundedStreamJoin joinProcessFunc = new RowTimeBoundedStreamJoin(
+ FlinkJoinType.INNER, -10, 20, 0, rowType, rowType, generatedFunction, 0, 0);
+
+ KeyedTwoInputStreamOperatorTestHarness<BaseRow, BaseRow, BaseRow, BaseRow> testHarness = createTestHarness(
+ joinProcessFunc);
+
+ testHarness.open();
+
+ testHarness.processWatermark1(new Watermark(1));
+ testHarness.processWatermark2(new Watermark(1));
+
+ // Test late data.
+ testHarness.processElement1(record(1L, "k1"));
+ // Though (1L, "k1") is actually late, it will also be cached.
+ assertEquals(1, testHarness.numEventTimeTimers());
+
+ testHarness.processElement1(record(2L, "k1"));
+ testHarness.processElement2(record(2L, "k1"));
+
+ assertEquals(2, testHarness.numEventTimeTimers());
+ assertEquals(4, testHarness.numKeyedStateEntries());
+
+ testHarness.processElement1(record(5L, "k1"));
+ testHarness.processElement2(record(15L, "k1"));
+ testHarness.processWatermark1(new Watermark(20));
+ testHarness.processWatermark2(new Watermark(20));
+ assertEquals(4, testHarness.numKeyedStateEntries());
+
+ testHarness.processElement1(record(35L, "k1"));
+
+ // The right rows with timestamp = 2 and 5 will be removed here.
+ // The left rows with timestamp = 2 and 15 will be removed here.
+ testHarness.processWatermark1(new Watermark(38));
+ testHarness.processWatermark2(new Watermark(38));
+
+ testHarness.processElement1(record(40L, "k2"));
+ testHarness.processElement2(record(39L, "k2"));
+ assertEquals(6, testHarness.numKeyedStateEntries());
+
+ // The right row with timestamp = 35 will be removed here.
+ testHarness.processWatermark1(new Watermark(61));
+ testHarness.processWatermark2(new Watermark(61));
+ assertEquals(4, testHarness.numKeyedStateEntries());
+
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(new Watermark(-19));
+ // This result is produced by the late row (1, "k1").
+ expectedOutput.add(record(1L, "k1", 2L, "k1"));
+ expectedOutput.add(record(2L, "k1", 2L, "k1"));
+ expectedOutput.add(record(5L, "k1", 2L, "k1"));
+ expectedOutput.add(record(5L, "k1", 15L, "k1"));
+ expectedOutput.add(new Watermark(0));
+ expectedOutput.add(record(35L, "k1", 15L, "k1"));
+ expectedOutput.add(new Watermark(18));
+ expectedOutput.add(record(40L, "k2", 39L, "k2"));
+ expectedOutput.add(new Watermark(41));
+
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
+ testHarness.close();
+ }
+
+ /** a.rowtime >= b.rowtime - 10 and a.rowtime <= b.rowtime - 7. **/
+ @Test
+ public void testRowTimeInnerJoinWithNegativeBounds() throws Exception {
+ RowTimeBoundedStreamJoin joinProcessFunc = new RowTimeBoundedStreamJoin(
+ FlinkJoinType.INNER, -10, -7, 0, rowType, rowType, generatedFunction, 0, 0);
+
+ KeyedTwoInputStreamOperatorTestHarness<BaseRow, BaseRow, BaseRow, BaseRow> testHarness = createTestHarness(
+ joinProcessFunc);
+
+ testHarness.open();
+
+ testHarness.processWatermark1(new Watermark(1));
+ testHarness.processWatermark2(new Watermark(1));
+
+ // This row will not be cached.
+ testHarness.processElement2(record(2L, "k1"));
+ assertEquals(0, testHarness.numKeyedStateEntries());
+
+ testHarness.processWatermark1(new Watermark(2));
+ testHarness.processWatermark2(new Watermark(2));
+ testHarness.processElement1(record(3L, "k1"));
+ testHarness.processElement2(record(3L, "k1"));
+
+ // Test for -10 boundary (13 - 10 = 3).
+ // This row from the right stream will be cached.
+ // The clean time for the left stream is 13 - 7 + 1 - 1 = 8
+ testHarness.processElement2(record(13L, "k1"));
+
+ // Test for -7 boundary (13 - 7 = 6).
+ testHarness.processElement1(record(6L, "k1"));
+ assertEquals(4, testHarness.numKeyedStateEntries());
+
+ // Trigger the left timer with timestamp 8.
+ // The row with timestamp = 13 will be removed here (13 < 10 + 7).
+ testHarness.processWatermark1(new Watermark(10));
+ testHarness.processWatermark2(new Watermark(10));
+ assertEquals(2, testHarness.numKeyedStateEntries());
+
+ // Clear the states.
+ testHarness.processWatermark1(new Watermark(18));
+ testHarness.processWatermark2(new Watermark(18));
+ assertEquals(0, testHarness.numKeyedStateEntries());
+
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(new Watermark(-9));
+ expectedOutput.add(new Watermark(-8));
+ expectedOutput.add(record(3L, "k1", 13L, "k1"));
+ expectedOutput.add(record(6L, "k1", 13L, "k1"));
+ expectedOutput.add(new Watermark(0));
+ expectedOutput.add(new Watermark(8));
+
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
+ testHarness.close();
+ }
+
+ @Test
+ public void testRowTimeLeftOuterJoin() throws Exception {
+ RowTimeBoundedStreamJoin joinProcessFunc = new RowTimeBoundedStreamJoin(
+ FlinkJoinType.LEFT, -5, 9, 0, rowType, rowType, generatedFunction, 0, 0);
+
+ KeyedTwoInputStreamOperatorTestHarness<BaseRow, BaseRow, BaseRow, BaseRow> testHarness = createTestHarness(
+ joinProcessFunc);
+
+ testHarness.open();
+
+ testHarness.processElement1(record(1L, "k1"));
+ testHarness.processElement2(record(1L, "k2"));
+ assertEquals(2, testHarness.numEventTimeTimers());
+ assertEquals(4, testHarness.numKeyedStateEntries());
+
+ // The left row with timestamp = 1 will be padded and removed (14=1+5+1+((5+9)/2)).
+ testHarness.processWatermark1(new Watermark(14));
+ testHarness.processWatermark2(new Watermark(14));
+ assertEquals(1, testHarness.numEventTimeTimers());
+ assertEquals(2, testHarness.numKeyedStateEntries());
+
+ // The right row with timestamp = 1 will be removed (18=1+9+1+((5+9)/2)).
+ testHarness.processWatermark1(new Watermark(18));
+ testHarness.processWatermark2(new Watermark(18));
+ assertEquals(0, testHarness.numEventTimeTimers());
+ assertEquals(0, testHarness.numKeyedStateEntries());
+
+ testHarness.processElement1(record(2L, "k1"));
+ testHarness.processElement2(record(2L, "k2"));
+ // The late rows with timestamp = 2 will not be cached, but a null padding result for the left
+ // row will be emitted.
+ assertEquals(0, testHarness.numKeyedStateEntries());
+ assertEquals(0, testHarness.numEventTimeTimers());
+
+ // Make sure the common (inner) join can be performed.
+ testHarness.processElement1(record(19L, "k1"));
+ testHarness.processElement1(record(20L, "k1"));
+ testHarness.processElement2(record(26L, "k1"));
+ testHarness.processElement2(record(25L, "k1"));
+ testHarness.processElement1(record(21L, "k1"));
+ testHarness.processElement2(record(39L, "k2"));
+ testHarness.processElement2(record(40L, "k2"));
+ testHarness.processElement1(record(50L, "k2"));
+ testHarness.processElement1(record(49L, "k2"));
+ testHarness.processElement2(record(41L, "k2"));
+ testHarness.processWatermark1(new Watermark(100));
+ testHarness.processWatermark2(new Watermark(100));
+
+ List<Object> expectedOutput = new ArrayList<>();
+ // The timestamp 14 is set with the triggered timer.
+ expectedOutput.add(record(1L, "k1", null, null));
+ expectedOutput.add(new Watermark(5));
+ expectedOutput.add(new Watermark(9));
+ expectedOutput.add(record(2L, "k1", null, null));
+ expectedOutput.add(record(20L, "k1", 25L, "k1"));
+ expectedOutput.add(record(21L, "k1", 25L, "k1"));
+ expectedOutput.add(record(21L, "k1", 26L, "k1"));
+ expectedOutput.add(record(49L, "k2", 40L, "k2"));
+ expectedOutput.add(record(49L, "k2", 41L, "k2"));
+ expectedOutput.add(record(50L, "k2", 41L, "k2"));
+ // The timestamp 32 is set with the triggered timer.
+ expectedOutput.add(record(19L, "k1", null, null));
+ expectedOutput.add(new Watermark(91));
+
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
+ testHarness.close();
+ }
+
+ @Test
+ public void testRowTimeRightOuterJoin() throws Exception {
+ RowTimeBoundedStreamJoin joinProcessFunc = new RowTimeBoundedStreamJoin(
+ FlinkJoinType.RIGHT, -5, 9, 0, rowType, rowType, generatedFunction, 0, 0);
+
+ KeyedTwoInputStreamOperatorTestHarness<BaseRow, BaseRow, BaseRow, BaseRow> testHarness = createTestHarness(
+ joinProcessFunc);
+
+ testHarness.open();
+
+ testHarness.processElement1(record(1L, "k1"));
+ testHarness.processElement2(record(1L, "k2"));
+ assertEquals(2, testHarness.numEventTimeTimers());
+ assertEquals(4, testHarness.numKeyedStateEntries());
+
+ // The left row with timestamp = 1 will be removed (14=1+5+1+((5+9)/2)).
+ testHarness.processWatermark1(new Watermark(14));
+ testHarness.processWatermark2(new Watermark(14));
+ assertEquals(1, testHarness.numEventTimeTimers());
+ assertEquals(2, testHarness.numKeyedStateEntries());
+
+ // The right row with timestamp = 1 will be padded and removed (18=1+9+1+((5+9)/2)).
+ testHarness.processWatermark1(new Watermark(18));
+ testHarness.processWatermark2(new Watermark(18));
+ assertEquals(0, testHarness.numEventTimeTimers());
+ assertEquals(0, testHarness.numKeyedStateEntries());
+
+ testHarness.processElement1(record(2L, "k1"));
+ testHarness.processElement2(record(2L, "k2"));
+ // The late rows with timestamp = 2 will not be cached, but a null padding result for the right
+ // row will be emitted.
+ assertEquals(0, testHarness.numKeyedStateEntries());
+ assertEquals(0, testHarness.numEventTimeTimers());
+
+ // Make sure the common (inner) join can be performed.
+ testHarness.processElement1(record(19L, "k1"));
+ testHarness.processElement1(record(20L, "k1"));
+ testHarness.processElement2(record(26L, "k1"));
+ testHarness.processElement2(record(25L, "k1"));
+ testHarness.processElement1(record(21L, "k1"));
+ testHarness.processElement2(record(39L, "k2"));
+ testHarness.processElement2(record(40L, "k2"));
+ testHarness.processElement1(record(50L, "k2"));
+ testHarness.processElement1(record(49L, "k2"));
+ testHarness.processElement2(record(41L, "k2"));
+ testHarness.processWatermark1(new Watermark(100));
+ testHarness.processWatermark2(new Watermark(100));
+
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(new Watermark(5));
+ // The timestamp 18 is set with the triggered timer.
+ expectedOutput.add(record(null, null, 1L, "k2"));
+ expectedOutput.add(new Watermark(9));
+ expectedOutput.add(record(null, null, 2L, "k2"));
+ expectedOutput.add(record(20L, "k1", 25L, "k1"));
+ expectedOutput.add(record(21L, "k1", 25L, "k1"));
+ expectedOutput.add(record(21L, "k1", 26L, "k1"));
+ expectedOutput.add(record(49L, "k2", 40L, "k2"));
+ expectedOutput.add(record(49L, "k2", 41L, "k2"));
+ expectedOutput.add(record(50L, "k2", 41L, "k2"));
+ // The timestamp 56 is set with the triggered timer.
+ expectedOutput.add(record(null, null, 39L, "k2"));
+ expectedOutput.add(new Watermark(91));
+
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
+ testHarness.close();
+ }
+
+ /** a.rowtime >= b.rowtime - 5 and a.rowtime <= b.rowtime + 9. **/
+ @Test
+ public void testRowTimeFullOuterJoin() throws Exception {
+ RowTimeBoundedStreamJoin joinProcessFunc = new RowTimeBoundedStreamJoin(
+ FlinkJoinType.FULL, -5, 9, 0, rowType, rowType, generatedFunction, 0, 0);
+
+ KeyedTwoInputStreamOperatorTestHarness<BaseRow, BaseRow, BaseRow, BaseRow> testHarness = createTestHarness(
+ joinProcessFunc);
+
+ testHarness.open();
+
+ testHarness.processElement1(record(1L, "k1"));
+ testHarness.processElement2(record(1L, "k2"));
+ assertEquals(2, testHarness.numEventTimeTimers());
+ assertEquals(4, testHarness.numKeyedStateEntries());
+
+ // The left row with timestamp = 1 will be padded and removed (14=1+5+1+((5+9)/2)).
+ testHarness.processWatermark1(new Watermark(14));
+ testHarness.processWatermark2(new Watermark(14));
+ assertEquals(1, testHarness.numEventTimeTimers());
+ assertEquals(2, testHarness.numKeyedStateEntries());
+
+ // The right row with timestamp = 1 will be padded and removed (18=1+9+1+((5+9)/2)).
+ testHarness.processWatermark1(new Watermark(18));
+ testHarness.processWatermark2(new Watermark(18));
+ assertEquals(0, testHarness.numEventTimeTimers());
+ assertEquals(0, testHarness.numKeyedStateEntries());
+
+ testHarness.processElement1(record(2L, "k1"));
+ testHarness.processElement2(record(2L, "k2"));
+ // The late rows with timestamp = 2 will not be cached, but a null padding result for the right
+ // row will be emitted.
+ assertEquals(0, testHarness.numKeyedStateEntries());
+ assertEquals(0, testHarness.numEventTimeTimers());
+
+ // Make sure the common (inner) join can be performed.
+ testHarness.processElement1(record(19L, "k1"));
+ testHarness.processElement1(record(20L, "k1"));
+ testHarness.processElement2(record(26L, "k1"));
+ testHarness.processElement2(record(25L, "k1"));
+ testHarness.processElement1(record(21L, "k1"));
+
+ testHarness.processElement2(record(39L, "k2"));
+ testHarness.processElement2(record(40L, "k2"));
+ testHarness.processElement1(record(50L, "k2"));
+ testHarness.processElement1(record(49L, "k2"));
+ testHarness.processElement2(record(41L, "k2"));
+ testHarness.processWatermark1(new Watermark(100));
+ testHarness.processWatermark2(new Watermark(100));
+
+ List<Object> expectedOutput = new ArrayList<>();
+ // The timestamp 14 is set with the triggered timer.
+ expectedOutput.add(record(1L, "k1", null, null));
+ expectedOutput.add(new Watermark(5));
+ // The timestamp 18 is set with the triggered timer.
+ expectedOutput.add(record(null, null, 1L, "k2"));
+ expectedOutput.add(new Watermark(9));
+ expectedOutput.add(record(2L, "k1", null, null));
+ expectedOutput.add(record(null, null, 2L, "k2"));
+ expectedOutput.add(record(20L, "k1", 25L, "k1"));
+ expectedOutput.add(record(21L, "k1", 25L, "k1"));
+ expectedOutput.add(record(21L, "k1", 26L, "k1"));
+ expectedOutput.add(record(49L, "k2", 40L, "k2"));
+ expectedOutput.add(record(49L, "k2", 41L, "k2"));
+ expectedOutput.add(record(50L, "k2", 41L, "k2"));
+ // The timestamp 32 is set with the triggered timer.
+ expectedOutput.add(record(19L, "k1", null, null));
+ // The timestamp 56 is set with the triggered timer.
+ expectedOutput.add(record(null, null, 39L, "k2"));
+ expectedOutput.add(new Watermark(91));
+
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
+ testHarness.close();
+ }
+
+ private KeyedTwoInputStreamOperatorTestHarness<BaseRow, BaseRow, BaseRow, BaseRow> createTestHarness(
+ RowTimeBoundedStreamJoin windowJoinFunc)
+ throws Exception {
+ KeyedCoProcessOperator<BaseRow, BaseRow, BaseRow, BaseRow> operator = new KeyedCoProcessOperatorWithWatermarkDelay<>(
+ windowJoinFunc, windowJoinFunc.getMaxOutputDelay());
+ KeyedTwoInputStreamOperatorTestHarness<BaseRow, BaseRow, BaseRow, BaseRow> testHarness =
+ new KeyedTwoInputStreamOperatorTestHarness<>(operator, keySelector, keySelector, keyType);
+ return testHarness;
+ }
+
+}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/String2SortMergeJoinOperatorTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/String2SortMergeJoinOperatorTest.java
index ccc2f98..f861a9f0 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/String2SortMergeJoinOperatorTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/String2SortMergeJoinOperatorTest.java
@@ -34,7 +34,6 @@ import org.apache.flink.table.generated.JoinCondition;
import org.apache.flink.table.generated.NormalizedKeyComputer;
import org.apache.flink.table.generated.Projection;
import org.apache.flink.table.generated.RecordComparator;
-import org.apache.flink.table.runtime.join.SortMergeJoinOperator.SortMergeJoinType;
import org.apache.flink.table.runtime.join.String2HashJoinOperatorTest.MyProjection;
import org.apache.flink.table.runtime.sort.StringNormalizedKeyComputer;
import org.apache.flink.table.runtime.sort.StringRecordComparator;
@@ -75,7 +74,7 @@ public class String2SortMergeJoinOperatorTest {
@Test
public void testInnerJoin() throws Exception {
- StreamOperator joinOperator = newOperator(SortMergeJoinType.INNER, leftIsSmall);
+ StreamOperator joinOperator = newOperator(FlinkJoinType.INNER, leftIsSmall);
TwoInputStreamTaskTestHarness<BinaryRow, BinaryRow, JoinedRow> testHarness =
buildSortMergeJoin(joinOperator);
@@ -90,7 +89,7 @@ public class String2SortMergeJoinOperatorTest {
@Test
public void testLeftOuterJoin() throws Exception {
- StreamOperator joinOperator = newOperator(SortMergeJoinType.LEFT, leftIsSmall);
+ StreamOperator joinOperator = newOperator(FlinkJoinType.LEFT, leftIsSmall);
TwoInputStreamTaskTestHarness<BinaryRow, BinaryRow, JoinedRow> testHarness =
buildSortMergeJoin(joinOperator);
@@ -106,7 +105,7 @@ public class String2SortMergeJoinOperatorTest {
@Test
public void testRightOuterJoin() throws Exception {
- StreamOperator joinOperator = newOperator(SortMergeJoinType.RIGHT, leftIsSmall);
+ StreamOperator joinOperator = newOperator(FlinkJoinType.RIGHT, leftIsSmall);
TwoInputStreamTaskTestHarness<BinaryRow, BinaryRow, JoinedRow> testHarness =
buildSortMergeJoin(joinOperator);
@@ -122,7 +121,7 @@ public class String2SortMergeJoinOperatorTest {
@Test
public void testFullJoin() throws Exception {
- StreamOperator joinOperator = newOperator(SortMergeJoinType.FULL, leftIsSmall);
+ StreamOperator joinOperator = newOperator(FlinkJoinType.FULL, leftIsSmall);
TwoInputStreamTaskTestHarness<BinaryRow, BinaryRow, JoinedRow> testHarness =
buildSortMergeJoin(joinOperator);
@@ -165,7 +164,7 @@ public class String2SortMergeJoinOperatorTest {
return testHarness;
}
- static StreamOperator newOperator(SortMergeJoinType type, boolean leftIsSmaller) {
+ static StreamOperator newOperator(FlinkJoinType type, boolean leftIsSmaller) {
return new SortMergeJoinOperator(
32 * 32 * 1024, 1024 * 1024, type, leftIsSmaller,
new GeneratedJoinCondition("", "", new Object[0]) {
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/TimeBoundedStreamJoinTestBase.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/TimeBoundedStreamJoinTestBase.java
new file mode 100644
index 0000000..6c243ac
--- /dev/null
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/join/TimeBoundedStreamJoinTestBase.java
@@ -0,0 +1,63 @@
+/*
+ * 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.runtime.join;
+
+import org.apache.flink.api.common.functions.FlatJoinFunction;
+import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.generated.GeneratedFunction;
+import org.apache.flink.table.runtime.util.BaseRowHarnessAssertor;
+import org.apache.flink.table.type.InternalTypes;
+import org.apache.flink.table.typeutils.BaseRowTypeInfo;
+
+/**
+ * Base Test for all subclass of {@link TimeBoundedStreamJoin}.
+ */
+abstract class TimeBoundedStreamJoinTestBase {
+ BaseRowTypeInfo rowType = new BaseRowTypeInfo(InternalTypes.LONG, InternalTypes.STRING);
+
+ private BaseRowTypeInfo outputRowType = new BaseRowTypeInfo(InternalTypes.LONG, InternalTypes.STRING,
+ InternalTypes.LONG, InternalTypes.STRING);
+ BaseRowHarnessAssertor assertor = new BaseRowHarnessAssertor(outputRowType.getFieldTypes());
+
+ private String funcCode =
+ "public class WindowJoinFunction\n" +
+ " extends org.apache.flink.api.common.functions.RichFlatJoinFunction {\n" +
+ " final org.apache.flink.table.dataformat.JoinedRow joinedRow = new org.apache.flink.table.dataformat.JoinedRow();\n" +
+
+ " public WindowJoinFunction(Object[] references) throws Exception {}\n" +
+
+ " @Override\n" +
+ " public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {}\n" +
+
+ " @Override\n" +
+ " public void join(Object _in1, Object _in2, org.apache.flink.util.Collector c) throws Exception {\n" +
+ " org.apache.flink.table.dataformat.BaseRow in1 = (org.apache.flink.table.dataformat.BaseRow) _in1;\n" +
+ " org.apache.flink.table.dataformat.BaseRow in2 = (org.apache.flink.table.dataformat.BaseRow) _in2;\n" +
+ " joinedRow.replace(in1,in2);\n" +
+ " c.collect(joinedRow);\n" +
+ " }\n" +
+
+ " @Override\n" +
+ " public void close() throws Exception {}\n" +
+ "}\n";
+
+ GeneratedFunction<FlatJoinFunction<BaseRow, BaseRow, BaseRow>> generatedFunction = new GeneratedFunction(
+ "WindowJoinFunction", funcCode, new Object[0]);
+
+}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/AppendOnlyTopNFunctionTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/AppendOnlyTopNFunctionTest.java
index d559830..7aa5694 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/AppendOnlyTopNFunctionTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/AppendOnlyTopNFunctionTest.java
@@ -55,16 +55,16 @@ public class AppendOnlyTopNFunctionTest extends TopNFunctionTestBase {
testHarness.processElement(record("fruit", 1L, 22));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 2L, 12));
- expectedOutputOutput.add(record("book", 2L, 19));
- expectedOutputOutput.add(retractRecord("book", 2L, 19));
- expectedOutputOutput.add(record("book", 2L, 11));
- expectedOutputOutput.add(record("fruit", 1L, 33));
- expectedOutputOutput.add(retractRecord("fruit", 1L, 33));
- expectedOutputOutput.add(record("fruit", 1L, 22));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 2L, 12));
+ expectedOutput.add(record("book", 2L, 19));
+ expectedOutput.add(retractRecord("book", 2L, 19));
+ expectedOutput.add(record("book", 2L, 11));
+ expectedOutput.add(record("fruit", 1L, 33));
+ expectedOutput.add(retractRecord("fruit", 1L, 33));
+ expectedOutput.add(record("fruit", 1L, 22));
assertorWithoutRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/RetractableTopNFunctionTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/RetractableTopNFunctionTest.java
index ac30d93..96a2904 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/RetractableTopNFunctionTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/RetractableTopNFunctionTest.java
@@ -60,19 +60,19 @@ public class RetractableTopNFunctionTest extends TopNFunctionTestBase {
testHarness.processElement(record("fruit", 5L, 22));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 1L, 12, 1L));
- expectedOutputOutput.add(record("book", 2L, 19, 2L));
- expectedOutputOutput.add(record("book", 4L, 11, 1L));
- expectedOutputOutput.add(record("book", 1L, 12, 2L));
- expectedOutputOutput.add(deleteRecord("book", 1L, 12, 2L));
- expectedOutputOutput.add(record("book", 2L, 19, 2L));
- expectedOutputOutput.add(record("book", 5L, 11, 2L));
- expectedOutputOutput.add(record("fruit", 4L, 33, 1L));
- expectedOutputOutput.add(record("fruit", 3L, 44, 2L));
- expectedOutputOutput.add(record("fruit", 5L, 22, 1L));
- expectedOutputOutput.add(record("fruit", 4L, 33, 2L));
- assertorWithRowNumber.assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 1L, 12, 1L));
+ expectedOutput.add(record("book", 2L, 19, 2L));
+ expectedOutput.add(record("book", 4L, 11, 1L));
+ expectedOutput.add(record("book", 1L, 12, 2L));
+ expectedOutput.add(deleteRecord("book", 1L, 12, 2L));
+ expectedOutput.add(record("book", 2L, 19, 2L));
+ expectedOutput.add(record("book", 5L, 11, 2L));
+ expectedOutput.add(record("fruit", 4L, 33, 1L));
+ expectedOutput.add(record("fruit", 3L, 44, 2L));
+ expectedOutput.add(record("fruit", 5L, 22, 1L));
+ expectedOutput.add(record("fruit", 4L, 33, 2L));
+ assertorWithRowNumber.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
@Test
@@ -91,24 +91,24 @@ public class RetractableTopNFunctionTest extends TopNFunctionTestBase {
testHarness.processElement(record("fruit", 5L, 22));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 1L, 12, 1L));
- expectedOutputOutput.add(record("book", 2L, 19, 2L));
- expectedOutputOutput.add(retractRecord("book", 2L, 19, 2L));
- expectedOutputOutput.add(retractRecord("book", 1L, 12, 1L));
- expectedOutputOutput.add(record("book", 4L, 11, 1L));
- expectedOutputOutput.add(record("book", 1L, 12, 2L));
- expectedOutputOutput.add(retractRecord("book", 1L, 12, 2L));
- expectedOutputOutput.add(record("book", 2L, 19, 2L));
- expectedOutputOutput.add(retractRecord("book", 2L, 19, 2L));
- expectedOutputOutput.add(record("book", 5L, 11, 2L));
- expectedOutputOutput.add(record("fruit", 4L, 33, 1L));
- expectedOutputOutput.add(record("fruit", 3L, 44, 2L));
- expectedOutputOutput.add(retractRecord("fruit", 4L, 33, 1L));
- expectedOutputOutput.add(retractRecord("fruit", 3L, 44, 2L));
- expectedOutputOutput.add(record("fruit", 5L, 22, 1L));
- expectedOutputOutput.add(record("fruit", 4L, 33, 2L));
- assertorWithRowNumber.assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 1L, 12, 1L));
+ expectedOutput.add(record("book", 2L, 19, 2L));
+ expectedOutput.add(retractRecord("book", 2L, 19, 2L));
+ expectedOutput.add(retractRecord("book", 1L, 12, 1L));
+ expectedOutput.add(record("book", 4L, 11, 1L));
+ expectedOutput.add(record("book", 1L, 12, 2L));
+ expectedOutput.add(retractRecord("book", 1L, 12, 2L));
+ expectedOutput.add(record("book", 2L, 19, 2L));
+ expectedOutput.add(retractRecord("book", 2L, 19, 2L));
+ expectedOutput.add(record("book", 5L, 11, 2L));
+ expectedOutput.add(record("fruit", 4L, 33, 1L));
+ expectedOutput.add(record("fruit", 3L, 44, 2L));
+ expectedOutput.add(retractRecord("fruit", 4L, 33, 1L));
+ expectedOutput.add(retractRecord("fruit", 3L, 44, 2L));
+ expectedOutput.add(record("fruit", 5L, 22, 1L));
+ expectedOutput.add(record("fruit", 4L, 33, 2L));
+ assertorWithRowNumber.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
// TODO RetractRankFunction could be sent less retraction message when does not need to retract row_number
@@ -126,26 +126,26 @@ public class RetractableTopNFunctionTest extends TopNFunctionTestBase {
testHarness.processElement(record("fruit", 3L, 44));
testHarness.processElement(record("fruit", 5L, 22));
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 1L, 12));
- expectedOutputOutput.add(record("book", 2L, 19));
- expectedOutputOutput.add(retractRecord("book", 1L, 12));
- expectedOutputOutput.add(record("book", 1L, 12));
- expectedOutputOutput.add(retractRecord("book", 2L, 19));
- expectedOutputOutput.add(record("book", 4L, 11));
- expectedOutputOutput.add(record("fruit", 4L, 33));
- expectedOutputOutput.add(record("fruit", 3L, 44));
- expectedOutputOutput.add(retractRecord("fruit", 4L, 33));
- expectedOutputOutput.add(retractRecord("fruit", 3L, 44));
- expectedOutputOutput.add(record("fruit", 4L, 33));
- expectedOutputOutput.add(record("fruit", 5L, 22));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 1L, 12));
+ expectedOutput.add(record("book", 2L, 19));
+ expectedOutput.add(retractRecord("book", 1L, 12));
+ expectedOutput.add(record("book", 1L, 12));
+ expectedOutput.add(retractRecord("book", 2L, 19));
+ expectedOutput.add(record("book", 4L, 11));
+ expectedOutput.add(record("fruit", 4L, 33));
+ expectedOutput.add(record("fruit", 3L, 44));
+ expectedOutput.add(retractRecord("fruit", 4L, 33));
+ expectedOutput.add(retractRecord("fruit", 3L, 44));
+ expectedOutput.add(record("fruit", 4L, 33));
+ expectedOutput.add(record("fruit", 5L, 22));
assertorWithoutRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
// do a snapshot, data could be recovered from state
OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
testHarness.close();
- expectedOutputOutput.clear();
+ expectedOutput.clear();
func = createFunction(RankType.ROW_NUMBER, new ConstantRankRange(1, 2), true, false);
testHarness = createTestHarness(func);
@@ -154,12 +154,12 @@ public class RetractableTopNFunctionTest extends TopNFunctionTestBase {
testHarness.open();
testHarness.processElement(record("book", 1L, 10));
- expectedOutputOutput.add(retractRecord("book", 1L, 12));
- expectedOutputOutput.add(retractRecord("book", 4L, 11));
- expectedOutputOutput.add(record("book", 4L, 11));
- expectedOutputOutput.add(record("book", 1L, 10));
+ expectedOutput.add(retractRecord("book", 1L, 12));
+ expectedOutput.add(retractRecord("book", 4L, 11));
+ expectedOutput.add(record("book", 4L, 11));
+ expectedOutput.add(record("book", 1L, 10));
assertorWithoutRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
testHarness.close();
}
@@ -177,18 +177,18 @@ public class RetractableTopNFunctionTest extends TopNFunctionTestBase {
testHarness.processElement(record("fruit", 1L, 22));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 2L, 12));
- expectedOutputOutput.add(record("book", 2L, 19));
- expectedOutputOutput.add(retractRecord("book", 2L, 19));
- expectedOutputOutput.add(retractRecord("book", 2L, 12));
- expectedOutputOutput.add(record("book", 2L, 12));
- expectedOutputOutput.add(record("book", 2L, 11));
- expectedOutputOutput.add(record("fruit", 1L, 33));
- expectedOutputOutput.add(retractRecord("fruit", 1L, 33));
- expectedOutputOutput.add(record("fruit", 1L, 22));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 2L, 12));
+ expectedOutput.add(record("book", 2L, 19));
+ expectedOutput.add(retractRecord("book", 2L, 19));
+ expectedOutput.add(retractRecord("book", 2L, 12));
+ expectedOutput.add(record("book", 2L, 12));
+ expectedOutput.add(record("book", 2L, 11));
+ expectedOutput.add(record("fruit", 1L, 33));
+ expectedOutput.add(retractRecord("fruit", 1L, 33));
+ expectedOutput.add(record("fruit", 1L, 22));
assertorWithoutRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
// TODO
@@ -206,17 +206,17 @@ public class RetractableTopNFunctionTest extends TopNFunctionTestBase {
testHarness.processElement(record("fruit", 5L, 22));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 1L, 12));
- expectedOutputOutput.add(record("book", 2L, 19));
- expectedOutputOutput.add(record("book", 1L, 12));
- expectedOutputOutput.add(record("book", 4L, 11));
- expectedOutputOutput.add(record("fruit", 4L, 33));
- expectedOutputOutput.add(record("fruit", 3L, 44));
- expectedOutputOutput.add(record("fruit", 4L, 33));
- expectedOutputOutput.add(record("fruit", 5L, 22));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 1L, 12));
+ expectedOutput.add(record("book", 2L, 19));
+ expectedOutput.add(record("book", 1L, 12));
+ expectedOutput.add(record("book", 4L, 11));
+ expectedOutput.add(record("fruit", 4L, 33));
+ expectedOutput.add(record("fruit", 3L, 44));
+ expectedOutput.add(record("fruit", 4L, 33));
+ expectedOutput.add(record("fruit", 5L, 22));
assertorWithoutRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/TopNFunctionTestBase.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/TopNFunctionTestBase.java
index 1fcb5c2..6cd215e 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/TopNFunctionTestBase.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/TopNFunctionTestBase.java
@@ -151,19 +151,19 @@ abstract class TopNFunctionTestBase {
testHarness.close();
// Notes: Delete message will be sent even disable generate retraction when not output rankNumber.
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 1L, 12));
- expectedOutputOutput.add(record("book", 2L, 19));
- expectedOutputOutput.add(deleteRecord("book", 2L, 19));
- expectedOutputOutput.add(record("book", 4L, 11));
- expectedOutputOutput.add(deleteRecord("book", 1L, 12));
- expectedOutputOutput.add(record("book", 5L, 11));
- expectedOutputOutput.add(record("fruit", 4L, 33));
- expectedOutputOutput.add(record("fruit", 3L, 44));
- expectedOutputOutput.add(deleteRecord("fruit", 3L, 44));
- expectedOutputOutput.add(record("fruit", 5L, 22));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 1L, 12));
+ expectedOutput.add(record("book", 2L, 19));
+ expectedOutput.add(deleteRecord("book", 2L, 19));
+ expectedOutput.add(record("book", 4L, 11));
+ expectedOutput.add(deleteRecord("book", 1L, 12));
+ expectedOutput.add(record("book", 5L, 11));
+ expectedOutput.add(record("fruit", 4L, 33));
+ expectedOutput.add(record("fruit", 3L, 44));
+ expectedOutput.add(deleteRecord("fruit", 3L, 44));
+ expectedOutput.add(record("fruit", 5L, 22));
assertorWithoutRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
@Test
@@ -183,17 +183,17 @@ abstract class TopNFunctionTestBase {
// Notes: Retract message will not be sent if disable generate retraction and output rankNumber.
// Because partition key + rankNumber decomposes a uniqueKey.
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 1L, 12, 1L));
- expectedOutputOutput.add(record("book", 2L, 19, 2L));
- expectedOutputOutput.add(record("book", 4L, 11, 1L));
- expectedOutputOutput.add(record("book", 1L, 12, 2L));
- expectedOutputOutput.add(record("book", 5L, 11, 2L));
- expectedOutputOutput.add(record("fruit", 4L, 33, 1L));
- expectedOutputOutput.add(record("fruit", 3L, 44, 2L));
- expectedOutputOutput.add(record("fruit", 5L, 22, 1L));
- expectedOutputOutput.add(record("fruit", 4L, 33, 2L));
- assertorWithRowNumber.assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 1L, 12, 1L));
+ expectedOutput.add(record("book", 2L, 19, 2L));
+ expectedOutput.add(record("book", 4L, 11, 1L));
+ expectedOutput.add(record("book", 1L, 12, 2L));
+ expectedOutput.add(record("book", 5L, 11, 2L));
+ expectedOutput.add(record("fruit", 4L, 33, 1L));
+ expectedOutput.add(record("fruit", 3L, 44, 2L));
+ expectedOutput.add(record("fruit", 5L, 22, 1L));
+ expectedOutput.add(record("fruit", 4L, 33, 2L));
+ assertorWithRowNumber.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
@Test
@@ -211,23 +211,23 @@ abstract class TopNFunctionTestBase {
testHarness.processElement(record("fruit", 5L, 22));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 1L, 12, 1L));
- expectedOutputOutput.add(record("book", 2L, 19, 2L));
- expectedOutputOutput.add(retractRecord("book", 1L, 12, 1L));
- expectedOutputOutput.add(retractRecord("book", 2L, 19, 2L));
- expectedOutputOutput.add(record("book", 4L, 11, 1L));
- expectedOutputOutput.add(record("book", 1L, 12, 2L));
- expectedOutputOutput.add(retractRecord("book", 1L, 12, 2L));
- expectedOutputOutput.add(record("book", 5L, 11, 2L));
- expectedOutputOutput.add(record("fruit", 4L, 33, 1L));
- expectedOutputOutput.add(record("fruit", 3L, 44, 2L));
- expectedOutputOutput.add(retractRecord("fruit", 4L, 33, 1L));
- expectedOutputOutput.add(retractRecord("fruit", 3L, 44, 2L));
- expectedOutputOutput.add(record("fruit", 4L, 33, 2L));
- expectedOutputOutput.add(record("fruit", 5L, 22, 1L));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 1L, 12, 1L));
+ expectedOutput.add(record("book", 2L, 19, 2L));
+ expectedOutput.add(retractRecord("book", 1L, 12, 1L));
+ expectedOutput.add(retractRecord("book", 2L, 19, 2L));
+ expectedOutput.add(record("book", 4L, 11, 1L));
+ expectedOutput.add(record("book", 1L, 12, 2L));
+ expectedOutput.add(retractRecord("book", 1L, 12, 2L));
+ expectedOutput.add(record("book", 5L, 11, 2L));
+ expectedOutput.add(record("fruit", 4L, 33, 1L));
+ expectedOutput.add(record("fruit", 3L, 44, 2L));
+ expectedOutput.add(retractRecord("fruit", 4L, 33, 1L));
+ expectedOutput.add(retractRecord("fruit", 3L, 44, 2L));
+ expectedOutput.add(record("fruit", 4L, 33, 2L));
+ expectedOutput.add(record("fruit", 5L, 22, 1L));
assertorWithRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
@Test
@@ -244,15 +244,15 @@ abstract class TopNFunctionTestBase {
testHarness.processElement(record("fruit", 5L, 22));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 2L, 19));
- expectedOutputOutput.add(retractRecord("book", 2L, 19));
- expectedOutputOutput.add(record("book", 1L, 12));
- expectedOutputOutput.add(record("fruit", 3L, 44));
- expectedOutputOutput.add(retractRecord("fruit", 3L, 44));
- expectedOutputOutput.add(record("fruit", 4L, 33));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 2L, 19));
+ expectedOutput.add(retractRecord("book", 2L, 19));
+ expectedOutput.add(record("book", 1L, 12));
+ expectedOutput.add(record("fruit", 3L, 44));
+ expectedOutput.add(retractRecord("fruit", 3L, 44));
+ expectedOutput.add(record("fruit", 4L, 33));
assertorWithoutRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
@Test
@@ -268,18 +268,18 @@ abstract class TopNFunctionTestBase {
testHarness.processElement(record("fruit", 1L, 22));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 2L, 12, 1L));
- expectedOutputOutput.add(record("book", 2L, 19, 2L));
- expectedOutputOutput.add(retractRecord("book", 2L, 12, 1L));
- expectedOutputOutput.add(retractRecord("book", 2L, 19, 2L));
- expectedOutputOutput.add(record("book", 2L, 11, 1L));
- expectedOutputOutput.add(record("book", 2L, 12, 2L));
- expectedOutputOutput.add(record("fruit", 1L, 33, 1L));
- expectedOutputOutput.add(retractRecord("fruit", 1L, 33, 1L));
- expectedOutputOutput.add(record("fruit", 1L, 22, 1L));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 2L, 12, 1L));
+ expectedOutput.add(record("book", 2L, 19, 2L));
+ expectedOutput.add(retractRecord("book", 2L, 12, 1L));
+ expectedOutput.add(retractRecord("book", 2L, 19, 2L));
+ expectedOutput.add(record("book", 2L, 11, 1L));
+ expectedOutput.add(record("book", 2L, 12, 2L));
+ expectedOutput.add(record("fruit", 1L, 33, 1L));
+ expectedOutput.add(retractRecord("fruit", 1L, 33, 1L));
+ expectedOutput.add(record("fruit", 1L, 22, 1L));
assertorWithRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
@Test
@@ -295,22 +295,22 @@ abstract class TopNFunctionTestBase {
testHarness.processElement(record("fruit", 3L, 44));
testHarness.processElement(record("fruit", 5L, 22));
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 1L, 12));
- expectedOutputOutput.add(record("book", 2L, 19));
- expectedOutputOutput.add(retractRecord("book", 2L, 19));
- expectedOutputOutput.add(record("book", 4L, 11));
- expectedOutputOutput.add(record("fruit", 4L, 33));
- expectedOutputOutput.add(record("fruit", 3L, 44));
- expectedOutputOutput.add(retractRecord("fruit", 3L, 44));
- expectedOutputOutput.add(record("fruit", 5L, 22));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 1L, 12));
+ expectedOutput.add(record("book", 2L, 19));
+ expectedOutput.add(retractRecord("book", 2L, 19));
+ expectedOutput.add(record("book", 4L, 11));
+ expectedOutput.add(record("fruit", 4L, 33));
+ expectedOutput.add(record("fruit", 3L, 44));
+ expectedOutput.add(retractRecord("fruit", 3L, 44));
+ expectedOutput.add(record("fruit", 5L, 22));
assertorWithoutRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
// do a snapshot, data could be recovered from state
OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
testHarness.close();
- expectedOutputOutput.clear();
+ expectedOutput.clear();
func = createFunction(RankType.ROW_NUMBER, new ConstantRankRange(1, 2), true, false);
testHarness = createTestHarness(func);
@@ -320,10 +320,10 @@ abstract class TopNFunctionTestBase {
testHarness.processElement(record("book", 1L, 10));
testHarness.close();
- expectedOutputOutput.add(retractRecord("book", 1L, 12));
- expectedOutputOutput.add(record("book", 1L, 10));
+ expectedOutput.add(retractRecord("book", 1L, 12));
+ expectedOutput.add(record("book", 1L, 10));
assertorWithoutRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
OneInputStreamOperatorTestHarness<BaseRow, BaseRow> createTestHarness(
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/UpdatableTopNFunctionTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/UpdatableTopNFunctionTest.java
index 26366a7..84e582d 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/UpdatableTopNFunctionTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/rank/UpdatableTopNFunctionTest.java
@@ -56,17 +56,17 @@ public class UpdatableTopNFunctionTest extends TopNFunctionTestBase {
testHarness.processElement(record("fruit", 1L, 22));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 2L, 19));
- expectedOutputOutput.add(retractRecord("book", 2L, 19));
- expectedOutputOutput.add(record("book", 2L, 18));
- expectedOutputOutput.add(record("fruit", 1L, 44));
- expectedOutputOutput.add(retractRecord("fruit", 1L, 44));
- expectedOutputOutput.add(record("fruit", 1L, 33));
- expectedOutputOutput.add(retractRecord("fruit", 1L, 33));
- expectedOutputOutput.add(record("fruit", 1L, 22));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 2L, 19));
+ expectedOutput.add(retractRecord("book", 2L, 19));
+ expectedOutput.add(record("book", 2L, 18));
+ expectedOutput.add(record("fruit", 1L, 44));
+ expectedOutput.add(retractRecord("fruit", 1L, 44));
+ expectedOutput.add(record("fruit", 1L, 33));
+ expectedOutput.add(retractRecord("fruit", 1L, 33));
+ expectedOutput.add(record("fruit", 1L, 22));
assertorWithoutRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
@Override
@@ -84,19 +84,19 @@ public class UpdatableTopNFunctionTest extends TopNFunctionTestBase {
testHarness.processElement(record("fruit", 1L, 22));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 2L, 19, 1L));
- expectedOutputOutput.add(retractRecord("book", 2L, 19, 1L));
- expectedOutputOutput.add(record("book", 2L, 12, 1L));
- expectedOutputOutput.add(retractRecord("book", 2L, 12, 1L));
- expectedOutputOutput.add(record("book", 2L, 11, 1L));
- expectedOutputOutput.add(record("fruit", 1L, 44, 1L));
- expectedOutputOutput.add(retractRecord("fruit", 1L, 44, 1L));
- expectedOutputOutput.add(record("fruit", 1L, 33, 1L));
- expectedOutputOutput.add(retractRecord("fruit", 1L, 33, 1L));
- expectedOutputOutput.add(record("fruit", 1L, 22, 1L));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 2L, 19, 1L));
+ expectedOutput.add(retractRecord("book", 2L, 19, 1L));
+ expectedOutput.add(record("book", 2L, 12, 1L));
+ expectedOutput.add(retractRecord("book", 2L, 12, 1L));
+ expectedOutput.add(record("book", 2L, 11, 1L));
+ expectedOutput.add(record("fruit", 1L, 44, 1L));
+ expectedOutput.add(retractRecord("fruit", 1L, 44, 1L));
+ expectedOutput.add(record("fruit", 1L, 33, 1L));
+ expectedOutput.add(retractRecord("fruit", 1L, 33, 1L));
+ expectedOutput.add(record("fruit", 1L, 22, 1L));
assertorWithRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
@Test
@@ -113,28 +113,28 @@ public class UpdatableTopNFunctionTest extends TopNFunctionTestBase {
testHarness.processElement(record("book", 2L, 1));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 2L, 19, 1L));
- expectedOutputOutput.add(retractRecord("book", 2L, 19, 1L));
- expectedOutputOutput.add(record("book", 2L, 19, 2L));
- expectedOutputOutput.add(record("book", 3L, 16, 1L));
- expectedOutputOutput.add(retractRecord("book", 3L, 16, 1L));
- expectedOutputOutput.add(record("book", 3L, 16, 2L));
- expectedOutputOutput.add(retractRecord("book", 2L, 19, 2L));
- expectedOutputOutput.add(record("book", 2L, 11, 1L));
- expectedOutputOutput.add(retractRecord("book", 3L, 16, 2L));
- expectedOutputOutput.add(record("book", 3L, 15, 2L));
- expectedOutputOutput.add(retractRecord("book", 3L, 15, 2L));
- expectedOutputOutput.add(retractRecord("book", 2L, 11, 1L));
- expectedOutputOutput.add(record("book", 2L, 11, 2L));
- expectedOutputOutput.add(record("book", 4L, 2, 1L));
- expectedOutputOutput.add(retractRecord("book", 2L, 11, 2L));
- expectedOutputOutput.add(retractRecord("book", 4L, 2, 1L));
- expectedOutputOutput.add(record("book", 2L, 1, 1L));
- expectedOutputOutput.add(record("book", 4L, 2, 2L));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 2L, 19, 1L));
+ expectedOutput.add(retractRecord("book", 2L, 19, 1L));
+ expectedOutput.add(record("book", 2L, 19, 2L));
+ expectedOutput.add(record("book", 3L, 16, 1L));
+ expectedOutput.add(retractRecord("book", 3L, 16, 1L));
+ expectedOutput.add(record("book", 3L, 16, 2L));
+ expectedOutput.add(retractRecord("book", 2L, 19, 2L));
+ expectedOutput.add(record("book", 2L, 11, 1L));
+ expectedOutput.add(retractRecord("book", 3L, 16, 2L));
+ expectedOutput.add(record("book", 3L, 15, 2L));
+ expectedOutput.add(retractRecord("book", 3L, 15, 2L));
+ expectedOutput.add(retractRecord("book", 2L, 11, 1L));
+ expectedOutput.add(record("book", 2L, 11, 2L));
+ expectedOutput.add(record("book", 4L, 2, 1L));
+ expectedOutput.add(retractRecord("book", 2L, 11, 2L));
+ expectedOutput.add(retractRecord("book", 4L, 2, 1L));
+ expectedOutput.add(record("book", 2L, 1, 1L));
+ expectedOutput.add(record("book", 4L, 2, 2L));
assertorWithRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
@Test
@@ -151,20 +151,20 @@ public class UpdatableTopNFunctionTest extends TopNFunctionTestBase {
testHarness.processElement(record("book", 2L, 1));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 2L, 19, 1L));
- expectedOutputOutput.add(record("book", 2L, 19, 2L));
- expectedOutputOutput.add(record("book", 3L, 16, 1L));
- expectedOutputOutput.add(record("book", 3L, 16, 2L));
- expectedOutputOutput.add(record("book", 2L, 11, 1L));
- expectedOutputOutput.add(record("book", 3L, 15, 2L));
- expectedOutputOutput.add(record("book", 2L, 11, 2L));
- expectedOutputOutput.add(record("book", 4L, 2, 1L));
- expectedOutputOutput.add(record("book", 2L, 1, 1L));
- expectedOutputOutput.add(record("book", 4L, 2, 2L));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 2L, 19, 1L));
+ expectedOutput.add(record("book", 2L, 19, 2L));
+ expectedOutput.add(record("book", 3L, 16, 1L));
+ expectedOutput.add(record("book", 3L, 16, 2L));
+ expectedOutput.add(record("book", 2L, 11, 1L));
+ expectedOutput.add(record("book", 3L, 15, 2L));
+ expectedOutput.add(record("book", 2L, 11, 2L));
+ expectedOutput.add(record("book", 4L, 2, 1L));
+ expectedOutput.add(record("book", 2L, 1, 1L));
+ expectedOutput.add(record("book", 4L, 2, 2L));
assertorWithRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
@Test
@@ -181,20 +181,20 @@ public class UpdatableTopNFunctionTest extends TopNFunctionTestBase {
testHarness.processElement(record("book", 2L, 1));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 2L, 19));
- expectedOutputOutput.add(record("book", 3L, 16));
- expectedOutputOutput.add(retractRecord("book", 2L, 19));
- expectedOutputOutput.add(record("book", 2L, 11));
- expectedOutputOutput.add(retractRecord("book", 3L, 16));
- expectedOutputOutput.add(record("book", 3L, 15));
- expectedOutputOutput.add(record("book", 4L, 2));
- expectedOutputOutput.add(retractRecord("book", 3L, 15));
- expectedOutputOutput.add(record("book", 2L, 1));
- expectedOutputOutput.add(retractRecord("book", 2L, 11));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 2L, 19));
+ expectedOutput.add(record("book", 3L, 16));
+ expectedOutput.add(retractRecord("book", 2L, 19));
+ expectedOutput.add(record("book", 2L, 11));
+ expectedOutput.add(retractRecord("book", 3L, 16));
+ expectedOutput.add(record("book", 3L, 15));
+ expectedOutput.add(record("book", 4L, 2));
+ expectedOutput.add(retractRecord("book", 3L, 15));
+ expectedOutput.add(record("book", 2L, 1));
+ expectedOutput.add(retractRecord("book", 2L, 11));
assertorWithRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
@Test
@@ -211,16 +211,16 @@ public class UpdatableTopNFunctionTest extends TopNFunctionTestBase {
testHarness.processElement(record("book", 2L, 1));
testHarness.close();
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("book", 2L, 19));
- expectedOutputOutput.add(record("book", 3L, 16));
- expectedOutputOutput.add(record("book", 2L, 11));
- expectedOutputOutput.add(record("book", 3L, 15));
- expectedOutputOutput.add(record("book", 4L, 2));
- expectedOutputOutput.add(deleteRecord("book", 3L, 15));
- expectedOutputOutput.add(record("book", 2L, 1));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("book", 2L, 19));
+ expectedOutput.add(record("book", 3L, 16));
+ expectedOutput.add(record("book", 2L, 11));
+ expectedOutput.add(record("book", 3L, 15));
+ expectedOutput.add(record("book", 4L, 2));
+ expectedOutput.add(deleteRecord("book", 3L, 15));
+ expectedOutput.add(record("book", 2L, 1));
assertorWithRowNumber
- .assertOutputEqualsSorted("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ .assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
}
}
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/ProcTimeSortOperatorTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/ProcTimeSortOperatorTest.java
index fa1e77c..c1a2b83 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/ProcTimeSortOperatorTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/ProcTimeSortOperatorTest.java
@@ -74,14 +74,14 @@ public class ProcTimeSortOperatorTest {
testHarness.processElement(record(9, 4L, "Comment#3", 9));
testHarness.setProcessingTime(1L);
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record(2, 2L, "Hello", 2));
- expectedOutputOutput.add(record(3, 3L, "Hello world", 3));
- expectedOutputOutput.add(record(5, 3L, "I am fine.", 5));
- expectedOutputOutput.add(record(6, 2L, "Luke Skywalker", 6));
- expectedOutputOutput.add(record(7, 1L, "Comment#1", 7));
- expectedOutputOutput.add(record(9, 4L, "Comment#3", 9));
- assertor.assertOutputEquals("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record(2, 2L, "Hello", 2));
+ expectedOutput.add(record(3, 3L, "Hello world", 3));
+ expectedOutput.add(record(5, 3L, "I am fine.", 5));
+ expectedOutput.add(record(6, 2L, "Luke Skywalker", 6));
+ expectedOutput.add(record(7, 1L, "Comment#1", 7));
+ expectedOutput.add(record(9, 4L, "Comment#3", 9));
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record(10, 4L, "Comment#4", 10));
testHarness.processElement(record(8, 4L, "Comment#2", 8));
@@ -94,7 +94,7 @@ public class ProcTimeSortOperatorTest {
OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
testHarness.close();
- expectedOutputOutput.clear();
+ expectedOutput.clear();
operator = createSortOperator();
testHarness = createTestHarness(operator);
@@ -103,14 +103,14 @@ public class ProcTimeSortOperatorTest {
testHarness.processElement(record(5, 3L, "I am fine.", 6));
testHarness.setProcessingTime(1L);
- expectedOutputOutput.add(record(1, 1L, "Hi", 2));
- expectedOutputOutput.add(record(1, 1L, "Hi", 1));
- expectedOutputOutput.add(record(4, 3L, "Helloworld, how are you?", 4));
- expectedOutputOutput.add(record(4, 5L, "Hello, how are you?", 4));
- expectedOutputOutput.add(record(5, 3L, "I am fine.", 6));
- expectedOutputOutput.add(record(8, 4L, "Comment#2", 8));
- expectedOutputOutput.add(record(10, 4L, "Comment#4", 10));
- assertor.assertOutputEquals("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record(1, 1L, "Hi", 2));
+ expectedOutput.add(record(1, 1L, "Hi", 1));
+ expectedOutput.add(record(4, 3L, "Helloworld, how are you?", 4));
+ expectedOutput.add(record(4, 5L, "Hello, how are you?", 4));
+ expectedOutput.add(record(5, 3L, "I am fine.", 6));
+ expectedOutput.add(record(8, 4L, "Comment#2", 8));
+ expectedOutput.add(record(10, 4L, "Comment#4", 10));
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
}
private ProcTimeSortOperator createSortOperator() {
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/RowTimeSortOperatorTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/RowTimeSortOperatorTest.java
index 684e05a..7066976 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/RowTimeSortOperatorTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/RowTimeSortOperatorTest.java
@@ -82,26 +82,26 @@ public class RowTimeSortOperatorTest {
testHarness.processElement(record(4, 5L, "Hello, how are you?", 4));
testHarness.processWatermark(new Watermark(4L));
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record(1, 1L, "Hi", 2));
- expectedOutputOutput.add(record(1, 1L, "Hi", 1));
- expectedOutputOutput.add(record(7, 1L, "Comment#1", 7));
- expectedOutputOutput.add(record(2, 2L, "Hello", 2));
- expectedOutputOutput.add(record(6, 2L, "Luke Skywalker", 6));
- expectedOutputOutput.add(record(3, 3L, "Hello world", 3));
- expectedOutputOutput.add(record(4, 3L, "Helloworld, how are you?", 4));
- expectedOutputOutput.add(record(5, 3L, "I am fine.", 5));
- expectedOutputOutput.add(record(8, 4L, "Comment#2", 8));
- expectedOutputOutput.add(record(9, 4L, "Comment#3", 9));
- expectedOutputOutput.add(record(10, 4L, "Comment#4", 10));
- expectedOutputOutput.add(new Watermark(4L));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record(1, 1L, "Hi", 2));
+ expectedOutput.add(record(1, 1L, "Hi", 1));
+ expectedOutput.add(record(7, 1L, "Comment#1", 7));
+ expectedOutput.add(record(2, 2L, "Hello", 2));
+ expectedOutput.add(record(6, 2L, "Luke Skywalker", 6));
+ expectedOutput.add(record(3, 3L, "Hello world", 3));
+ expectedOutput.add(record(4, 3L, "Helloworld, how are you?", 4));
+ expectedOutput.add(record(5, 3L, "I am fine.", 5));
+ expectedOutput.add(record(8, 4L, "Comment#2", 8));
+ expectedOutput.add(record(9, 4L, "Comment#3", 9));
+ expectedOutput.add(record(10, 4L, "Comment#4", 10));
+ expectedOutput.add(new Watermark(4L));
// do a snapshot, data could be recovered from state
OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
- assertor.assertOutputEquals("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
testHarness.close();
- expectedOutputOutput.clear();
+ expectedOutput.clear();
operator = createSortOperator(inputRowType, rowTimeIdx, gComparator);
testHarness = createTestHarness(operator);
@@ -111,18 +111,18 @@ public class RowTimeSortOperatorTest {
testHarness.processElement(record(5, 3L, "I am fine.", 6));
testHarness.processWatermark(new Watermark(5L));
- expectedOutputOutput.add(record(4, 5L, "Hello, how are you?", 4));
- expectedOutputOutput.add(new Watermark(5L));
+ expectedOutput.add(record(4, 5L, "Hello, how are you?", 4));
+ expectedOutput.add(new Watermark(5L));
- assertor.assertOutputEquals("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
// those watermark has no effect
testHarness.processWatermark(new Watermark(11L));
testHarness.processWatermark(new Watermark(12L));
- expectedOutputOutput.add(new Watermark(11L));
- expectedOutputOutput.add(new Watermark(12L));
+ expectedOutput.add(new Watermark(11L));
+ expectedOutput.add(new Watermark(12L));
- assertor.assertOutputEquals("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
}
@Test
@@ -150,25 +150,25 @@ public class RowTimeSortOperatorTest {
testHarness.processElement(record(4L, 3L, "Helloworld, how are you?", 4));
testHarness.processWatermark(new Watermark(9L));
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record(1L, 1L, "Hi", 2));
- expectedOutputOutput.add(record(1L, 1L, "Hi", 1));
- expectedOutputOutput.add(record(2L, 2L, "Hello", 2));
- expectedOutputOutput.add(record(3L, 2L, "Hello world", 3));
- expectedOutputOutput.add(record(4L, 3L, "Helloworld, how are you?", 4));
- expectedOutputOutput.add(record(5L, 3L, "I am fine.", 5));
- expectedOutputOutput.add(record(6L, 3L, "Luke Skywalker", 6));
- expectedOutputOutput.add(record(7L, 4L, "Comment#1", 7));
- expectedOutputOutput.add(record(8L, 4L, "Comment#2", 8));
- expectedOutputOutput.add(record(9L, 4L, "Comment#3", 9));
- expectedOutputOutput.add(new Watermark(9L));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record(1L, 1L, "Hi", 2));
+ expectedOutput.add(record(1L, 1L, "Hi", 1));
+ expectedOutput.add(record(2L, 2L, "Hello", 2));
+ expectedOutput.add(record(3L, 2L, "Hello world", 3));
+ expectedOutput.add(record(4L, 3L, "Helloworld, how are you?", 4));
+ expectedOutput.add(record(5L, 3L, "I am fine.", 5));
+ expectedOutput.add(record(6L, 3L, "Luke Skywalker", 6));
+ expectedOutput.add(record(7L, 4L, "Comment#1", 7));
+ expectedOutput.add(record(8L, 4L, "Comment#2", 8));
+ expectedOutput.add(record(9L, 4L, "Comment#3", 9));
+ expectedOutput.add(new Watermark(9L));
// do a snapshot, data could be recovered from state
OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
- assertor.assertOutputEquals("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
testHarness.close();
- expectedOutputOutput.clear();
+ expectedOutput.clear();
operator = createSortOperator(inputRowType, rowTimeIdx, null);
testHarness = createTestHarness(operator);
@@ -178,18 +178,18 @@ public class RowTimeSortOperatorTest {
testHarness.processElement(record(5L, 3L, "I am fine.", 6));
testHarness.processWatermark(new Watermark(10L));
- expectedOutputOutput.add(record(10L, 4L, "Comment#4", 10));
- expectedOutputOutput.add(new Watermark(10L));
+ expectedOutput.add(record(10L, 4L, "Comment#4", 10));
+ expectedOutput.add(new Watermark(10L));
- assertor.assertOutputEquals("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
// those watermark has no effect
testHarness.processWatermark(new Watermark(11L));
testHarness.processWatermark(new Watermark(12L));
- expectedOutputOutput.add(new Watermark(11L));
- expectedOutputOutput.add(new Watermark(12L));
+ expectedOutput.add(new Watermark(11L));
+ expectedOutput.add(new Watermark(12L));
- assertor.assertOutputEquals("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
}
private RowTimeSortOperator createSortOperator(BaseRowTypeInfo inputRowType, int rowTimeIdx,
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/StreamSortOperatorTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/StreamSortOperatorTest.java
index 62ce3d8..3105661 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/StreamSortOperatorTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/sort/StreamSortOperatorTest.java
@@ -67,18 +67,18 @@ public class StreamSortOperatorTest {
testHarness.processElement(record("world", 3));
testHarness.processElement(record("word", 4));
- List<Object> expectedOutputOutput = new ArrayList<>();
- expectedOutputOutput.add(record("hello", 2));
- expectedOutputOutput.add(record("hi", 1));
- expectedOutputOutput.add(record("word", 4));
- expectedOutputOutput.add(record("world", 3));
+ List<Object> expectedOutput = new ArrayList<>();
+ expectedOutput.add(record("hello", 2));
+ expectedOutput.add(record("hi", 1));
+ expectedOutput.add(record("word", 4));
+ expectedOutput.add(record("world", 3));
// do a snapshot, data could be recovered from state
OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
testHarness.close();
- assertor.assertOutputEquals("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
- expectedOutputOutput.clear();
+ expectedOutput.clear();
operator = createSortOperator();
testHarness = createTestHarness(operator);
@@ -88,13 +88,13 @@ public class StreamSortOperatorTest {
testHarness.processElement(record("aa", 1));
testHarness.close();
- expectedOutputOutput.add(record("aa", 1));
- expectedOutputOutput.add(record("abc", 1));
- expectedOutputOutput.add(record("hello", 2));
- expectedOutputOutput.add(record("hi", 1));
- expectedOutputOutput.add(record("word", 4));
- expectedOutputOutput.add(record("world", 3));
- assertor.assertOutputEquals("output wrong.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("aa", 1));
+ expectedOutput.add(record("abc", 1));
+ expectedOutput.add(record("hello", 2));
+ expectedOutput.add(record("hi", 1));
+ expectedOutput.add(record("word", 4));
+ expectedOutput.add(record("world", 3));
+ assertor.assertOutputEquals("output wrong.", expectedOutput, testHarness.getOutput());
}
private StreamSortOperator createSortOperator() {
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/window/WindowOperatorTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/window/WindowOperatorTest.java
index da9b403..1fad29c 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/window/WindowOperatorTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/window/WindowOperatorTest.java
@@ -109,7 +109,7 @@ public class WindowOperatorTest {
testHarness.open();
// process elements
- ConcurrentLinkedQueue<Object> expectedOutputOutput = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
// add elements out-of-order
testHarness.processElement(record("key2", 1, 3999L));
@@ -124,26 +124,26 @@ public class WindowOperatorTest {
testHarness.processElement(record("key2", 1, 1000L));
testHarness.processWatermark(new Watermark(999));
- expectedOutputOutput.add(record("key1", 3L, 3L, -2000L, 1000L, 999L));
- expectedOutputOutput.add(new Watermark(999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key1", 3L, 3L, -2000L, 1000L, 999L));
+ expectedOutput.add(new Watermark(999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processWatermark(new Watermark(1999));
- expectedOutputOutput.add(record("key1", 3L, 3L, -1000L, 2000L, 1999L));
- expectedOutputOutput.add(record("key2", 3L, 3L, -1000L, 2000L, 1999L));
- expectedOutputOutput.add(new Watermark(1999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key1", 3L, 3L, -1000L, 2000L, 1999L));
+ expectedOutput.add(record("key2", 3L, 3L, -1000L, 2000L, 1999L));
+ expectedOutput.add(new Watermark(1999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processWatermark(new Watermark(2999));
- expectedOutputOutput.add(record("key1", 3L, 3L, 0L, 3000L, 2999L));
- expectedOutputOutput.add(record("key2", 3L, 3L, 0L, 3000L, 2999L));
- expectedOutputOutput.add(new Watermark(2999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key1", 3L, 3L, 0L, 3000L, 2999L));
+ expectedOutput.add(record("key2", 3L, 3L, 0L, 3000L, 2999L));
+ expectedOutput.add(new Watermark(2999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// do a snapshot, close and restore again
OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
testHarness.close();
- expectedOutputOutput.clear();
+ expectedOutput.clear();
testHarness = createTestHarness(operator);
testHarness.setup();
@@ -151,27 +151,27 @@ public class WindowOperatorTest {
testHarness.open();
testHarness.processWatermark(new Watermark(3999));
- expectedOutputOutput.add(record("key2", 5L, 5L, 1000L, 4000L, 3999L));
- expectedOutputOutput.add(new Watermark(3999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key2", 5L, 5L, 1000L, 4000L, 3999L));
+ expectedOutput.add(new Watermark(3999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processWatermark(new Watermark(4999));
- expectedOutputOutput.add(record("key2", 2L, 2L, 2000L, 5000L, 4999L));
- expectedOutputOutput.add(new Watermark(4999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key2", 2L, 2L, 2000L, 5000L, 4999L));
+ expectedOutput.add(new Watermark(4999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processWatermark(new Watermark(5999));
- expectedOutputOutput.add(record("key2", 2L, 2L, 3000L, 6000L, 5999L));
- expectedOutputOutput.add(new Watermark(5999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key2", 2L, 2L, 3000L, 6000L, 5999L));
+ expectedOutput.add(new Watermark(5999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// those don't have any effect...
testHarness.processWatermark(new Watermark(6999));
testHarness.processWatermark(new Watermark(7999));
- expectedOutputOutput.add(new Watermark(6999));
- expectedOutputOutput.add(new Watermark(7999));
+ expectedOutput.add(new Watermark(6999));
+ expectedOutput.add(new Watermark(7999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.close();
@@ -193,7 +193,7 @@ public class WindowOperatorTest {
OneInputStreamOperatorTestHarness<BaseRow, BaseRow> testHarness = createTestHarness(operator);
- ConcurrentLinkedQueue<Object> expectedOutputOutput = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
testHarness.open();
@@ -203,27 +203,27 @@ public class WindowOperatorTest {
testHarness.setProcessingTime(1000);
- expectedOutputOutput.add(record("key2", 1L, 1L, -2000L, 1000L, 999L));
+ expectedOutput.add(record("key2", 1L, 1L, -2000L, 1000L, 999L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("key2", 1, Long.MAX_VALUE));
testHarness.processElement(record("key2", 1, Long.MAX_VALUE));
testHarness.setProcessingTime(2000);
- expectedOutputOutput.add(record("key2", 3L, 3L, -1000L, 2000L, 1999L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key2", 3L, 3L, -1000L, 2000L, 1999L));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("key1", 1, Long.MAX_VALUE));
testHarness.processElement(record("key1", 1, Long.MAX_VALUE));
testHarness.setProcessingTime(3000);
- expectedOutputOutput.add(record("key2", 3L, 3L, 0L, 3000L, 2999L));
- expectedOutputOutput.add(record("key1", 2L, 2L, 0L, 3000L, 2999L));
+ expectedOutput.add(record("key2", 3L, 3L, 0L, 3000L, 2999L));
+ expectedOutput.add(record("key1", 2L, 2L, 0L, 3000L, 2999L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("key1", 1, Long.MAX_VALUE));
testHarness.processElement(record("key1", 1, Long.MAX_VALUE));
@@ -231,12 +231,12 @@ public class WindowOperatorTest {
testHarness.setProcessingTime(7000);
- expectedOutputOutput.add(record("key2", 2L, 2L, 1000L, 4000L, 3999L));
- expectedOutputOutput.add(record("key1", 5L, 5L, 1000L, 4000L, 3999L));
- expectedOutputOutput.add(record("key1", 5L, 5L, 2000L, 5000L, 4999L));
- expectedOutputOutput.add(record("key1", 3L, 3L, 3000L, 6000L, 5999L));
+ expectedOutput.add(record("key2", 2L, 2L, 1000L, 4000L, 3999L));
+ expectedOutput.add(record("key1", 5L, 5L, 1000L, 4000L, 3999L));
+ expectedOutput.add(record("key1", 5L, 5L, 2000L, 5000L, 4999L));
+ expectedOutput.add(record("key1", 3L, 3L, 3000L, 6000L, 5999L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.close();
}
@@ -256,7 +256,7 @@ public class WindowOperatorTest {
OneInputStreamOperatorTestHarness<BaseRow, BaseRow> testHarness = createTestHarness(operator);
- ConcurrentLinkedQueue<Object> expectedOutputOutput = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
testHarness.open();
@@ -273,17 +273,17 @@ public class WindowOperatorTest {
testHarness.processElement(record("key2", 1, 1000L));
testHarness.processWatermark(new Watermark(999));
- expectedOutputOutput.add(new Watermark(999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processWatermark(new Watermark(1999));
- expectedOutputOutput.add(new Watermark(1999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(1999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// do a snapshot, close and restore again
OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
testHarness.close();
- expectedOutputOutput.clear();
+ expectedOutput.clear();
testHarness = createTestHarness(operator);
testHarness.setup();
@@ -291,31 +291,31 @@ public class WindowOperatorTest {
testHarness.open();
testHarness.processWatermark(new Watermark(2999));
- expectedOutputOutput.add(record("key1", 3L, 3L, 0L, 3000L, 2999L));
- expectedOutputOutput.add(record("key2", 3L, 3L, 0L, 3000L, 2999L));
- expectedOutputOutput.add(new Watermark(2999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key1", 3L, 3L, 0L, 3000L, 2999L));
+ expectedOutput.add(record("key2", 3L, 3L, 0L, 3000L, 2999L));
+ expectedOutput.add(new Watermark(2999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processWatermark(new Watermark(3999));
- expectedOutputOutput.add(new Watermark(3999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(3999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processWatermark(new Watermark(4999));
- expectedOutputOutput.add(new Watermark(4999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(4999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processWatermark(new Watermark(5999));
- expectedOutputOutput.add(record("key2", 2L, 2L, 3000L, 6000L, 5999L));
- expectedOutputOutput.add(new Watermark(5999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key2", 2L, 2L, 3000L, 6000L, 5999L));
+ expectedOutput.add(new Watermark(5999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// those don't have any effect...
testHarness.processWatermark(new Watermark(6999));
testHarness.processWatermark(new Watermark(7999));
- expectedOutputOutput.add(new Watermark(6999));
- expectedOutputOutput.add(new Watermark(7999));
+ expectedOutput.add(new Watermark(6999));
+ expectedOutput.add(new Watermark(7999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.close();
@@ -343,7 +343,7 @@ public class WindowOperatorTest {
OneInputStreamOperatorTestHarness<BaseRow, BaseRow> testHarness = createTestHarness(operator);
- ConcurrentLinkedQueue<Object> expectedOutputOutput = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
testHarness.open();
testHarness.setProcessingTime(0L);
@@ -362,24 +362,24 @@ public class WindowOperatorTest {
testHarness.processElement(record("key2", 1, 1000L));
testHarness.setProcessingTime(1000);
- expectedOutputOutput.add(record("key2", 2L, 2L, 3000L, 6000L, 5999L));
+ expectedOutput.add(record("key2", 2L, 2L, 3000L, 6000L, 5999L));
testHarness.processWatermark(new Watermark(999));
- expectedOutputOutput.add(new Watermark(999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.setProcessingTime(1001);
- expectedOutputOutput.add(record("key1", 3L, 3L, 0L, 3000L, 2999L));
- expectedOutputOutput.add(record("key2", 3L, 3L, 0L, 3000L, 2999L));
+ expectedOutput.add(record("key1", 3L, 3L, 0L, 3000L, 2999L));
+ expectedOutput.add(record("key2", 3L, 3L, 0L, 3000L, 2999L));
testHarness.processWatermark(new Watermark(1999));
testHarness.setProcessingTime(2001);
- expectedOutputOutput.add(new Watermark(1999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(1999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// do a snapshot, close and restore again
OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
testHarness.close();
- expectedOutputOutput.clear();
+ expectedOutput.clear();
// new a testHarness
testHarness = createTestHarness(operator);
@@ -390,50 +390,50 @@ public class WindowOperatorTest {
testHarness.setProcessingTime(3001);
testHarness.processWatermark(new Watermark(2999));
// on time fire key1 & key2 [0 ~ 3000) window, but because of early firing, on time result is ignored
- expectedOutputOutput.add(new Watermark(2999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(2999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("key2", 1, 4999L));
testHarness.processWatermark(new Watermark(3999));
testHarness.setProcessingTime(4001);
- expectedOutputOutput.add(new Watermark(3999));
- expectedOutputOutput.add(retractRecord("key2", 2L, 2L, 3000L, 6000L, 5999L));
- expectedOutputOutput.add(record("key2", 3L, 3L, 3000L, 6000L, 5999L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(3999));
+ expectedOutput.add(retractRecord("key2", 2L, 2L, 3000L, 6000L, 5999L));
+ expectedOutput.add(record("key2", 3L, 3L, 3000L, 6000L, 5999L));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// late arrival
testHarness.processElement(record("key2", 1, 2001L));
testHarness.processElement(record("key1", 1, 2030L));
// drop late elements
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.setProcessingTime(5100);
testHarness.processElement(record("key2", 1, 5122L));
testHarness.processWatermark(new Watermark(4999));
- expectedOutputOutput.add(new Watermark(4999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(4999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processWatermark(new Watermark(5999));
- expectedOutputOutput.add(retractRecord("key2", 3L, 3L, 3000L, 6000L, 5999L));
- expectedOutputOutput.add(record("key2", 4L, 4L, 3000L, 6000L, 5999L));
- expectedOutputOutput.add(new Watermark(5999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(retractRecord("key2", 3L, 3L, 3000L, 6000L, 5999L));
+ expectedOutput.add(record("key2", 4L, 4L, 3000L, 6000L, 5999L));
+ expectedOutput.add(new Watermark(5999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.setProcessingTime(6001);
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// those don't have any effect...
testHarness.processWatermark(new Watermark(6999));
testHarness.processWatermark(new Watermark(7999));
- expectedOutputOutput.add(new Watermark(6999));
- expectedOutputOutput.add(new Watermark(7999));
+ expectedOutput.add(new Watermark(6999));
+ expectedOutput.add(new Watermark(7999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// late arrival, drop
testHarness.processElement(record("key2", 1, 2877L));
testHarness.processElement(record("key1", 1, 2899L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.close();
@@ -463,7 +463,7 @@ public class WindowOperatorTest {
OneInputStreamOperatorTestHarness<BaseRow, BaseRow> testHarness = createTestHarness(operator);
- ConcurrentLinkedQueue<Object> expectedOutputOutput = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
testHarness.open();
testHarness.setProcessingTime(0L);
@@ -482,24 +482,24 @@ public class WindowOperatorTest {
testHarness.processElement(record("key2", 1, 1000L));
testHarness.setProcessingTime(1000);
- expectedOutputOutput.add(record("key2", 2L, 2L, 3000L, 6000L, 5999L));
+ expectedOutput.add(record("key2", 2L, 2L, 3000L, 6000L, 5999L));
testHarness.processWatermark(new Watermark(999));
- expectedOutputOutput.add(new Watermark(999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.setProcessingTime(1001);
- expectedOutputOutput.add(record("key1", 3L, 3L, 0L, 3000L, 2999L));
- expectedOutputOutput.add(record("key2", 3L, 3L, 0L, 3000L, 2999L));
+ expectedOutput.add(record("key1", 3L, 3L, 0L, 3000L, 2999L));
+ expectedOutput.add(record("key2", 3L, 3L, 0L, 3000L, 2999L));
testHarness.processWatermark(new Watermark(1999));
testHarness.setProcessingTime(2001);
- expectedOutputOutput.add(new Watermark(1999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(1999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// do a snapshot, close and restore again
OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
testHarness.close();
- expectedOutputOutput.clear();
+ expectedOutput.clear();
// new a testHarness
testHarness = createTestHarness(operator);
@@ -510,56 +510,56 @@ public class WindowOperatorTest {
testHarness.setProcessingTime(3001);
testHarness.processWatermark(new Watermark(2999));
// on time fire key1 & key2 [0 ~ 3000) window, but because of early firing, on time result is ignored
- expectedOutputOutput.add(new Watermark(2999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(2999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("key2", 1, 4999L));
testHarness.processWatermark(new Watermark(3999));
testHarness.setProcessingTime(4001);
- expectedOutputOutput.add(new Watermark(3999));
- expectedOutputOutput.add(retractRecord("key2", 2L, 2L, 3000L, 6000L, 5999L));
- expectedOutputOutput.add(record("key2", 3L, 3L, 3000L, 6000L, 5999L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(3999));
+ expectedOutput.add(retractRecord("key2", 2L, 2L, 3000L, 6000L, 5999L));
+ expectedOutput.add(record("key2", 3L, 3L, 3000L, 6000L, 5999L));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// late arrival
testHarness.processElement(record("key2", 1, 2001L));
- expectedOutputOutput.add(retractRecord("key2", 3L, 3L, 0L, 3000L, 2999L));
- expectedOutputOutput.add(record("key2", 4L, 4L, 0L, 3000L, 2999L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(retractRecord("key2", 3L, 3L, 0L, 3000L, 2999L));
+ expectedOutput.add(record("key2", 4L, 4L, 0L, 3000L, 2999L));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// late arrival
testHarness.processElement(record("key1", 1, 2030L));
- expectedOutputOutput.add(retractRecord("key1", 3L, 3L, 0L, 3000L, 2999L));
- expectedOutputOutput.add(record("key1", 4L, 4L, 0L, 3000L, 2999L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(retractRecord("key1", 3L, 3L, 0L, 3000L, 2999L));
+ expectedOutput.add(record("key1", 4L, 4L, 0L, 3000L, 2999L));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.setProcessingTime(5100);
testHarness.processElement(record("key2", 1, 5122L));
testHarness.processWatermark(new Watermark(4999));
- expectedOutputOutput.add(new Watermark(4999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(new Watermark(4999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processWatermark(new Watermark(5999));
- expectedOutputOutput.add(retractRecord("key2", 3L, 3L, 3000L, 6000L, 5999L));
- expectedOutputOutput.add(record("key2", 4L, 4L, 3000L, 6000L, 5999L));
- expectedOutputOutput.add(new Watermark(5999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(retractRecord("key2", 3L, 3L, 3000L, 6000L, 5999L));
+ expectedOutput.add(record("key2", 4L, 4L, 3000L, 6000L, 5999L));
+ expectedOutput.add(new Watermark(5999));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.setProcessingTime(6001);
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// those don't have any effect...
testHarness.processWatermark(new Watermark(6999));
testHarness.processWatermark(new Watermark(7999));
- expectedOutputOutput.add(new Watermark(6999));
- expectedOutputOutput.add(new Watermark(7999));
+ expectedOutput.add(new Watermark(6999));
+ expectedOutput.add(new Watermark(7999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// late arrival, but too late, drop
testHarness.processElement(record("key2", 1, 2877L));
testHarness.processElement(record("key1", 1, 2899L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.close();
@@ -582,7 +582,7 @@ public class WindowOperatorTest {
OneInputStreamOperatorTestHarness<BaseRow, BaseRow> testHarness = createTestHarness(operator);
- ConcurrentLinkedQueue<Object> expectedOutputOutput = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
testHarness.open();
@@ -598,10 +598,10 @@ public class WindowOperatorTest {
testHarness.setProcessingTime(5000);
- expectedOutputOutput.add(record("key2", 3L, 3L, 0L, 3000L, 2999L));
- expectedOutputOutput.add(record("key1", 2L, 2L, 0L, 3000L, 2999L));
+ expectedOutput.add(record("key2", 3L, 3L, 0L, 3000L, 2999L));
+ expectedOutput.add(record("key1", 2L, 2L, 0L, 3000L, 2999L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("key1", 1, 7000L));
testHarness.processElement(record("key1", 1, 7000L));
@@ -609,10 +609,10 @@ public class WindowOperatorTest {
testHarness.setProcessingTime(7000);
- expectedOutputOutput.add(record("key1", 3L, 3L, 3000L, 6000L, 5999L));
+ expectedOutput.add(record("key1", 3L, 3L, 3000L, 6000L, 5999L));
assertEquals(0L, operator.getWatermarkLatency().getValue());
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.close();
}
@@ -632,7 +632,7 @@ public class WindowOperatorTest {
OneInputStreamOperatorTestHarness<BaseRow, BaseRow> testHarness = createTestHarness(operator);
- ConcurrentLinkedQueue<Object> expectedOutputOutput = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
testHarness.open();
@@ -647,7 +647,7 @@ public class WindowOperatorTest {
// do a snapshot, close and restore again
OperatorSubtaskState snapshotV2 = testHarness.snapshot(0L, 0);
testHarness.close();
- expectedOutputOutput.clear();
+ expectedOutput.clear();
testHarness = createTestHarness(operator);
testHarness.setup();
@@ -665,11 +665,11 @@ public class WindowOperatorTest {
testHarness.processWatermark(new Watermark(12000));
- expectedOutputOutput.add(record("key1", 6L, 3L, 10L, 5500L, 5499L));
- expectedOutputOutput.add(record("key2", 6L, 3L, 0L, 5500L, 5499L));
+ expectedOutput.add(record("key1", 6L, 3L, 10L, 5500L, 5499L));
+ expectedOutput.add(record("key2", 6L, 3L, 0L, 5500L, 5499L));
- expectedOutputOutput.add(record("key2", 20L, 4L, 5501L, 9050L, 9049L));
- expectedOutputOutput.add(new Watermark(12000));
+ expectedOutput.add(record("key2", 20L, 4L, 5501L, 9050L, 9049L));
+ expectedOutput.add(new Watermark(12000));
// add a late data
testHarness.processElement(record("key1", 3, 4000L));
@@ -678,10 +678,10 @@ public class WindowOperatorTest {
testHarness.processWatermark(new Watermark(17999));
- expectedOutputOutput.add(record("key2", 30L, 2L, 15000L, 18000L, 17999L));
- expectedOutputOutput.add(new Watermark(17999));
+ expectedOutput.add(record("key2", 30L, 2L, 15000L, 18000L, 17999L));
+ expectedOutput.add(new Watermark(17999));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.setProcessingTime(18000);
assertEquals(1L, operator.getWatermarkLatency().getValue());
@@ -710,7 +710,7 @@ public class WindowOperatorTest {
BaseRowHarnessAssertor assertor = new BaseRowHarnessAssertor(
outputType.getFieldTypes(), new GenericRowRecordSortComparator(0, InternalTypes.STRING));
- ConcurrentLinkedQueue<Object> expectedOutputOutput = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
testHarness.open();
@@ -723,9 +723,9 @@ public class WindowOperatorTest {
testHarness.setProcessingTime(5000);
- expectedOutputOutput.add(record("key2", 2L, 2L, 3L, 4000L, 3999L));
+ expectedOutput.add(record("key2", 2L, 2L, 3L, 4000L, 3999L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("key2", 1, 5000L));
testHarness.processElement(record("key2", 1, 5000L));
@@ -735,10 +735,10 @@ public class WindowOperatorTest {
testHarness.setProcessingTime(10000);
- expectedOutputOutput.add(record("key2", 2L, 2L, 5000L, 8000L, 7999L));
- expectedOutputOutput.add(record("key1", 3L, 3L, 5000L, 8000L, 7999L));
+ expectedOutput.add(record("key2", 2L, 2L, 5000L, 8000L, 7999L));
+ expectedOutput.add(record("key1", 3L, 3L, 5000L, 8000L, 7999L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.close();
}
@@ -765,7 +765,7 @@ public class WindowOperatorTest {
OneInputStreamOperatorTestHarness<BaseRow, BaseRow> testHarness = createTestHarness(operator);
- ConcurrentLinkedQueue<Object> expectedOutputOutput = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
testHarness.open();
@@ -790,11 +790,11 @@ public class WindowOperatorTest {
testHarness.processWatermark(new Watermark(12000));
- expectedOutputOutput.add(record("key1", 36L, 3L, 10L, 4000L, 3999L));
- expectedOutputOutput.add(record("key2", 67L, 3L, 0L, 3000L, 2999L));
- expectedOutputOutput.add(new Watermark(12000));
+ expectedOutput.add(record("key1", 36L, 3L, 10L, 4000L, 3999L));
+ expectedOutput.add(record("key2", 67L, 3L, 0L, 3000L, 2999L));
+ expectedOutput.add(new Watermark(12000));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.close();
@@ -962,7 +962,7 @@ public class WindowOperatorTest {
OneInputStreamOperatorTestHarness<BaseRow, BaseRow> testHarness = createTestHarness(operator);
- ConcurrentLinkedQueue<Object> expectedOutputOutput = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
testHarness.open();
@@ -974,14 +974,14 @@ public class WindowOperatorTest {
testHarness.processWatermark(new Watermark(12000));
testHarness.setProcessingTime(12000L);
- expectedOutputOutput.add(record("key2", 6L, 3L, 0L));
- expectedOutputOutput.add(new Watermark(12000));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key2", 6L, 3L, 0L));
+ expectedOutput.add(new Watermark(12000));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// do a snapshot, close and restore again
OperatorSubtaskState snapshotV2 = testHarness.snapshot(0L, 0);
testHarness.close();
- expectedOutputOutput.clear();
+ expectedOutput.clear();
testHarness = createTestHarness(operator);
testHarness.setup();
@@ -989,27 +989,27 @@ public class WindowOperatorTest {
testHarness.open();
testHarness.processElement(record("key1", 2, 2500L));
- expectedOutputOutput.add(record("key1", 5L, 3L, 0L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key1", 5L, 3L, 0L));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("key2", 4, 5501L));
testHarness.processElement(record("key2", 5, 6000L));
testHarness.processElement(record("key2", 5, 6000L));
testHarness.processElement(record("key2", 6, 6050L));
- expectedOutputOutput.add(record("key2", 14L, 3L, 1L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key2", 14L, 3L, 1L));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("key1", 3, 4000L));
testHarness.processElement(record("key2", 10, 15000L));
testHarness.processElement(record("key2", 20, 15000L));
- expectedOutputOutput.add(record("key2", 36L, 3L, 2L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key2", 36L, 3L, 2L));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("key1", 2, 2500L));
testHarness.processElement(record("key1", 2, 2500L));
- expectedOutputOutput.add(record("key1", 7L, 3L, 1L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key1", 7L, 3L, 1L));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.close();
@@ -1032,7 +1032,7 @@ public class WindowOperatorTest {
OneInputStreamOperatorTestHarness<BaseRow, BaseRow> testHarness = createTestHarness(operator);
- ConcurrentLinkedQueue<Object> expectedOutputOutput = new ConcurrentLinkedQueue<>();
+ ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
testHarness.open();
@@ -1046,14 +1046,14 @@ public class WindowOperatorTest {
testHarness.processWatermark(new Watermark(12000));
testHarness.setProcessingTime(12000L);
- expectedOutputOutput.add(record("key2", 15L, 5L, 0L));
- expectedOutputOutput.add(new Watermark(12000));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key2", 15L, 5L, 0L));
+ expectedOutput.add(new Watermark(12000));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
// do a snapshot, close and restore again
OperatorSubtaskState snapshotV2 = testHarness.snapshot(0L, 0);
testHarness.close();
- expectedOutputOutput.clear();
+ expectedOutput.clear();
testHarness = createTestHarness(operator);
testHarness.setup();
@@ -1063,24 +1063,24 @@ public class WindowOperatorTest {
testHarness.processElement(record("key1", 3, 2500L));
testHarness.processElement(record("key1", 4, 2500L));
testHarness.processElement(record("key1", 5, 2500L));
- expectedOutputOutput.add(record("key1", 15L, 5L, 0L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key1", 15L, 5L, 0L));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("key2", 6, 6000L));
testHarness.processElement(record("key2", 7, 6000L));
testHarness.processElement(record("key2", 8, 6050L));
testHarness.processElement(record("key2", 9, 6050L));
- expectedOutputOutput.add(record("key2", 30L, 5L, 1L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key2", 30L, 5L, 1L));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.processElement(record("key1", 6, 4000L));
testHarness.processElement(record("key1", 7, 4000L));
testHarness.processElement(record("key1", 8, 4000L));
testHarness.processElement(record("key2", 10, 15000L));
testHarness.processElement(record("key2", 11, 15000L));
- expectedOutputOutput.add(record("key1", 30L, 5L, 1L));
- expectedOutputOutput.add(record("key2", 45L, 5L, 2L));
- assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutputOutput, testHarness.getOutput());
+ expectedOutput.add(record("key1", 30L, 5L, 1L));
+ expectedOutput.add(record("key2", 45L, 5L, 2L));
+ assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());
testHarness.close();