diff --git a/docs/content.zh/docs/libs/state_processor_api.md b/docs/content.zh/docs/libs/state_processor_api.md index e0e20bfaf4a3b5..14b63aefa0edb6 100644 --- a/docs/content.zh/docs/libs/state_processor_api.md +++ b/docs/content.zh/docs/libs/state_processor_api.md @@ -254,7 +254,6 @@ DataStream keyRange = savepoint.readKeyedState( * `SavepointKeyFilter.exact(K key)` / `SavepointKeyFilter.exact(Set 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`. Either bound may be `null` to leave that side unbounded. * `SavepointKeyFilter.range(K lower, boolean lowerInclusive, K upper, boolean upperInclusive, SerializableComparator comparator)` — same, but with an explicit comparator for key types that do not implement `Comparable`. 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` 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. @@ -314,8 +313,6 @@ DataStream 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，窗口分配器和聚合类型。 diff --git a/docs/content/docs/libs/state_processor_api.md b/docs/content/docs/libs/state_processor_api.md index 0b0472b7f39e40..6485e31cd5ce5a 100644 --- a/docs/content/docs/libs/state_processor_api.md +++ b/docs/content/docs/libs/state_processor_api.md @@ -268,7 +268,6 @@ DataStream keyRange = savepoint.readKeyedState( * `SavepointKeyFilter.exact(K key)` / `SavepointKeyFilter.exact(Set 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`. Either bound may be `null` to leave that side unbounded. * `SavepointKeyFilter.range(K lower, boolean lowerInclusive, K upper, boolean upperInclusive, SerializableComparator comparator)` — same, but with an explicit comparator for key types that do not implement `Comparable`. 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` 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. @@ -328,8 +327,6 @@ DataStream 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" >}}). 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 13f37554d92679..2cdf677252194e 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 392a06fbf8766d..ccef5909fa2b7b 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 static final long serialVersionUID = 4L; 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 c639c5c056d940..ebf059ec544207 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 f12cdf9a304e96..ec6af13c496320 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 92% 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 fe0b525cd1c7bf..dd1ea742bbeaa7 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,14 +16,14 @@ * limitations under the License. */ -package org.apache.flink.state.api.filter; +package org.apache.flink.state.table.filter; -import org.apache.flink.annotation.Experimental; +import org.apache.flink.annotation.Internal; import java.io.Serializable; /** Information about a bound in a range filter. */ -@Experimental +@Internal public final class BoundInfo implements Serializable { private static final long serialVersionUID = 4L; 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 160cc17d4a519a..ae5502cac8c729 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 00000000000000..23f870367d202b --- /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 00000000000000..8732f33a445b53 --- /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 00000000000000..21fe4f5c07898b --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/table/filter/SavepointKeyFilterPlan.java @@ -0,0 +1,106 @@ +/* + * 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.Internal; +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. + */ +@Internal +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); + + /** Returns a filter matching the subset of {@code keys} that {@code predicate} accepts. */ + 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); + } + + /** Returns a filter accepting only the given keys, or the empty filter if there are none. */ + static SavepointKeyFilterPlan exact(Set keys) { + if (keys.isEmpty()) { + return EmptyKeyFilterPlan.instance(); + } + return new ExactKeyFilterPlan<>(keys); + } + + /** Returns a filter accepting only the given key. */ + static SavepointKeyFilterPlan exact(K value) { + return new ExactKeyFilterPlan<>(Set.of(value)); + } + + /** Returns a range filter over the natural ordering of the key type. */ + static > SavepointKeyFilterPlan range( + @Nullable K lower, boolean lowerInclusive, @Nullable K upper, boolean upperInclusive) { + return range(lower, lowerInclusive, upper, upperInclusive, new NaturalOrderComparator<>()); + } + + /** Returns a range filter ordered by {@code comparator}; a {@code null} bound is unbounded. */ + 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); + } + + /** Returns a filter that rejects every key. */ + 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 00746a9285a94f..75dc9eb4a8f7ca 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; @@ -219,7 +220,7 @@ 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 bde20fd0f579cc..b1183f6bb9116a 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 @@ -55,6 +55,7 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.Set; @@ -138,7 +139,7 @@ void testExactKeyFilterPrunesInputSplits(boolean asyncState) throws Exception { @ParameterizedTest(name = "Enable async state = {0}") @ValueSource(booleans = {false, true}) - void testEmptyFilterProducesNoInputSplits(boolean asyncState) throws Exception { + void testEmptyExactFilterProducesNoInputSplits(boolean asyncState) throws Exception { OperatorID operatorID = OperatorIDGenerator.fromUid("uid"); OperatorSubtaskState state = @@ -153,10 +154,10 @@ void testEmptyFilterProducesNoInputSplits(boolean asyncState) throws Exception { new Configuration(), new KeyedStateReaderOperator<>(new ReaderFunction(), Types.INT), new ExecutionConfig(), - SavepointKeyFilter.empty()); + SavepointKeyFilter.exact(Collections.emptySet())); KeyGroupRangeInputSplit[] splits = format.createInputSplits(10); - assertThat(splits).isEmpty(); + assertThat(splits).as("A filter matching no key leaves nothing to read").isEmpty(); } @ParameterizedTest(name = "Enable async state = {0}") diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointDynamicTableSourceTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointDynamicTableSourceTest.java index e2a8ca574d0ce1..47277d5cd6e130 100644 --- a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointDynamicTableSourceTest.java +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/SavepointDynamicTableSourceTest.java @@ -394,6 +394,32 @@ void testOrAcrossKeyAndNonKeyColumnIsNotPushedDownButReturnsCorrectResult() thro assertThat(result.get(1).getField("k")).isEqualTo(5L); } + @Test + void testOrOfExactAndRangeOnKeyIsNotPushedDownButReturnsCorrectResult() throws Exception { + // The planner hands this over intact as or(equals(k, 1), greaterThan(k, 5)), but OR only + // merges finite key sets, so a range branch makes the whole disjunction non-pushable. + StreamTableEnvironment tEnv = createBatchTableEnv(); + tEnv.executeSql(STATE_TABLE_DDL); + + String sql = "SELECT k FROM state_table WHERE k = 1 OR k > 5 ORDER BY k"; + + assertThat(hasPushedDownFilter(tEnv, sql)).isFalse(); + assertThat(collectKeys(tEnv, sql)).containsExactly(1L, 6L, 7L, 8L, 9L); + } + + @Test + void testOrOfTwoRangesOnKeyIsNotPushedDownButReturnsCorrectResult() throws Exception { + // Same limitation for "outside a range". This is also the shape the planner produces + // when it expands a Sarg, which is why a range combined with <> is not pushed either. + StreamTableEnvironment tEnv = createBatchTableEnv(); + tEnv.executeSql(STATE_TABLE_DDL); + + String sql = "SELECT k FROM state_table WHERE k < 2 OR k > 7 ORDER BY k"; + + assertThat(hasPushedDownFilter(tEnv, sql)).isFalse(); + assertThat(collectKeys(tEnv, sql)).containsExactly(0L, 1L, 8L, 9L); + } + @Test void testUnsupportedFilterIsNotPushedDownButReturnsCorrectResult() throws Exception { StreamTableEnvironment tEnv = createBatchTableEnv(); @@ -410,10 +436,62 @@ void testUnsupportedFilterIsNotPushedDownButReturnsCorrectResult() throws Except assertThat(keys).containsExactly(0L, 2L, 4L, 6L, 8L); } + @Test + void testFilterPushDownUpperBoundReturnsCorrectResult() throws Exception { + StreamTableEnvironment tEnv = createBatchTableEnv(); + tEnv.executeSql(STATE_TABLE_DDL); + + String sql = "SELECT k FROM state_table WHERE k < 3 ORDER BY k"; + + assertThat(hasPushedDownFilter(tEnv, sql)).isTrue(); + assertThat(collectKeys(tEnv, sql)).containsExactly(0L, 1L, 2L); + } + + @Test + void testFilterPushDownStrictLowerBoundReturnsCorrectResult() throws Exception { + StreamTableEnvironment tEnv = createBatchTableEnv(); + tEnv.executeSql(STATE_TABLE_DDL); + + String sql = "SELECT k FROM state_table WHERE k > 7 ORDER BY k"; + + assertThat(hasPushedDownFilter(tEnv, sql)).isTrue(); + assertThat(collectKeys(tEnv, sql)).containsExactly(8L, 9L); + } + + @Test + void testFilterPushDownIntersectingRangesReturnsCorrectResult() throws Exception { + StreamTableEnvironment tEnv = createBatchTableEnv(); + tEnv.executeSql(STATE_TABLE_DDL); + + String sql = "SELECT k FROM state_table WHERE k >= 3 AND k <= 6 ORDER BY k"; + + assertThat(hasPushedDownFilter(tEnv, sql)).isTrue(); + assertThat(collectKeys(tEnv, sql)).containsExactly(3L, 4L, 5L, 6L); + } + + @Test + void testFilterPushDownComparisonWithLiteralOnLeftSide() throws Exception { + // verify that "5 < k" (literal on the left) works the same as "k > 5". + StreamTableEnvironment tEnv = createBatchTableEnv(); + tEnv.executeSql(STATE_TABLE_DDL); + + String sql = "SELECT k FROM state_table WHERE 5 < k ORDER BY k"; + + assertThat(hasPushedDownFilter(tEnv, sql)).isTrue(); + assertThat(collectKeys(tEnv, sql)).containsExactly(6L, 7L, 8L, 9L); + } + // ------------------------------------------------------------------------- // Helpers // ------------------------------------------------------------------------- + private static List collectKeys(StreamTableEnvironment tEnv, String sql) + throws Exception { + return tEnv.toDataStream(tEnv.sqlQuery(sql)).executeAndCollect(100).stream() + .map(r -> (Long) r.getField("k")) + .collect(Collectors.toList()); + } + private static StreamTableEnvironment createBatchTableEnv() { Configuration config = new Configuration(); config.set(RUNTIME_MODE, RuntimeExecutionMode.BATCH); 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 3e490b971602f2..6263d83ceb56f5 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 @@ -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.api.DataTypes; import org.apache.flink.table.expressions.CallExpression; import org.apache.flink.table.expressions.FieldReferenceExpression; @@ -29,10 +29,6 @@ import org.junit.jupiter.api.Test; -import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; -import java.io.ObjectInputStream; -import java.io.ObjectOutputStream; import java.math.BigDecimal; import java.util.Arrays; import java.util.Collections; @@ -53,7 +49,8 @@ class SavepointFilterTranslatorTest { @Test void equalsKeyOnLeft() { - SavepointKeyFilter filter = keyFilterOf(eq(longKeyRef(), longLit(42L))); + // key = 42 -> {42} + SavepointKeyFilterPlan filter = keyFilterOf(eq(longKeyRef(), longLit(42L))); assertNotNull(filter); assertThat(filter.getExactKeys()).containsExactly(42L); assertThat(filter.test(42L)).isTrue(); @@ -62,7 +59,8 @@ void equalsKeyOnLeft() { @Test void equalsKeyOnRight() { - SavepointKeyFilter filter = keyFilterOf(eq(longLit(42L), longKeyRef())); + // 42 = key -> {42} + SavepointKeyFilterPlan filter = keyFilterOf(eq(longLit(42L), longKeyRef())); assertNotNull(filter); assertThat(filter.getExactKeys()).containsExactly(42L); assertThat(filter.test(42L)).isTrue(); @@ -71,13 +69,15 @@ void equalsKeyOnRight() { @Test void equalsNeitherSideIsKeyColumn_returnsNull() { - SavepointKeyFilter filter = keyFilterOf(eq(otherRef(), longLit(42L))); + // val = 42 -> not the key column, not pushed + SavepointKeyFilterPlan filter = keyFilterOf(eq(otherRef(), longLit(42L))); assertThat(filter).isNull(); } @Test void equalsNeitherSideIsLiteral_returnsNull() { - SavepointKeyFilter filter = keyFilterOf(eq(longKeyRef(), otherRef())); + // key = val -> no literal to match, not pushed + SavepointKeyFilterPlan filter = keyFilterOf(eq(longKeyRef(), otherRef())); assertThat(filter).isNull(); } @@ -87,13 +87,14 @@ void equalsNeitherSideIsLiteral_returnsNull() { @Test void orOfEqualsProducesMergedExactFilter() { + // key = 1 OR key = 2 OR 3 = key -> {1, 2, 3} CallExpression expr = or( eq(longKeyRef(), longLit(1L)), 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,8 @@ void orWithNonPushableChild_returnsNull() { @Test void betweenProducesInclusiveRange() { - SavepointKeyFilter filter = + // key BETWEEN 10 AND 20 -> [10, 20] + SavepointKeyFilterPlan filter = keyFilterOf(between(longKeyRef(), longLit(10L), longLit(20L))); assertNotNull(filter); @@ -126,7 +128,8 @@ void betweenProducesInclusiveRange() { @Test void betweenWithNonKeyField_returnsNull() { - SavepointKeyFilter filter = + // val BETWEEN 1 AND 10 -> not the key column, not pushed + SavepointKeyFilterPlan filter = keyFilterOf(between(otherRef(), longLit(1L), longLit(10L))); assertThat(filter).isNull(); } @@ -137,7 +140,8 @@ void betweenWithNonKeyField_returnsNull() { @Test void greaterThanProducesExclusiveLowerBound() { - SavepointKeyFilter filter = keyFilterOf(gt(longKeyRef(), longLit(5L))); + // key > 5 -> (5, +∞) + SavepointKeyFilterPlan filter = keyFilterOf(gt(longKeyRef(), longLit(5L))); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); assertThat(filter.test(5L)).isFalse(); @@ -146,7 +150,8 @@ void greaterThanProducesExclusiveLowerBound() { @Test void greaterThanOrEqualProducesInclusiveLowerBound() { - SavepointKeyFilter filter = keyFilterOf(gte(longKeyRef(), longLit(5L))); + // key >= 5 -> [5, +∞) + SavepointKeyFilterPlan filter = keyFilterOf(gte(longKeyRef(), longLit(5L))); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); assertThat(filter.test(4L)).isFalse(); @@ -156,7 +161,8 @@ void greaterThanOrEqualProducesInclusiveLowerBound() { @Test void lessThanProducesExclusiveUpperBound() { - SavepointKeyFilter filter = keyFilterOf(lt(longKeyRef(), longLit(10L))); + // key < 10 -> (-∞, 10) + SavepointKeyFilterPlan filter = keyFilterOf(lt(longKeyRef(), longLit(10L))); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); assertThat(filter.test(9L)).isTrue(); @@ -165,7 +171,8 @@ void lessThanProducesExclusiveUpperBound() { @Test void lessThanOrEqualProducesInclusiveUpperBound() { - SavepointKeyFilter filter = keyFilterOf(lte(longKeyRef(), longLit(10L))); + // key <= 10 -> (-∞, 10] + SavepointKeyFilterPlan filter = keyFilterOf(lte(longKeyRef(), longLit(10L))); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); assertThat(filter.test(10L)).isTrue(); @@ -175,7 +182,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 +192,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 +206,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 +220,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(); @@ -232,11 +239,13 @@ void andWithExactKeyChildIsNotPushable() { @Test void nonCallExpressionReturnsNull() { + // A bare field reference is not a predicate, not pushed assertThat(keyFilterOf(longKeyRef())).isNull(); } @Test void unrecognizedFunctionReturnsNull() { + // key IS NULL -> unsupported function, not pushed CallExpression isNull = CallExpression.permanent( BuiltInFunctionDefinitions.IS_NULL, @@ -251,7 +260,8 @@ void unrecognizedFunctionReturnsNull() { @Test void rangeFilterOnStringKey() { - SavepointKeyFilter filter = + // key BETWEEN 'beta' AND 'delta' -> natural String order + SavepointKeyFilterPlan filter = keyFilterOf(between(stringKeyRef(), stringLit("beta"), stringLit("delta"))); assertNotNull(filter); @@ -264,6 +274,7 @@ void rangeFilterOnStringKey() { @Test void rangeFilterWithDoubleComparison() { + // key BETWEEN 1.5 AND 3.5 on a FLOAT key -> [1.5, 3.5] ValueLiteralExpression floatLower = new ValueLiteralExpression(1.5f, DataTypes.FLOAT().notNull()); ValueLiteralExpression floatUpper = @@ -271,7 +282,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(); @@ -281,85 +293,18 @@ void rangeFilterWithDoubleComparison() { assertThat(filter.test(4.0f)).isFalse(); } - // ------------------------------------------------------------------------- - // Range intersection - // ------------------------------------------------------------------------- - - @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); - - assertThat(result.isEmpty()).isFalse(); - assertThat(result.getExactKeys()).isNull(); - assertThat(result.test(4L)).isFalse(); - assertThat(result.test(5L)).isTrue(); - assertThat(result.test(10L)).isTrue(); - assertThat(result.test(11L)).isFalse(); - } - - @Test - void intersectDisjointRangesReturnsEmpty() { - // [10, ∞) ∩ (-∞, 5] — disjoint - SavepointKeyFilter a = SavepointKeyFilter.range(10L, true, null, true); - SavepointKeyFilter b = SavepointKeyFilter.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); - assertThat(result.isEmpty()).isFalse(); - assertThat(result.test(7L)).isTrue(); - assertThat(result.test(6L)).isFalse(); - assertThat(result.test(8L)).isFalse(); - } - - @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); - assertThat(a.intersect(b).isEmpty()).isTrue(); - } - - // ------------------------------------------------------------------------- - // Custom comparator - // ------------------------------------------------------------------------- - - @Test - void rangeWithCustomComparatorIsUsed() { - // Orders strings by length — clearly not the natural String order. - SavepointKeyFilter filter = - SavepointKeyFilter.range( - "aa", - true, - "cccc", - true, - (a, b) -> Integer.compare(a.length(), b.length())); - - // Length in [2, 4]: "abc" (3), passes; "a" (1) and "ccccc" (5), fail. - assertThat(filter.test("abc")).isTrue(); - assertThat(filter.test("a")).isFalse(); - assertThat(filter.test("ccccc")).isFalse(); - } - // ------------------------------------------------------------------------- // SavepointFilters.apply — intersection handling // ------------------------------------------------------------------------- @Test void applyAccumulatesRangeAndRange() { + // key >= 3, key <= 8 -> [3, 8] SavepointFilterTranslator.Result applied = 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); @@ -372,6 +317,7 @@ void applyAccumulatesRangeAndRange() { @Test void applyAccumulatesExactAndExact() { + // key IN (1, 2, 3), key IN (2, 3, 4) -> {2, 3} SavepointFilterTranslator.Result applied = apply( List.of( @@ -384,7 +330,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); @@ -394,11 +340,12 @@ void applyAccumulatesExactAndExact() { @Test void applyAccumulatesExactAndExactEmptyResult_matchesNothing() { + // key = 1, key = 2 -> disjoint, so nothing matches SavepointFilterTranslator.Result applied = 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); @@ -408,6 +355,7 @@ void applyAccumulatesExactAndExactEmptyResult_matchesNothing() { @Test void applyAccumulatesExactAndRange_keepsOnlyKeysInRange() { + // key IN (1, 5, 10, 15), key BETWEEN 4 AND 12 -> {5, 10} SavepointFilterTranslator.Result applied = apply( List.of( @@ -418,7 +366,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); @@ -428,6 +376,7 @@ void applyAccumulatesExactAndRange_keepsOnlyKeysInRange() { @Test void applyAccumulatesRangeAndExact_keepsOnlyKeysInRange() { + // key BETWEEN 4 AND 12, key IN (1, 5, 10, 15) -> {5, 10}, operands swapped SavepointFilterTranslator.Result applied = apply( List.of( @@ -438,7 +387,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); @@ -446,52 +395,6 @@ void applyAccumulatesRangeAndExact_keepsOnlyKeysInRange() { assertThat(result.getExactKeys()).containsExactlyInAnyOrder(5L, 10L); } - // ------------------------------------------------------------------------- - // Empty key filter - // ------------------------------------------------------------------------- - - @Test - void emptyKeyFilter_rejectsEverything() { - SavepointKeyFilter empty = SavepointKeyFilter.empty(); - assertThat(empty.isEmpty()).isTrue(); - assertThat(empty.getExactKeys()).isEmpty(); - assertThat(empty.test(42L)).isFalse(); - assertThat(empty.test("hello")).isFalse(); - } - - @Test - void exactWithEmptySetReturnsEmptyKeyFilter() { - SavepointKeyFilter filter = SavepointKeyFilter.exact(Collections.emptySet()); - assertThat(filter.isEmpty()).isTrue(); - } - - @Test - void emptyKeyFilterSingletonPreservedAcrossSerialization() throws Exception { - SavepointKeyFilter original = SavepointKeyFilter.empty(); - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - try (ObjectOutputStream oos = new ObjectOutputStream(baos)) { - oos.writeObject(original); - } - Object deserialized; - try (ObjectInputStream ois = - new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray()))) { - deserialized = ois.readObject(); - } - assertThat(deserialized).isSameAs(SavepointKeyFilter.empty()); - } - - // ------------------------------------------------------------------------- - // Exact key filter — single-value factory - // ------------------------------------------------------------------------- - - @Test - void exactSingleValueFactory() { - SavepointKeyFilter filter = SavepointKeyFilter.exact(42L); - assertThat(filter.getExactKeys()).containsExactly(42L); - assertThat(filter.test(42L)).isTrue(); - assertThat(filter.test(43L)).isFalse(); - } - // ------------------------------------------------------------------------- // AND with 3+ children // ------------------------------------------------------------------------- @@ -504,7 +407,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(); @@ -521,8 +424,9 @@ void andOfThreeRangesProducesIntersection() { @Test void orWithSingleChild_returnsExactFilter() { + // OR (key = 7) -> {7} CallExpression expr = or(eq(longKeyRef(), longLit(7L))); - SavepointKeyFilter filter = keyFilterOf(expr); + SavepointKeyFilterPlan filter = keyFilterOf(expr); assertNotNull(filter); assertThat(filter.getExactKeys()).containsExactly(7L); @@ -534,6 +438,7 @@ void orWithSingleChild_returnsExactFilter() { @Test void comparisonWithNeitherSideBeingKeyColumn_returnsNull() { + // val > 5 and 5 < val -> not the key column, not pushed assertThat(keyFilterOf(gt(otherRef(), longLit(5L)))).isNull(); assertThat(keyFilterOf(lt(longLit(5L), otherRef()))).isNull(); } @@ -544,13 +449,14 @@ void comparisonWithNeitherSideBeingKeyColumn_returnsNull() { @Test void applyWithEmptyThenRange_returnsEmpty() { + // (key > 10 AND key < 5) -> empty, key <= 10 -> empty absorbs the range SavepointFilterTranslator.Result applied = apply( List.of( 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); @@ -560,13 +466,14 @@ void applyWithEmptyThenRange_returnsEmpty() { @Test void applyWithRangeThenEmpty_returnsEmpty() { + // key <= 10, (key > 10 AND key < 5) -> empty, operands swapped SavepointFilterTranslator.Result applied = apply( List.of( 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); @@ -576,13 +483,14 @@ void applyWithRangeThenEmpty_returnsEmpty() { @Test void applyWithEmptyThenExact_returnsEmpty() { + // (key > 10 AND key < 5) -> empty, key = 1 -> empty absorbs the exact set SavepointFilterTranslator.Result applied = apply( List.of( 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); @@ -592,11 +500,12 @@ void applyWithEmptyThenExact_returnsEmpty() { @Test void applyWithConflictingExactPredicates_returnsEmptyFilter() { + // key = 1, key = 2 -> disjoint, so nothing matches SavepointFilterTranslator.Result applied = 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); @@ -610,6 +519,7 @@ void applyWithConflictingExactPredicates_returnsEmptyFilter() { @Test void betweenWithWrongArgCount_returnsNull() { + // BETWEEN with two children is malformed, not pushed CallExpression malformed = CallExpression.permanent( BuiltInFunctionDefinitions.BETWEEN, @@ -624,6 +534,7 @@ void betweenWithWrongArgCount_returnsNull() { @Test void betweenWithNonLiteralBound_returnsNull() { + // key BETWEEN val AND 10 -> non-literal bound, not pushed CallExpression expr = CallExpression.permanent( BuiltInFunctionDefinitions.BETWEEN, @@ -638,6 +549,7 @@ void betweenWithNonLiteralBound_returnsNull() { @Test void comparisonWithNonLiteralValue_returnsNull() { + // key > val -> no literal bound, not pushed assertThat(keyFilterOf(gt(longKeyRef(), otherRef()))).isNull(); } @@ -647,6 +559,7 @@ void comparisonWithNonLiteralValue_returnsNull() { @Test void equalsWithWrongArgCount_returnsNull() { + // EQUALS with one child is malformed, not pushed CallExpression malformed = CallExpression.permanent( BuiltInFunctionDefinitions.EQUALS, @@ -661,8 +574,9 @@ void equalsWithWrongArgCount_returnsNull() { @Test void equalsWithIntLiteralIsWidenedToBigintKeyAndPushed() { + // key = 5 (INT literal, BIGINT key) -> {5L} 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); @@ -672,9 +586,10 @@ void equalsWithIntLiteralIsWidenedToBigintKeyAndPushed() { @Test void betweenWithIntLiteralBoundsIsWidenedToBigintKeyAndPushed() { + // key BETWEEN 1 AND 10 (INT literals, BIGINT key) -> [1L, 10L] 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(); @@ -686,11 +601,12 @@ void betweenWithIntLiteralBoundsIsWidenedToBigintKeyAndPushed() { @Test void equalsWithIntLiteralIsWidenedToDoubleKeyAndPushed() { + // key = 5 (INT literal, DOUBLE key) -> {5.0} FieldReferenceExpression doubleKey = 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); @@ -700,11 +616,13 @@ void equalsWithIntLiteralIsWidenedToDoubleKeyAndPushed() { @Test void nonNumericLiteralAgainstNumericKeyIsNotWidenedAndNotPushed() { + // key = '5' (STRING literal, BIGINT key) -> no widening, not pushed assertThat(keyFilterOf(eq(longKeyRef(), stringLit("5")))).isNull(); } @Test void numericLiteralWithNonWidenableKeyTypeIsNotPushed() { + // key = 5L (BIGINT literal, INT key) -> narrowing is unsafe, not pushed FieldReferenceExpression intKey = new FieldReferenceExpression("key", DataTypes.INT().notNull(), 0, KEY_COL); assertThat(keyFilterOf(eq(intKey, longLit(5L)))).isNull(); @@ -712,13 +630,14 @@ void numericLiteralWithNonWidenableKeyTypeIsNotPushed() { @Test void decimalKeyEqualityIsPushedDownPreservingLiteralScale() { + // key = 5.00 on a DECIMAL(10, 2) key -> {5.00} FieldReferenceExpression decKey = new FieldReferenceExpression("key", DataTypes.DECIMAL(10, 2).notNull(), 0, KEY_COL); ValueLiteralExpression lit = 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. @@ -727,11 +646,121 @@ void decimalKeyEqualityIsPushedDownPreservingLiteralScale() { assertThat(filter.test(new BigDecimal("5.0"))).isFalse(); } + // ------------------------------------------------------------------------- + // apply — predicates that cannot be pushed stay in remaining() + // ------------------------------------------------------------------------- + + @Test + void applyKeepsNonPushablePredicatesInRemaining() { + // key = 5, key IS NULL -> only the first is pushed, the second must still be evaluated + SavepointFilterTranslator.Result applied = + apply(List.of(eq(longKeyRef(), longLit(5L)), isNull(longKeyRef())), LONG_KEY_TYPE); + SavepointKeyFilterPlan result = applied.keyFilter(); + + assertNotNull(result); + assertThat(applied.accepted()).hasSize(1); + assertThat(applied.remaining()).hasSize(1); + assertThat(result.getExactKeys()).containsExactly(5L); + } + + @Test + void applyWithOnlyNonPushablePredicateReturnsNoKeyFilter() { + // key IS NULL -> nothing to push, the predicate is handed back untouched + SavepointFilterTranslator.Result applied = + apply(List.of(isNull(longKeyRef())), LONG_KEY_TYPE); + + assertThat(applied.keyFilter()).isNull(); + assertThat(applied.accepted()).isEmpty(); + assertThat(applied.remaining()).hasSize(1); + } + + @Test + void applyWithNoPredicatesReturnsNoKeyFilter() { + // no predicates -> no filter, so the scan is not pruned at all + SavepointFilterTranslator.Result applied = apply(List.of(), LONG_KEY_TYPE); + + assertThat(applied.keyFilter()).isNull(); + assertThat(applied.accepted()).isEmpty(); + assertThat(applied.remaining()).isEmpty(); + } + + // ------------------------------------------------------------------------- + // Untranslatable children break their parent + // ------------------------------------------------------------------------- + + @Test + void andWithUntranslatableChild_returnsNull() { + // key > 5 AND key IS NULL -> the whole AND must not be pushed + assertThat(keyFilterOf(and(gt(longKeyRef(), longLit(5L)), isNull(longKeyRef())))).isNull(); + } + + @Test + void orWithUntranslatableChild_returnsNull() { + // key = 1 OR key IS NULL -> the whole OR must not be pushed + assertThat(keyFilterOf(or(eq(longKeyRef(), longLit(1L)), isNull(longKeyRef())))).isNull(); + } + + // ------------------------------------------------------------------------- + // Literals that cannot be read + // ------------------------------------------------------------------------- + + @Test + void equalsWithNullLiteral_returnsNull() { + // key = NULL -> the literal has no readable value, not pushed + assertThat(keyFilterOf(eq(longKeyRef(), nullLit()))).isNull(); + } + + @Test + void betweenWithNonComparableLiteral_returnsNull() { + // key BETWEEN x'01' AND x'02' on a BYTES key -> byte[] is not Comparable, not pushed + assertThat(keyFilterOf(between(bytesKeyRef(), bytesLit((byte) 1), bytesLit((byte) 2)))) + .isNull(); + } + + @Test + void comparisonWithNonComparableLiteral_returnsNull() { + // key > x'01' on a BYTES key -> byte[] is not Comparable, not pushed + assertThat(keyFilterOf(gt(bytesKeyRef(), bytesLit((byte) 1)))).isNull(); + } + + // ------------------------------------------------------------------------- + // Comparison — remaining flip directions and arity + // ------------------------------------------------------------------------- + + @Test + void comparisonWithLiteralOnLeft_gte_flipsDirection() { + // 10 >= key -> key <= 10 -> upper bound (inclusive) + SavepointKeyFilterPlan filter = keyFilterOf(gte(longLit(10L), longKeyRef())); + assertNotNull(filter); + assertThat(filter.test(10L)).isTrue(); + assertThat(filter.test(11L)).isFalse(); + } + + @Test + void comparisonWithLiteralOnLeft_lt_flipsDirection() { + // 5 < key -> key > 5 -> lower bound (exclusive) + SavepointKeyFilterPlan filter = keyFilterOf(lt(longLit(5L), longKeyRef())); + assertNotNull(filter); + assertThat(filter.test(5L)).isFalse(); + assertThat(filter.test(6L)).isTrue(); + } + + @Test + void comparisonWithWrongArgCount_returnsNull() { + // GREATER_THAN with one child is malformed, not pushed + CallExpression malformed = + CallExpression.permanent( + BuiltInFunctionDefinitions.GREATER_THAN, + Collections.singletonList(longKeyRef()), + DataTypes.BOOLEAN()); + assertThat(keyFilterOf(malformed)).isNull(); + } + // ------------------------------------------------------------------------- // 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(); } @@ -780,6 +809,25 @@ private static CallExpression eq(ResolvedExpression left, ResolvedExpression rig BuiltInFunctionDefinitions.EQUALS, Arrays.asList(left, right), DataTypes.BOOLEAN()); } + private static CallExpression isNull(ResolvedExpression arg) { + return CallExpression.permanent( + BuiltInFunctionDefinitions.IS_NULL, + Collections.singletonList(arg), + DataTypes.BOOLEAN()); + } + + private static ValueLiteralExpression nullLit() { + return new ValueLiteralExpression(null, DataTypes.BIGINT().nullable()); + } + + private static FieldReferenceExpression bytesKeyRef() { + return new FieldReferenceExpression("key", DataTypes.BYTES().notNull(), 0, KEY_COL); + } + + private static ValueLiteralExpression bytesLit(byte value) { + return new ValueLiteralExpression(new byte[] {value}, DataTypes.BYTES().notNull()); + } + private static CallExpression or(ResolvedExpression... args) { return CallExpression.permanent( BuiltInFunctionDefinitions.OR, Arrays.asList(args), DataTypes.BOOLEAN()); diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/filter/SavepointKeyFilterPlanTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/filter/SavepointKeyFilterPlanTest.java new file mode 100644 index 00000000000000..b57bf7ad31637e --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/table/filter/SavepointKeyFilterPlanTest.java @@ -0,0 +1,153 @@ +/* + * 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.SavepointKeyFilter; + +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.util.Collections; + +import static org.assertj.core.api.Assertions.assertThat; + +class SavepointKeyFilterPlanTest { + + // ------------------------------------------------------------------------- + // Range intersection + // ------------------------------------------------------------------------- + + @Test + void intersectNarrowsBounds() { + // [5, ∞) ∩ (-∞, 10] = [5, 10] + 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(); + assertThat(result.test(4L)).isFalse(); + assertThat(result.test(5L)).isTrue(); + assertThat(result.test(10L)).isTrue(); + assertThat(result.test(11L)).isFalse(); + } + + @Test + void intersectDisjointRangesReturnsEmpty() { + // [10, ∞) ∩ (-∞, 5] — disjoint + 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] + 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(); + assertThat(result.test(8L)).isFalse(); + } + + @Test + void intersectEqualBoundsOneExclusiveIsEmpty() { + // (7, ∞) ∩ (-∞, 7] — empty because lower is exclusive + SavepointKeyFilterPlan a = SavepointKeyFilterPlan.range(7L, false, null, true); + SavepointKeyFilterPlan b = SavepointKeyFilterPlan.range(null, true, 7L, true); + assertThat(a.intersect(b).isEmpty()).isTrue(); + } + + // ------------------------------------------------------------------------- + // Custom comparator + // ------------------------------------------------------------------------- + + @Test + void rangeWithCustomComparatorIsUsed() { + // Orders strings by length — clearly not the natural String order. + SavepointKeyFilter filter = + SavepointKeyFilterPlan.range( + "aa", + true, + "cccc", + true, + (a, b) -> Integer.compare(a.length(), b.length())); + + // Length in [2, 4]: "abc" (3), passes; "a" (1) and "ccccc" (5), fail. + assertThat(filter.test("abc")).isTrue(); + assertThat(filter.test("a")).isFalse(); + assertThat(filter.test("ccccc")).isFalse(); + } + + // ------------------------------------------------------------------------- + // Empty key filter + // ------------------------------------------------------------------------- + + @Test + void emptyKeyFilter_rejectsEverything() { + // empty -> matches nothing, of any key type + SavepointKeyFilterPlan empty = SavepointKeyFilterPlan.empty(); + assertThat(empty.isEmpty()).isTrue(); + assertThat(empty.getExactKeys()).isEmpty(); + assertThat(empty.test(42L)).isFalse(); + assertThat(empty.test("hello")).isFalse(); + } + + @Test + void exactWithEmptySetReturnsEmptyKeyFilter() { + // exact({}) -> collapses to empty + SavepointKeyFilterPlan filter = + SavepointKeyFilterPlan.exact(Collections.emptySet()); + assertThat(filter.isEmpty()).isTrue(); + } + + @Test + void emptyKeyFilterSingletonPreservedAcrossSerialization() throws Exception { + // empty is a singleton, so readResolve must return the same instance + SavepointKeyFilterPlan original = SavepointKeyFilterPlan.empty(); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (ObjectOutputStream oos = new ObjectOutputStream(baos)) { + oos.writeObject(original); + } + Object deserialized; + try (ObjectInputStream ois = + new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray()))) { + deserialized = ois.readObject(); + } + assertThat(deserialized).isSameAs(SavepointKeyFilterPlan.empty()); + } + + // ------------------------------------------------------------------------- + // Exact key filter — single-value factory + // ------------------------------------------------------------------------- + + @Test + void exactSingleValueFactory() { + // exact(42) -> {42} + SavepointKeyFilter filter = SavepointKeyFilter.exact(42L); + assertThat(filter.getExactKeys()).containsExactly(42L); + assertThat(filter.test(42L)).isTrue(); + assertThat(filter.test(43L)).isFalse(); + } +}