[flink]some fix

Author: Alibaba-HZY

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

@@ -369,6 +369,7 @@ public DynamicTableSource copy() {
         source.projectedFields = projectedFields;
         source.singleRowFilter = singleRowFilter;
         source.modificationScanType = modificationScanType;
+        source.predicate = predicate;
         return source;
     }
 

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

@@ -1179,6 +1179,76 @@ void testStreamingReadPartitionPushDownWithInExpr() throws Exception {
                         .collect();
 
         assertResultsIgnoreOrder(rowIter, expectedRowValues, true);
+
+        plan = tEnv.explainSql("select * from partitioned_table_in where c ='2025' or  c ='2026'");
+        assertThat(plan)
+                .contains(
+                        "TableSourceScan(table=[[testcatalog, defaultdb, partitioned_table_in, filter=[OR(=(c, _UTF-16LE'2025':VARCHAR(2147483647) CHARACTER SET \"UTF-16LE\"), =(c, _UTF-16LE'2026':VARCHAR(2147483647) CHARACTER SET \"UTF-16LE\"))]]], fields=[a, b, c])");
+
+        rowIter =
+                tEnv.executeSql("select * from partitioned_table_in where c ='2025' or  c ='2026'")
+                        .collect();
+
+        assertResultsIgnoreOrder(rowIter, expectedRowValues, true);
+    }
+
+    @Test
+    void testStreamingReadWithCombinedFiltersAndInExpr() throws Exception {
+        tEnv.executeSql(
+                "create table combined_filters_table_in"
+                        + " (a int not null, b varchar, c string, d int, primary key (a, c) NOT ENFORCED) partitioned by (c) ");
+        TablePath tablePath = TablePath.of(DEFAULT_DB, "combined_filters_table_in");
+        tEnv.executeSql("alter table combined_filters_table_in add partition (c=2025)");
+        tEnv.executeSql("alter table combined_filters_table_in add partition (c=2026)");
+        tEnv.executeSql("alter table combined_filters_table_in add partition (c=2027)");
+
+        List<InternalRow> rows = new ArrayList<>();
+        List<String> expectedRowValues = new ArrayList<>();
+
+        for (int i = 0; i < 10; i++) {
+            rows.add(row(i, "v" + i, "2025", i * 100));
+            if (i % 2 == 0) {
+                expectedRowValues.add(String.format("+I[%d, 2025, %d]", i, i * 100));
+            }
+        }
+        for (int i = 0; i < 10; i++) {
+            rows.add(row(i, "v" + i, "2026", i * 100));
+            if (i % 2 == 0) {
+                expectedRowValues.add(String.format("+I[%d, 2026, %d]", i, i * 100));
+            }
+        }
+        writeRows(conn, tablePath, rows, false);
+
+        for (int i = 0; i < 10; i++) {
+            rows.add(row(i, "v" + i, "2027", i * 100));
+        }
+
+        writeRows(conn, tablePath, rows, false);
+        waitUtilAllBucketFinishSnapshot(admin, tablePath, Arrays.asList("2025", "2026", "2027"));
+
+        String plan =
+                tEnv.explainSql(
+                        "select a,c,d from combined_filters_table_in where c in ('2025','2026') and d % 200 = 0");
+        assertThat(plan)
+                .contains(
+                        "TableSourceScan(table=[[testcatalog, defaultdb, combined_filters_table_in, filter=[OR(=(c, _UTF-16LE'2025'), =(c, _UTF-16LE'2026'))], project=[a, c, d]]], fields=[a, c, d])");
+
+        // test column filter、partition filter and flink runtime filter
+        org.apache.flink.util.CloseableIterator<Row> rowIter =
+                tEnv.executeSql(
+                                "select a,c,d from combined_filters_table_in where c in ('2025','2026') "
+                                        + "and d % 200 = 0")
+                        .collect();
+
+        assertResultsIgnoreOrder(rowIter, expectedRowValues, true);
+
+        rowIter =
+                tEnv.executeSql(
+                                "select a,c,d from combined_filters_table_in where (c ='2025' or  c ='2026') "
+                                        + "and d % 200 = 0")
+                        .collect();
+
+        assertResultsIgnoreOrder(rowIter, expectedRowValues, true);
     }
 
     private enum Caching {