{"record":{"id":"b314bd5a0ef0122e","repo":"risingwavelabs/risingwave","slug":"open-parquet-reader-for-path","errorCode":null,"errorMessage":"open parquet reader for {path}","messagePattern":"open parquet reader for (.+?)","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs","lineNumber":420,"sourceCode":"        .collect();\n    let projection = ProjectionMask::roots(parquet_schema, physical_indices);\n\n    Ok((projection, pk_order))\n}\n\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() {","sourceCodeStart":402,"sourceCodeEnd":438,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs#L402-L438","documentation":"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.","triggerScenarios":"`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).","commonSituations":"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.","solutions":["Check the error chain (`.context` preserves the source) to identify the root cause: NotFound vs permission vs corrupt footer.","Verify the file exists at `path` in the object store and was not deleted/expired by a lifecycle rule.","Validate object-store credentials, endpoint and region in the sink/connection configuration.","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."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"let metadata = file_io.metadata(path).await.map_err(|e| format!(\"input file missing/unreadable: {path}: {e}\"))?;","typeGuard":null,"tryCatchPattern":"match scan_input_pks_at_positions(...).await {\n    Ok(pks) => pks,\n    Err(e) if e.to_string().contains(\"open parquet reader\") => {\n        // inspect root cause: NotFound / PermissionDenied / corrupt footer\n        return Err(CompactionError::UnreadableInput(path.to_string(), e));\n    }\n    Err(e) => return Err(e.into()),\n}","preventionTips":["Pre-check file existence and size before compaction scans.","Exclude objects with lifecycle expiration from compaction input planning.","Validate object-store credentials/endpoint with a cheap HEAD call at sink startup.","Verify parquet footer readability when files are written by external tools."],"tags":["parquet","object-storage","iceberg-sink","compaction"],"backgroundTag":"file-open-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}