feat: implement read_rows #762
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Co-authored-by: Mariatta Wijaya <[email protected]>
daniel-sanche
commented
May 23, 2023
google/cloud/bigtable/_read_rows.py
Outdated
| - request: the request dict to send to the Bigtable API | ||
| - client: the Bigtable client to use to make the request | ||
| - operation_timeout: the timeout to use for the entire operation, in seconds | ||
| - buffer_size: the size of the buffer to use for caching rows from the network |
There was a problem hiding this comment.
Ok, if we still don’t have clear consensus on the buffer, maybe we should err towards simplicity and remove it then?
I do worry it could impact the perceived throughput if the gapic call is always blocked on the consumer. But we can wait for user feedback first, and could always add a buffer back later if that ends up being an issue.
google/cloud/bigtable/_read_rows.py
Outdated
Comment on lines
233
to
236
| buffer_task = asyncio.create_task( | ||
| self._generator_to_buffer(buffer, new_gapic_stream) | ||
| ) | ||
| buffered_stream = self._buffer_to_generator(buffer) |
There was a problem hiding this comment.
the only background task created should be the buffer_task, which is cleaned up in the finally black below.
We could add more tests/checks around this to be safe, but I'll hold off on any until we resolve https://github.com/googleapis/python-bigtable/pull/762/files#r1201158174
google/cloud/bigtable/_read_rows.py
Outdated
| self._emit_count += 1 | ||
| self._last_emitted_row_key = new_item.row_key | ||
| if total_row_limit and self._emit_count >= total_row_limit: | ||
| return |
There was a problem hiding this comment.
it will be closed when the buffer_task is closed in the finally block
google-cla
bot
added
cla: yes
google-cla
bot
added
cla: no
google-cla
bot
added
cla: yes
google/cloud/bigtable/_read_rows.py
Outdated
| self, | ||
| request: dict[str, Any], | ||
| client: BigtableAsyncClient, | ||
| operation_timeout: float = 600.0, |
There was a problem hiding this comment.
should operation_time be after the *?
google/cloud/bigtable/_read_rows.py
Outdated
| # revise next request's row limit based on number emitted | ||
| if total_row_limit: | ||
| new_limit = total_row_limit - self._emit_count | ||
| if new_limit <= 0: |
There was a problem hiding this comment.
I think we might have to raise an error if the count goes negative. The situation would imply that there is a bug in the client or server and the results cant be trusted
google/cloud/bigtable/_read_rows.py
Outdated
| params_str = f'table_name={self._request.get("table_name", "")}' | ||
| if self._request.get("app_profile_id", None): | ||
| params_str = ( | ||
| f'{params_str},app_profile_id={self._request.get("app_profile_id", "")}' |
There was a problem hiding this comment.
nit. is the default arg for get necessary here? it seems like L183 guarantees that the key is present
google/cloud/bigtable/_read_rows.py
Outdated
Comment on lines
233
to
236
| buffer_task = asyncio.create_task( | ||
| self._generator_to_buffer(buffer, new_gapic_stream) | ||
| ) | ||
| buffered_stream = self._buffer_to_generator(buffer) |
google-cla
bot
added
cla: no
google-cla
bot
added
cla: yes
igorbernstein2
approved these changes
May 24, 2023
daniel-sanche
added a commit
that referenced
this pull request
Feb 5, 2024
* feat: add new v3.0.0 API skeleton (#745) * feat: improve rows filters (#751) * feat: read rows query model class (#752) * feat: implement row and cell model classes (#753) * feat: add pooled grpc transport (#748) * feat: implement read_rows (#762) * feat: implement mutate rows (#769) * feat: literal value filter (#767) * feat: row_exists and read_row (#778) * feat: read_modify_write and check_and_mutate_row (#780) * feat: sharded read rows (#766) * feat: ping and warm with metadata (#810) * feat: mutate rows batching (#770) * chore: restructure module paths (#816) * feat: improve timeout structure (#819) * fix: api errors apply to all bulk mutations * chore: reduce public api surface (#820) * feat: improve error group tracebacks on < py11 (#825) * feat: optimize read_rows (#852) * chore: add user agent suffix (#842) * feat: optimize retries (#854) * feat: add test proxy (#836) * chore(tests): add conformance tests to CI for v3 (#870) * chore(tests): turn off fast fail for conformance tets (#882) * feat: add TABLE_DEFAULTS enum for table method arguments (#880) * fix: pass None for retry in gapic calls (#881) * feat: replace internal dictionaries with protos in gapic calls (#875) * chore: optimize gapic calls (#863) * feat: expose retryable error codes to users (#879) * chore: update api_core submodule (#897) * chore: merge main into experimental_v3 (#900) * chore: pin conformance tests to v0.0.2 (#903) * fix: bulk mutation eventual success (#909) --------- Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This PR implements the read_rows RPC call, along with related features like the row merging state machine and smart retries.
Most of the logic is implemented in private classes in
_read_rows:ReadRowsOperationis the highest level class, providing an interface for asynchronousmerging with or without retries
StateMachineis used internally to track the state of the merge, includingrows the current row and the keys of the rows that have been processed.
It processes a stream of chunks, and will raise
InvalidChunkif it reachesan invalid state.
Stateclasses track the current state of theStateMachine, and define whatto do on the next chunk.
RowBuilderis used by theStateMachineto build a Row object.I also added a
ReadRowsIteratorclass, which is what users will interact with when reading from a stream. It adds idle_timeouts and can providerequest_stats(though this isn't fully implemented yet)The changes in this PR have been tested with the read_rows conformance tests using my test proxy from #747