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

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: Box<[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::<Box<[_]>>();
                         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,13 @@ 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_type: Some(
+                        data_fields_info
+                            .primary_key_schema
+                            .iter()
+                            .map(|field| field.value_type.typ.clone())
+                            .collect::<Box<[_]>>(),
+                    ),
                 };
                 targets_analyzed_ss[*idx] = Some(export_op_ss);
 
@@ -940,6 +947,7 @@ impl AnalyzerContext {
                 setup_key,
                 desired_setup_state,
                 setup_by_user: false,
+                key_type: 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_type: Option<Box<[schema::ValueType]>>,
 }
 
 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_type) = &analyzed_target_ss.key_type
+            && let Some(existing_key_type) = &existing_state.common.key_type
+            && key_type != existing_key_type
+        {
+            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_type: analyzed_target_ss.key_type.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: Box<[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: Box<[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: Box<[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

@@ -132,7 +132,7 @@ fn bind_value_field<'arg>(
 pub struct ExportContext {
     db_ref: Option<spec::AuthEntryReference<DatabaseConnectionSpec>>,
     db_pool: PgPool,
-    key_fields_schema: Vec<FieldSchema>,
+    key_fields_schema: Box<[FieldSchema]>,
     value_fields_schema: Vec<FieldSchema>,
     upsert_sql_prefix: String,
     upsert_sql_suffix: String,
@@ -144,7 +144,7 @@ impl ExportContext {
         db_ref: Option<spec::AuthEntryReference<DatabaseConnectionSpec>>,
         db_pool: PgPool,
         table_name: String,
-        key_fields_schema: Vec<FieldSchema>,
+        key_fields_schema: Box<[FieldSchema]>,
         value_fields_schema: Vec<FieldSchema>,
     ) -> Result<Self> {
         let key_fields = key_fields_schema

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: Box<[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: Box<[FieldSchema]>,
     pub value_fields_schema: Vec<FieldSchema>,
 }
 

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_type: Option<Box<[schema::ValueType]>>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]