From 8fbd97290662e12559fd76717b5b7c39b1de5ff8 Mon Sep 17 00:00:00 2001 From: Ilya Soin Date: Fri, 10 Jul 2026 13:28:11 +0300 Subject: [PATCH 1/2] [FLINK-40116][state] Reorganize filters --- .../state/api/filter/ExactKeyFilter.java | 15 --- .../state/api/filter/RangeKeyFilter.java | 86 +++--------- .../state/api/filter/SavepointKeyFilter.java | 66 +-------- .../table/SavepointFilterTranslator.java | 65 ++++----- .../{api => table}/filter/BoundInfo.java | 7 +- .../filter/EmptyKeyFilterPlan.java} | 16 +-- .../table/filter/ExactKeyFilterPlan.java | 63 +++++++++ .../table/filter/RangeKeyFilterPlan.java | 126 ++++++++++++++++++ .../table/filter/SavepointKeyFilterPlan.java | 100 ++++++++++++++ .../api/SavepointReaderKeyedStateITCase.java | 3 +- .../api/input/KeyedStateInputFormatTest.java | 23 ---- .../table/SavepointFilterTranslatorTest.java | 99 +++++++------- 12 files changed, 403 insertions(+), 266 deletions(-) rename flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/{api => table}/filter/BoundInfo.java (88%) rename flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/{api/filter/EmptyKeyFilter.java => table/filter/EmptyKeyFilterPlan.java} (75%) create mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/ExactKeyFilterPlan.java create mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/RangeKeyFilterPlan.java create mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/SavepointKeyFilterPlan.java diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/ExactKeyFilter.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/ExactKeyFilter.java index 13f37554d9267..2cdf677252194 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/ExactKeyFilter.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/ExactKeyFilter.java @@ -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. */ @@ -42,20 +41,6 @@ public Set getExactKeys() { return keys; } - @Override - public SavepointKeyFilter intersect(SavepointKeyFilter other) { - if (other.isEmpty()) { - return other; - } - final Set otherKeys = other.getExactKeys(); - if (otherKeys != null) { - final Set intersection = new HashSet<>(keys); - intersection.retainAll(otherKeys); - return SavepointKeyFilter.exact(intersection); - } - return SavepointKeyFilter.filterKeys(keys, other); - } - @Override public String toString() { return "ExactKeyFilter" + keys; diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/RangeKeyFilter.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/RangeKeyFilter.java index 392a06fbf8766..22f99c3e1b339 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/RangeKeyFilter.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/RangeKeyFilter.java @@ -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 implements SavepointKeyFilter { private static final long serialVersionUID = 3L; private final SerializableComparator comparator; - @Nullable private final BoundInfo lower; - @Nullable private final BoundInfo upper; + @Nullable private final K lower; + private final boolean isLowerInclusive; + @Nullable private final K upper; + private final boolean isUpperInclusive; RangeKeyFilter( SerializableComparator comparator, - @Nullable BoundInfo lower, - @Nullable BoundInfo 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 getLowerBound() { - return lower; - } - - @Override - public BoundInfo getUpperBound() { - return upper; - } - - @Override - public SavepointKeyFilter intersect(SavepointKeyFilter other) { - if (other.isEmpty()) { - return other; - } - final Set otherExactKeys = other.getExactKeys(); - if (otherExactKeys != null) { - return SavepointKeyFilter.filterKeys(otherExactKeys, this); - } - return intersectRange(other.getLowerBound(), other.getUpperBound()); - } - - private SavepointKeyFilter intersectRange( - @Nullable BoundInfo otherLower, @Nullable BoundInfo otherUpper) { - BoundInfo newLower = tighter(lower, otherLower, true); - BoundInfo 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 tighter( - @Nullable BoundInfo a, @Nullable BoundInfo 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; } } diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/SavepointKeyFilter.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/SavepointKeyFilter.java index c639c5c056d94..ebf059ec54420 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/SavepointKeyFilter.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/SavepointKeyFilter.java @@ -23,7 +23,6 @@ import javax.annotation.Nullable; import java.io.Serializable; -import java.util.HashSet; import java.util.Set; /** @@ -37,15 +36,6 @@ public interface SavepointKeyFilter 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. - * - *

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. @@ -55,58 +45,12 @@ default Set getExactKeys() { return null; } - /** - * Returns the lower bound of this filter's range, or {@code null} if the filter does not define - * a lower bound. - * - *

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

Used only while combining filters during push-down translation, not during the scan. - */ - @Nullable - default BoundInfo getUpperBound() { - return null; - } - - /** - * Returns a filter that accepts a key if and only if both {@code this} and {@code other} accept - * it. - * - *

Used only while combining filters during push-down translation, not during the scan. - */ - default SavepointKeyFilter intersect(SavepointKeyFilter other) { - throw new UnsupportedOperationException( - getClass().getSimpleName() + " does not support intersect()"); - } - - static SavepointKeyFilter filterKeys(Set keys, SavepointKeyFilter predicate) { - final Set retained = new HashSet<>(); - for (K key : keys) { - if (predicate.test(key)) { - retained.add(key); - } - } - return exact(retained); - } - static SavepointKeyFilter exact(Set keys) { - if (keys.isEmpty()) { - return EmptyKeyFilter.instance(); - } return new ExactKeyFilter<>(keys); } static SavepointKeyFilter exact(K value) { - return new ExactKeyFilter<>(Set.of(value)); + return exact(Set.of(value)); } static > SavepointKeyFilter range( @@ -120,13 +64,7 @@ static SavepointKeyFilter range( @Nullable K upper, boolean upperInclusive, SerializableComparator comparator) { - BoundInfo lowerBoundInfo = lower != null ? new BoundInfo<>(lower, lowerInclusive) : null; - BoundInfo upperBoundInfo = upper != null ? new BoundInfo<>(upper, upperInclusive) : null; - return new RangeKeyFilter<>(comparator, lowerBoundInfo, upperBoundInfo); - } - - static SavepointKeyFilter empty() { - return EmptyKeyFilter.instance(); + return new RangeKeyFilter<>(comparator, lower, lowerInclusive, upper, upperInclusive); } final class NaturalOrderComparator> diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointFilterTranslator.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointFilterTranslator.java index f12cdf9a304e9..ec6af13c49632 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointFilterTranslator.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointFilterTranslator.java @@ -18,7 +18,7 @@ package org.apache.flink.state.table; -import org.apache.flink.state.api.filter.SavepointKeyFilter; +import org.apache.flink.state.table.filter.SavepointKeyFilterPlan; import org.apache.flink.table.expressions.CallExpression; import org.apache.flink.table.expressions.FieldReferenceExpression; import org.apache.flink.table.expressions.ResolvedExpression; @@ -40,7 +40,7 @@ import java.util.function.BiFunction; /** - * Converts {@link ResolvedExpression} key filter predicates into {@link SavepointKeyFilter} + * Converts {@link ResolvedExpression} key filter predicates into {@link SavepointKeyFilterPlan} * instances that can be used to prune key groups and key iterations during savepoint reads. */ @SuppressWarnings({"rawtypes", "unchecked"}) @@ -50,7 +50,7 @@ class SavepointFilterTranslator { private static final Map< FunctionDefinition, - BiFunction> + BiFunction> FILTERS = Map.of( BuiltInFunctionDefinitions.EQUALS, @@ -82,9 +82,9 @@ Result apply(List filters) { final List accepted = new ArrayList<>(); final List remaining = new ArrayList<>(); - SavepointKeyFilter keyFilter = null; + SavepointKeyFilterPlan keyFilter = null; for (ResolvedExpression filter : filters) { - SavepointKeyFilter extracted = extractFilter(filter); + SavepointKeyFilterPlan extracted = extractFilter(filter); if (extracted == null) { remaining.add(filter); continue; @@ -98,11 +98,12 @@ Result apply(List filters) { } @Nullable - private SavepointKeyFilter extractFilter(ResolvedExpression expr) { - final BiFunction extractor = - expr instanceof CallExpression - ? FILTERS.get(((CallExpression) expr).getFunctionDefinition()) - : null; + private SavepointKeyFilterPlan extractFilter(ResolvedExpression expr) { + final BiFunction + extractor = + expr instanceof CallExpression + ? FILTERS.get(((CallExpression) expr).getFunctionDefinition()) + : null; if (extractor == null) { LOG.debug( "Unsupported predicate [{}] cannot be pushed into savepoint key filter.", expr); @@ -116,7 +117,7 @@ private SavepointKeyFilter extractFilter(ResolvedExpression expr) { // ------------------------------------------------------------------------- @Nullable - private SavepointKeyFilter fromEquals(CallExpression call) { + private SavepointKeyFilterPlan fromEquals(CallExpression call) { if (!isBinaryValid(call)) { return null; } @@ -133,14 +134,14 @@ private SavepointKeyFilter fromEquals(CallExpression call) { if (value == null) { return null; } - return SavepointKeyFilter.exact(value); + return SavepointKeyFilterPlan.exact(value); } @Nullable - private SavepointKeyFilter fromOr(CallExpression call) { + private SavepointKeyFilterPlan fromOr(CallExpression call) { Set keys = new HashSet<>(); for (ResolvedExpression arg : call.getResolvedChildren()) { - SavepointKeyFilter sub = extractFilter(arg); + SavepointKeyFilterPlan sub = extractFilter(arg); if (sub == null) { return null; } @@ -151,7 +152,7 @@ private SavepointKeyFilter fromOr(CallExpression call) { } keys.addAll(subKeys); } - return SavepointKeyFilter.exact(keys); + return SavepointKeyFilterPlan.exact(keys); } // ------------------------------------------------------------------------- @@ -159,10 +160,10 @@ private SavepointKeyFilter fromOr(CallExpression call) { // ------------------------------------------------------------------------- @Nullable - private SavepointKeyFilter fromAnd(CallExpression call) { - SavepointKeyFilter merged = null; + private SavepointKeyFilterPlan fromAnd(CallExpression call) { + SavepointKeyFilterPlan merged = null; for (ResolvedExpression arg : call.getResolvedChildren()) { - SavepointKeyFilter sub = extractFilter(arg); + SavepointKeyFilterPlan sub = extractFilter(arg); // AND only absorbs range filters; exact (or null) children break pushdown. if (sub == null || sub.getExactKeys() != null) { return null; @@ -176,7 +177,7 @@ private SavepointKeyFilter fromAnd(CallExpression call) { } @Nullable - private SavepointKeyFilter fromBetween(CallExpression call) { + private SavepointKeyFilterPlan fromBetween(CallExpression call) { List args = call.getResolvedChildren(); if (args.size() != 3) { return null; @@ -200,33 +201,33 @@ private SavepointKeyFilter fromBetween(CallExpression call) { lower.getClass().getName()); return null; } - return SavepointKeyFilter.range( + return SavepointKeyFilterPlan.range( (Comparable) lower, true, (Comparable) upper, true); } @Nullable - private SavepointKeyFilter fromGreaterThan(CallExpression call) { + private SavepointKeyFilterPlan fromGreaterThan(CallExpression call) { return fromComparison(call, Comparison.GT); } @Nullable - private SavepointKeyFilter fromGreaterThanOrEqual(CallExpression call) { + private SavepointKeyFilterPlan fromGreaterThanOrEqual(CallExpression call) { return fromComparison(call, Comparison.GTE); } @Nullable - private SavepointKeyFilter fromLessThan(CallExpression call) { + private SavepointKeyFilterPlan fromLessThan(CallExpression call) { return fromComparison(call, Comparison.LT); } @Nullable - private SavepointKeyFilter fromLessThanOrEqual(CallExpression call) { + private SavepointKeyFilterPlan fromLessThanOrEqual(CallExpression call) { return fromComparison(call, Comparison.LTE); } @Nullable - private SavepointKeyFilter fromComparison(CallExpression call, Comparison cmp) { + private SavepointKeyFilterPlan fromComparison(CallExpression call, Comparison cmp) { if (!isBinaryValid(call)) { return null; } @@ -252,13 +253,13 @@ private SavepointKeyFilter fromComparison(CallExpression call, Comparison cmp) { Comparison keyLeftCmp = keyOnLeft ? cmp : cmp.flip(); switch (keyLeftCmp) { case GT: - return SavepointKeyFilter.range(b, false, null, true); + return SavepointKeyFilterPlan.range(b, false, null, true); case GTE: - return SavepointKeyFilter.range(b, true, null, true); + return SavepointKeyFilterPlan.range(b, true, null, true); case LT: - return SavepointKeyFilter.range(null, true, b, false); + return SavepointKeyFilterPlan.range(null, true, b, false); case LTE: - return SavepointKeyFilter.range(null, true, b, true); + return SavepointKeyFilterPlan.range(null, true, b, true); default: throw new IllegalStateException("Unknown Comparison: " + keyLeftCmp); } @@ -323,12 +324,12 @@ private Object widenToKeyType(Object value) { static final class Result { private final List accepted; private final List remaining; - @Nullable private final SavepointKeyFilter keyFilter; + @Nullable private final SavepointKeyFilterPlan keyFilter; private Result( List accepted, List remaining, - @Nullable SavepointKeyFilter keyFilter) { + @Nullable SavepointKeyFilterPlan keyFilter) { this.accepted = accepted; this.remaining = remaining; this.keyFilter = keyFilter; @@ -343,7 +344,7 @@ List remaining() { } @Nullable - SavepointKeyFilter keyFilter() { + SavepointKeyFilterPlan keyFilter() { return keyFilter; } } diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/BoundInfo.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/BoundInfo.java similarity index 88% rename from flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/BoundInfo.java rename to flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/BoundInfo.java index fe0b525cd1c7b..e2cd3f2088594 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/BoundInfo.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/BoundInfo.java @@ -16,15 +16,12 @@ * limitations under the License. */ -package org.apache.flink.state.api.filter; - -import org.apache.flink.annotation.Experimental; +package org.apache.flink.state.table.filter; import java.io.Serializable; /** Information about a bound in a range filter. */ -@Experimental -public final class BoundInfo implements Serializable { +final class BoundInfo implements Serializable { private static final long serialVersionUID = 4L; private final K value; diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/EmptyKeyFilter.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/EmptyKeyFilterPlan.java similarity index 75% rename from flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/EmptyKeyFilter.java rename to flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/EmptyKeyFilterPlan.java index 160cc17d4a519..ae5502cac8c72 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/EmptyKeyFilter.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/EmptyKeyFilterPlan.java @@ -16,24 +16,24 @@ * limitations under the License. */ -package org.apache.flink.state.api.filter; +package org.apache.flink.state.table.filter; import java.util.Collections; import java.util.Set; /** A filter that rejects every key. */ -final class EmptyKeyFilter implements SavepointKeyFilter { +final class EmptyKeyFilterPlan implements SavepointKeyFilterPlan { private static final long serialVersionUID = 1L; @SuppressWarnings("rawtypes") - private static final EmptyKeyFilter INSTANCE = new EmptyKeyFilter<>(); + private static final EmptyKeyFilterPlan INSTANCE = new EmptyKeyFilterPlan<>(); - private EmptyKeyFilter() {} + private EmptyKeyFilterPlan() {} @SuppressWarnings("unchecked") - static EmptyKeyFilter instance() { - return (EmptyKeyFilter) INSTANCE; + static EmptyKeyFilterPlan instance() { + return (EmptyKeyFilterPlan) INSTANCE; } @Override @@ -52,7 +52,7 @@ public Set getExactKeys() { } @Override - public SavepointKeyFilter intersect(SavepointKeyFilter other) { + public SavepointKeyFilterPlan intersect(SavepointKeyFilterPlan other) { return this; } @@ -62,6 +62,6 @@ private Object readResolve() { @Override public String toString() { - return "EmptyKeyFilter"; + return "EmptyKeyFilterPlan"; } } diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/ExactKeyFilterPlan.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/ExactKeyFilterPlan.java new file mode 100644 index 0000000000000..23f870367d202 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/ExactKeyFilterPlan.java @@ -0,0 +1,63 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table.filter; + +import java.util.HashSet; +import java.util.Set; + +/** A filter that accepts a finite set of keys. */ +final class ExactKeyFilterPlan implements SavepointKeyFilterPlan { + + private static final long serialVersionUID = 2L; + + private final Set keys; + + ExactKeyFilterPlan(Set keys) { + this.keys = Set.copyOf(keys); + } + + @Override + public boolean test(K key) { + return keys.contains(key); + } + + @Override + public Set getExactKeys() { + return keys; + } + + @Override + public SavepointKeyFilterPlan intersect(SavepointKeyFilterPlan other) { + if (other.isEmpty()) { + return other; + } + final Set otherKeys = other.getExactKeys(); + if (otherKeys != null) { + final Set intersection = new HashSet<>(keys); + intersection.retainAll(otherKeys); + return SavepointKeyFilterPlan.exact(intersection); + } + return SavepointKeyFilterPlan.filterKeys(keys, other); + } + + @Override + public String toString() { + return "ExactKeyFilterPlan" + keys; + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/RangeKeyFilterPlan.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/RangeKeyFilterPlan.java new file mode 100644 index 0000000000000..8732f33a445b5 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/RangeKeyFilterPlan.java @@ -0,0 +1,126 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table.filter; + +import org.apache.flink.state.api.filter.SerializableComparator; + +import javax.annotation.Nullable; + +import java.util.Set; + +/** A filter based on a range with an injected comparator. */ +final class RangeKeyFilterPlan implements SavepointKeyFilterPlan { + + private static final long serialVersionUID = 3L; + + private final SerializableComparator comparator; + @Nullable private final BoundInfo lower; + @Nullable private final BoundInfo upper; + + RangeKeyFilterPlan( + SerializableComparator comparator, + @Nullable BoundInfo lower, + @Nullable BoundInfo upper) { + this.comparator = comparator; + this.lower = lower; + this.upper = upper; + } + + @Override + public boolean test(K key) { + if (lower != null) { + int cmp = comparator.compare(lower.getValue(), key); + if (cmp > 0 || (cmp == 0 && !lower.isInclusive())) { + return false; + } + } + if (upper != null) { + int cmp = comparator.compare(upper.getValue(), key); + if (cmp < 0 || (cmp == 0 && !upper.isInclusive())) { + return false; + } + } + return true; + } + + @Override + public BoundInfo getLowerBound() { + return lower; + } + + @Override + public BoundInfo getUpperBound() { + return upper; + } + + @Override + public SavepointKeyFilterPlan intersect(SavepointKeyFilterPlan other) { + if (other.isEmpty()) { + return other; + } + final Set otherExactKeys = other.getExactKeys(); + if (otherExactKeys != null) { + return SavepointKeyFilterPlan.filterKeys(otherExactKeys, this); + } + return intersectRange(other.getLowerBound(), other.getUpperBound()); + } + + private SavepointKeyFilterPlan intersectRange( + @Nullable BoundInfo otherLower, @Nullable BoundInfo otherUpper) { + BoundInfo newLower = tighter(lower, otherLower, true); + BoundInfo newUpper = tighter(upper, otherUpper, false); + + if (newLower != null && newUpper != null) { + int cmp = comparator.compare(newLower.getValue(), newUpper.getValue()); + if (cmp > 0) { + return SavepointKeyFilterPlan.empty(); + } + if (cmp == 0 && (!newLower.isInclusive() || !newUpper.isInclusive())) { + return SavepointKeyFilterPlan.empty(); + } + } + return new RangeKeyFilterPlan<>(comparator, newLower, newUpper); + } + + @Nullable + private BoundInfo tighter( + @Nullable BoundInfo a, @Nullable BoundInfo 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() ? "]" : ")"); + return "RangeKeyFilterPlan" + lowerStr + ", " + upperStr; + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/SavepointKeyFilterPlan.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/SavepointKeyFilterPlan.java new file mode 100644 index 0000000000000..b3ddf00312952 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/SavepointKeyFilterPlan.java @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table.filter; + +import org.apache.flink.annotation.Experimental; +import org.apache.flink.state.api.filter.SavepointKeyFilter; +import org.apache.flink.state.api.filter.SerializableComparator; + +import javax.annotation.Nullable; + +import java.util.HashSet; +import java.util.Set; + +/** + * The internal, richer view of a {@link SavepointKeyFilter} used while combining and analyzing + * filters during push-down translation. + */ +@Experimental +public interface SavepointKeyFilterPlan extends SavepointKeyFilter { + + /** Returns {@code true} if this filter rejects every key. */ + default boolean isEmpty() { + return false; + } + + /** Returns the lower bound of this filter's range, or {@code null} if there is none. */ + @Nullable + default BoundInfo getLowerBound() { + return null; + } + + /** Returns the upper bound of this filter's range, or {@code null} if there is none. */ + @Nullable + default BoundInfo getUpperBound() { + return null; + } + + /** + * Returns a filter that accepts a key if and only if both {@code this} and {@code other} accept + * it. + */ + SavepointKeyFilterPlan intersect(SavepointKeyFilterPlan other); + + static SavepointKeyFilterPlan filterKeys(Set keys, SavepointKeyFilter predicate) { + final Set retained = new HashSet<>(); + for (K key : keys) { + if (predicate.test(key)) { + retained.add(key); + } + } + return exact(retained); + } + + static SavepointKeyFilterPlan exact(Set keys) { + if (keys.isEmpty()) { + return EmptyKeyFilterPlan.instance(); + } + return new ExactKeyFilterPlan<>(keys); + } + + static SavepointKeyFilterPlan exact(K value) { + return new ExactKeyFilterPlan<>(Set.of(value)); + } + + static > SavepointKeyFilterPlan range( + @Nullable K lower, boolean lowerInclusive, @Nullable K upper, boolean upperInclusive) { + return range(lower, lowerInclusive, upper, upperInclusive, new NaturalOrderComparator<>()); + } + + static SavepointKeyFilterPlan range( + @Nullable K lower, + boolean lowerInclusive, + @Nullable K upper, + boolean upperInclusive, + SerializableComparator comparator) { + BoundInfo lowerBoundInfo = lower != null ? new BoundInfo<>(lower, lowerInclusive) : null; + BoundInfo upperBoundInfo = upper != null ? new BoundInfo<>(upper, upperInclusive) : null; + return new RangeKeyFilterPlan<>(comparator, lowerBoundInfo, upperBoundInfo); + } + + static SavepointKeyFilterPlan empty() { + return EmptyKeyFilterPlan.instance(); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointReaderKeyedStateITCase.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointReaderKeyedStateITCase.java index 41dbc4e4b1148..c44169bfea0e8 100644 --- a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointReaderKeyedStateITCase.java +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/SavepointReaderKeyedStateITCase.java @@ -31,6 +31,7 @@ import org.apache.flink.state.api.functions.KeyedStateReaderFunction; import org.apache.flink.state.api.utils.JobResultRetriever; import org.apache.flink.state.api.utils.SavepointTestBase; +import org.apache.flink.state.table.filter.SavepointKeyFilterPlan; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.KeyedProcessFunction; @@ -220,7 +221,7 @@ public void testReadKeyedStateWithEmptyFilter() throws Exception { SavepointReader savepoint = SavepointReader.read(env, savepointPath, backendTuple.f1); CountingReadResult result = - readKeyedStateWithCountingReader(savepoint, SavepointKeyFilter.empty()); + readKeyedStateWithCountingReader(savepoint, SavepointKeyFilterPlan.empty()); // No key reaches the reader, so no state is read. assertThat(result.values).isEmpty(); assertThat(result.counter).isZero(); diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/KeyedStateInputFormatTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/KeyedStateInputFormatTest.java index 292eb61d2c44a..9f32c1db2d50a 100644 --- a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/KeyedStateInputFormatTest.java +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/input/KeyedStateInputFormatTest.java @@ -143,29 +143,6 @@ void testExactKeyFilterPrunesInputSplits(boolean asyncState) throws Exception { .hasSize(1); } - @ParameterizedTest(name = "Enable async state = {0}") - @ValueSource(booleans = {false, true}) - void testEmptyFilterProducesNoInputSplits(boolean asyncState) throws Exception { - OperatorID operatorID = OperatorIDGenerator.fromUid("uid"); - - OperatorSubtaskState state = - createOperatorSubtaskState(createFlatMap(asyncState), asyncState); - OperatorState operatorState = new OperatorState(null, null, operatorID, 1, 128); - operatorState.putState(0, state); - - KeyedStateInputFormat format = - new KeyedStateInputFormat<>( - operatorState, - new HashMapStateBackend(), - new Configuration(), - new KeyedStateReaderOperator<>(new ReaderFunction(), Types.INT), - new ExecutionConfig(), - SavepointKeyFilter.empty()); - KeyGroupRangeInputSplit[] splits = format.createInputSplits(10); - - assertThat(splits).isEmpty(); - } - @ParameterizedTest(name = "Enable async state = {0}") @ValueSource(booleans = {false, true}) void testRangeFilterDoesNotPruneInputSplits(boolean asyncState) throws Exception { diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointFilterTranslatorTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointFilterTranslatorTest.java index 3e490b971602f..8a5fbccc3e7a0 100644 --- a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointFilterTranslatorTest.java +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointFilterTranslatorTest.java @@ -19,6 +19,7 @@ package org.apache.flink.state.table; import org.apache.flink.state.api.filter.SavepointKeyFilter; +import org.apache.flink.state.table.filter.SavepointKeyFilterPlan; import org.apache.flink.table.api.DataTypes; import org.apache.flink.table.expressions.CallExpression; import org.apache.flink.table.expressions.FieldReferenceExpression; @@ -53,7 +54,7 @@ class SavepointFilterTranslatorTest { @Test void equalsKeyOnLeft() { - SavepointKeyFilter filter = keyFilterOf(eq(longKeyRef(), longLit(42L))); + SavepointKeyFilterPlan filter = keyFilterOf(eq(longKeyRef(), longLit(42L))); assertNotNull(filter); assertThat(filter.getExactKeys()).containsExactly(42L); assertThat(filter.test(42L)).isTrue(); @@ -62,7 +63,7 @@ void equalsKeyOnLeft() { @Test void equalsKeyOnRight() { - SavepointKeyFilter filter = keyFilterOf(eq(longLit(42L), longKeyRef())); + SavepointKeyFilterPlan filter = keyFilterOf(eq(longLit(42L), longKeyRef())); assertNotNull(filter); assertThat(filter.getExactKeys()).containsExactly(42L); assertThat(filter.test(42L)).isTrue(); @@ -71,13 +72,13 @@ void equalsKeyOnRight() { @Test void equalsNeitherSideIsKeyColumn_returnsNull() { - SavepointKeyFilter filter = keyFilterOf(eq(otherRef(), longLit(42L))); + SavepointKeyFilterPlan filter = keyFilterOf(eq(otherRef(), longLit(42L))); assertThat(filter).isNull(); } @Test void equalsNeitherSideIsLiteral_returnsNull() { - SavepointKeyFilter filter = keyFilterOf(eq(longKeyRef(), otherRef())); + SavepointKeyFilterPlan filter = keyFilterOf(eq(longKeyRef(), otherRef())); assertThat(filter).isNull(); } @@ -93,7 +94,7 @@ void orOfEqualsProducesMergedExactFilter() { eq(longKeyRef(), longLit(2L)), eq(longLit(3L), longKeyRef())); - SavepointKeyFilter filter = keyFilterOf(expr); + SavepointKeyFilterPlan filter = keyFilterOf(expr); assertNotNull(filter); assertThat(filter.getExactKeys()).containsExactlyInAnyOrder(1L, 2L, 3L); assertThat(filter.test(4L)).isFalse(); @@ -112,7 +113,7 @@ void orWithNonPushableChild_returnsNull() { @Test void betweenProducesInclusiveRange() { - SavepointKeyFilter filter = + SavepointKeyFilterPlan filter = keyFilterOf(between(longKeyRef(), longLit(10L), longLit(20L))); assertNotNull(filter); @@ -126,7 +127,7 @@ void betweenProducesInclusiveRange() { @Test void betweenWithNonKeyField_returnsNull() { - SavepointKeyFilter filter = + SavepointKeyFilterPlan filter = keyFilterOf(between(otherRef(), longLit(1L), longLit(10L))); assertThat(filter).isNull(); } @@ -137,7 +138,7 @@ void betweenWithNonKeyField_returnsNull() { @Test void greaterThanProducesExclusiveLowerBound() { - SavepointKeyFilter filter = keyFilterOf(gt(longKeyRef(), longLit(5L))); + SavepointKeyFilterPlan filter = keyFilterOf(gt(longKeyRef(), longLit(5L))); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); assertThat(filter.test(5L)).isFalse(); @@ -146,7 +147,7 @@ void greaterThanProducesExclusiveLowerBound() { @Test void greaterThanOrEqualProducesInclusiveLowerBound() { - SavepointKeyFilter filter = keyFilterOf(gte(longKeyRef(), longLit(5L))); + SavepointKeyFilterPlan filter = keyFilterOf(gte(longKeyRef(), longLit(5L))); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); assertThat(filter.test(4L)).isFalse(); @@ -156,7 +157,7 @@ void greaterThanOrEqualProducesInclusiveLowerBound() { @Test void lessThanProducesExclusiveUpperBound() { - SavepointKeyFilter filter = keyFilterOf(lt(longKeyRef(), longLit(10L))); + SavepointKeyFilterPlan filter = keyFilterOf(lt(longKeyRef(), longLit(10L))); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); assertThat(filter.test(9L)).isTrue(); @@ -165,7 +166,7 @@ void lessThanProducesExclusiveUpperBound() { @Test void lessThanOrEqualProducesInclusiveUpperBound() { - SavepointKeyFilter filter = keyFilterOf(lte(longKeyRef(), longLit(10L))); + SavepointKeyFilterPlan filter = keyFilterOf(lte(longKeyRef(), longLit(10L))); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); assertThat(filter.test(10L)).isTrue(); @@ -175,7 +176,7 @@ void lessThanOrEqualProducesInclusiveUpperBound() { @Test void comparisonWithLiteralOnLeft_flipsDirection() { // literal > key → key < literal → upper bound (exclusive) - SavepointKeyFilter filter = keyFilterOf(gt(longLit(10L), longKeyRef())); + SavepointKeyFilterPlan filter = keyFilterOf(gt(longLit(10L), longKeyRef())); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); assertThat(filter.test(9L)).isTrue(); @@ -185,7 +186,7 @@ void comparisonWithLiteralOnLeft_flipsDirection() { @Test void comparisonWithLiteralOnLeft_lte_flipsDirection() { // literal <= key → key >= literal → lower bound (inclusive) - SavepointKeyFilter filter = keyFilterOf(lte(longLit(5L), longKeyRef())); + SavepointKeyFilterPlan filter = keyFilterOf(lte(longLit(5L), longKeyRef())); assertThat(filter.getExactKeys()).isNull(); assertThat(filter.test(4L)).isFalse(); assertThat(filter.test(5L)).isTrue(); @@ -199,7 +200,7 @@ void comparisonWithLiteralOnLeft_lte_flipsDirection() { void andOfTwoRangesProducesIntersection() { // key >= 5 AND key <= 10 CallExpression expr = and(gte(longKeyRef(), longLit(5L)), lte(longKeyRef(), longLit(10L))); - SavepointKeyFilter filter = keyFilterOf(expr); + SavepointKeyFilterPlan filter = keyFilterOf(expr); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); @@ -213,7 +214,7 @@ void andOfTwoRangesProducesIntersection() { void andWithProvablyEmptyIntersection_matchesNothing() { // key > 10 AND key < 5 — disjoint CallExpression expr = and(gt(longKeyRef(), longLit(10L)), lt(longKeyRef(), longLit(5L))); - SavepointKeyFilter filter = keyFilterOf(expr); + SavepointKeyFilterPlan filter = keyFilterOf(expr); assertNotNull(filter); assertThat(filter.isEmpty()).isTrue(); @@ -251,7 +252,7 @@ void unrecognizedFunctionReturnsNull() { @Test void rangeFilterOnStringKey() { - SavepointKeyFilter filter = + SavepointKeyFilterPlan filter = keyFilterOf(between(stringKeyRef(), stringLit("beta"), stringLit("delta"))); assertNotNull(filter); @@ -271,7 +272,8 @@ void rangeFilterWithDoubleComparison() { FieldReferenceExpression floatKey = new FieldReferenceExpression("key", DataTypes.FLOAT().notNull(), 0, KEY_COL); - SavepointKeyFilter filter = keyFilterOf(between(floatKey, floatLower, floatUpper)); + SavepointKeyFilterPlan filter = + keyFilterOf(between(floatKey, floatLower, floatUpper)); assertNotNull(filter); assertThat(filter.test(1.5f)).isTrue(); @@ -288,9 +290,9 @@ void rangeFilterWithDoubleComparison() { @Test void intersectNarrowsBounds() { // [5, ∞) ∩ (-∞, 10] = [5, 10] - SavepointKeyFilter lower = SavepointKeyFilter.range(5L, true, null, true); - SavepointKeyFilter upper = SavepointKeyFilter.range(null, true, 10L, true); - SavepointKeyFilter result = lower.intersect(upper); + SavepointKeyFilterPlan lower = SavepointKeyFilterPlan.range(5L, true, null, true); + SavepointKeyFilterPlan upper = SavepointKeyFilterPlan.range(null, true, 10L, true); + SavepointKeyFilterPlan result = lower.intersect(upper); assertThat(result.isEmpty()).isFalse(); assertThat(result.getExactKeys()).isNull(); @@ -303,17 +305,17 @@ void intersectNarrowsBounds() { @Test void intersectDisjointRangesReturnsEmpty() { // [10, ∞) ∩ (-∞, 5] — disjoint - SavepointKeyFilter a = SavepointKeyFilter.range(10L, true, null, true); - SavepointKeyFilter b = SavepointKeyFilter.range(null, true, 5L, true); + SavepointKeyFilterPlan a = SavepointKeyFilterPlan.range(10L, true, null, true); + SavepointKeyFilterPlan b = SavepointKeyFilterPlan.range(null, true, 5L, true); assertThat(a.intersect(b).isEmpty()).isTrue(); } @Test void intersectEqualBoundsInclusiveIsNonEmpty() { // [7, ∞) ∩ (-∞, 7] = [7, 7] - SavepointKeyFilter a = SavepointKeyFilter.range(7L, true, null, true); - SavepointKeyFilter b = SavepointKeyFilter.range(null, true, 7L, true); - SavepointKeyFilter result = a.intersect(b); + SavepointKeyFilterPlan a = SavepointKeyFilterPlan.range(7L, true, null, true); + SavepointKeyFilterPlan b = SavepointKeyFilterPlan.range(null, true, 7L, true); + SavepointKeyFilterPlan result = a.intersect(b); assertThat(result.isEmpty()).isFalse(); assertThat(result.test(7L)).isTrue(); assertThat(result.test(6L)).isFalse(); @@ -323,8 +325,8 @@ void intersectEqualBoundsInclusiveIsNonEmpty() { @Test void intersectEqualBoundsOneExclusiveIsEmpty() { // (7, ∞) ∩ (-∞, 7] — empty because lower is exclusive - SavepointKeyFilter a = SavepointKeyFilter.range(7L, false, null, true); - SavepointKeyFilter b = SavepointKeyFilter.range(null, true, 7L, true); + SavepointKeyFilterPlan a = SavepointKeyFilterPlan.range(7L, false, null, true); + SavepointKeyFilterPlan b = SavepointKeyFilterPlan.range(null, true, 7L, true); assertThat(a.intersect(b).isEmpty()).isTrue(); } @@ -336,7 +338,7 @@ void intersectEqualBoundsOneExclusiveIsEmpty() { void rangeWithCustomComparatorIsUsed() { // Orders strings by length — clearly not the natural String order. SavepointKeyFilter filter = - SavepointKeyFilter.range( + SavepointKeyFilterPlan.range( "aa", true, "cccc", @@ -359,7 +361,7 @@ void applyAccumulatesRangeAndRange() { apply( List.of(gte(longKeyRef(), longLit(3L)), lte(longKeyRef(), longLit(8L))), LONG_KEY_TYPE); - SavepointKeyFilter result = applied.keyFilter(); + SavepointKeyFilterPlan result = applied.keyFilter(); assertNotNull(result); assertThat(applied.accepted()).hasSize(2); @@ -384,7 +386,7 @@ void applyAccumulatesExactAndExact() { eq(longKeyRef(), longLit(3L)), eq(longKeyRef(), longLit(4L)))), LONG_KEY_TYPE); - SavepointKeyFilter result = applied.keyFilter(); + SavepointKeyFilterPlan result = applied.keyFilter(); assertNotNull(result); assertThat(applied.accepted()).hasSize(2); @@ -398,7 +400,7 @@ void applyAccumulatesExactAndExactEmptyResult_matchesNothing() { apply( List.of(eq(longKeyRef(), longLit(1L)), eq(longKeyRef(), longLit(2L))), LONG_KEY_TYPE); - SavepointKeyFilter result = applied.keyFilter(); + SavepointKeyFilterPlan result = applied.keyFilter(); assertNotNull(result); assertThat(applied.accepted()).hasSize(2); @@ -418,7 +420,7 @@ void applyAccumulatesExactAndRange_keepsOnlyKeysInRange() { eq(longKeyRef(), longLit(15L))), between(longKeyRef(), longLit(4L), longLit(12L))), LONG_KEY_TYPE); - SavepointKeyFilter result = applied.keyFilter(); + SavepointKeyFilterPlan result = applied.keyFilter(); assertNotNull(result); assertThat(applied.accepted()).hasSize(2); @@ -438,7 +440,7 @@ void applyAccumulatesRangeAndExact_keepsOnlyKeysInRange() { eq(longKeyRef(), longLit(10L)), eq(longKeyRef(), longLit(15L)))), LONG_KEY_TYPE); - SavepointKeyFilter result = applied.keyFilter(); + SavepointKeyFilterPlan result = applied.keyFilter(); assertNotNull(result); assertThat(applied.accepted()).hasSize(2); @@ -452,7 +454,7 @@ void applyAccumulatesRangeAndExact_keepsOnlyKeysInRange() { @Test void emptyKeyFilter_rejectsEverything() { - SavepointKeyFilter empty = SavepointKeyFilter.empty(); + SavepointKeyFilterPlan empty = SavepointKeyFilterPlan.empty(); assertThat(empty.isEmpty()).isTrue(); assertThat(empty.getExactKeys()).isEmpty(); assertThat(empty.test(42L)).isFalse(); @@ -461,13 +463,14 @@ void emptyKeyFilter_rejectsEverything() { @Test void exactWithEmptySetReturnsEmptyKeyFilter() { - SavepointKeyFilter filter = SavepointKeyFilter.exact(Collections.emptySet()); + SavepointKeyFilterPlan filter = + SavepointKeyFilterPlan.exact(Collections.emptySet()); assertThat(filter.isEmpty()).isTrue(); } @Test void emptyKeyFilterSingletonPreservedAcrossSerialization() throws Exception { - SavepointKeyFilter original = SavepointKeyFilter.empty(); + SavepointKeyFilterPlan original = SavepointKeyFilterPlan.empty(); ByteArrayOutputStream baos = new ByteArrayOutputStream(); try (ObjectOutputStream oos = new ObjectOutputStream(baos)) { oos.writeObject(original); @@ -477,7 +480,7 @@ void emptyKeyFilterSingletonPreservedAcrossSerialization() throws Exception { new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray()))) { deserialized = ois.readObject(); } - assertThat(deserialized).isSameAs(SavepointKeyFilter.empty()); + assertThat(deserialized).isSameAs(SavepointKeyFilterPlan.empty()); } // ------------------------------------------------------------------------- @@ -504,7 +507,7 @@ void andOfThreeRangesProducesIntersection() { gte(longKeyRef(), longLit(3L)), lte(longKeyRef(), longLit(20L)), lt(longKeyRef(), longLit(10L))); - SavepointKeyFilter filter = keyFilterOf(expr); + SavepointKeyFilterPlan filter = keyFilterOf(expr); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); @@ -522,7 +525,7 @@ void andOfThreeRangesProducesIntersection() { @Test void orWithSingleChild_returnsExactFilter() { CallExpression expr = or(eq(longKeyRef(), longLit(7L))); - SavepointKeyFilter filter = keyFilterOf(expr); + SavepointKeyFilterPlan filter = keyFilterOf(expr); assertNotNull(filter); assertThat(filter.getExactKeys()).containsExactly(7L); @@ -550,7 +553,7 @@ void applyWithEmptyThenRange_returnsEmpty() { and(gt(longKeyRef(), longLit(10L)), lt(longKeyRef(), longLit(5L))), lte(longKeyRef(), longLit(10L))), LONG_KEY_TYPE); - SavepointKeyFilter result = applied.keyFilter(); + SavepointKeyFilterPlan result = applied.keyFilter(); assertNotNull(result); assertThat(applied.accepted()).hasSize(2); @@ -566,7 +569,7 @@ void applyWithRangeThenEmpty_returnsEmpty() { lte(longKeyRef(), longLit(10L)), and(gt(longKeyRef(), longLit(10L)), lt(longKeyRef(), longLit(5L)))), LONG_KEY_TYPE); - SavepointKeyFilter result = applied.keyFilter(); + SavepointKeyFilterPlan result = applied.keyFilter(); assertNotNull(result); assertThat(applied.accepted()).hasSize(2); @@ -582,7 +585,7 @@ void applyWithEmptyThenExact_returnsEmpty() { and(gt(longKeyRef(), longLit(10L)), lt(longKeyRef(), longLit(5L))), eq(longKeyRef(), longLit(1L))), LONG_KEY_TYPE); - SavepointKeyFilter result = applied.keyFilter(); + SavepointKeyFilterPlan result = applied.keyFilter(); assertNotNull(result); assertThat(applied.accepted()).hasSize(2); @@ -596,7 +599,7 @@ void applyWithConflictingExactPredicates_returnsEmptyFilter() { apply( List.of(eq(longKeyRef(), longLit(1L)), eq(longKeyRef(), longLit(2L))), LONG_KEY_TYPE); - SavepointKeyFilter result = applied.keyFilter(); + SavepointKeyFilterPlan result = applied.keyFilter(); assertNotNull(result); assertThat(applied.accepted()).hasSize(2); @@ -662,7 +665,7 @@ void equalsWithWrongArgCount_returnsNull() { @Test void equalsWithIntLiteralIsWidenedToBigintKeyAndPushed() { ValueLiteralExpression intLit = new ValueLiteralExpression(5, DataTypes.INT().notNull()); - SavepointKeyFilter filter = keyFilterOf(eq(longKeyRef(), intLit)); + SavepointKeyFilterPlan filter = keyFilterOf(eq(longKeyRef(), intLit)); assertNotNull(filter); assertThat(filter.getExactKeys()).containsExactly(5L); @@ -674,7 +677,7 @@ void equalsWithIntLiteralIsWidenedToBigintKeyAndPushed() { void betweenWithIntLiteralBoundsIsWidenedToBigintKeyAndPushed() { ValueLiteralExpression lower = new ValueLiteralExpression(1, DataTypes.INT().notNull()); ValueLiteralExpression upper = new ValueLiteralExpression(10, DataTypes.INT().notNull()); - SavepointKeyFilter filter = keyFilterOf(between(longKeyRef(), lower, upper)); + SavepointKeyFilterPlan filter = keyFilterOf(between(longKeyRef(), lower, upper)); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); @@ -690,7 +693,7 @@ void equalsWithIntLiteralIsWidenedToDoubleKeyAndPushed() { new FieldReferenceExpression("key", DataTypes.DOUBLE().notNull(), 0, KEY_COL); ValueLiteralExpression intLit = new ValueLiteralExpression(5, DataTypes.INT().notNull()); - SavepointKeyFilter filter = keyFilterOf(eq(doubleKey, intLit)); + SavepointKeyFilterPlan filter = keyFilterOf(eq(doubleKey, intLit)); assertNotNull(filter); assertThat(filter.getExactKeys()).containsExactly(5.0d); @@ -718,7 +721,7 @@ void decimalKeyEqualityIsPushedDownPreservingLiteralScale() { new ValueLiteralExpression( new BigDecimal("5.00"), DataTypes.DECIMAL(10, 2).notNull()); - SavepointKeyFilter filter = keyFilterOf(eq(decKey, lit)); + SavepointKeyFilterPlan filter = keyFilterOf(eq(decKey, lit)); assertNotNull(filter); // Literal scale is preserved, so exact matching is scale sensitive: 5.00 matches, 5.0 not. @@ -731,7 +734,7 @@ void decimalKeyEqualityIsPushedDownPreservingLiteralScale() { // Expression helpers // ------------------------------------------------------------------------- - private static SavepointKeyFilter keyFilterOf(ResolvedExpression expr) { + private static SavepointKeyFilterPlan keyFilterOf(ResolvedExpression expr) { DataType keyType = findKeyType(expr); return apply(Collections.singletonList(expr), keyType).keyFilter(); } From 3ad85b3349fe65ba90495bd078878befaecc233b Mon Sep 17 00:00:00 2001 From: Ilya Soin Date: Fri, 10 Jul 2026 17:53:18 +0300 Subject: [PATCH 2/2] [FLINK-40116][state] Add conjunction filter and exclusion filter --- .../table/SavepointFilterTranslator.java | 95 +++++++++++++++---- .../filter/ConjunctionKeyFilterPlan.java | 88 +++++++++++++++++ .../table/filter/ExclusionKeyFilterPlan.java | 75 +++++++++++++++ .../table/filter/RangeKeyFilterPlan.java | 12 ++- .../table/filter/SavepointKeyFilterPlan.java | 54 +++++++++++ .../table/SavepointFilterTranslatorTest.java | 8 +- 6 files changed, 309 insertions(+), 23 deletions(-) create mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/ConjunctionKeyFilterPlan.java create mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/ExclusionKeyFilterPlan.java diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointFilterTranslator.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointFilterTranslator.java index ec6af13c49632..3c0b213ecc6b4 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointFilterTranslator.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/SavepointFilterTranslator.java @@ -52,23 +52,37 @@ class SavepointFilterTranslator { FunctionDefinition, BiFunction> FILTERS = - Map.of( - BuiltInFunctionDefinitions.EQUALS, - SavepointFilterTranslator::fromEquals, - BuiltInFunctionDefinitions.OR, - SavepointFilterTranslator::fromOr, - BuiltInFunctionDefinitions.AND, - SavepointFilterTranslator::fromAnd, - BuiltInFunctionDefinitions.BETWEEN, - SavepointFilterTranslator::fromBetween, - BuiltInFunctionDefinitions.GREATER_THAN, - SavepointFilterTranslator::fromGreaterThan, - BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL, - SavepointFilterTranslator::fromGreaterThanOrEqual, - BuiltInFunctionDefinitions.LESS_THAN, - SavepointFilterTranslator::fromLessThan, - BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL, - SavepointFilterTranslator::fromLessThanOrEqual); + Map.ofEntries( + Map.entry( + BuiltInFunctionDefinitions.EQUALS, + SavepointFilterTranslator::fromEquals), + Map.entry( + BuiltInFunctionDefinitions.NOT_EQUALS, + SavepointFilterTranslator::fromNotEquals), + Map.entry( + BuiltInFunctionDefinitions.NOT, + SavepointFilterTranslator::fromNot), + Map.entry( + BuiltInFunctionDefinitions.OR, + SavepointFilterTranslator::fromOr), + Map.entry( + BuiltInFunctionDefinitions.AND, + SavepointFilterTranslator::fromAnd), + Map.entry( + BuiltInFunctionDefinitions.BETWEEN, + SavepointFilterTranslator::fromBetween), + Map.entry( + BuiltInFunctionDefinitions.GREATER_THAN, + SavepointFilterTranslator::fromGreaterThan), + Map.entry( + BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL, + SavepointFilterTranslator::fromGreaterThanOrEqual), + Map.entry( + BuiltInFunctionDefinitions.LESS_THAN, + SavepointFilterTranslator::fromLessThan), + Map.entry( + BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL, + SavepointFilterTranslator::fromLessThanOrEqual)); private final int keyColumnIndex; private final DataType keyColumnType; @@ -137,6 +151,49 @@ private SavepointKeyFilterPlan fromEquals(CallExpression call) { return SavepointKeyFilterPlan.exact(value); } + // ------------------------------------------------------------------------- + // Negation / membership + // ------------------------------------------------------------------------- + + @Nullable + private SavepointKeyFilterPlan fromNotEquals(CallExpression call) { + if (!isBinaryValid(call)) { + return null; + } + ResolvedExpression left = call.getResolvedChildren().get(0); + ResolvedExpression right = call.getResolvedChildren().get(1); + + Object value = null; + if (isKeyField(left)) { + value = extractValue(right); + } else if (isKeyField(right)) { + value = extractValue(left); + } + + if (value == null) { + return null; + } + return SavepointKeyFilterPlan.exclude(value); + } + + @Nullable + private SavepointKeyFilterPlan fromNot(CallExpression call) { + if (call.getResolvedChildren().size() != 1) { + return null; + } + SavepointKeyFilterPlan inner = extractFilter(call.getResolvedChildren().get(0)); + if (inner == null) { + return null; + } + Set keys = inner.getExactKeys(); + // Only NOT of a finite key set is an exclusion; NOT of a range is not pushed + // (will be supported in a future commit). + if (keys == null) { + return null; + } + return SavepointKeyFilterPlan.exclude(keys); + } + @Nullable private SavepointKeyFilterPlan fromOr(CallExpression call) { Set keys = new HashSet<>(); @@ -164,8 +221,8 @@ private SavepointKeyFilterPlan fromAnd(CallExpression call) { SavepointKeyFilterPlan merged = null; for (ResolvedExpression arg : call.getResolvedChildren()) { SavepointKeyFilterPlan sub = extractFilter(arg); - // AND only absorbs range filters; exact (or null) children break pushdown. - if (sub == null || sub.getExactKeys() != null) { + // A non-pushable child breaks pushdown of the whole AND. + if (sub == null) { return null; } merged = (merged == null) ? sub : merged.intersect(sub); diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/ConjunctionKeyFilterPlan.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/ConjunctionKeyFilterPlan.java new file mode 100644 index 0000000000000..2c735f70a6135 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/ConjunctionKeyFilterPlan.java @@ -0,0 +1,88 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table.filter; + +import javax.annotation.Nullable; + +import java.util.List; +import java.util.Set; + +/** + * A filter that accepts a key only if every branch accepts it (logical AND). + * + *

Used as the fallback for filter pairs that cannot be simplified into a single {@link + * ExactKeyFilterPlan} or {@link RangeKeyFilterPlan}, for example a range intersected with an + * exclusion. + */ +final class ConjunctionKeyFilterPlan implements SavepointKeyFilterPlan { + + private static final long serialVersionUID = 7L; + + private final List> branches; + + ConjunctionKeyFilterPlan(List> branches) { + this.branches = List.copyOf(branches); + } + + @Override + public List> getConjunctionBranches() { + return branches; + } + + @Override + public boolean test(K key) { + for (SavepointKeyFilterPlan branch : branches) { + if (!branch.test(key)) { + return false; + } + } + return true; + } + + @Override + @Nullable + public Set getExactKeys() { + // If any branch is a finite key set, the conjunction is at most that set narrowed by the + // other branches — still finite, so key groups can be pruned. + for (SavepointKeyFilterPlan branch : branches) { + final Set keys = branch.getExactKeys(); + if (keys != null) { + return SavepointKeyFilterPlan.filterKeys(keys, this).getExactKeys(); + } + } + return null; + } + + @Override + public SavepointKeyFilterPlan intersect(SavepointKeyFilterPlan other) { + if (other.isEmpty()) { + return other; + } + final Set otherKeys = other.getExactKeys(); + if (otherKeys != null) { + return SavepointKeyFilterPlan.filterKeys(otherKeys, this); + } + return SavepointKeyFilterPlan.conjunction(this, other); + } + + @Override + public String toString() { + return "ConjunctionKeyFilterPlan" + branches; + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/ExclusionKeyFilterPlan.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/ExclusionKeyFilterPlan.java new file mode 100644 index 0000000000000..1264b375bb133 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/ExclusionKeyFilterPlan.java @@ -0,0 +1,75 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.state.table.filter; + +import java.util.HashSet; +import java.util.Set; + +/** + * A filter that rejects a finite set of keys and accepts everything else (logical NOT of a finite + * set, e.g. {@code key <> v} or {@code key NOT IN (..)}). + */ +final class ExclusionKeyFilterPlan implements SavepointKeyFilterPlan { + + private static final long serialVersionUID = 6L; + + private final Set excluded; + + ExclusionKeyFilterPlan(Set excluded) { + this.excluded = Set.copyOf(excluded); + } + + @Override + public boolean test(K key) { + return !excluded.contains(key); + } + + @Override + public Set getExcludedKeys() { + return excluded; + } + + @Override + public SavepointKeyFilterPlan intersect(SavepointKeyFilterPlan other) { + if (other.isEmpty()) { + return other; + } + final Set otherKeys = other.getExactKeys(); + if (otherKeys != null) { + final Set retained = new HashSet<>(otherKeys); + retained.removeAll(excluded); + return SavepointKeyFilterPlan.exact(retained); + } + final Set otherExcluded = other.getExcludedKeys(); + if (otherExcluded != null) { + // NOT a AND NOT b = NOT (a OR b). + final Set merged = new HashSet<>(excluded); + merged.addAll(otherExcluded); + return new ExclusionKeyFilterPlan<>(merged); + } + // A range (or any other predicate) minus a finite set is not a single filter, so keep both + // constraints as a precise per-key conjunction. + return SavepointKeyFilterPlan.conjunction(this, other); + } + + @Override + public String toString() { + return "ExclusionKeyFilterPlan!" + excluded; + } +} diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/RangeKeyFilterPlan.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/RangeKeyFilterPlan.java index 8732f33a445b5..37d76f9cc94f4 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/RangeKeyFilterPlan.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/RangeKeyFilterPlan.java @@ -69,6 +69,11 @@ public BoundInfo getUpperBound() { return upper; } + @Override + public boolean isRange() { + return true; + } + @Override public SavepointKeyFilterPlan intersect(SavepointKeyFilterPlan other) { if (other.isEmpty()) { @@ -78,7 +83,12 @@ public SavepointKeyFilterPlan intersect(SavepointKeyFilterPlan other) { if (otherExactKeys != null) { return SavepointKeyFilterPlan.filterKeys(otherExactKeys, this); } - return intersectRange(other.getLowerBound(), other.getUpperBound()); + if (other.isRange()) { + return intersectRange(other.getLowerBound(), other.getUpperBound()); + } + // Anything else (e.g. an exclusion) intersected with a range is not a single range, so keep + // both constraints as a precise per-key conjunction. + return SavepointKeyFilterPlan.conjunction(this, other); } private SavepointKeyFilterPlan intersectRange( diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/SavepointKeyFilterPlan.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/SavepointKeyFilterPlan.java index b3ddf00312952..6a493fc4b1cb4 100644 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/SavepointKeyFilterPlan.java +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/SavepointKeyFilterPlan.java @@ -24,7 +24,10 @@ import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.Collections; import java.util.HashSet; +import java.util.List; import java.util.Set; /** @@ -51,6 +54,28 @@ default BoundInfo getUpperBound() { return null; } + /** Returns {@code true} if this filter is a single range, reporting its bounds. */ + default boolean isRange() { + return false; + } + + /** + * Returns the finite set of keys this filter rejects while accepting every other key, or {@code + * null} if this filter is not a pure exclusion. + */ + @Nullable + default Set getExcludedKeys() { + return null; + } + + /** + * Decomposes this filter into its top-level conjunctive (AND) branches; a non-conjunction + * filter is a single branch. + */ + default List> getConjunctionBranches() { + return Collections.singletonList(this); + } + /** * Returns a filter that accepts a key if and only if both {@code this} and {@code other} accept * it. @@ -97,4 +122,33 @@ static SavepointKeyFilterPlan range( static SavepointKeyFilterPlan empty() { return EmptyKeyFilterPlan.instance(); } + + /** + * Returns a filter accepting a key if and only if it is not one of the excluded keys + * ({@code key <> v} or {@code key NOT IN (..)}), or {@code null} if nothing is excluded. + */ + @Nullable + static SavepointKeyFilterPlan exclude(Set excluded) { + if (excluded.isEmpty()) { + return null; + } + return new ExclusionKeyFilterPlan<>(excluded); + } + + /** Returns a filter accepting a key if and only if it is not the excluded key. */ + static SavepointKeyFilterPlan exclude(K value) { + return new ExclusionKeyFilterPlan<>(Set.of(value)); + } + + /** Returns a filter accepting a key if and only if both {@code a} and {@code b} accept it. */ + static SavepointKeyFilterPlan conjunction( + SavepointKeyFilterPlan a, SavepointKeyFilterPlan b) { + if (a.isEmpty() || b.isEmpty()) { + return empty(); + } + final List> branches = new ArrayList<>(); + branches.addAll(a.getConjunctionBranches()); + branches.addAll(b.getConjunctionBranches()); + return new ConjunctionKeyFilterPlan<>(branches); + } } diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointFilterTranslatorTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointFilterTranslatorTest.java index 8a5fbccc3e7a0..ddbf15de3b8f4 100644 --- a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointFilterTranslatorTest.java +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointFilterTranslatorTest.java @@ -221,10 +221,12 @@ void andWithProvablyEmptyIntersection_matchesNothing() { } @Test - void andWithExactKeyChildIsNotPushable() { - // AND requires all children to be range filters; exact filter breaks pushdown + void andOfExactAndRangeNarrowsToExactSubset() { + // key = 5 AND key > 3 -> {5} CallExpression expr = and(eq(longKeyRef(), longLit(5L)), gt(longKeyRef(), longLit(3L))); - assertThat(keyFilterOf(expr)).isNull(); + SavepointKeyFilterPlan filter = keyFilterOf(expr); + assertNotNull(filter); + assertThat(filter.getExactKeys()).containsExactly(5L); } // -------------------------------------------------------------------------