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

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.

Solutions

1. Add `primary_key = '<col1>,<col2>'` to the sink's WITH options listing the Snowflake target table's key columns
2. Ensure the upstream materialized view/table has a primary key so pk_indices are non-empty
3. If the sink is truly append-only, create it with `append_only = true` so PK resolution is not required

Example fix

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

let has_pk = !sink_opts.get("primary_key").unwrap_or("").is_empty() || !pk_indices.is_empty();
if !has_pk {
    return Err("Snowflake sink requires `primary_key` option or upstream primary key columns");
}

When it happens

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

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

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

• ambiguous auth: multiple auth options provided; remove one…
• auth.method=key_pair_object must not set `password`
• database is required
• database is required for Snowflake JDBC sink
• intermediate.table.name is required

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

Appendix: source

Thrown at src/connector/src/sink/snowflake_redshift/snowflake.rs:428

            snowflake_task_ctx.task_serverless = self.task_serverless;
            snowflake_task_ctx.task_target_completion_interval =
                self.task_target_completion_interval.clone();
            if !self.task_serverless {
                snowflake_task_ctx.warehouse = Some(
                    self.snowflake_warehouse
                        .clone()
                        .ok_or(SinkError::Config(anyhow!("warehouse is required")))?,
                );
            }
            let pk_column_names: Vec<_> = schema
                .fields
                .iter()
                .enumerate()
                .filter(|(index, _)| pk_indices.contains(index))
                .map(|(_, field)| field.name.clone())
                .collect();
            if pk_column_names.is_empty() {
                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."
                )));
            }
            snowflake_task_ctx.pk_column_names = Some(pk_column_names);
            snowflake_task_ctx.all_column_names = Some(
                schema
                    .fields
                    .iter()
                    .map(|field| field.name.clone())
                    .collect(),
            );
            snowflake_task_ctx.task_name = Some(format!(
                "rw_snowflake_sink_from_{cdc_table_name}_to_{target_table_name}"
            ));
        }
        Ok(Some((snowflake_task_ctx, client)))
    }
}

View on GitHub (pinned to 6469eb736d)