[flink]Support partition pushdown for more filters in Flink connector #420

Alibaba-HZY · 2025-02-18T03:28:48Z

[flink]

Purpose

Support partition pushdown in Flink connector
Linked issue: close #1711

Tests

com.alibaba.fluss.connector.flink.source.FlinkTableSourceITCase#testStreamingReadPartitionPushDown
com.alibaba.fluss.predicate.PredicateBuilderTest
com.alibaba.fluss.predicate.PredicateTest

API and Format

no

Documentation

no

wuchong · 2025-02-26T12:15:11Z

@Alibaba-HZY Thanks for the contribution. We are reaching the code freeze deadline of v0.6. Therefore, I planned this issue into the next version. Besides, this pull request is quite huge. Splitting into multiple pull requests can speed up the review process.

Alibaba-HZY · 2025-03-03T03:15:52Z

@Alibaba-HZY感谢你的贡献。我们即将到达 v0.6 的代码冻结期限。因此，我将这个问题计划到了下一个版本。另外，这个 pull request 相当庞大。拆分成多个 pull request 可以加快审核速度。

ok i will submit 2pr.
first:Introduce Predicate #515
second:Support partition pushdown in Flink connector

platinumhamburg · 2025-06-24T03:02:13Z

...-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlinkTableSource.java

+                    remainingFilters.add(filter);
+                } else {
+                    Predicate p = predicateOptional.get();
+                    if (!p.visit(partitionPredicateVisitor)) {


The name PartitionPredicateVisitor is quite strange. Based on its implementation semantics and usage, perhaps it should be called PartitionPredicateMatcher? (Although I checked and found that Apache Paimon also uses this naming.)

If the class renamed PartitionPredicateMatcher, then his ’visit‘ methods should be renamed to ‘match’?

I propose a public method match() to wrap the internal visit() method as a better approach. match() serves as the semantic interface, and visit() is its implementation.

...-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlinkTableSource.java

...nk-common/src/main/java/com/alibaba/fluss/flink/source/enumerator/FlinkSourceEnumerator.java

platinumhamburg · 2025-06-30T03:28:40Z

@Alibaba-HZY It seems the pull request branch is currently far behind the main branch, and several conflicts exist. Could you kindly rebase it from the main branch when possible? This will help us proceed with more thorough testing for this PR.

platinumhamburg · 2025-07-01T02:31:19Z

@Alibaba-HZY I have tested this PR in several complex scenarios. Overall, the changes look good, but there are still a few minor issues that need attention:

When using partition filters with LIKE clauses such as "like 'xxx%'" or "like '%xxx%'", Flink throws an error: Caused by: java.io.NotSerializableException: com.alibaba.fluss.memory.MemorySegment.... This issue was straightforward to fix.
Currently, the test cases only cover a limited set of filter patterns. It would be better to add more comprehensive unit tests to cover the supported filtering scenarios.

Alibaba-HZY · 2025-07-01T02:34:15Z

@Alibaba-HZY I have tested this PR in several complex scenarios. Overall, the changes look good, but there are still a few minor issues that need attention:

When using partition filters with LIKE clauses such as "like 'xxx%'" or "like '%xxx%'", Flink throws an error: Caused by: java.io.NotSerializableException: com.alibaba.fluss.memory.MemorySegment.... This issue was straightforward to fix.

Currently, the test cases only cover a limited set of filter patterns. It would be better to add more comprehensive unit tests to cover the supported filtering scenarios.

thank you for your test, i will fix thoes problems

...-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlinkTableSource.java

platinumhamburg · 2025-07-08T12:05:23Z

.../fluss-flink-common/src/test/java/com/alibaba/fluss/flink/source/FlinkTableSourceITCase.java

    }

+    @Test
+    void testStreamingReadWithCombinedFilters2() throws Exception {


What is the difference between the test cases testStreamingReadWithCombinedFilters1 and testStreamingReadWithCombinedFilters2?

i think i can combine thoese

platinumhamburg · 2025-07-08T12:16:23Z

This PR LGTM. I've tested it across several rounds in different production scenarios, and it works almost well. Once https://github.com/apache/fluss/pull/515/files is merged, this PR should be ready for final review by @wuchong after cleaning up the commits.

platinumhamburg · 2025-07-10T02:46:17Z

...nk-common/src/main/java/com/alibaba/fluss/flink/source/enumerator/FlinkSourceEnumerator.java

+                                                    convertPartitionInfoToInternalRow(
+                                                            partitionInfo)))
+                            .collect(Collectors.toList());
+            LOG.info(


The listPartitions() will be called every 10 seconds (default setting) in streaming task, so this log will also be printed every 10 seconds, even if no new partitions are discovered. It will drown out other useful logs.

...nk-common/src/main/java/com/alibaba/fluss/flink/source/enumerator/FlinkSourceEnumerator.java

wuchong

Thanks @Alibaba-HZY , I added a commit to improve the code a bit.

Besides, please add UTs for the PredicateConverter, and try to support more data types if it is not complex since #1264 is merged.

wuchong · 2025-09-21T07:50:02Z

@Alibaba-HZY, just checking in how’s the progress on this PR? Let me know if you need any help or if there’s anything I can do to move it forward!

Alibaba-HZY · 2025-09-21T07:56:52Z

I'm almost done and will commit tonight or tomorrow.

…

---- Replied Message ---- | From | Jark ***@***.***> | | Date | 09/21/2025 15:50 | | To | apache/fluss ***@***.***> | | Cc | HZY ***@***.***>, Mention ***@***.***> | | Subject | Re: [apache/fluss] [flink]Support partition pushdown for more filters in Flink connector (PR #420) | wuchong left a comment (apache/fluss#420) @Alibaba-HZY, just checking in how’s the progress on this PR? Let me know if you need any help or if there’s anything I can do to move it forward! — Reply to this email directly, view it on GitHub, or unsubscribe. You are receiving this because you were mentioned.Message ID: ***@***.***>

Alibaba-HZY · 2025-09-21T12:12:38Z

just checking in how’s the progress on this PR? Let me know if you need any help or if there’s anything I can do to move it forward!

left comments in your dingding

…nnector (apache#420)

wuchong · 2025-09-26T15:46:55Z

@Alibaba-HZY I updated the implementation to fix bugs and add documentation.

…ushdown and fix bugs

wuchong · 2025-09-26T18:16:55Z

Merging...

…nnector (#420)

…nnector (apache#420)

Alibaba-HZY force-pushed the flink-support-patition-pushdown branch 2 times, most recently from 9a09bfe to 8d7cc2e Compare February 18, 2025 03:49

wuchong added this to the v0.7 milestone Feb 26, 2025

wuchong removed this from the v0.7 milestone Jun 18, 2025

platinumhamburg reviewed Jun 24, 2025

View reviewed changes

platinumhamburg reviewed Jun 26, 2025

View reviewed changes

...-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlinkTableSource.java Outdated Show resolved Hide resolved

...nk-common/src/main/java/com/alibaba/fluss/flink/source/enumerator/FlinkSourceEnumerator.java Outdated Show resolved Hide resolved

platinumhamburg suggested changes Jul 2, 2025

View reviewed changes

...-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlinkTableSource.java Outdated Show resolved Hide resolved

...-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/FlinkTableSource.java Outdated Show resolved Hide resolved

Alibaba-HZY requested a review from platinumhamburg July 2, 2025 09:28

platinumhamburg reviewed Jul 8, 2025

View reviewed changes

platinumhamburg reviewed Jul 10, 2025

View reviewed changes

Alibaba-HZY requested a review from platinumhamburg August 7, 2025 06:52

platinumhamburg reviewed Aug 28, 2025

View reviewed changes

...nk-common/src/main/java/com/alibaba/fluss/flink/source/enumerator/FlinkSourceEnumerator.java Show resolved Hide resolved

polyzos force-pushed the main branch 3 times, most recently from d88c76c to 434a4f4 Compare August 31, 2025 15:13

platinumhamburg mentioned this pull request Sep 9, 2025

[Client] Support prefix lookup with partial partition key columns #1656

Open

2 tasks

Alibaba-HZY force-pushed the flink-support-patition-pushdown branch from d1c103f to 2a3892b Compare September 10, 2025 10:00

wuchong reviewed Sep 16, 2025

View reviewed changes

wuchong linked an issue Sep 16, 2025 that may be closed by this pull request

Support partition pushdown for more filters #1711

Closed

2 tasks

wuchong changed the title ~~[flink]Support partition pushdown in Flink connector~~ [flink]Support partition pushdown for more filters in Flink connector Sep 16, 2025

wuchong force-pushed the flink-support-patition-pushdown branch from 463e7d0 to f0e7af6 Compare September 16, 2025 16:31

Alibaba-HZY force-pushed the flink-support-patition-pushdown branch from f0e7af6 to 9489f6a Compare September 21, 2025 14:37

[flink] Support general predicates for partition pushdown in Flink Co…

048dff7

…nnector (apache#420)

wuchong force-pushed the flink-support-patition-pushdown branch from 2b155cc to 774051b Compare September 26, 2025 15:44

wuchong force-pushed the flink-support-patition-pushdown branch from 774051b to c68b73d Compare September 26, 2025 15:57

[flink] Improve the implementation for general predicates partition p…

9fbfce4

…ushdown and fix bugs

wuchong force-pushed the flink-support-patition-pushdown branch from c68b73d to 9fbfce4 Compare September 26, 2025 16:59

wuchong merged commit 680ca7e into apache:main Sep 26, 2025
6 checks passed

wuchong pushed a commit that referenced this pull request Sep 26, 2025

[flink] Support general predicates for partition pushdown in Flink Co…

a688465

…nnector (#420)

leosanqing pushed a commit to leosanqing/fluss that referenced this pull request Sep 29, 2025

[flink] Support general predicates for partition pushdown in Flink Co…

484754c

…nnector (apache#420)

[flink]Support partition pushdown for more filters in Flink connector #420

[flink]Support partition pushdown for more filters in Flink connector #420

Uh oh!

Conversation

Alibaba-HZY commented Feb 18, 2025 • edited by wuchong Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Purpose

Tests

API and Format

Documentation

Uh oh!

wuchong commented Feb 26, 2025

Uh oh!

Alibaba-HZY commented Mar 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

platinumhamburg Jun 24, 2025

Choose a reason for hiding this comment

Uh oh!

Alibaba-HZY Jun 25, 2025

Choose a reason for hiding this comment

Uh oh!

platinumhamburg Jun 30, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

platinumhamburg commented Jun 30, 2025

Uh oh!

platinumhamburg commented Jul 1, 2025

Uh oh!

Alibaba-HZY commented Jul 1, 2025

Uh oh!

Uh oh!

Uh oh!

platinumhamburg Jul 8, 2025

Choose a reason for hiding this comment

Uh oh!

Alibaba-HZY Jul 8, 2025

Choose a reason for hiding this comment

Uh oh!

platinumhamburg commented Jul 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

platinumhamburg Jul 10, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

wuchong left a comment

Choose a reason for hiding this comment

Uh oh!

wuchong commented Sep 21, 2025

Uh oh!

Alibaba-HZY commented Sep 21, 2025 via email

Uh oh!

Alibaba-HZY commented Sep 21, 2025

Uh oh!

wuchong commented Sep 26, 2025

Uh oh!

wuchong commented Sep 26, 2025

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Alibaba-HZY commented Feb 18, 2025 •

edited by wuchong

Loading

Alibaba-HZY commented Mar 3, 2025 •

edited

Loading

platinumhamburg commented Jul 8, 2025 •

edited

Loading