[FLINK-40399][state] Separate SavepointKeyFilter runtime contract from push-down planning by soin08 · Pull Request #28982 · apache/flink

soin08 · 2026-08-16T21:09:48Z

What is the purpose of the change

SavepointKeyFilter currently combines two responsibilities:

the runtime contract used by savepoint scans (test and getExactKeys); and
planning algebra used only by SQL filter push-down (isEmpty, bounds, intersection, key filtering, and BoundInfo).

The planning methods are public even though DataStream API implementations do not use them. This change keeps only the runtime contract on SavepointKeyFilter and moves predicate-combination state into SavepointFilterTranslator.

There is no change to the supported predicates or their push-down behavior.

Brief change log

Keep SavepointKeyFilter focused on test(K), getExactKeys(), and the exact(...) / range(...) factories.
Remove the planning-only public methods, BoundInfo, and EmptyKeyFilter, without introducing a separate table-side filter-plan hierarchy.
Keep exact keys and range bounds in a private KeyFilterPlan inside SavepointFilterTranslator while walking expressions. It performs the existing intersection logic and creates the final SavepointKeyFilter.exact(...) or SavepointKeyFilter.range(...) only after translation.
Simplify ExactKeyFilter and RangeKeyFilter so they contain only runtime filtering behavior. An empty exact-key set now represents a filter that matches nothing.
Update the State Processor API documentation in English and Chinese to remove the deleted empty() factory and planning-method references.
Add direct tests for exact and range runtime filters, and expand translator and SQL-level push-down coverage.

Verifying this change

The tests cover:

exact-key and range runtime behavior, custom comparators, serialization, and empty/degenerate ranges;
exact/exact, range/range, and exact/range intersections performed during translation;
empty and contradictory intersections;
unsupported predicates being returned in remaining() so they are still evaluated;
an untranslatable child aborting the complete AND / OR push-down;
null and non-comparable literals, comparison arity and direction, and safe numeric widening; and
end-to-end SQL behavior for equality, bounds, intersecting ranges, partial push-down, and unsupported disjunctions.

The existing limitation is preserved: OR only combines finite exact-key predicates. For example, k = 5 OR k < 10 and disjunctions of ranges are not pushed down and remain for normal SQL evaluation.

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

Dependencies (does it add or upgrade a dependency): no
The public API: yes (@Experimental). Planning-only methods and empty() are removed from SavepointKeyFilter; its runtime contract (test, getExactKeys) and exact/range factories remain.
The serializers: no
The runtime per-record code paths (performance sensitive): no behavioral change
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
The State Processor API documentation is updated in both docs/content and docs/content.zh for the reduced interface.

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

Generated-by: Claude Code (Claude Opus 5)

Generated-by: Codex (GPT-5)

@experimental

…m push-down planning SavepointKeyFilter carried two unrelated responsibilities: the contract the savepoint scan needs, and the algebra used only while combining predicates during filter push-down. The second group was public and @experimental despite being unusable outside push-down translation, to the point that its javadoc had to disclaim it - "Used only while combining filters during push-down translation, not during the scan." Split the two: * org.apache.flink.state.api.filter.SavepointKeyFilter keeps only what the scan needs - test() and getExactKeys() - plus the public factories. * org.apache.flink.state.table.filter.SavepointKeyFilterPlan carries the planning algebra (isEmpty, bounds, intersect, filterKeys, empty), next to the connector that is its only user. SavepointKeyFilterPlan and BoundInfo are public only so the connector in org.apache.flink.state.table can reach them, and are marked @internal. The unit tests are split the same way: the plan algebra cases move to SavepointKeyFilterPlanTest, leaving SavepointFilterTranslatorTest to cover only expression translation. No behaviour change.

Adds cases for savepoint key filter push-down behaviour that had no test, most importantly that predicates which cannot be pushed are returned in remaining() so the runtime still evaluates them. Nothing covered that before: every apply() case asserted remaining() was empty, so a regression there would have silently dropped rows. Also covered: an untranslatable child aborting the whole AND/OR, "key = NULL" whose literal has no readable value, non-Comparable literals on a BYTES key, the two comparison flip directions that were missing, comparison arity, and at SQL level the upper-bound predicates, which had no end-to-end test at all. Finally, every test states the predicate under test as a leading comment, so the SQL shape being exercised is readable without decoding the expression builders.

flinkbot · 2026-08-16T21:13:38Z

CI report:

565434c UNKNOWN
9c1b945 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

soin08 · 2026-08-17T12:40:26Z

    }

+    @Test
+    void testOrOfExactAndRangeOnKeyIsNotPushedDownButReturnsCorrectResult() throws Exception {


this will be supported in the next MR

soin08 · 2026-08-17T12:40:54Z

+    }
+
+    @Test
+    void testOrOfTwoRangesOnKeyIsNotPushedDownButReturnsCorrectResult() throws Exception {


also will be supported in the next MR

soin08 · 2026-08-26T08:57:13Z

@flinkbot run azure

snuyanzin · 2026-08-26T10:26:32Z

fyi: merge commit is always a blocker

gaborgsomogyi · 2026-09-02T12:17:12Z

Thanks for the efforts!

Shrinking the public API surface here is the right direction, good call splitting the runtime contract from the push-down-only algebra.

That said, Flink's existing filter push-down connectors (e.g. SupportsFilterPushDown + FileSystemTableSource) generally avoid a second parallel class hierarchy for this: they keep one flat, already-serializable representation of a filter, computed once during push-down, and pass that same object through to runtime rather than shipping a richer "planning" type alongside a "runtime" type.

Right now test() is duplicated verbatim between RangeKeyFilter/RangeKeyFilterPlan and ExactKeyFilter/ExactKeyFilterPlan, and since the SQL path only exercises the *Plan variants, a future fix to one won't be caught by anything and will just quietly drift from the other.

Could this be done without the duplication by keeping the intersect/bounds combining logic as private computation inside SavepointFilterTranslator (plain local state while walking the expression tree), and only constructing the existing SavepointKeyFilter.range()/.exact() once at the end, so there's a single filter implementation used by both the SQL and direct API paths? What's the reasoning for a separate *Plan hierarchy instead?

gaborgsomogyi · 2026-09-04T12:51:56Z

Thanks for fixing, the code looks good but the description is stale.

Ilya Soin added 2 commits August 16, 2026 23:46

soin08 changed the title ~~Flink 40399~~ [FLINK-40399][state] Separate SavepointKeyFilter runtime contract from push-down planning Aug 16, 2026

[FLINK-40399][state] spotless

a2713e9

soin08 commented Aug 17, 2026

View reviewed changes

soin08 marked this pull request as ready for review August 17, 2026 12:46

github-actions Bot added the community-reviewed PR has been reviewed by the community. label Aug 18, 2026

Merge branch 'master' into FLINK-40399

fe172da

[FLINK-40399[state] fix typo

70868d1

Ilya Soin added 3 commits September 3, 2026 17:29

[FLINK-40399][state] remove table api filter hierarchy

ff20675

[FLINK-40399][state] add tests for public api filters

565434c

[FLINK-40399][state] spotless

ec6cc21

[FLINK-40399][state] fix bug in widenToKeyType and improve test coverage

9c1b945

Navigation Menu

Uh oh!

[FLINK-40399][state] Separate SavepointKeyFilter runtime contract from push-down planning - #28982

[FLINK-40399][state] Separate SavepointKeyFilter runtime contract from push-down planning#28982
soin08 wants to merge 9 commits into
apache:masterfrom
soin08:FLINK-40399

soin08 commented Aug 16, 2026 •

edited

Loading

Uh oh!

flinkbot commented Aug 16, 2026 •

edited

Loading

Uh oh!

soin08 Aug 17, 2026

Uh oh!

soin08 Aug 17, 2026

Uh oh!

soin08 commented Aug 26, 2026

Uh oh!

snuyanzin commented Aug 26, 2026

Uh oh!

gaborgsomogyi commented Sep 2, 2026

Uh oh!

gaborgsomogyi commented Sep 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Uh oh!

Conversation

soin08 commented Aug 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

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

Was generative AI tooling used to co-author this PR?

Uh oh!

flinkbot commented Aug 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

soin08 Aug 17, 2026

Choose a reason for hiding this comment

Uh oh!

soin08 Aug 17, 2026

Choose a reason for hiding this comment

Uh oh!

soin08 commented Aug 26, 2026

Uh oh!

snuyanzin commented Aug 26, 2026

Uh oh!

gaborgsomogyi commented Sep 2, 2026

Uh oh!

gaborgsomogyi commented Sep 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

soin08 commented Aug 16, 2026 •

edited

Loading

flinkbot commented Aug 16, 2026 •

edited

Loading