[Enhancement] Support partial pushdown for AND compound predicates in Iceberg connector (backport #70293) (#73085)

Signed-off-by: Jiwon Park <jpark92@outlook.kr>
Co-authored-by: Jiwon Park <jpark92@outlook.kr>

Author: mergify[bot]

fe/fe-core/src/main/java/com/starrocks/connector/iceberg/ScalarOperatorToIcebergExpr.java

@@ -110,9 +110,12 @@ public Expression convertStrict(List<ScalarOperator> operators, IcebergContext c
 
     public Expression convert(List<ScalarOperator> operators, IcebergContext context, boolean strict) {
         IcebergExprVisitor visitor = new IcebergExprVisitor();
+        IcebergContext effectiveContext = strict && !context.isStrict()
+                ? new IcebergContext(context.getSchema(), context.isInsideNot(), true)
+                : context;
         List<Expression> expressions = Lists.newArrayList();
         for (ScalarOperator operator : operators) {
-            Expression filterExpr = operator.accept(visitor, context);
+            Expression filterExpr = operator.accept(visitor, effectiveContext);
             if (filterExpr == null) {
                 if (strict) {
                     LOG.debug("Strict mode: cannot convert operator {}", operator.debugString());
@@ -141,14 +144,34 @@ public Expression convert(List<ScalarOperator> operators, IcebergContext context
 
     public static class IcebergContext {
         private final Types.StructType schema;
+        private final boolean insideNot;
+        private final boolean strict;
 
         public IcebergContext(Types.StructType schema) {
+            this(schema, false, false);
+        }
+
+        public IcebergContext(Types.StructType schema, boolean insideNot, boolean strict) {
             this.schema = schema;
+            this.insideNot = insideNot;
+            this.strict = strict;
         }
 
         public Types.StructType getSchema() {
             return schema;
         }
+
+        public boolean isInsideNot() {
+            return insideNot;
+        }
+
+        public boolean isStrict() {
+            return strict;
+        }
+
+        public IcebergContext withInsideNot() {
+            return new IcebergContext(schema, true, strict);
+        }
     }
 
     private static class IcebergExprVisitor extends ScalarOperatorVisitor<Expression, IcebergContext> {
@@ -185,7 +208,7 @@ public Expression visitCompoundPredicate(CompoundPredicateOperator operator, Ice
                 if (operator.getChild(0) instanceof LikePredicateOperator) {
                     return null;
                 }
-                Expression expression = operator.getChild(0).accept(this, context);
+                Expression expression = operator.getChild(0).accept(this, context.withInsideNot());
 
                 if (expression != null) {
                     return not(expression);
@@ -196,6 +219,21 @@ public Expression visitCompoundPredicate(CompoundPredicateOperator operator, Ice
                 if (left != null && right != null) {
                     return (op == CompoundPredicateOperator.CompoundType.OR) ? or(left, right) : and(left, right);
                 }
+                // For AND predicates outside of NOT, allow partial pushdown.
+                // If only one side converts successfully, push down that side alone.
+                // This is safe because AND(a, b) is more restrictive than just a (or just b),
+                // so pushing down one side still correctly filters data.
+                // This must NOT be done inside NOT, because NOT(AND(a, b)) = OR(NOT(a), NOT(b)),
+                // and pushing down NOT(a) alone would over-filter.
+                if (!context.isStrict() && !context.isInsideNot()
+                        && op == CompoundPredicateOperator.CompoundType.AND) {
+                    if (left != null) {
+                        return left;
+                    }
+                    if (right != null) {
+                        return right;
+                    }
+                }
             }
             return null;
         }

fe/fe-core/src/test/java/com/starrocks/connector/iceberg/IcebergExprVisitorTest.java

@@ -451,6 +451,92 @@ public void testToIcebergExpressionDotColumn() {
         Assertions.assertEquals(Expression.Operation.TRUE, convertedExpr.op());
     }
 
+    @Test
+    public void testAndPartialPushdown() {
+        ScalarOperatorToIcebergExpr.IcebergContext context = new ScalarOperatorToIcebergExpr.IcebergContext(SCHEMA.asStruct());
+        ScalarOperatorToIcebergExpr converter = new ScalarOperatorToIcebergExpr();
+
+        // Build a convertible predicate: k1 > 10
+        BinaryPredicateOperator convertible = new BinaryPredicateOperator(
+                BinaryType.GT, K1, ConstantOperator.createInt(10));
+
+        // Build an unconvertible predicate: CAST(k6 AS INT) < 5
+        // CastOperator(INT, K6) where K6 is a string column causes getLiteralValue to return null
+        CastOperator cast = new CastOperator(IntegerType.INT, K6);
+        BinaryPredicateOperator unconvertible = new BinaryPredicateOperator(
+                BinaryType.LT, cast, ConstantOperator.createInt(5));
+
+        // Verify that the unconvertible predicate alone produces alwaysTrue
+        Expression unconvertibleExpr = converter.convert(Lists.newArrayList(unconvertible), context);
+        Assertions.assertEquals(Expression.Operation.TRUE, unconvertibleExpr.op(),
+                "Unconvertible predicate should produce alwaysTrue");
+
+        Expression convertedExpr;
+        Expression expectedExpr;
+
+        // AND(convertible, unconvertible) -> returns the convertible side
+        convertedExpr = converter.convert(Lists.newArrayList(
+                new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.AND,
+                        convertible, unconvertible)), context);
+        expectedExpr = Expressions.greaterThan("k1", 10);
+        Assertions.assertEquals(expectedExpr.toString(), convertedExpr.toString(),
+                "AND(convertible, unconvertible) should push down the convertible side");
+
+        // AND(unconvertible, convertible) -> returns the convertible side
+        convertedExpr = converter.convert(Lists.newArrayList(
+                new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.AND,
+                        unconvertible, convertible)), context);
+        expectedExpr = Expressions.greaterThan("k1", 10);
+        Assertions.assertEquals(expectedExpr.toString(), convertedExpr.toString(),
+                "AND(unconvertible, convertible) should push down the convertible side");
+
+        // OR(convertible, unconvertible) -> returns alwaysTrue (no partial pushdown for OR)
+        convertedExpr = converter.convert(Lists.newArrayList(
+                new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.OR,
+                        convertible, unconvertible)), context);
+        Assertions.assertEquals(Expression.Operation.TRUE, convertedExpr.op(),
+                "OR(convertible, unconvertible) should NOT do partial pushdown");
+
+        // NOT(AND(convertible, unconvertible)) -> returns alwaysTrue (no partial pushdown inside NOT)
+        CompoundPredicateOperator andOp = new CompoundPredicateOperator(
+                CompoundPredicateOperator.CompoundType.AND, convertible, unconvertible);
+        convertedExpr = converter.convert(Lists.newArrayList(
+                new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.NOT, andOp)), context);
+        Assertions.assertEquals(Expression.Operation.TRUE, convertedExpr.op(),
+                "NOT(AND(convertible, unconvertible)) should NOT do partial pushdown");
+
+        // AND(convertible1, convertible2) -> returns and(left, right) (regression test)
+        BinaryPredicateOperator convertible2 = new BinaryPredicateOperator(
+                BinaryType.LT, K2, ConstantOperator.createInt(20));
+        convertedExpr = converter.convert(Lists.newArrayList(
+                new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.AND,
+                        convertible, convertible2)), context);
+        expectedExpr = Expressions.and(
+                Expressions.greaterThan("k1", 10),
+                Expressions.lessThan("k2", 20));
+        Assertions.assertEquals(expectedExpr.toString(), convertedExpr.toString(),
+                "AND(convertible1, convertible2) should return and(left, right)");
+
+        // Nested AND: AND(AND(convertible, unconvertible), convertible2) -> returns and(convertible, convertible2)
+        CompoundPredicateOperator innerAnd = new CompoundPredicateOperator(
+                CompoundPredicateOperator.CompoundType.AND, convertible, unconvertible);
+        convertedExpr = converter.convert(Lists.newArrayList(
+                new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.AND,
+                        innerAnd, convertible2)), context);
+        expectedExpr = Expressions.and(
+                Expressions.greaterThan("k1", 10),
+                Expressions.lessThan("k2", 20));
+        Assertions.assertEquals(expectedExpr.toString(), convertedExpr.toString(),
+                "AND(AND(convertible, unconvertible), convertible2) should return and(convertible, convertible2)");
+
+        // Strict mode: AND(convertible, unconvertible) -> returns null (no partial pushdown)
+        convertedExpr = converter.convertStrict(Lists.newArrayList(
+                new CompoundPredicateOperator(CompoundPredicateOperator.CompoundType.AND,
+                        convertible, unconvertible)), context);
+        Assertions.assertNull(convertedExpr,
+                "Strict mode: AND(convertible, unconvertible) should return null");
+    }
+
     @Test
     public void testConvertLastUpdatedSequenceNumberPredicate() {
         ScalarOperatorToIcebergExpr.IcebergContext context = new ScalarOperatorToIcebergExpr.IcebergContext(SCHEMA.asStruct());