feat(fast-fp): support fast collapse without reading source data

src/execution/db_tracking.rs

@@ -118,6 +118,7 @@ pub struct SourceTrackingInfoForPrecommit {
     pub staging_target_keys: sqlx::types::Json<TrackedTargetKeyForSource>,
 
     pub processed_source_ordinal: Option<i64>,
+    pub processed_source_fp: Option<Vec<u8>>,
     pub process_logic_fingerprint: Option<Vec<u8>>,
     pub process_ordinal: Option<i64>,
     pub target_keys: Option<sqlx::types::Json<TrackedTargetKeyForSource>>,
@@ -130,7 +131,12 @@ pub async fn read_source_tracking_info_for_precommit(
     db_executor: impl sqlx::Executor<'_, Database = sqlx::Postgres>,
 ) -> Result<Option<SourceTrackingInfoForPrecommit>> {
     let query_str = format!(
-        "SELECT max_process_ordinal, staging_target_keys, processed_source_ordinal, process_logic_fingerprint, process_ordinal, target_keys FROM {} WHERE source_id = $1 AND source_key = $2",
+        "SELECT max_process_ordinal, staging_target_keys, processed_source_ordinal, {}, process_logic_fingerprint, process_ordinal, target_keys FROM {} WHERE source_id = $1 AND source_key = $2",
+        if db_setup.has_fast_fingerprint_column {
+            "processed_source_fp"
+        } else {
+            "NULL::bytea AS processed_source_fp"
+        },
         db_setup.table_name
     );
     let precommit_tracking_info = sqlx::query_as(&query_str)
@@ -282,6 +288,7 @@ pub async fn delete_source_tracking_info(
 pub struct TrackedSourceKeyMetadata {
     pub source_key: serde_json::Value,
     pub processed_source_ordinal: Option<i64>,
+    pub processed_source_fp: Option<Vec<u8>>,
     pub process_logic_fingerprint: Option<Vec<u8>>,
 }
 
@@ -303,7 +310,12 @@ impl ListTrackedSourceKeyMetadataState {
         pool: &'a PgPool,
     ) -> impl Stream<Item = Result<TrackedSourceKeyMetadata, sqlx::Error>> + 'a {
         self.query_str = format!(
-            "SELECT source_key, processed_source_ordinal, process_logic_fingerprint FROM {} WHERE source_id = $1",
+            "SELECT source_key, processed_source_ordinal, {}, process_logic_fingerprint FROM {} WHERE source_id = $1",
+            if db_setup.has_fast_fingerprint_column {
+                "processed_source_fp"
+            } else {
+                "NULL::bytea AS processed_source_fp"
+            },
             db_setup.table_name
         );
         sqlx::query_as(&self.query_str).bind(source_id).fetch(pool)

src/execution/dumper.rs

@@ -193,6 +193,7 @@ impl<'a> Dumper<'a> {
 
         let mut rows_stream = import_op.executor.list(&SourceExecutorListOptions {
             include_ordinal: false,
+            include_content_version_fp: false,
         });
         while let Some(rows) = rows_stream.next().await {
             for row in rows?.into_iter() {

src/execution/indexing_status.rs

@@ -41,6 +41,7 @@ pub async fn get_source_row_indexing_status(
         &interface::SourceExecutorGetOptions {
             include_value: false,
             include_ordinal: true,
+            include_content_version_fp: false,
         },
     );
     let (last_processed, current) = try_join!(last_processed_fut, current_fut)?;

src/execution/live_updater.rs

@@ -1,5 +1,5 @@
 use crate::{
-    execution::{source_indexer::ProcessSourceKeyOptions, stats::UpdateStats},
+    execution::{source_indexer::ProcessSourceKeyInput, stats::UpdateStats},
     prelude::*,
 };
 
@@ -200,10 +200,9 @@ impl SourceUpdateTask {
                                             SharedAckFn::ack(&shared_ack_fn).await
                                         }),
                                         pool.clone(),
-                                        ProcessSourceKeyOptions {
+                                        ProcessSourceKeyInput {
                                             key_aux_info: Some(change.key_aux_info),
-                                            source_data: change.data,
-                                            ..Default::default()
+                                            data: change.data,
                                         },
                                     ));
                                 }