[FLINK-12936][table-planner-blink] Support intersect all / minus all to blink planner #8898

JingsongLi · 2019-06-26T10:04:00Z

What is the purpose of the change

Now, we just support intersect and minus, See ReplaceIntersectWithSemiJoinRule and ReplaceMinusWithAntiJoinRule, replace intersect with null aware semi-join and distinct aggregate.
We need support intersect all and minus all too.
Presto and Spark already support them:
prestodb/presto#4918
https://issues.apache.org/jira/browse/SPARK-21274

I think them have a good rewrite design and we can follow them:
1.For intersect all
Input Query

SELECT c1 FROM ut1 INTERSECT ALL SELECT c1 FROM ut2

Rewritten Query

  SELECT c1
    FROM (
         SELECT replicate_row(min_count, c1)
         FROM (
              SELECT c1,
                     IF (vcol1_cnt > vcol2_cnt, vcol2_cnt, vcol1_cnt) AS min_count
              FROM (
                   SELECT   c1, count(vcol1) as vcol1_cnt, count(vcol2) as vcol2_cnt
                   FROM (
                        SELECT c1, true as vcol1, null as vcol2 FROM ut1
                        UNION ALL
                        SELECT c1, null as vcol1, true as vcol2 FROM ut2
                        ) AS union_all
                   GROUP BY c1
                   HAVING vcol1_cnt >= 1 AND vcol2_cnt >= 1
                  )
              )
          )

2.For minus all:
Input Query

SELECT c1 FROM ut1 EXCEPT ALL SELECT c1 FROM ut2

Rewritten Query

 SELECT c1
    FROM (
     SELECT replicate_rows(sum_val, c1)
       FROM (
         SELECT c1, sum_val
           FROM (
             SELECT c1, sum(vcol) AS sum_val
               FROM (
                 SELECT 1L as vcol, c1 FROM ut1
                 UNION ALL
                 SELECT -1L as vcol, c1 FROM ut2
              ) AS union_all
            GROUP BY union_all.c1
          )
        WHERE sum_val > 0
       )
   )

Verifying this change

ut

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? JavaDocs

flinkbot · 2019-06-26T10:06:29Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

godfreyhe

thanks for this PR @JingsongLi . please add rule test for RewriteMinusAll and RewriteIntersectAll, just as ReplaceIntersectWithSemiJoinRuleTest and ReplaceMinusWithAntiJoinRule do.

...anner-blink/src/test/scala/org/apache/flink/table/runtime/batch/sql/SetOperatorsITCase.scala

.../src/main/scala/org/apache/flink/table/plan/rules/logical/ReplaceSetOpWithJoinRuleBase.scala

...ner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/RewriteIntersectAll.scala

...planner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/RewriteMinusAll.scala

...ner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/RewriteIntersectAll.scala

godfreyhe · 2019-06-27T04:32:11Z

...planner-blink/src/test/resources/org/apache/flink/table/plan/stream/sql/SetOperatorsTest.xml

+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testIntersectAll">


this case should be in second commit ?

Maybe I should just squash second and third to a one commit.

godfreyhe · 2019-06-27T05:11:14Z

update commit message as Support intersect all/minus all in blink planner ?

JingsongLi · 2019-06-27T06:03:42Z

thanks for this PR @JingsongLi . please add rule test for RewriteMinusAll and RewriteIntersectAll, just as ReplaceIntersectWithSemiJoinRuleTest and ReplaceMinusWithAntiJoinRule do.

OK, and I will change the names to RewriteMinusAllRule and RewriteIntersectAllRule.

godfreyhe

i left some minor comments

godfreyhe · 2019-07-05T03:06:29Z

.../main/scala/org/apache/flink/table/plan/rules/logical/ReplaceIntersectWithSemiJoinRule.scala

  "ReplaceIntersectWithSemiJoinRule") {

  override def matches(call: RelOptRuleCall): Boolean = {
    val intersect: Intersect = call.rel(0)
-    // not support intersect all now.
-    intersect.isDistinct
+    intersect.isDistinct && intersect.getInputs.size() == 2


intersect.isDistinct => !intersect.all

and add a TODO remove "intersect.getInputs.size() == 2" limit ?

intersect.isDistinct => !intersect.all

Why?

and add a TODO remove "intersect.getInputs.size() == 2" limit ?

I think this rule should never deal with size bigger than 2, we should introduce a new rule to split it.

OK, Use all SetOp.all and I will add comment to explain these rules just handle the case of input size 2

godfreyhe · 2019-07-05T03:07:02Z

.../src/main/scala/org/apache/flink/table/plan/rules/logical/ReplaceMinusWithAntiJoinRule.scala

  "ReplaceMinusWithAntiJoinRule") {

  override def matches(call: RelOptRuleCall): Boolean = {
    val minus: Minus = call.rel(0)
-    // not support minus all now.
-    minus.isDistinct
+    minus.isDistinct && minus.getInputs.size() == 2


minus.isDistinct => !minus.all
add a TODO

godfreyhe · 2019-07-05T03:08:27Z

...blink/src/main/scala/org/apache/flink/table/plan/rules/logical/RewriteIntersectAllRule.scala

+
+  override def matches(call: RelOptRuleCall): Boolean = {
+    val intersect: Intersect = call.rel(0)
+    intersect.all && intersect.getInputs.size() == 2


add a TODO

godfreyhe · 2019-07-05T03:10:23Z

...ner-blink/src/main/scala/org/apache/flink/table/plan/rules/logical/RewriteMinusAllRule.scala

+    val leftWithAddedVirtualCols = leftBuilder
+        .push(left)
+        .project(leftBuilder.fields(fields) ++
+            Seq(leftBuilder.alias(leftBuilder.cast(leftBuilder.literal(1L), BIGINT), "vcol")))


please give a more meaningful name instead of vcol

name it vcol_marker, there is no appropriate name to describe it.

godfreyhe · 2019-07-05T03:44:38Z

LGTM, +1 to merge

…l to blink planner

JingsongLi · 2019-07-05T05:47:34Z

Travis passed(blink planner) in https://travis-ci.org/JingsongLi/flink/builds/554510279

rmetzger added review=description? component=TableSQL/Planner labels Jun 26, 2019

godfreyhe reviewed Jun 27, 2019

View reviewed changes

JingsongLi changed the title ~~[FLINK-12936][table-planner-blink] Support intersect/minus all to blink planner~~ [FLINK-12936][table-planner-blink] Support intersect all / minus all to blink planner Jun 27, 2019

JingsongLi force-pushed the setall branch 2 times, most recently from 625cba0 to b186176 Compare July 2, 2019 09:07

godfreyhe reviewed Jul 5, 2019

View reviewed changes

[FLINK-12936][table-planner-blink] Add SetOperatorsITCase to blink

ea90805

JingsongLi force-pushed the setall branch from b186176 to 70dc24d Compare July 5, 2019 03:40

[FLINK-12936][table-planner-blink] Support intersect all and minus al…

9004cf5

…l to blink planner

JingsongLi force-pushed the setall branch from 70dc24d to 9004cf5 Compare July 5, 2019 03:47

KurtYoung merged commit b4403f2 into apache:master Jul 5, 2019

JingsongLi deleted the setall branch July 10, 2019 08:18

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-12936][table-planner-blink] Support intersect all / minus all to blink planner #8898

[FLINK-12936][table-planner-blink] Support intersect all / minus all to blink planner #8898

JingsongLi commented Jun 26, 2019

flinkbot commented Jun 26, 2019 •

edited

godfreyhe left a comment

godfreyhe Jun 27, 2019

JingsongLi Jun 27, 2019

godfreyhe commented Jun 27, 2019

JingsongLi commented Jun 27, 2019

godfreyhe left a comment

godfreyhe Jul 5, 2019

JingsongLi Jul 5, 2019

JingsongLi Jul 5, 2019

godfreyhe Jul 5, 2019

godfreyhe Jul 5, 2019

godfreyhe Jul 5, 2019

JingsongLi Jul 5, 2019

godfreyhe commented Jul 5, 2019

JingsongLi commented Jul 5, 2019

[FLINK-12936][table-planner-blink] Support intersect all / minus all to blink planner #8898

[FLINK-12936][table-planner-blink] Support intersect all / minus all to blink planner #8898

Conversation

JingsongLi commented Jun 26, 2019

What is the purpose of the change

Verifying this change

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

Documentation

flinkbot commented Jun 26, 2019 • edited

Review Progress

godfreyhe left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

godfreyhe commented Jun 27, 2019

JingsongLi commented Jun 27, 2019

godfreyhe 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

godfreyhe commented Jul 5, 2019

JingsongLi commented Jul 5, 2019

flinkbot commented Jun 26, 2019 •

edited