[FLINK-8428] [table] Implement stream-stream non-window left outer join #5327

hequn8128 · 2018-01-21T06:18:37Z

What is the purpose of the change

Implement stream-stream non-window left outer join for sql/table-api. A simple design doc can be found here

Brief change log

Add left join
- with non-equal predicates
- without non-equal predicates
Adapt retraction rules to left join. Outer join will generate retractions
Adapt UpsertTableSink. Table mode of dynamic table produced by left join is Update Mode, even if the table does not include a key definition
Add inner join test cases which consistent with test cases in batch.
Add left join test cases which consistent with test cases in batch.

Verifying this change

This change added tests and can be verified as follows:

Added integration tests for left join with or without non-equal predicates.
Added HarnessTests left join with or without non-equal predicates.
Add tests for AccMode generate by left join.
Add tests for UpsertSink followed left join.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (already docs)

hequn8128 · 2018-01-30T06:55:00Z

Hi, @twalthr It would be great if you can take a look at the pr. I'm looking forward to finish outer join (left/right/full) before the end of March. Besides, there are a few PRs planed to optimize inner/outer joins. Thanks :)

twalthr · 2018-01-31T13:23:33Z

Thanks for the reminder @hequn8128. I will review it in the next 2 weeks. If not, feel free to ping me again.

hequn8128 · 2018-02-24T11:16:50Z

hi, @twalthr Look forward to your review, thanks :-)

walterddr

Thanks for the pr @hequn8128 . Really looking forward to having LEFT JOIN support in datastream soon. I left a few comments and questions

Best,
Rong

walterddr · 2018-02-23T16:21:15Z

...raries/flink-table/src/main/scala/org/apache/flink/table/plan/util/UpdatingPlanChecker.scala

+          if (lInKeys.isEmpty || rInKeys.isEmpty) {
+            None
+          } else {
+            // Output of inner join must have keys if left and right both contain key(s).


remove "inner"?

Yes, thank you

walterddr · 2018-02-23T16:31:04Z

...libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/NonWindowJoin.scala

+    * @param defaultRow The result row used for output, right side fields will all be null.
+    * @param out        The collector for returning result values.
+    */
+  def collectWithNullRight(leftRow: Row, defaultRow: Row, out: Collector[Row]): Unit = {


We can probably reuse this function for RIGTH JOIN as well? maybe rename it to collectWithNull ?

I was trying to reduce the number of if else as much as possible which is inefficient. What do you think?

I was thinking about somthing like

def collectWithDefault(mainRow: Row, defaultRow: Row, out: Collector[Row]): Unit = { //... }

This way mainRow will be leftRow when LEFT JOIN and rightRow when RIGTH JOIN. It might be usable in FULL OUTER JOIN as well.

walterddr · 2018-02-26T14:09:52Z

...aries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/NonWindowLeftJoin.scala

+        if (rigthKeyNum == 1 && value.change) {
+          cRowWrapper.setChange(false)
+          collectWithNullRight(otherSideRow, resultRow, cRowWrapper)
+          retractFlag = true


Is retractFlag used anywhere?

Thanks, I will remove it :-)

walterddr · 2018-02-26T14:31:32Z

...aries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/NonWindowLeftJoin.scala

+        if (!value.change && rigthKeyNum == 0) {
+          cRowWrapper.setChange(true)
+          collectWithNullRight(otherSideRow, resultRow, cRowWrapper)
+          hasReEmittedNullRight = true


Same here, is hasReEmittedNullRight used?

walterddr · 2018-02-26T14:38:30Z

...ries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/table/JoinITCase.scala

@@ -201,18 +202,294 @@ class JoinITCase extends StreamingWithStateTestBase {
    // Proctime window output uncertain results, so assert has been ignored here.
  }

+  @Test
+  def testJoin(): Unit = {


Can this be more specific? like, inner equality join

walterddr · 2018-02-26T14:47:04Z

...aries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/NonWindowLeftJoin.scala

+      while (otherSideIterator.hasNext) {
+        val otherSideEntry = otherSideIterator.next()
+        val otherSideRow = otherSideEntry.getKey
+        val cntAndExpiredTime = otherSideEntry.getValue


cntAndExpiredTime is already defined in upper layer, maybe change the naming to avoid confusion?

Make sense.

walterddr · 2018-02-26T14:47:26Z

...aries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/NonWindowLeftJoin.scala

+      while (otherSideIterator.hasNext) {
+        val otherSideEntry = otherSideIterator.next()
+        val otherSideRow = otherSideEntry.getKey
+        val cntAndExpiredTime = otherSideEntry.getValue


cntAndExpiredTime is already defined in upper layer, maybe change the naming to avoid confusion?

walterddr · 2018-02-26T14:57:23Z

.../main/scala/org/apache/flink/table/runtime/join/NonWindowLeftJoinWithNonEquiPredicates.scala

+      }
+      // update matched cnt only when left row cnt is changed from 0 to 1. Each time encountered a
+      // new record from right, leftJoinCnt will also be updated.
+      if (cntAndExpiredTime.f0 == 1) {


Can't this check be triggered if cntAndExpiredTime was updated from 2 ==> 1?

Right, we should also take value.change into consideration.

walterddr · 2018-02-26T15:00:45Z

.../main/scala/org/apache/flink/table/runtime/join/NonWindowLeftJoinWithNonEquiPredicates.scala

+  override def open(parameters: Configuration): Unit = {
+    super.open(parameters)
+
+    val leftJoinCntDescriptor = new MapStateDescriptor[Row, Long](


probably use [Row, Int], to match with the type for count in LeftSideState and RightSideState?

Long is more safe. I will change all count type to Long. What do you think?

I think either is fine as long as they are consistent.

walterddr · 2018-02-26T15:03:56Z

...ries/flink-table/src/test/scala/org/apache/flink/table/runtime/harness/JoinHarnessTest.scala

@@ -985,4 +1017,232 @@ class JoinHarnessTest extends HarnessTestBase {

    testHarness.close()
  }
+
+  @Test
+  def testNonWindowLeftJoinWithOutNonEqualPred() {


WithOut ==> Without

hequn8128

Hi, @walterddr
Thanks very much for your review and suggestions. I will update it soon.
Best, Hequn

hequn8128 · 2018-02-27T14:21:47Z

...raries/flink-table/src/main/scala/org/apache/flink/table/plan/util/UpdatingPlanChecker.scala

+          if (lInKeys.isEmpty || rInKeys.isEmpty) {
+            None
+          } else {
+            // Output of inner join must have keys if left and right both contain key(s).


Yes, thank you

hequn8128 · 2018-02-27T14:22:05Z

...libraries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/NonWindowJoin.scala

+    * @param defaultRow The result row used for output, right side fields will all be null.
+    * @param out        The collector for returning result values.
+    */
+  def collectWithNullRight(leftRow: Row, defaultRow: Row, out: Collector[Row]): Unit = {


I was trying to reduce the number of if else as much as possible which is inefficient. What do you think?

hequn8128 · 2018-02-27T14:22:11Z

...aries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/NonWindowLeftJoin.scala

+      while (otherSideIterator.hasNext) {
+        val otherSideEntry = otherSideIterator.next()
+        val otherSideRow = otherSideEntry.getKey
+        val cntAndExpiredTime = otherSideEntry.getValue


Make sense.

hequn8128 · 2018-02-27T14:22:29Z

...aries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/NonWindowLeftJoin.scala

+        if (rigthKeyNum == 1 && value.change) {
+          cRowWrapper.setChange(false)
+          collectWithNullRight(otherSideRow, resultRow, cRowWrapper)
+          retractFlag = true


Thanks, I will remove it :-)

hequn8128 · 2018-02-27T14:22:38Z

.../main/scala/org/apache/flink/table/runtime/join/NonWindowLeftJoinWithNonEquiPredicates.scala

+  override def open(parameters: Configuration): Unit = {
+    super.open(parameters)
+
+    val leftJoinCntDescriptor = new MapStateDescriptor[Row, Long](


Long is more safe. I will change all count type to Long. What do you think?

hequn8128 · 2018-02-27T14:22:44Z

.../main/scala/org/apache/flink/table/runtime/join/NonWindowLeftJoinWithNonEquiPredicates.scala

+      }
+      // update matched cnt only when left row cnt is changed from 0 to 1. Each time encountered a
+      // new record from right, leftJoinCnt will also be updated.
+      if (cntAndExpiredTime.f0 == 1) {


Right, we should also take value.change into consideration.

hequn8128 · 2018-02-28T15:36:46Z

Update pr according to @walterddr 's suggestions.

hequn8128 · 2018-03-08T14:43:18Z

Hi @twalthr @walterddr
The latest update refactors interfaces and functions to make code more friendly to right/full join. The code of right/full joins are also ready and can be reached from https://github.com/hequn8128/flink/tree/outerjoin (branch:outerjoin).
@fhueske It would be great if you can also take a look.
Thanks all. Hequn

fhueske · 2018-03-15T10:14:05Z

Thanks @hequn8128! We're pretty busy with the Flink 1.5 release right now.
This will be one of the first features to add once 1.5 is out!

Best, Fabian

twalthr

Thank you very much for this PR @hequn8128 and sorry for the delay! The code is already in very good shape. I only added minor things:

I would move the enforceKeys change into a separate PR.
We need more tests at certain places because some code paths are never tested.
Would be great to add more inline and method comments to maintain the code in the future.

I will run a couple of TPC-H/TPC-DS queries all feedback has been addressed.

twalthr · 2018-04-18T12:27:47Z

.../flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamRel.scala

+  /**
+    * Whether the [[DataStreamRel]] produces retraction messages.
+    */
+  def producesRetractions: Boolean = false


Why isn't producesUpdates enough?

A join generates retraction if it's type is left/right/full. It is different from agg which generates retractions if sendsUpdatesAsRetraction(node) && node.producesUpdates is true.

Thanks for the explanation.

twalthr · 2018-04-18T12:28:49Z

...flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoin.scala

@@ -60,6 +60,9 @@ class DataStreamJoin(

  override def needsUpdatesAsRetraction: Boolean = true

+  // outer join will generate retractions
+  override def producesRetractions: Boolean = joinType != JoinRelType.INNER


Could you clarify this? A inner join is producing retractions.

Inner join doesn't produce retractions, left/right/full join does, for example, left join will retract the previous non-matched output when new matched row comes from the right side.

Thanks, now I understand the terminology between producing and just forwarding retractions.

twalthr · 2018-04-18T13:35:31Z

...ries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/NonWindowInnerJoin.scala

      isLeft: Boolean): Unit = {

    val inputRow = value.row
+    val (curProcessTime, _) = updateCurrentSide(value, ctx, timerState, currentSideState)


We should not use Scala sugar in runtime classes. This might create an tuple object for every processed element.

twalthr · 2018-04-18T13:46:38Z

.../flink-table/src/main/scala/org/apache/flink/table/runtime/join/NonWindowLeftRightJoin.scala

+    cRowWrapper.reset()
+    cRowWrapper.setCollector(out)
+    cRowWrapper.setChange(value.change)
+    cRowWrapper.setEmitCnt(0)


Remove this line.

twalthr · 2018-04-18T13:46:56Z

.../flink-table/src/main/scala/org/apache/flink/table/runtime/join/NonWindowLeftRightJoin.scala

+      recordFromLeft: Boolean): Unit = {
+
+    val inputRow = value.row
+    val (curProcessTime, _) = updateCurrentSide(value, ctx, timerState, currentSideState)


Same object creation issue as above.

twalthr · 2018-04-19T09:33:01Z

...ries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/NonWindowOuterJoin.scala

+    * Join current row with other side rows. Preserve current row if there are no matched rows
+    * from other side.
+    */
+  def preservedJoin(


Explain return type.

twalthr · 2018-04-19T09:46:13Z

...ries/flink-table/src/main/scala/org/apache/flink/table/runtime/join/NonWindowOuterJoin.scala

+    * return 1 if from right.
+    */
+  def getJoinCntIndex(isInputFromLeft: Boolean): Int = {
+    if (isInputFromLeft) {


You could return the state directly instead of returning the index.

twalthr · 2018-04-19T09:50:27Z

...k-libraries/flink-table/src/test/scala/org/apache/flink/table/plan/RetractionRulesTest.scala

+          "DataStreamCalc",
+          binaryNode(
+            "DataStreamJoin",
+            "DataStreamScan(true, Acc)",


Can you also add a test for a join that consumes from an aggregation?

testJoin() has covered this case.

twalthr · 2018-04-19T10:01:13Z

...flink-table/src/main/scala/org/apache/flink/table/plan/nodes/datastream/DataStreamJoin.scala

+          genFunction.code,
+          joinType == JoinRelType.LEFT,
+          queryConfig)
+      case JoinRelType.LEFT =>


Is there a reason why we don't support right outer joins here?

I planed to add right join in FLINK-8429. It's ok to add right join in this pr if you prefer.

We can also do it as part of FLINK-8429.

twalthr · 2018-04-19T10:21:29Z

...raries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/JoinITCase.scala

+  }
+
+  @Test
+  def testDataStreamJoinWithAggregation(): Unit = {


Can you try to use a consistent naming scheme for the test methods you added. Remove DataStream or Table from the names. And mark Inner, Outer joins correctly.

All names have been renamed, both stream and batch tests.

hequn8128 · 2018-04-20T12:18:41Z

@twalthr Hi, Great to see your review and valuable suggestions. I will update my pr late next week(maybe next weekend). Thanks very much.

hequn8128 · 2018-04-30T14:19:58Z

@twalthr Hi, thanks for your review. I have updated the pr according to your suggestions. Changes mainly include:

Remove changes about UpsertSink
Refactor test case name and add more test to cover code path
Add more method comments
Add another base class NonWindowOuterJoinWithNonEquiPredicates and move corresponding variables and functions into it.
Split CRowWrappingMultiOutputCollector into CRowWrappingMultiOutputCollector and LazyOutputCollector.

Best, Hequn.

twalthr · 2018-05-17T08:19:40Z

Thanks for the update @hequn8128. The changes look good. I tested your implementation on a cluster with TPC-H data. The results were equal to the batch results and the state clean-up worked. I will merge this :-)

Two different CoProcessFunctions are used to implement left join for performance reasons. One for left join with non-equal predicates, the other for left join without non-equal predicates. The main difference between them is, for left join without non-equal predicates, left rows can always find matching right rows as long as join keys are same. - Left join with non-equal predicates: Use a mapState to keep how many rows(joinCnt) from right table can be matched by current left row. If joinCnt is 0, output NULL right with left row. If joinCnt is changed from 0 to 1, retract the previous NULL right output and output the matched result. If joinCnt is changed from 1 to 0 when received a right retract input, retract the previous mathched result and output NULL right with left row. - Left join without non-equal predicates: We don't need to count joinCnt any more, because joinCnt is same with right state size, so check state size is ok. Table Modes: Left join will generate retractions, so DataStreamRel node of left join will working under AccRetract mode. Also, the table mode of dynamic table produced by left join is Update Mode, even if the table does not include a key definition. This closes apache#5327.

[FLINK-8428] [table] Implement stream-stream non-window left outer join

2766de5

walterddr reviewed Feb 26, 2018

View reviewed changes

hequn8128 commented Feb 27, 2018

View reviewed changes

1. Change all count type to Long 2.Remove useless variable

41b325d

refactor join function to be more friendly to right/full join

281e2f3

twalthr requested changes Apr 19, 2018

View reviewed changes

address comments

02f85a5

asfgit closed this in 8b95ba3 May 17, 2018

rmetzger added the component=API/TableSQL label Mar 18, 2019

flinkbot added component=TableSQL/API and removed component=API/TableSQL labels Mar 17, 2022

[FLINK-8428] [table] Implement stream-stream non-window left outer join #5327

[FLINK-8428] [table] Implement stream-stream non-window left outer join #5327

Conversation

hequn8128 commented Jan 21, 2018

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

hequn8128 commented Jan 30, 2018

twalthr commented Jan 31, 2018

hequn8128 commented Feb 24, 2018

walterddr left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hequn8128 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hequn8128 commented Feb 28, 2018

hequn8128 commented Mar 8, 2018 • edited

fhueske commented Mar 15, 2018

twalthr left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hequn8128 commented Apr 20, 2018

hequn8128 commented Apr 30, 2018

twalthr commented May 17, 2018

hequn8128 commented Mar 8, 2018 •

edited