risingwavelabs/risingwave · error · SinkError

duplicate pk index {pk_index}

Error message

duplicate pk index {pk_index}

What it means

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.

Source

Thrown at src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs:389

/// Builds a projection for PK root columns and maps the projected physical order back to the
/// downstream PK order.
fn pk_projection(
    parquet_schema: &SchemaDescriptor,
    pk_indices: &[usize],
) -> Result<(ProjectionMask, Vec<usize>), SinkError> {
    let root_count = parquet_schema.root_schema().get_fields().len();
    let mut physical_indices = Vec::with_capacity(pk_indices.len());
    let mut seen = HashSet::with_capacity(pk_indices.len());

    for &pk_index in pk_indices {
        if pk_index >= root_count {
            return Err(SinkError::Iceberg(anyhow!(
                "pk index {pk_index} is out of range for parquet schema with {root_count} root columns"
            )));
        }
        if !seen.insert(pk_index) {
            return Err(SinkError::Iceberg(anyhow!("duplicate pk index {pk_index}")));
        }
        physical_indices.push(pk_index);
    }

    physical_indices.sort_unstable();
    let pk_order = pk_indices
        .iter()
        .map(|pk_index| {
            physical_indices
                .binary_search(pk_index)
                .expect("validated pk index must be projected")
        })
        .collect();
    let projection = ProjectionMask::roots(parquet_schema, physical_indices);

    Ok((projection, pk_order))
}

View on GitHub (pinned to 6469eb736d)

Solutions

1. Deduplicate the pk_indices before calling the resolver, e.g. via a HashSet as the code itself does.
2. Find where pk_indices are constructed (sink definition / table properties) and fix the duplication at the source.
3. Add an assertion or unit test on pk_indices construction so duplicates are caught before compaction runs.

Example fix

// before
let pk_indices: Vec<usize> = identity_pks.into_iter().chain(upstream_pks).collect();

// after
let pk_indices: Vec<usize> = identity_pks.into_iter().chain(upstream_pks).collect::<HashSet<_>>().into_iter().collect();

let unique: HashSet<usize> = pk_indices.iter().copied().collect();
if unique.len() != pk_indices.len() {
    return Err("pk_indices contain duplicates".to_string());
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• pk index {pk_index} is out of range for parquet schema with
• open parquet reader for {path}
• build parquet stream for {path}
• read parquet batch of {path}
• compact_with_plan returned no result for a non-empty iceberg

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