Skip to content

feat(bigtable): Reroute Mutations Batcher to use data client - #18200

Draft
daniel-sanche wants to merge 26 commits into
shim/12-batcher-callbackfrom
shim/13-mutations-batcher
Draft

feat(bigtable): Reroute Mutations Batcher to use data client#18200
daniel-sanche wants to merge 26 commits into
shim/12-batcher-callbackfrom
shim/13-mutations-batcher

Conversation

@daniel-sanche

@daniel-sanche daniel-sanche commented Aug 21, 2026

Copy link
Copy Markdown
Contributor

Migrating over @gkevinzheng PR from bigtable monorepo googleapis/python-bigtable#1309

Original description:

Changes Made:

  • Replaced mutations batcher implementation with one based off of the data client.
  • Reworked unit tests.
  • Added additional system tests.

Additional Changes:

  • 785f138: fix references in propertoes to point to data client
  • ac62f4c: removed deprecation language around flush_interval, since it is supported in the data client

Note to reviewers: This PR has already been reviewed and merged to a staging branch, with the intention of doing a single merge to main. We are now planning to slowly rollout these changes back to the main branch. Minimal re-review should be necessary

@gemini-code-assist gemini-code-assist Bot left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Code Review

This pull request refactors the MutationsBatcher to delegate batching, queueing, and flow control to the underlying table implementation, removing redundant internal classes. It also updates the exception handling across both sync and async batchers to ensure that MutationsExceptionGroup only contains FailedMutationEntryError instances. A review comment points out a potential issue where unpacking error.__cause__ could result in None being added to the exceptions queue, and suggests a defensive fallback to the error itself.

Comment thread packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py Outdated
gkevinzheng and others added 26 commits September 2, 2026 11:42
**Changes made:**

- `Row` objects hold `Mutation` and `ReadModifyWriteRowRule` objects
from the data client rather than raw protos.
- Rerouted `ConditionalRow.commit` and `AppendRow.commit`
(CheckAndMutateRows and ReadModifyWriteRows respectively) to use the
data client, or more specifically, `self._table._table_impl`
- Added function `DirectRow._to_mutation_pbs` for retrieving mutations
in proto form for the current `MutateRows` implementation, as well as
for `DirectRow.get_mutations_size`.
- Removed unnecessary helper functions and tests for helper functions
**Changes Made:**

- Use MutateRow instead of MutateRows for DirectRow.commit instead of
Table.MutateRows
- Added system test for DirectRow.commit because of the decoupling of
DirectRow.commit and Table.MutateRows
- Adjusted input error system test because of slight changes in error
behavior
- Adjusted unit tests for DirectRow.commit
**Changes made:**

- Refactored `_MappableAttributesMixin` to `helpers.py` for use in
`RowRange`.
- Used `RowRange` and `ReadRowsQuery` from the data client as backing
data sources for `RowRange` and `RowSet` respectively.
According to go/rst_stream, `INTERNAL` errors with error messages
related to an `rst_stream` error should be interpreted as `UNAVAILABLE`
errors instead of internal errors. This PR creates a custom retry
predicate to allow retrying of `INTERNAL` errors with rst_stream
specific error messages if the `ServiceUnavailable` exception is allowed
to be retried.

---------

Co-authored-by: Daniel Sanche <sanche@google.com>
**Changes Made:**
- Added methods to convert `Row` and `Cell` objects in the data client
to `PartialRowData` and `Cell` objects in the legacy client.
- Removed legacy client code related to processing `ReadRowResponse`
chunks and testing `ReadRowResponse` chunks.
- Removed `_update_message_request` from `RowSet` because it's no longer
needed to create a `ReadRowQuery`
- Rerouted `read_row` and `read_rows` to use their data client
counterparts in `table.py`.
…atcher (#1308)

**Changes made:**

- Refactored logic from `Table.mutate_rows` from producing a list of
`Status` protos from a `MutationsExceptionGroup`

- Added private keyword argument for a batch completion callback in the
MutationsBatcher.

- Added unit tests/system tests.
**Changes Made:**

- Replaced mutations batcher implementation with one based off of the
data client.

- Reworked unit tests.

- Added additional system tests.
@daniel-sanche
daniel-sanche force-pushed the shim/13-mutations-batcher branch from dbfc051 to 57774e9 Compare September 2, 2026 18:42
exceptions.extend(
[
FailedMutationEntryError(
failed_idx=None, failed_mutation_entry=entry, cause=e

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what does field_index=None mean? should it be the index of the current entry?

)
return status_pb2.Status(
code=code_pb2.Code.UNKNOWN,
message="An unknown error has occurred",

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

maybe change this to GoogleApiCAllError with Unknown status code

# FailedMutationEntryError always has an Exception cause;
# defensively fall back to error itself if __cause__ is None.
cause = error.__cause__ if error.__cause__ is not None else error
self._exceptions.put(cause)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

could there be other types of error? we should put them in the exception list


return responses
self._close_batcher()
self._init_batcher()

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is it expensive to keep close and reopen batcher? would it be better if the data client batcher expose a flush method?

assert len(all_results) == num_sent


def test_mutations_batcher_exceptions(data_table, rows_to_delete):

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we add a test for RPC level error?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants