{"record":{"id":"e9327201d3344a48","repo":"risingwavelabs/risingwave","slug":"primary-key-not-defined-for-upsert-bigquery-sink","errorCode":null,"errorMessage":"Primary key not defined for upsert bigquery sink (please define in `primary_key` field)","messagePattern":"Primary key not defined for upsert bigquery sink \\(please define in `primary_key` field\\)","errorType":"validation","errorClass":"SinkError::Config","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/big_query.rs","lineNumber":557,"sourceCode":"            self.config.clone(),\n            self.schema.clone(),\n            self.pk_indices.clone(),\n            self.is_append_only,\n        )\n        .await?;\n        Ok(BigQueryLogSinker::new(\n            writer,\n            resp_stream,\n            BIGQUERY_SEND_FUTURE_BUFFER_MAX_SIZE,\n        ))\n    }\n\n    async fn validate(&self) -> Result<()> {\n        risingwave_common::license::Feature::BigQuerySink\n            .check_available()\n            .map_err(|e| anyhow::anyhow!(e))?;\n        if !self.is_append_only && self.pk_indices.is_empty() {\n            return Err(SinkError::Config(anyhow!(\n                \"Primary key not defined for upsert bigquery sink (please define in `primary_key` field)\"\n            )));\n        }\n        let client = self\n            .config\n            .common\n            .build_client(&self.config.aws_auth_props)\n            .await?;\n        let BigQueryCommon {\n            project: project_id,\n            dataset: dataset_id,\n            table: table_id,\n            ..\n        } = &self.config.common;\n\n        if self.config.common.auto_create {\n            match client\n                .table()","sourceCodeStart":539,"sourceCodeEnd":575,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/big_query.rs#L539-L575","documentation":"`validate` enforces that a non-append-only BigQuery sink has a primary key: if the sink is not append-only and `pk_indices` is empty, upserts cannot be keyed and the sink raises `SinkError::Config`. BigQuery upsert mode requires a primary key to identify rows to update.","triggerScenarios":"Creating a BigQuery sink with upsert semantics (non-append-only, e.g. from a table or MV with PKs not propagated, or explicit non-append-only mode) without specifying `primary_key` in the sink's WITH options.","commonSituations":"Users sink non-append-only materialized views to BigQuery for upserts but omit the `primary_key` option; MVs without PKs are sunk in upsert mode by mistake.","solutions":["Add `primary_key = '<col,...>'` to the sink's WITH/options so upserts can be keyed.","If the data is actually append-only, create the sink in append-only mode so no PK is required.","Ensure the underlying table/MV has a primary key and reference those columns as the sink's primary key."],"exampleFix":"-- before\nCREATE SINK s FROM mv WITH (\n  connector='bigquery', ..., type='upsert'\n);\n\n-- after\nCREATE SINK s FROM mv WITH (\n  connector='bigquery', ..., type='upsert', primary_key='id'\n);","handlingStrategy":"validation","validationCode":"-- Ensure the source MV/table has a PK and you specify primary_key for upsert sinks:\nSELECT relation, primary_key\nFROM rw_catalog.rw_relations\nWHERE name = '<mv_name>';\n-- Only use type='upsert' when primary_key is set in WITH options.","typeGuard":null,"tryCatchPattern":"// Catch config error on sink creation\ntry {\n  await client.query(\"CREATE SINK s FROM mv WITH (connector='bigquery', type='upsert', primary_key='id')\");\n} catch (e) {\n  if (e.message.includes('Primary key not defined for upsert')) {\n    // retry with primary_key option or switch to append-only type\n  }\n  throw e;\n}","preventionTips":["Always pair `type='upsert'` (or non-append-only sinks) with an explicit `primary_key` option.","Confirm the source table/MV has a primary key before configuring upsert sinks.","Use append-only sink mode when you only need inserts and no dedup/upserts."],"tags":["bigquery","sink","primary-key","configuration"],"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"}