damccorm opened a new pull request, #41301:
URL: https://github.com/apache/beam/pull/41301
Note: this PR was auto-generated by tooling. Before moving it out of draft,
the author has done a full review of the PR, specifically evaluating it on the
following criteria:
- The PR is an appropriate fix for the issue (e.g. we're not changing a test
to avoid a bug)
- The PR is a simple fix to the issue (e.g. we're not adding infra when we
could just modify a test)
- The PR is not covering for other problems (e.g. if we're bumping timeouts,
it is not because we're covering for a performance problem)
Full details of the methodology of generation and review can be found here -
https://docs.google.com/document/d/14lj7ECVov4T4EdwmIzNlWrRGE3lLDMOwX5xUMGzaLoc/edit?usp=sharing
If you have concerns with the PR or with the underlying methodology used to
generate this, please comment below - they will be addressed by the author not
the tooling :)
## Summary
Fixes a failure in `PreCommit Go` (`TestReadGzip` in [run
43447221716](https://github.com/apache/beam/actions/runs/43447221716/job/131959726687))
and a silent data truncation bug when reading gzip-compressed files in Go
`textio`:
```
--- FAIL: TestReadGzip (0.41s)
ptest.go:118:Failed to execute job: [2] error WaitForCompletion of job
"job-002": job go-job-2-1794581529400001809_go failed:
trying toReadGzip:
TextIO restriction lies outside the file being read. Restriction begins
at 25 bytes, but file is only 9 bytes.
```
### Root Cause
In `sdks/go/pkg/beam/io/textio/textio.go`, `readBaseFn` used
`file.Metadata.Size` (the **compressed** byte size on disk) as the
`offsetrange.Restriction` end bound `[0, file.Metadata.Size)`, while
`readBaseFn.process` opened the decompression stream (`gzip.Reader`) via
`file.Open(ctx)` and tracked **uncompressed** byte offsets `i +=
int64(len(line))`.
This mismatch between compressed restriction bounds and uncompressed stream
positions caused two bugs for gzip-compressed files:
1. **Dynamic/initial split failure (`TestReadGzip`)**:
`sdks/go/data/textio_test.gz` is 49 bytes compressed on disk but only 9 bytes
uncompressed (`"hello\ngo\n"`). When Prism dynamically split the `[0, 49)`
restriction at byte 25 while `process` was reading the first line, the residual
restriction `[25, 49)` attempted to discard 24 uncompressed bytes from a 9-byte
stream and failed with `Restriction begins at 25 bytes, but file is only 9
bytes`.
2. **Silent data truncation when uncompressed size exceeds compressed
size**: For any normal gzip file where the uncompressed text is larger than the
compressed `.gz` file on disk, `rt.TryClaim(i)` returned `false` as soon as the
uncompressed byte offset `i` reached `file.Metadata.Size`, causing `textio` to
exit the read loop early and silently drop the rest of the file.
### Fix
- Add `ReadableFile.IsCompressed()` in `sdks/go/pkg/beam/io/fileio/file.go`
to report whether a file resolves to a compressed format (explicit `ReadGzip()`
or auto-detected `.gz`).
- In `textio.readBaseFn`, treat non-empty compressed files as a single
unsplittable restriction `[0, 1)` (or `[0, 0)` if empty), claim `rest.Start`
once at the start of `process`, and read the decompressed stream to `io.EOF`.
- Add `TestReadGzipLargeUncompressed` in
`sdks/go/pkg/beam/io/textio/textio_test.go` to verify reading a gzip file whose
uncompressed contents are much larger than its compressed size on disk.
## Production Impact
**Impacts real Go SDK pipelines (including on Cloud Dataflow and Prism):**
- Any Go SDK pipeline using `textio.Read`, `textio.ReadAll`, or
`textio.ReadWithFilename` to read gzip-compressed files (`.gz` or
`textio.ReadGzip()`) where the uncompressed text is larger than the compressed
file size on disk previously stopped reading as soon as the uncompressed byte
offset reached `file.Metadata.Size`, **silently dropping all remaining lines in
the file**.
- Additionally, if a runner (such as Dataflow or Prism) dynamically split an
active `textio` restriction on a small gzip file (or initially split a gzip
file > 64 MB), the residual restriction failed or seeked to an uncompressed
byte offset computed from the compressed file size.
### Verification
- Triggered `beam_PreCommit_Go.yml` workflow run on branch:
[`43455232122`](https://github.com/apache/beam/actions/runs/43455232122)
--
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]
