feat(fast-fp): add process_logic_fingerprint for tracking table

Author: georgeh0

src/builder/exec_ctx.rs

@@ -288,6 +288,9 @@ pub fn build_flow_setup_execution_context(
                             db_tracking_setup::default_source_state_table_name(&flow_inst.name)
                         })
                 }),
+            has_fast_fingerprint_column: metadata
+                .features
+                .contains(setup::flow_features::FAST_FINGERPRINT),
         },
         targets: target_states,
         metadata,

src/execution/db_tracking_setup.rs

@@ -22,12 +22,16 @@ pub const CURRENT_TRACKING_TABLE_VERSION: i32 = 1;
 
 async fn upgrade_tracking_table(
     pool: &PgPool,
-    table_name: &str,
+    desired_state: &TrackingTableSetupState,
     existing_version_id: i32,
-    target_version_id: i32,
 ) -> Result<()> {
-    if existing_version_id < 1 && target_version_id >= 1 {
-        let query = format!(
+    if existing_version_id < 1 && desired_state.version_id >= 1 {
+        let table_name = &desired_state.table_name;
+        let opt_fast_fingerprint_column = desired_state
+            .has_fast_fingerprint_column
+            .then(|| "processed_source_fp BYTEA,")
+            .unwrap_or("");
+        let query =  format!(
             "CREATE TABLE IF NOT EXISTS {table_name} (
                 source_id INTEGER NOT NULL,
                 source_key JSONB NOT NULL,
@@ -39,6 +43,7 @@ async fn upgrade_tracking_table(
 
                 -- Update after applying the changes to the target storage.
                 processed_source_ordinal BIGINT,
+                {opt_fast_fingerprint_column}
                 process_logic_fingerprint BYTEA,
                 process_ordinal BIGINT,
                 process_time_micros BIGINT,
@@ -73,6 +78,8 @@ pub struct TrackingTableSetupState {
     pub version_id: i32,
     #[serde(default)]
     pub source_state_table_name: Option<String>,
+    #[serde(default)]
+    pub has_fast_fingerprint_column: bool,
 }
 
 #[derive(Debug)]
@@ -248,13 +255,8 @@ impl TrackingTableSetupChange {
             }
 
             if self.min_existing_version_id != Some(desired.version_id) {
-                upgrade_tracking_table(
-                    pool,
-                    &desired.table_name,
-                    self.min_existing_version_id.unwrap_or(0),
-                    desired.version_id,
-                )
-                .await?;
+                upgrade_tracking_table(pool, desired, self.min_existing_version_id.unwrap_or(0))
+                    .await?;
             }
         } else {
             for lagacy_name in self.legacy_tracking_table_names.iter() {

src/setup/flow_features.rs

@@ -1,7 +1,8 @@
 use crate::prelude::*;
 
 pub const SOURCE_STATE_TABLE: &str = "source_state_table";
+pub const FAST_FINGERPRINT: &str = "fast_fingerprint";
 
 pub fn default_features() -> BTreeSet<String> {
-    BTreeSet::new()
+    BTreeSet::from_iter([FAST_FINGERPRINT.to_string()])
 }