Extract the logic to evaluate a source entry into a function for reuse.

Author: badmonster0

src/execution/indexer.rs

@@ -418,13 +418,13 @@ async fn commit_source_tracking_info(
     Ok(WithApplyStatus::Normal(()))
 }
 
-/// Built an evaluation cache on the existing data.
-pub async fn evaluation_cache_on_existing_data(
+pub async fn evaluate_source_entry_with_cache(
     plan: &ExecutionPlan,
     source_op_idx: usize,
+    schema: &schema::DataSchema,
     key: &value::KeyValue,
     pool: &PgPool,
-) -> Result<EvaluationCache> {
+) -> Result<Option<value::ScopeValue>> {
     let source_id = plan.source_ops[source_op_idx].source_id;
     let source_key_json = serde_json::to_value(key)?;
     let existing_tracking_info = read_source_tracking_info(
@@ -438,10 +438,11 @@ pub async fn evaluation_cache_on_existing_data(
     let memoization_info = existing_tracking_info
         .and_then(|info| info.memoization_info.map(|info| info.0))
         .flatten();
-    Ok(EvaluationCache::new(
-        process_timestamp,
-        memoization_info.map(|info| info.cache),
-    ))
+    let evaluation_cache =
+        EvaluationCache::new(process_timestamp, memoization_info.map(|info| info.cache));
+    let data_builder =
+        evaluate_source_entry(plan, source_op_idx, schema, key, Some(&evaluation_cache)).await?;
+    Ok(data_builder.map(|builder| builder.into()))
 }
 
 pub async fn update_source_entry(

src/service/flows.rs

@@ -13,7 +13,7 @@ use super::error::ApiError;
 use crate::{
     api_bail, api_error,
     base::{schema, spec},
-    execution::{evaluator, indexer},
+    execution::indexer,
 };
 use crate::{execution::indexer::IndexUpdateInfo, lib_context::LibContext};
 
@@ -145,26 +145,19 @@ 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 evaluation_cache = indexer::evaluation_cache_on_existing_data(
-        &execution_plan,
-        source_op_idx,
-        &key,
-        &lib_context.pool,
-    )
-    .await?;
-    let data_builder = evaluator::evaluate_source_entry(
+    let data = indexer::evaluate_source_entry_with_cache(
         &execution_plan,
         source_op_idx,
         schema,
         &key,
-        Some(&evaluation_cache),
+        &lib_context.pool,
     )
     .await?
     .ok_or_else(|| api_error!("value not found for source at the specified key: {key:?}"))?;
 
     Ok(Json(EvaluateDataResponse {
         schema: schema.clone(),
-        data: data_builder.into(),
+        data,
     }))
 }