{"record":{"id":"6d3a5158c1b9bd5d","repo":"risingwavelabs/risingwave","slug":"primary-key-columns-not-found-please-set-the-pri","errorCode":null,"errorMessage":"Primary key columns not found. Please set the `primary_key` column in the sink properties, or ensure that the sink contains the primary key columns from the upstream.","messagePattern":"Primary key columns not found\\. Please set the `primary_key` column in the sink properties, or ensure that the sink contains the primary key columns from the upstream\\.","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/snowflake_redshift/redshift.rs","lineNumber":279,"sourceCode":"\n    fn is_coordinated_sink(&self) -> bool {\n        true\n    }\n\n    async fn new_coordinator(\n        &self,\n        _iceberg_compact_stat_sender: Option<UnboundedSender<IcebergSinkCompactionUpdate>>,\n    ) -> Result<SinkCommitCoordinator> {\n        let pk_column_names: Vec<_> = self\n            .schema\n            .fields\n            .iter()\n            .enumerate()\n            .filter(|(index, _)| self.pk_indices.contains(index))\n            .map(|(_, field)| field.name.clone())\n            .collect();\n        if pk_column_names.is_empty() && !self.is_append_only {\n            return Err(SinkError::Config(anyhow!(\n                \"Primary key columns not found. Please set the `primary_key` column in the sink properties, or ensure that the sink contains the primary key columns from the upstream.\"\n            )));\n        }\n        let all_column_names = self\n            .schema\n            .fields\n            .iter()\n            .map(|field| field.name.clone())\n            .collect();\n        let coordinator = RedshiftSinkCommitter::new(\n            self.config.clone(),\n            self.is_append_only,\n            &pk_column_names,\n            &all_column_names,\n            self.param.sink_id,\n        )?;\n        Ok(SinkCommitCoordinator::SinglePhase(Box::new(coordinator)))\n    }","sourceCodeStart":261,"sourceCodeEnd":297,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/snowflake_redshift/redshift.rs#L261-L297","documentation":"`RedshiftSink::new_coordinator` requires primary key column names for non-append-only sinks so it can build merge/upsert SQL. If the derived `pk_column_names` list is empty (no pk_indices mapped to schema fields) and the sink is not append-only, this SinkError::Config is returned.","triggerScenarios":"Creating a Redshift (or Snowflake S3) sink over a stream without a primary key and without an explicit `primary_key` property, while the sink performs non-append-only writes.","commonSituations":"Sources/materialized views without a defined primary key; users forgetting to set `primary_key` in sink properties; column name mismatches so mapped names end up empty.","solutions":["Set the `primary_key` property in the CREATE SINK statement listing the key columns","Ensure the upstream relation defines primary keys so pk_indices map to schema fields","If the data is truly append-only, declare the sink as append-only so primary keys are not required"],"exampleFix":"// before\nCREATE SINK s FROM mv WITH ( connector = 'redshift', ... );\n-- after\nCREATE SINK s FROM mv WITH ( connector = 'redshift', primary_key = 'id', ... );","handlingStrategy":"validation","validationCode":"// before creating the sink\nlet has_pk = !pk_indices.is_empty() || props.contains_key(\"primary_key\");\nif !has_pk && !is_append_only {\n    return Err(\"non-append-only sink requires primary_key columns\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Define primary keys on upstream relations when data has updates","Always set `primary_key` in sink properties for upsert sinks","Confirm column names in the sink schema match the upstream key fields"],"tags":["rust","risingwave","sink","redshift","primary-key","config"],"backgroundTag":"missing-required-config-field","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}