risingwavelabs/risingwave · error · SinkError

open parquet reader for {path}

Error message

open parquet reader for {path}

What it means

This context wrapper is added in `scan_input_pks_at_positions` when `ParquetRecordBatchStreamBuilder::new` fails while opening a parquet data file during compaction conflict resolution. The underlying error (from the object store reader or the parquet footer decode) is preserved and wrapped with the file path so the operator knows which file could not be opened. It is surfaced through `resolve` as part of the compaction resolver's error chain.

Source

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

        .collect();
    let projection = ProjectionMask::roots(parquet_schema, physical_indices);

    Ok((projection, pk_order))
}

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() {

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the error chain (`.context` preserves the source) to identify the root cause: NotFound vs permission vs corrupt footer.
2. Verify the file exists at `path` in the object store and was not deleted/expired by a lifecycle rule.
3. Validate object-store credentials, endpoint and region in the sink/connection configuration.
4. If the file is corrupt or missing, remove it from the Iceberg table's data files (or re-commit the snapshot) so compaction no longer references it.

let metadata = file_io.metadata(path).await.map_err(|e| format!("input file missing/unreadable: {path}: {e}"))?;

match scan_input_pks_at_positions(...).await {
    Ok(pks) => pks,
    Err(e) if e.to_string().contains("open parquet reader") => {
        // inspect root cause: NotFound / PermissionDenied / corrupt footer
        return Err(CompactionError::UnreadableInput(path.to_string(), e));
    }
    Err(e) => return Err(e.into()),
}

When it happens

Trigger: `resolve` scans input compaction files; `scan_input_pks_at_positions` calls `file_io.new_input(path)`, fetches metadata, opens a reader, and builds a `ParquetFileReader`. Failure occurs if the object does not exist, the object-store credentials/endpoint are wrong, the file is truncated/corrupt, or the parquet footer cannot be decoded (not a parquet file, unsupported version).

Common situations: Data file deleted or expired by an external lifecycle policy between planning and compaction; wrong S3/GCS/OSS credentials or region configured for the sink's file_io; network partition to object storage; file written by a newer parquet writer version than the reader supports.

Understand the failure class

Background: "open() failed", "failed to open file", "cannot create file" — what a file open error means and how to fix it — this error's family across 42 libraries.

Related errors

• build parquet stream 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/b314bd5a0ef0122e. Report an issue: GitHub.