{"record":{"id":"4b8a9d43bf7e0aa2","repo":"risingwavelabs/risingwave","slug":"duplicate-pk-index-pk-index","errorCode":null,"errorMessage":"duplicate pk index {pk_index}","messagePattern":"duplicate pk index (.+?)","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs","lineNumber":389,"sourceCode":"\n/// Builds a projection for PK root columns and maps the projected physical order back to the\n/// downstream PK order.\nfn pk_projection(\n    parquet_schema: &SchemaDescriptor,\n    pk_indices: &[usize],\n) -> Result<(ProjectionMask, Vec<usize>), SinkError> {\n    let root_count = parquet_schema.root_schema().get_fields().len();\n    let mut physical_indices = Vec::with_capacity(pk_indices.len());\n    let mut seen = HashSet::with_capacity(pk_indices.len());\n\n    for &pk_index in pk_indices {\n        if pk_index >= root_count {\n            return Err(SinkError::Iceberg(anyhow!(\n                \"pk index {pk_index} is out of range for parquet schema with {root_count} root columns\"\n            )));\n        }\n        if !seen.insert(pk_index) {\n            return Err(SinkError::Iceberg(anyhow!(\"duplicate pk index {pk_index}\")));\n        }\n        physical_indices.push(pk_index);\n    }\n\n    physical_indices.sort_unstable();\n    let pk_order = pk_indices\n        .iter()\n        .map(|pk_index| {\n            physical_indices\n                .binary_search(pk_index)\n                .expect(\"validated pk index must be projected\")\n        })\n        .collect();\n    let projection = ProjectionMask::roots(parquet_schema, physical_indices);\n\n    Ok((projection, pk_order))\n}\n","sourceCodeStart":371,"sourceCodeEnd":407,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs#L371-L407","documentation":"Thrown by `pk_projection` when the `pk_indices` slice contains the same index twice. A duplicate pk index would make the projection read the same parquet column twice and produce a bogus pk row order, so the resolver rejects it up front with this error. It is an invariant check on caller-provided input: pk indices must be a set of distinct column positions.","triggerScenarios":"Calling `pk_projection` (via `scan_input_pks_at_positions`, `scan_output_file_inner`, or tests) with `pk_indices` containing a repeated value, e.g. because upstream code built the list by concatenating per-column pk specs without deduplication.","commonSituations":"Bug in code that derives pk indices from multiple sources (identity columns + upstream pks) and appends instead of deduplicating; hand-edited sink configuration listing the same pk column twice.","solutions":["Deduplicate the pk_indices before calling the resolver, e.g. via a HashSet as the code itself does.","Find where pk_indices are constructed (sink definition / table properties) and fix the duplication at the source.","Add an assertion or unit test on pk_indices construction so duplicates are caught before compaction runs."],"exampleFix":"// before\nlet pk_indices: Vec<usize> = identity_pks.into_iter().chain(upstream_pks).collect();\n\n// after\nlet pk_indices: Vec<usize> = identity_pks.into_iter().chain(upstream_pks).collect::<HashSet<_>>().into_iter().collect();","handlingStrategy":"validation","validationCode":"let unique: HashSet<usize> = pk_indices.iter().copied().collect();\nif unique.len() != pk_indices.len() {\n    return Err(\"pk_indices contain duplicates\".to_string());\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Always deduplicate pk_indices at construction time using a HashSet or BTreeSet.","Unit-test pk index derivation so concatenating multiple sources cannot produce duplicates.","Keep pk column definitions in one place to avoid double-registration."],"tags":["parquet","invariant","iceberg-sink","compaction"],"backgroundTag":"internal-invariant-violation","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"}