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(); + } + } }