risingwavelabs/risingwave · error · SinkError::Iceberg

invalid extra partition column index

Error message

invalid extra partition column index {}

What it means

`process_chunk` rejected a pre-computed extra-partition column projection index that is out of bounds for the current StreamChunk. When `ProjectIdxVec::Prepare(idx)` is set (the extra partition column still physically present in the chunk), `idx` must point at an existing column so it can be split out for partitioned writing. RisingWave raises `SinkError::Iceberg` with the offending index.

Solutions

1. Recreate the sink so `project_idx_vec` is recomputed against the current MV projection.
2. Verify the MV/sink column list matches the configured extra partition columns in the sink WITH options.
3. If reproducible on a fresh sink, report it as an internal bug with the chunk column count and stored index.

// Guard the stored index against the chunk arity before using it
if let ProjectIdxVec::Prepare(idx) = &self.project_idx_vec {
    anyhow::ensure!(
        *idx < chunk.columns().len(),
        "extra partition column index {} out of bounds for chunk with {} columns; recreate the sink",
        idx, chunk.columns().len()
    );
}

fn idx_in_bounds(idx: &usize, len: usize) -> bool { idx < &len }

// Wrap the write so an arity mismatch triggers sink re-creation rather than silent data loss
match sink.write_batch(chunk).await {
    Err(e) if e.to_string().contains("invalid extra partition column index") => {
        recreate_sink().await?;
    }
    other => other?,
}

When it happens

Trigger: `write_batch`/`write_batch_with_position` delivering a chunk with fewer columns than the stored partition-projection index expects — typically after the MV/sink projection changed while the writer still holds a stale index, or an internal bug computing `project_idx_vec`.

Common situations: Altering the MV or sink column list while the sink is running; sink created with extra partition columns whose index no longer matches the chunk arity; internal invariant breakage in chunk column count (e.g. degenerate single-column chunks).

Related errors

• backfill iceberg pk-index sink delete files failed…
• Can't create iceberg sink write result from empty data!
• failed to add partition columns
• Partition source column does not exist in schema
• primary key column index

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

Appendix: source

Thrown at src/connector/src/sink/iceberg/writer.rs:702

                if writer.is_none() {
                    *writer = Some(Box::new(
                        writer_builder
                            .build()
                            .map_err(|err| SinkError::Iceberg(anyhow!(err)))?,
                    ));
                }
            }
        };
        Ok(())
    }

    fn process_chunk(&mut self, chunk: StreamChunk) -> Result<Option<(RecordBatch, usize)>> {
        let (mut chunk, ops) = chunk.compact_vis().into_parts();
        match &mut self.project_idx_vec {
            ProjectIdxVec::None => {}
            ProjectIdxVec::Prepare(idx) => {
                if *idx >= chunk.columns().len() {
                    return Err(SinkError::Iceberg(anyhow!(
                        "invalid extra partition column index {}",
                        idx
                    )));
                }
                let project_idx_vec = (0..*idx)
                    .chain(*idx + 1..chunk.columns().len())
                    .collect_vec();
                chunk = chunk.project(&project_idx_vec);
                self.project_idx_vec = ProjectIdxVec::Done(project_idx_vec);
            }
            ProjectIdxVec::Done(idx_vec) => {
                chunk = chunk.project(idx_vec);
            }
        }
        if ops.is_empty() {
            return Ok(None);
        }
        let write_batch_size = chunk.estimated_heap_size();

View on GitHub (pinned to 6469eb736d)