ManvithPanyam opened a new pull request, #39581:
URL: https://github.com/apache/beam/pull/39581
This is my first contribution to Apache Beam — happy to adjust the approach
if there's a preferred pattern I'm missing. Filing this as a combined bug
report + fix since the root cause was straightforward to trace and the repro is
small.
### What broke?
When reading an empty or header-only CSV file using `beam.io.ReadFromCsv`
(or `ReadViaPandas`), the pipeline fails at bundle completion with:
```text
ValueError: OffsetRestrictionTracker is not done since work in range [0, 16)
has not been claimed.
Traceback (most recent call last):
File "apache_beam/runners/common.py", line 1063, in
_invoke_process_per_window
self.threadsafe_restriction_tracker.check_done()
File "apache_beam/runners/sdf_utils.py", line 109, in check_done
return self._restriction_tracker.check_done()
File "apache_beam/io/restriction_trackers.py", line 89, in check_done
raise ValueError(
"OffsetRestrictionTracker is not done since work in range [%d, %d) "
"has not been claimed." % (self._current_position, self._range.stop))
Root Cause
In sdks/python/apache_beam/dataframe/io.py, _TruncatingFileHandle._read()
wraps the file handle for dynamic SDF restriction tracking during incremental
reads (ReadFromCsv / ReadViaPandas).
When reading an empty or header-only file, self._underlying.read(size)
returns empty bytes (b''), hitting lines 567–569:
python
if not self._buffer:
self._done = True
return self._empty
The method sets self._done = True and returns without claiming the
restriction range up to self._tracker.current_restriction().stop. At bundle
completion, OffsetRestrictionTracker.check_done() asserts that all work up to
stop was claimed, causing the ValueError.
This contrasts with the non-incremental branch in
_ReadFromPandasDoFn.process() (line 680), which already handles range
completion via:
python
if not self.incremental:
# Satisfy the SDF contract by claiming the whole range.
tracker.try_claim(tracker.current_restriction().stop)
Fix
In _TruncatingFileHandle._read(), claim the remaining restriction range
before setting self._done = True and returning self._empty, matching the
codebase's existing pattern for range completion.
python
if not self._buffer:
self._tracker.try_claim(self._tracker.current_restriction().stop)
self._done = True
return self._empty
Verification
Added test_empty_csv_read to IOTest in
sdks/python/apache_beam/dataframe/io_test.py.
Ran the full IOTest suite (36 tests):
Unmodified master: test_empty_csv_read fails with ValueError:
OffsetRestrictionTracker is not done...
With Fix: test_empty_csv_read passes cleanly (OK).
Pre-existing Windows test failures on standard file-writes
(test_double_write, test_read_write_csv, test_windowed_write) match baseline
unmodified master behavior due to known Windows file-locking issue #20642.
--
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]
