From ec460683c9d3e3f85455a4c6fd525e4c6514d3aa Mon Sep 17 00:00:00 2001 From: Anas Khan <83116240+anxkhn@users.noreply.github.com> Date: Mon, 6 Jul 2026 22:27:08 +0530 Subject: [PATCH] MINOR: Close already-opened input readers when ParquetRewriter setup fails ParquetRewriter.getFileReaders opens a TransParquetFileReader for each input file into a local list. If opening a later file throws IOException, it was rethrown as IllegalArgumentException without closing the readers already opened for the earlier files, leaking their SeekableInputStream handles (ParquetRewriter.close() only ends the writer and does not close input readers). One corrupt, missing, or permission-denied file among several valid ones could therefore leak up to N open input streams per failed rewrite/merge/join setup. Close the readers already opened before rethrowing, using AutoCloseables.close so every reader is released and any close failures are aggregated as suppressed exceptions on the original IllegalArgumentException. Add a ParquetRewriterTest case that supplies a valid input file followed by one that fails to open and asserts the first reader's stream is closed. Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com> --- .../hadoop/rewrite/ParquetRewriter.java | 12 ++- .../hadoop/rewrite/ParquetRewriterTest.java | 80 +++++++++++++++++++ 2 files changed, 91 insertions(+), 1 deletion(-) diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java index 88e41626dd..45fa7d4ede 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java @@ -87,6 +87,7 @@ import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.PrimitiveType; import org.apache.parquet.schema.Type; +import org.apache.parquet.util.AutoCloseables; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -336,7 +337,16 @@ private Queue getFileReaders(List inputFiles, inputFile, ParquetReadOptions.builder(conf).build()); inputFileReaders.add(reader); } catch (IOException e) { - throw new IllegalArgumentException("Failed to open input file: " + inputFile, e); + IllegalArgumentException failure = + new IllegalArgumentException("Failed to open input file: " + inputFile, e); + // Close the readers already opened so their input streams do not leak, aggregating any + // close failures as suppressed exceptions on the original failure. + try { + AutoCloseables.close(inputFileReaders); + } catch (Throwable t) { + failure.addSuppressed(t); + } + throw failure; } } return inputFileReaders; diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/rewrite/ParquetRewriterTest.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/rewrite/ParquetRewriterTest.java index f836feec55..ea8754aa6b 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/rewrite/ParquetRewriterTest.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/rewrite/ParquetRewriterTest.java @@ -49,6 +49,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.BiFunction; import java.util.function.Function; import java.util.stream.Collectors; @@ -91,6 +92,7 @@ import org.apache.parquet.hadoop.util.TestFileBuilder; import org.apache.parquet.internal.column.columnindex.ColumnIndex; import org.apache.parquet.internal.column.columnindex.OffsetIndex; +import org.apache.parquet.io.DelegatingSeekableInputStream; import org.apache.parquet.io.InputFile; import org.apache.parquet.io.InvalidRecordException; import org.apache.parquet.io.OutputFile; @@ -689,6 +691,84 @@ public void testMergeTwoFilesNullifyAndRenamedSameColumn() throws Exception { .hasMessage("Cannot nullify and rename the same column"); } + @Test + public void testInputFileReadersClosedWhenLaterInputFileFailsToOpen() throws Exception { + MessageType schema = new MessageType("schema", new PrimitiveType(OPTIONAL, INT64, "DocId")); + EncryptionTestFile file = new TestFileBuilder(conf, schema) + .withNumRecord(numRecord) + .withCodec("UNCOMPRESSED") + .withPageSize(ParquetProperties.DEFAULT_PAGE_SIZE) + .withWriterVersion(writerVersion) + .build(); + + CloseRecordingInputFile openable = + new CloseRecordingInputFile(HadoopInputFile.fromPath(new Path(file.getFileName()), conf)); + InputFile failing = new InputFile() { + @Override + public long getLength() { + return 0; + } + + @Override + public SeekableInputStream newStream() throws IOException { + throw new IOException("simulated open failure"); + } + }; + + OutputFile output = HadoopOutputFile.fromPath(new Path(outputFile), conf); + RewriteOptions options = + new RewriteOptions.Builder(parquetConf, Arrays.asList(openable, failing), output).build(); + + assertThrows(IllegalArgumentException.class, () -> new ParquetRewriter(options)); + // The reader opened for the first (valid) input file must be closed, not leaked, when a later + // input file fails to open. + assertTrue("Stream of the successfully opened input file was leaked", openable.isStreamClosed()); + } + + /** + * An {@link InputFile} that delegates to another one but records whether the {@link SeekableInputStream} it handed + * out was closed, so a test can assert that already-opened readers are released on a later failure. + */ + private static class CloseRecordingInputFile implements InputFile { + private final InputFile delegate; + private final AtomicBoolean streamClosed = new AtomicBoolean(false); + + CloseRecordingInputFile(InputFile delegate) { + this.delegate = delegate; + } + + boolean isStreamClosed() { + return streamClosed.get(); + } + + @Override + public long getLength() throws IOException { + return delegate.getLength(); + } + + @Override + public SeekableInputStream newStream() throws IOException { + SeekableInputStream stream = delegate.newStream(); + return new DelegatingSeekableInputStream(stream) { + @Override + public long getPos() throws IOException { + return stream.getPos(); + } + + @Override + public void seek(long newPos) throws IOException { + stream.seek(newPos); + } + + @Override + public void close() throws IOException { + streamClosed.set(true); + super.close(); + } + }; + } + } + public void testMergeTwoFilesWithDifferentSchemaSetup( Boolean wrongSchemaInInputFile, Map renameColumns, Map maskColumns) throws Exception {