Plubming EvaluatorCache through the evaluator.

Author: badmonster0

src/execution/evaluator.rs

@@ -11,7 +11,7 @@ use crate::{
     utils::immutable::RefList,
 };
 
-use super::memoization::{EvaluationCache, MemoizationInfo};
+use super::memoization::EvaluationCache;
 
 #[derive(Debug)]
 pub struct ScopeValueBuilder {
@@ -285,22 +285,19 @@ fn assemble_input_values(
         .collect()
 }
 
-struct CacheEntry {
-    value: serde_json::Value,
-    used: bool,
-}
-
 async fn evaluate_child_op_scope(
     op_scope: &AnalyzedOpScope,
     scoped_entries: RefList<'_, &ScopeEntry<'_>>,
     child_scope_entry: ScopeEntry<'_>,
+    cache: Option<&EvaluationCache>,
 ) -> Result<()> {
-    evaluate_op_scope(op_scope, scoped_entries.prepend(&child_scope_entry)).await
+    evaluate_op_scope(op_scope, scoped_entries.prepend(&child_scope_entry), cache).await
 }
 
 async fn evaluate_op_scope(
     op_scope: &AnalyzedOpScope,
     scoped_entries: RefList<'_, &ScopeEntry<'_>>,
+    cache: Option<&EvaluationCache>,
 ) -> Result<()> {
     let head_scope = *scoped_entries.head().unwrap();
     for reactive_op in op_scope.reactive_ops.iter() {
@@ -331,6 +328,7 @@ async fn evaluate_op_scope(
                                     value: &item,
                                     schema: &collection_schema.row,
                                 },
+                                cache,
                             )
                         })
                         .collect::<Vec<_>>(),
@@ -345,6 +343,7 @@ async fn evaluate_op_scope(
                                     value: v,
                                     schema: &collection_schema.row,
                                 },
+                                cache,
                             )
                         })
                         .collect::<Vec<_>>(),
@@ -360,6 +359,7 @@ async fn evaluate_op_scope(
                                     value: item,
                                     schema: &collection_schema.row,
                                 },
+                                cache,
                             )
                         })
                         .collect::<Vec<_>>(),
@@ -395,8 +395,8 @@ pub async fn evaluate_source_entry<'a>(
     source_op_idx: u32,
     schema: &schema::DataSchema,
     key: &value::KeyValue,
-    memoization_info: Option<MemoizationInfo>,
-) -> Result<Option<(ScopeValueBuilder, MemoizationInfo)>> {
+    cache: Option<&EvaluationCache>,
+) -> Result<Option<ScopeValueBuilder>> {
     let root_schema = &schema.schema;
     let root_scope_value =
         ScopeValueBuilder::new(root_schema.fields.len(), schema.collectors.len());
@@ -419,23 +419,20 @@ pub async fn evaluate_source_entry<'a>(
 
     let result = match source_op.executor.get_value(&key).await? {
         Some(val) => {
-            let cache = memoization_info
-                .map(|info| EvaluationCache::from_stored(info.cache))
-                .unwrap_or_default();
             let scope_value =
                 ScopeValueBuilder::augmented_from(value::ScopeValue(val), &collection_schema)?;
             root_scope_entry.define_field_w_builder(
                 &source_op.output,
                 value::Value::Table(BTreeMap::from([(key.clone(), scope_value)])),
             );
 
-            evaluate_op_scope(&plan.op_scope, RefList::Nil.prepend(&root_scope_entry)).await?;
-            Some((
-                root_scope_value,
-                MemoizationInfo {
-                    cache: cache.into_stored()?,
-                },
-            ))
+            evaluate_op_scope(
+                &plan.op_scope,
+                RefList::Nil.prepend(&root_scope_entry),
+                cache,
+            )
+            .await?;
+            Some(root_scope_value)
         }
         None => None,
     };
@@ -468,6 +465,7 @@ pub async fn evaluate_transient_flow(
     evaluate_op_scope(
         &flow.execution_plan.op_scope,
         RefList::Nil.prepend(&root_scope_entry),
+        None,
     )
     .await?;
     let output_value = assemble_value(

src/execution/indexer.rs

@@ -10,7 +10,7 @@ use sqlx::PgPool;
 
 use super::db_tracking::{self, read_source_tracking_info};
 use super::db_tracking_setup;
-use super::memoization::MemoizationInfo;
+use super::memoization::{EvaluationCache, MemoizationInfo};
 use crate::base::schema;
 use crate::base::spec::FlowInstanceSpec;
 use crate::base::value::{self, FieldValues, KeyValue};
@@ -446,28 +446,34 @@ pub async fn update_source_entry<'a>(
     let memoization_info = existing_tracking_info
         .map(|info| info.memoization_info.map(|info| info.0))
         .flatten();
-    let evaluate_output =
-        evaluate_source_entry(plan, source_op_idx, schema, key, memoization_info).await?;
+    let evaluation_cache = memoization_info
+        .map(|info| EvaluationCache::from_stored(info.cache))
+        .unwrap_or_default();
+    let value_builder =
+        evaluate_source_entry(plan, source_op_idx, schema, key, Some(&evaluation_cache)).await?;
 
     // Didn't exist and still doesn't exist. No need to apply any changes.
-    if !already_exists && evaluate_output.is_none() {
+    if !already_exists && value_builder.is_none() {
         return Ok(());
     }
 
-    let (source_ordinal, precommit_data) = match &evaluate_output {
-        Some((scope_value, memoization_info)) => {
-            // TODO: Generate the actual source ordinal.
+    let memoization_info = MemoizationInfo {
+        cache: evaluation_cache.into_stored()?,
+    };
+    let (source_ordinal, precommit_data) = match &value_builder {
+        Some(scope_value) => {
             (
+                // TODO: Generate the actual source ordinal.
                 Some(1),
                 Some(PrecommitData {
                     scope_value,
-                    memoization_info,
+                    memoization_info: &memoization_info,
                 }),
             )
         }
         None => (None, None),
     };
-    if evaluate_output.is_some() {
+    if value_builder.is_some() {
         Some(1)
     } else {
         None

src/service/flows.rs

@@ -144,7 +144,7 @@ pub async fn evaluate_data(
         .ok_or_else(|| api_error!("field {} does not have a key", query.field))?;
     let key = value::KeyValue::from_strs(query.key, &key_field.value_type.typ)?;
 
-    let (data_builder, _) =
+    let data_builder =
         evaluator::evaluate_source_entry(&execution_plan, field_idx as u32, &schema, &key, None)
             .await?
             .ok_or_else(|| {