[bugfix](paimon)add support for 'in' and 'not in' for 2.1 #38390 (#38576)

## Proposed changes

bp: #38390

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonPredicateConverter.java

@@ -21,9 +21,11 @@ import org.apache.doris.analysis.CastExpr;
 import org.apache.doris.analysis.CompoundPredicate;
 import org.apache.doris.analysis.Expr;
 import org.apache.doris.analysis.FunctionCallExpr;
+import org.apache.doris.analysis.InPredicate;
 import org.apache.doris.analysis.IsNullPredicate;
 import org.apache.doris.analysis.LiteralExpr;
 import org.apache.doris.analysis.SlotRef;
+import org.apache.doris.analysis.Subquery;
 import org.apache.doris.thrift.TExprOpcode;
 
 import org.apache.paimon.data.BinaryString;
@@ -85,11 +87,48 @@ public class PaimonPredicateConverter {
                 default:
                     return null;
             }
+        } else if (dorisExpr instanceof InPredicate) {
+            return doInPredicate((InPredicate) dorisExpr);
         } else {
             return binaryExprDesc(dorisExpr);
         }
     }
 
+    private Predicate doInPredicate(InPredicate predicate) {
+        // InPredicate, only support a in (1,2,3)
+        if (predicate.contains(Subquery.class)) {
+            return null;
+        }
+
+        SlotRef slotRef = convertDorisExprToSlotRef(predicate.getChild(0));
+        if (slotRef == null) {
+            return null;
+        }
+        String colName = slotRef.getColumnName();
+        int idx = fieldNames.indexOf(colName);
+        DataType dataType = paimonFieldTypes.get(idx);
+        List<Object> valueList = new ArrayList<>();
+        for (int i = 1; i < predicate.getChildren().size(); i++) {
+            if (!(predicate.getChild(i) instanceof LiteralExpr)) {
+                return null;
+            }
+            LiteralExpr literalExpr = convertDorisExprToLiteralExpr(predicate.getChild(i));
+            Object value = dataType.accept(new PaimonValueConverter(literalExpr));
+            if (value == null) {
+                return null;
+            }
+            valueList.add(value);
+        }
+
+        if (predicate.isNotIn()) {
+            // not in
+            return builder.notIn(idx, valueList);
+        } else {
+            // in
+            return builder.in(idx, valueList);
+        }
+    }
+
     private Predicate binaryExprDesc(Expr dorisExpr) {
         TExprOpcode opcode = dorisExpr.getOpcode();
         // Make sure the col slot is always first

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java

@@ -346,15 +346,26 @@ public class PaimonScanNode extends FileQueryScanNode {
 
     @Override
     public String getNodeExplainString(String prefix, TExplainLevel detailLevel) {
-        String result = super.getNodeExplainString(prefix, detailLevel)
-                + String.format("%spaimonNativeReadSplits=%d/%d\n",
-                        prefix, rawFileSplitNum, (paimonSplitNum + rawFileSplitNum));
-        if (detailLevel == TExplainLevel.VERBOSE) {
-            result += prefix + "PaimonSplitStats: \n";
-            for (SplitStat splitStat : splitStats) {
-                result += String.format("%s  %s\n", prefix, splitStat);
+        StringBuilder sb = new StringBuilder(super.getNodeExplainString(prefix, detailLevel));
+        sb.append(String.format("%spaimonNativeReadSplits=%d/%d\n",
+                prefix, rawFileSplitNum, (paimonSplitNum + rawFileSplitNum)));
+
+        sb.append(prefix).append("predicatesFromPaimon:");
+        if (predicates.isEmpty()) {
+            sb.append(" NONE\n");
+        } else {
+            sb.append("\n");
+            for (Predicate predicate : predicates) {
+                sb.append(prefix).append(prefix).append(predicate).append("\n");
             }
         }
-        return result;
+
+        if (detailLevel == TExplainLevel.VERBOSE) {
+            sb.append(prefix).append("PaimonSplitStats: \n");
+            for (SplitStat splitStat : splitStats) {
+                sb.append(String.format("%s  %s\n", prefix, splitStat));
+            }
+        }
+        return sb.toString();
     }
 }