perf: Wrap ProcessorRecord::lifetime in box to avoid occupying memo…

…ry when it's `None`

dozer-core/src/processor_record.rs

@@ -11,7 +11,7 @@ pub struct ProcessorRecord {
     total_len: u32,
 
     /// Time To Live for this record. If the value is None, the record will never expire.
-    pub lifetime: Option<Lifetime>,
+    lifetime: Option<Box<Lifetime>>,
 
     // Imagine that we flatten all the fields in `values` recursively, `index` is the index into the flattened vector.
     index: Vec<u32>,
@@ -69,10 +69,10 @@ impl ProcessorRecord {
     }
 
     pub fn get_lifetime(&self) -> Option<Lifetime> {
-        self.lifetime.clone()
+        self.lifetime.as_ref().map(|lifetime| *lifetime.clone())
     }
     pub fn set_lifetime(&mut self, lifetime: Option<Lifetime>) {
-        self.lifetime = lifetime;
+        self.lifetime = lifetime.map(Box::new);
     }
 
     pub fn get_field_indexes(&self) -> &[u32] {

dozer-sql/src/pipeline/product/join/operator.rs

@@ -467,7 +467,7 @@ impl JoinOperator {
 
                 add_join_record(&mut self.left_map, join_key, primary_key, &new);
 
-                if let Some(lifetime) = new.get_record().lifetime.clone() {
+                if let Some(lifetime) = new.get_record().get_lifetime() {
                     self.insert_evict_index(from, lifetime, join_key, primary_key)?
                 }
 
@@ -480,7 +480,7 @@ impl JoinOperator {
 
                 add_join_record(&mut self.right_map, join_key, primary_key, &new);
 
-                if let Some(lifetime) = new.get_record().lifetime.clone() {
+                if let Some(lifetime) = new.get_record().get_lifetime() {
                     self.insert_evict_index(from, lifetime, join_key, primary_key)?
                 }
 
@@ -494,7 +494,7 @@ impl JoinOperator {
 
                 add_join_record(&mut self.left_map, join_key, primary_key, &new);
 
-                if let Some(lifetime) = new.get_record().lifetime.clone() {
+                if let Some(lifetime) = new.get_record().get_lifetime() {
                     self.insert_evict_index(from, lifetime, join_key, primary_key)?
                 }
 
@@ -508,7 +508,7 @@ impl JoinOperator {
 
                 add_join_record(&mut self.right_map, join_key, primary_key, &new);
 
-                if let Some(lifetime) = new.get_record().lifetime.clone() {
+                if let Some(lifetime) = new.get_record().get_lifetime() {
                     self.insert_evict_index(from, lifetime, join_key, primary_key)?
                 }
 
@@ -522,7 +522,7 @@ impl JoinOperator {
 
                 add_join_record(&mut self.left_map, join_key, primary_key, &new);
 
-                if let Some(lifetime) = new.get_record().lifetime.clone() {
+                if let Some(lifetime) = new.get_record().get_lifetime() {
                     self.insert_evict_index(from, lifetime, join_key, primary_key)?
                 }
 
@@ -536,7 +536,7 @@ impl JoinOperator {
 
                 add_join_record(&mut self.right_map, join_key, primary_key, &new);
 
-                if let Some(lifetime) = new.get_record().lifetime.clone() {
+                if let Some(lifetime) = new.get_record().get_lifetime() {
                     self.insert_evict_index(from, lifetime, join_key, primary_key)?
                 }
 
@@ -619,8 +619,8 @@ fn join_records(
     left_record: ProcessorRecordRef,
     right_record: ProcessorRecordRef,
 ) -> ProcessorRecordRef {
-    let left_lifetime = left_record.get_record().lifetime.clone();
-    let right_lifetime = right_record.get_record().lifetime.clone();
+    let left_lifetime = left_record.get_record().get_lifetime();
+    let right_lifetime = right_record.get_record().get_lifetime();
 
     let mut output_record = ProcessorRecord::new();
     output_record.extend_referenced_record(left_record);

dozer-sql/src/pipeline/product/join/processor.rs

@@ -5,6 +5,7 @@ use dozer_core::node::{PortHandle, Processor};
 use dozer_core::DEFAULT_PORT_HANDLE;
 use dozer_types::errors::internal::BoxedError;
 use dozer_types::labels::Labels;
+use dozer_types::types::Lifetime;
 use metrics::{
     counter, describe_counter, describe_gauge, describe_histogram, gauge, histogram,
     increment_counter,
@@ -60,7 +61,7 @@ impl ProductProcessor {
         }
     }
 
-    fn update_eviction_index(&mut self, lifetime: &dozer_types::types::Lifetime) {
+    fn update_eviction_index(&mut self, lifetime: Lifetime) {
         let now = &lifetime.reference;
         let old_instants = self.join_operator.evict_index(&JoinBranch::Left, now);
         self.join_operator
@@ -91,7 +92,7 @@ impl Processor for ProductProcessor {
         let now = std::time::Instant::now();
         let records = match op {
             ProcessorOperation::Delete { old } => {
-                if let Some(lifetime) = &old.get_record().lifetime {
+                if let Some(lifetime) = old.get_record().get_lifetime() {
                     self.update_eviction_index(lifetime);
                 }
 
@@ -100,7 +101,7 @@ impl Processor for ProductProcessor {
                     .map_err(PipelineError::JoinError)?
             }
             ProcessorOperation::Insert { new } => {
-                if let Some(lifetime) = &new.get_record().lifetime {
+                if let Some(lifetime) = new.get_record().get_lifetime() {
                     self.update_eviction_index(lifetime);
                 }
 
@@ -109,7 +110,7 @@ impl Processor for ProductProcessor {
                     .map_err(PipelineError::JoinError)?
             }
             ProcessorOperation::Update { old, new } => {
-                if let Some(lifetime) = &old.get_record().lifetime {
+                if let Some(lifetime) = old.get_record().get_lifetime() {
                     self.update_eviction_index(lifetime);
                 }
 

dozer-sql/src/pipeline/projection/processor.rs

@@ -31,7 +31,7 @@ impl ProjectionProcessor {
                 .extend_direct_field(expr.evaluate(record.get_record(), &self.input_schema)?);
         }
 
-        output_record.set_lifetime(record.get_record().lifetime.to_owned());
+        output_record.set_lifetime(record.get_record().get_lifetime());
 
         Ok(ProcessorOperation::Delete {
             old: ProcessorRecordRef::new(output_record),
@@ -46,7 +46,7 @@ impl ProjectionProcessor {
                 .extend_direct_field(expr.evaluate(record.get_record(), &self.input_schema)?);
         }
 
-        output_record.set_lifetime(record.get_record().lifetime.to_owned());
+        output_record.set_lifetime(record.get_record().get_lifetime());
         Ok(ProcessorOperation::Insert {
             new: ProcessorRecordRef::new(output_record),
         })
@@ -66,9 +66,9 @@ impl ProjectionProcessor {
                 .extend_direct_field(expr.evaluate(new.get_record(), &self.input_schema)?);
         }
 
-        old_output_record.set_lifetime(old.get_record().lifetime.to_owned());
+        old_output_record.set_lifetime(old.get_record().get_lifetime());
 
-        new_output_record.set_lifetime(new.get_record().lifetime.to_owned());
+        new_output_record.set_lifetime(new.get_record().get_lifetime());
         Ok(ProcessorOperation::Update {
             old: ProcessorRecordRef::new(old_output_record),
             new: ProcessorRecordRef::new(new_output_record),