feat: Implement batch processing for stateless processors

dozer-sql/src/projection/processor.rs

@@ -48,7 +48,7 @@ impl ProjectionProcessor {
         Ok(Operation::Delete { old: output_record })
     }
 
-    fn insert(&mut self, record: &Record) -> Result<Operation, PipelineError> {
+    fn insert(&mut self, record: &Record) -> Result<Record, PipelineError> {
         let mut results = vec![];
 
         for expr in &mut self.expressions {
@@ -57,7 +57,7 @@ impl ProjectionProcessor {
 
         let mut output_record = Record::new(results);
         output_record.set_lifetime(record.lifetime.to_owned());
-        Ok(Operation::Insert { new: output_record })
+        Ok(output_record)
     }
 
     fn update(&mut self, old: &Record, new: &Record) -> Result<Operation, PipelineError> {
@@ -90,18 +90,16 @@ impl Processor for ProjectionProcessor {
     ) -> Result<(), BoxedError> {
         let output_op = match op {
             Operation::Delete { ref old } => self.delete(old)?,
-            Operation::Insert { ref new } => self.insert(new)?,
+            Operation::Insert { ref new } => Operation::Insert {
+                new: self.insert(new)?,
+            },
             Operation::Update { ref old, ref new } => self.update(old, new)?,
             Operation::BatchInsert { new } => {
-                for record in new {
-                    self.process(
-                        _from_port,
-                        _record_store,
-                        Operation::Insert { new: record },
-                        fw,
-                    )?;
-                }
-                return Ok(());
+                let records = new
+                    .iter()
+                    .map(|record| self.insert(record))
+                    .collect::<Result<Vec<_>, _>>()?;
+                Operation::BatchInsert { new: records }
             }
         };
         fw.send(output_op, DEFAULT_PORT_HANDLE);

dozer-sql/src/selection/processor.rs

@@ -7,7 +7,7 @@ use dozer_core::DEFAULT_PORT_HANDLE;
 use dozer_recordstore::ProcessorRecordStore;
 use dozer_sql_expression::execution::Expression;
 use dozer_types::errors::internal::BoxedError;
-use dozer_types::types::{Field, Operation, Schema};
+use dozer_types::types::{Field, Operation, Record, Schema};
 
 use crate::errors::PipelineError;
 
@@ -33,6 +33,10 @@ impl SelectionProcessor {
             expression,
         })
     }
+
+    fn filter(&mut self, record: &Record) -> Result<bool, PipelineError> {
+        Ok(self.expression.evaluate(record, &self.input_schema)? == Field::Boolean(true))
+    }
 }
 
 impl Processor for SelectionProcessor {
@@ -49,20 +53,18 @@ impl Processor for SelectionProcessor {
     ) -> Result<(), BoxedError> {
         match op {
             Operation::Delete { ref old } => {
-                if self.expression.evaluate(old, &self.input_schema)? == Field::Boolean(true) {
+                if self.filter(old)? {
                     fw.send(op, DEFAULT_PORT_HANDLE);
                 }
             }
             Operation::Insert { ref new } => {
-                if self.expression.evaluate(new, &self.input_schema)? == Field::Boolean(true) {
+                if self.filter(new)? {
                     fw.send(op, DEFAULT_PORT_HANDLE);
                 }
             }
             Operation::Update { old, new } => {
-                let old_fulfilled =
-                    self.expression.evaluate(&old, &self.input_schema)? == Field::Boolean(true);
-                let new_fulfilled =
-                    self.expression.evaluate(&new, &self.input_schema)? == Field::Boolean(true);
+                let old_fulfilled = self.filter(&old)?;
+                let new_fulfilled = self.filter(&new)?;
                 match (old_fulfilled, new_fulfilled) {
                     (true, true) => {
                         // both records fulfills the WHERE condition, forward the operation
@@ -82,13 +84,16 @@ impl Processor for SelectionProcessor {
                 }
             }
             Operation::BatchInsert { new } => {
-                for record in new {
-                    self.process(
-                        _from_port,
-                        _record_store,
-                        Operation::Insert { new: record },
-                        fw,
-                    )?;
+                let records = new
+                    .into_iter()
+                    .filter_map(|record| {
+                        self.filter(&record)
+                            .map(|fulfilled| if fulfilled { Some(record) } else { None })
+                            .transpose()
+                    })
+                    .collect::<Result<Vec<_>, _>>()?;
+                if !records.is_empty() {
+                    fw.send(Operation::BatchInsert { new: records }, DEFAULT_PORT_HANDLE);
                 }
             }
         }

dozer-sql/src/table_operator/processor.rs

@@ -82,14 +82,15 @@ impl Processor for TableOperatorProcessor {
                 }
             }
             Operation::BatchInsert { new } => {
+                let mut records = vec![];
                 for record in new {
-                    self.process(
-                        _from_port,
-                        record_store,
-                        Operation::Insert { new: record },
-                        fw,
-                    )?;
+                    records.extend(
+                        self.operator
+                            .execute(record_store, &record, &self.input_schema)
+                            .map_err(PipelineError::TableOperatorError)?,
+                    );
                 }
+                fw.send(Operation::BatchInsert { new: records }, DEFAULT_PORT_HANDLE);
             }
         }
         Ok(())

dozer-sql/src/window/processor.rs

@@ -69,14 +69,15 @@ impl Processor for WindowProcessor {
                 )?;
             }
             Operation::BatchInsert { new } => {
+                let mut records = vec![];
                 for record in new {
-                    self.process(
-                        _from_port,
-                        record_store,
-                        Operation::Insert { new: record },
-                        fw,
-                    )?;
+                    records.extend(
+                        self.window
+                            .execute(record_store, record)
+                            .map_err(PipelineError::WindowError)?,
+                    );
                 }
+                fw.send(Operation::BatchInsert { new: records }, DEFAULT_PORT_HANDLE);
             }
         }
         Ok(())