[flink]add index validate

Author: Alibaba-HZY

fluss-common/src/main/java/com/alibaba/fluss/predicate/PredicateBuilder.java

@@ -121,18 +121,21 @@ public Predicate contains(int idx, Object patternLiteral) {
     }
 
     public Predicate leaf(NullFalseLeafBinaryFunction function, int idx, Object literal) {
+        validateIndex(idx);
         DataField field = rowType.getFields().get(idx);
         return new LeafPredicate(
                 function, field.getType(), idx, field.getName(), singletonList(literal));
     }
 
     public Predicate leaf(LeafUnaryFunction function, int idx) {
+        validateIndex(idx);
         DataField field = rowType.getFields().get(idx);
         return new LeafPredicate(
                 function, field.getType(), idx, field.getName(), Collections.emptyList());
     }
 
     public Predicate in(int idx, List<Object> literals) {
+        validateIndex(idx);
         // In the IN predicate, 20 literals are critical for performance.
         // If there are more than 20 literals, the performance will decrease.
         if (literals.size() > 20) {
@@ -233,6 +236,12 @@ private static void splitCompound(
         }
     }
 
+    private void validateIndex(int idx) {
+        if (idx < 0 || idx >= rowType.getFieldCount()) {
+            throw new UnsupportedExpression("idx is not valid");
+        }
+    }
+
     public static Object convertJavaObject(DataType literalType, Object o) {
         if (o == null) {
             return null;

fluss-common/src/main/java/com/alibaba/fluss/predicate/UnsupportedExpression.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.fluss.predicate;
+
+/** Encounter an unsupported expression, the caller can choose to ignore this filter branch. */
+public class UnsupportedExpression extends RuntimeException {
+    public UnsupportedExpression(String message) {
+        super(message);
+    }
+
+    public UnsupportedExpression() {
+        super();
+    }
+}

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/PredicateConverter.java

@@ -22,6 +22,7 @@
 import com.alibaba.fluss.flink.utils.FlinkConversions;
 import com.alibaba.fluss.predicate.Predicate;
 import com.alibaba.fluss.predicate.PredicateBuilder;
+import com.alibaba.fluss.predicate.UnsupportedExpression;
 import com.alibaba.fluss.utils.TypeUtils;
 
 import org.apache.flink.table.expressions.CallExpression;
@@ -119,11 +120,12 @@ public Predicate visit(CallExpression call) {
             FieldReferenceExpression fieldRefExpr =
                     extractFieldReference(children.get(0)).orElseThrow(UnsupportedExpression::new);
             if (fieldRefExpr
-                    .getOutputDataType()
-                    .getLogicalType()
-                    .getTypeRoot()
-                    .getFamilies()
-                    .contains(LogicalTypeFamily.CHARACTER_STRING)) {
+                            .getOutputDataType()
+                            .getLogicalType()
+                            .getTypeRoot()
+                            .getFamilies()
+                            .contains(LogicalTypeFamily.CHARACTER_STRING)
+                    && builder.indexOf(fieldRefExpr.getName()) != -1) {
                 String sqlPattern =
                         Objects.requireNonNull(
                                         extractLiteral(
@@ -294,7 +296,4 @@ public static Optional<Predicate> convert(RowType rowType, ResolvedExpression fi
             return Optional.empty();
         }
     }
-
-    /** Encounter an unsupported expression, the caller can choose to ignore this filter branch. */
-    public static class UnsupportedExpression extends RuntimeException {}
 }

fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/source/FlinkTableSourceITCase.java

@@ -1185,8 +1185,9 @@ private List<String> writeRowsToTwoPartition(TablePath tablePath, Collection<Str
             values[1] = keyValuePairs[1].split("=")[1];
 
             for (int i = 0; i < 10; i++) {
-                rows.add(row(i, "v1", values[0], values[1]));
-                expectedRowValues.add(String.format("+I[%d, v1, %s, %s]", i, values[0], values[1]));
+                rows.add(row(i, "v" + i, values[0], values[1]));
+                expectedRowValues.add(
+                        String.format("+I[%d, v%d, %s, %s]", i, i, values[0], values[1]));
             }
         }
 
@@ -1354,6 +1355,47 @@ void testStreamingReadPartitionPushDownWithLikeExpr() throws Exception {
         assertResultsIgnoreOrder(rowIter, expectedRowValues, true);
     }
 
+    @Test
+    void testStreamingReadPartitionComplexPushDown() throws Exception {
+
+        tEnv.executeSql(
+                "create table partitioned_table_complex"
+                        + " (a int not null, b varchar, c string,d string, primary key (a, c, d) NOT ENFORCED) partitioned by (c,d) ");
+        TablePath tablePath = TablePath.of(DEFAULT_DB, "partitioned_table_complex");
+        tEnv.executeSql("alter table partitioned_table_complex add partition (c=2025,d=1)");
+        tEnv.executeSql("alter table partitioned_table_complex add partition (c=2025,d=2)");
+        tEnv.executeSql("alter table partitioned_table_complex add partition (c=2026,d=1)");
+
+        List<String> allData =
+                writeRowsToTwoPartition(
+                        tablePath, Arrays.asList("c=2025,d=1", "c=2025,d=2", "c=2026,d=1"));
+        List<String> expectedRowValues =
+                allData.stream()
+                        .filter(s -> s.contains("v3") && !s.contains("2025, 2"))
+                        .collect(Collectors.toList());
+        waitUtilAllBucketFinishSnapshot(
+                admin, tablePath, Arrays.asList("2025$1", "2025$2", "2026$1"));
+
+        String plan =
+                tEnv.explainSql(
+                        "select * from partitioned_table_complex where  a = 3\n"
+                                + "    and (c in ('2026')  or d like '%1%') "
+                                + "    and     b like '%v3%'");
+        assertThat(plan)
+                .contains(
+                        "Calc(select=[3 AS a, b, c, d], where=[((a = 3) AND LIKE(b, '%v3%'))])\n"
+                                + "+- TableSourceScan(table=[[testcatalog, defaultdb, partitioned_table_complex, filter=[OR(=(c, _UTF-16LE'2026'), LIKE(d, _UTF-16LE'%1%'))]]], fields=[a, b, c, d])");
+
+        org.apache.flink.util.CloseableIterator<Row> rowIter =
+                tEnv.executeSql(
+                                "select * from partitioned_table_complex where  a = 3\n"
+                                        + "    and (c in ('2026')  or d like '%1%') "
+                                        + "    and     b like '%v3%'")
+                        .collect();
+
+        assertResultsIgnoreOrder(rowIter, expectedRowValues, true);
+    }
+
     private enum Caching {
         ENABLE_CACHE,
         DISABLE_CACHE