[BugFix] If the operator is AND, return the predicates that are not null for paimon (#66038)

(cherry picked from commit 9313d06)

# Conflicts:
#	fe/fe-core/src/main/java/com/starrocks/connector/paimon/PaimonPredicateConverter.java

Author: liaowanyou

fe/fe-core/src/main/java/com/starrocks/connector/paimon/PaimonPredicateContext.java

@@ -0,0 +1,29 @@
+// Copyright 2021-present StarRocks, Inc. All rights reserved.
+//
+// Licensed 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
+//
+//     https://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.starrocks.connector.paimon;
+
+import com.starrocks.sql.optimizer.operator.scalar.CompoundPredicateOperator;
+
+public class PaimonPredicateContext {
+    private CompoundPredicateOperator parentPredicateOperator;
+
+    public CompoundPredicateOperator getParentPredicateOperator() {
+        return parentPredicateOperator;
+    }
+
+    public void setParentPredicateOperator(
+            CompoundPredicateOperator parentPredicateOperator) {
+        this.parentPredicateOperator = parentPredicateOperator;
+    }
+}

fe/fe-core/src/main/java/com/starrocks/connector/paimon/PaimonPredicateConverter.java

@@ -63,7 +63,7 @@
 
 import static org.apache.paimon.data.Timestamp.fromLocalDateTime;
 
-public class PaimonPredicateConverter extends ScalarOperatorVisitor<Predicate, Void> {
+public class PaimonPredicateConverter extends ScalarOperatorVisitor<Predicate, PaimonPredicateContext> {
     private static final Logger LOG = LogManager.getLogger(PaimonPredicateConverter.class);
     private final PredicateBuilder builder;
     private final List<String> fieldNames;
@@ -83,35 +83,48 @@ public Predicate convert(ScalarOperator operator) {
     }
 
     @Override
-    public Predicate visit(ScalarOperator scalarOperator, Void context) {
+    public Predicate visit(ScalarOperator scalarOperator, PaimonPredicateContext context) {
         return null;
     }
 
     @Override
-    public Predicate visitCompoundPredicate(CompoundPredicateOperator operator, Void context) {
+    public Predicate visitCompoundPredicate(CompoundPredicateOperator operator, PaimonPredicateContext context) {
         CompoundPredicateOperator.CompoundType op = operator.getCompoundType();
         if (op == CompoundPredicateOperator.CompoundType.NOT) {
             if (operator.getChild(0) instanceof LikePredicateOperator) {
                 return null;
             }
-            Predicate expression = operator.getChild(0).accept(this, null);
-
+            PaimonPredicateContext paimonPredicateContext = new PaimonPredicateContext();
+            paimonPredicateContext.setParentPredicateOperator(operator);
+            Predicate expression = operator.getChild(0).accept(this, paimonPredicateContext);
             if (expression != null) {
                 return expression.negate().orElse(null);
             }
         } else {
-            Predicate left = operator.getChild(0).accept(this, null);
-            Predicate right = operator.getChild(1).accept(this, null);
+            Predicate left = operator.getChild(0).accept(this, context);
+            Predicate right = operator.getChild(1).accept(this, context);
+            CompoundPredicateOperator parentPredicateOperator = null;
+            if (context != null && context.getParentPredicateOperator() != null) {
+                parentPredicateOperator = context.getParentPredicateOperator();
+            }
             if (left != null && right != null) {
                 return (op == CompoundPredicateOperator.CompoundType.OR) ? PredicateBuilder.or(left, right) :
                         PredicateBuilder.and(left, right);
+            } else if (parentPredicateOperator != null &&
+                    parentPredicateOperator.getCompoundType() == CompoundPredicateOperator.CompoundType.NOT) {
+                return null;
+            } else if (left != null && op == CompoundPredicateOperator.CompoundType.AND) {
+                //if op=and, return the predicates that are not null.
+                return left;
+            } else if (right != null && op == CompoundPredicateOperator.CompoundType.AND) {
+                return right;
             }
         }
         return null;
     }
 
     @Override
-    public Predicate visitIsNullPredicate(IsNullPredicateOperator operator, Void context) {
+    public Predicate visitIsNullPredicate(IsNullPredicateOperator operator, PaimonPredicateContext context) {
         String columnName = getColumnName(operator.getChild(0));
         if (columnName == null) {
             return null;
@@ -125,7 +138,7 @@ public Predicate visitIsNullPredicate(IsNullPredicateOperator operator, Void con
     }
 
     @Override
-    public Predicate visitBinaryPredicate(BinaryPredicateOperator operator, Void context) {
+    public Predicate visitBinaryPredicate(BinaryPredicateOperator operator, PaimonPredicateContext context) {
         String columnName = getColumnName(operator.getChild(0));
         if (columnName == null) {
             return null;
@@ -157,7 +170,16 @@ public Predicate visitBinaryPredicate(BinaryPredicateOperator operator, Void con
     }
 
     @Override
+<<<<<<< HEAD
     public Predicate visitInPredicate(InPredicateOperator operator, Void context) {
+=======
+    public Predicate visitLargeInPredicate(LargeInPredicateOperator operator, PaimonPredicateContext context) {
+        throw new UnsupportedOperationException("not support large in predicate in the PaimonPredicateConverter");
+    }
+
+    @Override
+    public Predicate visitInPredicate(InPredicateOperator operator, PaimonPredicateContext context) {
+>>>>>>> 9313d06954 ([BugFix] If the operator is AND, return the predicates that are not null for paimon (#66038))
         String columnName = getColumnName(operator.getChild(0));
         if (columnName == null) {
             return null;
@@ -184,7 +206,7 @@ public Predicate visitInPredicate(InPredicateOperator operator, Void context) {
     }
 
     @Override
-    public Predicate visitLikePredicateOperator(LikePredicateOperator operator, Void context) {
+    public Predicate visitLikePredicateOperator(LikePredicateOperator operator, PaimonPredicateContext context) {
         String columnName = getColumnName(operator.getChild(0));
         if (columnName == null) {
             return null;
@@ -375,4 +397,4 @@ public String visitCastOperator(CastOperator operator, Void context) {
             return operator.getChild(0).accept(this, context);
         }
     }
-}
+}

fe/fe-core/src/test/java/com/starrocks/connector/paimon/PaimonPredicateConverterTest.java

@@ -18,6 +18,7 @@
 import com.starrocks.analysis.BinaryType;
 import com.starrocks.catalog.Type;
 import com.starrocks.sql.optimizer.operator.scalar.BinaryPredicateOperator;
+import com.starrocks.sql.optimizer.operator.scalar.CaseWhenOperator;
 import com.starrocks.sql.optimizer.operator.scalar.CastOperator;
 import com.starrocks.sql.optimizer.operator.scalar.ColumnRefOperator;
 import com.starrocks.sql.optimizer.operator.scalar.CompoundPredicateOperator;
@@ -400,4 +401,53 @@ public void testPaimonCastPredicate() {
         result = CONVERTER.convert(new BinaryPredicateOperator(BinaryType.EQ, cast99, d));
         Assert.assertNull(result);
     }
+
+    @Test
+    public void testOrWithFunction() {
+        //    (f0 = 44 and (case when f0 = 44 then 'test' end) = 'test')
+        // OR (f0 <= 46 and (case when f0 = 44 then 'test' end) = 'test')
+        // OR((case when f0 = 44 then 'test' end) = 'test' and f1 like 'ttt%')
+        // return f0 = 44 OR f0 <= 46 OR f0 <= 20
+        BinaryPredicateOperator op2 = new BinaryPredicateOperator(
+                BinaryType.EQ, F0, ConstantOperator.createInt(44));
+        CaseWhenOperator caseWhenOperator = new CaseWhenOperator(IntegerType.INT, op2, null,
+                Lists.newArrayList(op2, ConstantOperator.createVarchar("test")));
+        BinaryPredicateOperator test =
+                new BinaryPredicateOperator(BinaryType.EQ, caseWhenOperator, ConstantOperator.createVarchar("test"));
+        ScalarOperator op20 = new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.AND, op2,
+                test.clone());
+
+        BinaryPredicateOperator op12 = new BinaryPredicateOperator(
+                BinaryType.LE, F0, ConstantOperator.createInt(46));
+        ScalarOperator op21 = new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.AND, op12,
+                test.clone());
+
+        ConstantOperator value = ConstantOperator.createVarchar("ttt%");
+        ScalarOperator op13 = new LikePredicateOperator(LikePredicateOperator.LikeType.LIKE, F1, value);
+        ScalarOperator op22 = new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.AND,
+                test.clone(), op13);
+
+        CompoundPredicateOperator compoundPredicateOperator =
+                new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.OR, op20, op21);
+        CompoundPredicateOperator compoundPredicateOperator1 =
+                new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.OR, compoundPredicateOperator,
+                        op22);
+        CompoundPredicate convert = (CompoundPredicate) CONVERTER.convert(compoundPredicateOperator1);
+        Assertions.assertTrue(
+                "Or([Or([Equal(f0, 44), LessOrEqual(f0, 46)]), StartsWith(f1, ttt)])".equals(convert.toString()));
+
+        // (case when f0 = 44 then 'test' end) = 'test'
+        // OR (f0 <= 46 and (case when f0 = 44 then 'test' end) = 'test')
+        // return null
+        ScalarOperator op40 = new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.OR, test.clone(),
+                op21.clone());
+        CompoundPredicate convert1 = (CompoundPredicate) CONVERTER.convert(op40.clone());
+        Assertions.assertTrue(convert1 == null);
+
+        // NOT ((case when f0 = 44 then 'test' end) = 'test' and  f1 like 'ttt%')
+        // return null
+        ScalarOperator op52 = new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.NOT, op22.clone());
+        Predicate convert2 = CONVERTER.convert(op52);
+        Assertions.assertTrue(convert2 == null);
+    }
 }