[FLINK-17426][blink planner] Dynamic Source supportsLimit pushdown #12964

liuyongvs · 2020-07-23T02:20:05Z

What is the purpose of the change

make the DynamicSource supports LimitPushDown Rule

Verifying this change

This change added tests and can be verified as follows:

Added LimitTest to verify the plan
Extended LimitITCase (only batch)to verify the result limit projection
make the TestValueSource supports LimitPushDown

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? (yes)
If yes, how is the feature documented? (JavaDocs)

…Down.

flinkbot · 2020-07-23T02:22:57Z

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 984b8a1 (Thu Jul 23 02:22:57 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-07-23T02:50:18Z

CI report:

34dd37d Azure: SUCCESS

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

liuyongvs · 2020-07-23T07:04:39Z

Hi @godfreyhe @wuchong ,it is here, thanks for your review.

godfreyhe

Thanks for the contribution, I left some comments

godfreyhe · 2020-07-24T06:28:44Z

...java/org/apache/flink/table/planner/plan/rules/logical/PushLimitIntoTableSourceScanRule.java

+		return onlyLimit
+			&& tableSourceTable != null
+			&& tableSourceTable.tableSource() instanceof SupportsLimitPushDown
+			&& !Arrays.stream(tableSourceTable.extraDigests()).anyMatch(str -> str.startsWith("limit=["));


nit: simpler way: use noneMatch instead of ! + anyMatch.
btw, it's better we can also update the matches method of PushFilterIntoTableSourceScanRule

godfreyhe · 2020-07-24T06:34:37Z

...java/org/apache/flink/table/planner/plan/rules/logical/PushLimitIntoTableSourceScanRule.java

+
+		// update extraDigests
+		String[] newExtraDigests = new String[0];
+		if (limit > 0) {


we should update the digest anyway, otherwise the rule will be applied endless loop if limit is 0.

But it will look strange, like this limit=[0] And this rule FlinkLimit0RemoveRule will remove it.

We can't rely on other rules to ensure the correctness of this rule, we must make sure each rule itself is correct.

godfreyhe · 2020-07-24T06:36:49Z

...java/org/apache/flink/table/planner/plan/rules/logical/PushLimitIntoTableSourceScanRule.java

+		TableSourceTable newTableSourceTable = applyLimit(limit, tableSourceTable);
+
+		FlinkLogicalTableSourceScan newScan = FlinkLogicalTableSourceScan.create(scan.getCluster(), newTableSourceTable);
+		call.transformTo(sort.copy(sort.getTraitSet(), Collections.singletonList(newScan)));


It's better we can put sort.copy(sort.getTraitSet(), Collections.singletonList(newScan)) in a single line, which could make debugging more easy.

godfreyhe · 2020-07-24T06:40:42Z

...ner-blink/src/test/java/org/apache/flink/table/planner/factories/TestValuesTableFactory.java

@@ -359,7 +361,7 @@ private ChangelogMode parseChangelogMode(String string) {
 	/**
 	 * Values {@link DynamicTableSource} for testing.
 	 */
-	private static class TestValuesTableSource implements ScanTableSource, LookupTableSource, SupportsProjectionPushDown, SupportsFilterPushDown {
+	private static class TestValuesTableSource implements ScanTableSource, LookupTableSource, SupportsProjectionPushDown, SupportsFilterPushDown, SupportsLimitPushDown {


this line is too long...

godfreyhe · 2020-07-24T06:48:25Z

...java/org/apache/flink/table/planner/plan/rules/logical/PushLimitIntoTableSourceScanRule.java

+ * implement.
+ * 3.We can support limit with offset, we can push down offset + fetch to table source.
+ */
+public class PushLimitIntoTableSourceScanRule extends RelOptRule {


please add a rule test to verify this rule， just like PushFilterIntoTableSourceScanRuleTest

the unit test is LimitTest.

Rule test is different from plan test. Rule test focuses on the new rule you implement and only a few must-involved rules can be added to the rule set to help the test. Plan test will verify the plan based on whole rule sets.

godfreyhe

Thanks for the update, LGTM overall, I left some minor comments.
btw, please fix the typo of title: plannger -> planner

godfreyhe · 2020-07-28T09:16:37Z

...java/org/apache/flink/table/planner/plan/rules/logical/PushLimitIntoTableSourceScanRule.java

+
+		// update extraDigests
+		String[] newExtraDigests = new String[0];
+		if (limit >= 0) {


how about we remove the if ?

I think it can not be removed. Although, it doesn't have problem now. But if the calcite supports this syntax limit x,y, which is the mysql limit offset syntax. y represents limit, which can be negative, such as -1 , you can refer this https://www.cnblogs.com/acm-bingzi/p/msqlLimit.html.
And you can also read the code of calcite FlinkSqlParserImpl.OrderedQueryOrExpr, where has some comments.
That is the reason why i add this test testMysqlLimit.

why not we add limitation limit >= 0 in matches method ?
btw, LIMIT x,y can be expressed as LIMIT y OFFSET x in sql standard.

Yeap, calcite fetch member is null when it doesn't have limit currently. And calcite doesn't support mysql limit x,y syntax. If it supports, limit 5, -1. The -1 represents the end. That is to say [offset5, end).
So the calcite may transform -1 to fetch null when it supports in parser. And we don't need to limit >=0.
So i just do as you say, remove the limit >= 0 and don't add limitation in matches.

godfreyhe · 2020-07-28T09:19:38Z

...apache/flink/table/planner/plan/rules/logical/PushLimitIntoLegacyTableSourceScanRuleTest.xml

+			<![CDATA[
+LogicalProject(a=[$0], b=[$1], c=[$2])
+- FlinkLogicalSort(fetch=[5])
+   +- FlinkLogicalLegacyTableSourceScan(table=[[default_catalog, default_database, LimitTable, source: [limit: 5]]], fields=[a, b, c])


why the digest pattern is not limit=5 ?

This is PushLimitIntoLegacyTableSourceScanRuleTest, which is legacy Rule.And read the code of TestLegacyLimitableTableSource.explainSource. You can compare the PushProjectIntoLegacyTableSourceScanRuleTest and PushProjectIntoTableSourceScanRuleTest. And there are the same.
new: LogicalTableScan(table=[[default_catalog, default_database, MyTable, project=[a, c]]])
Legacy: LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestSource(physical fields: a, c)]]])

sorry, this comment should be at the line 129 in PushLimitIntoTableSourceScanRuleTest.xml.

godfreyhe · 2020-07-28T09:23:54Z

...ache/flink/table/planner/plan/rules/logical/PushLimitIntoLegacyTableSourceScanRuleTest.scala

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


nit: unify the case of Pushdown ？

godfreyhe · 2020-07-29T07:59:11Z

...s/org/apache/flink/table/planner/plan/rules/logical/PushLimitIntoTableSourceScanRuleTest.xml

+			<![CDATA[
+LogicalProject(a=[$0], b=[$1], c=[$2])
+- FlinkLogicalSort(fetch=[5])
+   +- FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, LimitTable, source: [limit: 5]]], fields=[a, b, c])


why the digest pattern is not limit=5 ?

sorry, it should be removed. i have replace the testLimitWithoutOffset by testCanPushdownLimitWithoutOffset. And forgot to remove it.

godfreyhe

Thanks for the update, LGTM. cc @wuchong

godfreyhe · 2020-07-30T07:49:37Z

...java/org/apache/flink/table/planner/plan/rules/logical/PushLimitIntoTableSourceScanRule.java

+		TableStats newTableStats = new TableStats(newRowCount);
+		FlinkStatistic newStatistic = FlinkStatistic.builder()
+			.statistic(statistic)
+			.tableStats(newTableStats)


nit: -> .tableStats(new TableStats(newRowCount))

godfreyhe · 2020-07-30T07:50:49Z

...java/org/apache/flink/table/planner/plan/rules/logical/PushLimitIntoTableSourceScanRule.java

+		return oldTableSourceTable.copy(
+			newTableSource,
+			newStatistic,
+			newExtraDigests


nit: newExtraDigests is only used once, just move new String[] {"limit=[" + limit + "]"} into here.

wuchong

LGTM

…interface for ScanTableSource This closes apache#12964

liuyongvs added 4 commits July 22, 2020 11:50

[FLINK-17426][blink-planner] supportsLimitPushDown rule.

12d6026

[FLINK-17426][blink-planner] DynamicSource supports SupportsLimitPush…

65f49ba

…Down.

[FLINK-17426][blink-planner] fix checkstyle of line too long.

4eadef4

[FLINK-17425][blink-planner] rewrite the limit pushdown.

984b8a1

rmetzger added the review=description? label Jul 23, 2020

rmetzger added the component=TableSQL/Planner label Jul 23, 2020

godfreyhe reviewed Jul 24, 2020

View reviewed changes

[FLINK-17426][blink-planner] add unit tests.

0324c22

godfreyhe reviewed Jul 28, 2020

View reviewed changes

liuyongvs changed the title ~~[FLINK-17426][blink plannger] Dynamic Source supportsLimit pushdown~~ [FLINK-17426][blink planner] Dynamic Source supportsLimit pushdown Jul 28, 2020

[FLINK-17426][blink-planner] fix reviews to unify the unit test.

141bd8e

godfreyhe reviewed Jul 29, 2020

View reviewed changes

[FLINK-17426][blink-planner] remove the redundant content.

34dd37d

godfreyhe approved these changes Jul 30, 2020

View reviewed changes

wuchong approved these changes Jul 31, 2020

View reviewed changes

wuchong merged commit 3ab1a1c into apache:master Jul 31, 2020

twalthr pushed a commit to twalthr/flink that referenced this pull request Aug 3, 2020

[FLINK-17426][table-planner-blink] Support the SupportsLimitPushDown …

e5ab65e

…interface for ScanTableSource This closes apache#12964

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-17426][blink planner] Dynamic Source supportsLimit pushdown #12964

[FLINK-17426][blink planner] Dynamic Source supportsLimit pushdown #12964

liuyongvs commented Jul 23, 2020

flinkbot commented Jul 23, 2020

flinkbot commented Jul 23, 2020 •

edited

liuyongvs commented Jul 23, 2020

godfreyhe left a comment

godfreyhe Jul 24, 2020

liuyongvs Jul 24, 2020

godfreyhe Jul 24, 2020

liuyongvs Jul 24, 2020

godfreyhe Jul 24, 2020

godfreyhe Jul 24, 2020

liuyongvs Jul 24, 2020

godfreyhe Jul 24, 2020

liuyongvs Jul 24, 2020

godfreyhe Jul 24, 2020

liuyongvs Jul 24, 2020

godfreyhe Jul 24, 2020 •

edited

godfreyhe left a comment

godfreyhe Jul 28, 2020

liuyongvs Jul 28, 2020 •

edited

godfreyhe Jul 29, 2020

liuyongvs Jul 30, 2020

godfreyhe Jul 28, 2020

liuyongvs Jul 28, 2020 •

edited

godfreyhe Jul 29, 2020

godfreyhe Jul 28, 2020

liuyongvs Jul 28, 2020

godfreyhe Jul 29, 2020

liuyongvs Jul 30, 2020

godfreyhe left a comment

godfreyhe Jul 30, 2020

godfreyhe Jul 30, 2020

wuchong left a comment

[FLINK-17426][blink planner] Dynamic Source supportsLimit pushdown #12964

[FLINK-17426][blink planner] Dynamic Source supportsLimit pushdown #12964

Conversation

liuyongvs commented Jul 23, 2020

What is the purpose of the change

Verifying this change

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

Documentation

flinkbot commented Jul 23, 2020

Automated Checks

Review Progress

flinkbot commented Jul 23, 2020 • edited

CI report:

liuyongvs commented Jul 23, 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

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 Jul 24, 2020 • edited

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

liuyongvs Jul 28, 2020 • edited

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

liuyongvs Jul 28, 2020 • edited

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

wuchong left a comment

Choose a reason for hiding this comment

flinkbot commented Jul 23, 2020 •

edited

godfreyhe Jul 24, 2020 •

edited

liuyongvs Jul 28, 2020 •

edited

liuyongvs Jul 28, 2020 •

edited