perf(stream): add simple strategy for if the stream project compact t…

…he chunk (risingwavelabs#8758)

src/common/src/array/data_chunk.rs

@@ -138,6 +138,19 @@ impl DataChunk {
         &self.vis2
     }
 
+    pub fn selectivity(&self) -> f64 {
+        match &self.vis2 {
+            Vis::Bitmap(b) => {
+                if b.is_empty() {
+                    0.0
+                } else {
+                    b.count_ones() as f64 / b.len() as f64
+                }
+            }
+            Vis::Compact(_) => 1.0,
+        }
+    }
+
     pub fn with_visibility(&self, visibility: Bitmap) -> Self {
         DataChunk::new(self.columns.clone(), visibility)
     }

src/common/src/array/stream_chunk.rs

@@ -133,6 +133,10 @@ impl StreamChunk {
         self.data.capacity()
     }
 
+    pub fn selectivity(&self) -> f64 {
+        self.data.selectivity()
+    }
+
     /// Get the reference of the underlying data chunk.
     pub fn data_chunk(&self) -> &DataChunk {
         &self.data

src/stream/src/executor/integration_tests.rs

@@ -196,6 +196,7 @@ async fn test_merger_sum_aggr() {
         ],
         3,
         MultiMap::new(),
+        0.0,
     );
 
     let items = Arc::new(Mutex::new(vec![]));

src/stream/src/executor/project.rs

@@ -40,6 +40,10 @@ struct Inner {
     /// All the watermark derivations, (input_column_index, output_column_index). And the
     /// derivation expression is the project's expression itself.
     watermark_derivations: MultiMap<usize, usize>,
+
+    /// the selectivity threshold which should be in [0,1]. for the chunk with selectivity less
+    /// than the threshold, the Project executor will construct a new chunk before expr evaluation,
+    materialize_selectivity_threshold: f64,
 }
 
 impl ProjectExecutor {
@@ -50,6 +54,7 @@ impl ProjectExecutor {
         exprs: Vec<BoxedExpression>,
         executor_id: u64,
         watermark_derivations: MultiMap<usize, usize>,
+        materialize_selectivity_threshold: f64,
     ) -> Self {
         let info = ExecutorInfo {
             schema: input.schema().to_owned(),
@@ -74,6 +79,7 @@ impl ProjectExecutor {
                 },
                 exprs,
                 watermark_derivations,
+                materialize_selectivity_threshold,
             },
         }
     }
@@ -110,10 +116,12 @@ impl Inner {
         &self,
         chunk: StreamChunk,
     ) -> StreamExecutorResult<Option<StreamChunk>> {
-        let chunk = chunk.compact();
-
+        let chunk = if chunk.selectivity() <= self.materialize_selectivity_threshold {
+            chunk.compact()
+        } else {
+            chunk
+        };
         let (data_chunk, ops) = chunk.into_parts();
-
         let mut projected_columns = Vec::new();
 
         for expr in &self.exprs {
@@ -125,8 +133,9 @@ impl Inner {
             let new_column = Column::new(evaluated_expr);
             projected_columns.push(new_column);
         }
-
-        let new_chunk = StreamChunk::new(ops, projected_columns, None);
+        let (_, vis) = data_chunk.into_parts();
+        let vis = vis.into_visibility();
+        let new_chunk = StreamChunk::new(ops, projected_columns, vis);
         Ok(Some(new_chunk))
     }
 
@@ -233,6 +242,7 @@ mod tests {
             vec![test_expr],
             1,
             MultiMap::new(),
+            0.0,
         ));
         let mut project = project.execute();
 
@@ -299,6 +309,7 @@ mod tests {
             vec![a_expr, b_expr],
             1,
             MultiMap::from_iter(vec![(0, 0), (0, 1)].into_iter()),
+            0.0,
         ));
         let mut project = project.execute();
 

src/stream/src/from_proto/project.rs

@@ -15,6 +15,7 @@
 use multimap::MultiMap;
 use risingwave_common::util::iter_util::ZipEqFast;
 use risingwave_expr::expr::build_from_prost;
+use risingwave_pb::expr::expr_node;
 use risingwave_pb::stream_plan::ProjectNode;
 
 use super::*;
@@ -49,14 +50,19 @@ impl ExecutorBuilder for ProjectExecutorBuilder {
                         .map(|key| *key as usize),
                 ),
         );
-
+        let extremely_light = node.get_select_list().iter().all(|expr| {
+            let expr_type = expr.get_expr_type().unwrap();
+            expr_type == expr_node::Type::InputRef || expr_type == expr_node::Type::ConstantValue
+        });
+        let materialize_selectivity_threshold = if extremely_light { 0.0 } else { 0.5 };
         Ok(ProjectExecutor::new(
             params.actor_context,
             input,
             params.pk_indices,
             project_exprs,
             params.executor_id,
             watermark_derivations,
+            materialize_selectivity_threshold,
         )
         .boxed())
     }