feat: add content hash support for change detection in source processing

src/execution/db_tracking.rs

@@ -13,7 +13,6 @@ pub struct TrackedTargetKeyInfo {
     pub key: serde_json::Value,
     pub additional_key: serde_json::Value,
     pub process_ordinal: i64,
-    // None means deletion.
     pub fingerprint: Option<Fingerprint>,
 }
 
@@ -84,6 +83,8 @@ pub struct SourceTrackingInfoForProcessing {
 
     pub processed_source_ordinal: Option<i64>,
     pub process_logic_fingerprint: Option<Vec<u8>>,
+    pub max_process_ordinal: Option<i64>,
+    pub process_ordinal: Option<i64>,
 }
 
 pub async fn read_source_tracking_info_for_processing(
@@ -93,7 +94,7 @@ pub async fn read_source_tracking_info_for_processing(
     pool: &PgPool,
 ) -> Result<Option<SourceTrackingInfoForProcessing>> {
     let query_str = format!(
-        "SELECT memoization_info, processed_source_ordinal, process_logic_fingerprint FROM {} WHERE source_id = $1 AND source_key = $2",
+        "SELECT memoization_info, processed_source_ordinal, process_logic_fingerprint, max_process_ordinal, process_ordinal FROM {} WHERE source_id = $1 AND source_key = $2",
         db_setup.table_name
     );
     let tracking_info = sqlx::query_as(&query_str)
@@ -305,3 +306,23 @@ pub async fn read_source_last_processed_info(
         .await?;
     Ok(last_processed_info)
 }
+
+pub async fn update_source_tracking_ordinal(
+    source_id: i32,
+    source_key_json: &serde_json::Value,
+    processed_source_ordinal: Option<i64>,
+    db_setup: &TrackingTableSetupState,
+    db_executor: impl sqlx::Executor<'_, Database = sqlx::Postgres>,
+) -> Result<()> {
+    let query_str = format!(
+        "UPDATE {} SET processed_source_ordinal = $3 WHERE source_id = $1 AND source_key = $2",
+        db_setup.table_name
+    );
+    sqlx::query(&query_str)
+        .bind(source_id) // $1
+        .bind(source_key_json) // $2
+        .bind(processed_source_ordinal) // $3
+        .execute(db_executor)
+        .await?;
+    Ok(())
+}

src/execution/db_tracking_setup.rs

@@ -42,6 +42,7 @@ async fn upgrade_tracking_table(
         );
         sqlx::query(&query).execute(pool).await?;
     }
+
     Ok(())
 }
 
@@ -154,8 +155,6 @@ impl ResourceSetupStatus for TrackingTableSetupStatus {
             (None, None) => SetupChangeType::NoChange,
         }
     }
-
-
 }
 
 impl TrackingTableSetupStatus {

src/execution/dumper.rs

@@ -1,6 +1,6 @@
 use anyhow::Result;
-use futures::future::try_join_all;
 use futures::StreamExt;
+use futures::future::try_join_all;
 use indexmap::IndexMap;
 use itertools::Itertools;
 use serde::ser::SerializeSeq;

src/execution/memoization.rs

@@ -1,4 +1,4 @@
-use anyhow::{bail, Result};
+use anyhow::{Result, bail};
 use serde::{Deserialize, Serialize};
 use std::{
     borrow::Cow,
@@ -25,6 +25,9 @@ pub struct StoredMemoizationInfo {
 
     #[serde(default, skip_serializing_if = "HashMap::is_empty")]
     pub uuids: HashMap<Fingerprint, Vec<uuid::Uuid>>,
+
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub content_hash: Option<Fingerprint>,
 }
 
 pub type CacheEntryCell = Arc<tokio::sync::OnceCell<Result<value::Value, SharedError>>>;
@@ -156,7 +159,11 @@ impl EvaluationMemory {
             .into_iter()
             .filter_map(|(k, v)| v.into_stored().map(|uuids| (k, uuids)))
             .collect();
-        Ok(StoredMemoizationInfo { cache, uuids })
+        Ok(StoredMemoizationInfo {
+            cache,
+            uuids,
+            content_hash: None,
+        })
     }
 
     pub fn get_cache_entry(