[flink] Add missing predicat converters in PredicateConverter

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PredicateConverter.java

@@ -194,10 +194,56 @@ public Predicate visit(CallExpression call) {
             FieldReferenceExpression fieldRefExpr =
                     extractFieldReference(children.get(0)).orElseThrow(UnsupportedExpression::new);
             return builder.equal(builder.indexOf(fieldRefExpr.getName()), Boolean.FALSE);
+        } else if (func == BuiltInFunctionDefinitions.NOT) {
+            // NOT predicate - negate the inner predicate
+            Predicate innerPredicate = children.get(0).accept(this);
+            return innerPredicate.negate().orElseThrow(UnsupportedExpression::new);
+        } else if (func == BuiltInFunctionDefinitions.IS_NOT_TRUE) {
+            FieldReferenceExpression fieldRefExpr =
+                    extractFieldReference(children.get(0)).orElseThrow(UnsupportedExpression::new);
+            int fieldIndex = builder.indexOf(fieldRefExpr.getName());
+            // "x IS NOT TRUE" is true for both FALSE and NULL, unlike NotEqual which returns
+            // false when the field value is null.
+            return PredicateBuilder.or(
+                    builder.isNull(fieldIndex), builder.equal(fieldIndex, Boolean.FALSE));
+        } else if (func == BuiltInFunctionDefinitions.NOT_BETWEEN) {
+            FieldReferenceExpression fieldRefExpr =
+                    extractFieldReference(children.get(0)).orElseThrow(UnsupportedExpression::new);
+            return builder.between(
+                            builder.indexOf(fieldRefExpr.getName()),
+                            children.get(1),
+                            children.get(2))
+                    .negate()
+                    .orElseThrow(UnsupportedExpression::new);
+        } else if (func == BuiltInFunctionDefinitions.SIMILAR) {
+            FieldReferenceExpression fieldRefExpr =
+                    extractFieldReference(children.get(0)).orElseThrow(UnsupportedExpression::new);
+            if (fieldRefExpr
+                    .getOutputDataType()
+                    .getLogicalType()
+                    .getTypeRoot()
+                    .getFamilies()
+                    .contains(LogicalTypeFamily.CHARACTER_STRING)) {
+                String sqlPattern =
+                        Objects.requireNonNull(
+                                        extractLiteral(
+                                                fieldRefExpr.getOutputDataType(), children.get(1)))
+                                .toString();
+                String escape =
+                        children.size() <= 2
+                                ? null
+                                : Objects.requireNonNull(
+                                                extractLiteral(
+                                                        fieldRefExpr.getOutputDataType(),
+                                                        children.get(2)))
+                                        .toString();
+                String likePattern = convertSimilarToLike(sqlPattern, escape);
+                return builder.like(
+                        builder.indexOf(fieldRefExpr.getName()),
+                        BinaryString.fromString(likePattern));
+            }
         }
 
-        // TODO is_xxx, between_xxx, similar, in, not_in, not?
-
         throw new UnsupportedExpression();
     }
 
@@ -291,6 +337,87 @@ private boolean supportsPredicate(LogicalType type) {
         }
     }
 
+    /**
+     * Converts a SQL SIMILAR TO pattern to an equivalent SQL LIKE pattern so that it can be
+     * evaluated by {@link PredicateBuilder#like}.
+     *
+     * <p>The conversion handles only the subset of SIMILAR TO syntax that maps directly to SQL
+     * LIKE:
+     *
+     * <ul>
+     *   <li>{@code %} (any-string wildcard) is preserved as-is.
+     *   <li>{@code _} (single-character wildcard) is preserved as-is.
+     *   <li>Escape sequences: {@code escape + '_'} and {@code escape + '%'} become their literal
+     *       equivalents, emitted as {@code \ + char} so that the downstream {@link Like} function
+     *       (which uses {@code \} as its default escape) treats them as literals. {@code escape +
+     *       escape} becomes a literal escape character.
+     * </ul>
+     *
+     * <p>SIMILAR TO-only features (character classes {@code [...]}, alternation {@code |},
+     * quantifiers {@code *}, {@code +}, {@code ?}, {@code {m,n}}, and grouping {@code ()}) are not
+     * supported and will cause an {@link UnsupportedExpression} to be thrown.
+     *
+     * @param sqlPattern the SIMILAR TO pattern string
+     * @param escape the escape character string (single char), or {@code null} for no escaping
+     * @return an equivalent SQL LIKE pattern (using {@code \} as the escape character)
+     * @throws UnsupportedExpression if the pattern uses SIMILAR TO-only features
+     */
+    private String convertSimilarToLike(String sqlPattern, String escape) {
+        if (sqlPattern == null || sqlPattern.isEmpty()) {
+            return sqlPattern;
+        }
+
+        // Sentinel 0 means no escape character is defined.
+        char escapeChar = (escape != null && !escape.isEmpty()) ? escape.charAt(0) : 0;
+        // The output LIKE pattern will always use '\' as its escape char, because that is what
+        // Like.sqlToRegexLike uses by default.
+        final char outputEscape = '\\';
+
+        StringBuilder like = new StringBuilder();
+
+        for (int i = 0; i < sqlPattern.length(); i++) {
+            char c = sqlPattern.charAt(i);
+
+            if (escapeChar != 0 && c == escapeChar) {
+                // Escape sequence
+                if (i + 1 >= sqlPattern.length()) {
+                    throw new UnsupportedExpression();
+                }
+                char next = sqlPattern.charAt(i + 1);
+                if (next == '_' || next == '%') {
+                    // Escaped wildcard -> literal in the LIKE output.
+                    // Emit as outputEscape + wildcard so Like treats it as a literal.
+                    like.append(outputEscape).append(next);
+                } else if (next == escapeChar) {
+                    // Escaped escape char -> emit as a literal character.
+                    // If the escape char itself is special to LIKE (% or _), escape it; otherwise
+                    // emit it as-is since Like only special-cases % and _.
+                    like.append(escapeChar);
+                } else {
+                    // Unknown escape sequence - not supported
+                    throw new UnsupportedExpression();
+                }
+                i++;
+            } else if (c == '%' || c == '_') {
+                // SIMILAR TO wildcards are the same as SQL LIKE wildcards
+                like.append(c);
+            } else if (c == '[' || c == '|' || c == '(' || c == ')' || c == '*' || c == '+'
+                    || c == '?' || c == '{' || c == '}') {
+                // SIMILAR TO-only features (including {m,n} quantifiers): not representable in
+                // SQL LIKE
+                throw new UnsupportedExpression();
+            } else if (c == outputEscape) {
+                // A literal backslash in the pattern needs to be escaped in the output,
+                // since the output escape char is '\'.
+                like.append(outputEscape).append(outputEscape);
+            } else {
+                like.append(c);
+            }
+        }
+
+        return like.toString();
+    }
+
     @Override
     public Predicate visit(ValueLiteralExpression valueLiteralExpression) {
         throw new UnsupportedExpression();

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PredicateConverterTest.java

@@ -34,6 +34,7 @@
 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.BooleanType;
 import org.apache.flink.table.types.logical.DoubleType;
 import org.apache.flink.table.types.logical.IntType;
 import org.apache.flink.table.types.logical.RowType;
@@ -268,7 +269,13 @@ public static Stream<Arguments> provideResolvedExpression() {
                                 BuiltInFunctionDefinitions.IS_FALSE,
                                 Arrays.asList(boolRefExpr),
                                 DataTypes.BOOLEAN()),
-                        BUILDER.equal(3, false)));
+                        BUILDER.equal(3, false)),
+                Arguments.of(
+                        CallExpression.permanent(
+                                BuiltInFunctionDefinitions.IS_NOT_TRUE,
+                                Arrays.asList(boolRefExpr),
+                                DataTypes.BOOLEAN()),
+                        PredicateBuilder.or(BUILDER.isNull(3), BUILDER.equal(3, false))));
     }
 
     @MethodSource("provideLikeExpressions")
@@ -438,6 +445,19 @@ public static Stream<Arguments> provideLikeExpressions() {
                         expectedForStats4));
     }
 
+    @Test
+    public void testIsNotTrueNullSemantics() {
+        // "x IS NOT TRUE" must be true for both FALSE and NULL, matching SQL's three-valued
+        // logic, unlike a plain NotEqual(x, TRUE) which returns false for NULL rows.
+        CallExpression expr =
+                call(BuiltInFunctionDefinitions.IS_NOT_TRUE, field(0, DataTypes.BOOLEAN()));
+        Predicate predicate = expr.accept(new PredicateConverter(RowType.of(new BooleanType())));
+
+        assertThat(predicate.test(GenericRow.of(true))).isFalse();
+        assertThat(predicate.test(GenericRow.of(false))).isTrue();
+        assertThat(predicate.test(GenericRow.of((Object) null))).isTrue();
+    }
+
     @Test
     public void testUnsupportedExpression() {
         CallExpression expression =
@@ -808,6 +828,189 @@ public void testUnsupportedFieldReferenceExpression() {
                 .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
     }
 
+    // -------------------------------------------------------------------------
+    // Tests for convertSimilarToLike
+    // -------------------------------------------------------------------------
+
+    @Test
+    public void testConvertSimilarToLike() {
+        // '%' stays as '%' (SQL LIKE wildcard)
+        assertThat(convertSimilarToLike("abc%", null)).isEqualTo("abc%");
+        // '_' stays as '_' (SQL LIKE wildcard)
+        assertThat(convertSimilarToLike("a_c", null)).isEqualTo("a_c");
+        // Literal characters pass through unchanged
+        assertThat(convertSimilarToLike("hello", null)).isEqualTo("hello");
+        // '.' is just a literal dot in SIMILAR TO — kept as-is in the LIKE output
+        assertThat(convertSimilarToLike("3.14", null)).isEqualTo("3.14");
+        // Backslash in pattern is doubled so downstream Like sees it as a literal '\'
+        assertThat(convertSimilarToLike("a\\b", null)).isEqualTo("a\\\\b");
+        // Empty pattern returns empty
+        assertThat(convertSimilarToLike("", null)).isEqualTo("");
+
+        // --- SIMILAR TO-only features must throw UnsupportedExpression ---
+        assertThatThrownBy(() -> convertSimilarToLike("a|b", null))
+                .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+        assertThatThrownBy(() -> convertSimilarToLike("(a|b)%", null))
+                .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+        assertThatThrownBy(() -> convertSimilarToLike("a+", null))
+                .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+        assertThatThrownBy(() -> convertSimilarToLike("[a-z]", null))
+                .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+        // '{m,n}' quantifiers are SIMILAR TO-only syntax (e.g. 'a{2}' means 'aa') and must not
+        // pass through as literal LIKE characters.
+        assertThatThrownBy(() -> convertSimilarToLike("a{2}", null))
+                .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+        assertThatThrownBy(() -> convertSimilarToLike("a{2,3}", null))
+                .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+
+        // --- Escape-character tests (using '=' as the escape char) ---
+        // '=_' -> escaped underscore: output '\_' so Like treats '_' as literal
+        assertThat(convertSimilarToLike("a=_b", "=")).isEqualTo("a\\_b");
+        // '=%' -> escaped percent: output '\%' so Like treats '%' as literal
+        assertThat(convertSimilarToLike("a=%b", "=")).isEqualTo("a\\%b");
+        // '==' -> literal '=' (the escape char)
+        assertThat(convertSimilarToLike("a==b", "=")).isEqualTo("a=b");
+        // Trailing escape char alone must throw
+        assertThatThrownBy(() -> convertSimilarToLike("a=", "="))
+                .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+        // Unknown escape sequence (e.g. '=a') must throw
+        assertThatThrownBy(() -> convertSimilarToLike("a=b", "="))
+                .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+    }
+
+    /** Invokes the private {@code convertSimilarToLike} via reflection. */
+    private static String convertSimilarToLike(String sqlPattern, String escape) {
+        try {
+            java.lang.reflect.Method m =
+                    PredicateConverter.class.getDeclaredMethod(
+                            "convertSimilarToLike", String.class, String.class);
+            m.setAccessible(true);
+            return (String) m.invoke(new PredicateConverter(BUILDER), sqlPattern, escape);
+        } catch (java.lang.reflect.InvocationTargetException e) {
+            Throwable cause = e.getCause();
+            if (cause instanceof RuntimeException) {
+                throw (RuntimeException) cause;
+            }
+            throw new RuntimeException(cause);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    // -------------------------------------------------------------------------
+    // Tests for the SIMILAR (SIMILAR TO) branch in visit(CallExpression)
+    // -------------------------------------------------------------------------
+
+    @Test
+    public void testSimilarExpressionBasic() {
+        // 'abc%' SIMILAR TO: any string starting with 'abc'
+        PredicateConverter converter = new PredicateConverter(RowType.of(new VarCharType()));
+        CallExpression expr =
+                call(
+                        BuiltInFunctionDefinitions.SIMILAR,
+                        field(0, STRING()),
+                        literal("abc%", STRING()));
+        Predicate predicate = expr.accept(converter);
+
+        assertThat(predicate.test(GenericRow.of(BinaryString.fromString("abc")))).isTrue();
+        assertThat(predicate.test(GenericRow.of(BinaryString.fromString("abcdef")))).isTrue();
+        assertThat(predicate.test(GenericRow.of(BinaryString.fromString("ab")))).isFalse();
+        assertThat(predicate.test(GenericRow.of(BinaryString.fromString("ABC")))).isFalse();
+        assertThat(predicate.test(GenericRow.of((Object) null))).isFalse();
+    }
+
+    @Test
+    public void testSimilarExpressionUnderscore() {
+        // 'a_c' SIMILAR TO: exactly 3 chars starting with 'a', ending with 'c'
+        PredicateConverter converter = new PredicateConverter(RowType.of(new VarCharType()));
+        CallExpression expr =
+                call(
+                        BuiltInFunctionDefinitions.SIMILAR,
+                        field(0, STRING()),
+                        literal("a_c", STRING()));
+        Predicate predicate = expr.accept(converter);
+
+        assertThat(predicate.test(GenericRow.of(BinaryString.fromString("abc")))).isTrue();
+        assertThat(predicate.test(GenericRow.of(BinaryString.fromString("axc")))).isTrue();
+        assertThat(predicate.test(GenericRow.of(BinaryString.fromString("ac")))).isFalse();
+        assertThat(predicate.test(GenericRow.of(BinaryString.fromString("abbc")))).isFalse();
+    }
+
+    @Test
+    public void testSimilarExpressionAlternation() {
+        // '(cat|dog)%' uses SIMILAR TO-only syntax -> UnsupportedExpression
+        PredicateConverter converter = new PredicateConverter(RowType.of(new VarCharType()));
+        assertThatThrownBy(
+                        () ->
+                                call(
+                                                BuiltInFunctionDefinitions.SIMILAR,
+                                                field(0, STRING()),
+                                                literal("(cat|dog)%", STRING()))
+                                        .accept(converter))
+                .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+    }
+
+    @Test
+    public void testSimilarExpressionCharacterClass() {
+        // '[a-z]+' uses SIMILAR TO-only syntax -> UnsupportedExpression
+        PredicateConverter converter = new PredicateConverter(RowType.of(new VarCharType()));
+        assertThatThrownBy(
+                        () ->
+                                call(
+                                                BuiltInFunctionDefinitions.SIMILAR,
+                                                field(0, STRING()),
+                                                literal("[a-z]+", STRING()))
+                                        .accept(converter))
+                .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+    }
+
+    @Test
+    public void testSimilarExpressionQuantifier() {
+        // 'a{2}' uses the SIMILAR TO-only {m,n} quantifier syntax (equivalent to 'aa') and must
+        // not be pushed down as a literal LIKE pattern matching 'a{2}'.
+        PredicateConverter converter = new PredicateConverter(RowType.of(new VarCharType()));
+        assertThatThrownBy(
+                        () ->
+                                call(
+                                                BuiltInFunctionDefinitions.SIMILAR,
+                                                field(0, STRING()),
+                                                literal("a{2}", STRING()))
+                                        .accept(converter))
+                .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+    }
+
+    @Test
+    public void testSimilarExpressionEscapedWildcards() {
+        // 'a=%b' SIMILAR TO with escape '=': literal 'a%b' (% is not a wildcard)
+        // convertSimilarToLike converts this to 'a\%b', which Like then processes as 'a' + literal
+        // '%' + 'b'
+        PredicateConverter converter = new PredicateConverter(RowType.of(new VarCharType()));
+        CallExpression expr =
+                call(
+                        BuiltInFunctionDefinitions.SIMILAR,
+                        field(0, STRING()),
+                        literal("a=%b", STRING()),
+                        literal("=", STRING()));
+        Predicate predicate = expr.accept(converter);
+
+        assertThat(predicate.test(GenericRow.of(BinaryString.fromString("a%b")))).isTrue();
+        assertThat(predicate.test(GenericRow.of(BinaryString.fromString("axb")))).isFalse();
+        assertThat(predicate.test(GenericRow.of(BinaryString.fromString("ab")))).isFalse();
+    }
+
+    @Test
+    public void testSimilarExpressionNonStringTypeThrows() {
+        // SIMILAR on a non-string column must throw UnsupportedExpression
+        assertThatThrownBy(
+                        () ->
+                                call(
+                                                BuiltInFunctionDefinitions.SIMILAR,
+                                                field(0, DataTypes.INT()),
+                                                literal(5))
+                                        .accept(new PredicateConverter(RowType.of(new IntType()))))
+                .isInstanceOf(PredicateConverter.UnsupportedExpression.class);
+    }
+
     private static FieldReferenceExpression field(int i, DataType type) {
         return new FieldReferenceExpression("f" + i, type, Integer.MAX_VALUE, Integer.MAX_VALUE);
     }