jenwitteng commented on PR #3707: URL: https://github.com/apache/parquet-java/pull/3707#issuecomment-5946185074
Thanks, the `end()` reordering makes sense. There's a related gap on the `InternalParquetRecordWriter` side that this PR doesn't reach. It isn't blocking, and I can open a separate issue and PR if you'd rather keep this one as is. `InternalParquetRecordWriter.close()` still catches only `Exception`: https://github.com/apache/parquet-java/blob/921c3571f4086464c2888a9532ca6d925559c241/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java#L146-L151 An `Error` thrown in `close()` outside any `ParquetFileWriter` method never goes through `withAbortOnFailure`. So `aborted` stays `false`, and the `finally` calls `ParquetFileWriter.close()`, which flushes and closes `out`. This happens in two places: - `flushRowGroupToStore()` → `columnStore.flush()` ([L231](https://github.com/apache/parquet-java/blob/921c3571f4086464c2888a9532ca6d925559c241/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java#L231)). The last row group's pages are encoded and compressed in `ColumnChunkPageWriter.writePage`, after `startBlock()` and before `flushToFileWriter()`. - `writeSupport.finalizeWrite()` ([L138](https://github.com/apache/parquet-java/blob/921c3571f4086464c2888a9532ca6d925559c241/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java#L138)). On storage where the object only appears once the stream is closed (e.g. S3A), this publishes a footerless file, the same outcome as #3706. I reproduced it on 1.17.0, which has #3351 but not this PR. The test uses an `OutputFile` whose stream `close()` marks the object as committed: | Failure inside `close()` | 1.17.0 | 1.17.0 + this PR's `withAbortOnFailure` change | 1.17.0 + `catch (Throwable t)` in IPRW | |---|---|---|---| | `UnsatisfiedLinkError`: snappy-java native load fails (no fault injection). The file is small, so the first `compress()` runs in `close()` | committed, 4 bytes (`PAR1` only) | committed, 4 bytes | not committed | | `OutOfMemoryError` from the `ByteBufferAllocator` passed via `withAllocator()`, armed just before `close()` | committed, 8832 bytes, no footer | committed, 8832 bytes, no footer | not committed | | `RuntimeException` from the same allocator | not committed | not committed | not committed | None of the stacks contains a `ParquetFileWriter` frame, and `end()` is never reached. For the first row the stack is `SnappyCompressor.maxCompressedLength` ← `CodecFactory$HeapBytesCompressor.compress` ← `ColumnChunkPageWriter.writePage` ← `ColumnWriteStoreBase.flush` ← `InternalParquetRecordWriter.flushRowGroupToStore` ← `InternalParquetRecordWriter.close`. That is why the `withAbortOnFailure` change doesn't affect them. (The middle column comes from that change patched into the 1.17.0 classes, not from a build of this branch.) `write()` already uses `catch (Throwable t)` ([L164](https://github.com/apache/parquet-java/blob/921c3571f4086464c2888a9532ca6d925559c241/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java#L164)), so the matching fix would be: ```java } catch (Throwable t) { parquetFileWriter.abort(); throw t; } ``` Precise rethrow keeps the `throws IOException, InterruptedException` clause, and `abort()` only sets a flag. A test for this needs the failure to happen outside `ParquetFileWriter`, for example an allocator via `withAllocator()` or a `WriteSupport` whose `finalizeWrite()` throws an `Error`. `FaultInjectingOutputFile` doesn't exercise it, because its faults fire inside `ParquetFileWriter` methods, which this PR already covers. <details> <summary>Repro (no fault injection)</summary> Run with `-Dorg.xerial.snappy.disable.bundled.libs=true` to simulate snappy-java failing to load its native library: ```java import java.io.ByteArrayOutputStream; import org.apache.parquet.example.data.Group; import org.apache.parquet.example.data.simple.SimpleGroupFactory; import org.apache.parquet.hadoop.ParquetWriter; import org.apache.parquet.hadoop.example.ExampleParquetWriter; import org.apache.parquet.hadoop.metadata.CompressionCodecName; import org.apache.parquet.io.OutputFile; import org.apache.parquet.io.PositionOutputStream; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.MessageTypeParser; public class SnappyLoadFailInClose { public static void main(String[] args) throws Exception { boolean[] committed = {false}; ByteArrayOutputStream bytes = new ByteArrayOutputStream(); OutputFile out = new OutputFile() { public PositionOutputStream create(long h) { return createOrOverwrite(h); } public PositionOutputStream createOrOverwrite(long h) { return new PositionOutputStream() { public long getPos() { return bytes.size(); } public void write(int b) { bytes.write(b); } public void write(byte[] b, int o, int l) { bytes.write(b, o, l); } public void close() { committed[0] = true; } // object stores commit here }; } public boolean supportsBlockSize() { return false; } public long defaultBlockSize() { return 0; } }; MessageType schema = MessageTypeParser.parseMessageType("message m { required int64 id; }"); ParquetWriter<Group> w = ExampleParquetWriter.builder(out).withType(schema) .withCompressionCodec(CompressionCodecName.SNAPPY).build(); SimpleGroupFactory f = new SimpleGroupFactory(schema); for (long i = 0; i < 100; i++) w.write(f.newGroup().append("id", i)); try { w.close(); } catch (Throwable t) { System.out.println("close() threw " + t); } byte[] b = bytes.toByteArray(); System.out.println("committed=" + committed[0] + " size=" + b.length + " footer=" + (b.length >= 8 && new String(b, b.length - 4, 4, "US-ASCII").equals("PAR1"))); } } ``` Output on 1.17.0: ``` close() threw java.lang.UnsatisfiedLinkError: no snappyjava in java.library.path: ... committed=true size=4 footer=false ``` </details> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
