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 db3754f86ded..8e41c27c5ff1 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.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,93 @@ public PredicateConverter(PredicateBuilder builder) { @Override public Predicate visit(CallExpression call) { + return visit(call, false); + } + + private Predicate visit(CallExpression call, boolean negated) { FunctionDefinition func = call.getFunctionDefinition(); List children = call.getChildren(); if (func == BuiltInFunctionDefinitions.AND) { - return PredicateBuilder.and(flattenAndConvert(children, func)); + requireAtLeastArity(children, 2); + List 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 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 negated + ? visitBiFunction(children, builder::notEqual, builder::notEqual) + : visitBiFunction(children, builder::equal, builder::equal); } else if (func == BuiltInFunctionDefinitions.NOT_EQUALS) { - return visitBiFunction(children, builder::notEqual, builder::notEqual); + return negated + ? visitBiFunction(children, builder::equal, builder::equal) + : visitBiFunction(children, builder::notEqual, builder::notEqual); } else if (func == BuiltInFunctionDefinitions.GREATER_THAN) { - return visitBiFunction(children, builder::greaterThan, builder::lessThan); + return negated + ? visitBiFunction(children, builder::lessOrEqual, builder::greaterOrEqual) + : visitBiFunction(children, builder::greaterThan, builder::lessThan); } else if (func == BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL) { - return visitBiFunction(children, builder::greaterOrEqual, builder::lessOrEqual); + return negated + ? visitBiFunction(children, builder::lessThan, builder::greaterThan) + : visitBiFunction(children, builder::greaterOrEqual, builder::lessOrEqual); } else if (func == BuiltInFunctionDefinitions.LESS_THAN) { - return visitBiFunction(children, builder::lessThan, builder::greaterThan); + return negated + ? visitBiFunction(children, builder::greaterOrEqual, builder::lessOrEqual) + : visitBiFunction(children, builder::lessThan, builder::greaterThan); } else if (func == BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL) { - return visitBiFunction(children, builder::lessOrEqual, builder::greaterOrEqual); + return negated + ? visitBiFunction(children, builder::greaterThan, builder::lessThan) + : visitBiFunction(children, 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 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))); } - return builder.in(builder.indexOf(fieldRefExpr.getName()), literals); + return negated + ? builder.notIn(field.index, literals) + : 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)); + Object lower = extractLiteral(field.expression.getOutputDataType(), children.get(1)); + Object upper = extractLiteral(field.expression.getOutputDataType(), children.get(2)); + Predicate between = builder.between(field.index, lower, upper); + return negated ? negate(between) : 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 +204,57 @@ public Predicate visit(CallExpression call) { 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 +262,11 @@ public Predicate visit(CallExpression call) { * * @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 flattenAndConvert( - List children, FunctionDefinition targetFunc) { + List children, FunctionDefinition targetFunc, boolean negated) { List result = new ArrayList<>(); Deque stack = new ArrayDeque<>(); for (int i = children.size() - 1; i >= 0; i--) { @@ -228,14 +278,15 @@ private List flattenAndConvert( CallExpression ce = (CallExpression) expr; if (ce.getFunctionDefinition() == targetFunc) { List 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; @@ -245,23 +296,52 @@ private Predicate visitBiFunction( List children, BiFunction visit1, BiFunction visit2) { + requireArity(children, 2); Optional 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 children, int expected) { + if (children.size() != expected) { + throw new UnsupportedExpression(); + } + } + + private void requireAtLeastArity(List children, int minimum) { + if (children.size() < minimum) { + throw new UnsupportedExpression(); + } + } + private Optional extractFieldReference(Expression expression) { if (expression instanceof FieldReferenceExpression) { return Optional.of((FieldReferenceExpression) expression); @@ -304,6 +384,14 @@ private Object extractLiteral(DataType expectedType, Expression expression) { 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 +419,17 @@ private boolean supportsPredicate(LogicalType type) { } } + 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 e05773991ffb..21860be9dde5 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.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; @@ -46,6 +49,7 @@ 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 +83,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 +261,7 @@ public static Stream provideResolvedExpression() { 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 +317,281 @@ public void testBetweenWithNullBounds() { 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(builder.notIn(0, Arrays.asList(1L, null, 3L))); + 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 children = new ArrayList<>(); + List 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); + } + + @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 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))); + } + + @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 940dd95af0b0..c0b2f3f8a351 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 @@ -1125,6 +1125,51 @@ public void testIn() throws Exception { 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 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 ba7dbd08938e..e672f0e659bb 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.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.Schema; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.schema.TableSchema; @@ -139,6 +140,68 @@ public void testApplyPartitionTable() throws Exception { .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 SchemaManager(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(); + } + // ==================== Nested OR Tree Tests ==================== // // These tests construct OR trees in various shapes — mimicking what Flink's @@ -354,6 +417,17 @@ private ResolvedExpression p2Like(String literal) { 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( @@ -403,4 +477,15 @@ private ResolvedExpression and(ResolvedExpression e1, ResolvedExpression e2) { 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 DataTableSource dataTableSource(Table table) { + return new DataTableSource(ObjectIdentifier.of("catalog1", "db1", "T"), table, false, null); + } }