Skip to content

Bigtable: Add 'PartialRowsData.cancel'. #8176

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tseaver merged 9 commits into from
Jun 20, 2019
Merged

Bigtable: Add 'PartialRowsData.cancel'. #8176

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
d5a42a0
[BEAM-7249] Added code to stopiteration of PartialRowData()
mf2199 May 8, 2019
78abc7b
[BEAM-7249] Added code to stop iteration of PartialRowData()
mf2199 May 8, 2019
0f6d6cc
Merge branch 'bigtable_io' of https://github.com/mf2199/google-cloud-…
mf2199 May 8, 2019
d23f895
[BEAM-7249] Consensus version of addressing the stop iteration in Par…
mf2199 May 10, 2019
5804dd2
[BEAM-7249] PR version of addressing the stop iteration in PartialRow…
mf2199 May 10, 2019
372266e
[GCP-7760]: Addressing most recent PR comments
mf2199 Jun 2, 2019
ac0e77c
[GCP-7760]:
mf2199 Jun 2, 2019
53af715
Addressing the public PR comments
mf2199 Jun 19, 2019
b5692d5
PartialRowsData: self._stop flag renamed to self._cancelled
mf2199 Jun 20, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion bigtable/google/cloud/bigtable/row_data.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -376,6 +376,9 @@ def __init__(self, read_method, request, retry=DEFAULT_RETRY_READ_ROWS):
self.rows = {}
self._state = self.STATE_NEW_ROW

# Flag to stop iteration, for any reason not related to self.retry()
self._cancelled = False

@property
def state(self):
"""State machine state.
Expand All @@ -388,6 +391,7 @@ def state(self):

def cancel(self):
"""Cancels the iterator, closing the stream."""
self._cancelled = True
self.response_iterator.cancel()

def consume_all(self, max_loops=None):
Expand Down Expand Up @@ -436,7 +440,7 @@ def __iter__(self):
Parse the response and its chunks into a new/existing row in
:attr:`_rows`. Rows are returned in order by row key.
"""
while True:
while not self._cancelled:
try:
response = self._read_next_response()
except StopIteration:
Expand All @@ -445,6 +449,8 @@ def __iter__(self):
break

for chunk in response.chunks:
if self._cancelled:
break
self._process_chunk(chunk)
if chunk.commit_row:
self.last_scanned_row_key = self._previous_row.row_key
Expand Down
1 change: 1 addition & 0 deletions bigtable/tests/unit/test_row_data.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -529,6 +529,7 @@ def test_cancel(self):
self.assertEqual(response_iterator.cancel_calls, 0)
yield_rows_data.cancel()
self.assertEqual(response_iterator.cancel_calls, 1)
self.assertEqual(list(yield_rows_data), [])

# 'consume_next' tested via 'TestPartialRowsData_JSON_acceptance_tests'

Expand Down