feat(bigtable): Rerouted MutateRows to use the data client

[daniel-sanche]

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

Original description:

Changes Made:

• Rerouted MutateRows to use the data client's Table.bulk_mutate_rows function.
• Reworked error handling in Table.mutate_rows. Since we must return a list of Status objects, we process the list of errors obtained from the _MutateRowsOperation back into a corresponding list of error statuses.
• Reworked retries in Table.mutate_rows as follows:
  • retry.deadline -> operation_timeout
  • timeout/mutation_timeout -> attempt_timeout
  • A null retry deadline now defaults to the table default mutate rows operation timeout instead of being defined as no timeout. The docstring has been updated to reflect that. A retry deadline of 0.0 will still be a no-retry option, and is implemented by setting the list of retriable errors to the empty list, and the operation timeout to the default mutate rows operation timeout.
• A timestamp of None in set_cell, which is the default, will now generate a client-side timestamp representing the current time, rather than defaulting to using the server-side timestamp. The docstrings for this have been updated to reflect this.

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]

Code Review

This pull request refactors the mutate_rows method in google/cloud/bigtable/table.py to delegate bulk mutation execution directly to the underlying data client implementation (self._table_impl.bulk_mutate_rows), allowing the removal of the legacy _RetryableMutateRowsWorker class and its associated helper functions. The review feedback correctly identifies a critical bug where an AttributeError will occur if retry is None, and points out the loss of defensive input validation checks for row types and table mismatches that were previously present.

[gemini-code-assist]

Issues Identified:

1. Potential AttributeError when retry is None:
   If a user calls mutate_rows with retry=None (a common pattern to disable retries), accessing retry.deadline on line 753 will raise an AttributeError: 'NoneType' object has no attribute 'deadline'. We should check if retry is None first and disable retries by setting retryable_errors = [].

2. Missing Input Validation for Row Type and Table Mismatch:
   The previous helper functions _check_row_type and _check_row_table_name were removed. Without these checks:

   • Passing a ConditionalRow or AppendRow will raise an unhelpful AttributeError (due to missing _get_mutations) instead of a clear TypeError.
   • Passing a row belonging to a different table will silently succeed and write to the current table, potentially causing silent data corruption.

We should restore these defensive checks and handle retry is None gracefully.

        if timeout is DEFAULT:
            timeout = self.mutation_timeout

        retryable_errors = RETRYABLE_MUTATION_ERRORS

        # The data client cannot take in zero or null values for deadline, so we set it to
        # the default if that is the case.
        if retry is None:
            operation_timeout = TABLE_DEFAULT.MUTATE_ROWS
            retryable_errors = []
        elif retry.deadline is None:
            operation_timeout = TABLE_DEFAULT.MUTATE_ROWS

        # To adhere to the retry strategy of do-nothing being achievable with a deadline
        # of 0.0, we modify the retryable errors to be empty if such a deadline is passed.
        elif retry.deadline == 0:
            operation_timeout = TABLE_DEFAULT.MUTATE_ROWS
            retryable_errors = []
        else:
            operation_timeout = retry.deadline

        attempt_timeout = timeout
        mutation_entries = []
        for row in rows:
            if not isinstance(row, DirectRow):
                raise TypeError(
                    "Bulk processing can not be applied for conditional or append mutations."
                )
            if row.table is not None and row.table.name != self.name:
                raise TableMismatchError(
                    "Row %s is a part of %s table. Current table: %s"
                    % (row.row_key, row.table.name, self.name)
                )
            mutation_entries.append(
                RowMutationEntry(row.row_key, row._get_mutations())
            )
        return_statuses = [status_pb2.Status(code=code_pb2.Code.OK)] * len(
            mutation_entries
        )  # By default, return status OKs for everything