feat: add content hash support for change detection in source processing (#629)

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

* refactor: remove content hash comments from tracking and indexing structures

* style: clean up whitespace and formatting in test_content_hash.py

* refactor: add type hints to test methods in test_content_hash.py

* refactor: remove content hash handling from source executor options and streamline related logic

* implement update function for source ordinal and fingerprints, enhancing tracking efficiency

* enhance fast path logic in update_source_row to include version verification

* revert all changes for files under ops

* remove processed_source_content_hash from tracking structures and optimize update logic for source processing

* implement content hash computation in EvaluationMemory and optimize update_source_row logic for efficient source processing

* update source tracking logic to include process ordinal in database updates

* refactor: optimize update_source_tracking_ordinal_and_logic by removing process_logic_fingerprint and enhancing content hash optimization logic in update_source_row

* update EvaluationMemory initialization and streamline content hash handling in update_source_row for improved efficiency

* remove test suite for content hash functionality in test_content_hash.py to streamline test coverage

* simplify update_source_tracking_ordinal_and_logic by removing unnecessary parameters and enhance EvaluationMemory initialization for better clarity and performance

* rename update_source_tracking_ordinal_and_logic to update_source_tracking_ordinal and streamline memoization info handling in update_source_row for improved clarity

* refactor memoization info handling

Author: vumichien

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(())
 }
 

src/execution/memoization.rs

@@ -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(