[FLINK-17425][blink-planner] supportsFilterPushDown rule in DynamicSource. #12866

liuyongvs · 2020-07-10T08:54:00Z

What is the purpose of the change

make the DynamicSource supports FilterPushDown Rule

Verifying this change

This change added tests and can be verified as follows:

Added PushFilterIntoTableSourceScanRuleTest to verify the plan
Extended TableSourceITCase (batch and stream )to verify the result filter projection

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)

flinkbot · 2020-07-10T08:56:12Z

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 dcdd7b7 (Fri Jul 10 08:56:12 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

liuyongvs · 2020-07-10T08:57:38Z

@godfreyhe ,because of my careless operation, i drop my github repo. So i open a open PR. And the third commit is modification for all your reviews.

flinkbot · 2020-07-10T09:16:57Z

CI report:

1ca506e UNKNOWN
9942ebd 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

liuyongvs · 2020-07-14T05:49:36Z

hi @godfreyhe, you can review it now. The blink planner tests passed.

liuyongvs · 2020-07-14T06:50:42Z

PR before #12851
Add this for tracking.

godfreyhe · 2020-07-14T08:15:59Z

...ava/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRule.java

+
+ LogicalTableScan scan = call.rel(1);
+ TableSourceTable tableSourceTable = scan.getTable().unwrap(TableSourceTable.class);
+ //we can not push filter twice


nit: add a space after //

godfreyhe

Thanks for the update, I left some comments.
There are some comments in #12851 are not been addressed, WDYT?

godfreyhe · 2020-07-14T08:18:44Z

...ava/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRule.java

+ }),
+ context.getCatalogManager().getDataTypeFactory())
+ .build();
+ SupportsFilterPushDown.Result result = ((SupportsFilterPushDown) newTableSource).applyFilters(resolver.resolve(remainingPredicates));


nit: wrap the line, it's too long

godfreyhe · 2020-07-14T08:20:49Z

...ava/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRule.java

+ getNewFlinkStatistic(oldTableSourceTable, originPredicatesSize, updatedPredicatesSize),
+ getNewExtraDigests(oldTableSourceTable, result.getAcceptedFilters())
+ );
+ TableScan newScan = new LogicalTableScan(scan.getCluster(), scan.getTraitSet(), scan.getHints(), newTableSourceTable);


use LogicalTableScan.create instead of new LogicalTableScan

i want to know why?

because the traits of Scan may be changed after filter push down, It's better the traits could be recomputed which is done in LogicalTableScan#create method. Currently, new LogicalTableScan is ok, but I think using LogicalTableScan#create is better.

godfreyhe · 2020-07-14T08:26:39Z

...ava/org/apache/flink/table/planner/plan/rules/logical/PushFilterIntoTableSourceScanRule.java

+ //we can not push filter twice
+ return tableSourceTable != null
+ && tableSourceTable.tableSource() instanceof SupportsFilterPushDown
+ && !Arrays.stream(tableSourceTable.extraDigests()).anyMatch(str -> str.contains("filter"));


what if a table contain a field named filter ?
use str.startWith("filter=[")

liuyongvs · 2020-07-15T02:20:12Z

Hi @godfreyhe . Yes, i find some type is not Comparable, such as Period int getValueAs(Class clazz).
Thanks for you review in great details. And I learned a lot

godfreyhe · 2020-07-15T03:27:44Z

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

+ Object lhsValue = getValue(children.get(0), row);
+ Object rhsValue = getValue(children.get(1), row);
+ // validate that literal is comparable
+ if (!isComparable(lhsValue, binExpr) || !isComparable(rhsValue, binExpr)) {


how about check whether the type of a field is comparable in shouldPushDownUnaryExpression method ? then the logic of binaryFilterApplies need not to change

godfreyhe · 2020-07-16T10:03:00Z

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

+
+ if (expr instanceof ValueLiteralExpression) {
+ // validate that literal is comparable
+ Optional value = ((ValueLiteralExpression) expr).getValueAs(((ValueLiteralExpression) expr).getOutputDataType().getConversionClass());


we should also consider whether the type of FieldReferenceExpression is comparable, consider the following pattern:
the filterableFields is a and b and the pushed pattern a > b

godfreyhe · 2020-07-17T01:40:02Z

thanks for the update. LGTM overall, last comment: please remove the import: import static org.apache.flink.runtime.state.CheckpointStreamWithResultProvider.LOG; in TestValuesTableFactory

godfreyhe · 2020-07-17T05:38:51Z

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

@@ -565,14 +562,12 @@ private boolean binaryFilterApplies(CallExpression binExpr, Row row) {
 }
 }

- private boolean isComparable(Class<?> clazz) {
- // validate that literal is comparable
+ private void validateTypeComparable(Class<?> clazz) {


I think this method should be

private boolean isComparable(Class<?> clazz) { return Comparable.class.isAssignableFrom(clazz); }

we do not need throw exception if a class is non-comparable.TestValuesTableFactory just does not support non-comparable type. just like the case that TestValuesTableFactory just supports UPPER and LOWER UDF.

@godfreyhe , But need it log some warning info here?

I don't think so. We can add some java doc at the class level to explain which patterns this class supports

good idea +1

…urce.

…PushDown logic.

godfreyhe · 2020-07-17T06:23:35Z

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

- * Test implementation of {@link DynamicTableSourceFactory} that creates
- * a source that produces a sequence of values.
+ * Test implementation of {@link DynamicTableSourceFactory} that creates a source that produces a sequence of values.
+ * And this source {@link TestValuesTableSource} supports FilterPushDown. And it has some limitations.


And {@link TestValuesTableSource} can push down filter into table source. A predicate can be pushed down only if it satisfies the following conditions:

field name is in filterable-fields which is defined in with properties

the field type is comparable

UDF is UPPER or LOWER

godfreyhe

Thanks for the update, LGTM. cc @wuchong

liuyongvs · 2020-07-21T01:31:56Z

@godfreyhe @wuchong , could it merge to master . The error is test_rocksdb_state_memory_control, which it is not relevant to my commits.

wuchong

Sorry for the late response. LGTM.

I will merge this once the build is passed in my repo.

…ScanTableSource in planner This closes apache#12866

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

liuyongvs mentioned this pull request Jul 10, 2020

[FLINK-17425][blink-planner] supportsFilterPushDown rule in DynamicSource. #12851

Closed

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

liuyongvs force-pushed the filter_pushdown branch from b745ddd to 33ee4b7 Compare July 13, 2020 08:51

godfreyhe reviewed Jul 14, 2020

View reviewed changes

godfreyhe reviewed Jul 15, 2020

View reviewed changes

liuyongvs force-pushed the filter_pushdown branch from 21ff7c2 to b9ae3fb Compare July 16, 2020 05:50

godfreyhe reviewed Jul 16, 2020

View reviewed changes

liuyongvs force-pushed the filter_pushdown branch from 77a0241 to a78a2f7 Compare July 17, 2020 01:58

godfreyhe reviewed Jul 17, 2020

View reviewed changes

liuyongvs added 12 commits July 17, 2020 14:08

[FLINK-17425][blink-planner] supportsFilterPushDown rule in DynamicSo…

c311bf0

…urce.

[FLINK-17426][blink-planner] fix TableScanTest conflicts with master.

e68d9f6

[FLINK-17425][blink-planner] fixs problems of review.

727bca6

[FLINK-17425][blink-planner] fix bug and comflicts with master.

2887155

[FLINK-17425][blink-planner] fix bug.

f692615

[FLINK-17425][blink-planner]fix problem of reviews.

6ea5ce8

[FLINK-17425][blink-planner] fix review of some type is not comparable.

02f8b28

[FLINK-17425][blink-planner] put the checking of comparable at should…

c053da5

…PushDown logic.

[FLINK-17425][blink-planner] verify the clazz is comparable.

8523a56

[FLINK-17425][blink-planner] fix bug.

75dacdd

[FLINK-17425][blink-planner] replace log by RuntimeException.

bb091e2

[FLINK-17425][blink-planner] add some comments.

1ca506e

liuyongvs force-pushed the filter_pushdown branch from a78a2f7 to 1ca506e Compare July 17, 2020 06:09

godfreyhe reviewed Jul 17, 2020

View reviewed changes

[FLINK-17425][blink-planner] modify comments.

9942ebd

godfreyhe approved these changes Jul 17, 2020

View reviewed changes

wuchong approved these changes Jul 21, 2020

View reviewed changes

wuchong pushed a commit to wuchong/flink that referenced this pull request Jul 21, 2020

[FLINK-17425][table-planner-blink] Support SupportsFilterPushDown on …

f65b687

…ScanTableSource in planner This closes apache#12866

wuchong closed this in fe0d001 Jul 22, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-17425][blink-planner] supportsFilterPushDown rule in DynamicSource. #12866

[FLINK-17425][blink-planner] supportsFilterPushDown rule in DynamicSource. #12866

liuyongvs commented Jul 10, 2020 •

edited

Loading

flinkbot commented Jul 10, 2020

liuyongvs commented Jul 10, 2020

flinkbot commented Jul 10, 2020 •

edited

Loading

liuyongvs commented Jul 14, 2020

liuyongvs commented Jul 14, 2020 •

edited

Loading

godfreyhe Jul 14, 2020

godfreyhe left a comment

godfreyhe Jul 14, 2020

liuyongvs Jul 14, 2020

godfreyhe Jul 14, 2020

liuyongvs Jul 14, 2020

godfreyhe Jul 14, 2020

godfreyhe Jul 14, 2020

liuyongvs commented Jul 15, 2020

godfreyhe Jul 15, 2020

liuyongvs Jul 17, 2020

godfreyhe Jul 16, 2020

godfreyhe commented Jul 17, 2020

godfreyhe Jul 17, 2020

liuyongvs Jul 17, 2020

godfreyhe Jul 17, 2020

liuyongvs Jul 17, 2020

godfreyhe Jul 17, 2020

liuyongvs Jul 17, 2020

godfreyhe left a comment

liuyongvs commented Jul 21, 2020

wuchong left a comment

[FLINK-17425][blink-planner] supportsFilterPushDown rule in DynamicSource. #12866

[FLINK-17425][blink-planner] supportsFilterPushDown rule in DynamicSource. #12866

Conversation

liuyongvs commented Jul 10, 2020 • edited Loading

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 10, 2020

Automated Checks

Review Progress

liuyongvs commented Jul 10, 2020

flinkbot commented Jul 10, 2020 • edited Loading

CI report:

liuyongvs commented Jul 14, 2020

liuyongvs commented Jul 14, 2020 • edited Loading

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

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 commented Jul 15, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

godfreyhe commented Jul 17, 2020

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

liuyongvs commented Jul 21, 2020

wuchong left a comment

Choose a reason for hiding this comment

liuyongvs commented Jul 10, 2020 •

edited

Loading

flinkbot commented Jul 10, 2020 •

edited

Loading

liuyongvs commented Jul 14, 2020 •

edited

Loading