This is an automated email from the ASF dual-hosted git repository.
shunping pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new c315ceacb5c Fix truncating file handle (#38425)
c315ceacb5c is described below
commit c315ceacb5c4cd745fa1c4a646828af8b112a2e9
Author: Shunping Huang <[email protected]>
AuthorDate: Tue Jun 2 04:24:54 2026 -0400
Fix truncating file handle (#38425)
---
sdks/python/apache_beam/dataframe/io.py | 15 ++++++++--
sdks/python/apache_beam/dataframe/io_test.py | 45 ++++++++++++++++++++++++++++
2 files changed, 58 insertions(+), 2 deletions(-)
diff --git a/sdks/python/apache_beam/dataframe/io.py
b/sdks/python/apache_beam/dataframe/io.py
index cabf6ccd5c8..21eab0b82fa 100644
--- a/sdks/python/apache_beam/dataframe/io.py
+++ b/sdks/python/apache_beam/dataframe/io.py
@@ -515,7 +515,7 @@ class _TruncatingFileHandle(object):
@property
def closed(self):
- return False
+ return getattr(self._underlying, 'closed', False)
def __iter__(self):
# For pandas is_file_like.
@@ -584,7 +584,18 @@ class _TruncatingFileHandle(object):
return res
def flush(self):
- self._underlying.flush()
+ if not self.closed:
+ try:
+ self._underlying.flush()
+ except ValueError:
+ pass
+
+ def close(self):
+ if not self.closed and hasattr(self._underlying, 'close'):
+ try:
+ self._underlying.close()
+ except (OSError, ValueError):
+ pass
class _ReadFromPandasDoFn(beam.DoFn, beam.RestrictionProvider):
diff --git a/sdks/python/apache_beam/dataframe/io_test.py
b/sdks/python/apache_beam/dataframe/io_test.py
index 313d955b455..4cd502d1b8d 100644
--- a/sdks/python/apache_beam/dataframe/io_test.py
+++ b/sdks/python/apache_beam/dataframe/io_test.py
@@ -296,6 +296,51 @@ A B
self._run_truncating_file_handle_iter_test('aaa b cccccccccccccccccccc')
self._run_truncating_file_handle_iter_test('aaa b ccccccccccccccccc ')
+ def test_truncating_filehandle_flush_on_closed_stream(self):
+ class ClosedFlushingStream(StringIO):
+ def flush(self):
+ if self.closed:
+ raise ValueError("I/O operation on closed file.")
+ super().flush()
+
+ s = 'a b c'
+ tracker = restriction_trackers.OffsetRestrictionTracker(
+ restriction_trackers.OffsetRange(0, len(s)))
+ underlying = ClosedFlushingStream(s)
+ handle = io._TruncatingFileHandle(
+ underlying, tracker, splitter=io._DelimSplitter(' ', 10))
+
+ # Verify that calling flush() when the underlying stream is closed
+ # succeeds without raising ValueError.
+ underlying.close()
+ handle.flush()
+ handle.close()
+
+ def test_truncating_filehandle_exception_suppression(self):
+ class FaultyStream(StringIO):
+ @property
+ def closed(self):
+ return False
+
+ def flush(self):
+ raise ValueError("Simulated flush error")
+
+ def close(self):
+ raise OSError("Simulated close error")
+
+ s = 'a b c'
+ tracker = restriction_trackers.OffsetRestrictionTracker(
+ restriction_trackers.OffsetRange(0, len(s)))
+ underlying = FaultyStream(s)
+ handle = io._TruncatingFileHandle(
+ underlying, tracker, splitter=io._DelimSplitter(' ', 10))
+
+ # Verify that ValueError raised during flush() is safely suppressed.
+ handle.flush()
+
+ # Verify that OSError raised during close() is safely suppressed.
+ handle.close()
+
@parameterized.expand([
('defaults', {}),
('header', dict(header=1)),
