Repository navigation
feat: implement read_rows - #762
Conversation
Co-authored-by: Mariatta Wijaya <Mariatta@users.noreply.github.com>
daniel-sanche
left a comment
There was a problem hiding this comment.
made changes based on PR feedback, but held off on adjusting tests until we finish the discussions
| - 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.
| 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
| 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
| 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 *?
| # 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
| 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
| buffer_task = asyncio.create_task( | ||
| self._generator_to_buffer(buffer, new_gapic_stream) | ||
| ) | ||
| buffered_stream = self._buffer_to_generator(buffer) |
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