Fix.

src/query/storages/parquet/src/parquet_rs/parquet_reader.rs

@@ -265,30 +265,30 @@ impl ParquetRSReader {
                 )
                 .await?;
 
-            let mut reader = ParquetRecordBatchReader::try_new_with_row_groups(
+            let reader = ParquetRecordBatchReader::try_new_with_row_groups(
                 predicate.field_levels(),
                 &row_group,
-                usize::MAX, // Read all rows for eval prewhere predicate.
+                self.batch_size,
                 selection.clone(),
             )?;
 
-            if let Some(batch) = reader.next() {
-                assert!(reader.next().is_none());
+            let mut filters = vec![];
+            for batch in reader {
                 let batch = batch?;
                 let filter = predicate.evaluate(&batch)?;
-                let sel = RowSelection::from_filters(&[filter]);
-                if !sel.selects_any() {
-                    // All rows in current row group are filtered out.
-                    return Ok(None);
+                filters.push(filter);
+            }
+            let sel = RowSelection::from_filters(&filters);
+            if !sel.selects_any() {
+                // All rows in current row group are filtered out.
+                return Ok(None);
+            }
+            match selection.as_mut() {
+                Some(selection) => {
+                    selection.and_then(&sel);
                 }
-
-                match selection.as_mut() {
-                    Some(selection) => {
-                        selection.and_then(&sel);
-                    }
-                    None => {
-                        selection = Some(sel);
-                    }
+                None => {
+                    selection = Some(sel);
                 }
             }
         }