feat(bigtable): route read_row/mutate_row through the accelerator with native fallback - #2
Conversation
934f2cd to
b86802c
Compare
8543853 to
86edefc
Compare
mutianf
changed the title
1fc9db4 to
6d4ace5
Compare
|
|
||
|
|
||
| def _grpc_code(exc: BaseException) -> StatusCode | None: | ||
| """Best-effort extraction of a gRPC status code from an exception.""" |
There was a problem hiding this comment.
This would work for grpc.RpcError exceptions. But there's also api_core.exceptions.GoogleAPICallError errors, which store this as grpc_status_code.
I don't know if that'll be relevant here
(We have similar code here. Maybe we should make this a shared helper?)
|
|
||
| Never escapes the Table method that raises it: the method catches it and | ||
| falls through to the native code path. | ||
| """ |
There was a problem hiding this comment.
This could also live in google.cloud.bigtable.data.exceptions, depending on the usage
| f"to the native client: {exc}", | ||
| RuntimeWarning, | ||
| stacklevel=2, | ||
| ) |
There was a problem hiding this comment.
Do we want to prevent this fall-back if explicit is set?
| metric=ActiveOperationMetric(OperationType.READ_ROWS, is_streaming=False), | ||
| retryable_exceptions=(), | ||
| ) | ||
| row_merger._operation_metric.start_attempt() |
There was a problem hiding this comment.
Do we still need to collect the metrics in Python? Or does the accelerator handle that too?
I might need to double-check that read_rows still works as expected without the tracked_retry wrapper, unless you already verified that
There was a problem hiding this comment.
Good point, I think this line should be removed.
| row_merger.request, timeout=operation_timeout | ||
| ) | ||
| chunked_stream = row_merger.chunk_stream(stream) | ||
| results = [a async for a in row_merger.merge_rows(chunked_stream)] |
There was a problem hiding this comment.
Is read_rows_stream out of scope for now? That's the main part that I'd be worried about, since we have to be careful about how generators are wrapped/errors are propagated
There was a problem hiding this comment.
Yeah, it's out of scope in this iteration.
| ) | ||
| except _AcceleratorFallback: | ||
| # Daemon can't serve this call; fall through to the native path. | ||
| pass |
There was a problem hiding this comment.
I wonder if there's a solution here using decorators?
I like the idea of being able to mark each method with @accelerator_route('mutate_rows_via_accelerator'), and then have generic wrapper logic that handles the routing/fallback, with the native python implementation left mostly unchanged. That feels like it could give us some nice abstractions
But maybe that would introduce too many complications
There was a problem hiding this comment.
This feels like a bigger change, maybe i'll do it in a future refactor.
| return [row async for row in row_generator] | ||
|
|
||
| @CrossSync.convert | ||
| async def _read_row_via_accelerator( |
There was a problem hiding this comment.
Could it make sense to move this logic into accelerator_client.read_rows, to keep the accelerator implementation in one place?
There was a problem hiding this comment.
accelerator implements bigtable.proto which doesn't have a read_row rpc. This already routes the request to accelerator_client.read_rows?
…h native fallback Change-Id: I6fc1d2563ca3185b12efb87b96eff2b7be438405
…back comments Change-Id: I191c52745e0cd10e5ab99e5e06135cd2e518bb9f
Change-Id: I72d5e456cdf722498ace021cc1f976cc1469a4dc
…NIMPLEMENTED Change-Id: Ifad0d9eb9d7b4d03fff07de2549a6d3294cf2f16
…lback Change-Id: I1b7382f6bfd8e70e7c6035e4b8ebb93ee3e282f3
1b2b842 to
4419ba3
Compare
2 participants
Stack PR 2/7 — base:
accel-01-daemon-subprocess-wrapperRoutes
read_rowandmutate_rowthrough the accelerator daemon when eligible, via the method-name gate in_accelerator/_routing.pyand per-method dispatch branches.Accelerator stack