diff --git a/google/cloud/spanner_admin_database_v1/__init__.py b/google/cloud/spanner_admin_database_v1/__init__.py
index e587590c9a..ee52bda123 100644
--- a/google/cloud/spanner_admin_database_v1/__init__.py
+++ b/google/cloud/spanner_admin_database_v1/__init__.py
@@ -19,6 +19,9 @@
 
 from .types.backup import Backup
 from .types.backup import BackupInfo
+from .types.backup import CopyBackupEncryptionConfig
+from .types.backup import CopyBackupMetadata
+from .types.backup import CopyBackupRequest
 from .types.backup import CreateBackupEncryptionConfig
 from .types.backup import CreateBackupMetadata
 from .types.backup import CreateBackupRequest
@@ -57,6 +60,9 @@
     "DatabaseAdminAsyncClient",
     "Backup",
     "BackupInfo",
+    "CopyBackupEncryptionConfig",
+    "CopyBackupMetadata",
+    "CopyBackupRequest",
     "CreateBackupEncryptionConfig",
     "CreateBackupMetadata",
     "CreateBackupRequest",
diff --git a/google/cloud/spanner_admin_database_v1/gapic_metadata.json b/google/cloud/spanner_admin_database_v1/gapic_metadata.json
index 1460097dc3..f7272318ef 100644
--- a/google/cloud/spanner_admin_database_v1/gapic_metadata.json
+++ b/google/cloud/spanner_admin_database_v1/gapic_metadata.json
@@ -10,6 +10,11 @@
         "grpc": {
           "libraryClient": "DatabaseAdminClient",
           "rpcs": {
+            "CopyBackup": {
+              "methods": [
+                "copy_backup"
+              ]
+            },
             "CreateBackup": {
               "methods": [
                 "create_backup"
@@ -100,6 +105,11 @@
         "grpc-async": {
           "libraryClient": "DatabaseAdminAsyncClient",
           "rpcs": {
+            "CopyBackup": {
+              "methods": [
+                "copy_backup"
+              ]
+            },
             "CreateBackup": {
               "methods": [
                 "create_backup"
diff --git a/google/cloud/spanner_admin_database_v1/services/database_admin/async_client.py b/google/cloud/spanner_admin_database_v1/services/database_admin/async_client.py
index add0829bc8..e4793ae26b 100644
--- a/google/cloud/spanner_admin_database_v1/services/database_admin/async_client.py
+++ b/google/cloud/spanner_admin_database_v1/services/database_admin/async_client.py
@@ -1506,6 +1506,168 @@ def sample_create_backup():
         # Done; return the response.
         return response
 
+    async def copy_backup(
+        self,
+        request: Union[backup.CopyBackupRequest, dict] = None,
+        *,
+        parent: str = None,
+        backup_id: str = None,
+        source_backup: str = None,
+        expire_time: timestamp_pb2.Timestamp = None,
+        retry: OptionalRetry = gapic_v1.method.DEFAULT,
+        timeout: float = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> operation_async.AsyncOperation:
+        r"""Starts copying a Cloud Spanner Backup. The returned backup
+        [long-running operation][google.longrunning.Operation] will have
+        a name of the format
+        ``projects/<project>/instances/<instance>/backups/<backup>/operations/<operation_id>``
+        and can be used to track copying of the backup. The operation is
+        associated with the destination backup. The
+        [metadata][google.longrunning.Operation.metadata] field type is
+        [CopyBackupMetadata][google.spanner.admin.database.v1.CopyBackupMetadata].
+        The [response][google.longrunning.Operation.response] field type
+        is [Backup][google.spanner.admin.database.v1.Backup], if
+        successful. Cancelling the returned operation will stop the
+        copying and delete the backup. Concurrent CopyBackup requests
+        can run on the same source backup.
+
+
+        .. code-block:: python
+
+            from google.cloud import spanner_admin_database_v1
+
+            def sample_copy_backup():
+                # Create a client
+                client = spanner_admin_database_v1.DatabaseAdminClient()
+
+                # Initialize request argument(s)
+                request = spanner_admin_database_v1.CopyBackupRequest(
+                    parent="parent_value",
+                    backup_id="backup_id_value",
+                    source_backup="source_backup_value",
+                )
+
+                # Make the request
+                operation = client.copy_backup(request=request)
+
+                print("Waiting for operation to complete...")
+
+                response = operation.result()
+
+                # Handle the response
+                print(response)
+
+        Args:
+            request (Union[google.cloud.spanner_admin_database_v1.types.CopyBackupRequest, dict]):
+                The request object. The request for
+                [CopyBackup][google.spanner.admin.database.v1.DatabaseAdmin.CopyBackup].
+            parent (:class:`str`):
+                Required. The name of the destination instance that will
+                contain the backup copy. Values are of the form:
+                ``projects/<project>/instances/<instance>``.
+
+                This corresponds to the ``parent`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            backup_id (:class:`str`):
+                Required. The id of the backup copy. The ``backup_id``
+                appended to ``parent`` forms the full backup_uri of the
+                form
+                ``projects/<project>/instances/<instance>/backups/<backup>``.
+
+                This corresponds to the ``backup_id`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            source_backup (:class:`str`):
+                Required. The source backup to be copied. The source
+                backup needs to be in READY state for it to be copied.
+                Once CopyBackup is in progress, the source backup cannot
+                be deleted or cleaned up on expiration until CopyBackup
+                is finished. Values are of the form:
+                ``projects/<project>/instances/<instance>/backups/<backup>``.
+
+                This corresponds to the ``source_backup`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            expire_time (:class:`google.protobuf.timestamp_pb2.Timestamp`):
+                Required. The expiration time of the backup in
+                microsecond granularity. The expiration time must be at
+                least 6 hours and at most 366 days from the
+                ``create_time`` of the source backup. Once the
+                ``expire_time`` has passed, the backup is eligible to be
+                automatically deleted by Cloud Spanner to free the
+                resources used by the backup.
+
+                This corresponds to the ``expire_time`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            retry (google.api_core.retry.Retry): Designation of what errors, if any,
+                should be retried.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+
+        Returns:
+            google.api_core.operation_async.AsyncOperation:
+                An object representing a long-running operation.
+
+                The result type for the operation will be
+                :class:`google.cloud.spanner_admin_database_v1.types.Backup`
+                A backup of a Cloud Spanner database.
+
+        """
+        # Create or coerce a protobuf request object.
+        # Quick check: If we got a request object, we should *not* have
+        # gotten any keyword arguments that map to the request.
+        has_flattened_params = any([parent, backup_id, source_backup, expire_time])
+        if request is not None and has_flattened_params:
+            raise ValueError(
+                "If the `request` argument is set, then none of "
+                "the individual field arguments should be set."
+            )
+
+        request = backup.CopyBackupRequest(request)
+
+        # If we have keyword arguments corresponding to fields on the
+        # request, apply these.
+        if parent is not None:
+            request.parent = parent
+        if backup_id is not None:
+            request.backup_id = backup_id
+        if source_backup is not None:
+            request.source_backup = source_backup
+        if expire_time is not None:
+            request.expire_time = expire_time
+
+        # Wrap the RPC method; this adds retry and timeout information,
+        # and friendly error handling.
+        rpc = gapic_v1.method_async.wrap_method(
+            self._client._transport.copy_backup,
+            default_timeout=3600.0,
+            client_info=DEFAULT_CLIENT_INFO,
+        )
+
+        # Certain fields should be provided within the metadata header;
+        # add these here.
+        metadata = tuple(metadata) + (
+            gapic_v1.routing_header.to_grpc_metadata((("parent", request.parent),)),
+        )
+
+        # Send the request.
+        response = await rpc(request, retry=retry, timeout=timeout, metadata=metadata,)
+
+        # Wrap the response in an operation future.
+        response = operation_async.from_gapic(
+            response,
+            self._client._transport.operations_client,
+            backup.Backup,
+            metadata_type=backup.CopyBackupMetadata,
+        )
+
+        # Done; return the response.
+        return response
+
     async def get_backup(
         self,
         request: Union[backup.GetBackupRequest, dict] = None,
diff --git a/google/cloud/spanner_admin_database_v1/services/database_admin/client.py b/google/cloud/spanner_admin_database_v1/services/database_admin/client.py
index 120dec124a..a7106d7aa7 100644
--- a/google/cloud/spanner_admin_database_v1/services/database_admin/client.py
+++ b/google/cloud/spanner_admin_database_v1/services/database_admin/client.py
@@ -1704,6 +1704,168 @@ def sample_create_backup():
         # Done; return the response.
         return response
 
+    def copy_backup(
+        self,
+        request: Union[backup.CopyBackupRequest, dict] = None,
+        *,
+        parent: str = None,
+        backup_id: str = None,
+        source_backup: str = None,
+        expire_time: timestamp_pb2.Timestamp = None,
+        retry: OptionalRetry = gapic_v1.method.DEFAULT,
+        timeout: float = None,
+        metadata: Sequence[Tuple[str, str]] = (),
+    ) -> operation.Operation:
+        r"""Starts copying a Cloud Spanner Backup. The returned backup
+        [long-running operation][google.longrunning.Operation] will have
+        a name of the format
+        ``projects/<project>/instances/<instance>/backups/<backup>/operations/<operation_id>``
+        and can be used to track copying of the backup. The operation is
+        associated with the destination backup. The
+        [metadata][google.longrunning.Operation.metadata] field type is
+        [CopyBackupMetadata][google.spanner.admin.database.v1.CopyBackupMetadata].
+        The [response][google.longrunning.Operation.response] field type
+        is [Backup][google.spanner.admin.database.v1.Backup], if
+        successful. Cancelling the returned operation will stop the
+        copying and delete the backup. Concurrent CopyBackup requests
+        can run on the same source backup.
+
+
+        .. code-block:: python
+
+            from google.cloud import spanner_admin_database_v1
+
+            def sample_copy_backup():
+                # Create a client
+                client = spanner_admin_database_v1.DatabaseAdminClient()
+
+                # Initialize request argument(s)
+                request = spanner_admin_database_v1.CopyBackupRequest(
+                    parent="parent_value",
+                    backup_id="backup_id_value",
+                    source_backup="source_backup_value",
+                )
+
+                # Make the request
+                operation = client.copy_backup(request=request)
+
+                print("Waiting for operation to complete...")
+
+                response = operation.result()
+
+                # Handle the response
+                print(response)
+
+        Args:
+            request (Union[google.cloud.spanner_admin_database_v1.types.CopyBackupRequest, dict]):
+                The request object. The request for
+                [CopyBackup][google.spanner.admin.database.v1.DatabaseAdmin.CopyBackup].
+            parent (str):
+                Required. The name of the destination instance that will
+                contain the backup copy. Values are of the form:
+                ``projects/<project>/instances/<instance>``.
+
+                This corresponds to the ``parent`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            backup_id (str):
+                Required. The id of the backup copy. The ``backup_id``
+                appended to ``parent`` forms the full backup_uri of the
+                form
+                ``projects/<project>/instances/<instance>/backups/<backup>``.
+
+                This corresponds to the ``backup_id`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            source_backup (str):
+                Required. The source backup to be copied. The source
+                backup needs to be in READY state for it to be copied.
+                Once CopyBackup is in progress, the source backup cannot
+                be deleted or cleaned up on expiration until CopyBackup
+                is finished. Values are of the form:
+                ``projects/<project>/instances/<instance>/backups/<backup>``.
+
+                This corresponds to the ``source_backup`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            expire_time (google.protobuf.timestamp_pb2.Timestamp):
+                Required. The expiration time of the backup in
+                microsecond granularity. The expiration time must be at
+                least 6 hours and at most 366 days from the
+                ``create_time`` of the source backup. Once the
+                ``expire_time`` has passed, the backup is eligible to be
+                automatically deleted by Cloud Spanner to free the
+                resources used by the backup.
+
+                This corresponds to the ``expire_time`` field
+                on the ``request`` instance; if ``request`` is provided, this
+                should not be set.
+            retry (google.api_core.retry.Retry): Designation of what errors, if any,
+                should be retried.
+            timeout (float): The timeout for this request.
+            metadata (Sequence[Tuple[str, str]]): Strings which should be
+                sent along with the request as metadata.
+
+        Returns:
+            google.api_core.operation.Operation:
+                An object representing a long-running operation.
+
+                The result type for the operation will be
+                :class:`google.cloud.spanner_admin_database_v1.types.Backup`
+                A backup of a Cloud Spanner database.
+
+        """
+        # Create or coerce a protobuf request object.
+        # Quick check: If we got a request object, we should *not* have
+        # gotten any keyword arguments that map to the request.
+        has_flattened_params = any([parent, backup_id, source_backup, expire_time])
+        if request is not None and has_flattened_params:
+            raise ValueError(
+                "If the `request` argument is set, then none of "
+                "the individual field arguments should be set."
+            )
+
+        # Minor optimization to avoid making a copy if the user passes
+        # in a backup.CopyBackupRequest.
+        # There's no risk of modifying the input as we've already verified
+        # there are no flattened fields.
+        if not isinstance(request, backup.CopyBackupRequest):
+            request = backup.CopyBackupRequest(request)
+            # If we have keyword arguments corresponding to fields on the
+            # request, apply these.
+            if parent is not None:
+                request.parent = parent
+            if backup_id is not None:
+                request.backup_id = backup_id
+            if source_backup is not None:
+                request.source_backup = source_backup
+            if expire_time is not None:
+                request.expire_time = expire_time
+
+        # Wrap the RPC method; this adds retry and timeout information,
+        # and friendly error handling.
+        rpc = self._transport._wrapped_methods[self._transport.copy_backup]
+
+        # Certain fields should be provided within the metadata header;
+        # add these here.
+        metadata = tuple(metadata) + (
+            gapic_v1.routing_header.to_grpc_metadata((("parent", request.parent),)),
+        )
+
+        # Send the request.
+        response = rpc(request, retry=retry, timeout=timeout, metadata=metadata,)
+
+        # Wrap the response in an operation future.
+        response = operation.from_gapic(
+            response,
+            self._transport.operations_client,
+            backup.Backup,
+            metadata_type=backup.CopyBackupMetadata,
+        )
+
+        # Done; return the response.
+        return response
+
     def get_backup(
         self,
         request: Union[backup.GetBackupRequest, dict] = None,
diff --git a/google/cloud/spanner_admin_database_v1/services/database_admin/transports/base.py b/google/cloud/spanner_admin_database_v1/services/database_admin/transports/base.py
index 090e2a954e..18dfc4074c 100644
--- a/google/cloud/spanner_admin_database_v1/services/database_admin/transports/base.py
+++ b/google/cloud/spanner_admin_database_v1/services/database_admin/transports/base.py
@@ -234,6 +234,9 @@ def _prep_wrapped_messages(self, client_info):
             self.create_backup: gapic_v1.method.wrap_method(
                 self.create_backup, default_timeout=3600.0, client_info=client_info,
             ),
+            self.copy_backup: gapic_v1.method.wrap_method(
+                self.copy_backup, default_timeout=3600.0, client_info=client_info,
+            ),
             self.get_backup: gapic_v1.method.wrap_method(
                 self.get_backup,
                 default_retry=retries.Retry(
@@ -444,6 +447,15 @@ def create_backup(
     ]:
         raise NotImplementedError()
 
+    @property
+    def copy_backup(
+        self,
+    ) -> Callable[
+        [backup.CopyBackupRequest],
+        Union[operations_pb2.Operation, Awaitable[operations_pb2.Operation]],
+    ]:
+        raise NotImplementedError()
+
     @property
     def get_backup(
         self,
diff --git a/google/cloud/spanner_admin_database_v1/services/database_admin/transports/grpc.py b/google/cloud/spanner_admin_database_v1/services/database_admin/transports/grpc.py
index 9c0d1ea4d0..6f1d695122 100644
--- a/google/cloud/spanner_admin_database_v1/services/database_admin/transports/grpc.py
+++ b/google/cloud/spanner_admin_database_v1/services/database_admin/transports/grpc.py
@@ -593,6 +593,44 @@ def create_backup(
             )
         return self._stubs["create_backup"]
 
+    @property
+    def copy_backup(
+        self,
+    ) -> Callable[[backup.CopyBackupRequest], operations_pb2.Operation]:
+        r"""Return a callable for the copy backup method over gRPC.
+
+        Starts copying a Cloud Spanner Backup. The returned backup
+        [long-running operation][google.longrunning.Operation] will have
+        a name of the format
+        ``projects/<project>/instances/<instance>/backups/<backup>/operations/<operation_id>``
+        and can be used to track copying of the backup. The operation is
+        associated with the destination backup. The
+        [metadata][google.longrunning.Operation.metadata] field type is
+        [CopyBackupMetadata][google.spanner.admin.database.v1.CopyBackupMetadata].
+        The [response][google.longrunning.Operation.response] field type
+        is [Backup][google.spanner.admin.database.v1.Backup], if
+        successful. Cancelling the returned operation will stop the
+        copying and delete the backup. Concurrent CopyBackup requests
+        can run on the same source backup.
+
+        Returns:
+            Callable[[~.CopyBackupRequest],
+                    ~.Operation]:
+                A function that, when called, will call the underlying RPC
+                on the server.
+        """
+        # Generate a "stub function" on-the-fly which will actually make
+        # the request.
+        # gRPC handles serialization and deserialization, so we just need
+        # to pass in the functions for each.
+        if "copy_backup" not in self._stubs:
+            self._stubs["copy_backup"] = self.grpc_channel.unary_unary(
+                "/google.spanner.admin.database.v1.DatabaseAdmin/CopyBackup",
+                request_serializer=backup.CopyBackupRequest.serialize,
+                response_deserializer=operations_pb2.Operation.FromString,
+            )
+        return self._stubs["copy_backup"]
+
     @property
     def get_backup(self) -> Callable[[backup.GetBackupRequest], backup.Backup]:
         r"""Return a callable for the get backup method over gRPC.
diff --git a/google/cloud/spanner_admin_database_v1/services/database_admin/transports/grpc_asyncio.py b/google/cloud/spanner_admin_database_v1/services/database_admin/transports/grpc_asyncio.py
index fd35a3eaf5..2a3200a882 100644
--- a/google/cloud/spanner_admin_database_v1/services/database_admin/transports/grpc_asyncio.py
+++ b/google/cloud/spanner_admin_database_v1/services/database_admin/transports/grpc_asyncio.py
@@ -604,6 +604,44 @@ def create_backup(
             )
         return self._stubs["create_backup"]
 
+    @property
+    def copy_backup(
+        self,
+    ) -> Callable[[backup.CopyBackupRequest], Awaitable[operations_pb2.Operation]]:
+        r"""Return a callable for the copy backup method over gRPC.
+
+        Starts copying a Cloud Spanner Backup. The returned backup
+        [long-running operation][google.longrunning.Operation] will have
+        a name of the format
+        ``projects/<project>/instances/<instance>/backups/<backup>/operations/<operation_id>``
+        and can be used to track copying of the backup. The operation is
+        associated with the destination backup. The
+        [metadata][google.longrunning.Operation.metadata] field type is
+        [CopyBackupMetadata][google.spanner.admin.database.v1.CopyBackupMetadata].
+        The [response][google.longrunning.Operation.response] field type
+        is [Backup][google.spanner.admin.database.v1.Backup], if
+        successful. Cancelling the returned operation will stop the
+        copying and delete the backup. Concurrent CopyBackup requests
+        can run on the same source backup.
+
+        Returns:
+            Callable[[~.CopyBackupRequest],
+                    Awaitable[~.Operation]]:
+                A function that, when called, will call the underlying RPC
+                on the server.
+        """
+        # Generate a "stub function" on-the-fly which will actually make
+        # the request.
+        # gRPC handles serialization and deserialization, so we just need
+        # to pass in the functions for each.
+        if "copy_backup" not in self._stubs:
+            self._stubs["copy_backup"] = self.grpc_channel.unary_unary(
+                "/google.spanner.admin.database.v1.DatabaseAdmin/CopyBackup",
+                request_serializer=backup.CopyBackupRequest.serialize,
+                response_deserializer=operations_pb2.Operation.FromString,
+            )
+        return self._stubs["copy_backup"]
+
     @property
     def get_backup(
         self,
diff --git a/google/cloud/spanner_admin_database_v1/types/__init__.py b/google/cloud/spanner_admin_database_v1/types/__init__.py
index 8a7e38d1ab..8d4b5f4094 100644
--- a/google/cloud/spanner_admin_database_v1/types/__init__.py
+++ b/google/cloud/spanner_admin_database_v1/types/__init__.py
@@ -16,6 +16,9 @@
 from .backup import (
     Backup,
     BackupInfo,
+    CopyBackupEncryptionConfig,
+    CopyBackupMetadata,
+    CopyBackupRequest,
     CreateBackupEncryptionConfig,
     CreateBackupMetadata,
     CreateBackupRequest,
@@ -58,6 +61,9 @@
 __all__ = (
     "Backup",
     "BackupInfo",
+    "CopyBackupEncryptionConfig",
+    "CopyBackupMetadata",
+    "CopyBackupRequest",
     "CreateBackupEncryptionConfig",
     "CreateBackupMetadata",
     "CreateBackupRequest",
diff --git a/google/cloud/spanner_admin_database_v1/types/backup.py b/google/cloud/spanner_admin_database_v1/types/backup.py
index da5f4d4b2e..b4cff201a2 100644
--- a/google/cloud/spanner_admin_database_v1/types/backup.py
+++ b/google/cloud/spanner_admin_database_v1/types/backup.py
@@ -27,6 +27,8 @@
         "Backup",
         "CreateBackupRequest",
         "CreateBackupMetadata",
+        "CopyBackupRequest",
+        "CopyBackupMetadata",
         "UpdateBackupRequest",
         "GetBackupRequest",
         "DeleteBackupRequest",
@@ -36,6 +38,7 @@
         "ListBackupOperationsResponse",
         "BackupInfo",
         "CreateBackupEncryptionConfig",
+        "CopyBackupEncryptionConfig",
     },
 )
 
@@ -107,6 +110,23 @@ class Backup(proto.Message):
         database_dialect (google.cloud.spanner_admin_database_v1.types.DatabaseDialect):
             Output only. The database dialect information
             for the backup.
+        referencing_backups (Sequence[str]):
+            Output only. The names of the destination backups being
+            created by copying this source backup. The backup names are
+            of the form
+            ``projects/<project>/instances/<instance>/backups/<backup>``.
+            Referencing backups may exist in different instances. The
+            existence of any referencing backup prevents the backup from
+            being deleted. When the copy operation is done (either
+            successfully completed or cancelled or the destination
+            backup is deleted), the reference to the backup is removed.
+        max_expire_time (google.protobuf.timestamp_pb2.Timestamp):
+            Output only. The max allowed expiration time of the backup,
+            with microseconds granularity. A backup's expiration time
+            can be configured in multiple APIs: CreateBackup,
+            UpdateBackup, CopyBackup. When updating or copying an
+            existing backup, the expiration time specified must be less
+            than ``Backup.max_expire_time``.
     """
 
     class State(proto.Enum):
@@ -129,6 +149,10 @@ class State(proto.Enum):
         proto.MESSAGE, number=8, message=common.EncryptionInfo,
     )
     database_dialect = proto.Field(proto.ENUM, number=10, enum=common.DatabaseDialect,)
+    referencing_backups = proto.RepeatedField(proto.STRING, number=11,)
+    max_expire_time = proto.Field(
+        proto.MESSAGE, number=12, message=timestamp_pb2.Timestamp,
+    )
 
 
 class CreateBackupRequest(proto.Message):
@@ -204,6 +228,91 @@ class CreateBackupMetadata(proto.Message):
     cancel_time = proto.Field(proto.MESSAGE, number=4, message=timestamp_pb2.Timestamp,)
 
 
+class CopyBackupRequest(proto.Message):
+    r"""The request for
+    [CopyBackup][google.spanner.admin.database.v1.DatabaseAdmin.CopyBackup].
+
+    Attributes:
+        parent (str):
+            Required. The name of the destination instance that will
+            contain the backup copy. Values are of the form:
+            ``projects/<project>/instances/<instance>``.
+        backup_id (str):
+            Required. The id of the backup copy. The ``backup_id``
+            appended to ``parent`` forms the full backup_uri of the form
+            ``projects/<project>/instances/<instance>/backups/<backup>``.
+        source_backup (str):
+            Required. The source backup to be copied. The source backup
+            needs to be in READY state for it to be copied. Once
+            CopyBackup is in progress, the source backup cannot be
+            deleted or cleaned up on expiration until CopyBackup is
+            finished. Values are of the form:
+            ``projects/<project>/instances/<instance>/backups/<backup>``.
+        expire_time (google.protobuf.timestamp_pb2.Timestamp):
+            Required. The expiration time of the backup in microsecond
+            granularity. The expiration time must be at least 6 hours
+            and at most 366 days from the ``create_time`` of the source
+            backup. Once the ``expire_time`` has passed, the backup is
+            eligible to be automatically deleted by Cloud Spanner to
+            free the resources used by the backup.
+        encryption_config (google.cloud.spanner_admin_database_v1.types.CopyBackupEncryptionConfig):
+            Optional. The encryption configuration used to encrypt the
+            backup. If this field is not specified, the backup will use
+            the same encryption configuration as the source backup by
+            default, namely
+            [encryption_type][google.spanner.admin.database.v1.CopyBackupEncryptionConfig.encryption_type]
+            = ``USE_CONFIG_DEFAULT_OR_BACKUP_ENCRYPTION``.
+    """
+
+    parent = proto.Field(proto.STRING, number=1,)
+    backup_id = proto.Field(proto.STRING, number=2,)
+    source_backup = proto.Field(proto.STRING, number=3,)
+    expire_time = proto.Field(proto.MESSAGE, number=4, message=timestamp_pb2.Timestamp,)
+    encryption_config = proto.Field(
+        proto.MESSAGE, number=5, message="CopyBackupEncryptionConfig",
+    )
+
+
+class CopyBackupMetadata(proto.Message):
+    r"""Metadata type for the google.longrunning.Operation returned by
+    [CopyBackup][google.spanner.admin.database.v1.DatabaseAdmin.CopyBackup].
+
+    Attributes:
+        name (str):
+            The name of the backup being created through the copy
+            operation. Values are of the form
+            ``projects/<project>/instances/<instance>/backups/<backup>``.
+        source_backup (str):
+            The name of the source backup that is being copied. Values
+            are of the form
+            ``projects/<project>/instances/<instance>/backups/<backup>``.
+        progress (google.cloud.spanner_admin_database_v1.types.OperationProgress):
+            The progress of the
+            [CopyBackup][google.spanner.admin.database.v1.DatabaseAdmin.CopyBackup]
+            operation.
+        cancel_time (google.protobuf.timestamp_pb2.Timestamp):
+            The time at which cancellation of CopyBackup operation was
+            received.
+            [Operations.CancelOperation][google.longrunning.Operations.CancelOperation]
+            starts asynchronous cancellation on a long-running
+            operation. The server makes a best effort to cancel the
+            operation, but success is not guaranteed. Clients can use
+            [Operations.GetOperation][google.longrunning.Operations.GetOperation]
+            or other methods to check whether the cancellation succeeded
+            or whether the operation completed despite cancellation. On
+            successful cancellation, the operation is not deleted;
+            instead, it becomes an operation with an
+            [Operation.error][google.longrunning.Operation.error] value
+            with a [google.rpc.Status.code][google.rpc.Status.code] of
+            1, corresponding to ``Code.CANCELLED``.
+    """
+
+    name = proto.Field(proto.STRING, number=1,)
+    source_backup = proto.Field(proto.STRING, number=2,)
+    progress = proto.Field(proto.MESSAGE, number=3, message=common.OperationProgress,)
+    cancel_time = proto.Field(proto.MESSAGE, number=4, message=timestamp_pb2.Timestamp,)
+
+
 class UpdateBackupRequest(proto.Message):
     r"""The request for
     [UpdateBackup][google.spanner.admin.database.v1.DatabaseAdmin.UpdateBackup].
@@ -386,6 +495,8 @@ class ListBackupOperationsRequest(proto.Message):
                is
                ``type.googleapis.com/google.spanner.admin.database.v1.CreateBackupMetadata``.
             -  ``metadata.<field_name>`` - any field in metadata.value.
+               ``metadata.@type`` must be specified first if filtering
+               on metadata fields.
             -  ``error`` - Error associated with the long-running
                operation.
             -  ``response.@type`` - the type of response.
@@ -399,8 +510,14 @@ class ListBackupOperationsRequest(proto.Message):
             Here are a few examples:
 
             -  ``done:true`` - The operation is complete.
-            -  ``metadata.database:prod`` - The database the backup was
-               taken from has a name containing the string "prod".
+            -  ``(metadata.@type=type.googleapis.com/google.spanner.admin.database.v1.CreateBackupMetadata) AND``
+               ``metadata.database:prod`` - Returns operations where:
+
+               -  The operation's metadata type is
+                  [CreateBackupMetadata][google.spanner.admin.database.v1.CreateBackupMetadata].
+               -  The database the backup was taken from has a name
+                  containing the string "prod".
+
             -  ``(metadata.@type=type.googleapis.com/google.spanner.admin.database.v1.CreateBackupMetadata) AND``
                ``(metadata.name:howl) AND``
                ``(metadata.progress.start_time < \"2018-03-28T14:50:00Z\") AND``
@@ -411,6 +528,37 @@ class ListBackupOperationsRequest(proto.Message):
                -  The backup name contains the string "howl".
                -  The operation started before 2018-03-28T14:50:00Z.
                -  The operation resulted in an error.
+
+            -  ``(metadata.@type=type.googleapis.com/google.spanner.admin.database.v1.CopyBackupMetadata) AND``
+               ``(metadata.source_backup:test) AND``
+               ``(metadata.progress.start_time < \"2022-01-18T14:50:00Z\") AND``
+               ``(error:*)`` - Returns operations where:
+
+               -  The operation's metadata type is
+                  [CopyBackupMetadata][google.spanner.admin.database.v1.CopyBackupMetadata].
+               -  The source backup of the copied backup name contains
+                  the string "test".
+               -  The operation started before 2022-01-18T14:50:00Z.
+               -  The operation resulted in an error.
+
+            -  ``((metadata.@type=type.googleapis.com/google.spanner.admin.database.v1.CreateBackupMetadata) AND``
+               ``(metadata.database:test_db)) OR``
+               ``((metadata.@type=type.googleapis.com/google.spanner.admin.database.v1.CopyBackupMetadata) AND``
+               ``(metadata.source_backup:test_bkp)) AND``
+               ``(error:*)`` - Returns operations where:
+
+               -  The operation's metadata matches either of criteria:
+
+                  -  The operation's metadata type is
+                     [CreateBackupMetadata][google.spanner.admin.database.v1.CreateBackupMetadata]
+                     AND the database the backup was taken from has name
+                     containing string "test_db"
+                  -  The operation's metadata type is
+                     [CopyBackupMetadata][google.spanner.admin.database.v1.CopyBackupMetadata]
+                     AND the backup the backup was copied from has name
+                     containing string "test_bkp"
+
+               -  The operation resulted in an error.
         page_size (int):
             Number of operations to be returned in the
             response. If 0 or less, defaults to the server's
@@ -437,11 +585,9 @@ class ListBackupOperationsResponse(proto.Message):
         operations (Sequence[google.longrunning.operations_pb2.Operation]):
             The list of matching backup [long-running
             operations][google.longrunning.Operation]. Each operation's
-            name will be prefixed by the backup's name and the
-            operation's
-            [metadata][google.longrunning.Operation.metadata] will be of
-            type
-            [CreateBackupMetadata][google.spanner.admin.database.v1.CreateBackupMetadata].
+            name will be prefixed by the backup's name. The operation's
+            [metadata][google.longrunning.Operation.metadata] field type
+            ``metadata.type_url`` describes the type of the metadata.
             Operations returned include those that are pending or have
             completed/failed/canceled within the last 7 days. Operations
             returned are ordered by
@@ -520,4 +666,29 @@ class EncryptionType(proto.Enum):
     kms_key_name = proto.Field(proto.STRING, number=2,)
 
 
+class CopyBackupEncryptionConfig(proto.Message):
+    r"""Encryption configuration for the copied backup.
+
+    Attributes:
+        encryption_type (google.cloud.spanner_admin_database_v1.types.CopyBackupEncryptionConfig.EncryptionType):
+            Required. The encryption type of the backup.
+        kms_key_name (str):
+            Optional. The Cloud KMS key that will be used to protect the
+            backup. This field should be set only when
+            [encryption_type][google.spanner.admin.database.v1.CopyBackupEncryptionConfig.encryption_type]
+            is ``CUSTOMER_MANAGED_ENCRYPTION``. Values are of the form
+            ``projects/<project>/locations/<location>/keyRings/<key_ring>/cryptoKeys/<kms_key_name>``.
+    """
+
+    class EncryptionType(proto.Enum):
+        r"""Encryption types for the backup."""
+        ENCRYPTION_TYPE_UNSPECIFIED = 0
+        USE_CONFIG_DEFAULT_OR_BACKUP_ENCRYPTION = 1
+        GOOGLE_DEFAULT_ENCRYPTION = 2
+        CUSTOMER_MANAGED_ENCRYPTION = 3
+
+    encryption_type = proto.Field(proto.ENUM, number=1, enum=EncryptionType,)
+    kms_key_name = proto.Field(proto.STRING, number=2,)
+
+
 __all__ = tuple(sorted(__protobuf__.manifest))
diff --git a/google/cloud/spanner_admin_database_v1/types/spanner_database_admin.py b/google/cloud/spanner_admin_database_v1/types/spanner_database_admin.py
index 42cf4f484f..c9c519334b 100644
--- a/google/cloud/spanner_admin_database_v1/types/spanner_database_admin.py
+++ b/google/cloud/spanner_admin_database_v1/types/spanner_database_admin.py
@@ -447,6 +447,8 @@ class ListDatabaseOperationsRequest(proto.Message):
                is
                ``type.googleapis.com/google.spanner.admin.database.v1.RestoreDatabaseMetadata``.
             -  ``metadata.<field_name>`` - any field in metadata.value.
+               ``metadata.@type`` must be specified first, if filtering
+               on metadata fields.
             -  ``error`` - Error associated with the long-running
                operation.
             -  ``response.@type`` - the type of response.
diff --git a/samples/generated_samples/snippet_metadata_spanner admin database_v1.json b/samples/generated_samples/snippet_metadata_spanner admin database_v1.json
index 10a85bf3f2..5564ff3d37 100644
--- a/samples/generated_samples/snippet_metadata_spanner admin database_v1.json	
+++ b/samples/generated_samples/snippet_metadata_spanner admin database_v1.json	
@@ -1,5 +1,94 @@
 {
   "snippets": [
+    {
+      "clientMethod": {
+        "async": true,
+        "method": {
+          "service": {
+            "shortName": "DatabaseAdmin"
+          },
+          "shortName": "CopyBackup"
+        }
+      },
+      "file": "spanner_v1_generated_database_admin_copy_backup_async.py",
+      "regionTag": "spanner_v1_generated_DatabaseAdmin_CopyBackup_async",
+      "segments": [
+        {
+          "end": 50,
+          "start": 27,
+          "type": "FULL"
+        },
+        {
+          "end": 50,
+          "start": 27,
+          "type": "SHORT"
+        },
+        {
+          "end": 33,
+          "start": 31,
+          "type": "CLIENT_INITIALIZATION"
+        },
+        {
+          "end": 40,
+          "start": 34,
+          "type": "REQUEST_INITIALIZATION"
+        },
+        {
+          "end": 47,
+          "start": 41,
+          "type": "REQUEST_EXECUTION"
+        },
+        {
+          "end": 51,
+          "start": 48,
+          "type": "RESPONSE_HANDLING"
+        }
+      ]
+    },
+    {
+      "clientMethod": {
+        "method": {
+          "service": {
+            "shortName": "DatabaseAdmin"
+          },
+          "shortName": "CopyBackup"
+        }
+      },
+      "file": "spanner_v1_generated_database_admin_copy_backup_sync.py",
+      "regionTag": "spanner_v1_generated_DatabaseAdmin_CopyBackup_sync",
+      "segments": [
+        {
+          "end": 50,
+          "start": 27,
+          "type": "FULL"
+        },
+        {
+          "end": 50,
+          "start": 27,
+          "type": "SHORT"
+        },
+        {
+          "end": 33,
+          "start": 31,
+          "type": "CLIENT_INITIALIZATION"
+        },
+        {
+          "end": 40,
+          "start": 34,
+          "type": "REQUEST_INITIALIZATION"
+        },
+        {
+          "end": 47,
+          "start": 41,
+          "type": "REQUEST_EXECUTION"
+        },
+        {
+          "end": 51,
+          "start": 48,
+          "type": "RESPONSE_HANDLING"
+        }
+      ]
+    },
     {
       "clientMethod": {
         "async": true,
diff --git a/samples/generated_samples/spanner_v1_generated_database_admin_copy_backup_async.py b/samples/generated_samples/spanner_v1_generated_database_admin_copy_backup_async.py
new file mode 100644
index 0000000000..645e606faf
--- /dev/null
+++ b/samples/generated_samples/spanner_v1_generated_database_admin_copy_backup_async.py
@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+# Copyright 2022 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Generated code. DO NOT EDIT!
+#
+# Snippet for CopyBackup
+# NOTE: This snippet has been automatically generated for illustrative purposes only.
+# It may require modifications to work in your environment.
+
+# To install the latest published package dependency, execute the following:
+#   python3 -m pip install google-cloud-spanner-admin-database
+
+
+# [START spanner_v1_generated_DatabaseAdmin_CopyBackup_async]
+from google.cloud import spanner_admin_database_v1
+
+
+async def sample_copy_backup():
+    # Create a client
+    client = spanner_admin_database_v1.DatabaseAdminAsyncClient()
+
+    # Initialize request argument(s)
+    request = spanner_admin_database_v1.CopyBackupRequest(
+        parent="parent_value",
+        backup_id="backup_id_value",
+        source_backup="source_backup_value",
+    )
+
+    # Make the request
+    operation = client.copy_backup(request=request)
+
+    print("Waiting for operation to complete...")
+
+    response = await operation.result()
+
+    # Handle the response
+    print(response)
+
+# [END spanner_v1_generated_DatabaseAdmin_CopyBackup_async]
diff --git a/samples/generated_samples/spanner_v1_generated_database_admin_copy_backup_sync.py b/samples/generated_samples/spanner_v1_generated_database_admin_copy_backup_sync.py
new file mode 100644
index 0000000000..f5babd289c
--- /dev/null
+++ b/samples/generated_samples/spanner_v1_generated_database_admin_copy_backup_sync.py
@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+# Copyright 2022 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Generated code. DO NOT EDIT!
+#
+# Snippet for CopyBackup
+# NOTE: This snippet has been automatically generated for illustrative purposes only.
+# It may require modifications to work in your environment.
+
+# To install the latest published package dependency, execute the following:
+#   python3 -m pip install google-cloud-spanner-admin-database
+
+
+# [START spanner_v1_generated_DatabaseAdmin_CopyBackup_sync]
+from google.cloud import spanner_admin_database_v1
+
+
+def sample_copy_backup():
+    # Create a client
+    client = spanner_admin_database_v1.DatabaseAdminClient()
+
+    # Initialize request argument(s)
+    request = spanner_admin_database_v1.CopyBackupRequest(
+        parent="parent_value",
+        backup_id="backup_id_value",
+        source_backup="source_backup_value",
+    )
+
+    # Make the request
+    operation = client.copy_backup(request=request)
+
+    print("Waiting for operation to complete...")
+
+    response = operation.result()
+
+    # Handle the response
+    print(response)
+
+# [END spanner_v1_generated_DatabaseAdmin_CopyBackup_sync]
diff --git a/scripts/fixup_spanner_admin_database_v1_keywords.py b/scripts/fixup_spanner_admin_database_v1_keywords.py
index 5a0630802f..5c11670473 100644
--- a/scripts/fixup_spanner_admin_database_v1_keywords.py
+++ b/scripts/fixup_spanner_admin_database_v1_keywords.py
@@ -39,6 +39,7 @@ def partition(
 class spanner_admin_databaseCallTransformer(cst.CSTTransformer):
     CTRL_PARAMS: Tuple[str] = ('retry', 'timeout', 'metadata')
     METHOD_TO_PARAMS: Dict[str, Tuple[str]] = {
+        'copy_backup': ('parent', 'backup_id', 'source_backup', 'expire_time', 'encryption_config', ),
         'create_backup': ('parent', 'backup_id', 'backup', 'encryption_config', ),
         'create_database': ('parent', 'create_statement', 'extra_statements', 'encryption_config', 'database_dialect', ),
         'delete_backup': ('name', ),
diff --git a/tests/unit/gapic/spanner_admin_database_v1/test_database_admin.py b/tests/unit/gapic/spanner_admin_database_v1/test_database_admin.py
index de918f8c79..71fb398101 100644
--- a/tests/unit/gapic/spanner_admin_database_v1/test_database_admin.py
+++ b/tests/unit/gapic/spanner_admin_database_v1/test_database_admin.py
@@ -3005,6 +3005,241 @@ async def test_create_backup_flattened_error_async():
         )
 
 
+@pytest.mark.parametrize("request_type", [backup.CopyBackupRequest, dict,])
+def test_copy_backup(request_type, transport: str = "grpc"):
+    client = DatabaseAdminClient(
+        credentials=ga_credentials.AnonymousCredentials(), transport=transport,
+    )
+
+    # Everything is optional in proto3 as far as the runtime is concerned,
+    # and we are mocking out the actual API, so just send an empty request.
+    request = request_type()
+
+    # Mock the actual call within the gRPC stub, and fake the request.
+    with mock.patch.object(type(client.transport.copy_backup), "__call__") as call:
+        # Designate an appropriate return value for the call.
+        call.return_value = operations_pb2.Operation(name="operations/spam")
+        response = client.copy_backup(request)
+
+        # Establish that the underlying gRPC stub method was called.
+        assert len(call.mock_calls) == 1
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == backup.CopyBackupRequest()
+
+    # Establish that the response is the type that we expect.
+    assert isinstance(response, future.Future)
+
+
+def test_copy_backup_empty_call():
+    # This test is a coverage failsafe to make sure that totally empty calls,
+    # i.e. request == None and no flattened fields passed, work.
+    client = DatabaseAdminClient(
+        credentials=ga_credentials.AnonymousCredentials(), transport="grpc",
+    )
+
+    # Mock the actual call within the gRPC stub, and fake the request.
+    with mock.patch.object(type(client.transport.copy_backup), "__call__") as call:
+        client.copy_backup()
+        call.assert_called()
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == backup.CopyBackupRequest()
+
+
+@pytest.mark.asyncio
+async def test_copy_backup_async(
+    transport: str = "grpc_asyncio", request_type=backup.CopyBackupRequest
+):
+    client = DatabaseAdminAsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(), transport=transport,
+    )
+
+    # Everything is optional in proto3 as far as the runtime is concerned,
+    # and we are mocking out the actual API, so just send an empty request.
+    request = request_type()
+
+    # Mock the actual call within the gRPC stub, and fake the request.
+    with mock.patch.object(type(client.transport.copy_backup), "__call__") as call:
+        # Designate an appropriate return value for the call.
+        call.return_value = grpc_helpers_async.FakeUnaryUnaryCall(
+            operations_pb2.Operation(name="operations/spam")
+        )
+        response = await client.copy_backup(request)
+
+        # Establish that the underlying gRPC stub method was called.
+        assert len(call.mock_calls)
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == backup.CopyBackupRequest()
+
+    # Establish that the response is the type that we expect.
+    assert isinstance(response, future.Future)
+
+
+@pytest.mark.asyncio
+async def test_copy_backup_async_from_dict():
+    await test_copy_backup_async(request_type=dict)
+
+
+def test_copy_backup_field_headers():
+    client = DatabaseAdminClient(credentials=ga_credentials.AnonymousCredentials(),)
+
+    # Any value that is part of the HTTP/1.1 URI should be sent as
+    # a field header. Set these to a non-empty value.
+    request = backup.CopyBackupRequest()
+
+    request.parent = "parent/value"
+
+    # Mock the actual call within the gRPC stub, and fake the request.
+    with mock.patch.object(type(client.transport.copy_backup), "__call__") as call:
+        call.return_value = operations_pb2.Operation(name="operations/op")
+        client.copy_backup(request)
+
+        # Establish that the underlying gRPC stub method was called.
+        assert len(call.mock_calls) == 1
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == request
+
+    # Establish that the field header was sent.
+    _, _, kw = call.mock_calls[0]
+    assert ("x-goog-request-params", "parent=parent/value",) in kw["metadata"]
+
+
+@pytest.mark.asyncio
+async def test_copy_backup_field_headers_async():
+    client = DatabaseAdminAsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+    )
+
+    # Any value that is part of the HTTP/1.1 URI should be sent as
+    # a field header. Set these to a non-empty value.
+    request = backup.CopyBackupRequest()
+
+    request.parent = "parent/value"
+
+    # Mock the actual call within the gRPC stub, and fake the request.
+    with mock.patch.object(type(client.transport.copy_backup), "__call__") as call:
+        call.return_value = grpc_helpers_async.FakeUnaryUnaryCall(
+            operations_pb2.Operation(name="operations/op")
+        )
+        await client.copy_backup(request)
+
+        # Establish that the underlying gRPC stub method was called.
+        assert len(call.mock_calls)
+        _, args, _ = call.mock_calls[0]
+        assert args[0] == request
+
+    # Establish that the field header was sent.
+    _, _, kw = call.mock_calls[0]
+    assert ("x-goog-request-params", "parent=parent/value",) in kw["metadata"]
+
+
+def test_copy_backup_flattened():
+    client = DatabaseAdminClient(credentials=ga_credentials.AnonymousCredentials(),)
+
+    # Mock the actual call within the gRPC stub, and fake the request.
+    with mock.patch.object(type(client.transport.copy_backup), "__call__") as call:
+        # Designate an appropriate return value for the call.
+        call.return_value = operations_pb2.Operation(name="operations/op")
+        # Call the method with a truthy value for each flattened field,
+        # using the keyword arguments to the method.
+        client.copy_backup(
+            parent="parent_value",
+            backup_id="backup_id_value",
+            source_backup="source_backup_value",
+            expire_time=timestamp_pb2.Timestamp(seconds=751),
+        )
+
+        # Establish that the underlying call was made with the expected
+        # request object values.
+        assert len(call.mock_calls) == 1
+        _, args, _ = call.mock_calls[0]
+        arg = args[0].parent
+        mock_val = "parent_value"
+        assert arg == mock_val
+        arg = args[0].backup_id
+        mock_val = "backup_id_value"
+        assert arg == mock_val
+        arg = args[0].source_backup
+        mock_val = "source_backup_value"
+        assert arg == mock_val
+        assert TimestampRule().to_proto(args[0].expire_time) == timestamp_pb2.Timestamp(
+            seconds=751
+        )
+
+
+def test_copy_backup_flattened_error():
+    client = DatabaseAdminClient(credentials=ga_credentials.AnonymousCredentials(),)
+
+    # Attempting to call a method with both a request object and flattened
+    # fields is an error.
+    with pytest.raises(ValueError):
+        client.copy_backup(
+            backup.CopyBackupRequest(),
+            parent="parent_value",
+            backup_id="backup_id_value",
+            source_backup="source_backup_value",
+            expire_time=timestamp_pb2.Timestamp(seconds=751),
+        )
+
+
+@pytest.mark.asyncio
+async def test_copy_backup_flattened_async():
+    client = DatabaseAdminAsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+    )
+
+    # Mock the actual call within the gRPC stub, and fake the request.
+    with mock.patch.object(type(client.transport.copy_backup), "__call__") as call:
+        # Designate an appropriate return value for the call.
+        call.return_value = operations_pb2.Operation(name="operations/op")
+
+        call.return_value = grpc_helpers_async.FakeUnaryUnaryCall(
+            operations_pb2.Operation(name="operations/spam")
+        )
+        # Call the method with a truthy value for each flattened field,
+        # using the keyword arguments to the method.
+        response = await client.copy_backup(
+            parent="parent_value",
+            backup_id="backup_id_value",
+            source_backup="source_backup_value",
+            expire_time=timestamp_pb2.Timestamp(seconds=751),
+        )
+
+        # Establish that the underlying call was made with the expected
+        # request object values.
+        assert len(call.mock_calls)
+        _, args, _ = call.mock_calls[0]
+        arg = args[0].parent
+        mock_val = "parent_value"
+        assert arg == mock_val
+        arg = args[0].backup_id
+        mock_val = "backup_id_value"
+        assert arg == mock_val
+        arg = args[0].source_backup
+        mock_val = "source_backup_value"
+        assert arg == mock_val
+        assert TimestampRule().to_proto(args[0].expire_time) == timestamp_pb2.Timestamp(
+            seconds=751
+        )
+
+
+@pytest.mark.asyncio
+async def test_copy_backup_flattened_error_async():
+    client = DatabaseAdminAsyncClient(
+        credentials=ga_credentials.AnonymousCredentials(),
+    )
+
+    # Attempting to call a method with both a request object and flattened
+    # fields is an error.
+    with pytest.raises(ValueError):
+        await client.copy_backup(
+            backup.CopyBackupRequest(),
+            parent="parent_value",
+            backup_id="backup_id_value",
+            source_backup="source_backup_value",
+            expire_time=timestamp_pb2.Timestamp(seconds=751),
+        )
+
+
 @pytest.mark.parametrize("request_type", [backup.GetBackupRequest, dict,])
 def test_get_backup(request_type, transport: str = "grpc"):
     client = DatabaseAdminClient(
@@ -3025,6 +3260,7 @@ def test_get_backup(request_type, transport: str = "grpc"):
             state=backup.Backup.State.CREATING,
             referencing_databases=["referencing_databases_value"],
             database_dialect=common.DatabaseDialect.GOOGLE_STANDARD_SQL,
+            referencing_backups=["referencing_backups_value"],
         )
         response = client.get_backup(request)
 
@@ -3041,6 +3277,7 @@ def test_get_backup(request_type, transport: str = "grpc"):
     assert response.state == backup.Backup.State.CREATING
     assert response.referencing_databases == ["referencing_databases_value"]
     assert response.database_dialect == common.DatabaseDialect.GOOGLE_STANDARD_SQL
+    assert response.referencing_backups == ["referencing_backups_value"]
 
 
 def test_get_backup_empty_call():
@@ -3081,6 +3318,7 @@ async def test_get_backup_async(
                 state=backup.Backup.State.CREATING,
                 referencing_databases=["referencing_databases_value"],
                 database_dialect=common.DatabaseDialect.GOOGLE_STANDARD_SQL,
+                referencing_backups=["referencing_backups_value"],
             )
         )
         response = await client.get_backup(request)
@@ -3098,6 +3336,7 @@ async def test_get_backup_async(
     assert response.state == backup.Backup.State.CREATING
     assert response.referencing_databases == ["referencing_databases_value"]
     assert response.database_dialect == common.DatabaseDialect.GOOGLE_STANDARD_SQL
+    assert response.referencing_backups == ["referencing_backups_value"]
 
 
 @pytest.mark.asyncio
@@ -3246,6 +3485,7 @@ def test_update_backup(request_type, transport: str = "grpc"):
             state=gsad_backup.Backup.State.CREATING,
             referencing_databases=["referencing_databases_value"],
             database_dialect=common.DatabaseDialect.GOOGLE_STANDARD_SQL,
+            referencing_backups=["referencing_backups_value"],
         )
         response = client.update_backup(request)
 
@@ -3262,6 +3502,7 @@ def test_update_backup(request_type, transport: str = "grpc"):
     assert response.state == gsad_backup.Backup.State.CREATING
     assert response.referencing_databases == ["referencing_databases_value"]
     assert response.database_dialect == common.DatabaseDialect.GOOGLE_STANDARD_SQL
+    assert response.referencing_backups == ["referencing_backups_value"]
 
 
 def test_update_backup_empty_call():
@@ -3302,6 +3543,7 @@ async def test_update_backup_async(
                 state=gsad_backup.Backup.State.CREATING,
                 referencing_databases=["referencing_databases_value"],
                 database_dialect=common.DatabaseDialect.GOOGLE_STANDARD_SQL,
+                referencing_backups=["referencing_backups_value"],
             )
         )
         response = await client.update_backup(request)
@@ -3319,6 +3561,7 @@ async def test_update_backup_async(
     assert response.state == gsad_backup.Backup.State.CREATING
     assert response.referencing_databases == ["referencing_databases_value"]
     assert response.database_dialect == common.DatabaseDialect.GOOGLE_STANDARD_SQL
+    assert response.referencing_backups == ["referencing_backups_value"]
 
 
 @pytest.mark.asyncio
@@ -5076,6 +5319,7 @@ def test_database_admin_base_transport():
         "get_iam_policy",
         "test_iam_permissions",
         "create_backup",
+        "copy_backup",
         "get_backup",
         "update_backup",
         "delete_backup",
