[FLINK-18988][table] Continuous query with LATERAL and LIMIT produces… #13291

danny0405 · 2020-09-01T03:31:26Z

… wrong result

The batch mode rank only supports RANK function, so we only rewrite the
stream mode query.

What is the purpose of the change

Fix the query of pattern:

SELECT state, name
FROM
  (SELECT DISTINCT state FROM cities) states,
  LATERAL (
    SELECT name, pop
    FROM cities
    WHERE state = states.state
    ORDER BY pop
    DESC LIMIT 3
  )

Before the patch, the query generates a wrong plan then wrong results.

Brief change log

Add a new rule CorrelateSortToRankRule for the rewrite
Add plan test and IT test

Verifying this change

Added tests.

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, Kubernetes/Yarn/Mesos, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? no
If yes, how is the feature documented? not documented

flinkbot · 2020-09-01T03:33:51Z

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.

Automated Checks

Last check on commit 8406cd5 (Tue Sep 01 03:33:51 UTC 2020)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

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

flinkbot · 2020-09-01T03:47:40Z

CI report:

b02876f Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

twalthr · 2020-09-01T06:40:27Z

...c/main/scala/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.scala

+ *
+ * <p>This rule can only be used in HepPlanner.
+ */
+class CorrelateSortToRankRule extends RelOptRule(


side comment: Our long-term goal is to get rid of Scala. This class could have been implemented easily in Java. Please keep that in mind for future contributions.

Thanks for the reminder, i saw most of the rules are implemented as scala code when contribution, do you mean we prefer java rules in the future ?

See FLIP-32 Appendix: Porting Guidelines.

A new planner rule or node that only depends on Calcite and runtime classes should be implemented in Java.

I see, thanks for the share ~

We rework so many classes all the time, eventually the Scala code will hopefully be gone at some point.

Sure, it would be exciting if all the code can switch to Java.

twalthr

I'm trying to base an example on this PR. But the results differ between batch and streaming mode. It seems that the batch mode now outputs the global maximum.

danny0405 · 2020-09-17T04:19:07Z

I'm trying to base an example on this PR. But the results differ between batch and streaming mode. It seems that the batch mode now outputs the global maximum.

The patch is only for streaming purpose, let me check if batch mode is supported.

… wrong result The batch mode rank only supports RANK function, so we only rewrite the stream mode query.

godfreyhe

Thanks for the fix @danny0405, I left some comments.
btw, I find the blink batch planner does not support the given query, and I get some errors in sql-client, like "org.apache.flink.table.api.TableException: unexpected correlate variable $cor1 in the plan"

godfreyhe · 2020-11-17T09:27:07Z

...src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkStreamProgram.scala

+    // rewrite before decorrelation
+    chainedProgram.addLast(
+      PRE_DECORRELATE_REWRITE,
+      FlinkHepRuleSetProgramBuilder.newBuilder
+        .setHepRulesExecutionType(HEP_RULES_EXECUTION_TYPE.RULE_SEQUENCE)
+        .setHepMatchOrder(HepMatchOrder.BOTTOM_UP)
+        .add(FlinkStreamRuleSets.PRE_DECORRELATION_RULES)
+        .build())


We should merge this program into DECORRELATE

godfreyhe · 2020-11-17T09:29:53Z

...c/main/scala/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.scala

+ * {{{
+ *   LogicalProject(state=[$0], name=[$1])
+ *   +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{0}])
+ *      :- LogicalAggregate(group=[{0}])
+ *      :  +- LogicalProject(state=[$1])
+ *      :     +- LogicalTableScan(table=[[default_catalog, default_database, cities]])
+ *      +- LogicalSort(sort0=[$1], dir0=[DESC-nulls-last], fetch=[3])
+ *         +- LogicalProject(name=[$0], pop=[$2])
+ *            +- LogicalFilter(condition=[=($1, $cor0.state)])
+ *               +- LogicalTableScan(table=[[default_catalog, default_database, cities]])
+ * }}}
+ *
+ * <p>would be transformed to
+ *
+ * {{{
+ *   LogicalProject(state=[$0], name=[$1])
+ *    +- LogicalProject(state=[$1], name=[$0], pop=[$2])
+ *       +- LogicalRank(rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=3],
+ *            partitionBy=[$1], orderBy=[$2 DESC], select=[name=$0, state=$1, pop=$2])
+ *          +- LogicalTableScan(table=[[default_catalog, default_database, cities]])
+ * }}}


it seems the rewrite is not correct, consider the following example:

SELECT state, name FROM (SELECT DISTINCT state FROM cities1) states, LATERAL ( SELECT name, pop FROM cities2 WHERE state = states.state ORDER BY pop DESC LIMIT 3 );

The outer table (cities1) and the inner table (cities2) are different.

Yes, that is why i added the match condition aggInput.getInput.getDigest.equals(filter.getInput.getDigest). If the outer and inner table are different, the rule can not be matched.

Thanks for the explanation

danny0405 · 2020-11-17T12:30:25Z

Thanks for the fix @danny0405, I left some comments.
btw, I find the blink batch planner does not support the given query, and I get some errors in sql-client, like "org.apache.flink.table.api.TableException: unexpected correlate variable $cor1 in the plan"

Yes, because our batch rank only supports rank type as RANK, see BatchExecRankRule match condition, while here we use the ROW_NUMBER.

godfreyhe

We should add CorrelateSortToRankRuleTest which only involves the minimal rule set to verify the logic of CorrelateSortToRankRule, including the supported cases and unsupported cases

godfreyhe · 2020-11-18T02:29:54Z

...c/main/scala/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.scala

+ * {{{
+ *   LogicalProject(state=[$0], name=[$1])
+ *   +- LogicalCorrelate(correlation=[$cor0], joinType=[inner], requiredColumns=[{0}])
+ *      :- LogicalAggregate(group=[{0}])
+ *      :  +- LogicalProject(state=[$1])
+ *      :     +- LogicalTableScan(table=[[default_catalog, default_database, cities]])
+ *      +- LogicalSort(sort0=[$1], dir0=[DESC-nulls-last], fetch=[3])
+ *         +- LogicalProject(name=[$0], pop=[$2])
+ *            +- LogicalFilter(condition=[=($1, $cor0.state)])
+ *               +- LogicalTableScan(table=[[default_catalog, default_database, cities]])
+ * }}}
+ *
+ * <p>would be transformed to
+ *
+ * {{{
+ *   LogicalProject(state=[$0], name=[$1])
+ *    +- LogicalProject(state=[$1], name=[$0], pop=[$2])
+ *       +- LogicalRank(rankType=[ROW_NUMBER], rankRange=[rankStart=1, rankEnd=3],
+ *            partitionBy=[$1], orderBy=[$2 DESC], select=[name=$0, state=$1, pop=$2])
+ *          +- LogicalTableScan(table=[[default_catalog, default_database, cities]])
+ * }}}


Thanks for the explanation

godfreyhe · 2020-11-18T02:31:56Z

...c/main/scala/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.scala

+    val agg: Aggregate = call.rel(1)
+    if (agg.getAggCallList.size() > 0
+      || agg.getGroupSets.size() > 1
+      || agg.getGroupSet.cardinality() != 1) {


agg.getGroupSet.cardinality() != 1

we should support multiple equal conditions, such as: state = states.state and name = states.name

Yes, but the code would complicate a lot, we can support it in the future. In general, one equal condition is enough, the project distinct value in order to avoid unnecessary Cartesian Product.

godfreyhe · 2020-11-18T02:42:57Z

...c/main/scala/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.scala

+    val oriBuilder = call.builder()
+    val builder = FlinkRelBuilder.of(oriBuilder.getCluster, oriBuilder.getRelOptSchema)


nit: one simply way is: using FlinkRelFactories.FLINK_REL_BUILDER to construct RelOptRule, and then we can cast call.builder() as FlinkRelBuilder. one example is FlinkSubQueryRemoveRule

godfreyhe · 2020-11-18T02:46:34Z

...c/main/scala/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.scala

+        RankType.ROW_NUMBER,
+        new ConstantRankRange(
+          1,
+          sort.fetch.asInstanceOf[RexLiteral].getValueAs(classOf[java.lang.Long])),


we can use SortUtil.getLimitEnd

godfreyhe · 2020-11-18T02:50:39Z

...c/main/scala/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.scala

+  val INSTANCE = new CorrelateSortToRankRule
+}
+
+


nit: redundant line

godfreyhe

Thanks for the update, LGTM overall, there is only a minor comment. @twalthr do you have any other concern?

godfreyhe · 2020-11-19T02:33:47Z

...c/main/scala/org/apache/flink/table/planner/plan/rules/logical/CorrelateSortToRankRule.scala

+    val newCollation = RelCollations.of(newFieldCollations)
+
+    val newRel = builder
+      .push(filter.getInput()).asInstanceOf[FlinkRelBuilder]


cast is redundant

Updated, thanks ~

danny0405 · 2020-11-20T09:20:22Z

Thanks @godfreyhe , i have addressed the review comments, can you take a look again ~ Thanks so much in advance ~

godfreyhe

LGTM

twalthr · 2020-11-24T10:54:09Z

Thanks @danny0405 and @godfreyhe. I will merge this now...

… wrong result This closes apache#13291. The batch mode rank only supports RANK function, so we only rewrite the stream mode query.

rmetzger added the review=description? label Sep 1, 2020

rmetzger added the component=TableSQL/Planner label Sep 1, 2020

twalthr reviewed Sep 1, 2020

View reviewed changes

danny0405 force-pushed the FLINK-18988 branch from 8406cd5 to 3dff129 Compare September 1, 2020 08:58

twalthr requested changes Sep 14, 2020

View reviewed changes

twalthr mentioned this pull request Sep 17, 2020

[FLINK-19188][examples-table] Add a new streaming SQL example #13413

Closed

[FLINK-18988][table] Continuous query with LATERAL and LIMIT produces…

73fdc9a

… wrong result The batch mode rank only supports RANK function, so we only rewrite the stream mode query.

danny0405 force-pushed the FLINK-18988 branch from 3dff129 to 73fdc9a Compare November 17, 2020 04:32

godfreyhe reviewed Nov 17, 2020

View reviewed changes

Fix the review comments

dbbea58

godfreyhe reviewed Nov 18, 2020

View reviewed changes

Fix the review comments

6ac74e1

godfreyhe reviewed Nov 19, 2020

View reviewed changes

Fix the review comments

b02876f

godfreyhe approved these changes Nov 23, 2020

View reviewed changes

twalthr closed this in a3320d1 Nov 24, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-18988][table] Continuous query with LATERAL and LIMIT produces… #13291

[FLINK-18988][table] Continuous query with LATERAL and LIMIT produces… #13291

danny0405 commented Sep 1, 2020

flinkbot commented Sep 1, 2020

flinkbot commented Sep 1, 2020 •

edited

twalthr Sep 1, 2020

danny0405 Sep 1, 2020

twalthr Sep 1, 2020

danny0405 Sep 1, 2020

twalthr Sep 1, 2020

danny0405 Sep 1, 2020

twalthr left a comment •

edited

danny0405 commented Sep 17, 2020

godfreyhe left a comment

godfreyhe Nov 17, 2020

godfreyhe Nov 17, 2020

danny0405 Nov 17, 2020

godfreyhe Nov 18, 2020

danny0405 commented Nov 17, 2020 •

edited

godfreyhe left a comment

godfreyhe Nov 18, 2020

godfreyhe Nov 18, 2020

danny0405 Nov 18, 2020

godfreyhe Nov 18, 2020

godfreyhe Nov 18, 2020

godfreyhe Nov 18, 2020

godfreyhe left a comment

godfreyhe Nov 19, 2020

danny0405 Nov 19, 2020

danny0405 commented Nov 20, 2020

godfreyhe left a comment

twalthr commented Nov 24, 2020

		val oriBuilder = call.builder()
		val builder = FlinkRelBuilder.of(oriBuilder.getCluster, oriBuilder.getRelOptSchema)

[FLINK-18988][table] Continuous query with LATERAL and LIMIT produces… #13291

[FLINK-18988][table] Continuous query with LATERAL and LIMIT produces… #13291

Conversation

danny0405 commented Sep 1, 2020

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

flinkbot commented Sep 1, 2020

Automated Checks

Review Progress

flinkbot commented Sep 1, 2020 • edited

CI report:

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

twalthr left a comment • edited

Choose a reason for hiding this comment

danny0405 commented Sep 17, 2020

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

danny0405 commented Nov 17, 2020 • edited

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

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

danny0405 commented Nov 20, 2020

godfreyhe left a comment

Choose a reason for hiding this comment

twalthr commented Nov 24, 2020

flinkbot commented Sep 1, 2020 •

edited

twalthr left a comment •

edited

danny0405 commented Nov 17, 2020 •

edited