{"record":{"id":"9568091b1767cab2","repo":"risingwavelabs/risingwave","slug":"pk-index-pk-index-is-out-of-range-for-parquet-sc","errorCode":null,"errorMessage":"pk index {pk_index} is out of range for parquet schema with {root_count} root columns","messagePattern":"pk index (.+?) is out of range for parquet schema with (.+?) root columns","errorType":"validation","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs","lineNumber":384,"sourceCode":"            }\n        }\n    }\n    Ok(map)\n}\n\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();","sourceCodeStart":366,"sourceCodeEnd":402,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs#L366-L402","documentation":"This error is thrown by `pk_projection` in the Iceberg sink compaction resolver when a primary-key column index supplied by the caller points past the end of the parquet file's root schema. The parquet file on disk has fewer root columns than the pk index implies, so projecting that index would read out of bounds of the schema. The check exists to fail fast with a descriptive message instead of letting arrow projection panic deeper in the scan pipeline.","triggerScenarios":"Calling `pk_projection` (directly or via `scan_input_pks_at_positions` / `scan_output_file_inner` during compaction conflict resolution) with `pk_indices` containing an index >= the number of root fields in the parquet schema being scanned. This happens when the pk_indices configured for the sink do not match the actual schema of the parquet data file, e.g. after a schema evolution or when a stale pk_index set is applied to a newly written output file.","commonSituations":"Schema evolved (columns dropped/reordered) after the sink was created so cached pk indices no longer align with the parquet root schema; compaction resolver mixing files written under an older schema; a bug or misconfiguration where pk_indices refer to the stream schema instead of the parquet file schema.","solutions":["Verify the pk_indices passed to the compaction resolver correspond to the current parquet schema; recompute them from the sink's current table schema.","Check whether the Iceberg table schema changed (column drops/reorders) since the data files were written; if so, rewrite or re-register affected files.","Log `parquet_schema.root_schema().get_fields().len()` and the full pk_indices at the error site to confirm which index is stale.","If this arises from a RisingWave upgrade, check release notes for schema/pk-index mapping changes in the iceberg sink and resink the affected table."],"exampleFix":"// before: blindly using stream-schema indices\nlet (projection, pk_order) = pk_projection(builder.parquet_schema(), &pk_indices)?;\n\n// after: validate against the actual file schema first\nlet root_count = builder.parquet_schema().root_schema().get_fields().len();\nlet pk_indices: Vec<usize> = pk_indices.iter().copied().filter(|&i| i < root_count).collect();\nlet (projection, pk_order) = pk_projection(builder.parquet_schema(), &pk_indices)?;","handlingStrategy":"validation","validationCode":"let root_count = parquet_schema.root_schema().get_fields().len();\nif pk_indices.iter().any(|&i| i >= root_count) {\n    return Err(format!(\"pk_indices {:?} exceed {} root columns\", pk_indices, root_count));\n}","typeGuard":"fn pk_indices_in_range(pk_indices: &[usize], root_count: usize) -> bool {\n    pk_indices.iter().all(|&i| i < root_count)\n}","tryCatchPattern":null,"preventionTips":["Recompute pk_indices from the current table schema instead of caching them across schema changes.","Assert pk_indices.len() <= root_count and monotonic ordering when building the sink.","Log schema fingerprint (column count + names) alongside pk_indices for diagnosability."],"tags":["parquet","schema","iceberg-sink","compaction"],"backgroundTag":"index-out-of-range","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}