This is an automated email from the ASF dual-hosted git repository.
gaborgsomogyi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new d8f92d44ba1 [FLINK-40399][state] Separate key filter runtime contract
from push-down planning
d8f92d44ba1 is described below
commit d8f92d44ba1941e3dc6f5690c4b9de769806bc1b
Author: Ilya Soin <[email protected]>
AuthorDate: Tue Sep 8 12:23:55 2026 +0300
[FLINK-40399][state] Separate key filter runtime contract from push-down
planning
---
docs/content.zh/docs/libs/state_processor_api.md | 3 -
docs/content/docs/libs/state_processor_api.md | 3 -
.../apache/flink/state/api/filter/BoundInfo.java | 45 --
.../flink/state/api/filter/EmptyKeyFilter.java | 67 ---
.../flink/state/api/filter/ExactKeyFilter.java | 15 -
.../flink/state/api/filter/RangeKeyFilter.java | 88 +---
.../flink/state/api/filter/SavepointKeyFilter.java | 66 +--
.../state/table/SavepointFilterTranslator.java | 179 +++++--
.../state/api/SavepointReaderKeyedStateITCase.java | 3 +-
.../flink/state/api/filter/ExactKeyFilterTest.java | 103 ++++
.../flink/state/api/filter/RangeKeyFilterTest.java | 215 ++++++++
.../state/api/input/KeyedStateInputFormatTest.java | 7 +-
.../table/SavepointDynamicTableSourceTest.java | 232 ++++++++-
.../state/table/SavepointFilterTranslatorTest.java | 580 ++++++++++++++++-----
.../resources/table-state-typed-keys/_metadata | Bin 0 -> 26047 bytes
15 files changed, 1179 insertions(+), 427 deletions(-)
diff --git a/docs/content.zh/docs/libs/state_processor_api.md
b/docs/content.zh/docs/libs/state_processor_api.md
index e0e20bfaf4a..14b63aefa0e 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<KeyedState> keyRange = savepoint.readKeyedState(
* `SavepointKeyFilter.exact(K key)` / `SavepointKeyFilter.exact(Set<K> keys)`
— match a single key or a finite set of keys.
* `SavepointKeyFilter.range(K lower, boolean lowerInclusive, K upper, boolean
upperInclusive)` — match a range; `K` must implement `Comparable<K>`. Either
bound may be `null` to leave that side unbounded.
* `SavepointKeyFilter.range(K lower, boolean lowerInclusive, K upper, boolean
upperInclusive, SerializableComparator<K> comparator)` — same, but with an
explicit comparator for key types that do not implement `Comparable<K>`. The
comparator must be serializable because the filter is shipped with the job;
lambdas and method references assigned to `SerializableComparator` satisfy this
automatically.
-* `SavepointKeyFilter.empty()` — match no keys. Not intended for direct use —
it only serves as an internal building block for the Table API filter pushdown.
When the built-in filters are not enough, you can implement the
`SavepointKeyFilter<K>` interface yourself.
For use with the DataStream API, only `test(K key)` has to be implemented; it
is called for every key in each opened split and decides whether that key will
be read.
@@ -314,8 +313,6 @@ DataStream<KeyedState> firstKeys = savepoint.readKeyedState(
new UpToKeyFilter(100));
```
-The remaining interface methods can be left at their defaults for DataStream
API usage, as they are only used internally in the Table API during push-down
handling.
-
### 窗口状态 Window State
State Processor API 支持读取[窗口算子]({{< ref "docs/dev/datastream/operators/windows"
>}})的状态，当读取窗口状态时，需要指定算子 id，窗口分配器和聚合类型。
diff --git a/docs/content/docs/libs/state_processor_api.md
b/docs/content/docs/libs/state_processor_api.md
index 336502ab3f2..6b378cf110b 100644
--- a/docs/content/docs/libs/state_processor_api.md
+++ b/docs/content/docs/libs/state_processor_api.md
@@ -268,7 +268,6 @@ DataStream<KeyedState> keyRange = savepoint.readKeyedState(
* `SavepointKeyFilter.exact(K key)` / `SavepointKeyFilter.exact(Set<K> keys)`
— match a single key or a finite set of keys.
* `SavepointKeyFilter.range(K lower, boolean lowerInclusive, K upper, boolean
upperInclusive)` — match a range; `K` must implement `Comparable<K>`. Either
bound may be `null` to leave that side unbounded.
* `SavepointKeyFilter.range(K lower, boolean lowerInclusive, K upper, boolean
upperInclusive, SerializableComparator<K> comparator)` — same, but with an
explicit comparator for key types that do not implement `Comparable<K>`. The
comparator must be serializable because the filter is shipped with the job;
lambdas and method references assigned to `SerializableComparator` satisfy this
automatically.
-* `SavepointKeyFilter.empty()` — match no keys. Not intended for direct use —
it only serves as an internal building block for the Table API filter pushdown.
When the built-in filters are not enough, you can implement the
`SavepointKeyFilter<K>` interface yourself.
For use with the DataStream API, only `test(K key)` has to be implemented; it
is called for every key in each opened split and decides whether that key will
be read.
@@ -328,8 +327,6 @@ DataStream<KeyedState> firstKeys = savepoint.readKeyedState(
new UpToKeyFilter(100));
```
-The remaining interface methods can be left at their defaults for DataStream
API usage, as they are only used internally in the Table API during push-down
handling.
-
#### Window State
The state processor API supports reading state from a [window operator]({{<
ref "docs/dev/datastream/operators/windows" >}}).
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 fe0b525cd1c..00000000000
---
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<K> 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 160cc17d4a5..00000000000
---
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<K> implements SavepointKeyFilter<K> {
-
- private static final long serialVersionUID = 1L;
-
- @SuppressWarnings("rawtypes")
- private static final EmptyKeyFilter INSTANCE = new EmptyKeyFilter<>();
-
- private EmptyKeyFilter() {}
-
- @SuppressWarnings("unchecked")
- static <K> EmptyKeyFilter<K> instance() {
- return (EmptyKeyFilter<K>) INSTANCE;
- }
-
- @Override
- public boolean test(K key) {
- return false;
- }
-
- @Override
- public boolean isEmpty() {
- return true;
- }
-
- @Override
- public Set<K> getExactKeys() {
- return Collections.emptySet();
- }
-
- @Override
- public SavepointKeyFilter<K> intersect(SavepointKeyFilter<K> 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 13f37554d92..2cdf6772521 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 @@ final class ExactKeyFilter<K> implements
SavepointKeyFilter<K> {
return keys;
}
- @Override
- public SavepointKeyFilter<K> intersect(SavepointKeyFilter<K> other) {
- if (other.isEmpty()) {
- return other;
- }
- final Set<K> otherKeys = other.getExactKeys();
- if (otherKeys != null) {
- final Set<K> intersection = new HashSet<>(keys);
- intersection.retainAll(otherKeys);
- return SavepointKeyFilter.exact(intersection);
- }
- return SavepointKeyFilter.filterKeys(keys, other);
- }
-
@Override
public String toString() {
return "ExactKeyFilter" + keys;
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 392a06fbf87..ccef5909fa2 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 @@ package org.apache.flink.state.api.filter;
import javax.annotation.Nullable;
-import java.util.Set;
-
/** A filter based on a range with an injected comparator. */
final class RangeKeyFilter<K> implements SavepointKeyFilter<K> {
- private static final long serialVersionUID = 3L;
+ private static final long serialVersionUID = 4L;
private final SerializableComparator<K> comparator;
- @Nullable private final BoundInfo<K> lower;
- @Nullable private final BoundInfo<K> upper;
+ @Nullable private final K lower;
+ private final boolean isLowerInclusive;
+ @Nullable private final K upper;
+ private final boolean isUpperInclusive;
RangeKeyFilter(
SerializableComparator<K> comparator,
- @Nullable BoundInfo<K> lower,
- @Nullable BoundInfo<K> upper) {
+ @Nullable K lower,
+ boolean isLowerInclusive,
+ @Nullable K upper,
+ boolean isUpperInclusive) {
this.comparator = comparator;
this.lower = lower;
+ this.isLowerInclusive = isLowerInclusive;
this.upper = upper;
+ this.isUpperInclusive = isUpperInclusive;
}
@Override
public boolean test(K key) {
if (lower != null) {
- int cmp = comparator.compare(lower.getValue(), key);
- if (cmp > 0 || (cmp == 0 && !lower.isInclusive())) {
+ int cmp = comparator.compare(lower, key);
+ if (cmp > 0 || (cmp == 0 && !isLowerInclusive)) {
return false;
}
}
if (upper != null) {
- int cmp = comparator.compare(upper.getValue(), key);
- if (cmp < 0 || (cmp == 0 && !upper.isInclusive())) {
+ int cmp = comparator.compare(upper, key);
+ if (cmp < 0 || (cmp == 0 && !isUpperInclusive)) {
return false;
}
}
return true;
}
- @Override
- public BoundInfo<K> getLowerBound() {
- return lower;
- }
-
- @Override
- public BoundInfo<K> getUpperBound() {
- return upper;
- }
-
- @Override
- public SavepointKeyFilter<K> intersect(SavepointKeyFilter<K> other) {
- if (other.isEmpty()) {
- return other;
- }
- final Set<K> otherExactKeys = other.getExactKeys();
- if (otherExactKeys != null) {
- return SavepointKeyFilter.filterKeys(otherExactKeys, this);
- }
- return intersectRange(other.getLowerBound(), other.getUpperBound());
- }
-
- private SavepointKeyFilter<K> intersectRange(
- @Nullable BoundInfo<K> otherLower, @Nullable BoundInfo<K>
otherUpper) {
- BoundInfo<K> newLower = tighter(lower, otherLower, true);
- BoundInfo<K> newUpper = tighter(upper, otherUpper, false);
-
- if (newLower != null && newUpper != null) {
- int cmp = comparator.compare(newLower.getValue(),
newUpper.getValue());
- if (cmp > 0) {
- return SavepointKeyFilter.empty();
- }
- if (cmp == 0 && (!newLower.isInclusive() ||
!newUpper.isInclusive())) {
- return SavepointKeyFilter.empty();
- }
- }
- return new RangeKeyFilter<>(comparator, newLower, newUpper);
- }
-
- @Nullable
- private BoundInfo<K> tighter(
- @Nullable BoundInfo<K> a, @Nullable BoundInfo<K> b, boolean
preferHigher) {
- if (a == null) {
- return b;
- }
- if (b == null) {
- return a;
- }
- int c = comparator.compare(a.getValue(), b.getValue());
- if (c == 0) {
- return new BoundInfo<>(a.getValue(), a.isInclusive() &&
b.isInclusive());
- }
- boolean aWins = preferHigher ? c > 0 : c < 0;
- return aWins ? a : b;
- }
-
@Override
public String toString() {
- String lowerStr =
- lower == null ? "(-∞" : (lower.isInclusive() ? "[" : "(") +
lower.getValue();
- String upperStr =
- upper == null ? "+∞)" : upper.getValue() +
(upper.isInclusive() ? "]" : ")");
+ String lowerStr = lower == null ? "(-∞" : (isLowerInclusive ? "[" :
"(") + lower;
+ String upperStr = upper == null ? "+∞)" : upper + (isUpperInclusive ?
"]" : ")");
return "RangeKeyFilter" + lowerStr + ", " + upperStr;
}
}
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 c639c5c056d..ebf059ec544 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 org.apache.flink.annotation.Experimental;
import javax.annotation.Nullable;
import java.io.Serializable;
-import java.util.HashSet;
import java.util.Set;
/**
@@ -37,15 +36,6 @@ public interface SavepointKeyFilter<K> extends Serializable {
/** Returns {@code true} if the given key passes this filter. */
boolean test(K key);
- /**
- * Returns {@code true} if this filter rejects every key.
- *
- * <p>Used only while combining filters during push-down translation, not
during the scan.
- */
- default boolean isEmpty() {
- return false;
- }
-
/**
* Returns the finite set of keys this filter matches, or {@code null} if
the filter does not
* resolve to a finite key set.
@@ -55,58 +45,12 @@ public interface SavepointKeyFilter<K> extends Serializable
{
return null;
}
- /**
- * Returns the lower bound of this filter's range, or {@code null} if the
filter does not define
- * a lower bound.
- *
- * <p>Used only while combining filters during push-down translation, not
during the scan.
- */
- @Nullable
- default BoundInfo<K> getLowerBound() {
- return null;
- }
-
- /**
- * Returns the upper bound of this filter's range, or {@code null} if the
filter does not define
- * an upper bound.
- *
- * <p>Used only while combining filters during push-down translation, not
during the scan.
- */
- @Nullable
- default BoundInfo<K> getUpperBound() {
- return null;
- }
-
- /**
- * Returns a filter that accepts a key if and only if both {@code this}
and {@code other} accept
- * it.
- *
- * <p>Used only while combining filters during push-down translation, not
during the scan.
- */
- default SavepointKeyFilter<K> intersect(SavepointKeyFilter<K> other) {
- throw new UnsupportedOperationException(
- getClass().getSimpleName() + " does not support intersect()");
- }
-
- static <K> SavepointKeyFilter<K> filterKeys(Set<K> keys,
SavepointKeyFilter<K> predicate) {
- final Set<K> retained = new HashSet<>();
- for (K key : keys) {
- if (predicate.test(key)) {
- retained.add(key);
- }
- }
- return exact(retained);
- }
-
static <K> SavepointKeyFilter<K> exact(Set<K> keys) {
- if (keys.isEmpty()) {
- return EmptyKeyFilter.instance();
- }
return new ExactKeyFilter<>(keys);
}
static <K> SavepointKeyFilter<K> exact(K value) {
- return new ExactKeyFilter<>(Set.of(value));
+ return exact(Set.of(value));
}
static <K extends Comparable<K>> SavepointKeyFilter<K> range(
@@ -120,13 +64,7 @@ public interface SavepointKeyFilter<K> extends Serializable
{
@Nullable K upper,
boolean upperInclusive,
SerializableComparator<K> comparator) {
- BoundInfo<K> lowerBoundInfo = lower != null ? new BoundInfo<>(lower,
lowerInclusive) : null;
- BoundInfo<K> upperBoundInfo = upper != null ? new BoundInfo<>(upper,
upperInclusive) : null;
- return new RangeKeyFilter<>(comparator, lowerBoundInfo,
upperBoundInfo);
- }
-
- static <K> SavepointKeyFilter<K> empty() {
- return EmptyKeyFilter.instance();
+ return new RangeKeyFilter<>(comparator, lower, lowerInclusive, upper,
upperInclusive);
}
final class NaturalOrderComparator<K extends Comparable<K>>
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 d225accb804..e64592811a3 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 org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
@@ -51,10 +52,12 @@ import java.util.function.Consumer;
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<SavepointFilterTranslator, CallExpression,
SavepointKeyFilter>>
+ BiFunction<SavepointFilterTranslator, CallExpression,
KeyFilterPlan>>
FILTERS =
Map.of(
BuiltInFunctionDefinitions.EQUALS,
@@ -86,9 +89,9 @@ class SavepointFilterTranslator {
final List<ResolvedExpression> accepted = new ArrayList<>();
final List<ResolvedExpression> 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 @@ class SavepointFilterTranslator {
accepted.add(filter);
}
- return new Result(accepted, remaining, keyFilter);
+ return new Result(
+ accepted, remaining, keyFilter == null ? null :
keyFilter.toSavepointKeyFilter());
}
/**
@@ -123,8 +127,8 @@ class SavepointFilterTranslator {
}
@Nullable
- private SavepointKeyFilter extractFilter(ResolvedExpression expr) {
- final BiFunction<SavepointFilterTranslator, CallExpression,
SavepointKeyFilter> extractor =
+ private KeyFilterPlan extractFilter(ResolvedExpression expr) {
+ final BiFunction<SavepointFilterTranslator, CallExpression,
KeyFilterPlan> extractor =
expr instanceof CallExpression
? FILTERS.get(((CallExpression)
expr).getFunctionDefinition())
: null;
@@ -141,7 +145,7 @@ class SavepointFilterTranslator {
//
-------------------------------------------------------------------------
@Nullable
- private SavepointKeyFilter fromEquals(CallExpression call) {
+ private KeyFilterPlan fromEquals(CallExpression call) {
if (!isBinaryValid(call)) {
return null;
}
@@ -158,25 +162,25 @@ class SavepointFilterTranslator {
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<Object> keys = new HashSet<>();
for (ResolvedExpression arg : call.getResolvedChildren()) {
- SavepointKeyFilter sub = extractFilter(arg);
+ KeyFilterPlan sub = extractFilter(arg);
if (sub == null) {
return null;
}
- Set<Object> subKeys = sub.getExactKeys();
+ Set<Object> 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 @@ class SavepointFilterTranslator {
//
-------------------------------------------------------------------------
@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 @@ class SavepointFilterTranslator {
}
@Nullable
- private SavepointKeyFilter fromBetween(CallExpression call) {
+ private KeyFilterPlan fromBetween(CallExpression call) {
List<ResolvedExpression> args = call.getResolvedChildren();
if (args.size() != 3) {
return null;
@@ -225,13 +229,11 @@ class SavepointFilterTranslator {
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 @@ class SavepointFilterTranslator {
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 @@ class SavepointFilterTranslator {
}
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 @@ class SavepointFilterTranslator {
}
}
+ private static final class KeyFilterPlan {
+ @Nullable private final Set<Object> exactKeys;
+ @Nullable private final Comparable lower;
+ private final boolean lowerInclusive;
+ @Nullable private final Comparable upper;
+ private final boolean upperInclusive;
+
+ private KeyFilterPlan(
+ @Nullable Set<Object> 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<Object> 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<Object> 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<Object> toSavepointKeyFilter() {
+ if (isEmpty()) {
+ return SavepointKeyFilter.exact(Set.of());
+ }
+ if (exactKeys != null && lower == null && upper == null) {
+ return SavepointKeyFilter.exact(exactKeys);
+ }
+ final SavepointKeyFilter<Object> rangeFilter =
+ SavepointKeyFilter.range(lower, lowerInclusive, upper,
upperInclusive);
+ if (exactKeys == null) {
+ return rangeFilter;
+ }
+ final Set<Object> 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 00746a9285a..7da2167f144 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 @@ abstract class SavepointReaderKeyedStateITCase<B extends
StateBackend> extends S
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 00000000000..dd79101ad5f
--- /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<Long> 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<Long> 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<Object> 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<Long> keys = new HashSet<>(Set.of(1L, 2L));
+ SavepointKeyFilter<Long> filter = SavepointKeyFilter.exact(keys);
+
+ keys.add(3L);
+
+ assertThat(filter.getExactKeys()).containsExactlyInAnyOrder(1L, 2L);
+ assertThat(filter.test(3L)).isFalse();
+ }
+
+ @Test
+ void exactKeysAreUnmodifiable() {
+ SavepointKeyFilter<Long> 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<Long> filter = SavepointKeyFilter.exact(Set.of(1L,
2L));
+ SavepointKeyFilter<Long> 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 00000000000..8eb2fa49b33
--- /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<Long> 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<Long> 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<Long> lowerInclusive = SavepointKeyFilter.range(5L,
true, 10L, false);
+ assertThat(lowerInclusive.test(5L)).isTrue();
+ assertThat(lowerInclusive.test(10L)).isFalse();
+
+ SavepointKeyFilter<Long> 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<Long> 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<Long> 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<Long> 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<Long> 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<Long> 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<String> 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<String> 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<Integer> 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<Long> filter = SavepointKeyFilter.range(5L, true,
10L, false);
+ SavepointKeyFilter<Long> 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<Integer> descending = (a, b) ->
Integer.compare(b, a);
+ SavepointKeyFilter<Integer> filter = SavepointKeyFilter.range(6, true,
3, true, descending);
+ SavepointKeyFilter<Integer> 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 bde20fd0f57..b1183f6bb91 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 javax.annotation.Nonnull;
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 @@ class KeyedStateInputFormatTest {
@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 @@ class KeyedStateInputFormatTest {
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 3b587f7ce6e..ba10ef10e4b 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.apache.flink.types.Row;
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 @@ class SavepointDynamicTableSourceTest {
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 @@ class SavepointDynamicTableSourceTest {
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 @@ class SavepointDynamicTableSourceTest {
// Helpers
//
-------------------------------------------------------------------------
+ private static List<Object> 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<Long> 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 @@ class SavepointDynamicTableSourceTest {
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 3e490b97160..685b4145b61 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.apache.flink.table.types.DataType;
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<Object> filter = keyFilterOf(eq(longKeyRef(),
longLit(42L)));
assertNotNull(filter);
assertThat(filter.getExactKeys()).containsExactly(42L);
@@ -62,6 +59,7 @@ class SavepointFilterTranslatorTest {
@Test
void equalsKeyOnRight() {
+ // 42 = key -> {42}
SavepointKeyFilter<Object> filter = keyFilterOf(eq(longLit(42L),
longKeyRef()));
assertNotNull(filter);
assertThat(filter.getExactKeys()).containsExactly(42L);
@@ -71,12 +69,14 @@ class SavepointFilterTranslatorTest {
@Test
void equalsNeitherSideIsKeyColumn_returnsNull() {
+ // val = 42 -> not the key column, not pushed
SavepointKeyFilter<Object> filter = keyFilterOf(eq(otherRef(),
longLit(42L)));
assertThat(filter).isNull();
}
@Test
void equalsNeitherSideIsLiteral_returnsNull() {
+ // key = val -> no literal to match, not pushed
SavepointKeyFilter<Object> filter = keyFilterOf(eq(longKeyRef(),
otherRef()));
assertThat(filter).isNull();
}
@@ -87,6 +87,7 @@ class SavepointFilterTranslatorTest {
@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 @@ class SavepointFilterTranslatorTest {
@Test
void betweenProducesInclusiveRange() {
+ // key BETWEEN 10 AND 20 -> [10, 20]
SavepointKeyFilter<Object> filter =
keyFilterOf(between(longKeyRef(), longLit(10L), longLit(20L)));
@@ -126,6 +128,7 @@ class SavepointFilterTranslatorTest {
@Test
void betweenWithNonKeyField_returnsNull() {
+ // val BETWEEN 1 AND 10 -> not the key column, not pushed
SavepointKeyFilter<Object> filter =
keyFilterOf(between(otherRef(), longLit(1L), longLit(10L)));
assertThat(filter).isNull();
@@ -137,6 +140,7 @@ class SavepointFilterTranslatorTest {
@Test
void greaterThanProducesExclusiveLowerBound() {
+ // key > 5 -> (5, +∞)
SavepointKeyFilter<Object> filter = keyFilterOf(gt(longKeyRef(),
longLit(5L)));
assertNotNull(filter);
assertThat(filter.getExactKeys()).isNull();
@@ -146,6 +150,7 @@ class SavepointFilterTranslatorTest {
@Test
void greaterThanOrEqualProducesInclusiveLowerBound() {
+ // key >= 5 -> [5, +∞)
SavepointKeyFilter<Object> filter = keyFilterOf(gte(longKeyRef(),
longLit(5L)));
assertNotNull(filter);
assertThat(filter.getExactKeys()).isNull();
@@ -156,6 +161,7 @@ class SavepointFilterTranslatorTest {
@Test
void lessThanProducesExclusiveUpperBound() {
+ // key < 10 -> (-∞, 10)
SavepointKeyFilter<Object> filter = keyFilterOf(lt(longKeyRef(),
longLit(10L)));
assertNotNull(filter);
assertThat(filter.getExactKeys()).isNull();
@@ -165,6 +171,7 @@ class SavepointFilterTranslatorTest {
@Test
void lessThanOrEqualProducesInclusiveUpperBound() {
+ // key <= 10 -> (-∞, 10]
SavepointKeyFilter<Object> filter = keyFilterOf(lte(longKeyRef(),
longLit(10L)));
assertNotNull(filter);
assertThat(filter.getExactKeys()).isNull();
@@ -216,7 +223,30 @@ class SavepointFilterTranslatorTest {
SavepointKeyFilter<Object> 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<Object> 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<Object> filter = keyFilterOf(expr);
+
+ assertNotNull(filter);
+ assertThat(filter.getExactKeys()).isEmpty();
}
@Test
@@ -232,11 +262,13 @@ class SavepointFilterTranslatorTest {
@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 @@ class SavepointFilterTranslatorTest {
@Test
void rangeFilterOnStringKey() {
+ // key BETWEEN 'beta' AND 'delta' -> natural String order
SavepointKeyFilter<Object> filter =
keyFilterOf(between(stringKeyRef(), stringLit("beta"),
stringLit("delta")));
@@ -264,6 +297,7 @@ class SavepointFilterTranslatorTest {
@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 @@ class SavepointFilterTranslatorTest {
assertThat(filter.test(4.0f)).isFalse();
}
- //
-------------------------------------------------------------------------
- // Range intersection
- //
-------------------------------------------------------------------------
-
- @Test
- void intersectNarrowsBounds() {
- // [5, ∞) ∩ (-∞, 10] = [5, 10]
- SavepointKeyFilter<Long> lower = SavepointKeyFilter.range(5L, true,
null, true);
- SavepointKeyFilter<Long> upper = SavepointKeyFilter.range(null, true,
10L, true);
- SavepointKeyFilter<Long> 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<Long> a = SavepointKeyFilter.range(10L, true, null,
true);
- SavepointKeyFilter<Long> b = SavepointKeyFilter.range(null, true, 5L,
true);
- assertThat(a.intersect(b).isEmpty()).isTrue();
- }
-
- @Test
- void intersectEqualBoundsInclusiveIsNonEmpty() {
- // [7, ∞) ∩ (-∞, 7] = [7, 7]
- SavepointKeyFilter<Long> a = SavepointKeyFilter.range(7L, true, null,
true);
- SavepointKeyFilter<Long> b = SavepointKeyFilter.range(null, true, 7L,
true);
- SavepointKeyFilter<Long> 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<Long> a = SavepointKeyFilter.range(7L, false, null,
true);
- SavepointKeyFilter<Long> 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<String> 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 @@ class SavepointFilterTranslatorTest {
@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 @@ class SavepointFilterTranslatorTest {
@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 @@ class SavepointFilterTranslatorTest {
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 @@ class SavepointFilterTranslatorTest {
@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 @@ class SavepointFilterTranslatorTest {
assertThat(result.getExactKeys()).containsExactlyInAnyOrder(5L, 10L);
}
- //
-------------------------------------------------------------------------
- // Empty key filter
- //
-------------------------------------------------------------------------
-
- @Test
- void emptyKeyFilter_rejectsEverything() {
- SavepointKeyFilter<Object> 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<Object> filter =
SavepointKeyFilter.exact(Collections.emptySet());
- assertThat(filter.isEmpty()).isTrue();
- }
-
- @Test
- void emptyKeyFilterSingletonPreservedAcrossSerialization() throws
Exception {
- SavepointKeyFilter<Object> 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<Long> 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 @@ class SavepointFilterTranslatorTest {
@Test
void orWithSingleChild_returnsExactFilter() {
+ // OR (key = 7) -> {7}
CallExpression expr = or(eq(longKeyRef(), longLit(7L)));
SavepointKeyFilter<Object> filter = keyFilterOf(expr);
@@ -534,6 +460,7 @@ class SavepointFilterTranslatorTest {
@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 @@ class SavepointFilterTranslatorTest {
@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 @@ class SavepointFilterTranslatorTest {
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 @@ class SavepointFilterTranslatorTest {
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 @@ class SavepointFilterTranslatorTest {
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 @@ class SavepointFilterTranslatorTest {
assertNotNull(result);
assertThat(applied.accepted()).hasSize(2);
assertThat(applied.remaining()).isEmpty();
- assertThat(result.isEmpty()).isTrue();
+ assertThat(result.getExactKeys()).isEmpty();
}
//
-------------------------------------------------------------------------
@@ -610,6 +541,7 @@ class SavepointFilterTranslatorTest {
@Test
void betweenWithWrongArgCount_returnsNull() {
+ // BETWEEN with two children is malformed, not pushed
CallExpression malformed =
CallExpression.permanent(
BuiltInFunctionDefinitions.BETWEEN,
@@ -624,6 +556,7 @@ class SavepointFilterTranslatorTest {
@Test
void betweenWithNonLiteralBound_returnsNull() {
+ // key BETWEEN val AND 10 -> non-literal bound, not pushed
CallExpression expr =
CallExpression.permanent(
BuiltInFunctionDefinitions.BETWEEN,
@@ -638,6 +571,7 @@ class SavepointFilterTranslatorTest {
@Test
void comparisonWithNonLiteralValue_returnsNull() {
+ // key > val -> no literal bound, not pushed
assertThat(keyFilterOf(gt(longKeyRef(), otherRef()))).isNull();
}
@@ -647,6 +581,7 @@ class SavepointFilterTranslatorTest {
@Test
void equalsWithWrongArgCount_returnsNull() {
+ // EQUALS with one child is malformed, not pushed
CallExpression malformed =
CallExpression.permanent(
BuiltInFunctionDefinitions.EQUALS,
@@ -661,6 +596,7 @@ class SavepointFilterTranslatorTest {
@Test
void equalsWithIntLiteralIsWidenedToBigintKeyAndPushed() {
+ // key = 5 (INT literal, BIGINT key) -> {5L}
ValueLiteralExpression intLit = new ValueLiteralExpression(5,
DataTypes.INT().notNull());
SavepointKeyFilter<Object> filter = keyFilterOf(eq(longKeyRef(),
intLit));
@@ -672,6 +608,7 @@ class SavepointFilterTranslatorTest {
@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<Object> filter = keyFilterOf(between(longKeyRef(),
lower, upper));
@@ -686,6 +623,7 @@ class SavepointFilterTranslatorTest {
@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 @@ class SavepointFilterTranslatorTest {
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<Object> 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<Object> filter = keyFilterOf(eq(longKeyRef(),
floatLit));
+
+ assertNotNull(filter);
+ assertThat(filter.getExactKeys()).containsExactly(5L);
+ }
+
+ @Test
+ void approximateLiteralAtBigintExactIntegerLimitIsNotPushed() {
+ List<ValueLiteralExpression> 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<ValueLiteralExpression> 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<Object> 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<Object> 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<Object> 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<Object> 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<Object> 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<Object> 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<Object> 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 @@ class SavepointFilterTranslatorTest {
@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 @@ class SavepointFilterTranslatorTest {
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<Object> 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<Object> 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<Object> 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 @@ class SavepointFilterTranslatorTest {
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 @@ class SavepointFilterTranslatorTest {
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 00000000000..b14c172192b
Binary files /dev/null and
b/flink-libraries/flink-state-processing-api/src/test/resources/table-state-typed-keys/_metadata
differ
