{"record":{"id":"40b8921054053400","repo":"risingwavelabs/risingwave","slug":"read-parquet-batch-of-path","errorCode":null,"errorMessage":"read parquet batch of {path}","messagePattern":"read parquet batch of (.+?)","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs","lineNumber":431,"sourceCode":") -> 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();\n        }\n        base_pos += chunk.capacity();\n    }\n\n    Ok(results)\n}","sourceCodeStart":413,"sourceCodeEnd":449,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs#L413-L449","documentation":"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.","triggerScenarios":"`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.","commonSituations":"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.","solutions":["Read the source error to distinguish I/O (retryable) from decode corruption (file is bad).","If retryable, retry the compaction resolve; transient S3 errors often clear.","If corrupt, restore the file from backup or drop/rewrite it via Iceberg repair, then re-run compaction.","Enable object-store integrity checks (ETag/ checksum validation) to catch truncation early."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"let mut attempt = 0;\nloop {\n    match scan_input_pks_at_positions(...).await {\n        Ok(v) => break Ok(v),\n        Err(e) if is_transient_io(&e) && attempt < 2 => { attempt += 1; continue; }\n        Err(e) => break Err(e),\n    }\n}","preventionTips":["Distinguish I/O errors (retry) from decode corruption (repair/regenerate) by inspecting the error chain.","Enable checksum/ETag validation on object storage reads.","Avoid referencing files from interrupted uploads; only commit fully uploaded objects."],"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"}