{"record":{"id":"fb21649779e3bfcd","repo":"risingwavelabs/risingwave","slug":"the-primary-key-column-index-is-out-of-bounds-i","errorCode":null,"errorMessage":"The primary key column index {} is out of bounds in schema {:?}","messagePattern":"The primary key column index (.+?) is out of bounds in schema (.+?)","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/formatter/mod.rs","lineNumber":207,"sourceCode":"    params: &'a EncoderParams<'_>,\n    pk_indices: &'a Option<Vec<usize>>,\n) -> Result<(usize, &'a Field)> {\n    let Some(pk_indices) = pk_indices else {\n        return Err(SinkError::Config(anyhow!(\n            \"{}Encoder requires primary key columns to be specified\",\n            data_type_name\n        )));\n    };\n    if pk_indices.len() != 1 {\n        return Err(SinkError::Config(anyhow!(\n            \"KEY ENCODE {} expects only one primary key, but got {}\",\n            data_type_name,\n            pk_indices.len(),\n        )));\n    }\n\n    let schema_ref = params.schema.fields().get(pk_indices[0]).ok_or_else(|| {\n        SinkError::Config(anyhow!(\n            \"The primary key column index {} is out of bounds in schema {:?}\",\n            pk_indices[0],\n            params.schema\n        ))\n    })?;\n\n    Ok((pk_indices[0], schema_ref))\n}\n\nimpl EncoderBuild for BytesEncoder {\n    async fn build(params: EncoderParams<'_>, pk_indices: Option<Vec<usize>>) -> Result<Self> {\n        match pk_indices {\n            // This is being used as a key encoder\n            Some(_) => {\n                let (pk_index, schema_ref) = ensure_only_one_pk(\"BYTES\", &params, &pk_indices)?;\n                if let DataType::Bytea = schema_ref.data_type() {\n                    Ok(BytesEncoder::new(params.schema, pk_index))\n                } else {","sourceCodeStart":189,"sourceCodeEnd":225,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/formatter/mod.rs#L189-L225","documentation":"The single primary key index supplied to the encoder points past the end of the sink schema's fields, so no column can be resolved. This is an internal consistency violation between the pk indices and the schema passed into EncoderParams.","triggerScenarios":"build() resolves pk_indices[0] via params.schema.fields().get(pk_indices[0]) and it is None — pk index computed against a different/wider schema than the one given to the encoder.","commonSituations":"Custom sink code passing a pruned or reordered schema while reusing pk indices from the original relation; version drift where downstream column elimination shrank the schema.","solutions":["Recompute pk_indices against the exact schema passed in params.schema","Verify the sink's downstream dispatcher did not drop/reorder columns before encoder build","Ensure pk index refers to a column still present in the sink's output schema"],"exampleFix":"// before: stale indices\nlet encoder = BytesEncoder::build(params_with_full_schema, pk_from_relation).await?;\n// after\nlet pk_indices = pk_from_relation.filter(|i| *i < params.schema.len());","handlingStrategy":"validation","validationCode":"let pk = pk_indices.expect(\"pk required\")[0];\nassert!(pk < schema.len(), \"pk index {} out of bounds for schema len {}\", pk, schema.len());","typeGuard":"fn pk_in_schema(pk: usize, schema: &Schema) -> bool { pk < schema.fields().len() }","tryCatchPattern":null,"preventionTips":["Recompute pk indices from the exact schema handed to the encoder","Log schema + pk indices together when debugging sink builds"],"tags":["sink","schema","index-out-of-bounds"],"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-23T08:17:48.524Z"}