From 398b563daac6e69d3304f83c139e26082804b877 Mon Sep 17 00:00:00 2001 From: Ilya Soin Date: Mon, 7 Sep 2026 17:32:02 +0300 Subject: [PATCH] [FLINK-40399][state] Separate key filter runtime contract from push-down planning Keep SavepointKeyFilter focused on runtime filtering and move predicate-combination state into SavepointFilterTranslator. Preserve SQL comparison semantics when converting numeric literals for BIGINT and DOUBLE key push-down. --- .../docs/libs/state_processor_api.md | 3 - docs/content/docs/libs/state_processor_api.md | 3 - .../flink/state/api/filter/BoundInfo.java | 45 -- .../state/api/filter/EmptyKeyFilter.java | 67 -- .../state/api/filter/ExactKeyFilter.java | 15 - .../state/api/filter/RangeKeyFilter.java | 88 +-- .../state/api/filter/SavepointKeyFilter.java | 66 +- .../table/SavepointFilterTranslator.java | 179 +++++- .../api/SavepointReaderKeyedStateITCase.java | 3 +- .../state/api/filter/ExactKeyFilterTest.java | 103 ++++ .../state/api/filter/RangeKeyFilterTest.java | 215 +++++++ .../api/input/KeyedStateInputFormatTest.java | 7 +- .../SavepointDynamicTableSourceTest.java | 232 ++++++- .../table/SavepointFilterTranslatorTest.java | 580 ++++++++++++++---- .../table-state-typed-keys/_metadata | Bin 0 -> 26047 bytes 15 files changed, 1179 insertions(+), 427 deletions(-) delete mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/BoundInfo.java delete mode 100644 flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/EmptyKeyFilter.java create mode 100644 flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/filter/ExactKeyFilterTest.java create mode 100644 flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/filter/RangeKeyFilterTest.java create mode 100644 flink-libraries/flink-state-processing-api/src/test/resources/table-state-typed-keys/_metadata 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 336502ab3f2173..6b378cf110b569 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/BoundInfo.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/BoundInfo.java deleted file mode 100644 index fe0b525cd1c7bf..00000000000000 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/BoundInfo.java +++ /dev/null @@ -1,45 +0,0 @@ -/* - * 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.api.filter; - -import org.apache.flink.annotation.Experimental; - -import java.io.Serializable; - -/** Information about a bound in a range filter. */ -@Experimental -public final class BoundInfo implements Serializable { - private static final long serialVersionUID = 4L; - - private final K value; - private final boolean inclusive; - - public BoundInfo(K value, boolean inclusive) { - this.value = value; - this.inclusive = inclusive; - } - - public K getValue() { - return value; - } - - public boolean isInclusive() { - return inclusive; - } -} 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/api/filter/EmptyKeyFilter.java deleted file mode 100644 index 160cc17d4a519a..00000000000000 --- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/filter/EmptyKeyFilter.java +++ /dev/null @@ -1,67 +0,0 @@ -/* - * 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.api.filter; - -import java.util.Collections; -import java.util.Set; - -/** A filter that rejects every key. */ -final class EmptyKeyFilter implements SavepointKeyFilter { - - private static final long serialVersionUID = 1L; - - @SuppressWarnings("rawtypes") - private static final EmptyKeyFilter INSTANCE = new EmptyKeyFilter<>(); - - private EmptyKeyFilter() {} - - @SuppressWarnings("unchecked") - static EmptyKeyFilter instance() { - return (EmptyKeyFilter) INSTANCE; - } - - @Override - public boolean test(K key) { - return false; - } - - @Override - public boolean isEmpty() { - return true; - } - - @Override - public Set getExactKeys() { - return Collections.emptySet(); - } - - @Override - public SavepointKeyFilter intersect(SavepointKeyFilter other) { - return this; - } - - private Object readResolve() { - return INSTANCE; - } - - @Override - public String toString() { - return "EmptyKeyFilter"; - } -} 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 d225accb804dee..e64592811a32cc 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 @@ -35,6 +35,7 @@ import javax.annotation.Nullable; +import java.math.BigDecimal; import java.util.ArrayList; import java.util.HashSet; import java.util.List; @@ -51,10 +52,12 @@ class SavepointFilterTranslator { private static final Logger LOG = LoggerFactory.getLogger(SavepointFilterTranslator.class); + private static final long FLOAT_EXACT_INTEGER_LIMIT = 1L << 24; + private static final long DOUBLE_EXACT_INTEGER_LIMIT = 1L << 53; private static final Map< FunctionDefinition, - BiFunction> + BiFunction> FILTERS = Map.of( BuiltInFunctionDefinitions.EQUALS, @@ -86,9 +89,9 @@ Result apply(List filters) { final List accepted = new ArrayList<>(); final List remaining = new ArrayList<>(); - SavepointKeyFilter keyFilter = null; + KeyFilterPlan keyFilter = null; for (ResolvedExpression filter : filters) { - SavepointKeyFilter extracted = extractFilter(filter); + KeyFilterPlan extracted = extractFilter(filter); if (extracted == null) { remaining.add(filter); continue; @@ -98,7 +101,8 @@ Result apply(List filters) { accepted.add(filter); } - return new Result(accepted, remaining, keyFilter); + return new Result( + accepted, remaining, keyFilter == null ? null : keyFilter.toSavepointKeyFilter()); } /** @@ -123,8 +127,8 @@ static SupportsFilterPushDown.Result applyKeyColumnFilters( } @Nullable - private SavepointKeyFilter extractFilter(ResolvedExpression expr) { - final BiFunction extractor = + private KeyFilterPlan extractFilter(ResolvedExpression expr) { + final BiFunction extractor = expr instanceof CallExpression ? FILTERS.get(((CallExpression) expr).getFunctionDefinition()) : null; @@ -141,7 +145,7 @@ private SavepointKeyFilter extractFilter(ResolvedExpression expr) { // ------------------------------------------------------------------------- @Nullable - private SavepointKeyFilter fromEquals(CallExpression call) { + private KeyFilterPlan fromEquals(CallExpression call) { if (!isBinaryValid(call)) { return null; } @@ -158,25 +162,25 @@ private SavepointKeyFilter fromEquals(CallExpression call) { if (value == null) { return null; } - return SavepointKeyFilter.exact(value); + return KeyFilterPlan.exact(value); } @Nullable - private SavepointKeyFilter fromOr(CallExpression call) { + private KeyFilterPlan fromOr(CallExpression call) { Set keys = new HashSet<>(); for (ResolvedExpression arg : call.getResolvedChildren()) { - SavepointKeyFilter sub = extractFilter(arg); + KeyFilterPlan sub = extractFilter(arg); if (sub == null) { return null; } - Set subKeys = sub.getExactKeys(); + Set subKeys = sub.exactKeys; // OR can only absorb finite key sets; a range branch cannot be merged via union. if (subKeys == null) { return null; } keys.addAll(subKeys); } - return SavepointKeyFilter.exact(keys); + return KeyFilterPlan.exact(keys); } // ------------------------------------------------------------------------- @@ -184,12 +188,12 @@ private SavepointKeyFilter fromOr(CallExpression call) { // ------------------------------------------------------------------------- @Nullable - private SavepointKeyFilter fromAnd(CallExpression call) { - SavepointKeyFilter merged = null; + private KeyFilterPlan fromAnd(CallExpression call) { + KeyFilterPlan merged = null; for (ResolvedExpression arg : call.getResolvedChildren()) { - SavepointKeyFilter sub = extractFilter(arg); + KeyFilterPlan sub = extractFilter(arg); // AND only absorbs range filters; exact (or null) children break pushdown. - if (sub == null || sub.getExactKeys() != null) { + if (sub == null || sub.exactKeys != null) { return null; } merged = (merged == null) ? sub : merged.intersect(sub); @@ -201,7 +205,7 @@ private SavepointKeyFilter fromAnd(CallExpression call) { } @Nullable - private SavepointKeyFilter fromBetween(CallExpression call) { + private KeyFilterPlan fromBetween(CallExpression call) { List args = call.getResolvedChildren(); if (args.size() != 3) { return null; @@ -225,13 +229,11 @@ private SavepointKeyFilter fromBetween(CallExpression call) { lower.getClass().getName()); return null; } - return SavepointKeyFilter.range( - (Comparable) lower, true, - (Comparable) upper, true); + return KeyFilterPlan.range((Comparable) lower, true, (Comparable) upper, true); } @Nullable - private SavepointKeyFilter fromComparison(CallExpression call, Comparison cmp) { + private KeyFilterPlan fromComparison(CallExpression call, Comparison cmp) { if (!isBinaryValid(call)) { return null; } @@ -257,13 +259,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 KeyFilterPlan.range(b, false, null, true); case GTE: - return SavepointKeyFilter.range(b, true, null, true); + return KeyFilterPlan.range(b, true, null, true); case LT: - return SavepointKeyFilter.range(null, true, b, false); + return KeyFilterPlan.range(null, true, b, false); case LTE: - return SavepointKeyFilter.range(null, true, b, true); + return KeyFilterPlan.range(null, true, b, true); default: throw new IllegalStateException("Unknown Comparison: " + keyLeftCmp); } @@ -311,14 +313,29 @@ private Object widenToKeyType(Object value) { } if (value instanceof Number) { if (keyClass == Long.class) { - return ((Number) value).longValue(); - } - if (keyClass == Double.class) { - return ((Number) value).doubleValue(); + final boolean unsafeLongConversion = + (value instanceof Float + && Math.abs((Float) value) >= FLOAT_EXACT_INTEGER_LIMIT) + || (value instanceof Double + && Math.abs((Double) value) >= DOUBLE_EXACT_INTEGER_LIMIT); + if (!unsafeLongConversion) { + try { + return new BigDecimal(value.toString()).longValueExact(); + } catch (NumberFormatException | ArithmeticException lossy) { + // Not a decimal (NaN, Infinity), fractional, or outside the long range: + // fall through and refuse the pushdown. + } + } + } else if (keyClass == Double.class) { + final double converted = ((Number) value).doubleValue(); + if (Double.isFinite(converted)) { + return converted; + } } } LOG.debug( - "Refusing pushdown: literal value {} of type {} cannot be widened to key type {}.", + "Refusing pushdown: literal value {} of type {} cannot be widened to key type {}" + + " without loss.", value, value.getClass().getName(), keyColumnType); @@ -353,6 +370,108 @@ SavepointKeyFilter keyFilter() { } } + private static final class KeyFilterPlan { + @Nullable private final Set exactKeys; + @Nullable private final Comparable lower; + private final boolean lowerInclusive; + @Nullable private final Comparable upper; + private final boolean upperInclusive; + + private KeyFilterPlan( + @Nullable Set exactKeys, + @Nullable Comparable lower, + boolean lowerInclusive, + @Nullable Comparable upper, + boolean upperInclusive) { + this.exactKeys = exactKeys; + this.lower = lower; + this.lowerInclusive = lowerInclusive; + this.upper = upper; + this.upperInclusive = upperInclusive; + } + + private static KeyFilterPlan exact(Object key) { + return exact(Set.of(key)); + } + + private static KeyFilterPlan exact(Set keys) { + return new KeyFilterPlan(Set.copyOf(keys), null, true, null, true); + } + + private static KeyFilterPlan range( + @Nullable Comparable lower, + boolean lowerInclusive, + @Nullable Comparable upper, + boolean upperInclusive) { + return new KeyFilterPlan(null, lower, lowerInclusive, upper, upperInclusive); + } + + private KeyFilterPlan intersect(KeyFilterPlan other) { + Set newExactKeys = exactKeys; + if (newExactKeys == null) { + newExactKeys = other.exactKeys; + } else if (other.exactKeys != null) { + newExactKeys = new HashSet<>(newExactKeys); + newExactKeys.retainAll(other.exactKeys); + } + + Comparable newLower = lower; + boolean newLowerInclusive = lowerInclusive; + if (other.lower != null) { + final int comparison = newLower == null ? -1 : newLower.compareTo(other.lower); + if (newLower == null || comparison < 0) { + newLower = other.lower; + newLowerInclusive = other.lowerInclusive; + } else if (comparison == 0) { + newLowerInclusive &= other.lowerInclusive; + } + } + + Comparable newUpper = upper; + boolean newUpperInclusive = upperInclusive; + if (other.upper != null) { + final int comparison = newUpper == null ? 1 : newUpper.compareTo(other.upper); + if (newUpper == null || comparison > 0) { + newUpper = other.upper; + newUpperInclusive = other.upperInclusive; + } else if (comparison == 0) { + newUpperInclusive &= other.upperInclusive; + } + } + + return new KeyFilterPlan( + newExactKeys, newLower, newLowerInclusive, newUpper, newUpperInclusive); + } + + private boolean isEmpty() { + if (exactKeys != null && exactKeys.isEmpty()) { + return true; + } + if (lower == null || upper == null) { + return false; + } + final int comparison = lower.compareTo(upper); + return comparison > 0 || (comparison == 0 && (!lowerInclusive || !upperInclusive)); + } + + private SavepointKeyFilter toSavepointKeyFilter() { + if (isEmpty()) { + return SavepointKeyFilter.exact(Set.of()); + } + if (exactKeys != null && lower == null && upper == null) { + return SavepointKeyFilter.exact(exactKeys); + } + final SavepointKeyFilter rangeFilter = + SavepointKeyFilter.range(lower, lowerInclusive, upper, upperInclusive); + if (exactKeys == null) { + return rangeFilter; + } + final Set retainedKeys = new HashSet<>(exactKeys); + retainedKeys.removeIf(key -> !rangeFilter.test(key)); + return SavepointKeyFilter.exact(retainedKeys); + } + } + private enum Comparison { GT, GTE, 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..7da2167f144e1b 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 @@ -219,7 +219,8 @@ void testReadKeyedStateWithEmptyFilter() throws Exception { SavepointReader savepoint = SavepointReader.read(env, savepointPath, backendTuple.f1); CountingReadResult result = - readKeyedStateWithCountingReader(savepoint, SavepointKeyFilter.empty()); + readKeyedStateWithCountingReader( + savepoint, SavepointKeyFilter.exact(Collections.emptySet())); // 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/filter/ExactKeyFilterTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/filter/ExactKeyFilterTest.java new file mode 100644 index 00000000000000..dd79101ad5f534 --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/filter/ExactKeyFilterTest.java @@ -0,0 +1,103 @@ +/* + * 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.api.filter; + +import org.apache.flink.util.InstantiationUtil; + +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Unit tests for {@link ExactKeyFilter}, built through {@link SavepointKeyFilter#exact}. */ +class ExactKeyFilterTest { + + @Test + void singleValueFactoryMatchesOnlyThatKey() { + // exact(42) -> {42} + SavepointKeyFilter filter = SavepointKeyFilter.exact(42L); + + assertThat(filter.getExactKeys()).containsExactly(42L); + assertThat(filter.test(42L)).isTrue(); + assertThat(filter.test(43L)).isFalse(); + } + + @Test + void setFactoryMatchesEveryKeyInTheSet() { + // exact({1, 2, 3}) -> membership test + SavepointKeyFilter filter = SavepointKeyFilter.exact(Set.of(1L, 2L, 3L)); + + assertThat(filter.getExactKeys()).containsExactlyInAnyOrder(1L, 2L, 3L); + assertThat(filter.test(1L)).isTrue(); + assertThat(filter.test(2L)).isTrue(); + assertThat(filter.test(3L)).isTrue(); + assertThat(filter.test(0L)).isFalse(); + assertThat(filter.test(4L)).isFalse(); + } + + @Test + void emptySetMatchesNothing() { + // exact({}) -> the scan can be pruned entirely + SavepointKeyFilter filter = SavepointKeyFilter.exact(Collections.emptySet()); + + assertThat(filter.getExactKeys()).isEmpty(); + assertThat(filter.test(42L)).isFalse(); + assertThat(filter.test("hello")).isFalse(); + } + + @Test + void keysAreCopiedDefensively() { + // Mutating the source set after construction must not change the filter. + Set keys = new HashSet<>(Set.of(1L, 2L)); + SavepointKeyFilter filter = SavepointKeyFilter.exact(keys); + + keys.add(3L); + + assertThat(filter.getExactKeys()).containsExactlyInAnyOrder(1L, 2L); + assertThat(filter.test(3L)).isFalse(); + } + + @Test + void exactKeysAreUnmodifiable() { + SavepointKeyFilter filter = SavepointKeyFilter.exact(Set.of(1L)); + + assertThatThrownBy(() -> filter.getExactKeys().add(2L)) + .isInstanceOf(UnsupportedOperationException.class); + } + + @Test + void survivesSerialization() throws Exception { + // The filter is shipped with the job, so it must round-trip unchanged. + SavepointKeyFilter filter = SavepointKeyFilter.exact(Set.of(1L, 2L)); + SavepointKeyFilter copy = InstantiationUtil.clone(filter); + + assertThat(copy.getExactKeys()).containsExactlyInAnyOrder(1L, 2L); + assertThat(copy.test(1L)).isTrue(); + assertThat(copy.test(3L)).isFalse(); + } + + @Test + void toStringListsTheKeys() { + assertThat(SavepointKeyFilter.exact(7L)).hasToString("ExactKeyFilter[7]"); + } +} diff --git a/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/filter/RangeKeyFilterTest.java b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/filter/RangeKeyFilterTest.java new file mode 100644 index 00000000000000..8eb2fa49b3389a --- /dev/null +++ b/flink-libraries/flink-state-processing-api/src/test/java/org/apache/flink/state/api/filter/RangeKeyFilterTest.java @@ -0,0 +1,215 @@ +/* + * 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.api.filter; + +import org.apache.flink.util.InstantiationUtil; + +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Unit tests for {@link RangeKeyFilter}, built through {@link SavepointKeyFilter#range}. */ +class RangeKeyFilterTest { + + // ------------------------------------------------------------------------- + // Natural order — bound inclusiveness + // ------------------------------------------------------------------------- + + @Test + void closedRangeIncludesBothBounds() { + // [5, 10] + SavepointKeyFilter filter = SavepointKeyFilter.range(5L, true, 10L, true); + + assertThat(filter.test(4L)).isFalse(); + assertThat(filter.test(5L)).isTrue(); + assertThat(filter.test(7L)).isTrue(); + assertThat(filter.test(10L)).isTrue(); + assertThat(filter.test(11L)).isFalse(); + } + + @Test + void openRangeExcludesBothBounds() { + // (5, 10) + SavepointKeyFilter filter = SavepointKeyFilter.range(5L, false, 10L, false); + + assertThat(filter.test(5L)).isFalse(); + assertThat(filter.test(6L)).isTrue(); + assertThat(filter.test(9L)).isTrue(); + assertThat(filter.test(10L)).isFalse(); + } + + @Test + void halfOpenRangesExcludeOnlyTheExclusiveBound() { + // [5, 10) and (5, 10] + SavepointKeyFilter lowerInclusive = SavepointKeyFilter.range(5L, true, 10L, false); + assertThat(lowerInclusive.test(5L)).isTrue(); + assertThat(lowerInclusive.test(10L)).isFalse(); + + SavepointKeyFilter upperInclusive = SavepointKeyFilter.range(5L, false, 10L, true); + assertThat(upperInclusive.test(5L)).isFalse(); + assertThat(upperInclusive.test(10L)).isTrue(); + } + + // ------------------------------------------------------------------------- + // Natural order — unbounded and degenerate ranges + // ------------------------------------------------------------------------- + + @Test + void nullLowerBoundMeansUnboundedBelow() { + // (-∞, 10] + SavepointKeyFilter filter = SavepointKeyFilter.range(null, true, 10L, true); + + assertThat(filter.test(Long.MIN_VALUE)).isTrue(); + assertThat(filter.test(10L)).isTrue(); + assertThat(filter.test(11L)).isFalse(); + } + + @Test + void nullUpperBoundMeansUnboundedAbove() { + // [5, +∞) + SavepointKeyFilter filter = SavepointKeyFilter.range(5L, true, null, true); + + assertThat(filter.test(4L)).isFalse(); + assertThat(filter.test(5L)).isTrue(); + assertThat(filter.test(Long.MAX_VALUE)).isTrue(); + } + + @Test + void bothBoundsNullMatchesEverything() { + // (-∞, +∞) + SavepointKeyFilter filter = SavepointKeyFilter.range(null, true, null, true); + + assertThat(filter.test(Long.MIN_VALUE)).isTrue(); + assertThat(filter.test(0L)).isTrue(); + assertThat(filter.test(Long.MAX_VALUE)).isTrue(); + } + + @Test + void equalInclusiveBoundsMatchOnlyThatKey() { + // [7, 7] + SavepointKeyFilter filter = SavepointKeyFilter.range(7L, true, 7L, true); + + assertThat(filter.test(6L)).isFalse(); + assertThat(filter.test(7L)).isTrue(); + assertThat(filter.test(8L)).isFalse(); + } + + @Test + void equalBoundsWithAnExclusiveSideMatchNothing() { + // (7, 7] and [7, 7) + assertThat(SavepointKeyFilter.range(7L, false, 7L, true).test(7L)).isFalse(); + assertThat(SavepointKeyFilter.range(7L, true, 7L, false).test(7L)).isFalse(); + } + + @Test + void invertedBoundsMatchNothingUnderNaturalOrder() { + // [10, 5] — the lower bound is above the upper one + SavepointKeyFilter filter = SavepointKeyFilter.range(10L, true, 5L, true); + + assertThat(filter.test(5L)).isFalse(); + assertThat(filter.test(7L)).isFalse(); + assertThat(filter.test(10L)).isFalse(); + } + + @Test + void doesNotResolveToAFiniteKeySet() { + assertThat(SavepointKeyFilter.range(5L, true, 10L, true).getExactKeys()).isNull(); + } + + @Test + void stringKeysUseNaturalOrder() { + // ['beta', 'delta'] + SavepointKeyFilter filter = SavepointKeyFilter.range("beta", true, "delta", true); + + assertThat(filter.test("alpha")).isFalse(); + assertThat(filter.test("beta")).isTrue(); + assertThat(filter.test("charlie")).isTrue(); + assertThat(filter.test("delta")).isTrue(); + assertThat(filter.test("epsilon")).isFalse(); + } + + // ------------------------------------------------------------------------- + // Custom comparator + // ------------------------------------------------------------------------- + + @Test + void customComparatorIsUsed() { + // 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(); + } + + @Test + void descendingComparatorMakesInvertedBoundsValid() { + // Under a descending comparator, [6, 3] covers 3, 4, 5 and 6. + SavepointKeyFilter filter = + SavepointKeyFilter.range(6, true, 3, true, (a, b) -> Integer.compare(b, a)); + + assertThat(filter.test(2)).isFalse(); + assertThat(filter.test(3)).isTrue(); + assertThat(filter.test(6)).isTrue(); + assertThat(filter.test(7)).isFalse(); + } + + // ------------------------------------------------------------------------- + // Serialization and toString + // ------------------------------------------------------------------------- + + @Test + void survivesSerialization() throws Exception { + // The filter is shipped with the job, so it must round-trip unchanged. + SavepointKeyFilter filter = SavepointKeyFilter.range(5L, true, 10L, false); + SavepointKeyFilter copy = InstantiationUtil.clone(filter); + + assertThat(copy.test(4L)).isFalse(); + assertThat(copy.test(5L)).isTrue(); + assertThat(copy.test(10L)).isFalse(); + } + + @Test + void survivesSerializationWithACustomComparator() throws Exception { + // A lambda assigned to SerializableComparator is serializable too. + SerializableComparator descending = (a, b) -> Integer.compare(b, a); + SavepointKeyFilter filter = SavepointKeyFilter.range(6, true, 3, true, descending); + SavepointKeyFilter copy = InstantiationUtil.clone(filter); + + assertThat(copy.test(4)).isTrue(); + assertThat(copy.test(7)).isFalse(); + } + + @Test + void toStringRendersTheBounds() { + assertThat(SavepointKeyFilter.range(5L, true, 10L, false)) + .hasToString("RangeKeyFilter[5, 10)"); + assertThat(SavepointKeyFilter.range(null, true, 10L, true)) + .hasToString("RangeKeyFilter(-∞, 10]"); + assertThat(SavepointKeyFilter.range(5L, false, null, true)) + .hasToString("RangeKeyFilter(5, +∞)"); + } +} 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 3b587f7ce6e73b..ba10ef10e4bae4 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 @@ -28,6 +28,7 @@ import org.junit.jupiter.api.Test; +import java.math.BigDecimal; import java.util.List; import java.util.Map; import java.util.Set; @@ -57,6 +58,8 @@ class SavepointDynamicTableSourceTest { + " 'operator.uid' = 'keyed-state-process-uid'\n" + ")"; + private static final String TYPED_KEY_STATE_PATH = "src/test/resources/table-state-typed-keys"; + @Test @SuppressWarnings("unchecked") public void testReadKeyedState() throws Exception { @@ -395,6 +398,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(); @@ -411,6 +440,171 @@ 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); + } + + @Test + void testFilterPushDownOnIntKey() throws Exception { + StreamTableEnvironment tEnv = createBatchTableEnv(); + tEnv.executeSql(typedKeyDdl("int_key_table", "int", "int-key-state-op")); + + String sql = "SELECT k FROM int_key_table WHERE k = 5"; + + assertThat(hasPushedDownFilter(tEnv, sql)).isTrue(); + assertThat(collectTypedKeys(tEnv, sql)).containsExactly(5); + + String rangeSql = "SELECT k FROM int_key_table WHERE k BETWEEN 3 AND 6 ORDER BY k"; + + assertThat(hasPushedDownFilter(tEnv, rangeSql)).isTrue(); + assertThat(collectTypedKeys(tEnv, rangeSql)).containsExactly(3, 4, 5, 6); + } + + @Test + void testFilterPushDownOnDoubleKey() throws Exception { + StreamTableEnvironment tEnv = createBatchTableEnv(); + tEnv.executeSql(typedKeyDdl("double_key_table", "double", "double-key-state-op")); + + String equalitySql = "SELECT k FROM double_key_table WHERE k = 5"; + + assertThat(hasPushedDownFilter(tEnv, equalitySql)).isTrue(); + assertThat(collectTypedKeys(tEnv, equalitySql)).containsExactly(5.0d); + + // Bounds keep their own type, so the INT literals are converted to the key type here. + String rangeSql = "SELECT k FROM double_key_table WHERE k BETWEEN 1 AND 3 ORDER BY k"; + + assertThat(hasPushedDownFilter(tEnv, rangeSql)).isTrue(); + assertThat(collectTypedKeys(tEnv, rangeSql)).containsExactly(1.0d, 2.0d, 3.0d); + + // A BIGINT bound beyond the range where doubles are exact still reaches the filter as a + // BIGINT literal, so the conversion happens on our side. + String largeBoundSql = "SELECT k FROM double_key_table WHERE k > 9007199254740000"; + + assertThat(hasPushedDownFilter(tEnv, largeBoundSql)).isTrue(); + assertThat(collectTypedKeys(tEnv, largeBoundSql)).containsExactly(9007199254740992.0d); + + // 9007199254740993 is 2^53 + 1, which no double holds. The planner folds the literal to + // the nearest double itself, so the row it asks for is the one keyed 2^53, and the pushed + // filter agrees with it rather than rounding on its own. + String beyondExactRangeSql = "SELECT k FROM double_key_table WHERE k = 9007199254740993"; + + assertThat(hasPushedDownFilter(tEnv, beyondExactRangeSql)).isTrue(); + assertThat(collectTypedKeys(tEnv, beyondExactRangeSql)) + .containsExactly(9007199254740992.0d); + } + + @Test + void testFilterPushDownOnSmallintKey() throws Exception { + StreamTableEnvironment tEnv = createBatchTableEnv(); + tEnv.executeSql(typedKeyDdl("smallint_key_table", "smallint", "smallint-key-state-op")); + + // The key column is bare here, but an INT literal is not converted to a SMALLINT key: + // only BIGINT and DOUBLE keys take a literal of another numeric type. + String sql = "SELECT k FROM smallint_key_table WHERE k BETWEEN 3 AND 6 ORDER BY k"; + + assertThat(hasPushedDownFilter(tEnv, sql)).isFalse(); + assertThat(collectTypedKeys(tEnv, sql)) + .containsExactly((short) 3, (short) 4, (short) 5, (short) 6); + + // The planner rewrites k = 5 into CAST(k AS INT) = 5, leaving no key column to push on. + // The predicate stays in the query and still returns the right row. + String equalitySql = "SELECT k FROM smallint_key_table WHERE k = 5"; + + assertThat(hasPushedDownFilter(tEnv, equalitySql)).isFalse(); + assertThat(collectTypedKeys(tEnv, equalitySql)).containsExactly((short) 5); + } + + @Test + void testFilterPushDownOnTinyintKey() throws Exception { + StreamTableEnvironment tEnv = createBatchTableEnv(); + tEnv.executeSql(typedKeyDdl("tinyint_key_table", "tinyint", "tinyint-key-state-op")); + + // As for SMALLINT, an INT literal is not converted to a TINYINT key. + String sql = "SELECT k FROM tinyint_key_table WHERE k BETWEEN 3 AND 6 ORDER BY k"; + + assertThat(hasPushedDownFilter(tEnv, sql)).isFalse(); + assertThat(collectTypedKeys(tEnv, sql)) + .containsExactly((byte) 3, (byte) 4, (byte) 5, (byte) 6); + + // As for SMALLINT, k = 5 is rewritten to CAST(k AS INT) = 5 and cannot be pushed. + String equalitySql = "SELECT k FROM tinyint_key_table WHERE k = 5"; + + assertThat(hasPushedDownFilter(tEnv, equalitySql)).isFalse(); + assertThat(collectTypedKeys(tEnv, equalitySql)).containsExactly((byte) 5); + } + + @Test + void testFilterPushDownOnDecimalKey() throws Exception { + StreamTableEnvironment tEnv = createBatchTableEnv(); + tEnv.executeSql(typedKeyDdl("decimal_key_table", "decimal(10, 2)", "decimal-key-state-op")); + + // Literal scale equal to the key scale. + String sameScaleSql = "SELECT k FROM decimal_key_table WHERE k = 5.00"; + + assertThat(hasPushedDownFilter(tEnv, sameScaleSql)).isTrue(); + assertThat(collectTypedKeys(tEnv, sameScaleSql)).containsExactly(new BigDecimal("5.00")); + + // Literal scale above the key scale: the planner widens the comparison and casts the key + // column, so nothing is pushed and the predicate is evaluated on the read rows instead. + String largerScaleSql = "SELECT k FROM decimal_key_table WHERE k = 5.000"; + + assertThat(hasPushedDownFilter(tEnv, largerScaleSql)).isFalse(); + assertThat(collectTypedKeys(tEnv, largerScaleSql)).containsExactly(new BigDecimal("5.00")); + + // A range keeps the key column bare, but an INT bound is not converted to a DECIMAL key, + // so this is not pushed either. + String rangeSql = "SELECT k FROM decimal_key_table WHERE k BETWEEN 3 AND 6 ORDER BY k"; + + assertThat(hasPushedDownFilter(tEnv, rangeSql)).isFalse(); + assertThat(collectTypedKeys(tEnv, rangeSql)) + .containsExactly( + new BigDecimal("3.00"), + new BigDecimal("4.00"), + new BigDecimal("5.00"), + new BigDecimal("6.00")); + } + // ------------------------------------------------------------------------- // Projection push-down tests // ------------------------------------------------------------------------- @@ -503,6 +697,41 @@ void testPlanningSucceedsWithNonexistentSavepointPath() { // Helpers // ------------------------------------------------------------------------- + private static List collectTypedKeys(StreamTableEnvironment tEnv, String sql) + throws Exception { + return tEnv.toDataStream(tEnv.sqlQuery(sql)).executeAndCollect(100).stream() + .map(r -> r.getField("k")) + .collect(Collectors.toList()); + } + + private static String typedKeyDdl(String table, String keyType, String uid) { + return "CREATE TABLE " + + table + + " (\n" + + " k " + + keyType + + ",\n" + + " v bigint,\n" + + " PRIMARY KEY (k) NOT ENFORCED\n" + + ")\n" + + "with (\n" + + " 'connector' = 'savepoint',\n" + + " 'state.path' = '" + + TYPED_KEY_STATE_PATH + + "',\n" + + " 'operator.uid' = '" + + uid + + "'\n" + + ")"; + } + + 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); @@ -512,7 +741,8 @@ private static StreamTableEnvironment createBatchTableEnv() { private static final Pattern PUSHED_DOWN_FILTER = Pattern.compile( - "TableSourceScan\\(table=\\[\\[default_catalog, default_database, state_table, filter=\\[[^\\]]+\\]"); + "TableSourceScan\\(table=\\[\\[default_catalog, default_database, \\w+," + + " filter=\\[[^\\]]+\\]"); private static boolean hasPushedDownFilter(StreamTableEnvironment tEnv, String sql) { return PUSHED_DOWN_FILTER.matcher(tEnv.explainSql(sql)).find(); 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..685b4145b61e3a 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 @@ -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,6 +49,7 @@ class SavepointFilterTranslatorTest { @Test void equalsKeyOnLeft() { + // key = 42 -> {42} SavepointKeyFilter filter = keyFilterOf(eq(longKeyRef(), longLit(42L))); assertNotNull(filter); assertThat(filter.getExactKeys()).containsExactly(42L); @@ -62,6 +59,7 @@ void equalsKeyOnLeft() { @Test void equalsKeyOnRight() { + // 42 = key -> {42} SavepointKeyFilter filter = keyFilterOf(eq(longLit(42L), longKeyRef())); assertNotNull(filter); assertThat(filter.getExactKeys()).containsExactly(42L); @@ -71,12 +69,14 @@ void equalsKeyOnRight() { @Test void equalsNeitherSideIsKeyColumn_returnsNull() { + // val = 42 -> not the key column, not pushed SavepointKeyFilter filter = keyFilterOf(eq(otherRef(), longLit(42L))); assertThat(filter).isNull(); } @Test void equalsNeitherSideIsLiteral_returnsNull() { + // key = val -> no literal to match, not pushed SavepointKeyFilter filter = keyFilterOf(eq(longKeyRef(), otherRef())); assertThat(filter).isNull(); } @@ -87,6 +87,7 @@ void equalsNeitherSideIsLiteral_returnsNull() { @Test void orOfEqualsProducesMergedExactFilter() { + // key = 1 OR key = 2 OR 3 = key -> {1, 2, 3} CallExpression expr = or( eq(longKeyRef(), longLit(1L)), @@ -112,6 +113,7 @@ void orWithNonPushableChild_returnsNull() { @Test void betweenProducesInclusiveRange() { + // key BETWEEN 10 AND 20 -> [10, 20] SavepointKeyFilter filter = keyFilterOf(between(longKeyRef(), longLit(10L), longLit(20L))); @@ -126,6 +128,7 @@ void betweenProducesInclusiveRange() { @Test void betweenWithNonKeyField_returnsNull() { + // val BETWEEN 1 AND 10 -> not the key column, not pushed SavepointKeyFilter filter = keyFilterOf(between(otherRef(), longLit(1L), longLit(10L))); assertThat(filter).isNull(); @@ -137,6 +140,7 @@ void betweenWithNonKeyField_returnsNull() { @Test void greaterThanProducesExclusiveLowerBound() { + // key > 5 -> (5, +∞) SavepointKeyFilter filter = keyFilterOf(gt(longKeyRef(), longLit(5L))); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); @@ -146,6 +150,7 @@ void greaterThanProducesExclusiveLowerBound() { @Test void greaterThanOrEqualProducesInclusiveLowerBound() { + // key >= 5 -> [5, +∞) SavepointKeyFilter filter = keyFilterOf(gte(longKeyRef(), longLit(5L))); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); @@ -156,6 +161,7 @@ void greaterThanOrEqualProducesInclusiveLowerBound() { @Test void lessThanProducesExclusiveUpperBound() { + // key < 10 -> (-∞, 10) SavepointKeyFilter filter = keyFilterOf(lt(longKeyRef(), longLit(10L))); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); @@ -165,6 +171,7 @@ void lessThanProducesExclusiveUpperBound() { @Test void lessThanOrEqualProducesInclusiveUpperBound() { + // key <= 10 -> (-∞, 10] SavepointKeyFilter filter = keyFilterOf(lte(longKeyRef(), longLit(10L))); assertNotNull(filter); assertThat(filter.getExactKeys()).isNull(); @@ -216,7 +223,30 @@ void andWithProvablyEmptyIntersection_matchesNothing() { SavepointKeyFilter filter = keyFilterOf(expr); assertNotNull(filter); - assertThat(filter.isEmpty()).isTrue(); + assertThat(filter.getExactKeys()).isEmpty(); + } + + @Test + void andWithEqualInclusiveBounds_matchesOnlyBound() { + // key >= 7 AND key <= 7 -> [7, 7] + CallExpression expr = and(gte(longKeyRef(), longLit(7L)), lte(longKeyRef(), longLit(7L))); + SavepointKeyFilter filter = keyFilterOf(expr); + + assertNotNull(filter); + assertThat(filter.getExactKeys()).isNull(); + assertThat(filter.test(6L)).isFalse(); + assertThat(filter.test(7L)).isTrue(); + assertThat(filter.test(8L)).isFalse(); + } + + @Test + void andWithEqualBoundsAndExclusiveLower_matchesNothing() { + // key > 7 AND key <= 7 -> empty + CallExpression expr = and(gt(longKeyRef(), longLit(7L)), lte(longKeyRef(), longLit(7L))); + SavepointKeyFilter filter = keyFilterOf(expr); + + assertNotNull(filter); + assertThat(filter.getExactKeys()).isEmpty(); } @Test @@ -232,11 +262,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,6 +283,7 @@ void unrecognizedFunctionReturnsNull() { @Test void rangeFilterOnStringKey() { + // key BETWEEN 'beta' AND 'delta' -> natural String order SavepointKeyFilter filter = keyFilterOf(between(stringKeyRef(), stringLit("beta"), stringLit("delta"))); @@ -264,6 +297,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 = @@ -281,80 +315,13 @@ 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))), @@ -372,6 +339,7 @@ void applyAccumulatesRangeAndRange() { @Test void applyAccumulatesExactAndExact() { + // key IN (1, 2, 3), key IN (2, 3, 4) -> {2, 3} SavepointFilterTranslator.Result applied = apply( List.of( @@ -394,6 +362,7 @@ 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))), @@ -403,11 +372,12 @@ void applyAccumulatesExactAndExactEmptyResult_matchesNothing() { assertNotNull(result); assertThat(applied.accepted()).hasSize(2); assertThat(applied.remaining()).isEmpty(); - assertThat(result.isEmpty()).isTrue(); + assertThat(result.getExactKeys()).isEmpty(); } @Test void applyAccumulatesExactAndRange_keepsOnlyKeysInRange() { + // key IN (1, 5, 10, 15), key BETWEEN 4 AND 12 -> {5, 10} SavepointFilterTranslator.Result applied = apply( List.of( @@ -428,6 +398,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( @@ -446,52 +417,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 // ------------------------------------------------------------------------- @@ -521,6 +446,7 @@ void andOfThreeRangesProducesIntersection() { @Test void orWithSingleChild_returnsExactFilter() { + // OR (key = 7) -> {7} CallExpression expr = or(eq(longKeyRef(), longLit(7L))); SavepointKeyFilter filter = keyFilterOf(expr); @@ -534,6 +460,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,6 +471,7 @@ 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( @@ -555,11 +483,12 @@ void applyWithEmptyThenRange_returnsEmpty() { assertNotNull(result); assertThat(applied.accepted()).hasSize(2); assertThat(applied.remaining()).isEmpty(); - assertThat(result.isEmpty()).isTrue(); + assertThat(result.getExactKeys()).isEmpty(); } @Test void applyWithRangeThenEmpty_returnsEmpty() { + // key <= 10, (key > 10 AND key < 5) -> empty, operands swapped SavepointFilterTranslator.Result applied = apply( List.of( @@ -571,11 +500,12 @@ void applyWithRangeThenEmpty_returnsEmpty() { assertNotNull(result); assertThat(applied.accepted()).hasSize(2); assertThat(applied.remaining()).isEmpty(); - assertThat(result.isEmpty()).isTrue(); + assertThat(result.getExactKeys()).isEmpty(); } @Test void applyWithEmptyThenExact_returnsEmpty() { + // (key > 10 AND key < 5) -> empty, key = 1 -> empty absorbs the exact set SavepointFilterTranslator.Result applied = apply( List.of( @@ -587,11 +517,12 @@ void applyWithEmptyThenExact_returnsEmpty() { assertNotNull(result); assertThat(applied.accepted()).hasSize(2); assertThat(applied.remaining()).isEmpty(); - assertThat(result.isEmpty()).isTrue(); + assertThat(result.getExactKeys()).isEmpty(); } @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))), @@ -601,7 +532,7 @@ void applyWithConflictingExactPredicates_returnsEmptyFilter() { assertNotNull(result); assertThat(applied.accepted()).hasSize(2); assertThat(applied.remaining()).isEmpty(); - assertThat(result.isEmpty()).isTrue(); + assertThat(result.getExactKeys()).isEmpty(); } // ------------------------------------------------------------------------- @@ -610,6 +541,7 @@ void applyWithConflictingExactPredicates_returnsEmptyFilter() { @Test void betweenWithWrongArgCount_returnsNull() { + // BETWEEN with two children is malformed, not pushed CallExpression malformed = CallExpression.permanent( BuiltInFunctionDefinitions.BETWEEN, @@ -624,6 +556,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 +571,7 @@ void betweenWithNonLiteralBound_returnsNull() { @Test void comparisonWithNonLiteralValue_returnsNull() { + // key > val -> no literal bound, not pushed assertThat(keyFilterOf(gt(longKeyRef(), otherRef()))).isNull(); } @@ -647,6 +581,7 @@ void comparisonWithNonLiteralValue_returnsNull() { @Test void equalsWithWrongArgCount_returnsNull() { + // EQUALS with one child is malformed, not pushed CallExpression malformed = CallExpression.permanent( BuiltInFunctionDefinitions.EQUALS, @@ -661,6 +596,7 @@ 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)); @@ -672,6 +608,7 @@ 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)); @@ -686,6 +623,7 @@ 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()); @@ -698,13 +636,271 @@ void equalsWithIntLiteralIsWidenedToDoubleKeyAndPushed() { assertThat(filter.test(6.0d)).isFalse(); } + @Test + void lessThanFractionalLiteralAgainstBigintKeyIsNotPushed() { + // key < 1.5 (DECIMAL literal, BIGINT key) -> no long bound is exact here, and a + // truncated 1 would drop key 1, so the predicate is not pushed + SavepointFilterTranslator.Result applied = + apply(List.of(lt(longKeyRef(), decLit("1.5"))), LONG_KEY_TYPE); + + assertThat(applied.keyFilter()).isNull(); + // The predicate must be handed back, or nothing would evaluate it. + assertThat(applied.accepted()).isEmpty(); + assertThat(applied.remaining()).hasSize(1); + } + + @Test + void greaterThanOrEqualFractionalLiteralAgainstBigintKeyIsNotPushed() { + // key >= 1.5 -> a truncated bound of 1 would wrongly admit key 1, not pushed + SavepointFilterTranslator.Result applied = + apply(List.of(gte(longKeyRef(), decLit("1.5"))), LONG_KEY_TYPE); + + assertThat(applied.keyFilter()).isNull(); + assertThat(applied.accepted()).isEmpty(); + assertThat(applied.remaining()).hasSize(1); + } + + @Test + void betweenFractionalLiteralBoundsAgainstBigintKeyIsNotPushed() { + // key BETWEEN 1.5 AND 4.5 -> both bounds are lossy, not pushed + assertThat(keyFilterOf(between(longKeyRef(), decLit("1.5"), decLit("4.5")))).isNull(); + } + + @Test + void equalsFractionalLiteralAgainstBigintKeyIsNotPushed() { + // key = 1.5 -> no BIGINT key equals 1.5, and a truncated 1 would match the wrong key + assertThat(keyFilterOf(eq(longKeyRef(), decLit("1.5")))).isNull(); + } + + @Test + void integralDecimalLiteralAgainstBigintKeyIsConvertedAndPushed() { + // key = 5.0 -> the conversion to 5L is exact, so the pushdown is still safe + SavepointKeyFilter filter = keyFilterOf(eq(longKeyRef(), decLit("5.0"))); + + assertNotNull(filter); + assertThat(filter.getExactKeys()).containsExactly(5L); + assertThat(filter.test(5L)).isTrue(); + } + + @Test + void outOfRangeLiteralAgainstBigintKeyIsNotPushed() { + // key < 10^30 -> the value does not fit in a long, not pushed + ValueLiteralExpression hugeLit = + new ValueLiteralExpression( + new BigDecimal("1000000000000000000000000000000"), + DataTypes.DECIMAL(31, 0).notNull()); + + assertThat(keyFilterOf(lt(longKeyRef(), hugeLit))).isNull(); + } + + @Test + void negativeFractionalLiteralAgainstBigintKeyIsNotPushed() { + // key > -1.5 -> fractional, so no long bound represents it exactly, not pushed + assertThat(keyFilterOf(gt(longKeyRef(), decLit("-1.5")))).isNull(); + } + + @Test + void floatLiteralWithFractionAgainstBigintKeyIsNotPushed() { + // key < 1.5 (FLOAT literal, BIGINT key) -> lossy, not pushed + ValueLiteralExpression floatLit = + new ValueLiteralExpression(1.5f, DataTypes.FLOAT().notNull()); + + assertThat(keyFilterOf(lt(longKeyRef(), floatLit))).isNull(); + } + + @Test + void integralFloatLiteralAgainstBigintKeyIsConvertedAndPushed() { + // key = 5.0 (FLOAT literal, BIGINT key) -> the conversion is exact, so still pushed + ValueLiteralExpression floatLit = + new ValueLiteralExpression(5.0f, DataTypes.FLOAT().notNull()); + + SavepointKeyFilter filter = keyFilterOf(eq(longKeyRef(), floatLit)); + + assertNotNull(filter); + assertThat(filter.getExactKeys()).containsExactly(5L); + } + + @Test + void approximateLiteralAtBigintExactIntegerLimitIsNotPushed() { + List aliasedLiterals = + List.of( + new ValueLiteralExpression(16777216f, DataTypes.FLOAT().notNull()), + new ValueLiteralExpression(-16777216f, DataTypes.FLOAT().notNull()), + new ValueLiteralExpression(9007199254740992d, DataTypes.DOUBLE().notNull()), + new ValueLiteralExpression( + -9007199254740992d, DataTypes.DOUBLE().notNull())); + + for (ValueLiteralExpression literal : aliasedLiterals) { + for (ResolvedExpression predicate : + List.of(eq(longKeyRef(), literal), gt(longKeyRef(), literal))) { + SavepointFilterTranslator.Result applied = apply(List.of(predicate), LONG_KEY_TYPE); + + assertThat(applied.keyFilter()).isNull(); + assertThat(applied.accepted()).isEmpty(); + assertThat(applied.remaining()).containsExactly(predicate); + } + } + } + + @Test + void approximateLiteralBelowBigintExactIntegerLimitIsPushed() { + List exactLiterals = + List.of( + new ValueLiteralExpression(16777215f, DataTypes.FLOAT().notNull()), + new ValueLiteralExpression(-16777215f, DataTypes.FLOAT().notNull()), + new ValueLiteralExpression(9007199254740991d, DataTypes.DOUBLE().notNull()), + new ValueLiteralExpression( + -9007199254740991d, DataTypes.DOUBLE().notNull())); + + for (ValueLiteralExpression literal : exactLiterals) { + assertThat(keyFilterOf(eq(longKeyRef(), literal))).isNotNull(); + } + } + + @Test + void tinyintLiteralAgainstBigintKeyIsConvertedAndPushed() { + // key = 5 (TINYINT literal, BIGINT key) -> {5L} + ValueLiteralExpression tinyLit = + new ValueLiteralExpression((byte) 5, DataTypes.TINYINT().notNull()); + + SavepointKeyFilter filter = keyFilterOf(eq(longKeyRef(), tinyLit)); + + assertNotNull(filter); + assertThat(filter.getExactKeys()).containsExactly(5L); + } + + @Test + void smallintLiteralAgainstBigintKeyIsConvertedAndPushed() { + // key = 5 (SMALLINT literal, BIGINT key) -> {5L} + ValueLiteralExpression smallLit = + new ValueLiteralExpression((short) 5, DataTypes.SMALLINT().notNull()); + + SavepointKeyFilter filter = keyFilterOf(eq(longKeyRef(), smallLit)); + + assertNotNull(filter); + assertThat(filter.getExactKeys()).containsExactly(5L); + } + + @Test + void nanLiteralAgainstBigintKeyIsNotPushed() { + // key = NaN -> no decimal value at all, not pushed + ValueLiteralExpression nanLit = + new ValueLiteralExpression(Double.NaN, DataTypes.DOUBLE().notNull()); + + assertThat(keyFilterOf(eq(longKeyRef(), nanLit))).isNull(); + } + + @Test + void infiniteLiteralAgainstBigintKeyIsNotPushed() { + // key < +Infinity -> no decimal value at all, not pushed + ValueLiteralExpression infLit = + new ValueLiteralExpression(Double.POSITIVE_INFINITY, DataTypes.DOUBLE().notNull()); + + assertThat(keyFilterOf(lt(longKeyRef(), infLit))).isNull(); + } + + @Test + void fractionalLiteralAgainstDoubleKeyIsConvertedAndPushed() { + // key BETWEEN 1.5 AND 3.5 (DECIMAL literals, DOUBLE key) -> [1.5, 3.5] + FieldReferenceExpression doubleKey = + new FieldReferenceExpression("key", DataTypes.DOUBLE().notNull(), 0, KEY_COL); + + SavepointKeyFilter filter = + keyFilterOf(between(doubleKey, decLit("1.5"), decLit("3.5"))); + + assertNotNull(filter); + assertThat(filter.test(1.4d)).isFalse(); + assertThat(filter.test(1.5d)).isTrue(); + assertThat(filter.test(3.5d)).isTrue(); + assertThat(filter.test(3.6d)).isFalse(); + } + + @Test + void floatLiteralAgainstDoubleKeyIsConvertedAndPushed() { + // key = 0.5 (FLOAT literal, DOUBLE key) -> exactly representable, so pushed + FieldReferenceExpression doubleKey = + new FieldReferenceExpression("key", DataTypes.DOUBLE().notNull(), 0, KEY_COL); + ValueLiteralExpression floatLit = + new ValueLiteralExpression(0.5f, DataTypes.FLOAT().notNull()); + + SavepointKeyFilter filter = keyFilterOf(eq(doubleKey, floatLit)); + + assertNotNull(filter); + assertThat(filter.getExactKeys()).containsExactly(0.5d); + } + + @Test + void literalNotRepresentableAsDoubleIsConvertedToTheNearestDouble() { + // key > 9007199254740993 (BIGINT literal, DOUBLE key) -> 2^53 + 1 is not representable, so + // the bound becomes the nearest double, which is the value SQL compares against too. + FieldReferenceExpression doubleKey = + new FieldReferenceExpression("key", DataTypes.DOUBLE().notNull(), 0, KEY_COL); + + SavepointKeyFilter filter = keyFilterOf(gt(doubleKey, longLit(9007199254740993L))); + + assertNotNull(filter); + assertThat(filter.test(9007199254740992.0d)).isFalse(); + assertThat(filter.test(9007199254740994.0d)).isTrue(); + } + + @Test + void equalsWithLongLiteralNotRepresentableAsDoubleUsesNearestDouble() { + FieldReferenceExpression doubleKey = + new FieldReferenceExpression("key", DataTypes.DOUBLE().notNull(), 0, KEY_COL); + + SavepointKeyFilter filter = keyFilterOf(eq(doubleKey, longLit(9007199254740993L))); + + assertNotNull(filter); + assertThat(filter.getExactKeys()).containsExactly(9007199254740992.0d); + } + + @Test + void outOfRangeLiteralAgainstDoubleKeyIsNotPushed() { + // key < 10^400 (DOUBLE key) -> the value overflows to +Infinity, not pushed + FieldReferenceExpression doubleKey = + new FieldReferenceExpression("key", DataTypes.DOUBLE().notNull(), 0, KEY_COL); + ValueLiteralExpression hugeLit = + new ValueLiteralExpression( + new BigDecimal("1E+400"), DataTypes.DECIMAL(38, 0).notNull()); + + assertThat(keyFilterOf(lt(doubleKey, hugeLit))).isNull(); + + // A literal of the same shape that still fits in a double is pushed, so the rejection + // above is the overflow guard rather than an unreadable literal. + ValueLiteralExpression largeLit = + new ValueLiteralExpression( + new BigDecimal("1E+300"), DataTypes.DECIMAL(38, 0).notNull()); + assertNotNull(keyFilterOf(lt(doubleKey, largeLit))); + } + + @Test + void doubleLiteralAgainstDoubleKeyIsPushed() { + // key BETWEEN 1.5 AND 3.5 (DOUBLE literals, DOUBLE key) -> [1.5, 3.5] + FieldReferenceExpression doubleKey = + new FieldReferenceExpression("key", DataTypes.DOUBLE().notNull(), 0, KEY_COL); + ValueLiteralExpression lower = + new ValueLiteralExpression(1.5d, DataTypes.DOUBLE().notNull()); + ValueLiteralExpression upper = + new ValueLiteralExpression(3.5d, DataTypes.DOUBLE().notNull()); + + SavepointKeyFilter filter = keyFilterOf(between(doubleKey, lower, upper)); + + assertNotNull(filter); + assertThat(filter.test(1.4d)).isFalse(); + assertThat(filter.test(1.5d)).isTrue(); + assertThat(filter.test(3.5d)).isTrue(); + assertThat(filter.test(3.6d)).isFalse(); + } + @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,6 +908,7 @@ 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 = @@ -727,6 +924,116 @@ 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); + SavepointKeyFilter 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) + SavepointKeyFilter 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) + SavepointKeyFilter 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 // ------------------------------------------------------------------------- @@ -771,6 +1078,12 @@ private static ValueLiteralExpression longLit(long value) { return new ValueLiteralExpression(value, DataTypes.BIGINT().notNull()); } + private static ValueLiteralExpression decLit(String value) { + BigDecimal decimal = new BigDecimal(value); + return new ValueLiteralExpression( + decimal, DataTypes.DECIMAL(decimal.precision(), decimal.scale()).notNull()); + } + private static ValueLiteralExpression stringLit(String value) { return new ValueLiteralExpression(value, DataTypes.STRING().notNull()); } @@ -780,6 +1093,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/resources/table-state-typed-keys/_metadata b/flink-libraries/flink-state-processing-api/src/test/resources/table-state-typed-keys/_metadata new file mode 100644 index 0000000000000000000000000000000000000000..b14c172192b606f767ab0da7de45ccc9746fb59a GIT binary patch literal 26047 zcmeHPPi!4W9eyv*c3annAOhmRA)s84I?m4S%5SSC{iUFh*SxL%B@@|g(_T-;6y8+5*Jhz!2wn35rL>w35n8%Z}#ojZ=anw>p#yB z?MSmbJG1}y+i&Lg&2Q%T*>j}R=}aIFhVjm!Q{G*)-hE}sFYDFC<+)Y8>N^ML7gos= z{t{UUn`Ck6wI9Fm`Nj9e&kue}KKIlEFa822xfk(rI9W)ng?p;h!L^)iCA#z5I&AHr z?z7#_iIru0bk(n{9$oe;ix-!zUpcy}&3Ruv)#{}s@4{2A9F@XTnR-LC^p-HHGa^$d ziEzr9LKv5uqSNWVg!7lL@IdFp;_|s^y`=5qe!4h6zwiV$=BMrAg$s)d)8Wh)SLe^K zOdGx8r;jhJ&iUneeSZFFzkH;&HMgLbRvuqm#ff{JZm0WH^#t9{!7t4|Hv90Ixzoo^ z&z?E`*vZ+>_@l=j`RXioIM6v*CeO^xo;rT)k>lT(JyqtOY;aGHEH0cIxT*SP?XC`; zIQHe)xs%5pnjOC0qZ_Wbd~soQ{(_$_VKe>c;=DVdFZdO_CO>eoTjbZS8s5XiRV>Cr zDne}R=FPC#nYb1+W68Z$Nm-0fRjoS#j{%Ffz~fGF7B92)x)C1JRM?#Il;qMF!j$#I zP$h^IlnQTo#=XZwK8wq?cwEE(p5yU)eBTju6Lr)bh%gAKJB)Y>)WzNQe%%OlUgo0E zG9jrHU_p~OqFE-1DwH&-;CYcMpe|5zi@K4at^!Fff+b)pLEHjcd(Ui*u*I_?Gdc65 zuv!q7rk*IDGGcs|GcFiqp0$iN@*4IAfOKC#R%EpM>Wo&_zM(vzZU%7+)M1Z34vkO; zz@?d!p2%D$gn6flM8gj4I$?e8a$e+F%V?uUU4@YGh?Ib>lZe~I)=rtOY(|y$(g<71 zC56_Kl7cf$Snfe)84L=?tfn~^N@W(P#>E39BEoSo+=+eZAoDo)_^=1lLkd#v8Qk&6~IbiD#IC*Os zjLb5acoOfsvSh=9j zL;9*Kh40+_*e7TH{o;S<@(<0_E3-esiNiTBBd!x~*oeQx(4P>BnuP9!CVOw(w|k}0 zpCGm6D)Sj=H9@VPK_papp@}lSuxVmg#)BO*Q0t%CppxBMDV%$Jad}WH9NMlaWUX4b zuAGL?I(Bt)t#IsGA9oSi+f@p8h_j8zyH!pjJZe#Rk#a$tN+e;9Lc=yNa|Cymnx`W!iI{i5$2K zKTV6fky=0%t(5J8kPh{1g1ALJyVH6!T|Bd zQVpr2F3mLu9yvS>l1xFxcARjL3z=jx$(d>SZa=(VD!z*{N!gBDLgr9AZtvS|{To?R z4qt1;caEX~muPshh`@J(<-{cLF0kl;^GY$}Sg22bkBhW+!bt6eu_rfbhhy*cWx2?8 zQ8^s@Xmj0g>}F%t@X+e~!lhx+@P7XE+*3zBJpJ-tzPZt?C7WfII$nYm5+GV-1iiU3Jpv_l5K@(TL2QRG-de_xATghA?v$rc4?hs>lj?P9{OsUXJCxR4(g_Bnb?OLirgsvRP ztus93Pz$=xv{;vG99-b&n(#18j*`R+t_tggO^TN3Mn62RYP@YJ5blt{s>^J->b}_DrTa=F zrpv6cju#2BmO&s)a*2*!xE5;IquN+&y$bf|Z*cO~G#H_2(0j793L>S2joy1D)?!u^ z2YVlF>QmUe+0>!%kn?sPK83ZB$8TQ$=F9&W|JUbcZ(RN51Alw$FzzGVz-7cW8V>2c zaEbMhK3} zpKjn!_~87xhrL}_z>W-O)FiRiqp)}S26NKJ)N$P=C44VTw^XYuh+ zUWV}Z1avk;X#e2dTsA_e1VX{jaDh+-tn1V|q8x*X&EbMW6Yq2H@Q~p*Y7shO3%deE ziO!G?2s@1U|AnxECyD2l5-ZXD%%}p^GGPb@9m20V71D_oVf!-&1HcqQQ-!d9)b1@e z*3ykQj2oksNiE!LC`=em6_KGAnp4U%4GTtC355LwH{5dAh&gQP3*oIF_&S!gS~h%$ zsb7b(?gPVRy}{8b)f^$%r{upGtQ+#*Ms$UfZ4BKPHs)=f(+HgzoN)3CW8?G0p$({n z!n_M6C4$PracT=k13F*DrCM~37@haOfww-622LHEQTUm9GklcpcZbVkS2FTLC=H~L z2o>~6#D)M3hwg4@1W?REV^cUL5|@QOV_v{@K{AVpL9`U;P!>kD74$w$l2pN>B!-X< z2s?{dA?!}64M*&9XoRrDP%AY24ik#8r3{Qm6zGp&K)n`@dg$S%t)Ta(2{?5Y$e8FI z#KlqQf->|kqc;Xzhu#>k;Iz#P zLli3nW=v8_(;(9P6c5%`&?By(r#=X8{lKrVtf4cilapa+bC6(arL`WK&4)LC+$~-V zlLZb=gZPa6+PkkZ@%)(&p8eD7%b%IVY5%9i$nfM1ra?5JYf+>a&CC_dB&}g4EZrHWu{NQH1ny!(ti5xa&gR( z48Gjj5_tc>AIJtHzK~&b7a**P)E1crcDE^$4@0%ZZkS59S@7`Fyh1;dN@5a(b_)C?PnnGQ)Mig-5PT^^W$?MSbEDgG$4W?DJFyw+Y z*V-Q=wzDVyICD6D)OF9BI5ukX^vrtUnqN*ZxhLP6iD1Q6OvVnJ$v@A0di}syx@Nmq zL$CXL#L}AWULhguF{FI1pLW!#Mws_T*&I^>kc8#nTo8k4tkq>=Rf>*pOc7=