This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 79530d9bb4 [flink] Support negated predicates in PredicateConverter
(#9427)
79530d9bb4 is described below
commit 79530d9bb4f7938fe3d750f499a425f53c23acd3
Author: Wenchao Wu <[email protected]>
AuthorDate: Sun Sep 20 13:13:08 2026 +0800
[flink] Support negated predicates in PredicateConverter (#9427)
---
.../apache/paimon/flink/PredicateConverter.java | 292 ++++++++++---
.../paimon/flink/PredicateConverterTest.java | 454 ++++++++++++++++++++-
.../apache/paimon/flink/ReadWriteTableITCase.java | 92 +++++
.../paimon/flink/source/FlinkTableSourceTest.java | 235 +++++++++++
4 files changed, 1016 insertions(+), 57 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PredicateConverter.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PredicateConverter.java
index db3754f86d..d14b1ccce1 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PredicateConverter.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PredicateConverter.java
@@ -36,13 +36,13 @@ import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.LogicalTypeFamily;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.flink.table.types.logical.RowType;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;
-import java.util.Objects;
import java.util.Optional;
import java.util.function.BiFunction;
import java.util.regex.Matcher;
@@ -74,74 +74,136 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
@Override
public Predicate visit(CallExpression call) {
+ return visit(call, false);
+ }
+
+ private Predicate visit(CallExpression call, boolean negated) {
FunctionDefinition func = call.getFunctionDefinition();
List<Expression> children = call.getChildren();
if (func == BuiltInFunctionDefinitions.AND) {
- return PredicateBuilder.and(flattenAndConvert(children, func));
+ requireAtLeastArity(children, 2);
+ List<Predicate> predicates = flattenAndConvert(children, func,
negated);
+ return negated ? PredicateBuilder.or(predicates) :
PredicateBuilder.and(predicates);
} else if (func == BuiltInFunctionDefinitions.OR) {
- return PredicateBuilder.or(flattenAndConvert(children, func));
+ requireAtLeastArity(children, 2);
+ List<Predicate> predicates = flattenAndConvert(children, func,
negated);
+ return negated ? PredicateBuilder.and(predicates) :
PredicateBuilder.or(predicates);
+ } else if (func == BuiltInFunctionDefinitions.NOT) {
+ requireArity(children, 1);
+ return visit(children.get(0), !negated);
} else if (func == BuiltInFunctionDefinitions.EQUALS) {
- return visitBiFunction(children, builder::equal, builder::equal);
+ return visitComparison(
+ children,
+ negated,
+ builder::notEqual,
+ builder::notEqual,
+ builder::equal,
+ builder::equal);
} else if (func == BuiltInFunctionDefinitions.NOT_EQUALS) {
- return visitBiFunction(children, builder::notEqual,
builder::notEqual);
+ return visitComparison(
+ children,
+ negated,
+ builder::equal,
+ builder::equal,
+ builder::notEqual,
+ builder::notEqual);
} else if (func == BuiltInFunctionDefinitions.GREATER_THAN) {
- return visitBiFunction(children, builder::greaterThan,
builder::lessThan);
+ return visitComparison(
+ children,
+ negated,
+ builder::lessOrEqual,
+ builder::greaterOrEqual,
+ builder::greaterThan,
+ builder::lessThan);
} else if (func == BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL) {
- return visitBiFunction(children, builder::greaterOrEqual,
builder::lessOrEqual);
+ return visitComparison(
+ children,
+ negated,
+ builder::lessThan,
+ builder::greaterThan,
+ builder::greaterOrEqual,
+ builder::lessOrEqual);
} else if (func == BuiltInFunctionDefinitions.LESS_THAN) {
- return visitBiFunction(children, builder::lessThan,
builder::greaterThan);
+ return visitComparison(
+ children,
+ negated,
+ builder::greaterOrEqual,
+ builder::lessOrEqual,
+ builder::lessThan,
+ builder::greaterThan);
} else if (func == BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL) {
- return visitBiFunction(children, builder::lessOrEqual,
builder::greaterOrEqual);
+ return visitComparison(
+ children,
+ negated,
+ builder::greaterThan,
+ builder::lessThan,
+ builder::lessOrEqual,
+ builder::greaterOrEqual);
} else if (func == BuiltInFunctionDefinitions.IN) {
- FieldReferenceExpression fieldRefExpr =
-
extractFieldReference(children.get(0)).orElseThrow(UnsupportedExpression::new);
+ requireAtLeastArity(children, 2);
+ ResolvedField field = resolveField(children.get(0));
List<Object> literals = new ArrayList<>();
for (int i = 1; i < children.size(); i++) {
- literals.add(extractLiteral(fieldRefExpr.getOutputDataType(),
children.get(i)));
+
literals.add(extractLiteral(field.expression.getOutputDataType(),
children.get(i)));
+ }
+ if (negated) {
+ // SQL WHERE: v NOT IN (..., NULL, ...) is never true
regardless of
+ // type, so this runs before the float residual. Passing NULL
to
+ // notIn is also unsafe for BSI/range-bitmap file-index
evaluation.
+ if (literals.contains(null)) {
+ return PredicateBuilder.alwaysFalse();
+ }
+ rejectNegatedFloatingPoint(field.expression);
+ return builder.notIn(field.index, literals);
}
- return builder.in(builder.indexOf(fieldRefExpr.getName()),
literals);
+ return builder.in(field.index, literals);
} else if (func == BuiltInFunctionDefinitions.IS_NULL) {
- return extractFieldReference(children.get(0))
- .map(FieldReferenceExpression::getName)
- .map(builder::indexOf)
- .map(builder::isNull)
- .orElseThrow(UnsupportedExpression::new);
+ requireArity(children, 1);
+ ResolvedField field = resolveField(children.get(0));
+ return negated ? builder.isNotNull(field.index) :
builder.isNull(field.index);
} else if (func == BuiltInFunctionDefinitions.IS_NOT_NULL) {
- return extractFieldReference(children.get(0))
- .map(FieldReferenceExpression::getName)
- .map(builder::indexOf)
- .map(builder::isNotNull)
- .orElseThrow(UnsupportedExpression::new);
+ requireArity(children, 1);
+ ResolvedField field = resolveField(children.get(0));
+ return negated ? builder.isNull(field.index) :
builder.isNotNull(field.index);
} else if (func == BuiltInFunctionDefinitions.BETWEEN) {
- FieldReferenceExpression fieldRefExpr =
-
extractFieldReference(children.get(0)).orElseThrow(UnsupportedExpression::new);
- DataType fieldType = fieldRefExpr.getOutputDataType();
- return builder.between(
- builder.indexOf(fieldRefExpr.getName()),
- extractLiteral(fieldType, children.get(1)),
- extractLiteral(fieldType, children.get(2)));
+ requireArity(children, 3);
+ ResolvedField field = resolveField(children.get(0));
+ DataType fieldType = field.expression.getOutputDataType();
+ Object lower = extractLiteral(fieldType, children.get(1));
+ Object upper = extractLiteral(fieldType, children.get(2));
+ Predicate between = builder.between(field.index, lower, upper);
+ if (negated) {
+ rejectNegatedFloatingPoint(field.expression);
+ // LeafTernaryFunction.test returns false if any literal is
null, but
+ // 12 NOT BETWEEN 15 AND NULL is TRUE (TRUE OR UNKNOWN). Keep
residual
+ // so Flink can evaluate the three-valued cases.
+ if (lower == null || upper == null) {
+ throw new UnsupportedExpression();
+ }
+ return negate(between);
+ }
+ return between;
} else if (func == BuiltInFunctionDefinitions.LIKE) {
- FieldReferenceExpression fieldRefExpr =
-
extractFieldReference(children.get(0)).orElseThrow(UnsupportedExpression::new);
- if (fieldRefExpr
+ if (children.size() != 2 && children.size() != 3) {
+ throw new UnsupportedExpression();
+ }
+ ResolvedField field = resolveField(children.get(0));
+ if (field.expression
.getOutputDataType()
.getLogicalType()
.getTypeRoot()
.getFamilies()
.contains(LogicalTypeFamily.CHARACTER_STRING)) {
String sqlPattern =
- Objects.requireNonNull(
- extractLiteral(
-
fieldRefExpr.getOutputDataType(), children.get(1)))
+
extractNonNullLiteral(field.expression.getOutputDataType(), children.get(1))
.toString();
String escape =
children.size() <= 2
? null
- : Objects.requireNonNull(
- extractLiteral(
-
fieldRefExpr.getOutputDataType(),
- children.get(2)))
+ : extractNonNullLiteral(
+
field.expression.getOutputDataType(),
+ children.get(2))
.toString();
String escapedSqlPattern = sqlPattern;
boolean allowQuick = false;
@@ -185,27 +247,57 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
if (allowQuick) {
Matcher beginMatcher =
BEGIN_PATTERN.matcher(escapedSqlPattern);
if (beginMatcher.matches()) {
+ if (negated) {
+ // StartsWith has no negated predicate, so NOT
LIKE must remain a
+ // residual filter evaluated by Flink.
+ throw new UnsupportedExpression();
+ }
return builder.startsWith(
- builder.indexOf(fieldRefExpr.getName()),
-
BinaryString.fromString(beginMatcher.group(1)));
+ field.index,
BinaryString.fromString(beginMatcher.group(1)));
}
}
}
} else if (func == BuiltInFunctionDefinitions.IS_TRUE) {
- FieldReferenceExpression fieldRefExpr =
-
extractFieldReference(children.get(0)).orElseThrow(UnsupportedExpression::new);
- return builder.equal(builder.indexOf(fieldRefExpr.getName()),
Boolean.TRUE);
+ requireArity(children, 1);
+ return booleanTest(resolveField(children.get(0)), true, negated);
} else if (func == BuiltInFunctionDefinitions.IS_FALSE) {
- FieldReferenceExpression fieldRefExpr =
-
extractFieldReference(children.get(0)).orElseThrow(UnsupportedExpression::new);
- return builder.equal(builder.indexOf(fieldRefExpr.getName()),
Boolean.FALSE);
+ requireArity(children, 1);
+ return booleanTest(resolveField(children.get(0)), false, negated);
+ } else if (func == BuiltInFunctionDefinitions.IS_NOT_TRUE) {
+ requireArity(children, 1);
+ return booleanTest(resolveField(children.get(0)), true, !negated);
+ } else if (func == BuiltInFunctionDefinitions.IS_NOT_FALSE) {
+ requireArity(children, 1);
+ return booleanTest(resolveField(children.get(0)), false, !negated);
}
- // TODO is_xxx, between_xxx, similar, in, not_in, not?
+ throw new UnsupportedExpression();
+ }
+ private Predicate visit(Expression expression, boolean negated) {
+ if (expression instanceof CallExpression) {
+ return visit((CallExpression) expression, negated);
+ }
throw new UnsupportedExpression();
}
+ private Predicate booleanTest(ResolvedField field, boolean expected,
boolean complement) {
+ if (field.expression.getOutputDataType().getLogicalType().getTypeRoot()
+ != LogicalTypeRoot.BOOLEAN) {
+ throw new UnsupportedExpression();
+ }
+ Predicate equals = builder.equal(field.index, expected);
+ if (!complement) {
+ return equals;
+ }
+ return PredicateBuilder.or(
+ builder.isNull(field.index), builder.notEqual(field.index,
expected));
+ }
+
+ private Predicate negate(Predicate predicate) {
+ return predicate.negate().orElseThrow(UnsupportedExpression::new);
+ }
+
/**
* Iteratively flattens a nested AND/OR expression tree into a flat list
of child predicates,
* avoiding stack overflow caused by recursive {@code accept} calls on
deeply nested trees (e.g.
@@ -213,10 +305,11 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
*
* @param children the children of the top-level AND/OR {@link
CallExpression}
* @param targetFunc the function definition to flatten ({@code AND} or
{@code OR})
+ * @param negated whether to negate every flattened child and combine them
using De Morgan's law
* @return a flat list of converted child predicates in original order
*/
private List<Predicate> flattenAndConvert(
- List<Expression> children, FunctionDefinition targetFunc) {
+ List<Expression> children, FunctionDefinition targetFunc, boolean
negated) {
List<Predicate> result = new ArrayList<>();
Deque<Expression> stack = new ArrayDeque<>();
for (int i = children.size() - 1; i >= 0; i--) {
@@ -228,40 +321,112 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
CallExpression ce = (CallExpression) expr;
if (ce.getFunctionDefinition() == targetFunc) {
List<Expression> ceChildren = ce.getChildren();
+ requireAtLeastArity(ceChildren, 2);
for (int i = ceChildren.size() - 1; i >= 0; i--) {
stack.push(ceChildren.get(i));
}
} else {
- result.add(ce.accept(this));
+ result.add(visit(ce, negated));
}
} else {
- result.add(expr.accept(this));
+ result.add(visit(expr, negated));
}
}
return result;
}
+ private Predicate visitComparison(
+ List<Expression> children,
+ boolean negated,
+ BiFunction<Integer, Object, Predicate> negatedVisit1,
+ BiFunction<Integer, Object, Predicate> negatedVisit2,
+ BiFunction<Integer, Object, Predicate> visit1,
+ BiFunction<Integer, Object, Predicate> visit2) {
+ // Flink FLOAT/DOUBLE comparisons use Java operators; Paimon uses
+ // Float/Double.compareTo. Negated equality, inequality, IN and BETWEEN
+ // are therefore not equivalent (NaN identity and signed zeros). Simple
+ // SQL such as NOT (d > 1.0) is often simplified to d <= 1.0 before
+ // pushdown; keep this guard for unsimplified NOT that still reaches
+ // applyFilters (for example De Morgan over AND/OR).
+ if (negated && isFloatingPointComparison(children)) {
+ throw new UnsupportedExpression();
+ }
+ return negated
+ ? visitBiFunction(children, negatedVisit1, negatedVisit2)
+ : visitBiFunction(children, visit1, visit2);
+ }
+
+ private void rejectNegatedFloatingPoint(FieldReferenceExpression field) {
+ if (isFloatingPointField(field)) {
+ throw new UnsupportedExpression();
+ }
+ }
+
+ private boolean isFloatingPointComparison(List<Expression> children) {
+ for (Expression child : children) {
+ Optional<FieldReferenceExpression> field =
extractFieldReference(child);
+ if (field.isPresent() && isFloatingPointField(field.get())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private boolean isFloatingPointField(FieldReferenceExpression field) {
+ LogicalTypeRoot root =
field.getOutputDataType().getLogicalType().getTypeRoot();
+ return root == LogicalTypeRoot.FLOAT || root == LogicalTypeRoot.DOUBLE;
+ }
+
private Predicate visitBiFunction(
List<Expression> children,
BiFunction<Integer, Object, Predicate> visit1,
BiFunction<Integer, Object, Predicate> visit2) {
+ requireArity(children, 2);
Optional<FieldReferenceExpression> fieldRefExpr =
extractFieldReference(children.get(0));
if (fieldRefExpr.isPresent()) {
+ int fieldIndex = resolveFieldIndex(fieldRefExpr.get());
Object literal =
extractLiteral(fieldRefExpr.get().getOutputDataType(),
children.get(1));
- return visit1.apply(builder.indexOf(fieldRefExpr.get().getName()),
literal);
+ return visit1.apply(fieldIndex, literal);
} else {
fieldRefExpr = extractFieldReference(children.get(1));
if (fieldRefExpr.isPresent()) {
+ int fieldIndex = resolveFieldIndex(fieldRefExpr.get());
Object literal =
extractLiteral(fieldRefExpr.get().getOutputDataType(),
children.get(0));
- return
visit2.apply(builder.indexOf(fieldRefExpr.get().getName()), literal);
+ return visit2.apply(fieldIndex, literal);
}
}
throw new UnsupportedExpression();
}
+ private ResolvedField resolveField(Expression expression) {
+ FieldReferenceExpression field =
+
extractFieldReference(expression).orElseThrow(UnsupportedExpression::new);
+ return new ResolvedField(field, resolveFieldIndex(field));
+ }
+
+ private int resolveFieldIndex(FieldReferenceExpression field) {
+ int index = builder.indexOf(field.getName());
+ if (index < 0) {
+ throw new UnsupportedExpression();
+ }
+ return index;
+ }
+
+ private void requireArity(List<Expression> children, int expected) {
+ if (children.size() != expected) {
+ throw new UnsupportedExpression();
+ }
+ }
+
+ private void requireAtLeastArity(List<Expression> children, int minimum) {
+ if (children.size() < minimum) {
+ throw new UnsupportedExpression();
+ }
+ }
+
private Optional<FieldReferenceExpression>
extractFieldReference(Expression expression) {
if (expression instanceof FieldReferenceExpression) {
return Optional.of((FieldReferenceExpression) expression);
@@ -304,6 +469,14 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
throw new UnsupportedExpression();
}
+ private Object extractNonNullLiteral(DataType expectedType, Expression
expression) {
+ Object literal = extractLiteral(expectedType, expression);
+ if (literal == null) {
+ throw new UnsupportedExpression();
+ }
+ return literal;
+ }
+
private boolean supportsPredicate(LogicalType type) {
switch (type.getTypeRoot()) {
case CHAR:
@@ -331,6 +504,17 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
}
}
+ private static class ResolvedField {
+
+ private final FieldReferenceExpression expression;
+ private final int index;
+
+ private ResolvedField(FieldReferenceExpression expression, int index) {
+ this.expression = expression;
+ this.index = index;
+ }
+ }
+
@Override
public Predicate visit(ValueLiteralExpression valueLiteralExpression) {
throw new UnsupportedExpression();
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PredicateConverterTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PredicateConverterTest.java
index e05773991f..b472fdafd8 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PredicateConverterTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PredicateConverterTest.java
@@ -22,6 +22,9 @@ import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.format.SimpleColStats;
import org.apache.paimon.predicate.CompoundPredicate;
+import org.apache.paimon.predicate.In;
+import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.NotBetween;
import org.apache.paimon.predicate.Or;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
@@ -37,7 +40,9 @@ import org.apache.flink.table.functions.FunctionDefinition;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.DoubleType;
+import org.apache.flink.table.types.logical.FloatType;
import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.types.Row;
@@ -46,6 +51,7 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -79,8 +85,7 @@ public class PredicateConverterTest {
@ParameterizedTest
public void testVisitAndAutoTypeInference(ResolvedExpression expression,
Predicate expected) {
if (expression instanceof CallExpression) {
- assertThat(CONVERTER.visit((CallExpression) expression).toString())
- .isEqualTo(expected.toString());
+ assertThat(CONVERTER.visit((CallExpression)
expression)).isEqualTo(expected);
} else {
assertThatThrownBy(() -> CONVERTER.visit(expression))
.isInstanceOf(PredicateConverter.UnsupportedExpression.class);
@@ -258,7 +263,7 @@ public class PredicateConverterTest {
BuiltInFunctionDefinitions.BETWEEN,
Arrays.asList(longRefExpr, intLitExpr,
intLitExpr2),
DataTypes.BOOLEAN()),
- BUILDER.between(0, 10, 20)),
+ BUILDER.between(0, 10L, 20L)),
Arguments.of(
CallExpression.permanent(
BuiltInFunctionDefinitions.IS_TRUE,
@@ -314,6 +319,449 @@ public class PredicateConverterTest {
assertThat(nullUpperBound.test(GenericRow.of(15L))).isFalse();
}
+ @Test
+ public void testInAndNotInRowSemantics() {
+ PredicateConverter converter = new PredicateConverter(RowType.of(new
BigIntType()));
+ PredicateBuilder builder = predicateBuilder(RowType.of(new
BigIntType()));
+ CallExpression in =
+ call(
+ BuiltInFunctionDefinitions.IN,
+ field(0, DataTypes.BIGINT()),
+ literal(1, DataTypes.INT()),
+ literal(null, DataTypes.BIGINT()),
+ literal(3, DataTypes.INT()));
+
+ Predicate inPredicate = in.accept(converter);
+ Predicate notInPredicate = call(BuiltInFunctionDefinitions.NOT,
in).accept(converter);
+
+ assertThat(inPredicate).isEqualTo(builder.in(0, Arrays.asList(1L,
null, 3L)));
+ assertThat(notInPredicate).isEqualTo(PredicateBuilder.alwaysFalse());
+ assertThat(inPredicate.test(GenericRow.of(1L))).isTrue();
+ assertThat(inPredicate.test(GenericRow.of(2L))).isFalse();
+ assertThat(inPredicate.test(GenericRow.of(3L))).isTrue();
+ assertThat(inPredicate.test(GenericRow.of((Object) null))).isFalse();
+ for (Object value : Arrays.asList(null, 1L, 2L, 3L, 4L)) {
+ assertThat(notInPredicate.test(GenericRow.of(value))).isFalse();
+ }
+ }
+
+ @Test
+ public void testLargeInAndNotIn() {
+ PredicateConverter converter = new PredicateConverter(RowType.of(new
BigIntType()));
+ PredicateBuilder builder = predicateBuilder(RowType.of(new
BigIntType()));
+ List<ResolvedExpression> children = new ArrayList<>();
+ List<Object> expectedLiterals = new ArrayList<>();
+ children.add(field(0, DataTypes.BIGINT()));
+ for (int i = 0; i < 21; i++) {
+ children.add(literal(i, DataTypes.INT()));
+ expectedLiterals.add((long) i);
+ }
+ CallExpression in =
+ new CallExpression(
+ false, null, BuiltInFunctionDefinitions.IN, children,
DataTypes.BOOLEAN());
+
+ Predicate inPredicate = in.accept(converter);
+ Predicate notInPredicate = call(BuiltInFunctionDefinitions.NOT,
in).accept(converter);
+
+ assertThat(inPredicate).isEqualTo(builder.in(0, expectedLiterals));
+ assertThat(inPredicate).isInstanceOf(LeafPredicate.class);
+ assertThat(((LeafPredicate)
inPredicate).function()).isEqualTo(In.INSTANCE);
+ assertThat(((LeafPredicate) inPredicate).literals())
+ .containsExactlyElementsOf(expectedLiterals);
+ assertThat(notInPredicate).isEqualTo(builder.notIn(0,
expectedLiterals));
+ assertThat(inPredicate.test(GenericRow.of(20L))).isTrue();
+ assertThat(inPredicate.test(GenericRow.of(21L))).isFalse();
+ assertThat(inPredicate.test(GenericRow.of((Object) null))).isFalse();
+ assertThat(notInPredicate.test(GenericRow.of(20L))).isFalse();
+ assertThat(notInPredicate.test(GenericRow.of(21L))).isTrue();
+ assertThat(notInPredicate.test(GenericRow.of((Object)
null))).isFalse();
+ }
+
+ @Test
+ public void testNotBetweenStructure() {
+ PredicateConverter converter = new PredicateConverter(RowType.of(new
BigIntType()));
+ PredicateBuilder builder = predicateBuilder(RowType.of(new
BigIntType()));
+ CallExpression between =
+ call(
+ BuiltInFunctionDefinitions.BETWEEN,
+ field(0, DataTypes.BIGINT()),
+ literal(10, DataTypes.INT()),
+ literal(20, DataTypes.INT()));
+
+ Predicate predicate = call(BuiltInFunctionDefinitions.NOT,
between).accept(converter);
+
+ assertThat(predicate).isEqualTo(builder.between(0, 10L,
20L).negate().get());
+ assertThat(predicate).isInstanceOf(LeafPredicate.class);
+ assertThat(((LeafPredicate)
predicate).function()).isEqualTo(NotBetween.INSTANCE);
+ assertThat(((LeafPredicate)
predicate).literals()).containsExactly(10L, 20L);
+ }
+
+ @Test
+ public void testNotBetweenWithNullBoundsRemainUnsupported() {
+ RowType rowType = RowType.of(new BigIntType());
+ PredicateConverter converter = new PredicateConverter(rowType);
+ FieldReferenceExpression field = field(0, DataTypes.BIGINT());
+ CallExpression nullUpper =
+ call(
+ BuiltInFunctionDefinitions.NOT,
+ call(
+ BuiltInFunctionDefinitions.BETWEEN,
+ field,
+ literal(15, DataTypes.INT()),
+ literal(null, DataTypes.BIGINT())));
+ CallExpression nullLower =
+ call(
+ BuiltInFunctionDefinitions.NOT,
+ call(
+ BuiltInFunctionDefinitions.BETWEEN,
+ field,
+ literal(null, DataTypes.BIGINT()),
+ literal(10, DataTypes.INT())));
+
+ assertThatThrownBy(() -> nullUpper.accept(converter))
+ .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+ assertThatThrownBy(() -> nullLower.accept(converter))
+ .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+ assertThat(PredicateConverter.convert(rowType, nullUpper)).isEmpty();
+ assertThat(PredicateConverter.convert(rowType, nullLower)).isEmpty();
+ }
+
+ @MethodSource("provideNegatedComparisons")
+ @ParameterizedTest
+ public void testNegatedComparisons(
+ FunctionDefinition function, boolean literalOnLeft, Predicate
expected) {
+ PredicateConverter converter = new PredicateConverter(RowType.of(new
BigIntType()));
+ ResolvedExpression field = field(0, DataTypes.BIGINT());
+ ResolvedExpression literal = literal(1, DataTypes.INT());
+ CallExpression comparison =
+ literalOnLeft ? call(function, literal, field) :
call(function, field, literal);
+
+ assertThat(call(BuiltInFunctionDefinitions.NOT,
comparison).accept(converter))
+ .isEqualTo(expected);
+ }
+
+ public static Stream<Arguments> provideNegatedComparisons() {
+ PredicateBuilder builder = predicateBuilder(RowType.of(new
BigIntType()));
+ return Stream.of(
+ Arguments.of(BuiltInFunctionDefinitions.EQUALS, false,
builder.notEqual(0, 1L)),
+ Arguments.of(BuiltInFunctionDefinitions.NOT_EQUALS, false,
builder.equal(0, 1L)),
+ Arguments.of(
+ BuiltInFunctionDefinitions.GREATER_THAN, false,
builder.lessOrEqual(0, 1L)),
+ Arguments.of(
+ BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL,
+ false,
+ builder.lessThan(0, 1L)),
+ Arguments.of(
+ BuiltInFunctionDefinitions.LESS_THAN, false,
builder.greaterOrEqual(0, 1L)),
+ Arguments.of(
+ BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL,
+ false,
+ builder.greaterThan(0, 1L)),
+ Arguments.of(BuiltInFunctionDefinitions.EQUALS, true,
builder.notEqual(0, 1L)),
+ Arguments.of(BuiltInFunctionDefinitions.NOT_EQUALS, true,
builder.equal(0, 1L)),
+ Arguments.of(
+ BuiltInFunctionDefinitions.GREATER_THAN,
+ true,
+ builder.greaterOrEqual(0, 1L)),
+ Arguments.of(
+ BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL,
+ true,
+ builder.greaterThan(0, 1L)),
+ Arguments.of(
+ BuiltInFunctionDefinitions.LESS_THAN, true,
builder.lessOrEqual(0, 1L)),
+ Arguments.of(
+ BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL,
+ true,
+ builder.lessThan(0, 1L)));
+ }
+
+ @MethodSource("provideNegatedFloatingPointComparisons")
+ @ParameterizedTest
+ public void testNegatedFloatingPointComparisonsRemainUnsupported(
+ LogicalType fieldType,
+ DataType fieldDataType,
+ ResolvedExpression literal,
+ Object greater,
+ Object lesser,
+ Object nan) {
+ RowType rowType = RowType.of(fieldType);
+ PredicateConverter converter = new PredicateConverter(rowType);
+ ResolvedExpression field = field(0, fieldDataType);
+ for (FunctionDefinition function :
+ Arrays.asList(
+ BuiltInFunctionDefinitions.EQUALS,
+ BuiltInFunctionDefinitions.NOT_EQUALS,
+ BuiltInFunctionDefinitions.GREATER_THAN,
+ BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL,
+ BuiltInFunctionDefinitions.LESS_THAN,
+ BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL)) {
+ CallExpression negated =
+ call(BuiltInFunctionDefinitions.NOT, call(function, field,
literal));
+ CallExpression negatedLiteralOnLeft =
+ call(BuiltInFunctionDefinitions.NOT, call(function,
literal, field));
+ assertThatThrownBy(() -> negated.accept(converter))
+
.isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+ assertThatThrownBy(() -> negatedLiteralOnLeft.accept(converter))
+
.isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+ assertThat(PredicateConverter.convert(rowType, negated)).isEmpty();
+ assertThat(PredicateConverter.convert(rowType,
negatedLiteralOnLeft)).isEmpty();
+ }
+
+ ResolvedExpression nanLiteral = literal(nan, fieldDataType);
+ CallExpression notIn =
+ call(
+ BuiltInFunctionDefinitions.NOT,
+ call(BuiltInFunctionDefinitions.IN, field,
nanLiteral));
+ CallExpression notBetween =
+ call(
+ BuiltInFunctionDefinitions.NOT,
+ call(BuiltInFunctionDefinitions.BETWEEN, field,
literal, nanLiteral));
+ assertThatThrownBy(() -> notIn.accept(converter))
+ .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+ assertThatThrownBy(() -> notBetween.accept(converter))
+ .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+ assertThat(PredicateConverter.convert(rowType, notIn)).isEmpty();
+ assertThat(PredicateConverter.convert(rowType, notBetween)).isEmpty();
+
+ // Non-negated comparisons stay supported. Paimon orders NaN via
compareTo,
+ // so GreaterThan(NaN, 1.0) is true, unlike Flink's Java operators.
+ Predicate greaterThan =
+ call(BuiltInFunctionDefinitions.GREATER_THAN, field,
literal).accept(converter);
+ assertThat(greaterThan.test(GenericRow.of(greater))).isTrue();
+ assertThat(greaterThan.test(GenericRow.of(lesser))).isFalse();
+ assertThat(greaterThan.test(GenericRow.of(nan))).isTrue();
+ Predicate equal =
+ call(BuiltInFunctionDefinitions.EQUALS, field,
nanLiteral).accept(converter);
+ assertThat(equal.test(GenericRow.of(nan))).isTrue();
+ Predicate in = call(BuiltInFunctionDefinitions.IN, field,
nanLiteral).accept(converter);
+ assertThat(in.test(GenericRow.of(nan))).isTrue();
+ Predicate between =
+ call(BuiltInFunctionDefinitions.BETWEEN, field, literal,
nanLiteral)
+ .accept(converter);
+ assertThat(between.test(GenericRow.of(greater))).isTrue();
+ }
+
+ public static Stream<Arguments> provideNegatedFloatingPointComparisons() {
+ return Stream.of(
+ Arguments.of(
+ new DoubleType(),
+ DataTypes.DOUBLE(),
+ literal(1.0d, DataTypes.DOUBLE()),
+ 2.0d,
+ 0.5d,
+ Double.NaN),
+ Arguments.of(
+ new FloatType(),
+ DataTypes.FLOAT(),
+ literal(1.0f, DataTypes.FLOAT()),
+ 2.0f,
+ 0.5f,
+ Float.NaN));
+ }
+
+ @Test
+ public void
testNegatedFloatingPointComparisonInCompoundRemainsUnsupported() {
+ // Covers unsimplified NOT(AND(float_cmp, ...)) reaching the converter,
+ // which simple SQL NOT (d > 1.0) does not (Flink rewrites that to d
<= 1.0).
+ RowType rowType = RowType.of(new DoubleType(), new BigIntType());
+ PredicateConverter converter = new PredicateConverter(rowType);
+ CallExpression negated =
+ call(
+ BuiltInFunctionDefinitions.NOT,
+ call(
+ BuiltInFunctionDefinitions.AND,
+ call(
+
BuiltInFunctionDefinitions.GREATER_THAN,
+ field(0, DataTypes.DOUBLE()),
+ literal(1.0d, DataTypes.DOUBLE())),
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ field(1, DataTypes.BIGINT()),
+ literal(10L, DataTypes.BIGINT()))));
+
+ assertThatThrownBy(() -> negated.accept(converter))
+ .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+ assertThat(PredicateConverter.convert(rowType, negated)).isEmpty();
+ }
+
+ @Test
+ public void testNegatedFloatingPointSignedZerosRemainUnsupported() {
+ RowType rowType = RowType.of(new DoubleType());
+ PredicateConverter converter = new PredicateConverter(rowType);
+ ResolvedExpression field = field(0, DataTypes.DOUBLE());
+ ResolvedExpression zero = literal(0.0d, DataTypes.DOUBLE());
+ // Java: -0.0 == 0.0, so NOT (d <> 0.0) keeps -0.0. Double.compare
does not.
+ CallExpression negatedNotEquals =
+ call(
+ BuiltInFunctionDefinitions.NOT,
+ call(BuiltInFunctionDefinitions.NOT_EQUALS, field,
zero));
+ CallExpression negatedEquals =
+ call(
+ BuiltInFunctionDefinitions.NOT,
+ call(BuiltInFunctionDefinitions.EQUALS, field, zero));
+
+ assertThatThrownBy(() -> negatedNotEquals.accept(converter))
+ .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+ assertThatThrownBy(() -> negatedEquals.accept(converter))
+ .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+ assertThat(PredicateConverter.convert(rowType,
negatedNotEquals)).isEmpty();
+ assertThat(PredicateConverter.convert(rowType,
negatedEquals)).isEmpty();
+
+ // Non-negated equal stays on the pre-existing compareTo path.
+ Predicate equal = call(BuiltInFunctionDefinitions.EQUALS, field,
zero).accept(converter);
+ assertThat(equal.test(GenericRow.of(0.0d))).isTrue();
+ assertThat(equal.test(GenericRow.of(-0.0d))).isFalse();
+ }
+
+ @Test
+ public void testGenericNotComparisonAndDoubleNot() {
+ PredicateBuilder builder = predicateBuilder(RowType.of(new
BigIntType()));
+ PredicateConverter converter = new PredicateConverter(RowType.of(new
BigIntType()));
+ CallExpression equal =
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ field(0, DataTypes.BIGINT()),
+ literal(10, DataTypes.INT()));
+ CallExpression equalNull =
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ field(0, DataTypes.BIGINT()),
+ literal(null, DataTypes.BIGINT()));
+
+ Predicate notEqual = call(BuiltInFunctionDefinitions.NOT,
equal).accept(converter);
+ Predicate notEqualNull = call(BuiltInFunctionDefinitions.NOT,
equalNull).accept(converter);
+ Predicate doubleNot =
+ call(BuiltInFunctionDefinitions.NOT,
call(BuiltInFunctionDefinitions.NOT, equal))
+ .accept(converter);
+
+ assertThat(notEqual).isEqualTo(builder.notEqual(0, 10L));
+ assertThat(notEqualNull).isEqualTo(builder.notEqual(0, null));
+ assertThat(doubleNot).isEqualTo(builder.equal(0, 10L));
+ assertThat(notEqual.test(GenericRow.of(9L))).isTrue();
+ assertThat(notEqual.test(GenericRow.of(10L))).isFalse();
+ assertThat(notEqual.test(GenericRow.of((Object) null))).isFalse();
+ assertThat(notEqualNull.test(GenericRow.of(10L))).isFalse();
+ assertThat(notEqualNull.test(GenericRow.of((Object) null))).isFalse();
+ }
+
+ @Test
+ public void testGenericNotAndOr() {
+ PredicateBuilder builder = predicateBuilder(RowType.of(new
BigIntType()));
+ PredicateConverter converter = new PredicateConverter(RowType.of(new
BigIntType()));
+ CallExpression equal10 =
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ field(0, DataTypes.BIGINT()),
+ literal(10L, DataTypes.BIGINT()));
+ CallExpression equal20 =
+ call(
+ BuiltInFunctionDefinitions.EQUALS,
+ field(0, DataTypes.BIGINT()),
+ literal(20L, DataTypes.BIGINT()));
+
+ Predicate notAnd =
+ call(
+ BuiltInFunctionDefinitions.NOT,
+ call(BuiltInFunctionDefinitions.AND, equal10,
equal20))
+ .accept(converter);
+ Predicate notOr =
+ call(
+ BuiltInFunctionDefinitions.NOT,
+ call(BuiltInFunctionDefinitions.OR, equal10,
equal20))
+ .accept(converter);
+
+ assertThat(notAnd)
+ .isEqualTo(PredicateBuilder.or(builder.notEqual(0, 10L),
builder.notEqual(0, 20L)));
+ assertThat(notOr)
+ .isEqualTo(
+ PredicateBuilder.and(builder.notEqual(0, 10L),
builder.notEqual(0, 20L)));
+ assertThat(notAnd.test(GenericRow.of(10L))).isTrue();
+ assertThat(notAnd.test(GenericRow.of((Object) null))).isFalse();
+ assertThat(notOr.test(GenericRow.of(10L))).isFalse();
+ assertThat(notOr.test(GenericRow.of(15L))).isTrue();
+ assertThat(notOr.test(GenericRow.of((Object) null))).isFalse();
+ }
+
+ @Test
+ public void testBooleanTruthPredicatesAndNot() {
+ PredicateBuilder builder =
+
predicateBuilder(RowType.of(DataTypes.BOOLEAN().getLogicalType()));
+ PredicateConverter converter =
+ new
PredicateConverter(RowType.of(DataTypes.BOOLEAN().getLogicalType()));
+ ResolvedExpression boolField = field(0, DataTypes.BOOLEAN());
+ CallExpression isTrue = call(BuiltInFunctionDefinitions.IS_TRUE,
boolField);
+ CallExpression isFalse = call(BuiltInFunctionDefinitions.IS_FALSE,
boolField);
+ CallExpression isNotTrue =
call(BuiltInFunctionDefinitions.IS_NOT_TRUE, boolField);
+ CallExpression isNotFalse =
call(BuiltInFunctionDefinitions.IS_NOT_FALSE, boolField);
+
+ Predicate truePredicate = isTrue.accept(converter);
+ Predicate falsePredicate = isFalse.accept(converter);
+ Predicate notTruePredicate = isNotTrue.accept(converter);
+ Predicate notFalsePredicate = isNotFalse.accept(converter);
+
+ assertThat(truePredicate).isEqualTo(builder.equal(0, true));
+ assertThat(falsePredicate).isEqualTo(builder.equal(0, false));
+ assertBooleanResults(truePredicate, true, false, false);
+ assertBooleanResults(falsePredicate, false, true, false);
+ assertBooleanResults(notTruePredicate, false, true, true);
+ assertBooleanResults(notFalsePredicate, true, false, true);
+ assertBooleanResults(
+ call(BuiltInFunctionDefinitions.NOT,
isTrue).accept(converter), false, true, true);
+ assertBooleanResults(
+ call(BuiltInFunctionDefinitions.NOT,
isFalse).accept(converter), true, false, true);
+ assertBooleanResults(
+ call(BuiltInFunctionDefinitions.NOT,
isNotTrue).accept(converter),
+ true,
+ false,
+ false);
+ assertBooleanResults(
+ call(BuiltInFunctionDefinitions.NOT,
isNotFalse).accept(converter),
+ false,
+ true,
+ false);
+ }
+
+ @Test
+ public void testUnsupportedNotLike() {
+ RowType rowType = RowType.of(new VarCharType());
+ PredicateConverter converter = new PredicateConverter(RowType.of(new
VarCharType()));
+ CallExpression unsupportedLike =
+ call(
+ BuiltInFunctionDefinitions.LIKE,
+ field(0, STRING()),
+ literal("%middle%", STRING()));
+
+ assertThatThrownBy(
+ () ->
+ call(BuiltInFunctionDefinitions.NOT,
unsupportedLike)
+ .accept(converter))
+ .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+
+ CallExpression prefixLike =
+ call(
+ BuiltInFunctionDefinitions.LIKE,
+ field(0, STRING()),
+ literal("prefix%", STRING()));
+ CallExpression notPrefixLike = call(BuiltInFunctionDefinitions.NOT,
prefixLike);
+ assertThatThrownBy(() -> notPrefixLike.accept(converter))
+ .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+ assertThat(PredicateConverter.convert(rowType,
notPrefixLike)).isEmpty();
+ }
+
+ private static void assertBooleanResults(
+ Predicate predicate,
+ boolean expectedForTrue,
+ boolean expectedForFalse,
+ boolean expectedForNull) {
+
assertThat(predicate.test(GenericRow.of(true))).isEqualTo(expectedForTrue);
+
assertThat(predicate.test(GenericRow.of(false))).isEqualTo(expectedForFalse);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(expectedForNull);
+ }
+
+ private static PredicateBuilder predicateBuilder(RowType rowType) {
+ return new PredicateBuilder(LogicalTypeConversion.toDataType(rowType));
+ }
+
@MethodSource("provideLikeExpressions")
@ParameterizedTest
public void testStartsWith(
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
index e0b4172448..0b1d05ea60 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java
@@ -1127,6 +1127,98 @@ public class ReadWriteTableITCase extends
AbstractTestBase {
expected);
}
+ @Test
+ public void testNullablePredicateThreeValuedLogic() throws Exception {
+ String table =
+ createTable(
+ Arrays.asList("id INT", "v INT", "flag BOOLEAN"),
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ Collections.emptyList());
+
+ insertInto(
+ table,
+ "(1, CAST(NULL AS INT), CAST(NULL AS BOOLEAN))",
+ "(2, 1, TRUE)",
+ "(3, 2, FALSE)",
+ "(4, 3, CAST(NULL AS BOOLEAN))",
+ "(5, 4, TRUE)");
+
+ testBatchRead(
+ buildQuery(table, "id", "WHERE v BETWEEN 1 AND 3"),
+ Arrays.asList(changelogRow("+I", 2), changelogRow("+I", 3),
changelogRow("+I", 4)));
+ testBatchRead(
+ buildQuery(table, "id", "WHERE v NOT BETWEEN 1 AND 3"),
+ Collections.singletonList(changelogRow("+I", 5)));
+ testBatchRead(
+ buildQuery(table, "id", "WHERE v IN (1, 3)"),
+ Arrays.asList(changelogRow("+I", 2), changelogRow("+I", 4)));
+ testBatchRead(
+ buildQuery(table, "id", "WHERE v NOT IN (1, 3)"),
+ Arrays.asList(changelogRow("+I", 3), changelogRow("+I", 5)));
+ testBatchRead(
+ buildQuery(table, "id", "WHERE v IN (1, NULL, 3)"),
+ Arrays.asList(changelogRow("+I", 2), changelogRow("+I", 4)));
+ testBatchRead(
+ buildQuery(table, "id", "WHERE v NOT IN (1, NULL, 3)"),
Collections.emptyList());
+ testBatchRead(
+ buildQuery(table, "id", "WHERE flag IS TRUE"),
+ Arrays.asList(changelogRow("+I", 2), changelogRow("+I", 5)));
+ testBatchRead(
+ buildQuery(table, "id", "WHERE flag IS NOT TRUE"),
+ Arrays.asList(changelogRow("+I", 1), changelogRow("+I", 3),
changelogRow("+I", 4)));
+ testBatchRead(
+ buildQuery(table, "id", "WHERE NOT (flag IS TRUE)"),
+ Arrays.asList(changelogRow("+I", 1), changelogRow("+I", 3),
changelogRow("+I", 4)));
+ }
+
+ @Test
+ public void testNotBetweenWithNullBoundsThreeValuedLogic() throws
Exception {
+ String table =
+ createTable(
+ Arrays.asList("id INT", "v INT"),
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ Collections.emptyList());
+
+ insertInto(table, "(1, 12)", "(2, 16)", "(3, 8)", "(4, CAST(NULL AS
INT))");
+
+ // 12 NOT BETWEEN 15 AND NULL = 12 < 15 OR 12 > NULL = TRUE
+ // 16 NOT BETWEEN 15 AND NULL = UNKNOWN (dropped by WHERE)
+ testBatchRead(
+ buildQuery(table, "id", "WHERE v NOT BETWEEN 15 AND NULL"),
+ Arrays.asList(changelogRow("+I", 1), changelogRow("+I", 3)));
+
+ // 12 NOT BETWEEN NULL AND 10 = 12 < NULL OR 12 > 10 = TRUE
+ // 8 NOT BETWEEN NULL AND 10 = UNKNOWN
+ testBatchRead(
+ buildQuery(table, "id", "WHERE v NOT BETWEEN NULL AND 10"),
+ Arrays.asList(changelogRow("+I", 1), changelogRow("+I", 2)));
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"file-index.bsi.columns",
"file-index.range-bitmap.columns"})
+ public void testNotInNullWithFileIndex(String indexOption) throws
Exception {
+ Map<String, String> options = new HashMap<>();
+ options.put(indexOption, "v");
+ options.put("file-index.in-manifest-threshold", "1B");
+ String table =
+ createTable(
+ Arrays.asList("id INT", "v INT"),
+ Collections.emptyList(),
+ Collections.singletonList("id"),
+ Collections.emptyList(),
+ options);
+
+ insertInto(table, "(1, CAST(NULL AS INT))", "(2, 1)", "(3, 2)", "(4,
3)", "(5, 4)");
+
+ testBatchRead(
+ buildQuery(table, "id", "WHERE v NOT IN (1, NULL, 3)"),
Collections.emptyList());
+ testBatchRead(
+ buildQuery(table, "id", "WHERE v NOT IN (1, 3)"),
+ Arrays.asList(changelogRow("+I", 3), changelogRow("+I", 5)));
+ }
+
@Test
public void testUnsupportedPredicate() throws Exception {
String table =
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkTableSourceTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkTableSourceTest.java
index 74183a40e2..b8e236ae60 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkTableSourceTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkTableSourceTest.java
@@ -23,6 +23,7 @@ import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.predicate.CompoundPredicate;
import org.apache.paimon.predicate.Or;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.FileSystemSchemaManager;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.TableSchema;
@@ -141,6 +142,119 @@ public class FlinkTableSourceTest extends TableTestBase {
.isEqualTo(ImmutableList.of(filters.get(1)));
}
+ @Test
+ public void testApplyNotFilters() throws Exception {
+ FileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(String.format("%s/%s.db/%s", warehouse,
database, "T"));
+ Schema schema =
+ Schema.newBuilder()
+ .column("col1", DataTypes.INT())
+ .column("col2", DataTypes.INT())
+ .column("p1", DataTypes.INT())
+ .column("p2", DataTypes.STRING())
+ .partitionKeys("p1", "p2")
+ .build();
+ TableSchema tableSchema =
+ new FileSystemSchemaManager(fileIO,
tablePath).createTable(schema);
+ Table table = FileStoreTableFactory.create(LocalFileIO.create(),
tablePath, tableSchema);
+ PredicateBuilder builder = new PredicateBuilder(table.rowType());
+
+ ResolvedExpression supportedPartitionFilter = not(p1Equal1());
+ FlinkTableSource tableSource = dataTableSource(table);
+ Assertions.assertThat(
+ tableSource
+
.applyFilters(ImmutableList.of(supportedPartitionFilter))
+ .getRemainingFilters())
+ .isEmpty();
+
Assertions.assertThat(tableSource.predicate).isEqualTo(builder.notEqual(2, 1));
+
+ ResolvedExpression supportedNonPartitionFilter = not(col1Equal1());
+ tableSource = dataTableSource(table);
+ Assertions.assertThat(
+ tableSource
+
.applyFilters(ImmutableList.of(supportedNonPartitionFilter))
+ .getRemainingFilters())
+ .containsExactly(supportedNonPartitionFilter);
+
Assertions.assertThat(tableSource.predicate).isEqualTo(builder.notEqual(0, 1));
+
+ ResolvedExpression unsupportedLike = not(p2Like("%a"));
+ tableSource = dataTableSource(table);
+ Assertions.assertThat(
+ tableSource
+
.applyFilters(ImmutableList.of(unsupportedLike))
+ .getRemainingFilters())
+ .containsExactly(unsupportedLike);
+ Assertions.assertThat(tableSource.predicate).isNull();
+
+ ResolvedExpression unsupportedPrefixLike = not(p2Like("prefix%"));
+ tableSource = dataTableSource(table);
+ Assertions.assertThat(
+ tableSource
+
.applyFilters(ImmutableList.of(unsupportedPrefixLike))
+ .getRemainingFilters())
+ .containsExactly(unsupportedPrefixLike);
+ Assertions.assertThat(tableSource.predicate).isNull();
+
+ ResolvedExpression unsupportedSimilar = not(p2Similar("a.*"));
+ tableSource = dataTableSource(table);
+ Assertions.assertThat(
+ tableSource
+
.applyFilters(ImmutableList.of(unsupportedSimilar))
+ .getRemainingFilters())
+ .containsExactly(unsupportedSimilar);
+ Assertions.assertThat(tableSource.predicate).isNull();
+ }
+
+ @Test
+ public void testApplyNegatedFloatingPointComparison() throws Exception {
+ // Simple SQL such as NOT (d = NaN), d NOT IN (NaN) and NOT (d <> 0.0)
+ // is normalized by Flink to <>(d, NaN) or =(d, 0.0) before
applyFilters,
+ // so those queries never hit this residual. CAST(-0.0 AS DOUBLE) is
also
+ // +0.0 in Flink. Coverage is the unsimplified AST below.
+ FileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(String.format("%s/%s.db/%s", warehouse,
database, "T"));
+ Schema schema = Schema.newBuilder().column("d",
DataTypes.DOUBLE()).build();
+ TableSchema tableSchema =
+ new FileSystemSchemaManager(fileIO,
tablePath).createTable(schema);
+ Table table = FileStoreTableFactory.create(LocalFileIO.create(),
tablePath, tableSchema);
+
+ assertResidual(table, not(doubleGreaterThan1()));
+ assertResidual(table, not(doubleEqual(1.0d)));
+ assertResidual(table, not(doubleNotEqual(0.0d)));
+ assertResidual(table, not(doubleIn(Double.NaN)));
+ assertResidual(table, not(doubleBetween(1.0d, Double.NaN)));
+ }
+
+ @Test
+ public void testApplyNotBetweenWithNullBounds() throws Exception {
+ FileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(String.format("%s/%s.db/%s", warehouse,
database, "T"));
+ Schema schema = Schema.newBuilder().column("col1",
DataTypes.INT()).build();
+ TableSchema tableSchema =
+ new FileSystemSchemaManager(fileIO,
tablePath).createTable(schema);
+ Table table = FileStoreTableFactory.create(LocalFileIO.create(),
tablePath, tableSchema);
+
+ assertResidual(table, not(col1Between(15, null)));
+ assertResidual(table, not(col1Between(null, 10)));
+ }
+
+ @Test
+ public void testApplyNotInWithNull() throws Exception {
+ FileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(String.format("%s/%s.db/%s", warehouse,
database, "T"));
+ Schema schema = Schema.newBuilder().column("col1",
DataTypes.INT()).build();
+ TableSchema tableSchema =
+ new FileSystemSchemaManager(fileIO,
tablePath).createTable(schema);
+ Table table = FileStoreTableFactory.create(LocalFileIO.create(),
tablePath, tableSchema);
+
+ ResolvedExpression notInNull = not(col1InWithNull());
+ FlinkTableSource tableSource = dataTableSource(table);
+ Assertions.assertThat(
+
tableSource.applyFilters(ImmutableList.of(notInNull)).getRemainingFilters())
+ .containsExactly(notInNull);
+
Assertions.assertThat(tableSource.predicate).isEqualTo(PredicateBuilder.alwaysFalse());
+ }
+
// ==================== Nested OR Tree Tests ====================
//
// These tests construct OR trees in various shapes — mimicking what
Flink's
@@ -335,6 +449,97 @@ public class FlinkTableSourceTest extends TableTestBase {
org.apache.flink.table.api.DataTypes.BOOLEAN());
}
+ private ResolvedExpression col1InWithNull() {
+ return CallExpression.anonymous(
+ BuiltInFunctionDefinitions.IN,
+ ImmutableList.of(
+ new FieldReferenceExpression(
+ "col1",
org.apache.flink.table.api.DataTypes.INT(), 0, 0),
+ new ValueLiteralExpression(
+ 1,
org.apache.flink.table.api.DataTypes.INT().notNull()),
+ new ValueLiteralExpression(
+ null,
org.apache.flink.table.api.DataTypes.INT()),
+ new ValueLiteralExpression(
+ 3,
org.apache.flink.table.api.DataTypes.INT().notNull())),
+ org.apache.flink.table.api.DataTypes.BOOLEAN());
+ }
+
+ private ResolvedExpression doubleGreaterThan1() {
+ return CallExpression.anonymous(
+ BuiltInFunctionDefinitions.GREATER_THAN,
+ ImmutableList.of(
+ new FieldReferenceExpression(
+ "d",
org.apache.flink.table.api.DataTypes.DOUBLE(), 0, 0),
+ new ValueLiteralExpression(
+ 1.0d,
org.apache.flink.table.api.DataTypes.DOUBLE().notNull())),
+ org.apache.flink.table.api.DataTypes.BOOLEAN());
+ }
+
+ private ResolvedExpression doubleEqual(double value) {
+ return CallExpression.anonymous(
+ BuiltInFunctionDefinitions.EQUALS,
+ ImmutableList.of(
+ new FieldReferenceExpression(
+ "d",
org.apache.flink.table.api.DataTypes.DOUBLE(), 0, 0),
+ new ValueLiteralExpression(
+ value,
org.apache.flink.table.api.DataTypes.DOUBLE().notNull())),
+ org.apache.flink.table.api.DataTypes.BOOLEAN());
+ }
+
+ private ResolvedExpression doubleNotEqual(double value) {
+ return CallExpression.anonymous(
+ BuiltInFunctionDefinitions.NOT_EQUALS,
+ ImmutableList.of(
+ new FieldReferenceExpression(
+ "d",
org.apache.flink.table.api.DataTypes.DOUBLE(), 0, 0),
+ new ValueLiteralExpression(
+ value,
org.apache.flink.table.api.DataTypes.DOUBLE().notNull())),
+ org.apache.flink.table.api.DataTypes.BOOLEAN());
+ }
+
+ private ResolvedExpression doubleIn(double value) {
+ return CallExpression.anonymous(
+ BuiltInFunctionDefinitions.IN,
+ ImmutableList.of(
+ new FieldReferenceExpression(
+ "d",
org.apache.flink.table.api.DataTypes.DOUBLE(), 0, 0),
+ new ValueLiteralExpression(
+ value,
org.apache.flink.table.api.DataTypes.DOUBLE().notNull())),
+ org.apache.flink.table.api.DataTypes.BOOLEAN());
+ }
+
+ private ResolvedExpression doubleBetween(double lower, double upper) {
+ return CallExpression.anonymous(
+ BuiltInFunctionDefinitions.BETWEEN,
+ ImmutableList.of(
+ new FieldReferenceExpression(
+ "d",
org.apache.flink.table.api.DataTypes.DOUBLE(), 0, 0),
+ new ValueLiteralExpression(
+ lower,
org.apache.flink.table.api.DataTypes.DOUBLE().notNull()),
+ new ValueLiteralExpression(
+ upper,
org.apache.flink.table.api.DataTypes.DOUBLE().notNull())),
+ org.apache.flink.table.api.DataTypes.BOOLEAN());
+ }
+
+ private ResolvedExpression col1Between(Integer lower, Integer upper) {
+ return CallExpression.anonymous(
+ BuiltInFunctionDefinitions.BETWEEN,
+ ImmutableList.of(
+ new FieldReferenceExpression(
+ "col1",
org.apache.flink.table.api.DataTypes.INT(), 0, 0),
+ intLiteral(lower),
+ intLiteral(upper)),
+ org.apache.flink.table.api.DataTypes.BOOLEAN());
+ }
+
+ private ValueLiteralExpression intLiteral(Integer value) {
+ if (value == null) {
+ return new ValueLiteralExpression(null,
org.apache.flink.table.api.DataTypes.INT());
+ }
+ return new ValueLiteralExpression(
+ value, org.apache.flink.table.api.DataTypes.INT().notNull());
+ }
+
private ResolvedExpression p1Equal1() {
return CallExpression.anonymous(
BuiltInFunctionDefinitions.EQUALS,
@@ -357,6 +562,17 @@ public class FlinkTableSourceTest extends TableTestBase {
org.apache.flink.table.api.DataTypes.BOOLEAN());
}
+ private ResolvedExpression p2Similar(String literal) {
+ return CallExpression.anonymous(
+ BuiltInFunctionDefinitions.SIMILAR,
+ ImmutableList.of(
+ new FieldReferenceExpression(
+ "p2",
org.apache.flink.table.api.DataTypes.STRING(), 0, 3),
+ new ValueLiteralExpression(
+ literal,
org.apache.flink.table.api.DataTypes.STRING().notNull())),
+ org.apache.flink.table.api.DataTypes.BOOLEAN());
+ }
+
// where rand(42) > 0.1
private ResolvedExpression rand() {
return CallExpression.anonymous(
@@ -406,4 +622,23 @@ public class FlinkTableSourceTest extends TableTestBase {
ImmutableList.of(e1, e2),
org.apache.flink.table.api.DataTypes.BOOLEAN());
}
+
+ private ResolvedExpression not(ResolvedExpression expression) {
+ return CallExpression.anonymous(
+ BuiltInFunctionDefinitions.NOT,
+ ImmutableList.of(expression),
+ org.apache.flink.table.api.DataTypes.BOOLEAN());
+ }
+
+ private void assertResidual(Table table, ResolvedExpression filter) {
+ FlinkTableSource tableSource = dataTableSource(table);
+ Assertions.assertThat(
+
tableSource.applyFilters(ImmutableList.of(filter)).getRemainingFilters())
+ .containsExactly(filter);
+ Assertions.assertThat(tableSource.predicate).isNull();
+ }
+
+ private DataTableSource dataTableSource(Table table) {
+ return new DataTableSource(ObjectIdentifier.of("catalog1", "db1",
"T"), table, false, null);
+ }
}
