fix: always treat target as not-compatible when key schema changed

Author: georgeh0

src/builder/analyzer.rs

@@ -643,7 +643,7 @@ fn add_collector(
 struct ExportDataFieldsInfo {
     local_collector_ref: AnalyzedLocalCollectorReference,
     primary_key_def: AnalyzedPrimaryKeyDef,
-    primary_key_schema: Vec<FieldSchema>,
+    primary_key_schema: Arc<[FieldSchema]>,
     value_fields_idx: Vec<u32>,
     value_stable: bool,
 }
@@ -851,8 +851,8 @@ impl AnalyzerContext {
 
                         let primary_key_schema = pk_fields_idx
                             .iter()
-                            .map(|idx| collector_schema.fields[*idx].clone())
-                            .collect::<Vec<_>>();
+                            .map(|idx| collector_schema.fields[*idx].without_attrs())
+                            .collect::<Arc<[_]>>();
                         let mut value_fields_schema: Vec<FieldSchema> = vec![];
                         let mut value_fields_idx = vec![];
                         for (idx, field) in collector_schema.fields.iter().enumerate() {
@@ -911,6 +911,7 @@ impl AnalyzerContext {
                     setup_key: data_coll_output.setup_key,
                     desired_setup_state: data_coll_output.desired_setup_state,
                     setup_by_user: export_op.spec.setup_by_user,
+                    key_schema: Some(data_fields_info.primary_key_schema.clone()),
                 };
                 targets_analyzed_ss[*idx] = Some(export_op_ss);
 
@@ -940,6 +941,7 @@ impl AnalyzerContext {
                 setup_key,
                 desired_setup_state,
                 setup_by_user: false,
+                key_schema: None,
             };
             declarations_analyzed_ss.push(decl_ss);
         }

src/builder/exec_ctx.rs

@@ -23,6 +23,8 @@ pub struct AnalyzedTargetSetupState {
     pub setup_key: serde_json::Value,
     pub desired_setup_state: serde_json::Value,
     pub setup_by_user: bool,
+    /// None for declarations.
+    pub key_schema: Option<Arc<[schema::FieldSchema]>>,
 }
 
 pub struct AnalyzedSetupState {
@@ -121,15 +123,19 @@ fn build_target_id(
     let mut compatible_target_ids = HashSet::<Option<i32>>::new();
     let mut reusable_schema_version_ids = HashSet::<Option<i32>>::new();
     for existing_state in existing_target_states.iter().flat_map(|v| v.iter()) {
-        let compatibility =
-            if analyzed_target_ss.setup_by_user == existing_state.common.setup_by_user {
-                target_factory.check_state_compatibility(
-                    &analyzed_target_ss.desired_setup_state,
-                    &existing_state.state,
-                )?
-            } else {
-                SetupStateCompatibility::NotCompatible
-            };
+        let compatibility = if let Some(key_schema) = &analyzed_target_ss.key_schema
+            && let Some(existing_key_schema) = &existing_state.common.key_schema
+            && key_schema != existing_key_schema
+        {
+            SetupStateCompatibility::NotCompatible
+        } else if analyzed_target_ss.setup_by_user != existing_state.common.setup_by_user {
+            SetupStateCompatibility::NotCompatible
+        } else {
+            target_factory.check_state_compatibility(
+                &analyzed_target_ss.desired_setup_state,
+                &existing_state.state,
+            )?
+        };
         let compatible_target_id = if compatibility != SetupStateCompatibility::NotCompatible {
             reusable_schema_version_ids.insert(
                 (compatibility == SetupStateCompatibility::Compatible)
@@ -184,6 +190,7 @@ fn build_target_id(
                     schema_version_id,
                     max_schema_version_id: max_schema_version_id.max(schema_version_id),
                     setup_by_user: analyzed_target_ss.setup_by_user,
+                    key_schema: analyzed_target_ss.key_schema.clone(),
                 },
                 state: analyzed_target_ss.desired_setup_state.clone(),
             });

src/builder/plan.rs

@@ -105,7 +105,7 @@ pub struct AnalyzedExportOp {
     pub export_target_factory: Arc<dyn TargetFactory + Send + Sync>,
     pub export_context: Arc<dyn Any + Send + Sync>,
     pub primary_key_def: AnalyzedPrimaryKeyDef,
-    pub primary_key_schema: Vec<FieldSchema>,
+    pub primary_key_schema: Arc<[FieldSchema]>,
     /// idx for value fields - excluding the primary key field.
     pub value_fields: Vec<u32>,
     /// If true, value is never changed on the same primary key.

src/ops/factory_bases.rs

@@ -363,7 +363,7 @@ pub struct TypedExportDataCollectionBuildOutput<F: TargetFactoryBase + ?Sized> {
 pub struct TypedExportDataCollectionSpec<F: TargetFactoryBase + ?Sized> {
     pub name: String,
     pub spec: F::Spec,
-    pub key_fields_schema: Vec<FieldSchema>,
+    pub key_fields_schema: Arc<[FieldSchema]>,
     pub value_fields_schema: Vec<FieldSchema>,
     pub index_options: IndexOptions,
 }

src/ops/interface.rs

@@ -258,7 +258,7 @@ pub struct ExportDataCollectionBuildOutput {
 pub struct ExportDataCollectionSpec {
     pub name: String,
     pub spec: serde_json::Value,
-    pub key_fields_schema: Vec<FieldSchema>,
+    pub key_fields_schema: Arc<[FieldSchema]>,
     pub value_fields_schema: Vec<FieldSchema>,
     pub index_options: IndexOptions,
 }

src/ops/py_factory.rs

@@ -356,11 +356,14 @@ impl interface::TargetFactory for PyExportTargetFactory {
 
     fn check_state_compatibility(
         &self,
-        _desired_state: &serde_json::Value,
-        _existing_state: &serde_json::Value,
+        desired_state: &serde_json::Value,
+        existing_state: &serde_json::Value,
     ) -> Result<SetupStateCompatibility> {
-        // The Python target connector doesn't support state update yet.
-        Ok(SetupStateCompatibility::Compatible)
+        Ok(if desired_state == existing_state {
+            SetupStateCompatibility::Compatible
+        } else {
+            SetupStateCompatibility::PartialCompatible
+        })
     }
 
     fn describe_resource(&self, key: &serde_json::Value) -> Result<String> {

src/ops/targets/postgres.rs

@@ -632,7 +632,7 @@ impl TargetFactoryBase for Factory {
                         db_ref,
                         db_pool.clone(),
                         table_name,
-                        d.key_fields_schema,
+                        d.key_fields_schema.iter().cloned().collect(),
                         d.value_fields_schema,
                     )?);
                     Ok(export_context)

src/ops/targets/shared/property_graph.rs

@@ -112,7 +112,7 @@ impl<AuthEntry> std::fmt::Display for GraphElementType<AuthEntry> {
 
 pub struct GraphElementSchema {
     pub elem_type: ElementType,
-    pub key_fields: Vec<schema::FieldSchema>,
+    pub key_fields: Arc<[schema::FieldSchema]>,
     pub value_fields: Vec<schema::FieldSchema>,
 }
 
@@ -256,7 +256,7 @@ impl GraphElementSchemaBuilder {
         }
         Ok(GraphElementSchema {
             elem_type: self.elem_type,
-            key_fields: self.key_fields,
+            key_fields: self.key_fields.into(),
             value_fields: self.value_fields,
         })
     }
@@ -349,7 +349,7 @@ pub struct DataCollectionGraphMappingInput<'a, AuthEntry> {
     pub mapping: &'a GraphElementMapping,
     pub index_options: &'a spec::IndexOptions,
 
-    pub key_fields_schema: Vec<FieldSchema>,
+    pub key_fields_schema: Arc<[FieldSchema]>,
     pub value_fields_schema: Vec<FieldSchema>,
 }
 
@@ -435,6 +435,7 @@ pub fn analyze_graph_mappings<'a, AuthEntry: 'a>(
                                 .key_fields_schema
                                 .into_iter()
                                 .enumerate()
+                                .map(|(idx, f)| (idx, f.clone()))
                                 .collect(),
                             data_coll_input
                                 .value_fields_schema

src/setup/states.rs

@@ -180,6 +180,8 @@ pub struct TargetSetupStateCommon {
     pub max_schema_version_id: i32,
     #[serde(default)]
     pub setup_by_user: bool,
+    #[serde(default)]
+    pub key_schema: Option<Arc<[schema::FieldSchema]>>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]