Skip to content

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

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

Draft
soin08 wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8fbd972
[FLINK-40116][state] Reorganize filters
Jul 10, 2026
3ad85b3
[FLINK-40116][state] Add conjunction filter and exclusion filter
Jul 10, 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
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
86 changes: 16 additions & 70 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 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