refactor: projection executor to reuse arena tmp values buffer

Author: KKould

src/execution/dql/join/nested_loop_join.rs

@@ -16,7 +16,6 @@
 //! [`JoinType::RightOuter`], [`JoinType::Cross`], [`JoinType::Full`].
 
 use crate::errors::DatabaseError;
-use crate::execution::dql::projection::Projection;
 use crate::execution::{
     build_read, ExecArena, ExecId, ExecNode, ExecutionCaches, ExecutorNode, ReadExecutor,
 };
@@ -65,10 +64,14 @@ impl EqualCondition {
         if self.on_left_keys.is_empty() {
             return Ok(true);
         }
-        let left_values = Projection::projection(left_tuple, &self.on_left_keys)?;
-        let right_values = Projection::projection(right_tuple, &self.on_right_keys)?;
 
-        Ok(left_values == right_values)
+        for (left_expr, right_expr) in self.on_left_keys.iter().zip(self.on_right_keys.iter()) {
+            if left_expr.eval(Some(left_tuple))? != right_expr.eval(Some(right_tuple))? {
+                return Ok(false);
+            }
+        }
+
+        Ok(true)
     }
 }
 
@@ -458,6 +461,7 @@ mod test {
     use crate::utils::lru::SharedLruCache;
     use std::collections::HashSet;
     use std::hash::RandomState;
+    use std::sync::Arc;
     use tempfile::TempDir;
 
     fn optimize_exprs(plan: LogicalPlan) -> Result<LogicalPlan, DatabaseError> {

src/execution/dql/projection.rs

@@ -20,10 +20,10 @@ use crate::planner::LogicalPlan;
 use crate::storage::Transaction;
 use crate::types::tuple::Tuple;
 use crate::types::value::DataValue;
+
 pub struct Projection {
     exprs: Vec<ScalarExpression>,
     input: ExecId,
-    scratch: Tuple,
 }
 
 impl<'a, T: Transaction + 'a> ExecutorNode<'a, T> for Projection {
@@ -36,11 +36,7 @@ impl<'a, T: Transaction + 'a> ExecutorNode<'a, T> for Projection {
         transaction: *mut T,
     ) -> ExecId {
         let input = build_read(arena, input, cache, transaction);
-        arena.push(ExecNode::Projection(Projection {
-            exprs,
-            input,
-            scratch: Tuple::default(),
-        }))
+        arena.push(ExecNode::Projection(Projection { exprs, input }))
     }
 
     fn next_tuple(&mut self, arena: &mut ExecArena<'a, T>) -> Result<(), DatabaseError> {
@@ -49,15 +45,14 @@ impl<'a, T: Transaction + 'a> ExecutorNode<'a, T> for Projection {
             return Ok(());
         }
 
-        std::mem::swap(&mut self.scratch, arena.result_tuple_mut());
-        let tuple = &self.scratch;
-        let output = arena.result_tuple_mut();
-        output.pk.clone_from(&tuple.pk);
-        output.values.clear();
-        output.values.reserve(self.exprs.len());
-        for expr in self.exprs.iter() {
-            output.values.push(expr.eval(Some(tuple))?);
-        }
+        arena.with_projection_tmp(|tuple, projection_tmp| {
+            projection_tmp.clear();
+            projection_tmp.reserve(self.exprs.len());
+            for expr in self.exprs.iter() {
+                projection_tmp.push(expr.eval(Some(tuple))?);
+            }
+            Ok::<_, DatabaseError>(())
+        })?;
         arena.resume();
         Ok(())
     }

src/execution/mod.rs

@@ -63,6 +63,7 @@ use crate::planner::LogicalPlan;
 use crate::storage::{StatisticsMetaCache, TableCache, Transaction, ViewCache};
 use crate::types::index::RuntimeIndexProbe;
 use crate::types::tuple::Tuple;
+use crate::types::value::DataValue;
 
 pub(crate) type ExecutionCaches<'a> = (&'a TableCache, &'a ViewCache, &'a StatisticsMetaCache);
 pub(crate) type ExecId = usize;
@@ -264,6 +265,7 @@ impl<'a, T: Transaction + 'a> ExecNode<'a, T> {
 pub(crate) struct ExecArena<'a, T: Transaction + 'a> {
     nodes: Vec<ExecNode<'a, T>>,
     result: ExecResult,
+    projection_tmp: Vec<DataValue>,
     cache: Option<ExecutionCaches<'a>>,
     transaction: *mut T,
     runtime_probe_stack: Vec<RuntimeIndexProbe>,
@@ -274,6 +276,7 @@ impl<'a, T: Transaction + 'a> Default for ExecArena<'a, T> {
         Self {
             nodes: Vec::new(),
             result: ExecResult::default(),
+            projection_tmp: Vec::new(),
             cache: None,
             transaction: std::ptr::null_mut(),
             runtime_probe_stack: Vec::new(),
@@ -344,6 +347,21 @@ impl<'a, T: Transaction + 'a> ExecArena<'a, T> {
         &mut self.result.tuple
     }
 
+    #[inline]
+    pub(crate) fn with_projection_tmp<R, E>(
+        &mut self,
+        f: impl FnOnce(&Tuple, &mut Vec<DataValue>) -> Result<R, E>,
+    ) -> Result<R, E> {
+        let ExecArena {
+            result,
+            projection_tmp,
+            ..
+        } = self;
+        let ret = f(&result.tuple, projection_tmp)?;
+        std::mem::swap(&mut result.tuple.values, projection_tmp);
+        Ok(ret)
+    }
+
     #[inline]
     pub(crate) fn resume(&mut self) {
         self.result.status = Some(ExecStatus::Continue);