[FLINK-40116][state] Support more predicates for filter pushdown during savepoint scan

[soin08]

What is the purpose of the change

This pull request improves filter pushdown logic in the savepoint Table API connector by supporting more predicates:

• exclusion (WHERE k <> 5 / k NOT IN (3, 7) / k NOT BETWEEN 10 AND 20)
• union of exact and range (k = 1 OR k > 5 )
• union of ranges (k < 5 OR k > 100)
• intersection of range and exact (narrowing to exact or empty: WHERE k = 5 AND k > 3 / WHERE K = 5 and K > 6)
• intersection of exact and exclusion (WHERE k IN (1,2,3) AND k <> 2)
• intersection of range and exclusion (WHERE k BETWEEN 1 AND 10 AND k <> 5)

At the current stage I've implemented this approach and added 2 new filters ExclusionKeyFilterPlan and ConjunctionKeyFilterPlan, to demonstrate.

Brief change log

TO DISCUSS:

Currently filters live in the package org.apache.flink.state.api.filter. Users of DataStream API can import SavepointKeyFilter and create a new filter with SavepointKeyFilter.exact(), SavepointKeyFilter.range(), or implement their own filter. savepoint connector in the Table API also uses these filters to prune input splits and / or optimize state reading by excluding reading state for the keys the user didn't request.

Inside SavepointKeyFilter, only .test() and .getExactKeys() methods are used during the execution of the job. The rest of the methods, such as getLowerBound(), getUpperBound() and intersect() are only used during the planning phase.

In the future, more planning-state-only methods will be added to this interface. I think it will be a better design if we only leave user-facing methods inside this interface (.test() and .getExactKeys()), since those are the only ones needed for the runtime.

Proposition:

• Create a new interface SavepointKeyFilterPlan implements SavepointKeyFilter which will be used only by the Table API. It will contain all the methods needed for filter resolution: intersect(), getLowerBound(), getUpperBound(). When we want to support more filters, we will add methods to SavepointKeyFilterPlan instead of the user-facing SavepointKeyFilter.
• SavepointKeyFilter becomes simple, no need to explain in the docs that some methods are only used internally by the table api
• Remove SavepointKeyFilter.empty(), since it will no longer make sense
• filters ExactKeyFilter and RangeKeyFilter become common predefined filters to be used with the DataStream API, Table API no longer depends on them
• RangeKeyFilter no longer needs getUpperBound and getLowerBound, so BoundInfo class becomes internal to the Table API.

Expected result:

• cleaner public API
• greater separation of concerns

A downside:

• since now filters for table api have to implement SavepointKeyFilterPlan, they need to duplicate logic for range filter and exact filter. Alternatively, ExactKeyFilterPlan can instantiate ExactKeyFilter and delegate to it its .test() and getExactKeys. Same for RangeKeyFilterPlan.

Verifying this change

This change is coveted by existing and new tests: TODO

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, 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? docs / JavaDoc

---

Was generative AI tooling used to co-author this PR

• [ X] Yes (please specify the tool below)

Generated-by: Claude

[flinkbot]

CI report:

• 3ad85b3 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build