[Flink 16024][Connector][JDBC] Support FilterPushdown #20140

qingwei91 · 2022-07-03T08:45:39Z

What is the purpose of the change

Implement Filter Pushdown in JDBC Connector Source, this is an optimization that will avoid scanning the whole table into Flink when we only need a subset of the table.

Brief change log

Implement a SQL Visitor that converts Flink SQL AST to JDBC SQL String
Implement FilterPushdown in JDBCDynamicSource using the Visitor
Add tests for both Visitor and JDBCDynamicSource

Verifying this change

This change added tests and can be verified as follows:

Added Integrated Test in JdbcDynamicTableSourceITCase to make sure there is no regression on SQL Filtering
Added JdbcFilterPushdownVisitorTest to verify we are generating SQL correctly

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): (yes)
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, 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 · 2022-07-03T08:49:21Z

CI report:

4780c64 UNKNOWN
c7bab56 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

qingwei91 · 2022-07-03T08:46:59Z

.../flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/dialect/JdbcDialect.java

+     *
+     * @return {@link JdbcFilterPushdownVisitor}
+     */
+    default JdbcFilterPushdownVisitor getFilterPushdownVisitor() {


This should be dialect specific so that it can be specialized to specific SQL syntax.

I noticed we removed all default implementation from this interface, should I do the same for this method?

qingwei91 · 2022-07-03T08:47:45Z

...nnector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSource.java


 /** A {@link DynamicTableSource} for JDBC. */
 @Internal
 public class JdbcDynamicTableSource
        implements ScanTableSource,
                LookupTableSource,
                SupportsProjectionPushDown,
-                SupportsLimitPushDown {
+                SupportsLimitPushDown,
+                SupportsFilterPushDown {


Support Filter Pushdown in JDBC source

qingwei91 · 2022-07-03T08:49:48Z

...nnector-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcDynamicTableSource.java

+     * then be handled in Flink runtime.
+     */
+    @Override
+    public Result applyFilters(List<ResolvedExpression> filters) {


Core implementation of this change:

We traverse the ResolvedExpression, and produce a String if we know how to push it to SQL Database, returning None if we cannot handle it.

The unhandled expression will be kept as Flink SQL and run in the job

MartijnVisser · 2022-07-05T11:12:52Z

@qingwei91 Thanks for the PR. Can you make sure that your PR title and especially your commit messages are confirming to the Code Contribution guide? https://flink.apache.org/contributing/contribute-code.html

For example, your commit message should start with [FLINK-16024][Connector][JDBC] in this case.

qingwei91 · 2022-07-13T16:28:08Z

@flinkbot run azure

qingwei91 · 2022-07-13T18:05:09Z

@flinkbot run azure

qingwei91 · 2022-07-14T07:08:59Z

@flinkbot run azure

JingGe

Thanks @qingwei91 for driving this PR.

JingGe · 2022-07-15T06:46:45Z

...ctor-jdbc/src/main/java/org/apache/flink/connector/jdbc/table/JdbcFilterPushdownVisitor.java

@@ -53,7 +52,6 @@
 * Visitor to walk a Expression AST. Produces a String that can be used to pushdown the filter,
 * return Optional.empty() if we cannot pushdown the filter.
 */
-@Public


Changing public interface is not allowed. Please check FLIP-196: Source API stability guarantees for more information.

Hi @JingGe, thanks for the comment.

This is a file I newly added, and in my latest commit, I marked it as PublicEvolving: 6c97da0

Do you mean we cannot expose new API?
For instance, I added this here: https://github.com/apache/flink/pull/20140/files#diff-ae60653ffe2ac890a3c1b01da41405bcc4e6913949176c36edc009df5090c38fR157, which adds a new method and a new return type to JdbcDialect which is PublicEvolving, because filter pushdown might differ across JDBC dialect. Is this approach a problem?

Hi @JingGe did you get a chance to check on this?

Do you mean we cannot expose new API?

You can definitely add a new API to the master branch, but the reason why it's problematic here is that you've changed the status of the API in your different commits. That means that when a reviewer is checking individual commits, it shows that you've removed @Public. In a later commit, you've again added @PublicEvolving. It's better to squash your commits in this case to avoid this type of confusion.

Hi @qingwei91 Could you please explain why a public API is required? Does @internal work too? Thanks.

Hi @JingGe , I believe it has to be PublicEvolving because this class is returned by a public method in JdbcDialect, which is a PublicEvolving class. Here's an example where its failing if I mark it as Internal: https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=38493&view=logs&j=fc5181b0-e452-5c8f-68de-1097947f6483&t=995c650b-6573-581c-9ce6-7ad4cc038461

I will sort out the commit, sorry for the confusion

I've squash the commits now and we have a green build.

Can anyone of you review this again? Thank you!

MartijnVisser · 2022-07-29T09:36:22Z

@qingwei91 I can't review this since this is not my expertise. I would like to ask @hailin0 to help with a review due to #20304 - Else I would like to ask @leonardBang if he can help out

libenchao · 2022-08-08T03:55:08Z

@qingwei91 Thanks for your contribution. For the current design, I see you use ValueLiteralExpression#toString() to generate the string for literals. This may work for some cases, but not for all cases IMHO.
Consider following cases:

ValueLiteralExpression#toString() uses Flink dialect to represent string, and it hard-coded the quote with '. However, in many DBMS, ' is not the only choice.
ValueLiteralExpression#toString() only handles special character ' using escape. In many DBMS, they need more special character handing, e.g. mysql driver
For other types, e.g. TIMESTAMP, DATE, TIME, INTERVAL and so on, they may suffers from this too, because we cannot assume that all DB dialects handle them in the same way.

Another more general way to handle this is to use PrepareStatement.setXXX just like we already did in TableSimpleStatementExecutor and JdbcRowDataLookupFunction. WDYT?

qingwei91 · 2022-08-08T20:47:04Z

Hi @libenchao , thanks for the review!

Thanks for pointing out the flaw, you're right. 🙇

On your recommended approach, I believe JdbcFilterPushdownVisitor needs to produce strings or some equivalent data, so that it can be used by JdbcDynamicTableSource::getScanRuntimeProvider, how should I make use of PrepareStatement here? Maybe I am missing something?

I think you're pointing out a fundamental issue with this PR, SQL statement generation has to be dialect-specific, and me trying to provide a default implementation might be a lost cause here.

If we cannot go down the prepared statement route, I can think of 2 ideas:

Implement dialect-specific method that converts ValueLiteralExpression to SQL string literal, and have JdbcFilterPushdownVisitor to make use of it. We don't have to support all dialect in 1 go, and we can simply fallback to in-memory filter when its not supported.
Let JdbcFilterPushdownVisitor be dialect-specific, then every implementation will need to deal with dialect specific thing including literal stringification. This is similar to current approach, the only difference is that currently this PR provides a default implementation, where this option will stop doing it. My implementation is tested and used in Production for SQL Server. Maybe I can rename it and make it specific to SQL Server jdbc dialect? (SQL Server dialect is still opened [Flink-14101][Connectors][Jdbc] SQL Server dialect #20235), so this will have a dependency to it. Likewise, we can fallback to in-memory filtering when dialect without an implementation.

I think option 1 is less code, but probably more fiddly and easier to break. Option 2 is likely gonna be more code, but the separation is cleaner and less likely to break.

Let me know your thoughts. 😄

libenchao · 2022-08-09T02:32:13Z

I believe JdbcFilterPushdownVisitor needs to produce strings or some equivalent data, so that it can be used by JdbcDynamicTableSource::getScanRuntimeProvider, how should I make use of PrepareStatement here? Maybe I am missing something?

@qingwei91 Currently JdbcRowDataInputFormat already uses PreparedStatement, and 'scan.partition' is implemented using it, hence we can do it. JdbcFilterPushdownVisitor does not necessarily need to return String, it can return anything we need.
We may need to rethink about the design, especially for the following points:

What functions should we support, e.g. IN, BETWEEN?
Are these functions supported in all JDBC dialects, and do they use the same operator? If not, how should we abstract it out to make it dialect specific?

qingwei91 · 2022-08-09T08:51:26Z

Hi @libenchao, oh I see, I didnt spot that JdbcRowDataInputFormat is using PreparedStatement under the hood.

Thanks for pointing it out, this will be larger change than I expect, I will give it a go.

What functions should we support, e.g. IN, BETWEEN?

I think we can tackle this incrementally? I believe IN is supported out of the box, because Flink compile IN into multiple X=Y condition chained together by OR, I never looked into BETWEEN though

Are these functions supported in all JDBC dialects, and do they use the same operator? If not, how should we abstract it out to make it dialect specific?

I think ultimate we need to allow dialect specificness, right now I design it such that the query generator (ie. JdbcFilterPushdownVisitor) is part of JDBCDialect, so each dialect can provide their own instance to deal with it. Do you think this design is okay? Or is there a better way?

libenchao · 2022-08-09T09:09:11Z

I believe IN is supported out of the box, because Flink compile IN into multiple X=Y condition chained together by OR

Not exactly, we have a threshold (default 4):

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

Line 47 in 208f08b

private val THRESHOLD: Int = 4

I think we can tackle this incrementally?

I agree, we can start from some common functions, such as =, <>, <, >, <=, >=, IS NULL, IS NOT NULL, IN, and leave others to the future.

HeChuanXUPT · 2022-09-06T06:10:12Z

Wow, Thanks, It works for me~

Implement Filter Pushdown for JDBC connector source using expression visitor pattern.

qingwei91 · 2022-10-23T20:02:51Z

@flinkbot run azure

qingwei91 · 2022-10-24T18:52:03Z

Hi @libenchao , thank you very much for your review 👍

I've addressed all of your concern. On the TablePlanTest, do you mind to check if that's how it supposed to work? I don't think I understand internal good enough to judge

libenchao

One small tip, do not rebase/force-pushing before the reviewer asks you because this will make the reviewer hard to do incremental review.

libenchao · 2022-10-25T01:52:43Z

...onnector-jdbc/src/test/resources/org/apache/flink/connector/jdbc/table/JdbcTablePlanTest.xml

+	</Resource>
+	<Resource name="optimized exec plan">
+		<![CDATA[
+Calc(select=[id, time_col, real_col], where=[((time_col <> '11:11:11.000111') OR (double_col >= -1000.23))])


why the condition still exists in the Calc?

I figure out, its because the column type Time wasn't supported, I've added support here: 74fe5ee

Thanks for calling it out.

qingwei91 · 2022-10-28T21:27:15Z

One small tip, do not rebase/force-pushing before the reviewer asks you because this will make the reviewer hard to do incremental review.

Sorry, my bad, I was advised to squashed in prev PR, but of course that should only be done right before merging.

qingwei91 · 2022-10-28T21:27:38Z

@flinkbot run azure

libenchao · 2022-10-30T09:38:16Z

...n/java/org/apache/flink/connector/jdbc/table/JdbcFilterPushdownPreparedStatementVisitor.java

+        LogicalType tpe = litExp.getOutputDataType().getLogicalType();
+        Class<?> typeCs = tpe.getClass();
+
+        if (SUPPORTED_DATA_TYPES.contains(typeCs)) {


How about change this to something like:

switch(tpe.getTypeRoot()) { case INTEGER: ... case VARCHAR: ... ... default: return Optional.empty(); }

qingwei91 · 2022-10-30T14:27:48Z

@flinkbot run azure

qingwei91 · 2022-10-30T21:56:56Z

@flinkbot run azure

libenchao

@qingwei91 Thanks for the update. Please do not use 'squash' or 'force-push' unless you must or the reviewer asks you. (I go through all the codes again, and left several minor comment)

libenchao · 2022-10-31T04:57:44Z

...n/java/org/apache/flink/connector/jdbc/table/JdbcFilterPushdownPreparedStatementVisitor.java

+
+    @Override
+    public Optional<ParameterizedPredicate> visit(FieldReferenceExpression fieldReference) {
+        String predicateStr = (this.quoteIdentifierFunction.apply(fieldReference.toString()));


useless ( )

...n/java/org/apache/flink/connector/jdbc/table/JdbcFilterPushdownPreparedStatementVisitor.java

libenchao · 2022-10-31T05:11:03Z

...n/java/org/apache/flink/connector/jdbc/table/JdbcFilterPushdownPreparedStatementVisitor.java

+public class JdbcFilterPushdownPreparedStatementVisitor
+        extends ExpressionDefaultVisitor<Optional<ParameterizedPredicate>> {
+
+    private Function<String, String> quoteIdentifierFunction;


qingwei91 · 2022-10-31T18:44:56Z

@libenchao

Thanks for the update. Please do not use 'squash' or 'force-push' unless you must or the reviewer asks you. (I go through all the codes again, and left several minor comment)

Sorry, in the last commit, I thought to bundle all changes I've made since your last review into 55d5227, so that you can review just that without noise. I wont do it again.

I will go through the comments and fix them

Add test for IS NULL and IS NOT NULL

qingwei91 · 2022-11-01T18:40:41Z

Hi @libenchao , this is the new commit I added to address your comment.

42078a4

I also added support IS NULL and IS NOT NULL as these 2 are quite common

libenchao

I also added support IS NULL and IS NOT NULL as these 2 are quite common

This is great!
The PR LGTM with only one minor comment, after fixing that, I'll merge this.

...n/java/org/apache/flink/connector/jdbc/table/JdbcFilterPushdownPreparedStatementVisitor.java

…/flink/connector/jdbc/table/JdbcFilterPushdownPreparedStatementVisitor.java Prefer primitive Co-authored-by: Benchao Li <libenchao@gmail.com>

libenchao

LGTM, merging. @qingwei91 Thanks for your contribution and the patience and consistence during the review.

This closes apache#20140

grzegorz8 · 2023-02-21T19:55:20Z

...n/java/org/apache/flink/connector/jdbc/table/JdbcFilterPushdownPreparedStatementVisitor.java

+    }
+
+    @Override
+    public Optional<ParameterizedPredicate> visit(CallExpression call) {


@qingwei91 - Is there any reason why LIKE was not added here? It seems to be as easy as other binary operators below.
I will gladly add LIKE but I just want to ensure I didn't overlook something tricky.

Hi @grzegorz8 , I think we simply missed it, thanks for noticing this.

qingwei91 commented Jul 3, 2022

View reviewed changes

This was referenced Jul 3, 2022

FLINK-16024 [jdbc-connector] JDBC Filter Pushdown #20138

Closed

FLINK-16024 [jdbc-connector] JDBC Filter Pushdown #20136

Closed

qingwei91 force-pushed the FLINK-16024/1.16-jdbc-filter branch from 3672733 to 16a7978 Compare July 5, 2022 17:52

zhuyufeng0809 mentioned this pull request Jul 6, 2022

[FLINK-19651][Connectors/JDBC] Support the filter push down for the JDBC connector #20126

Closed

qingwei91 changed the title ~~Flink 16024/1.16 jdbc filter~~ [Flink 16024][Connector][JDBC] Support FilterPushdown Jul 10, 2022

qingwei91 force-pushed the FLINK-16024/1.16-jdbc-filter branch from 8ed32ca to 6c97da0 Compare July 13, 2022 19:41

JingGe requested changes Jul 15, 2022

View reviewed changes

MartijnVisser mentioned this pull request Jul 20, 2022

[FLINK-28600][Connector][JDBC] Support FilterPushDown in flink-connector-jdbc #20304

Closed

qingwei91 force-pushed the FLINK-16024/1.16-jdbc-filter branch 3 times, most recently from 57590d6 to d08882b Compare July 20, 2022 20:45

qingwei91 requested a review from MartijnVisser July 28, 2022 18:00

MartijnVisser removed their request for review July 29, 2022 09:36

qingwei91 force-pushed the FLINK-16024/1.16-jdbc-filter branch 4 times, most recently from cf84956 to f5516a5 Compare October 2, 2022 18:24

[FLINK-16024][Connector][JDBC] Filter Pushdown

cd71a47

Implement Filter Pushdown for JDBC connector source using expression visitor pattern.

qingwei91 force-pushed the FLINK-16024/1.16-jdbc-filter branch from db4b708 to cd71a47 Compare October 23, 2022 19:31

qingwei91 requested a review from libenchao October 24, 2022 18:50

libenchao reviewed Oct 25, 2022

View reviewed changes

Support TimeType in filter pushdown

74fe5ee

qingwei91 requested a review from libenchao October 29, 2022 08:36

libenchao reviewed Oct 30, 2022

View reviewed changes

qingwei91 force-pushed the FLINK-16024/1.16-jdbc-filter branch from d9b1457 to dd54aed Compare October 30, 2022 11:18

Change ifelse to switch and drop useless variable

55d5227

qingwei91 force-pushed the FLINK-16024/1.16-jdbc-filter branch from dd54aed to 55d5227 Compare October 30, 2022 18:01

libenchao reviewed Oct 31, 2022

View reviewed changes

Support more data type and operator

42078a4

Add test for IS NULL and IS NOT NULL

qingwei91 requested a review from libenchao November 1, 2022 18:39

libenchao reviewed Nov 2, 2022

View reviewed changes

...n/java/org/apache/flink/connector/jdbc/table/JdbcFilterPushdownPreparedStatementVisitor.java Outdated Show resolved Hide resolved

Update flink-connectors/flink-connector-jdbc/src/main/java/org/apache…

c7bab56

…/flink/connector/jdbc/table/JdbcFilterPushdownPreparedStatementVisitor.java Prefer primitive Co-authored-by: Benchao Li <libenchao@gmail.com>

libenchao approved these changes Nov 2, 2022

View reviewed changes

libenchao closed this in 4967001 Nov 2, 2022

huangxiaofeng10047 pushed a commit to huangxiaofeng10047/flink that referenced this pull request Nov 3, 2022

[FLINK-16024][connector/jdbc] Support filter pushdown

5e80382

This closes apache#20140

dchristle pushed a commit to dchristle/flink that referenced this pull request Nov 18, 2022

[FLINK-16024][connector/jdbc] Support filter pushdown

4fbb408

This closes apache#20140

akkinenivijay pushed a commit to krisnaru/flink that referenced this pull request Feb 11, 2023

[FLINK-16024][connector/jdbc] Support filter pushdown

63154fc

This closes apache#20140

grzegorz8 reviewed Feb 21, 2023

View reviewed changes

[Flink 16024][Connector][JDBC] Support FilterPushdown #20140

[Flink 16024][Connector][JDBC] Support FilterPushdown #20140

Conversation

qingwei91 commented Jul 3, 2022 • edited

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 Jul 3, 2022 • edited

CI report:

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MartijnVisser commented Jul 5, 2022

qingwei91 commented Jul 13, 2022

qingwei91 commented Jul 13, 2022

qingwei91 commented Jul 14, 2022

JingGe left a comment

Choose a reason for hiding this comment

JingGe Jul 15, 2022 • 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

JingGe Jul 19, 2022 • edited

Choose a reason for hiding this comment

qingwei91 Jul 20, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MartijnVisser commented Jul 29, 2022

libenchao commented Aug 8, 2022

qingwei91 commented Aug 8, 2022

libenchao commented Aug 9, 2022

qingwei91 commented Aug 9, 2022

libenchao commented Aug 9, 2022

HeChuanXUPT commented Sep 6, 2022

qingwei91 commented Oct 23, 2022

qingwei91 commented Oct 24, 2022

libenchao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

qingwei91 commented Oct 28, 2022

qingwei91 commented Oct 28, 2022

Choose a reason for hiding this comment

qingwei91 commented Oct 30, 2022

qingwei91 commented Oct 30, 2022

libenchao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

qingwei91 commented Oct 31, 2022

qingwei91 commented Nov 1, 2022 • edited

libenchao left a comment

Choose a reason for hiding this comment

libenchao left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

qingwei91 commented Jul 3, 2022 •

edited

flinkbot commented Jul 3, 2022 •

edited

JingGe Jul 15, 2022 •

edited

JingGe Jul 19, 2022 •

edited

qingwei91 Jul 20, 2022 •

edited

qingwei91 commented Nov 1, 2022 •

edited