risingwavelabs/risingwave · error · SinkError

pk index {pk_index} is out of range for parquet schema with

Error message

pk index {pk_index} is out of range for parquet schema with {root_count} root columns

What it means

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.

Source

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

            }
        }
    }
    Ok(map)
}

/// 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();

View on GitHub (pinned to 6469eb736d)

Solutions

1. Verify the pk_indices passed to the compaction resolver correspond to the current parquet schema; recompute them from the sink's current table schema.
2. Check whether the Iceberg table schema changed (column drops/reorders) since the data files were written; if so, rewrite or re-register affected files.
3. Log `parquet_schema.root_schema().get_fields().len()` and the full pk_indices at the error site to confirm which index is stale.
4. 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.

Example fix

// before: blindly using stream-schema indices
let (projection, pk_order) = pk_projection(builder.parquet_schema(), &pk_indices)?;

// after: validate against the actual file schema first
let root_count = builder.parquet_schema().root_schema().get_fields().len();
let pk_indices: Vec<usize> = pk_indices.iter().copied().filter(|&i| i < root_count).collect();
let (projection, pk_order) = pk_projection(builder.parquet_schema(), &pk_indices)?;

let root_count = parquet_schema.root_schema().get_fields().len();
if pk_indices.iter().any(|&i| i >= root_count) {
    return Err(format!("pk_indices {:?} exceed {} root columns", pk_indices, root_count));
}

fn pk_indices_in_range(pk_indices: &[usize], root_count: usize) -> bool {
    pk_indices.iter().all(|&i| i < root_count)
}

When it happens

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

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

Related errors

• duplicate pk index {pk_index}
• open parquet reader for {path}
• build parquet stream for {path}
• read parquet batch of {path}
• {message}

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