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

[soin08]

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.

The supported predicate shapes are unchanged, but numeric literal conversion now preserves SQL comparison semantics. Safe conversions to BIGINT or DOUBLE are pushed down. Lossy BIGINT conversions—such as truncating 1.5 to 1—and approximate values that can alias multiple BIGINT keys are no longer pushed down.

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: Codex (GPT-5)

[flinkbot]

CI report:

• 565434c UNKNOWN
• 398b563 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[soin08]

this will be supported in the next MR

[soin08]

also will be supported in the next MR

[soin08]

@flinkbot run azure

[snuyanzin]

fyi: merge commit is always a blocker

[gaborgsomogyi]

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]

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

[soin08]

This fixes two cases where converting numeric literals to the key type could produce an incorrect pushed-down filter.

First, widenToKeyType previously used Number.longValue() for BIGINT keys. This conversion silently truncates fractional values and wraps values outside the long range. For example:

• key < 1.5 was pushed as key < 1, incorrectly removing key = 1.
• key >= 1.5 was pushed as key >= 1, incorrectly adding key = 1.
• A bound such as 10^30 could wrap to an unrelated long value.

Second, converting an integral FLOAT or DOUBLE literal to one exact BIGINT bound is not always safe. When a BIGINT key is compared with a FLOAT or DOUBLE literal, SQL coerces the BIGINT operand to an approximate numeric type. At large magnitudes, adjacent BIGINT values can then round to the same value. For example:

• 16777217L = 16777216f is true, but pushing key = 16777216L omits 16777217L.
• 9007199254740993L > 9007199254740992d is false, but pushing key > 9007199254740992L incorrectly includes it.

The fix uses BigDecimal.longValueExact() for BIGINT bounds, rejecting fractional, non-finite, and out-of-range literals. It also rejects FLOAT literals at abs(value) >= 2^24 and DOUBLE literals at abs(value) >= 2^53, where integer aliasing becomes possible. Rejected predicates remain in the query and are evaluated using the original SQL semantics.

For DOUBLE keys, finite numeric literals are converted to DOUBLE because SQL applies the same conversion to the literal. Tests cover fractional and out-of-range values, NaN and infinity, positive and negative precision boundaries, equality and range predicates, and values immediately below the boundaries that remain safe to push.

[gaborgsomogyi]

I'm fine with the bugfix, found some gaps:

Test gap: add (or restore, adapted) a test for = against a DOUBLE key with a Long literal that isn't exactly representable as a double (e.g. key = 9007199254740993), asserting it's now pushed down with the nearest-double value mirroring the existing gt() test (literalNotRepresentableAsDoubleIsConvertedToTheNearestDouble). Right now only the comparison case is covered; equality relies on an unverified assumption about Calcite pre-folding.

Description: rephrase "There is no change to the supported predicates or their push-down behavior" since it's no longer accurate since the DOUBLE-key precision-loss case now pushes down where it previously didn't.

[gaborgsomogyi]

Seems like unrelated test issues, can you plz kick it green?

[snuyanzin]

Please remove merge commit from PR first fe172da

[snuyanzin]

as it was mentioned at #28982 (comment)

[gaborgsomogyi]

Good catch, just for my own understanding squash and merge won't resolve that? Or what would happen?

[snuyanzin]

just for my own understanding squash and merge won't resolve that?

based on my experience (a couple of years ago) it does not

Not sure if Github changed this behavior

[gaborgsomogyi]

Good to be on safe side, thanks!

merged commit removed, dismissing requesting changes