risingwavelabs/risingwave · error · SinkError::Config

Primary key columns not found. Please set the `primary_key`

Error message

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.

What it means

`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.

Source

Thrown at src/connector/src/sink/snowflake_redshift/redshift.rs:279

    fn is_coordinated_sink(&self) -> bool {
        true
    }

    async fn new_coordinator(
        &self,
        _iceberg_compact_stat_sender: Option<UnboundedSender<IcebergSinkCompactionUpdate>>,
    ) -> Result<SinkCommitCoordinator> {
        let pk_column_names: Vec<_> = self
            .schema
            .fields
            .iter()
            .enumerate()
            .filter(|(index, _)| self.pk_indices.contains(index))
            .map(|(_, field)| field.name.clone())
            .collect();
        if pk_column_names.is_empty() && !self.is_append_only {
            return Err(SinkError::Config(anyhow!(
                "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."
            )));
        }
        let all_column_names = self
            .schema
            .fields
            .iter()
            .map(|field| field.name.clone())
            .collect();
        let coordinator = RedshiftSinkCommitter::new(
            self.config.clone(),
            self.is_append_only,
            &pk_column_names,
            &all_column_names,
            self.param.sink_id,
        )?;
        Ok(SinkCommitCoordinator::SinglePhase(Box::new(coordinator)))
    }

View on GitHub (pinned to 6469eb736d)

Solutions

1. Set the `primary_key` property in the CREATE SINK statement listing the key columns
2. Ensure the upstream relation defines primary keys so pk_indices map to schema fields
3. If the data is truly append-only, declare the sink as append-only so primary keys are not required

Example fix

// before
CREATE SINK s FROM mv WITH ( connector = 'redshift', ... );
-- after
CREATE SINK s FROM mv WITH ( connector = 'redshift', primary_key = 'id', ... );

// before creating the sink
let has_pk = !pk_indices.is_empty() || props.contains_key("primary_key");
if !has_pk && !is_append_only {
    return Err("non-append-only sink requires primary_key columns");
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "is required", "must be set", "missing required field": configuration validation errors across open-source libraries — this error's family across 36 libraries.

Related errors

• SinkError::Config(anyhow!(e))
• intermediate.table.name is required for append-only sink
• intermediate.table.name is required for non-append-only sink
• Primary key not defined for upsert doris sink (please define
• please set the separator in the with option, when there are

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/6d3a5158c1b9bd5d. Report an issue: GitHub.