Support auto UUID in StoredMemoizationInfo and EvaluationMemory.

Author: badmonster0

Cargo.toml

@@ -89,4 +89,4 @@ rustls = { version = "0.23.25" }
 http-body-util = "0.1.3"
 yaml-rust2 = "0.10.0"
 urlencoding = "2.1.3"
-uuid = "1.16.0"
+uuid = { version = "1.16.0", features = ["serde", "v4", "v8"] }

src/execution/db_tracking.rs

@@ -1,4 +1,4 @@
-use super::{db_tracking_setup::TrackingTableSetupState, memoization::MemoizationInfo};
+use super::{db_tracking_setup::TrackingTableSetupState, memoization::StoredMemoizationInfo};
 use crate::utils::{db::WriteAction, fingerprint::Fingerprint};
 use anyhow::Result;
 use sqlx::PgPool;
@@ -12,7 +12,7 @@ pub type TrackedTargetKeyForSource = Vec<(i32, Vec<TrackedTargetKey>)>;
 pub struct SourceTrackingInfo {
     pub max_process_ordinal: i64,
     pub staging_target_keys: sqlx::types::Json<TrackedTargetKeyForSource>,
-    pub memoization_info: Option<sqlx::types::Json<Option<MemoizationInfo>>>,
+    pub memoization_info: Option<sqlx::types::Json<Option<StoredMemoizationInfo>>>,
 
     pub processed_source_ordinal: Option<i64>,
     pub process_logic_fingerprint: Option<Vec<u8>>,
@@ -74,7 +74,7 @@ pub async fn precommit_source_tracking_info(
     source_key_json: &serde_json::Value,
     max_process_ordinal: i64,
     staging_target_keys: TrackedTargetKeyForSource,
-    memoization_info: Option<&MemoizationInfo>,
+    memoization_info: Option<&StoredMemoizationInfo>,
     db_setup: &TrackingTableSetupState,
     db_executor: impl sqlx::Executor<'_, Database = sqlx::Postgres>,
     action: WriteAction,

src/execution/dumper.rs

@@ -11,6 +11,7 @@ use std::path::{Path, PathBuf};
 use yaml_rust2::YamlEmitter;
 
 use super::indexer;
+use super::memoization::EvaluationMemoryOptions;
 use crate::base::{schema, value};
 use crate::builder::plan::{AnalyzedSourceOp, ExecutionPlan};
 use crate::utils::yaml_ser::YamlSerializer;
@@ -73,18 +74,16 @@ impl<'a> Dumper<'a> {
     where
         'a: 'b,
     {
-        let cache_option = if self.options.use_cache {
-            indexer::EvaluationCacheOption::UseCache(self.pool)
-        } else {
-            indexer::EvaluationCacheOption::NoCache
-        };
-
-        let data_builder = indexer::evaluate_source_entry_with_cache(
+        let data_builder = indexer::evaluate_source_entry_with_memory(
             self.plan,
             source_op,
             self.schema,
             key,
-            cache_option,
+            EvaluationMemoryOptions {
+                enable_cache: self.options.use_cache,
+                evaluation_only: true,
+            },
+            self.pool,
         )
         .await?;
 

src/execution/evaluator.rs

@@ -11,7 +11,7 @@ use crate::{
     utils::immutable::RefList,
 };
 
-use super::memoization::{evaluate_with_cell, EvaluationCache};
+use super::memoization::{evaluate_with_cell, EvaluationMemory, EvaluationMemoryOptions};
 
 #[derive(Debug)]
 pub struct ScopeValueBuilder {
@@ -292,9 +292,9 @@ async fn evaluate_child_op_scope(
     op_scope: &AnalyzedOpScope,
     scoped_entries: RefList<'_, &ScopeEntry<'_>>,
     child_scope_entry: ScopeEntry<'_>,
-    cache: Option<&EvaluationCache>,
+    memory: &EvaluationMemory,
 ) -> Result<()> {
-    evaluate_op_scope(op_scope, scoped_entries.prepend(&child_scope_entry), cache)
+    evaluate_op_scope(op_scope, scoped_entries.prepend(&child_scope_entry), memory)
         .await
         .with_context(|| {
             format!(
@@ -310,30 +310,25 @@ async fn evaluate_child_op_scope(
 async fn evaluate_op_scope(
     op_scope: &AnalyzedOpScope,
     scoped_entries: RefList<'_, &ScopeEntry<'_>>,
-    cache: Option<&EvaluationCache>,
+    memory: &EvaluationMemory,
 ) -> Result<()> {
     let head_scope = *scoped_entries.head().unwrap();
     for reactive_op in op_scope.reactive_ops.iter() {
         match reactive_op {
             AnalyzedReactiveOp::Transform(op) => {
                 let input_values = assemble_input_values(&op.inputs, scoped_entries);
-
-                let output_value_cell = match (op.function_exec_info.enable_cache, cache) {
-                    (true, Some(cache)) => {
-                        let key = op
+                let output_value_cell = memory.get_cache_entry(
+                    || {
+                        Ok(op
                             .function_exec_info
                             .fingerprinter
                             .clone()
                             .with(&input_values)?
-                            .into_fingerprint();
-                        Some(cache.get(
-                            key,
-                            &op.function_exec_info.output_type,
-                            /*ttl=*/ None,
-                        )?)
-                    }
-                    _ => None,
-                };
+                            .into_fingerprint())
+                    },
+                    &op.function_exec_info.output_type,
+                    /*ttl=*/ None,
+                )?;
                 let output_value = evaluate_with_cell(output_value_cell.as_ref(), move || {
                     op.executor.evaluate(input_values)
                 })
@@ -362,7 +357,7 @@ async fn evaluate_op_scope(
                                     value: item,
                                     schema: &collection_schema.row,
                                 },
-                                cache,
+                                memory,
                             )
                         })
                         .collect::<Vec<_>>(),
@@ -377,7 +372,7 @@ async fn evaluate_op_scope(
                                     value: v,
                                     schema: &collection_schema.row,
                                 },
-                                cache,
+                                memory,
                             )
                         })
                         .collect::<Vec<_>>(),
@@ -393,7 +388,7 @@ async fn evaluate_op_scope(
                                     value: item,
                                     schema: &collection_schema.row,
                                 },
-                                cache,
+                                memory,
                             )
                         })
                         .collect::<Vec<_>>(),
@@ -431,7 +426,7 @@ pub async fn evaluate_source_entry(
     source_op: &AnalyzedSourceOp,
     schema: &schema::DataSchema,
     key: &value::KeyValue,
-    cache: Option<&EvaluationCache>,
+    memory: &EvaluationMemory,
 ) -> Result<Option<ScopeValueBuilder>> {
     let root_schema = &schema.schema;
     let root_scope_value =
@@ -464,7 +459,7 @@ pub async fn evaluate_source_entry(
             evaluate_op_scope(
                 &plan.op_scope,
                 RefList::Nil.prepend(&root_scope_entry),
-                cache,
+                memory,
             )
             .await?;
             Some(root_scope_value)
@@ -497,10 +492,18 @@ pub async fn evaluate_transient_flow(
     for (field, value) in flow.execution_plan.input_fields.iter().zip(input_values) {
         root_scope_entry.define_field(field, value)?;
     }
+    let eval_memory = EvaluationMemory::new(
+        chrono::Utc::now(),
+        None,
+        EvaluationMemoryOptions {
+            enable_cache: false,
+            evaluation_only: true,
+        },
+    );
     evaluate_op_scope(
         &flow.execution_plan.op_scope,
         RefList::Nil.prepend(&root_scope_entry),
-        None,
+        &eval_memory,
     )
     .await?;
     let output_value = assemble_value(

src/execution/indexer.rs

@@ -9,7 +9,7 @@ use std::sync::atomic::{AtomicUsize, Ordering::Relaxed};
 
 use super::db_tracking::{self, read_source_tracking_info, TrackedTargetKey};
 use super::db_tracking_setup;
-use super::memoization::{EvaluationCache, MemoizationInfo};
+use super::memoization::{EvaluationMemory, EvaluationMemoryOptions, StoredMemoizationInfo};
 use crate::base::schema;
 use crate::base::value::{self, FieldValues, KeyValue};
 use crate::builder::plan::*;
@@ -118,7 +118,7 @@ struct TrackingInfoForTarget<'a> {
 #[derive(Debug)]
 struct PrecommitData<'a> {
     scope_value: &'a ScopeValueBuilder,
-    memoization_info: &'a MemoizationInfo,
+    memoization_info: &'a StoredMemoizationInfo,
 }
 struct PrecommitMetadata {
     source_entry_exists: bool,
@@ -417,40 +417,31 @@ async fn commit_source_tracking_info(
     Ok(WithApplyStatus::Normal(()))
 }
 
-pub enum EvaluationCacheOption<'a> {
-    NoCache,
-    UseCache(&'a PgPool),
-}
-
-pub async fn evaluate_source_entry_with_cache(
+pub async fn evaluate_source_entry_with_memory(
     plan: &ExecutionPlan,
     source_op: &AnalyzedSourceOp,
     schema: &schema::DataSchema,
     key: &value::KeyValue,
-    cache_option: EvaluationCacheOption<'_>,
+    options: EvaluationMemoryOptions,
+    pool: &PgPool,
 ) -> Result<Option<ScopeValueBuilder>> {
-    let cache = match cache_option {
-        EvaluationCacheOption::NoCache => None,
-        EvaluationCacheOption::UseCache(pool) => {
-            let source_key_json = serde_json::to_value(key)?;
-            let existing_tracking_info = read_source_tracking_info(
-                source_op.source_id,
-                &source_key_json,
-                &plan.tracking_table_setup,
-                pool,
-            )
-            .await?;
-            let process_timestamp = chrono::Utc::now();
-            let memoization_info = existing_tracking_info
-                .and_then(|info| info.memoization_info.map(|info| info.0))
-                .flatten();
-            Some(EvaluationCache::new(
-                process_timestamp,
-                memoization_info.map(|info| info.cache),
-            ))
-        }
+    let stored_info = if options.enable_cache || !options.evaluation_only {
+        let source_key_json = serde_json::to_value(key)?;
+        let existing_tracking_info = read_source_tracking_info(
+            source_op.source_id,
+            &source_key_json,
+            &plan.tracking_table_setup,
+            pool,
+        )
+        .await?;
+        existing_tracking_info
+            .and_then(|info| info.memoization_info.map(|info| info.0))
+            .flatten()
+    } else {
+        None
     };
-    let data_builder = evaluate_source_entry(plan, source_op, schema, key, cache.as_ref()).await?;
+    let memory = EvaluationMemory::new(chrono::Utc::now(), stored_info, options);
+    let data_builder = evaluate_source_entry(plan, source_op, schema, key, &memory).await?;
     Ok(data_builder)
 }
 
@@ -480,10 +471,16 @@ pub async fn update_source_entry(
     let memoization_info = existing_tracking_info
         .and_then(|info| info.memoization_info.map(|info| info.0))
         .flatten();
-    let evaluation_cache =
-        EvaluationCache::new(process_timestamp, memoization_info.map(|info| info.cache));
+    let evaluation_memory = EvaluationMemory::new(
+        process_timestamp,
+        memoization_info,
+        EvaluationMemoryOptions {
+            enable_cache: true,
+            evaluation_only: false,
+        },
+    );
     let value_builder = if !only_for_deletion {
-        evaluate_source_entry(plan, source_op, schema, key, Some(&evaluation_cache)).await?
+        evaluate_source_entry(plan, source_op, schema, key, &evaluation_memory).await?
     } else {
         None
     };
@@ -501,9 +498,7 @@ pub async fn update_source_entry(
         return Ok(());
     }
 
-    let memoization_info = MemoizationInfo {
-        cache: evaluation_cache.into_stored()?,
-    };
+    let memoization_info = evaluation_memory.into_stored()?;
     let (source_ordinal, precommit_data) = match &value_builder {
         Some(scope_value) => {
             (