[damccorm]

Summary

Fixes a splittable CSV/text read bug in _TruncatingFileHandle (sdks/python/apache_beam/dataframe/io.py) where a record is silently dropped when a restriction's stop offset lands on the exact record boundary immediately following its start offset. This was exposed by CsvTest.test_non_utf8_csv_read_write in PreCommit Python (3.14) (run 42817903097).

Example Reproducing the Problem

By design, _TruncatingFileHandle breaks files at record delimiters (\n): any restriction with start > len(header) skips from start to the first delimiter at or after start (because the previous restriction reads through that delimiter), and reads through the first delimiter at or after stop.

Consider a 30-byte CSV file with three 10-byte lines:

Bytes [0, 10):  b"col1,col2\n"   (header, len = 10)
Bytes [10, 20): b"row1,val1\n"   (Record 1)
Bytes [20, 30): b"row2,val2\n"   (Record 2)

Suppose the file range [0, 30) is split into three contiguous restrictions [0, 15), [15, 20), and [20, 30) (where 15 is mid-way through Record 1 and 20 is the exact boundary between Record 1 and Record 2):

1. Restriction [0, 15):
   • Starts after the header (_buffer_start_pos = 10).
   • offset = stop - _buffer_start_pos = 15 - 10 = 5 > 0, so it reads bytes [10, 15) (b"row1,") plus read_to_record_boundary through byte 20 (b"val1\n").
   • Emits: Record 1 (b"row1,val1\n").
2. Restriction [15, 20):
   • In __init__, because start = 15 > len(header) (10), it seeks to byte 15 and calls read_to_record_boundary to skip the tail of Record 1 (b"val1\n"), advancing _buffer_start_pos to 20 (the start of Record 2, b"row2,val2\n").
   • In _read, offset = stop - _buffer_start_pos = 20 - 20 = 0.
   • Before this PR: if offset <= 0: treated offset == 0 as empty and returned self._empty.
   • Emits: nothing.
3. Restriction [20, 30):
   • In __init__, because start = 20 > len(header) (10), it seeks to byte 20 and calls read_to_record_boundary to skip to the first delimiter after 20 — skipping all of Record 2 (b"row2,val2\n", bytes [20, 30)) and advancing _buffer_start_pos to 30.
   • Emits: nothing.

As a result, concatenating the three contiguous restrictions [0, 15), [15, 20), and [20, 30) silently drops Record 2 (b"row2,val2\n"):

from io import BytesIO
import pandas as pd
from apache_beam.dataframe import io
from apache_beam.io import restriction_trackers

csv_bytes = b"col1,col2\nrow1,val1\nrow2,val2\n"  # 30 bytes (10 bytes per line)

def read_range(start, stop):
  handle = io._TruncatingFileHandle(
      BytesIO(csv_bytes),
      restriction_trackers.OffsetRestrictionTracker(
          restriction_trackers.OffsetRange(start, stop)),
      splitter=io._TextFileSplitter((), {}))
  return pd.read_csv(handle)

df = pd.concat([read_range(0, 15), read_range(15, 20), read_range(20, 30)], ignore_index=True)
# Before fix: len(df) == 1 ("row2,val2" is dropped!)
# After fix:  len(df) == 2 (read_range(15, 20) emits "row2,val2")

Fix

In _TruncatingFileHandle._read, only return self._empty when offset < 0 or stop <= self._header_len. When stop > self._header_len and offset == 0 (_buffer_start_pos == stop), _TruncatingFileHandle reads through the next record boundary from _buffer[0:], emitting the record starting at stop that the subsequent restriction [stop, ...) skips.

Verification

• Added unit test coverage in TruncatingFileHandleTest (sdks/python/apache_beam/dataframe/io_test.py) testing splits [0, split), [split, first_record_end), [first_record_end, len(contents)).
• Triggered beam_PreCommit_Python.yml workflow run on branch: 42832034988

[damccorm]

/gemini review

[github-actions]

📋 Review Summary

This PR fixes a data-loss bug in _TruncatingFileHandle when a post-header restriction [start, stop) has its stop offset aligned with the first record boundary after start. The fix accurately distinguishes between a restriction ending within the header (stop <= self._header_len) and a post-header restriction whose initial skipped record lands directly on stop (offset == 0), ensuring the record starting at stop is read rather than skipped by both adjacent restrictions.

🔍 General Feedback

• The boundary condition analysis and fix in sdks/python/apache_beam/dataframe/io.py are sound and apply cleanly to both header-bearing (_TextFileSplitter) and headerless (_DelimSplitter) splittable reads.
• The new unit test in sdks/python/apache_beam/dataframe/io_test.py deterministically exercises the [0, split), [split, first_record_end), [first_record_end, len(contents)) split alignment that previously dropped a record.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 59.14%. Comparing base (0fa1aca) to head (ff6dbaf).
⚠️ Report is 33 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #41163      +/-   ##
============================================
+ Coverage     59.13%   59.14%   +0.01%     
  Complexity     4894     4894              
============================================
  Files          1468     1461       -7     
  Lines        186188   186010     -178     
  Branches       3645     3645              
============================================
- Hits         110096   110018      -78     
+ Misses        72743    72643     -100     
  Partials       3349     3349              

Flag	Coverage Δ
python	80.55% <100.00%> (+0.07%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[damccorm]

/gemini review

[damccorm]

To the reviewer: this one is subtle/took me a while to understand after AI discovered it - I tried to leave explanatory comments, but let me know if you need more context

[github-actions]

Assigning reviewers:

R: @shunping for label python.
Backup reviewer: claudevdm (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).