[FLINK-5256] Extend DataSetSingleRowJoin to support Left and Right joins #3673

DmytroShkvyra · 2017-04-05T09:33:47Z

Thanks for contributing to Apache Flink. Before you open your pull request, please take the following check list into consideration.
If your changes take all of the items into account, feel free to open your pull request. For more information and/or questions please refer to the How To Contribute guide.
In addition to going through the list, please provide a meaningful description of your changes.

General
- JIRA issue [FLINK-5256] Extend DataSetSingleRowJoin to support Left and Right joins
- Continue of [FLINK-5256] Extend DataSetSingleRowJoin to support Left and Right joins #3033 PR (by @ex00)
Documentation
- No
Tests & Build
- Functionality added by the pull request is covered by tests
- mvn clean verify has been executed successfully locally or a Travis build has passed

DmytroShkvyra · 2017-04-06T09:27:02Z

Hi @fhueske could you take attention to this PR because previous #3033 (which you reviewed) was closed (author can't contribute to this jira any more)

fhueske · 2017-04-07T21:41:55Z

Thanks for working on this @DmytroShkvyra. I'll be on vacation next week but will review your PR when I am back.

Thanks, Fabian

DmytroShkvyra · 2017-04-19T17:17:17Z

Hi @fhueske could you review this variant. I had to removed tests with non-equality predicates (https://issues.apache.org/jira/browse/FLINK-5520)

fhueske

Hi @DmytroShkvyra, sorry for the delayed review.
We are currently working on getting the last features for the upcoming 1.3 release in and there are some open ends that I had to prioritize.

The approach of the PR looks good in general but needs a few changes. I left a couple of comments.

Thank you, Fabian

fhueske · 2017-04-26T16:14:55Z

...able/src/main/scala/org/apache/flink/table/plan/rules/dataSet/DataSetSingleRowJoinRule.scala

-    } else {
-      false
+    join.getJoinType match {
+      case JoinRelType.INNER | JoinRelType.LEFT | JoinRelType.RIGHT =>


We can only support this for outer joins if the single-row input is on the inner side of the join.
If the single row is on the outer side, we would need check that it does not join with any of the inner side before we send out a result with null fields. Since the single row is broadcasted / replicated, it is not possible to check this.

Hence, the condition should be:

case JoinRelType.INNER if isSingleRow(join.getLeft) || isSingleRow(join.getRight) => true case JoinRelType.LEFT if isSingleRow(join.getRight) => true case JoinRelType.RIGHT if isSingleRow(join.getLeft) => true case _ => false

fhueske · 2017-04-26T16:23:00Z

...nk-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetSingleRowJoin.scala

+          .map(field => getRowType.getFieldNames.indexOf(field.getName))
+          .map(i => s"${conversion.resultTerm}.setField($i,null);")
+
+        if (joinType == JoinRelType.LEFT && leftIsSingle) {


we only need to to handle the cases JoinRelType.LEFT && rightIsSingle and JoinRelType.RIGHT && leftIsSingle.

fhueske · 2017-04-27T09:51:05Z

...able/src/main/scala/org/apache/flink/table/plan/rules/dataSet/DataSetSingleRowJoinRule.scala

      dataSetLeftNode,
      dataSetRightNode,
      leftIsSingle,
+      rightIsSingle,


rightIsSingle is !leftIsSingle, so we can omit this parameter, IMO

fhueske · 2017-04-27T09:57:41Z

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

    broadcastSet match {
      case Some(singleInput) => function.join(multiInput, singleInput, out)
-      case None =>
+      case None => {


If we add a parameter `outerJoin: Boolean, we can do

case None if outerJoin => function.join(multiInput, null, out)

if we generate the function such that it checks if the single input is null

We cant pass null value to join function, but adding of outerJoin parameter is useful. It will fix cross join testcase.
Thanks!!!
Done.

Why not? We configure the CodeGenerator to expect null as input (see DataSetSingleRowJoin line 134).
The generated code will make sure we do not run into a NPE and the generated condition will be false if one of the input is null.

So, we can simply do case None if outerJoin => function.join(multiInput, null.asInstanceOf[IN2], out) here and case None if outerJoin => function.join(null.asInstanceOf[IN1], multiInput, out) in MapJoinRightRunner.

I check the generated code and this seems to be fine. Also the tests that you provided pass with that change.

@fhueske, If I use your code in MapJoinLeftRunner I get following spooky error:

scala.MatchError: None (of class scala.None$) at org.apache.flink.table.runtime.MapJoinLeftRunner.flatMap(MapJoinLeftRunner.scala:34) at org.apache.flink.api.common.operators.base.FlatMapOperatorBase.executeOnCollections(FlatMapOperatorBase.java:73)

I have googled this issue and find out that https://groups.google.com/forum/#!topic/akka-user/jvbbilXDKdA it could be related to akka.
So, I left code in MapJoinLeftRunner as was

A MatchError is thrown when no suitable pattern is found (Akka uses a lot of pattern matching, that's probably why you found that error, but it's actual a Scala thing).

This means that there is the case of an empty singleElement in case of an inner join. The problem can be easily solved by not replacing case None => with case None if outerJoin => but just inserting case None if outerJoin => in between. Pattern matching works from top to bottom and takes the first match (following matches are not executed).

fhueske · 2017-04-27T09:58:10Z

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

    broadcastSet match {
      case Some(singleInput) => function.join(singleInput, multiInput, out)
      case None =>
+        if (isRowClass(multiInput) && returnType.getTypeClass.equals(classOf[Row])) {


See comment on MapJoinLeftRunner

See reply on MapJoinLeftRunner
Done

fhueske · 2017-04-27T10:28:23Z

flink-libraries/flink-table/src/main/scala/org/apache/flink/table/codegen/generated.scala

-    resultType: TypeInformation[_])
+    resultType: TypeInformation[_]) {
+
+  var leftNullTerm: String = _


I don't think we need these attributes.
We can check with ${codeGenerator.input1Term} == null whether an input record is null.

@fhueske It doesn't work because ${codeGenerator.input1Term} is Row
@Override public void join(Object _in1, Object _in2, org.apache.flink.util.Collector c) throws Exception { org.apache.flink.types.Row in1 = (org.apache.flink.types.Row) _in1; org.apache.flink.types.Row in2 = (org.apache.flink.types.Row) _in2;
But I need know which field in join expression is null

fhueske · 2017-04-27T10:31:32Z

...nk-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetSingleRowJoin.scala

+          s"""
+             |${condition.code}
+             |${conversion.code}
+             |if(!${condition.resultTerm}){


we can add || ${codeGenerator.input1Term} == null to check if the first input is null for the JoinRelType.RIGHT && leftIsSingle case and || ${codeGenerator.input2Term} == null for the JoinRelType.LEFT && rightIsSingle case

fhueske · 2017-04-27T10:33:34Z

...nk-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetSingleRowJoin.scala

-        genFunction.code,
-        genFunction.returnType,
-        broadcastInputSetName)
+    if (joinType == JoinRelType.RIGHT) {


Do not invert the runners for JoinRelType.RIGHT the join input sides should be independent of the join type.

fhueske · 2017-04-27T10:41:37Z

...aries/flink-table/src/test/scala/org/apache/flink/table/api/scala/batch/sql/JoinITCase.scala

+    tEnv.registerTable("B", ds1)
+
+    val result = tEnv.sql(sqlQuery)
+    val expected = Seq(


The result should be Seq("3,3", "3,3", "3,3").
The outer side may only produce a single row with null if no other row matched.

fhueske · 2017-04-27T10:43:41Z

...aries/flink-table/src/test/scala/org/apache/flink/table/api/scala/batch/sql/JoinITCase.scala

+    val env = ExecutionEnvironment.getExecutionEnvironment
+    val tEnv = TableEnvironment.getTableEnvironment(env, config)
+    val sqlQuery =
+      "SELECT a, cnt " +


Please format all test queries in a consistent style

fhueske

Thanks for the update @DmytroShkvyra!

This looks pretty good. I made some minor comments inline.

Aside of those minor issues I noticed that the tests only cover equality joins. In fact, the main purpose of the SingleRowJoin is to support arbitrary join conditions that cannot be addressed by the default join which only handles equality joins.

When I tried to change some of the test cases to use inequality predicates (e.g., a < cnt), I notice that these joins are "blocked" by the FlinkLogicalJoinConverter rule.

We should

change that rule to also convert joins that can be translated by the SingleRowJoin (LEFT / RIGHT joins with a single row on the inner side).
adapt all test cases to use inequality predicates to enforce SingleRowJoins. Some of the current tests will simply use the default join, which is not what we want to test here.

What do you think?
Fabian

fhueske · 2017-05-09T16:27:53Z

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

    broadcastSet match {
      case Some(singleInput) => function.join(multiInput, singleInput, out)
-      case None =>
+      case None => {


Why not? We configure the CodeGenerator to expect null as input (see DataSetSingleRowJoin line 134).
The generated code will make sure we do not run into a NPE and the generated condition will be false if one of the input is null.

So, we can simply do case None if outerJoin => function.join(multiInput, null.asInstanceOf[IN2], out) here and case None if outerJoin => function.join(null.asInstanceOf[IN1], multiInput, out) in MapJoinRightRunner.

I check the generated code and this seems to be fine. Also the tests that you provided pass with that change.

fhueske · 2017-05-09T16:29:35Z

...nk-table/src/main/scala/org/apache/flink/table/plan/nodes/dataset/DataSetSingleRowJoin.scala

-        genFunction.returnType,
-        broadcastInputSetName)
-    }
+      if (!leftIsSingle) {


un-indent by 2 spaces

fhueske · 2017-05-09T16:30:44Z

...ble/src/test/scala/org/apache/flink/table/api/scala/batch/sql/DataSetSingleRowJoinTest.scala

+        "ON a1 = cnt"
+
+    //val queryRightJoin =
+    //  "SELECT a2 FROM (SELECT COUNT(*) AS cnt FROM B) RIGHT JOIN A ON a1 < cnt"


fhueske · 2017-05-09T16:32:00Z

...ble/src/test/scala/org/apache/flink/table/api/scala/batch/sql/DataSetSingleRowJoinTest.scala

+    util.addTable[(Long, Int)]("A", 'a1, 'a2)
+    util.addTable[(Int, Int)]("B", 'b1, 'b2)
+
+    val queryRightJoin =


The plan for this query does not include a single row join.

@fhueske Why??? We have single row at right side.

Because the optimizer decides to execute the join as a regular join (see DataSetJoin in the plan). If we change the predicate to a non-equality predicate (a1 < cnt) the regular join cannot be used, and we can force a single row join.

@fhueske see term("joinType", "RightOuterJoin") in plan.
I was really confused about "Disable outer joins with non-equality predicates" in https://issues.apache.org/jira/browse/FLINK-5520. Everything works fine until this update.
Do you want extend scope of this PR for support non-equality predicates?

The generate join is a RightOuterJoin but not a SingleRowJoin, which this test should verify.

We had to disable outer joins with predicates that include non-equi conditions in FLINK-5520 because they were not properly implemented. That implementation was based on splitting the join predicate into equi-conditions which were evaluated by the join and non-equi-conditions which were evaluated in a subsequent filter step. However, this split did not work correctly, because it would generate too many null rows if records passed the equi-join predicate in the join but not the non-equi predicate in the filter (since each filter call did only see a single row and would not know if all other rows had been filtered as well).

In our case the situation is different. We are translating the join into a NestedLoopJoin (where one side is at most one record), which can evaluate the full predicate including the non-equi conditions inside the join and know if we need to emit a null` result because there is only a single row that either matches the predicate or not.

DmytroShkvyra · 2017-05-10T16:02:40Z

@fhueske

When I tried to change some of the test cases to use inequality predicates (e.g., a < cnt), I notice that these joins are "blocked" by the FlinkLogicalJoinConverter rule.

We should

change that rule to also convert joins that can be translated by the SingleRowJoin (LEFT / RIGHT joins with a single row on the inner side).
adapt all test cases to use inequality predicates to enforce SingleRowJoins. Some of the current tests will simply use the default join, which is not what we want to test here.

What do you think?

I think we should reopen FLINK-5520. It's not my fault. I have removed wrong test as you wished.
Thanks for your guidance but I have no any clue how to FlinkLogicalJoinConverter rule works.
I'm going to vacation and possible I will be off-line.

fhueske

Thanks for the update @DmytroShkvyra.
I left a couple of comments.

Please let me know if you have questions.
Thank you,
Fabian

fhueske · 2017-05-10T16:09:09Z

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

-      case None =>
+      case None if outerJoin => function.
+                                join(multiInput, null.asInstanceOf[IN2], out)
+      case None => {


The case None => will only be reached for inner joins where the single row input is empty. Since an inner join does not return anything, this case can be empty as before.

In addition, the condition will never evaluate to true because the case of outerJoin == true is caught by the case above.

fhueske · 2017-05-10T16:10:29Z

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

+      case None if outerJoin => function.
+                                join(null.asInstanceOf[IN1], multiInput, out)
      case None =>
+        if (outerJoin && isRowClass(multiInput) && returnType.getTypeClass.equals(classOf[Row])) {


same as above.

fhueske · 2017-05-10T16:10:50Z

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

    }
  }
+
+  private def isRowClass(obj: Any) = obj match {


Can be removed

fhueske · 2017-05-10T16:10:57Z

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

    }
  }
+
+  private def isRowClass(obj: Any) = obj match {


can be removed

fhueske · 2017-05-10T16:17:34Z

...ble/src/test/scala/org/apache/flink/table/api/scala/batch/sql/DataSetSingleRowJoinTest.scala

+        )
+
+    util.verifySql(query, expected)
+  }


I add the following additional unit tests here:

left join with inequality join predicate

right join with equality join predicate

right join with inequality join predicate

You will find that the left/right joins with inequality predicates cannot be translated because of the too strict condition in FlinkLogicalJoinConverter. This condition needs to be adapted to accept single row outer joins with inequality predicates.

fhueske · 2017-05-10T16:21:18Z

...aries/flink-table/src/test/scala/org/apache/flink/table/api/scala/batch/sql/JoinITCase.scala

  }

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


Several of these tests are executed with regular joins because of the restriction that for an outer single row join, the single row input must be on the inner side. Some of these tests have the single row input on the outer side which will translate them into regular joins (which is only possible because the join conditions are equality predicates). These tests can be removed, because they test the existing join implementation.

In order to enforce tests of the new outer single row joins, the join condition must be an inequality predicate. It would be good to adapt the tests queries accordingly.

We need to check the following cases:

LEFT JOIN, single row on the right

LEFT JOIN, empty input on the right

RIGHT JOIN, single row on the left

RIGHT JOIN; empty input on the left

DmytroShkvyra · 2017-05-11T14:01:51Z

Hi @fhueske thanks for help. I have fixed your comments.

DmytroShkvyra · 2017-05-12T12:01:29Z

Hi @fhueske I have fixed all your notes and build almost green. :(
I have got only yarn test fail:

Failed tests: 
  YARNSessionFIFOITCase.testJavaAPI:249 Error while deploying YARN cluster: Couldn't deploy Yarn cluster

I think it not related to flink-tables

fhueske · 2017-05-12T12:13:40Z

Thanks for the update @DmytroShkvyra!

There are a few instable tests in the build, so failure is likely to be unrelated.
I'll have a look soon.

DmytroShkvyra · 2017-05-12T12:18:01Z

Thanks a lot Fabian!

fhueske

Hi @DmytroShkvyra,
this looks really good!
If found one thing that needs to be fixed, but I'll do that myself before merging.

Thanks for working on this.
Fabian

fhueske · 2017-05-12T14:56:51Z

.../flink-table/src/main/scala/org/apache/flink/table/plan/nodes/logical/FlinkLogicalJoin.scala

-    hasEqualityPredicates(join, joinInfo) || isSingleRowInnerJoin(join)
+    (hasEqualityPredicates(join, joinInfo)
+      || isSingleRowInnerJoin(join)
+      || isOuterJoinWithSingleRowAtOuterSide(join, joinInfo))


We need here the same condition as in DataSetSingleRowJoinRule.matches():

join.getJoinType match { case JoinRelType.INNER if isSingleRow(join.getRight) || isSingleRow(join.getLeft) => true case JoinRelType.LEFT if isSingleRow(join.getRight) => true case JoinRelType.RIGHT if isSingleRow(join.getLeft) => true case _ => false }

@fhueske Thanks for your guidances!

…Right joins. This closes #3673.

…Right joins. This closes apache#3673.

DmytroShkvyra force-pushed the FLINK-5256 branch 2 times, most recently from 447ae9b to 2cb9c45 Compare April 19, 2017 14:38

fhueske reviewed Apr 27, 2017

View reviewed changes

DmytroShkvyra force-pushed the FLINK-5256 branch 5 times, most recently from 2d88e18 to bdfac39 Compare May 8, 2017 15:24

fhueske reviewed May 9, 2017

View reviewed changes

DmytroShkvyra force-pushed the FLINK-5256 branch from bdfac39 to 4501995 Compare May 10, 2017 14:40

fhueske reviewed May 10, 2017

View reviewed changes

DmytroShkvyra force-pushed the FLINK-5256 branch from 4501995 to 9924e3f Compare May 11, 2017 13:56

DmytroShkvyra added 12 commits May 12, 2017 11:48

[FLINK-5256] Extend DataSetSingleRowJoin to support Left and Right joins

30b5bb2

[FLINK-5256] Extend DataSetSingleRowJoin to support Left and Right joins

49fe926

[FLINK-5256] Change tests accordingly FLINK-5520

18ce319

[FLINK-5256] Extend DataSetSingleRowJoin to support Left and Right joins

d49f283

[FLINK-5256] Extend DataSetSingleRowJoin to support Left and Right joins

6ef3bb6

[FLINK-5256] Extend DataSetSingleRowJoin to support Left and Right joins

397377a

[FLINK-5256] Add joinType to explain method

4acaa72

[FLINK-5256] Fix style

1e56704

[FLINK-5256] Fix conflict

c53710e

[FLINK-5256] Extend DataSetSingleRowJoin to support Left and Right joins

417d0a6

[FLINK-5256] Review fix

ef2fc0e

[FLINK-5256] Fix style

6f9b311

DmytroShkvyra force-pushed the FLINK-5256 branch from 9924e3f to 6f9b311 Compare May 12, 2017 08:48

fhueske reviewed May 12, 2017

View reviewed changes

asfgit closed this in ef751b2 May 12, 2017

asfgit pushed a commit that referenced this pull request May 12, 2017

[FLINK-5256] [table] Extend DataSetSingleRowJoin to support Left and …

01171e8

…Right joins. This closes #3673.

fanyon pushed a commit to fanyon/flink that referenced this pull request May 15, 2017

[FLINK-5256] [table] Extend DataSetSingleRowJoin to support Left and …

4d65d53

…Right joins. This closes apache#3673.

hequn8128 pushed a commit to hequn8128/flink that referenced this pull request Jun 22, 2017

[FLINK-5256] [table] Extend DataSetSingleRowJoin to support Left and …

b2a4928

…Right joins. This closes apache#3673.

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

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

[FLINK-5256] Extend DataSetSingleRowJoin to support Left and Right joins #3673

[FLINK-5256] Extend DataSetSingleRowJoin to support Left and Right joins #3673

Uh oh!

Conversation

DmytroShkvyra commented Apr 5, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

DmytroShkvyra commented Apr 6, 2017

Uh oh!

fhueske commented Apr 7, 2017

Uh oh!

DmytroShkvyra commented Apr 19, 2017

Uh oh!

fhueske left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

DmytroShkvyra May 4, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fhueske May 9, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

DmytroShkvyra May 4, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

fhueske left a comment

Choose a reason for hiding this comment

Uh oh!

fhueske May 9, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

DmytroShkvyra commented Apr 5, 2017 •

edited

Loading

DmytroShkvyra May 4, 2017 •

edited

Loading

fhueske May 9, 2017 •

edited

Loading

DmytroShkvyra May 4, 2017 •

edited

Loading

fhueske May 9, 2017 •

edited

Loading

DmytroShkvyra commented May 10, 2017 •

edited

Loading