{"record":{"id":"361e15d5f678042d","repo":"risingwavelabs/risingwave","slug":"primary-key-columns-not-found-please-set-the-pri-361e15","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/snowflake.rs","lineNumber":428,"sourceCode":"            snowflake_task_ctx.task_serverless = self.task_serverless;\n            snowflake_task_ctx.task_target_completion_interval =\n                self.task_target_completion_interval.clone();\n            if !self.task_serverless {\n                snowflake_task_ctx.warehouse = Some(\n                    self.snowflake_warehouse\n                        .clone()\n                        .ok_or(SinkError::Config(anyhow!(\"warehouse is required\")))?,\n                );\n            }\n            let pk_column_names: Vec<_> = schema\n                .fields\n                .iter()\n                .enumerate()\n                .filter(|(index, _)| pk_indices.contains(index))\n                .map(|(_, field)| field.name.clone())\n                .collect();\n            if pk_column_names.is_empty() {\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            snowflake_task_ctx.pk_column_names = Some(pk_column_names);\n            snowflake_task_ctx.all_column_names = Some(\n                schema\n                    .fields\n                    .iter()\n                    .map(|field| field.name.clone())\n                    .collect(),\n            );\n            snowflake_task_ctx.task_name = Some(format!(\n                \"rw_snowflake_sink_from_{cdc_table_name}_to_{target_table_name}\"\n            ));\n        }\n        Ok(Some((snowflake_task_ctx, client)))\n    }\n}","sourceCodeStart":410,"sourceCodeEnd":446,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/snowflake_redshift/snowflake.rs#L410-L446","documentation":"This error is thrown when building the Snowflake JDBC sink task context: the resolved primary key column name list is empty. A Snowflake sink performing upserts must know which columns form the primary key, either via the `primary_key` sink property or from the upstream table's PK columns. Without it the sink cannot generate correct MERGE/DELETE statements, so creation fails.","triggerScenarios":"Calling `SnowflakeConfig::build_snowflake_task_ctx_jdbc_client` (via `SnowflakeSink::new`) with a non-append-only sink where `pk_indices` is empty, or pk_indices resolve to no fields in the schema (no `primary_key` property set).","commonSituations":"Creating a Snowflake sink on a table/source with no primary key while omitting the `primary_key = '...'` property in the WITH options; upstream where PK columns were dropped; version changes requiring PK for non-append-only sinks.","solutions":["Add `primary_key = '<col1>,<col2>'` to the sink's WITH options listing the Snowflake target table's key columns","Ensure the upstream materialized view/table has a primary key so pk_indices are non-empty","If the sink is truly append-only, create it with `append_only = true` so PK resolution is not required"],"exampleFix":"-- before\nCREATE SINK s FROM mv INTO snowflake WITH (\n  connector='snowflake', ...\n);\n-- after\nCREATE SINK s FROM mv INTO snowflake WITH (\n  connector='snowflake',\n  primary_key = 'id',\n  ...\n);","handlingStrategy":"validation","validationCode":"let has_pk = !sink_opts.get(\"primary_key\").unwrap_or(\"\").is_empty() || !pk_indices.is_empty();\nif !has_pk {\n    return Err(\"Snowflake sink requires `primary_key` option or upstream primary key columns\");\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always set `primary_key` in Snowflake sink WITH options unless the source has a PK","Check upstream table/mv has a primary key before creating non-append-only sinks"],"tags":["snowflake","sink","config","primary-key"],"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-23T08:17:48.524Z"}