risingwavelabs/risingwave · error · SinkError

read parquet batch of {path}

Error message

read parquet batch of {path}

What it means

This error context is attached in `scan_input_pks_at_positions` when a `RecordBatch` read from the parquet stream fails while iterating `stream.next()`. The file opened fine and the stream was built; a specific batch of rows could not be decoded (bad page, checksum/dictionary corruption, or an I/O error mid-stream). The path in the message identifies which data file contains the bad batch.

Source

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

) -> 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();
        }
        base_pos += chunk.capacity();
    }

    Ok(results)
}

View on GitHub (pinned to 6469eb736d)

Solutions

1. Read the source error to distinguish I/O (retryable) from decode corruption (file is bad).
2. If retryable, retry the compaction resolve; transient S3 errors often clear.
3. If corrupt, restore the file from backup or drop/rewrite it via Iceberg repair, then re-run compaction.
4. Enable object-store integrity checks (ETag/ checksum validation) to catch truncation early.

let mut attempt = 0;
loop {
    match scan_input_pks_at_positions(...).await {
        Ok(v) => break Ok(v),
        Err(e) if is_transient_io(&e) && attempt < 2 => { attempt += 1; continue; }
        Err(e) => break Err(e),
    }
}

When it happens

Trigger: `resolve` → `scan_input_pks_at_positions` iterates a projected parquet stream to locate rows at `want_positions`. Any batch decode failure during iteration produces this message: corrupt data pages, truncated file body (footer readable but data incomplete), or a mid-stream object-store I/O error.

Common situations: Interrupted multipart upload leaving a file with valid footer but missing data pages; disk/network failure while streaming from S3/GCS; bit-rot or failed checksum on a long-lived object.

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}
• build parquet stream for {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/40b8921054053400. Report an issue: GitHub.