feat(fine-grain-fp): use fine grain data lineage to detect logic changes (#1292)

* feat(fine-grain-fp): use fine grain data lineage to detect logic changes

* fix: incorporate target schema id, and use tuple for encoding

* minor consolidate

Author: georgeh0

rust/cocoindex/src/builder/analyzer.rs

@@ -1,7 +1,9 @@
 use crate::{
-    base::schema::EnrichedValueType, prelude::*, py::Pythonized, setup::ObjectSetupChange,
+    base::schema::EnrichedValueType, builder::plan::FieldDefFingerprint, prelude::*,
+    py::Pythonized, setup::ObjectSetupChange,
 };
 
+use cocoindex_utils::fingerprint::Fingerprinter;
 use pyo3::{exceptions::PyException, prelude::*};
 use pyo3_async_runtimes::tokio::future_into_py;
 use std::{collections::btree_map, ops::Deref};
@@ -119,8 +121,8 @@ impl DataSlice {
             spec::ValueMapping::Field(spec::FieldMapping { scope, field_path }) => {
                 let data_scope_builder = self.scope.data.lock().unwrap();
                 let struct_schema = {
-                    let (_, val_type) = data_scope_builder
-                        .analyze_field_path(field_path)
+                    let (_, val_type, _) = data_scope_builder
+                        .analyze_field_path(field_path, self.scope.base_value_def_fp.clone())
                         .into_py_result()?;
                     match &val_type.typ {
                         ValueTypeBuilder::Struct(struct_type) => struct_type,
@@ -171,7 +173,8 @@ impl DataSlice {
             spec::ValueMapping::Constant(c) => c.schema.clone(),
             spec::ValueMapping::Field(v) => {
                 let data_scope_builder = self.scope.data.lock().unwrap();
-                let (_, val_type) = data_scope_builder.analyze_field_path(&v.field_path)?;
+                let (_, val_type, _) = data_scope_builder
+                    .analyze_field_path(&v.field_path, self.scope.base_value_def_fp.clone())?;
                 EnrichedValueType::from_alternative(val_type)?
             }
         };
@@ -257,6 +260,7 @@ impl FlowBuilder {
             spec::ROOT_SCOPE_NAME.to_string(),
             None,
             Arc::new(Mutex::new(DataScopeBuilder::new())),
+            FieldDefFingerprint::default(),
         );
         let flow_inst_context = build_flow_instance_context(
             name,
@@ -366,7 +370,19 @@ impl FlowBuilder {
         {
             let mut root_data_scope = self.root_op_scope.data.lock().unwrap();
             root_data_scope
-                .add_field(name.clone(), &value_type)
+                .add_field(
+                    name.clone(),
+                    &value_type,
+                    FieldDefFingerprint {
+                        source_op_names: HashSet::from([name.clone()]),
+                        fingerprint: Fingerprinter::default()
+                            .with("input")
+                            .into_py_result()?
+                            .with(&name)
+                            .into_py_result()?
+                            .into_fingerprint(),
+                    },
+                )
                 .into_py_result()?;
         }
         let result = Self::last_field_to_data_slice(&self.root_op_scope).into_py_result()?;
@@ -545,11 +561,17 @@ impl FlowBuilder {
             auto_uuid_field,
         );
         {
+            // TODO: Pass in the right field def fingerprint
             let mut collector = collector.collector.lock().unwrap();
             if let Some(collector) = collector.as_mut() {
-                collector.merge_schema(&collector_schema).into_py_result()?;
+                collector
+                    .collect(&collector_schema, FieldDefFingerprint::default())
+                    .into_py_result()?;
             } else {
-                *collector = Some(CollectorBuilder::new(Arc::new(collector_schema)));
+                *collector = Some(CollectorBuilder::new(
+                    Arc::new(collector_schema),
+                    FieldDefFingerprint::default(),
+                ));
             }
         }
 

rust/cocoindex/src/builder/flow_builder.rs

@@ -53,6 +53,24 @@ pub struct AnalyzedOpOutput {
     pub field_idx: u32,
 }
 
+/// Tracks which affects value of the field, to detect changes of logic.
+#[derive(Debug, Clone)]
+pub struct FieldDefFingerprint {
+    /// Name of sources that affect value of the field.
+    pub source_op_names: HashSet<String>,
+    /// Fingerprint of the logic that affects value of the field.
+    pub fingerprint: Fingerprint,
+}
+
+impl Default for FieldDefFingerprint {
+    fn default() -> Self {
+        Self {
+            source_op_names: HashSet::new(),
+            fingerprint: Fingerprinter::default().into_fingerprint(),
+        }
+    }
+}
+
 pub struct AnalyzedImportOp {
     pub name: String,
     pub executor: Box<dyn SourceExecutor>,
@@ -121,6 +139,7 @@ pub struct AnalyzedExportOp {
     pub value_stable: bool,
     /// Fingerprinter of the output value.
     pub output_value_fingerprinter: Fingerprinter,
+    pub def_fp: FieldDefFingerprint,
 }
 
 pub struct AnalyzedExportTargetOpGroup {
@@ -141,20 +160,8 @@ pub struct AnalyzedOpScope {
     pub scope_qualifier: String,
 }
 
-pub struct ExecutionPlanLogicFingerprint {
-    pub current: Fingerprint,
-    pub legacy: Fingerprint,
-}
-
-impl ExecutionPlanLogicFingerprint {
-    pub fn matches(&self, other: impl AsRef<[u8]>) -> bool {
-        self.current.as_slice() == other.as_ref() || self.legacy.as_slice() == other.as_ref()
-    }
-}
-
 pub struct ExecutionPlan {
-    pub logic_fingerprint: ExecutionPlanLogicFingerprint,
-
+    pub legacy_fingerprint: Vec<Fingerprint>,
     pub import_ops: Vec<AnalyzedImportOp>,
     pub op_scope: AnalyzedOpScope,
     pub export_ops: Vec<AnalyzedExportOp>,

rust/cocoindex/src/builder/plan.rs

@@ -1,3 +1,4 @@
+use crate::execution::indexing_status::SourceLogicFingerprint;
 use crate::prelude::*;
 
 use futures::{StreamExt, future::try_join_all};
@@ -71,6 +72,7 @@ impl<'a> Dumper<'a> {
         import_op: &'a AnalyzedImportOp,
         key: &value::KeyValue,
         key_aux_info: &serde_json::Value,
+        source_logic_fp: &SourceLogicFingerprint,
         collected_values_buffer: &'b mut Vec<Vec<value::FieldValues>>,
     ) -> Result<Option<IndexMap<&'b str, TargetExportData<'b>>>>
     where
@@ -83,6 +85,7 @@ impl<'a> Dumper<'a> {
                 schema: self.schema,
                 key,
                 import_op_idx,
+                source_logic_fp,
             },
             key_aux_info,
             self.setup_execution_ctx,
@@ -139,6 +142,12 @@ impl<'a> Dumper<'a> {
         key_aux_info: serde_json::Value,
         file_path: PathBuf,
     ) -> Result<()> {
+        let source_logic_fp = SourceLogicFingerprint::new(
+            self.plan,
+            import_op_idx,
+            &self.setup_execution_ctx.export_ops,
+            self.plan.legacy_fingerprint.clone(),
+        )?;
         let _permit = import_op
             .concurrency_controller
             .acquire(concur_control::BYTES_UNKNOWN_YET)
@@ -150,6 +159,7 @@ impl<'a> Dumper<'a> {
                 import_op,
                 &key,
                 &key_aux_info,
+                &source_logic_fp,
                 &mut collected_values_buffer,
             )
             .await

rust/cocoindex/src/execution/dumper.rs

@@ -1,3 +1,4 @@
+use crate::execution::indexing_status::SourceLogicFingerprint;
 use crate::prelude::*;
 
 use anyhow::{Context, Ok};
@@ -634,6 +635,7 @@ pub struct SourceRowEvaluationContext<'a> {
     pub schema: &'a schema::FlowSchema,
     pub key: &'a value::KeyValue,
     pub import_op_idx: usize,
+    pub source_logic_fp: &'a SourceLogicFingerprint,
 }
 
 #[derive(Debug)]

rust/cocoindex/src/execution/evaluator.rs

@@ -3,6 +3,47 @@ use crate::prelude::*;
 use super::db_tracking;
 use super::evaluator;
 use futures::try_join;
+use utils::fingerprint::{Fingerprint, Fingerprinter};
+
+pub struct SourceLogicFingerprint {
+    pub current: Fingerprint,
+    pub legacy: Vec<Fingerprint>,
+}
+
+impl SourceLogicFingerprint {
+    pub fn new(
+        exec_plan: &plan::ExecutionPlan,
+        source_idx: usize,
+        export_exec_ctx: &[exec_ctx::ExportOpExecutionContext],
+        legacy: Vec<Fingerprint>,
+    ) -> Result<Self> {
+        let import_op = &exec_plan.import_ops[source_idx];
+        let mut fp = Fingerprinter::default();
+        if exec_plan.import_ops.len() != export_exec_ctx.len() {
+            bail!("Import op count does not match export op count");
+        }
+        for (export_op, export_op_exec_ctx) in
+            std::iter::zip(exec_plan.export_ops.iter(), export_exec_ctx.iter())
+        {
+            if export_op.def_fp.source_op_names.contains(&import_op.name) {
+                fp = fp.with(&(
+                    &export_op.def_fp.fingerprint,
+                    &export_op_exec_ctx.target_id,
+                    &export_op_exec_ctx.schema_version_id,
+                ))?;
+            }
+        }
+        Ok(Self {
+            current: fp.into_fingerprint(),
+            legacy,
+        })
+    }
+
+    pub fn matches(&self, other: impl AsRef<[u8]>) -> bool {
+        self.current.as_slice() == other.as_ref()
+            || self.legacy.iter().any(|fp| fp.as_slice() == other.as_ref())
+    }
+}
 
 #[derive(Debug, Serialize)]
 pub struct SourceRowLastProcessedInfo {
@@ -54,7 +95,7 @@ pub async fn get_source_row_indexing_status(
         is_logic_current: l
             .process_logic_fingerprint
             .as_ref()
-            .map_or(false, |fp| src_eval_ctx.plan.logic_fingerprint.matches(fp)),
+            .map_or(false, |fp| src_eval_ctx.source_logic_fp.matches(fp)),
     });
     let current = SourceRowInfo {
         ordinal: current.ordinal,

rust/cocoindex/src/execution/indexing_status.rs

@@ -1,3 +1,4 @@
+use crate::execution::indexing_status::SourceLogicFingerprint;
 use crate::prelude::*;
 
 use base64::Engine;
@@ -55,7 +56,7 @@ impl SourceVersion {
     pub fn from_stored(
         stored_ordinal: Option<i64>,
         stored_fp: &Option<Vec<u8>>,
-        curr_fp: &ExecutionPlanLogicFingerprint,
+        curr_fp: &SourceLogicFingerprint,
     ) -> Self {
         Self {
             ordinal: Ordinal(stored_ordinal),
@@ -74,7 +75,7 @@ impl SourceVersion {
 
     pub fn from_stored_processing_info(
         info: &db_tracking::SourceTrackingInfoForProcessing,
-        curr_fp: &ExecutionPlanLogicFingerprint,
+        curr_fp: &SourceLogicFingerprint,
     ) -> Self {
         Self::from_stored(
             info.processed_source_ordinal,
@@ -85,7 +86,7 @@ impl SourceVersion {
 
     pub fn from_stored_precommit_info(
         info: &db_tracking::SourceTrackingInfoForPrecommit,
-        curr_fp: &ExecutionPlanLogicFingerprint,
+        curr_fp: &SourceLogicFingerprint,
     ) -> Self {
         Self::from_stored(
             info.processed_source_ordinal,
@@ -240,7 +241,7 @@ impl<'a> RowIndexer<'a> {
             Some(info) => {
                 let existing_version = SourceVersion::from_stored_processing_info(
                     info,
-                    &self.src_eval_ctx.plan.logic_fingerprint,
+                    &self.src_eval_ctx.source_logic_fp,
                 );
 
                 // First check ordinal-based skipping
@@ -486,7 +487,7 @@ impl<'a> RowIndexer<'a> {
         // Check 1: Same check as precommit - verify no newer version exists
         let existing_source_version = SourceVersion::from_stored_precommit_info(
             &existing_tracking_info,
-            &self.src_eval_ctx.plan.logic_fingerprint,
+            &self.src_eval_ctx.source_logic_fp,
         );
         if existing_source_version.should_skip(source_version, Some(self.update_stats)) {
             return Ok(Some(SkippedOr::Skipped(
@@ -537,7 +538,6 @@ impl<'a> RowIndexer<'a> {
         let db_setup = &self.setup_execution_ctx.setup_state.tracking_table;
         let export_ops = &self.src_eval_ctx.plan.export_ops;
         let export_ops_exec_ctx = &self.setup_execution_ctx.export_ops;
-        let logic_fp = &self.src_eval_ctx.plan.logic_fingerprint;
 
         let mut txn = self.pool.begin().await?;
 
@@ -551,8 +551,10 @@ impl<'a> RowIndexer<'a> {
         if self.mode == super::source_indexer::UpdateMode::Normal
             && let Some(tracking_info) = &tracking_info
         {
-            let existing_source_version =
-                SourceVersion::from_stored_precommit_info(&tracking_info, logic_fp);
+            let existing_source_version = SourceVersion::from_stored_precommit_info(
+                &tracking_info,
+                &self.src_eval_ctx.source_logic_fp,
+            );
             if existing_source_version.should_skip(source_version, Some(self.update_stats)) {
                 return Ok(SkippedOr::Skipped(
                     existing_source_version,
@@ -834,7 +836,7 @@ impl<'a> RowIndexer<'a> {
                 cleaned_staging_target_keys,
                 source_version.ordinal.into(),
                 source_fp,
-                &self.src_eval_ctx.plan.logic_fingerprint.current.0,
+                &self.src_eval_ctx.source_logic_fp.current.0,
                 precommit_metadata.process_ordinal,
                 self.process_time.timestamp_micros(),
                 precommit_metadata.new_target_keys,

rust/cocoindex/src/execution/row_indexer.rs

@@ -1,4 +1,9 @@
-use crate::{execution::row_indexer::ContentHashBasedCollapsingBaseline, prelude::*};
+use crate::{
+    execution::{
+        indexing_status::SourceLogicFingerprint, row_indexer::ContentHashBasedCollapsingBaseline,
+    },
+    prelude::*,
+};
 use utils::batching;
 
 use futures::future::Ready;
@@ -62,6 +67,7 @@ pub struct SourceIndexingContext {
     needs_to_track_rows_to_retry: bool,
 
     update_once_batcher: batching::Batcher<UpdateOnceRunner>,
+    source_logic_fp: SourceLogicFingerprint,
 }
 
 pub const NO_ACK: Option<fn() -> Ready<Result<()>>> = None;
@@ -258,6 +264,12 @@ impl SourceIndexingContext {
         let mut rows = HashMap::new();
         let mut rows_to_retry: Option<HashSet<value::KeyValue>> = None;
         let scan_generation = 0;
+        let source_logic_fp = SourceLogicFingerprint::new(
+            &plan,
+            source_idx,
+            &setup_execution_ctx.export_ops,
+            plan.legacy_fingerprint.clone(),
+        )?;
         {
             let mut key_metadata_stream = list_state.list(
                 setup_execution_ctx.import_ops[source_idx].source_id,
@@ -282,7 +294,7 @@ impl SourceIndexingContext {
                             source_version: SourceVersion::from_stored(
                                 key_metadata.processed_source_ordinal,
                                 &key_metadata.process_logic_fingerprint,
-                                &plan.logic_fingerprint,
+                                &source_logic_fp,
                             ),
                             content_version_fp: key_metadata.processed_source_fp,
                         },
@@ -307,6 +319,7 @@ impl SourceIndexingContext {
                 UpdateOnceRunner,
                 batching::BatchingOptions::default(),
             ),
+            source_logic_fp,
         }))
     }
 
@@ -347,6 +360,7 @@ impl SourceIndexingContext {
                 schema,
                 key: &row_input.key,
                 import_op_idx: self.source_idx,
+                source_logic_fp: &self.source_logic_fp,
             };
             let process_time = chrono::Utc::now();
             let operation_in_process_stats_cloned = operation_in_process_stats.clone();