update source tracking logic to include process ordinal in database updates

Author: vumichien

src/execution/db_tracking.rs

@@ -312,6 +312,7 @@ pub async fn update_source_tracking_ordinal_and_logic(
     source_key_json: &serde_json::Value,
     processed_source_ordinal: Option<i64>,
     process_logic_fingerprint: &[u8],
+    process_ordinal: i64,
     process_time_micros: i64,
     db_setup: &TrackingTableSetupState,
     db_executor: impl sqlx::Executor<'_, Database = sqlx::Postgres>,
@@ -320,7 +321,8 @@ pub async fn update_source_tracking_ordinal_and_logic(
         "UPDATE {} SET \
          processed_source_ordinal = $3, \
          process_logic_fingerprint = $4, \
-         process_time_micros = $5 \
+         process_ordinal = $5, \
+         process_time_micros = $6 \
          WHERE source_id = $1 AND source_key = $2",
         db_setup.table_name
     );
@@ -329,7 +331,8 @@ pub async fn update_source_tracking_ordinal_and_logic(
         .bind(source_key_json) // $2
         .bind(processed_source_ordinal) // $3
         .bind(process_logic_fingerprint) // $4
-        .bind(process_time_micros) // $5
+        .bind(process_ordinal) // $5
+        .bind(process_time_micros) // $6
         .execute(db_executor)
         .await?;
     Ok(())

src/execution/row_indexer.rs

@@ -634,11 +634,15 @@ pub async fn update_source_row(
                             .and_then(|info| info.process_ordinal);
                         if current_info.process_ordinal == original_process_ordinal {
                             // Safe to apply optimization - just update tracking table
+                            let new_process_ordinal = (current_info.max_process_ordinal + 1)
+                                .max(process_time.timestamp_millis());
+
                             db_tracking::update_source_tracking_ordinal_and_logic(
                                 src_eval_ctx.import_op.source_id,
                                 &source_key_json,
                                 source_version.ordinal.0,
                                 &src_eval_ctx.plan.logic_fingerprint.0,
+                                new_process_ordinal,
                                 process_time.timestamp_micros(),
                                 &src_eval_ctx.plan.tracking_table_setup,
                                 &mut *txn,