{connector_name} snapshot primary-key index {index} is out o

Error message

{connector_name} snapshot primary-key index {index} is out of bounds for {} columns

What it means

`decode_row_with_strict_pk` validates that every primary key index refers to an actual field of the decoded row schema before decoding datums. When a `pk_indices` entry is greater than or equal to the number of schema fields (`schema.fields.len()`), the index cannot name a primary key, so decoding aborts with this out-of-bounds error naming the connector and the column count.

Source

Thrown at src/connector/src/parser/mod.rs:81

use crate::source::{
    BoxSourceMessageEventStream, SourceChunkStream, SourceColumnDesc, SourceColumnType,
    SourceContext, SourceContextRef, SourceCtrlOpts, SourceMessageEvent, SourceMeta,
    SourceReaderEvent,
};

fn decode_row_with_strict_pk(
    connector_name: &str,
    schema: &Schema,
    pk_indices: &[usize],
    mut decode: impl FnMut(usize, &Field) -> anyhow::Result<Datum>,
    mut log_non_pk_error: impl FnMut(&str, anyhow::Error),
) -> anyhow::Result<OwnedRow> {
    if let Some(index) = pk_indices
        .iter()
        .copied()
        .find(|index| *index >= schema.fields.len())
    {
        bail!(
            "{connector_name} snapshot primary-key index {index} is out of bounds for {} columns",
            schema.fields.len()
        );
    }

    let mut datums = Vec::with_capacity(schema.fields.len());
    for (index, field) in schema.fields.iter().enumerate() {
        let is_pk = pk_indices.contains(&index);
        let decode_result = decode(index, field);
        let datum = if is_pk {
            decode_result.with_context(|| {
                format!(
                    "failed to decode {connector_name} snapshot primary key `{}`",
                    field.name
                )
            })?
        } else {
            match decode_result {

View on GitHub (pinned to 6469eb736d)

Solutions

Align the upstream table schema with the RW table definition (re-add the missing column) so decoded row width covers all pk indices.
Verify pk_indices are derived from the same schema that is passed to `decode_row_with_strict_pk`.
Re-create the source/table so the plan's schema and pk indices are recomputed against the current upstream schema.

Defensive patterns

Strategy: validation

Validate before calling

if pk_indices.iter().any(|&i| i >= schema.fields.len()) {
    return Err(anyhow!("pk index out of range for decoded schema"));
}

Try / catch

match decode_row_with_strict_pk(...) {
    Err(e) if e.to_string().contains("out of bounds") => resync_schema_and_retry(),
    other => other,
}

Prevention

Derive pk_indices and the decode schema from the same plan/schema snapshot.
Watch for upstream ALTER TABLE during snapshots; re-create the source after schema changes.
Assert schema field count matches the table's column count before starting a snapshot.

When it happens

Trigger: Calling `decode_row_with_strict_pk` with a `pk_indices` vector computed from the target table schema while the decoder's `schema` (Data schema from the CDC payload) has fewer fields — e.g. the upstream message schema changed and no longer matches the table definition.

Common situations: Snapshot backfill after the upstream table was altered (columns dropped), so pk indices from the RW table exceed the incoming schema width; a connector misconfiguration where the declared schema and pk indices are built from different sources; version skew between connector message schema and table plan.

Related errors

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