{"record":{"id":"548fcf37f6909b0d","repo":"apache/flink","slug":"failed-to-read-from-input-stream","errorCode":null,"errorMessage":"Failed to read from input stream","messagePattern":"Failed to read from input stream","errorType":"exception","errorClass":"ParquetDecodingException","httpStatus":null,"severity":"error","filePath":"flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/RunLengthDecoder.java","lineNumber":286,"sourceCode":"                        if (buffer.hasArray()) {\n                            // byte array has better performance than ByteBuffer\n                            this.packer.unpack8Values(\n                                    buffer.array(),\n                                    buffer.arrayOffset() + buffer.position(),\n                                    this.currentBuffer,\n                                    valueIndex);\n                        } else {\n                            this.packer.unpack8Values(\n                                    buffer, buffer.position(), this.currentBuffer, valueIndex);\n                        }\n                        valueIndex += 8;\n                    }\n                    return;\n                default:\n                    throw new ParquetDecodingException(\"not a valid mode \" + this.mode);\n            }\n        } catch (IOException e) {\n            throw new ParquetDecodingException(\"Failed to read from input stream\", e);\n        }\n    }\n\n    enum MODE {\n        RLE,\n        PACKED\n    }\n}\n","sourceCodeStart":268,"sourceCodeEnd":295,"githubUrl":"https://github.com/apache/flink/blob/2f3c205e9266cb30240eb7f4fdab15cad629a70f/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/RunLengthDecoder.java#L268-L295","documentation":"Thrown by Flink's Parquet vectorized reader when the underlying RLE/bit-packing decoder (used for definition levels, repetition levels, or dictionary-encoded values) hits an IOException while reading bytes from the column page input stream. The ParquetDecodingException wraps the original IOException, so the real cause is always in the 'caused by' chain. It almost always means the byte stream ended prematurely, the page data is corrupt, or the column encoding cannot be decoded with the current reader implementation.","triggerScenarios":"Reading a Parquet file whose column chunk pages are truncated or corrupt (interrupted write, truncated FTP/HDFS upload); reading a file with a dictionary/RLE encoding variant the reader cannot decode; reading from a split whose byte range is wrong (bad InputSplit offsets); a schema/encoding produced by a Parquet writer version incompatible with the bundled parquet-mr.","commonSituations":"Jobs reading files that were still being written when the query started; files copied incompletely; compressed blocks damaged by transfer; Parquet files written by Spark/Impala/Hive with encodings the Flink parquet reader mishandles; FS connectors with wrong file size metadata.","solutions":["Check the wrapped cause: catch ParquetDecodingException and inspect getCause() — EOFException means truncated file, CRC/codec errors mean corruption.","Verify the file is complete and readable with an external tool (parquet-tools / 'parquet cat') on the same storage.","Re-produce or re-upload the affected Parquet file; if the producer was interrupted mid-write, regenerate it.","If the file is valid, test with a recent parquet-mr writer/reader combination and check Flink JIRAs for encoding-specific reader bugs; report with a reproducer.","As a job-level guard, skip bad files/splits via a custom reader wrapper or isolate the failing file by binary-searching the input set."],"exampleFix":"// before: silently failing task\ntry {\n    columnReader.readBatch();\n} catch (ParquetDecodingException e) {\n    throw e;\n}\n// after: surface root cause\ntry {\n    columnReader.readBatch();\n} catch (ParquetDecodingException e) {\n    Throwable root = e.getCause();\n    LOG.error(\"Parquet decode failed, likely corrupt/truncated page: {}\", root, e);\n    throw e;\n}","handlingStrategy":"try-catch","validationCode":"// Validate file integrity before reading (best-effort)\nParquetFileReader reader = ParquetFileReader.open(conf, path);\nif (reader.getFooter().getBlocks().isEmpty()) {\n    throw new IOException(\"Empty/corrupt parquet footer: \" + path);\n}","typeGuard":null,"tryCatchPattern":"try {\n    vectorizedReader.readBatch();\n} catch (ParquetDecodingException e) {\n    if (e.getCause() instanceof EOFException) {\n        // truncated file: quarantine and skip, or fail the split\n        LOG.warn(\"Truncated parquet split: {}\", split, e);\n        throw e;\n    }\n    throw e;\n}","preventionTips":["Only expose fully-written files to Flink (write to a temp directory, atomically rename / use the '_'-prefix convention that readers skip).","Verify checksums after transferring parquet files between storage systems.","Log the wrapped IOException cause, not just the wrapper, when triaging."],"tags":["parquet","io","data-corruption","flink-formats"],"backgroundTag":null,"analyzedSha":"2f3c205e9266cb30240eb7f4fdab15cad629a70f","analyzedAt":"2026-08-14T08:48:24.518Z","schemaVersion":2},"datasetVersion":"2026-08-14T10:17:34.591Z"}