risingwavelabs/risingwave · error · SinkError

build parquet stream for {path}

Error message

build parquet stream for {path}

What it means

Thrown in `scan_input_pks_at_positions` after the parquet reader is opened but `builder.with_projection(projection).build()` fails while constructing the record-batch stream. The projection mask was already computed from the file's schema, so this error points to a malformed or unreadable parquet file structure (bad page index, column chunk metadata, or unsupported encoding) rather than a schema/pk mismatch.

Source

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

async fn scan_input_pks_at_positions(
    file_io: &FileIO,
    path: &str,
    pk_indices: &[usize],
    want_positions: &DeleteVector,
) -> Result<Vec<OwnedRow>, SinkError> {
    let mut results = Vec::new();
    let input_file = file_io.new_input(path)?;
    let metadata = input_file.metadata().await?;
    let reader = input_file.reader().await?;
    let builder = ParquetRecordBatchStreamBuilder::new(ParquetFileReader::new(metadata, reader))
        .await
        .map_err(|e| anyhow!(e).context(format!("open parquet reader for {path}")))?;
    let (projection, pk_order) = pk_projection(builder.parquet_schema(), pk_indices)?;
    let mut stream = builder
        .with_projection(projection)
        .build()
        .map_err(|e| anyhow!(e).context(format!("build parquet stream for {path}")))?;

    let mut base_pos = 0;
    let mut iter = want_positions.iter().peekable();
    while let Some(batch) = stream.next().await {
        let batch =
            batch.map_err(|e| anyhow!(e).context(format!("read parquet batch of {path}")))?;
        let chunk = IcebergArrowConvert
            .chunk_from_record_batch(&batch)
            .map_err(|e| anyhow!(e).context(format!("convert parquet batch of {path}")))?
            .project(&pk_order);
        while let Some(&iter_pos) = iter.peek() {
            let chunk_pos = iter_pos as usize - base_pos;
            if chunk_pos >= chunk.capacity() {
                break;
            }
            let pk = chunk.row_at(chunk_pos).0.to_owned_row();
            results.push(pk);
            iter.next();

View on GitHub (pinned to 6469eb736d)

Solutions

1. Inspect the underlying error in the chain for the failing column/encoding; if unsupported, rewrite the file with compatible encodings.
2. Check file integrity: compare object size/ETag against the Iceberg manifest entry to detect truncation.
3. Remove or rewrite the corrupt file via an Iceberg rewrite/repair action so compaction no longer scans it.
4. Retry once in case the object store returned a transient bad read; persistent failure indicates corruption.

let stream = match builder.with_projection(projection).build() {
    Ok(s) => s,
    Err(e) => return Err(CompactionError::BadParquetFile(path.to_string(), e)),
};

When it happens

Trigger: `resolve` → `scan_input_pks_at_positions` builds a projected `ParquetRecordBatchStream` for an input file of the compaction. Triggered by corrupt column-chunk metadata, an unsupported parquet encoding/compression in the projection, or the file having been modified/truncated after the footer was read.

Common situations: Data file truncated by an interrupted upload to object storage; files written by an external writer (Spark/Trino) using encodings this reader does not support; concurrent overwrite of the object by a data-repair job.

Understand the failure class

Background: "failed to read file", EACCES, ENOENT and "could not read <path>" errors: when a program can't read a file from disk — this error's family across 49 libraries.

Related errors

• open parquet reader for {path}
• read parquet batch of {path}
• pk index {pk_index} is out of range for parquet schema with
• duplicate pk index {pk_index}
• convert parquet batch of {path}

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