{"record":{"id":"7db328f71e6af585","repo":"risingwavelabs/risingwave","slug":"connector-name-snapshot-primary-key-index-index","errorCode":null,"errorMessage":"{connector_name} snapshot primary-key index {index} is out of bounds for {} columns","messagePattern":"(.+?) snapshot primary-key index (.+?) is out of bounds for (.+?) columns","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/parser/mod.rs","lineNumber":81,"sourceCode":"use crate::source::{\n    BoxSourceMessageEventStream, SourceChunkStream, SourceColumnDesc, SourceColumnType,\n    SourceContext, SourceContextRef, SourceCtrlOpts, SourceMessageEvent, SourceMeta,\n    SourceReaderEvent,\n};\n\nfn decode_row_with_strict_pk(\n    connector_name: &str,\n    schema: &Schema,\n    pk_indices: &[usize],\n    mut decode: impl FnMut(usize, &Field) -> anyhow::Result<Datum>,\n    mut log_non_pk_error: impl FnMut(&str, anyhow::Error),\n) -> anyhow::Result<OwnedRow> {\n    if let Some(index) = pk_indices\n        .iter()\n        .copied()\n        .find(|index| *index >= schema.fields.len())\n    {\n        bail!(\n            \"{connector_name} snapshot primary-key index {index} is out of bounds for {} columns\",\n            schema.fields.len()\n        );\n    }\n\n    let mut datums = Vec::with_capacity(schema.fields.len());\n    for (index, field) in schema.fields.iter().enumerate() {\n        let is_pk = pk_indices.contains(&index);\n        let decode_result = decode(index, field);\n        let datum = if is_pk {\n            decode_result.with_context(|| {\n                format!(\n                    \"failed to decode {connector_name} snapshot primary key `{}`\",\n                    field.name\n                )\n            })?\n        } else {\n            match decode_result {","sourceCodeStart":63,"sourceCodeEnd":99,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/parser/mod.rs#L63-L99","documentation":"`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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"if pk_indices.iter().any(|&i| i >= schema.fields.len()) {\n    return Err(anyhow!(\"pk index out of range for decoded schema\"));\n}","typeGuard":null,"tryCatchPattern":"match decode_row_with_strict_pk(...) {\n    Err(e) if e.to_string().contains(\"out of bounds\") => resync_schema_and_retry(),\n    other => other,\n}","preventionTips":["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."],"tags":["cdc","snapshot","primary-key","index-out-of-bounds","schema"],"backgroundTag":"index-out-of-bounds","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"}