perf: refactor slow path

src/flow/src/compute/render/reduce.rs

@@ -378,7 +378,7 @@ fn reduce_batch_subgraph(
 
             // optimize use bool mask to avoid unnecessary slice
             for row_idx in 0..key_batch.row_count() {
-                let key_row = key_batch.get_row(row_idx).unwrap();
+                let key_row = key_batch.get_row(row_idx)?;
                 let key_row = Row::new(key_row);
 
                 // if the same key exist then it's already filtered all values in this batch

src/flow/src/expr/linear.rs

@@ -17,6 +17,7 @@
 use std::collections::{BTreeMap, BTreeSet};
 
 use arrow::array::BooleanArray;
+use arrow::buffer::BooleanBuffer;
 use arrow::compute::FilterBuilder;
 use common_telemetry::debug;
 use datatypes::prelude::ConcreteDataType;
@@ -523,7 +524,9 @@ impl SafeMfpPlan {
         // mark the columns that have been evaluated and appended to the `batch`
         let mut expression = 0;
         // preds default to true and will be updated as we evaluate each predicate
-        let mut all_preds = BooleanVector::from(vec![Some(true); batch.row_count()]);
+        let buf = BooleanBuffer::new_set(batch.row_count());
+        let arr = BooleanArray::new(buf, None);
+        let mut all_preds = BooleanVector::from(arr);
 
         // to compute predicate, need to first compute all expressions used in predicates
         for (support, predicate) in self.mfp.predicates.iter() {