You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by bo...@apache.org on 2019/07/18 21:44:38 UTC
[beam] branch master updated: Add last_attempted_record_start to
python OffsetRangeTracker (#9058)
This is an automated email from the ASF dual-hosted git repository.
boyuanz 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 8c8a902 Add last_attempted_record_start to python OffsetRangeTracker (#9058)
8c8a902 is described below
commit 8c8a902cfd3bb5502c757aa32d84ebc30c768fac
Author: Boyuan Zhang <36...@users.noreply.github.com>
AuthorDate: Thu Jul 18 14:44:12 2019 -0700
Add last_attempted_record_start to python OffsetRangeTracker (#9058)
* Add last_attempted_record_start to python OffsetRangeTracker
* Fix read loop to deal with unsplittable range_tracker
* Adjust record_start validation rule
* Only check last_attempted_claim in try_claim()
* fix lint
* Fix test_split_at_offset
* Add comment
* Fix test_set_current_position
* Update last_attempted_record_start docstring
---
sdks/python/apache_beam/io/filebasedsource_test.py | 6 +++--
sdks/python/apache_beam/io/range_trackers.py | 18 +++++++++++++++
sdks/python/apache_beam/io/range_trackers_test.py | 27 +++++++++++++++++++++-
sdks/python/apache_beam/io/sources_test.py | 8 +++++--
4 files changed, 54 insertions(+), 5 deletions(-)
diff --git a/sdks/python/apache_beam/io/filebasedsource_test.py b/sdks/python/apache_beam/io/filebasedsource_test.py
index 27b8d41..5f969d8 100644
--- a/sdks/python/apache_beam/io/filebasedsource_test.py
+++ b/sdks/python/apache_beam/io/filebasedsource_test.py
@@ -67,8 +67,10 @@ class LineSource(FileBasedSource):
start += len(line)
current = start
line = f.readline()
- while line:
- if not range_tracker.try_claim(current):
+ while range_tracker.try_claim(current):
+ # When the source is unsplittable, try_claim is not enough to determine
+ # whether the file has reached to the end.
+ if not line:
return
yield line.rstrip(b'\n')
current += len(line)
diff --git a/sdks/python/apache_beam/io/range_trackers.py b/sdks/python/apache_beam/io/range_trackers.py
index e02bea7..5bf4898 100644
--- a/sdks/python/apache_beam/io/range_trackers.py
+++ b/sdks/python/apache_beam/io/range_trackers.py
@@ -61,6 +61,7 @@ class OffsetRangeTracker(iobase.RangeTracker):
self._stop_offset = end
self._last_record_start = -1
+ self._last_attempted_record_start = -1
self._offset_of_last_split_point = -1
self._lock = threading.Lock()
@@ -77,6 +78,16 @@ class OffsetRangeTracker(iobase.RangeTracker):
def last_record_start(self):
return self._last_record_start
+ @property
+ def last_attempted_record_start(self):
+ """Return current value of last_attempted_record_start.
+
+ last_attempted_record_start records a valid position that tried to be
+ claimed by calling try_claim(). This value is only updated by `try_claim()`
+ no matter `try_claim()` returns `True` or `False`.
+ """
+ return self._last_attempted_record_start
+
def _validate_record_start(self, record_start, split_point):
# This function must only be called under the lock self.lock.
if not self._lock.locked():
@@ -102,7 +113,14 @@ class OffsetRangeTracker(iobase.RangeTracker):
def try_claim(self, record_start):
with self._lock:
+ # Attempted claim should be monotonous.
+ if record_start <= self._last_attempted_record_start:
+ raise ValueError(
+ 'Trying to return a record [starting at %d] which is not greater'
+ 'than the last-attempted record [starting at %d]' %
+ (record_start, self._last_attempted_record_start))
self._validate_record_start(record_start, True)
+ self._last_attempted_record_start = record_start
if record_start >= self.stop_position():
return False
self._offset_of_last_split_point = record_start
diff --git a/sdks/python/apache_beam/io/range_trackers_test.py b/sdks/python/apache_beam/io/range_trackers_test.py
index c9a9c42..e80401f 100644
--- a/sdks/python/apache_beam/io/range_trackers_test.py
+++ b/sdks/python/apache_beam/io/range_trackers_test.py
@@ -45,6 +45,32 @@ class OffsetRangeTrackerTest(unittest.TestCase):
self.assertTrue(tracker.try_claim(5))
self.assertFalse(tracker.try_claim(6))
+ def test_try_claim_update_last_attempt(self):
+ tracker = range_trackers.OffsetRangeTracker(1, 2)
+ self.assertTrue(tracker.try_claim(1))
+ self.assertEqual(1, tracker.last_attempted_record_start)
+
+ self.assertFalse(tracker.try_claim(3))
+ self.assertEqual(3, tracker.last_attempted_record_start)
+
+ self.assertFalse(tracker.try_claim(6))
+ self.assertEqual(6, tracker.last_attempted_record_start)
+
+ with self.assertRaises(Exception):
+ tracker.try_claim(6)
+
+ def test_set_current_position(self):
+ tracker = range_trackers.OffsetRangeTracker(0, 6)
+ self.assertTrue(tracker.try_claim(2))
+ # Cannot set current position before successful claimed pos.
+ with self.assertRaises(Exception):
+ tracker.set_current_position(1)
+
+ self.assertFalse(tracker.try_claim(10))
+ tracker.set_current_position(11)
+ self.assertEqual(10, tracker.last_attempted_record_start)
+ self.assertEqual(11, tracker.last_record_start)
+
def test_try_return_record_continuous_until_split_point(self):
tracker = range_trackers.OffsetRangeTracker(9, 18)
# Return records with gaps of 2; every 3rd record is a split point.
@@ -93,7 +119,6 @@ class OffsetRangeTrackerTest(unittest.TestCase):
self.assertFalse(tracker.try_claim(150))
self.assertFalse(tracker.try_claim(151))
# Should accept non-splitpoint records starting after stop offset.
- tracker.set_current_position(135)
tracker.set_current_position(152)
tracker.set_current_position(160)
tracker.set_current_position(171)
diff --git a/sdks/python/apache_beam/io/sources_test.py b/sdks/python/apache_beam/io/sources_test.py
index 93c9920..8908681 100644
--- a/sdks/python/apache_beam/io/sources_test.py
+++ b/sdks/python/apache_beam/io/sources_test.py
@@ -50,11 +50,13 @@ class LineSource(iobase.BoundedSource):
start -= 1
start += len(f.readline())
current = start
- for line in f:
- if not range_tracker.try_claim(current):
+ line = f.readline()
+ while range_tracker.try_claim(current):
+ if not line:
return
yield line.rstrip(b'\n')
current += len(line)
+ line = f.readline()
def split(self, desired_bundle_size, start_position=None, stop_position=None):
assert start_position is None
@@ -110,6 +112,8 @@ class SourcesTest(unittest.TestCase):
result = [line for line in source.read(range_tracker)]
self.assertCountEqual([b'aaaa', b'bbbb', b'cccc', b'dddd'], result)
+ self.assertTrue(range_tracker.last_attempted_record_start
+ >= range_tracker.stop_position())
def test_source_estimated_size(self):
file_name = self._create_temp_file(b'aaaa\n')