Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
soin08 wants to merge 3 commits into
base: master
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a8ee4dd
[FLINK-40399][state] Separate SavepointKeyFilter runtime contract fro…
Aug 16, 2026
27f72a2
[FLINK-40399][tests] Cover remaining key filter push-down paths
Aug 16, 2026
a2713e9
[FLINK-40399][state] spotless
Aug 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 0 additions & 3 deletions docs/content.zh/docs/libs/state_processor_api.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -254,7 +254,6 @@ DataStream<KeyedState> keyRange = savepoint.readKeyedState(
* `SavepointKeyFilter.exact(K key)` / `SavepointKeyFilter.exact(Set<K> keys)` — match a single key or a finite set of keys.
* `SavepointKeyFilter.range(K lower, boolean lowerInclusive, K upper, boolean upperInclusive)` — match a range; `K` must implement `Comparable<K>`. Either bound may be `null` to leave that side unbounded.
* `SavepointKeyFilter.range(K lower, boolean lowerInclusive, K upper, boolean upperInclusive, SerializableComparator<K> comparator)` — same, but with an explicit comparator for key types that do not implement `Comparable<K>`. The comparator must be serializable because the filter is shipped with the job; lambdas and method references assigned to `SerializableComparator` satisfy this automatically.
* `SavepointKeyFilter.empty()` — match no keys. Not intended for direct use — it only serves as an internal building block for the Table API filter pushdown.

When the built-in filters are not enough, you can implement the `SavepointKeyFilter<K>` interface yourself.
For use with the DataStream API, only `test(K key)` has to be implemented; it is called for every key in each opened split and decides whether that key will be read.
Expand Down Expand Up @@ -314,8 +313,6 @@ DataStream<KeyedState> firstKeys = savepoint.readKeyedState(
new UpToKeyFilter(100));
```

The remaining interface methods can be left at their defaults for DataStream API usage, as they are only used internally in the Table API during push-down handling.

### 窗口状态 Window State

State Processor API 支持读取[窗口算子]({{< ref "docs/dev/datastream/operators/windows" >}})的状态，当读取窗口状态时，需要指定算子 id，窗口分配器和聚合类型。
Expand Down
3 changes: 0 additions & 3 deletions docs/content/docs/libs/state_processor_api.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -268,7 +268,6 @@ DataStream<KeyedState> keyRange = savepoint.readKeyedState(
* `SavepointKeyFilter.exact(K key)` / `SavepointKeyFilter.exact(Set<K> keys)` — match a single key or a finite set of keys.
* `SavepointKeyFilter.range(K lower, boolean lowerInclusive, K upper, boolean upperInclusive)` — match a range; `K` must implement `Comparable<K>`. Either bound may be `null` to leave that side unbounded.
* `SavepointKeyFilter.range(K lower, boolean lowerInclusive, K upper, boolean upperInclusive, SerializableComparator<K> comparator)` — same, but with an explicit comparator for key types that do not implement `Comparable<K>`. The comparator must be serializable because the filter is shipped with the job; lambdas and method references assigned to `SerializableComparator` satisfy this automatically.
* `SavepointKeyFilter.empty()` — match no keys. Not intended for direct use — it only serves as an internal building block for the Table API filter pushdown.

When the built-in filters are not enough, you can implement the `SavepointKeyFilter<K>` interface yourself.
For use with the DataStream API, only `test(K key)` has to be implemented; it is called for every key in each opened split and decides whether that key will be read.
Expand Down Expand Up @@ -328,8 +327,6 @@ DataStream<KeyedState> firstKeys = savepoint.readKeyedState(
new UpToKeyFilter(100));
```

The remaining interface methods can be left at their defaults for DataStream API usage, as they are only used internally in the Table API during push-down handling.

#### Window State

The state processor API supports reading state from a [window operator]({{< ref "docs/dev/datastream/operators/windows" >}}).
Expand Down
15 changes: 0 additions & 15 deletions ...-state-processing-api/src/main/java/org/apache/flink/state/api/filter/ExactKeyFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@

package org.apache.flink.state.api.filter;

import java.util.HashSet;
import java.util.Set;

/** A filter that accepts a finite set of keys. */
Expand All @@ -42,20 +41,6 @@ public Set<K> getExactKeys() {
return keys;
}

@Override
public SavepointKeyFilter<K> intersect(SavepointKeyFilter<K> other) {
if (other.isEmpty()) {
return other;
}
final Set<K> otherKeys = other.getExactKeys();
if (otherKeys != null) {
final Set<K> intersection = new HashSet<>(keys);
intersection.retainAll(otherKeys);
return SavepointKeyFilter.exact(intersection);
}
return SavepointKeyFilter.filterKeys(keys, other);
}

@Override
public String toString() {
return "ExactKeyFilter" + keys;
Expand Down
88 changes: 17 additions & 71 deletions ...-state-processing-api/src/main/java/org/apache/flink/state/api/filter/RangeKeyFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,105 +20,51 @@

import javax.annotation.Nullable;

import java.util.Set;

/** A filter based on a range with an injected comparator. */
final class RangeKeyFilter<K> implements SavepointKeyFilter<K> {

private static final long serialVersionUID = 3L;
private static final long serialVersionUID = 4L;

private final SerializableComparator<K> comparator;
@Nullable private final BoundInfo<K> lower;
@Nullable private final BoundInfo<K> upper;
@Nullable private final K lower;
private final boolean isLowerInclusive;
@Nullable private final K upper;
private final boolean isUpperInclusive;

RangeKeyFilter(
SerializableComparator<K> comparator,
@Nullable BoundInfo<K> lower,
@Nullable BoundInfo<K> upper) {
@Nullable K lower,
boolean isLowerInclusive,
@Nullable K upper,
boolean isUpperInclusive) {
this.comparator = comparator;
this.lower = lower;
this.isLowerInclusive = isLowerInclusive;
this.upper = upper;
this.isUpperInclusive = isUpperInclusive;
}

@Override
public boolean test(K key) {
if (lower != null) {
int cmp = comparator.compare(lower.getValue(), key);
if (cmp > 0 || (cmp == 0 && !lower.isInclusive())) {
int cmp = comparator.compare(lower, key);
if (cmp > 0 || (cmp == 0 && !isLowerInclusive)) {
return false;
}
}
if (upper != null) {
int cmp = comparator.compare(upper.getValue(), key);
if (cmp < 0 || (cmp == 0 && !upper.isInclusive())) {
int cmp = comparator.compare(upper, key);
if (cmp < 0 || (cmp == 0 && !isUpperInclusive)) {
return false;
}
}
return true;
}

@Override
public BoundInfo<K> getLowerBound() {
return lower;
}

@Override
public BoundInfo<K> getUpperBound() {
return upper;
}

@Override
public SavepointKeyFilter<K> intersect(SavepointKeyFilter<K> other) {
if (other.isEmpty()) {
return other;
}
final Set<K> otherExactKeys = other.getExactKeys();
if (otherExactKeys != null) {
return SavepointKeyFilter.filterKeys(otherExactKeys, this);
}
return intersectRange(other.getLowerBound(), other.getUpperBound());
}

private SavepointKeyFilter<K> intersectRange(
@Nullable BoundInfo<K> otherLower, @Nullable BoundInfo<K> otherUpper) {
BoundInfo<K> newLower = tighter(lower, otherLower, true);
BoundInfo<K> newUpper = tighter(upper, otherUpper, false);

if (newLower != null && newUpper != null) {
int cmp = comparator.compare(newLower.getValue(), newUpper.getValue());
if (cmp > 0) {
return SavepointKeyFilter.empty();
}
if (cmp == 0 && (!newLower.isInclusive() || !newUpper.isInclusive())) {
return SavepointKeyFilter.empty();
}
}
return new RangeKeyFilter<>(comparator, newLower, newUpper);
}

@Nullable
private BoundInfo<K> tighter(
@Nullable BoundInfo<K> a, @Nullable BoundInfo<K> b, boolean preferHigher) {
if (a == null) {
return b;
}
if (b == null) {
return a;
}
int c = comparator.compare(a.getValue(), b.getValue());
if (c == 0) {
return new BoundInfo<>(a.getValue(), a.isInclusive() && b.isInclusive());
}
boolean aWins = preferHigher ? c > 0 : c < 0;
return aWins ? a : b;
}

@Override
public String toString() {
String lowerStr =
lower == null ? "(-∞" : (lower.isInclusive() ? "[" : "(") + lower.getValue();
String upperStr =
upper == null ? "+∞)" : upper.getValue() + (upper.isInclusive() ? "]" : ")");
String lowerStr = lower == null ? "(-∞" : (isLowerInclusive ? "[" : "(") + lower;
String upperStr = upper == null ? "+∞)" : upper + (isUpperInclusive ? "]" : ")");
return "RangeKeyFilter" + lowerStr + ", " + upperStr;
}
}
66 changes: 2 additions & 64 deletions ...te-processing-api/src/main/java/org/apache/flink/state/api/filter/SavepointKeyFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import javax.annotation.Nullable;

import java.io.Serializable;
import java.util.HashSet;
import java.util.Set;

/**
Expand All @@ -37,15 +36,6 @@ public interface SavepointKeyFilter<K> extends Serializable {
/** Returns {@code true} if the given key passes this filter. */
boolean test(K key);

/**
* Returns {@code true} if this filter rejects every key.
*
* <p>Used only while combining filters during push-down translation, not during the scan.
*/
default boolean isEmpty() {
return false;
}

/**
* Returns the finite set of keys this filter matches, or {@code null} if the filter does not
* resolve to a finite key set.
Expand All @@ -55,58 +45,12 @@ default Set<K> getExactKeys() {
return null;
}

/**
* Returns the lower bound of this filter's range, or {@code null} if the filter does not define
* a lower bound.
*
* <p>Used only while combining filters during push-down translation, not during the scan.
*/
@Nullable
default BoundInfo<K> getLowerBound() {
return null;
}

/**
* Returns the upper bound of this filter's range, or {@code null} if the filter does not define
* an upper bound.
*
* <p>Used only while combining filters during push-down translation, not during the scan.
*/
@Nullable
default BoundInfo<K> getUpperBound() {
return null;
}

/**
* Returns a filter that accepts a key if and only if both {@code this} and {@code other} accept
* it.
*
* <p>Used only while combining filters during push-down translation, not during the scan.
*/
default SavepointKeyFilter<K> intersect(SavepointKeyFilter<K> other) {
throw new UnsupportedOperationException(
getClass().getSimpleName() + " does not support intersect()");
}

static <K> SavepointKeyFilter<K> filterKeys(Set<K> keys, SavepointKeyFilter<K> predicate) {
final Set<K> retained = new HashSet<>();
for (K key : keys) {
if (predicate.test(key)) {
retained.add(key);
}
}
return exact(retained);
}

static <K> SavepointKeyFilter<K> exact(Set<K> keys) {
if (keys.isEmpty()) {
return EmptyKeyFilter.instance();
}
return new ExactKeyFilter<>(keys);
}

static <K> SavepointKeyFilter<K> exact(K value) {
return new ExactKeyFilter<>(Set.of(value));
return exact(Set.of(value));
}

static <K extends Comparable<K>> SavepointKeyFilter<K> range(
Expand All @@ -120,13 +64,7 @@ static <K> SavepointKeyFilter<K> range(
@Nullable K upper,
boolean upperInclusive,
SerializableComparator<K> comparator) {
BoundInfo<K> lowerBoundInfo = lower != null ? new BoundInfo<>(lower, lowerInclusive) : null;
BoundInfo<K> upperBoundInfo = upper != null ? new BoundInfo<>(upper, upperInclusive) : null;
return new RangeKeyFilter<>(comparator, lowerBoundInfo, upperBoundInfo);
}

static <K> SavepointKeyFilter<K> empty() {
return EmptyKeyFilter.instance();
return new RangeKeyFilter<>(comparator, lower, lowerInclusive, upper, upperInclusive);
}

final class NaturalOrderComparator<K extends Comparable<K>>
Expand Down
Loading