{"record":{"id":"3bab9572f7133d9d","repo":"risingwavelabs/risingwave","slug":"build-parquet-stream-for-path","errorCode":null,"errorMessage":"build parquet stream for {path}","messagePattern":"build parquet stream for (.+?)","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs","lineNumber":425,"sourceCode":"\nasync fn scan_input_pks_at_positions(\n    file_io: &FileIO,\n    path: &str,\n    pk_indices: &[usize],\n    want_positions: &DeleteVector,\n) -> Result<Vec<OwnedRow>, SinkError> {\n    let mut results = Vec::new();\n    let input_file = file_io.new_input(path)?;\n    let metadata = input_file.metadata().await?;\n    let reader = input_file.reader().await?;\n    let builder = ParquetRecordBatchStreamBuilder::new(ParquetFileReader::new(metadata, reader))\n        .await\n        .map_err(|e| anyhow!(e).context(format!(\"open parquet reader for {path}\")))?;\n    let (projection, pk_order) = pk_projection(builder.parquet_schema(), pk_indices)?;\n    let mut stream = builder\n        .with_projection(projection)\n        .build()\n        .map_err(|e| anyhow!(e).context(format!(\"build parquet stream for {path}\")))?;\n\n    let mut base_pos = 0;\n    let mut iter = want_positions.iter().peekable();\n    while let Some(batch) = stream.next().await {\n        let batch =\n            batch.map_err(|e| anyhow!(e).context(format!(\"read parquet batch of {path}\")))?;\n        let chunk = IcebergArrowConvert\n            .chunk_from_record_batch(&batch)\n            .map_err(|e| anyhow!(e).context(format!(\"convert parquet batch of {path}\")))?\n            .project(&pk_order);\n        while let Some(&iter_pos) = iter.peek() {\n            let chunk_pos = iter_pos as usize - base_pos;\n            if chunk_pos >= chunk.capacity() {\n                break;\n            }\n            let pk = chunk.row_at(chunk_pos).0.to_owned_row();\n            results.push(pk);\n            iter.next();","sourceCodeStart":407,"sourceCodeEnd":443,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs#L407-L443","documentation":"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.","triggerScenarios":"`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.","commonSituations":"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.","solutions":["Inspect the underlying error in the chain for the failing column/encoding; if unsupported, rewrite the file with compatible encodings.","Check file integrity: compare object size/ETag against the Iceberg manifest entry to detect truncation.","Remove or rewrite the corrupt file via an Iceberg rewrite/repair action so compaction no longer scans it.","Retry once in case the object store returned a transient bad read; persistent failure indicates corruption."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"let stream = match builder.with_projection(projection).build() {\n    Ok(s) => s,\n    Err(e) => return Err(CompactionError::BadParquetFile(path.to_string(), e)),\n};","preventionTips":["Compare object size/ETag against the Iceberg manifest to detect truncated files before scanning.","Standardize writer encodings/versions for files this reader must consume.","Add a one-shot retry for transient object-store errors before failing compaction."],"tags":["parquet","object-storage","iceberg-sink","compaction"],"backgroundTag":"file-read-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}