PDGGK opened a new pull request, #40966:
URL: https://github.com/apache/beam/pull/40966
Fixes #31863. #35363 hit the same failure with gzip.
`fileio.WriteToFiles` fails when a sink asks for a compressed file through
`create_metadata`, which is how the `FileSink` docstring says to choose the
metadata passed to `FileSystems.create`:
```python
class GzipTextSink(fileio.TextSink):
def create_metadata(self, destination, full_file_name):
return fileio.FileMetadata(
mime_type='application/octet-stream',
compression_type=CompressionTypes.GZIP)
```
`WriteToFiles` calls `sink.flush()` and then `writer.close()`.
`TextSink.flush()` calls `CompressedFile.flush()`, which called the
compressor's `flush()` with its default mode. That mode finishes the compressed
stream. `close()` then finished it a second time, and every compression type
raises at that point:
| compression | error from `CompressedFile.close()` |
|---|---|
| gzip, deflate | `zlib.error: Error -2 while flushing: inconsistent stream
state` |
| bzip2, lzma | `ValueError: Repeated call to flush()` |
| zstd | `ZstdError: compressor object already finished` |
A write after a flush also raises, because the compressor is already
finished.
### Change
`CompressedFile.flush()` now writes buffered data to the file without
finishing the stream. Only `close()` finishes it.
- gzip and deflate: `flush(zlib.Z_SYNC_FLUSH)`, the same mode
`gzip.GzipFile.flush()` uses.
- zstd: `flush(zstandard.COMPRESSOBJ_FLUSH_BLOCK)`.
- bzip2 and lzma: the compressor is not flushed, because these compressors
cannot flush without finishing. `bz2.BZ2File.flush()` and
`lzma.LZMAFile.flush()` do not flush the compressor either. The data is written
by `close()`.
The underlying file is still flushed in every case. `close()` is unchanged.
### Before and after
A pipeline that writes five lines through `WriteToFiles` with the sink
above, once per compression type, on the FnApiRunner. Each written file is then
decompressed with the standard library or `zstandard`.
Beam 2.77.0 (current master is the same):
```
gzip FAILED: zlib.error: Error -2 while flushing: inconsistent stream
state [while running
'WriteToFiles/ParDo(_WriteUnshardedRecordsFn)/ParDo(_WriteUnshardedRecordsFn)']
bzip2 FAILED: ValueError: Repeated call to flush() [while running
'WriteToFiles/ParDo(_WriteUnshardedRecordsFn)/ParDo(_WriteUnshardedRecordsFn)']
deflate FAILED: zlib.error: Error -2 while flushing: inconsistent stream
state [while running
'WriteToFiles/ParDo(_WriteUnshardedRecordsFn)/ParDo(_WriteUnshardedRecordsFn)']
zstd FAILED: zstandard.backend_c.ZstdError: compressor object already
finished [while running
'WriteToFiles/ParDo(_WriteUnshardedRecordsFn)/ParDo(_WriteUnshardedRecordsFn)']
lzma FAILED: ValueError: Repeated call to flush() [while running
'WriteToFiles/ParDo(_WriteUnshardedRecordsFn)/ParDo(_WriteUnshardedRecordsFn)']
```
With this change:
```
gzip OK: read back 5 lines ['line-0', 'line-1', 'line-2', 'line-3',
'line-4']
bzip2 OK: read back 5 lines ['line-0', 'line-1', 'line-2', 'line-3',
'line-4']
deflate OK: read back 5 lines ['line-0', 'line-1', 'line-2', 'line-3',
'line-4']
zstd OK: read back 5 lines ['line-0', 'line-1', 'line-2', 'line-3',
'line-4']
lzma OK: read back 5 lines ['line-0', 'line-1', 'line-2', 'line-3',
'line-4']
```
### Tests
- `filesystem_test.py`: `test_close_after_flush` (write, flush, close) and
`test_write_after_flush` (write, flush, write, close). Each runs for all five
compression types and reads the file back through `CompressedFile`.
- `fileio_test.py`: `test_write_compressed_files` runs `WriteToFiles` with a
sink that asks for bzip2 and then gzip, and reads the output back with
`ReadMatches`. The compression is inferred from the `.bz2` and `.gz` suffixes.
With the `filesystem.py` change reverted, all 10 `CompressedFile` subtests
fail. The close-after-flush cases fail in `close()` with the errors in the
table above, and the write-after-flush cases fail on the second `write()`.
`test_write_compressed_files` fails with `ValueError: Repeated call to flush()
[while running 'WriteToFiles/...']`. With the change, they all pass.
Run locally on Python 3.12, macOS, one process:
- `filesystem_test.py`: 41 passed, 10 subtests passed.
- `fileio_test.py`, `WriteFilesTest` and `ReadTest`: 13 passed and 1 skipped
on the FnApiRunner. The two streaming tests (`test_streaming_complex_timing`,
`test_streaming_different_file_types`) passed on the BundleBasedDirectRunner,
because they use `TestStream`.
- `textio_test.py`, the compressed `TextSinkTest` cases: 10 passed.
`WriteToText` closes without flushing, so this checks that path is unchanged.
- `yapf --diff` (0.43.0), `ruff check` (0.15.22) and `pyrefly check` (1.1.1)
on the three changed Python files: clean.
I did not run the tests on the default Prism runner, on other Python
versions, or the full Python suite. That is left to CI.
`CHANGES.md` has a Bugfixes entry.
--
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]
