From 201ee57fcd1bfc6f1a753e6ef6d6db51b554a627 Mon Sep 17 00:00:00 2001 From: Yigit Colakoglu Date: Tue, 4 Aug 2026 15:55:15 +0200 Subject: [PATCH 1/2] GH-3706: Avoid flushing incomplete files when finalization fails --- .../parquet/hadoop/ParquetFileWriter.java | 18 +- .../parquet/hadoop/TestParquetFileWriter.java | 156 ++++++++++++++++++ 2 files changed, 166 insertions(+), 8 deletions(-) diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java index 82f4577b83..6cb4b50335 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java @@ -344,7 +344,7 @@ interface IOCallable { private T withAbortOnFailure(IOCallable action) throws IOException { try { return action.call(); - } catch (IOException e) { + } catch (Throwable e) { aborted = true; throw e; } @@ -358,7 +358,7 @@ interface IORunnable { private void withAbortOnFailure(IORunnable action) throws IOException { try { action.run(); - } catch (IOException e) { + } catch (Throwable e) { aborted = true; throw e; } @@ -1872,8 +1872,10 @@ private static void copy(SeekableInputStream from, PositionOutputStream to, long * @throws IOException if there is an error while writing */ public void end(Map extraMetaData) throws IOException { - withAbortOnFailure(() -> { - try { + // withAbortOnFailure marks the writer aborted on any throwable before the outer finally runs + // close(), so an incomplete file is never flushed if finalizing the file fails. + try { + withAbortOnFailure(() -> { state = state.end(); serializeColumnIndexes(columnIndexes, blocks, out, fileEncryptor); serializeOffsetIndexes(offsetIndexes, blocks, out, fileEncryptor); @@ -1882,10 +1884,10 @@ public void end(Map extraMetaData) throws IOException { this.footer = new ParquetMetadata(new FileMetaData(schema, extraMetaData, Version.FULL_VERSION), blocks); serializeFooter(footer, out, fileEncryptor, metadataConverter); - } finally { - close(); - } - }); + }); + } finally { + close(); + } } /* Mark the writer as aborted to avoid flushing incomplete data. */ diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java index 69e6bc3f2a..d153d71d44 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java @@ -98,7 +98,9 @@ import org.apache.parquet.internal.column.columnindex.BoundaryOrder; import org.apache.parquet.internal.column.columnindex.ColumnIndex; import org.apache.parquet.internal.column.columnindex.OffsetIndex; +import org.apache.parquet.io.OutputFile; import org.apache.parquet.io.ParquetEncodingException; +import org.apache.parquet.io.PositionOutputStream; import org.apache.parquet.io.api.Binary; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.MessageTypeParser; @@ -1476,4 +1478,158 @@ private Path newTempPath() { private Path existingTempPath() throws IOException { return new Path(Files.createTempFile(tempDir, "test", ".tmp").toUri()); } + + @Test + public void testNoFlushWhenEndFailsWithIOException() throws Exception { + assertEndFailureDoesNotFlush(new IOException("injected end() failure")); + } + + @Test + public void testNoFlushWhenEndFailsWithRuntimeException() throws Exception { + // Validate that non-IOException failures are also caught and mark the writer aborted. + assertEndFailureDoesNotFlush(new RuntimeException("injected end() failure")); + } + + private void assertEndFailureDoesNotFlush(Throwable toThrow) throws Exception { + Path path = newTempPath(); + Configuration conf = getTestConfiguration(false); + + FaultInjectingOutputFile out = new FaultInjectingOutputFile(HadoopOutputFile.fromPath(path, conf)); + ParquetFileWriter w = new ParquetFileWriter( + out, + SCHEMA, + ParquetFileWriter.Mode.CREATE, + DEFAULT_BLOCK_SIZE, + MAX_PADDING_SIZE_DEFAULT, + null, + ParquetProperties.builder().withAllocator(allocator).build()); + + // Write one minimal, complete block. + w.start(); + w.startBlock(1); + w.startColumn(C1, 1, CODEC); + w.writeDataPage(1, 4, BytesInput.from(BYTES1), EMPTY_STATS, BIT_PACKED, BIT_PACKED, PLAIN); + w.endColumn(); + w.startColumn(C2, 1, CODEC); + w.writeDataPage(1, 4, BytesInput.from(BYTES2), EMPTY_STATS, BIT_PACKED, BIT_PACKED, PLAIN); + w.endColumn(); + w.endBlock(); + + // Arm the fault so serialization inside end() throws, then reset the flush counter so we only + // measure flushes that happen during the failing end()/close(). + out.flushCount = 0; + out.toThrow = toThrow; + + try { + w.end(new HashMap()); + throw new AssertionError("Expected the injected failure from end()"); + } catch (IOException | RuntimeException expected) { + // A RuntimeException propagates unwrapped; an IOException is wrapped by serializeFooter + // (Util.writeFileMetaData -> "can not write"), so unwrap to the injected root cause. + Throwable root = expected; + while (root.getCause() != null) { + root = root.getCause(); + } + assertThat(root.getMessage()).isEqualTo("injected end() failure"); + } finally { + // end()'s abort path intentionally leaves the underlying stream unclosed (so the incomplete + // file is never committed); close it here to make the resource contract explicit. Swallow any + // close failure so it can't mask the flushCount assertion below. + try { + out.stream.close(); + } catch (IOException ignored) { + } + } + + // The writer is marked aborted on failure before close() runs, so close() skips the flush and + // the incomplete stream is never flushed to storage. + assertThat(out.flushCount) + .as("end() must not flush the output stream when it fails") + .isEqualTo(0); + } + + /** + * OutputFile whose stream throws a preset Throwable on write (simulating a mid-write failure) and + * which counts flush() calls, so a test can assert nothing was flushed on the failure path. + */ + private static class FaultInjectingOutputFile implements OutputFile { + private final OutputFile delegate; + volatile Throwable toThrow = null; // when non-null, the next write(s) throw it + int flushCount = 0; + PositionOutputStream stream; // the most recently created wrapping stream + + FaultInjectingOutputFile(OutputFile delegate) { + this.delegate = delegate; + } + + private void failIfArmed() throws IOException { + Throwable t = toThrow; + if (t == null) { + return; + } + if (t instanceof IOException) { + throw (IOException) t; + } + if (t instanceof RuntimeException) { + throw (RuntimeException) t; + } + if (t instanceof Error) { + throw (Error) t; + } + throw new IOException(t); + } + + private PositionOutputStream wrap(PositionOutputStream out) { + stream = new PositionOutputStream() { + @Override + public long getPos() throws IOException { + return out.getPos(); + } + + @Override + public void write(int b) throws IOException { + failIfArmed(); + out.write(b); + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + failIfArmed(); + out.write(b, off, len); + } + + @Override + public void flush() throws IOException { + flushCount++; + out.flush(); + } + + @Override + public void close() throws IOException { + out.close(); + } + }; + return stream; + } + + @Override + public PositionOutputStream create(long blockSizeHint) throws IOException { + return wrap(delegate.create(blockSizeHint)); + } + + @Override + public PositionOutputStream createOrOverwrite(long blockSizeHint) throws IOException { + return wrap(delegate.createOrOverwrite(blockSizeHint)); + } + + @Override + public boolean supportsBlockSize() { + return delegate.supportsBlockSize(); + } + + @Override + public long defaultBlockSize() { + return delegate.defaultBlockSize(); + } + } } From d5af1cabc355c40606f88f40f73f93edaca641a9 Mon Sep 17 00:00:00 2001 From: Yigit Colakoglu Date: Fri, 2 Oct 2026 15:26:05 +0200 Subject: [PATCH 2/2] GH-3706: Catch Throwable in InternalParquetRecordWriter.close() to abort on Error Address review feedback on #3707: close() caught only Exception, so an Error thrown outside ParquetFileWriter (e.g. OOM during flushRowGroupToStore, or finalizeWrite) left the writer un-aborted and the finally flushed an incomplete, footerless file. Match write()'s catch (Throwable t) so abort runs before close(). --- .../hadoop/InternalParquetRecordWriter.java | 4 +- .../parquet/hadoop/TestParquetFileWriter.java | 50 +++++++++++++++++++ 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java index d503cbb4eb..4c43b0652a 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java @@ -143,9 +143,9 @@ public void close() throws IOException, InterruptedException { } finalMetadata.putAll(finalWriteContext.getExtraMetaData()); parquetFileWriter.end(finalMetadata); - } catch (Exception e) { + } catch (Throwable t) { parquetFileWriter.abort(); - throw e; + throw t; } finally { AutoCloseables.uncheckedClose(columnStore, pageStore, bloomFilterWriteStore, parquetFileWriter); closed = true; diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java index d153d71d44..cfe4a1e5bd 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetFileWriter.java @@ -78,10 +78,13 @@ import org.apache.parquet.column.values.bloomfilter.BloomFilter; import org.apache.parquet.example.data.Group; import org.apache.parquet.example.data.simple.SimpleGroup; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; import org.apache.parquet.format.Statistics; import org.apache.parquet.hadoop.ParquetOutputFormat.JobSummaryLevel; import org.apache.parquet.hadoop.example.GroupReadSupport; import org.apache.parquet.hadoop.example.GroupWriteSupport; +import org.apache.parquet.hadoop.api.WriteSupport; +import org.apache.parquet.hadoop.api.WriteSupport.FinalizedWriteContext; import org.apache.parquet.hadoop.metadata.BlockMetaData; import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; import org.apache.parquet.hadoop.metadata.CompressionCodecName; @@ -98,6 +101,7 @@ import org.apache.parquet.internal.column.columnindex.BoundaryOrder; import org.apache.parquet.internal.column.columnindex.ColumnIndex; import org.apache.parquet.internal.column.columnindex.OffsetIndex; +import org.apache.parquet.io.LocalOutputFile; import org.apache.parquet.io.OutputFile; import org.apache.parquet.io.ParquetEncodingException; import org.apache.parquet.io.PositionOutputStream; @@ -1548,6 +1552,52 @@ private void assertEndFailureDoesNotFlush(Throwable toThrow) throws Exception { .isEqualTo(0); } + @Test + public void testNoFlushWhenRecordWriterCloseFailsWithError() throws Exception { + // An Error thrown from close() outside ParquetFileWriter (e.g. finalizeWrite) must abort + // before the finally flushes, so an incomplete file is never committed. + java.nio.file.Path path = tempDir.resolve(java.util.UUID.randomUUID() + ".tmp"); + Configuration conf = getTestConfiguration(false); + + FaultInjectingOutputFile out = new FaultInjectingOutputFile(new LocalOutputFile(path)); + MessageType schema = MessageTypeParser.parseMessageType("message m { required int64 id; }"); + WriteSupport writeSupport = new GroupWriteSupport() { + @Override + public FinalizedWriteContext finalizeWrite() { + throw new AssertionError("injected finalizeWrite() failure"); + } + }; + GroupWriteSupport.setSchema(schema, conf); + + ParquetWriter writer = new ParquetWriter( + out, + ParquetFileWriter.Mode.CREATE, + writeSupport, + CompressionCodecName.UNCOMPRESSED, + DEFAULT_BLOCK_SIZE, + false, + conf, + MAX_PADDING_SIZE_DEFAULT, + ParquetProperties.builder().withAllocator(allocator).build(), + null); + + // Write enough rows to form a row group, so close() flushes pages before finalizeWrite(). + SimpleGroupFactory groupFactory = new SimpleGroupFactory(schema); + for (int i = 0; i < 100; i++) { + writer.write(groupFactory.newGroup().append("id", (long) i)); + } + + out.flushCount = 0; + assertThatThrownBy(writer::close) + .isInstanceOf(AssertionError.class) + .hasMessageContaining("injected finalizeWrite() failure"); + + // Aborted before close(), so the flush is skipped and the incomplete file is never committed. + assertThat(out.flushCount) + .as("close() must not flush the output stream when finalizeWrite() fails with an Error") + .isEqualTo(0); + } + /** * OutputFile whose stream throws a preset Throwable on write (simulating a mid-write failure) and * which counts flush() calls, so a test can assert nothing was flushed on the failure path.