Repository: googleapis/python-bigtable

Issue:
BigTable: 'Table.mutate_rows' deadline exceeded for large mutations
Deadline exceeded for large mutation of BigQuery table; not necessarily a bug perhaps it would be helpful to other to add this to the docs, automatically batch mutations by size, make the timeout configurable, or document large data transfers as a potential cause of exceeding the deadline.

#### Environment details

* Cloud BigTable
* Ubuntu 18.04
* Python 3.6.4
* google-cloud-bigtable==0.31.1

#### Steps to reproduce and code example

Changing the value of `mutation_batch_size` in the following, too large and the deadline is exceeded; with tfexample serialized video examples with 4 frames of size 224x224. Not necessarily a bug if this is the expected behavior and users should handle this kind of batching themselves.

```python

def iterable_dataset_from_file(filename):
    dataset = tf.data.TFRecordDataset(filename)

    iterator = dataset.make_initializable_iterator()

    next_element = iterator.get_next()

    with tf.Session() as sess:

      sess.run(iterator.initializer)

      i = 0
      while True:
        try:
          if i % 1000 == 0:
            print("Processed %s examples..." % i)
          yield sess.run(next_element)
          i += 1
        except tf.errors.OutOfRangeError:
          print("Ran out of examples (processed %s), exiting..." % i)
          break


def tfrecord_files_to_cbt_table(glob, table, selection, max_records=100000000,
                                mutation_batch_size=250):

  mutation_index = 0

  def new_mutation_batch():
    return [None for _ in range(mutation_batch_size)]

  files = tf.gfile.Glob(glob)

  for file_path in files:
 
    row_mutation_batch = new_mutation_batch()

    for i, example in enumerate(iterable_dataset_from_file(file_path)):

      idx = hashlib.md5(example).hexdigest()
    
      # DEV: To check "shuffle" effect add the id suffix
      idx = "_".join([selection.prefix, idx, str(i)])

      row = table.row(idx)
      row.set_cell(column_family_id=selection.column_family,
                   column=selection.column_qualifier,
                   value=example,
                   timestamp=datetime.datetime.utcnow())

      row_mutation_batch[mutation_index] = row

      if mutation_index == (mutation_batch_size - 1):
        table.mutate_rows(row_mutation_batch)
        row_mutation_batch = new_mutation_batch()
        mutation_index = 0
      else:
        mutation_index += 1

    final_mutation = row_mutation_batch[:(mutation_index-1)]
    if final_mutation:
      table.mutate_rows(final_mutation)
```

#### Stack trace
```
Traceback (most recent call last):
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/api_core/grpc_helpers.py", line 79, in next
    return six.next(self._wrapped)
  File "/opt/conda/lib/python3.6/site-packages/grpc/_channel.py", line 341, in __next__
    return self._next()
  File "/opt/conda/lib/python3.6/site-packages/grpc/_channel.py", line 335, in _next
    raise self
grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with (StatusCode.DEADLINE_EXCEEDED, Deadline Exceeded)>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/opt/conda/lib/python3.6/runpy.py", line 193, in _run_module_as_main
    "__main__", mod_spec)
  File "/opt/conda/lib/python3.6/runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "/home/jovyan/work/pcml/pcml/operations/tfrecord2bigtable.py", line 330, in <module>
    tf.app.run()
  File "/opt/conda/lib/python3.6/site-packages/tensorflow/python/platform/app.py", line 125, in run
    _sys.exit(main(argv))
  File "/home/jovyan/work/pcml/pcml/operations/tfrecord2bigtable.py", line 300, in main
    max_records=FLAGS.max_records)
  File "/home/jovyan/work/pcml/pcml/operations/tfrecord2bigtable.py", line 228, in tfrecord_files_to_cbt_table
    table.mutate_rows(row_mutation_batch)
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/cloud/bigtable/table.py", line 423, in mutate_rows
    return retryable_mutate_rows(retry=retry)
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/cloud/bigtable/table.py", line 571, in __call__
    mutate_rows()
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/api_core/retry.py", line 270, in retry_wrapped_func
    on_error=on_error,
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/api_core/retry.py", line 179, in retry_target
    return target()
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/cloud/bigtable/table.py", line 634, in _do_mutate_retryable_rows
    for response in responses:
  File "/home/jovyan/.local/lib/python3.6/site-packages/google/api_core/grpc_helpers.py", line 81, in next
    six.raise_from(exceptions.from_grpc_error(exc), exc)
  File "<string>", line 3, in raise_from
google.api_core.exceptions.DeadlineExceeded: 504 Deadline Exceeded
```


Discussion:

Patch:
diff --git a/google/cloud/bigtable/backup.py b/google/cloud/bigtable/backup.py
index 03a1c894..291ac783 100644
--- a/google/cloud/bigtable/backup.py
+++ b/google/cloud/bigtable/backup.py
@@ -21,6 +21,7 @@ from google.cloud.bigtable_admin_v2.gapic.bigtable_table_admin_client import (
     BigtableTableAdminClient,
 )
 from google.cloud.bigtable_admin_v2.types import table_pb2
+from google.cloud.bigtable.policy import Policy
 from google.cloud.exceptions import NotFound
 from google.protobuf import field_mask_pb2
 
@@ -392,3 +393,54 @@ class Backup(object):
         """
         api = self._instance._client.table_admin_client
         return api.restore_table(self._instance.name, table_id, self.name)
+
+    def get_iam_policy(self):
+        """Gets the IAM access control policy for this backup.
+
+        :rtype: :class:`google.cloud.bigtable.policy.Policy`
+        :returns: The current IAM policy of this backup.
+        """
+        table_api = self._instance._client.table_admin_client
+        args = {"resource": self.name}
+        response = table_api.get_iam_policy(**args)
+        return Policy.from_pb(response)
+
+    def set_iam_policy(self, policy):
+        """Sets the IAM access control policy for this backup. Replaces any
+        existing policy.
+
+        For more information about policy, please see documentation of
+        class `google.cloud.bigtable.policy.Policy`
+
+        :type policy: :class:`google.cloud.bigtable.policy.Policy`
+        :param policy: A new IAM policy to replace the current IAM policy
+                       of this backup.
+
+        :rtype: :class:`google.cloud.bigtable.policy.Policy`
+        :returns: The current IAM policy of this backup.
+        """
+        table_api = self._instance._client.table_admin_client
+        response = table_api.set_iam_policy(resource=self.name, policy=policy.to_pb())
+        return Policy.from_pb(response)
+
+    def test_iam_permissions(self, permissions):
+        """Tests whether the caller has the given permissions for this backup.
+        Returns the permissions that the caller has.
+
+        :type permissions: list
+        :param permissions: The set of permissions to check for
+               the ``resource``. Permissions with wildcards (such as '*'
+               or 'storage.*') are not allowed. For more information see
+               `IAM Overview
+               <https://cloud.google.com/iam/docs/overview#permissions>`_.
+               `Bigtable Permissions
+               <https://cloud.google.com/bigtable/docs/access-control>`_.
+
+        :rtype: list
+        :returns: A List(string) of permissions allowed on the backup.
+        """
+        table_api = self._instance._client.table_admin_client
+        response = table_api.test_iam_permissions(
+            resource=self.name, permissions=permissions
+        )
+        return list(response.permissions)
diff --git a/google/cloud/bigtable/table.py b/google/cloud/bigtable/table.py
index 950a8c3f..35ca43d2 100644
--- a/google/cloud/bigtable/table.py
+++ b/google/cloud/bigtable/table.py
@@ -20,9 +20,9 @@ from google.api_core.exceptions import DeadlineExceeded
 from google.api_core.exceptions import NotFound
 from google.api_core.exceptions import RetryError
 from google.api_core.exceptions import ServiceUnavailable
+from google.api_core.gapic_v1.method import DEFAULT
 from google.api_core.retry import if_exception_type
 from google.api_core.retry import Retry
-from google.api_core.gapic_v1.method import wrap_method
 from google.cloud._helpers import _to_bytes
 from google.cloud.bigtable.backup import Backup
 from google.cloud.bigtable.column_family import _gc_rule_from_pb
@@ -625,7 +625,7 @@ class Table(object):
         )
         return self.read_rows(**kwargs)
 
-    def mutate_rows(self, rows, retry=DEFAULT_RETRY):
+    def mutate_rows(self, rows, retry=DEFAULT_RETRY, timeout=DEFAULT):
         """Mutates multiple rows in bulk.
 
         For example:
@@ -656,17 +656,23 @@ class Table(object):
             the :meth:`~google.api_core.retry.Retry.with_delay` method or the
             :meth:`~google.api_core.retry.Retry.with_deadline` method.
 
+        :type timeout: float
+        :param timeout: number of seconds bounding retries for the call
+
         :rtype: list
         :returns: A list of response statuses (`google.rpc.status_pb2.Status`)
                   corresponding to success or failure of each row mutation
                   sent. These will be in the same order as the `rows`.
         """
+        if timeout is DEFAULT:
+            timeout = self.mutation_timeout
+
         retryable_mutate_rows = _RetryableMutateRowsWorker(
             self._instance._client,
             self.name,
             rows,
             app_profile_id=self._app_profile_id,
-            timeout=self.mutation_timeout,
+            timeout=timeout,
         )
         return retryable_mutate_rows(retry=retry)
 
@@ -1058,27 +1064,20 @@ class _RetryableMutateRowsWorker(object):
             # All mutations are either successful or non-retryable now.
             return self.responses_statuses
 
-        mutate_rows_request = _mutate_rows_request(
-            self.table_name, retryable_rows, app_profile_id=self.app_profile_id
-        )
+        entries = _compile_mutation_entries(self.table_name, retryable_rows)
         data_client = self.client.table_data_client
-        inner_api_calls = data_client._inner_api_calls
-        if "mutate_rows" not in inner_api_calls:
-            default_retry = (data_client._method_configs["MutateRows"].retry,)
-            if self.timeout is None:
-                default_timeout = data_client._method_configs["MutateRows"].timeout
-            else:
-                default_timeout = timeout.ExponentialTimeout(deadline=self.timeout)
-            data_client._inner_api_calls["mutate_rows"] = wrap_method(
-                data_client.transport.mutate_rows,
-                default_retry=default_retry,
-                default_timeout=default_timeout,
-                client_info=data_client._client_info,
-            )
+
+        kwargs = {}
+        if self.timeout is not None:
+            kwargs["timeout"] = timeout.ExponentialTimeout(deadline=self.timeout)
 
         try:
-            responses = data_client._inner_api_calls["mutate_rows"](
-                mutate_rows_request, retry=None
+            responses = data_client.mutate_rows(
+                self.table_name,
+                entries,
+                app_profile_id=self.app_profile_id,
+                retry=None,
+                **kwargs
             )
         except (ServiceUnavailable, DeadlineExceeded, Aborted):
             # If an exception, considered retryable by `RETRY_CODES`, is
@@ -1260,8 +1259,8 @@ def _create_row_request(
     return message
 
 
-def _mutate_rows_request(table_name, rows, app_profile_id=None):
-    """Creates a request to mutate rows in a table.
+def _compile_mutation_entries(table_name, rows):
+    """Create list of mutation entries
 
     :type table_name: str
     :param table_name: The name of the table to write to.
@@ -1269,29 +1268,29 @@ def _mutate_rows_request(table_name, rows, app_profile_id=None):
     :type rows: list
     :param rows: List or other iterable of :class:`.DirectRow` instances.
 
-    :type: app_profile_id: str
-    :param app_profile_id: (Optional) The unique name of the AppProfile.
-
-    :rtype: :class:`data_messages_v2_pb2.MutateRowsRequest`
-    :returns: The ``MutateRowsRequest`` protobuf corresponding to the inputs.
+    :rtype: List[:class:`data_messages_v2_pb2.MutateRowsRequest.Entry`]
+    :returns: entries corresponding to the inputs.
     :raises: :exc:`~.table.TooManyMutationsError` if the number of mutations is
-             greater than 100,000
-    """
-    request_pb = data_messages_v2_pb2.MutateRowsRequest(
-        table_name=table_name, app_profile_id=app_profile_id
+             greater than the max ({})
+    """.format(
+        _MAX_BULK_MUTATIONS
     )
+    entries = []
     mutations_count = 0
+    entry_klass = data_messages_v2_pb2.MutateRowsRequest.Entry
+
     for row in rows:
         _check_row_table_name(table_name, row)
         _check_row_type(row)
         mutations = row._get_mutations()
-        request_pb.entries.add(row_key=row.row_key, mutations=mutations)
+        entries.append(entry_klass(row_key=row.row_key, mutations=mutations))
         mutations_count += len(mutations)
+
     if mutations_count > _MAX_BULK_MUTATIONS:
         raise TooManyMutationsError(
             "Maximum number of mutations is %s" % (_MAX_BULK_MUTATIONS,)
         )
-    return request_pb
+    return entries
 
 
 def _check_row_table_name(table_name, row):
diff --git a/synth.py b/synth.py
index 8a2fed1c..21100c74 100644
--- a/synth.py
+++ b/synth.py
@@ -45,40 +45,28 @@ library = gapic.py_library(
 s.move(library / "google/cloud/bigtable_admin_v2")
 s.move(library / "tests")
 
-s.replace(
-    [
-        "google/cloud/bigtable_admin_v2/gapic/bigtable_instance_admin_client.py",
-        "google/cloud/bigtable_admin_v2/gapic/bigtable_table_admin_client.py",
-    ],
-    "'google-cloud-bigtable-admin'",
-    "'google-cloud-bigtable'",
-)
-
-s.replace(
-    "google/**/*.py",
-    "from google\.cloud\.bigtable\.admin_v2.proto",
-    "from google.cloud.bigtable_admin_v2.proto",
-)
+# Work around non-standard installations
 
-s.replace(
-    ["google/cloud/bigtable_admin_v2/__init__.py"],
-    "    __doc__ = bigtable_instance_admin_client."
-    "BigtableInstanceAdminClient.__doc__\n",
-    "    __doc__ = (\n"
-    "        bigtable_instance_admin_client.BigtableInstanceAdminClient."
-    "__doc__)\n",
-)
-
-s.replace(
-    ["google/cloud/bigtable_v2/gapic/bigtable_client.py"],
-    "if ``true_mutations`` is empty, and at most\n\n\s*100000.",
-    "if ``true_mutations`` is empty, and at most 100000.",
-)
+admin_clients = [
+    "google/cloud/bigtable_admin_v2/gapic/bigtable_instance_admin_client.py",
+    "google/cloud/bigtable_admin_v2/gapic/bigtable_table_admin_client.py",
+]
 
 s.replace(
-    ["google/cloud/bigtable_v2/gapic/bigtable_client.py"],
-    "if ``false_mutations`` is empty, and at most\n\n\s*100000.",
-    "if ``false_mutations`` is empty, and at most 100000.",
+    admin_clients,
+    """\
+_GAPIC_LIBRARY_VERSION = pkg_resources.get_distribution\(
+    'google-cloud-bigtable-admin',
+\).version
+""",
+    """\
+try:
+    _GAPIC_LIBRARY_VERSION = pkg_resources.get_distribution(
+        "google-cloud-bigtable"
+    ).version
+except pkg_resources.DistributionNotFound:  # pragma: NO COVER
+    _GAPIC_LIBRARY_VERSION = None
+"""
 )
 
 # ----------------------------------------------------------------------------


Test patch:
diff --git a/tests/unit/test_backup.py b/tests/unit/test_backup.py
index 2f263dff..0285d668 100644
--- a/tests/unit/test_backup.py
+++ b/tests/unit/test_backup.py
@@ -734,6 +734,113 @@ class TestBackup(unittest.TestCase):
             backup=self.BACKUP_NAME,
         )
 
+    def test_get_iam_policy(self):
+        from google.cloud.bigtable.client import Client
+        from google.cloud.bigtable_admin_v2.gapic import bigtable_table_admin_client
+        from google.iam.v1 import policy_pb2
+        from google.cloud.bigtable.policy import BIGTABLE_ADMIN_ROLE
+
+        credentials = _make_credentials()
+        client = Client(project=self.PROJECT_ID, credentials=credentials, admin=True)
+
+        instance = client.instance(instance_id=self.INSTANCE_ID)
+        backup = self._make_one(self.BACKUP_ID, instance, cluster_id=self.CLUSTER_ID)
+
+        version = 1
+        etag = b"etag_v1"
+        members = ["serviceAccount:service_acc1@test.com", "user:user1@test.com"]
+        bindings = [{"role": BIGTABLE_ADMIN_ROLE, "members": members}]
+        iam_policy = policy_pb2.Policy(version=version, etag=etag, bindings=bindings)
+
+        table_api = mock.create_autospec(
+            bigtable_table_admin_client.BigtableTableAdminClient
+        )
+        client._table_admin_client = table_api
+        table_api.get_iam_policy.return_value = iam_policy
+
+        result = backup.get_iam_policy()
+
+        table_api.get_iam_policy.assert_called_once_with(resource=backup.name)
+        self.assertEqual(result.version, version)
+        self.assertEqual(result.etag, etag)
+
+        admins = result.bigtable_admins
+        self.assertEqual(len(admins), len(members))
+        for found, expected in zip(sorted(admins), sorted(members)):
+            self.assertEqual(found, expected)
+
+    def test_set_iam_policy(self):
+        from google.cloud.bigtable.client import Client
+        from google.cloud.bigtable_admin_v2.gapic import bigtable_table_admin_client
+        from google.iam.v1 import policy_pb2
+        from google.cloud.bigtable.policy import Policy
+        from google.cloud.bigtable.policy import BIGTABLE_ADMIN_ROLE
+
+        credentials = _make_credentials()
+        client = Client(project=self.PROJECT_ID, credentials=credentials, admin=True)
+
+        instance = client.instance(instance_id=self.INSTANCE_ID)
+        backup = self._make_one(self.BACKUP_ID, instance, cluster_id=self.CLUSTER_ID)
+
+        version = 1
+        etag = b"etag_v1"
+        members = ["serviceAccount:service_acc1@test.com", "user:user1@test.com"]
+        bindings = [{"role": BIGTABLE_ADMIN_ROLE, "members": sorted(members)}]
+        iam_policy_pb = policy_pb2.Policy(version=version, etag=etag, bindings=bindings)
+
+        table_api = mock.create_autospec(
+            bigtable_table_admin_client.BigtableTableAdminClient
+        )
+        client._table_admin_client = table_api
+        table_api.set_iam_policy.return_value = iam_policy_pb
+
+        iam_policy = Policy(etag=etag, version=version)
+        iam_policy[BIGTABLE_ADMIN_ROLE] = [
+            Policy.user("user1@test.com"),
+            Policy.service_account("service_acc1@test.com"),
+        ]
+
+        result = backup.set_iam_policy(iam_policy)
+
+        table_api.set_iam_policy.assert_called_once_with(
+            resource=backup.name, policy=iam_policy_pb
+        )
+        self.assertEqual(result.version, version)
+        self.assertEqual(result.etag, etag)
+
+        admins = result.bigtable_admins
+        self.assertEqual(len(admins), len(members))
+        for found, expected in zip(sorted(admins), sorted(members)):
+            self.assertEqual(found, expected)
+
+    def test_test_iam_permissions(self):
+        from google.cloud.bigtable.client import Client
+        from google.cloud.bigtable_admin_v2.gapic import bigtable_table_admin_client
+        from google.iam.v1 import iam_policy_pb2
+
+        credentials = _make_credentials()
+        client = Client(project=self.PROJECT_ID, credentials=credentials, admin=True)
+
+        instance = client.instance(instance_id=self.INSTANCE_ID)
+        backup = self._make_one(self.BACKUP_ID, instance, cluster_id=self.CLUSTER_ID)
+
+        permissions = ["bigtable.backups.create", "bigtable.backups.list"]
+
+        response = iam_policy_pb2.TestIamPermissionsResponse(permissions=permissions)
+
+        table_api = mock.create_autospec(
+            bigtable_table_admin_client.BigtableTableAdminClient
+        )
+        table_api.test_iam_permissions.return_value = response
+        client._table_admin_client = table_api
+
+        result = backup.test_iam_permissions(permissions)
+
+        self.assertEqual(result, permissions)
+        table_api.test_iam_permissions.assert_called_once_with(
+            resource=backup.name, permissions=permissions
+        )
+
 
 class _Client(object):
     def __init__(self, project=TestBackup.PROJECT_ID):
diff --git a/tests/unit/test_table.py b/tests/unit/test_table.py
index c99cd659..4469846b 100644
--- a/tests/unit/test_table.py
+++ b/tests/unit/test_table.py
@@ -20,14 +20,14 @@ from ._testing import _make_credentials
 from google.api_core.exceptions import DeadlineExceeded
 
 
-class Test___mutate_rows_request(unittest.TestCase):
+class Test__compile_mutation_entries(unittest.TestCase):
     def _call_fut(self, table_name, rows):
-        from google.cloud.bigtable.table import _mutate_rows_request
+        from google.cloud.bigtable.table import _compile_mutation_entries
 
-        return _mutate_rows_request(table_name, rows)
+        return _compile_mutation_entries(table_name, rows)
 
     @mock.patch("google.cloud.bigtable.table._MAX_BULK_MUTATIONS", new=3)
-    def test__mutate_rows_too_many_mutations(self):
+    def test_w_too_many_mutations(self):
         from google.cloud.bigtable.row import DirectRow
         from google.cloud.bigtable.table import TooManyMutationsError
 
@@ -41,13 +41,15 @@ class Test___mutate_rows_request(unittest.TestCase):
         rows[0].set_cell("cf1", b"c1", 2)
         rows[1].set_cell("cf1", b"c1", 3)
         rows[1].set_cell("cf1", b"c1", 4)
+
         with self.assertRaises(TooManyMutationsError):
             self._call_fut("table", rows)
 
-    def test__mutate_rows_request(self):
+    def test_normal(self):
         from google.cloud.bigtable.row import DirectRow
+        from google.cloud.bigtable_v2.proto import bigtable_pb2
 
-        table = mock.Mock(name="table", spec=["name"])
+        table = mock.Mock(spec=["name"])
         table.name = "table"
         rows = [
             DirectRow(row_key=b"row_key", table=table),
@@ -55,25 +57,26 @@ class Test___mutate_rows_request(unittest.TestCase):
         ]
         rows[0].set_cell("cf1", b"c1", b"1")
         rows[1].set_cell("cf1", b"c1", b"2")
+
         result = self._call_fut("table", rows)
 
-        expected_result = _mutate_rows_request_pb(table_name="table")
-        entry1 = expected_result.entries.add()
-        entry1.row_key = b"row_key"
-        mutations1 = entry1.mutations.add()
-        mutations1.set_cell.family_name = "cf1"
-        mutations1.set_cell.column_qualifier = b"c1"
-        mutations1.set_cell.timestamp_micros = -1
-        mutations1.set_cell.value = b"1"
-        entry2 = expected_result.entries.add()
-        entry2.row_key = b"row_key_2"
-        mutations2 = entry2.mutations.add()
-        mutations2.set_cell.family_name = "cf1"
-        mutations2.set_cell.column_qualifier = b"c1"
-        mutations2.set_cell.timestamp_micros = -1
-        mutations2.set_cell.value = b"2"
+        Entry = bigtable_pb2.MutateRowsRequest.Entry
 
-        self.assertEqual(result, expected_result)
+        entry_1 = Entry(row_key=b"row_key")
+        mutations_1 = entry_1.mutations.add()
+        mutations_1.set_cell.family_name = "cf1"
+        mutations_1.set_cell.column_qualifier = b"c1"
+        mutations_1.set_cell.timestamp_micros = -1
+        mutations_1.set_cell.value = b"1"
+
+        entry_2 = Entry(row_key=b"row_key_2")
+        mutations_2 = entry_2.mutations.add()
+        mutations_2.set_cell.family_name = "cf1"
+        mutations_2.set_cell.column_qualifier = b"c1"
+        mutations_2.set_cell.timestamp_micros = -1
+        mutations_2.set_cell.value = b"2"
+
+        self.assertEqual(result, [entry_1, entry_2])
 
 
 class Test__check_row_table_name(unittest.TestCase):
@@ -162,27 +165,49 @@ class TestTable(unittest.TestCase):
     def _make_client(self, *args, **kwargs):
         return self._get_target_client_class()(*args, **kwargs)
 
-    def test_constructor_w_admin(self):
-        credentials = _make_credentials()
-        client = self._make_client(
-            project=self.PROJECT_ID, credentials=credentials, admin=True
-        )
-        instance = client.instance(instance_id=self.INSTANCE_ID)
+    def test_constructor_defaults(self):
+        instance = mock.Mock(spec=[])
+
         table = self._make_one(self.TABLE_ID, instance)
+
+        self.assertEqual(table.table_id, self.TABLE_ID)
+        self.assertIs(table._instance, instance)
+        self.assertIsNone(table.mutation_timeout)
+        self.assertIsNone(table._app_profile_id)
+
+    def test_constructor_explicit(self):
+        instance = mock.Mock(spec=[])
+        mutation_timeout = 123
+        app_profile_id = "profile-123"
+
+        table = self._make_one(
+            self.TABLE_ID,
+            instance,
+            mutation_timeout=mutation_timeout,
+            app_profile_id=app_profile_id,
+        )
+
         self.assertEqual(table.table_id, self.TABLE_ID)
-        self.assertIs(table._instance._client, client)
-        self.assertEqual(table.name, self.TABLE_NAME)
+        self.assertIs(table._instance, instance)
+        self.assertEqual(table.mutation_timeout, mutation_timeout)
+        self.assertEqual(table._app_profile_id, app_profile_id)
 
-    def test_constructor_wo_admin(self):
-        credentials = _make_credentials()
-        client = self._make_client(
-            project=self.PROJECT_ID, credentials=credentials, admin=False
+    def test_name(self):
+        table_data_client = mock.Mock(spec=["table_path"])
+        client = mock.Mock(
+            project=self.PROJECT_ID,
+            table_data_client=table_data_client,
+            spec=["project", "table_data_client"],
         )
-        instance = client.instance(instance_id=self.INSTANCE_ID)
+        instance = mock.Mock(
+            _client=client,
+            instance_id=self.INSTANCE_ID,
+            spec=["_client", "instance_id"],
+        )
+
         table = self._make_one(self.TABLE_ID, instance)
-        self.assertEqual(table.table_id, self.TABLE_ID)
-        self.assertIs(table._instance._client, client)
-        self.assertEqual(table.name, self.TABLE_NAME)
+
+        self.assertEqual(table.name, table_data_client.table_path.return_value)
 
     def _row_methods_helper(self):
         client = self._make_client(
@@ -620,8 +645,11 @@ class TestTable(unittest.TestCase):
         with self.assertRaises(ValueError):
             self._read_row_helper(chunks, None)
 
-    def test_mutate_rows(self):
+    def _mutate_rows_helper(
+        self, mutation_timeout=None, app_profile_id=None, retry=None, timeout=None
+    ):
         from google.rpc.status_pb2 import Status
+        from google.cloud.bigtable.table import DEFAULT_RETRY
         from google.cloud.bigtable_admin_v2.gapic import bigtable_table_admin_client
 
         table_api = mock.create_autospec(
@@ -633,21 +661,78 @@ class TestTable(unittest.TestCase):
         )
         instance = client.instance(instance_id=self.INSTANCE_ID)
         client._table_admin_client = table_api
-        table = self._make_one(self.TABLE_ID, instance)
+        ctor_kwargs = {}
 
-        response = [Status(code=0), Status(code=1)]
+        if mutation_timeout is not None:
+            ctor_kwargs["mutation_timeout"] = mutation_timeout
+
+        if app_profile_id is not None:
+            ctor_kwargs["app_profile_id"] = app_profile_id
 
-        mock_worker = mock.Mock(return_value=response)
-        with mock.patch(
+        table = self._make_one(self.TABLE_ID, instance, **ctor_kwargs)
+
+        rows = [mock.MagicMock(), mock.MagicMock()]
+        response = [Status(code=0), Status(code=1)]
+        instance_mock = mock.Mock(return_value=response)
+        klass_mock = mock.patch(
             "google.cloud.bigtable.table._RetryableMutateRowsWorker",
-            new=mock.MagicMock(return_value=mock_worker),
-        ):
-            statuses = table.mutate_rows([mock.MagicMock(), mock.MagicMock()])
+            new=mock.MagicMock(return_value=instance_mock),
+        )
+
+        call_kwargs = {}
+
+        if retry is not None:
+            call_kwargs["retry"] = retry
+
+        if timeout is not None:
+            expected_timeout = call_kwargs["timeout"] = timeout
+        else:
+            expected_timeout = mutation_timeout
+
+        with klass_mock:
+            statuses = table.mutate_rows(rows, **call_kwargs)
+
         result = [status.code for status in statuses]
         expected_result = [0, 1]
-
         self.assertEqual(result, expected_result)
 
+        klass_mock.new.assert_called_once_with(
+            client,
+            self.TABLE_NAME,
+            rows,
+            app_profile_id=app_profile_id,
+            timeout=expected_timeout,
+        )
+
+        if retry is not None:
+            instance_mock.assert_called_once_with(retry=retry)
+        else:
+            instance_mock.assert_called_once_with(retry=DEFAULT_RETRY)
+
+    def test_mutate_rows_w_default_mutation_timeout_app_profile_id(self):
+        self._mutate_rows_helper()
+
+    def test_mutate_rows_w_mutation_timeout(self):
+        mutation_timeout = 123
+        self._mutate_rows_helper(mutation_timeout=mutation_timeout)
+
+    def test_mutate_rows_w_app_profile_id(self):
+        app_profile_id = "profile-123"
+        self._mutate_rows_helper(app_profile_id=app_profile_id)
+
+    def test_mutate_rows_w_retry(self):
+        retry = mock.Mock()
+        self._mutate_rows_helper(retry=retry)
+
+    def test_mutate_rows_w_timeout_arg(self):
+        timeout = 123
+        self._mutate_rows_helper(timeout=timeout)
+
+    def test_mutate_rows_w_mutation_timeout_and_timeout_arg(self):
+        mutation_timeout = 123
+        timeout = 456
+        self._mutate_rows_helper(mutation_timeout=mutation_timeout, timeout=timeout)
+
     def test_read_rows(self):
         from google.cloud._testing import _Monkey
         from google.cloud.bigtable.row_data import PartialRowsData
@@ -1424,21 +1509,18 @@ class Test__RetryableMutateRowsWorker(unittest.TestCase):
         row_3 = DirectRow(row_key=b"row_key_3", table=table)
         row_3.set_cell("cf", b"col", b"value3")
 
-        response = self._make_responses(
-            [self.SUCCESS, self.RETRYABLE_1, self.NON_RETRYABLE]
-        )
+        worker = self._make_worker(client, table.name, [row_1, row_2, row_3])
 
-        with mock.patch("google.cloud.bigtable.table.wrap_method") as patched:
-            patched.return_value = mock.Mock(return_value=[response])
+        response_codes = [self.SUCCESS, self.RETRYABLE_1, self.NON_RETRYABLE]
+        response = self._make_responses(response_codes)
+        data_api.mutate_rows = mock.MagicMock(return_value=[response])
 
-            worker = self._make_worker(client, table.name, [row_1, row_2, row_3])
-            statuses = worker(retry=None)
+        statuses = worker(retry=None)
 
         result = [status.code for status in statuses]
-        expected_result = [self.SUCCESS, self.RETRYABLE_1, self.NON_RETRYABLE]
+        self.assertEqual(result, response_codes)
 
-        client._table_data_client._inner_api_calls["mutate_rows"].assert_called_once()
-        self.assertEqual(result, expected_result)
+        data_api.mutate_rows.assert_called_once()
 
     def test_callable_retry(self):
         from google.cloud.bigtable.row import DirectRow