<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def partition_read( self, session, table, key_set, transaction=None, index=None, columns=None, partition_options=None, retry=google.api_core.gapic_v1.method.DEFAULT, timeout=google.api_core.gapic_v1.method.DEFAULT, metadata=None, ):
 """ Creates a set of partition tokens that can be used to execute a read operation in parallel. Each of the returned partition tokens can be used by ``StreamingRead`` to specify a subset of the read result to read. The same session and read-only transaction must be used by the PartitionReadRequest used to create the partition tokens and the ReadRequests that use the partition tokens. There are no ordering guarantees on rows returned among the returned partition tokens, or even within each individual StreamingRead call issued with a partition\_token. Partition tokens become invalid when the session used to create them is deleted, is idle for too long, begins a new transaction, or becomes too old. When any of these happen, it is not possible to resume the read, and the whole operation must be restarted from the beginning. Example: Args: session (str):
 Required. The session used to create the partitions. table (str):
 Required. The name of the table in the database to be read. key_set (Union[dict, ~google.cloud.spanner_v1.types.KeySet]):
 Required. ``key_set`` identifies the rows to be yielded. ``key_set`` names the primary keys of the rows in ``table`` to be yielded, unless ``index`` is present. If ``index`` is present, then ``key_set`` instead names index keys in ``index``. It is not an error for the ``key_set`` to name rows that do not exist in the database. Read yields nothing for nonexistent rows. If a dict is provided, it must be of the same form as the protobuf message :class:`~google.cloud.spanner_v1.types.KeySet` transaction (Union[dict, ~google.cloud.spanner_v1.types.TransactionSelector]):
 Read only snapshot transactions are supported, read/write and single use transactions are not. If a dict is provided, it must be of the same form as the protobuf message :class:`~google.cloud.spanner_v1.types.TransactionSelector` index (str):
 If non-empty, the name of an index on ``table``. This index is used instead of the table primary key when interpreting ``key_set`` and sorting result rows. See ``key_set`` for further information. columns (list[str]):
 The columns of ``table`` to be returned for each row matching this request. partition_options (Union[dict, ~google.cloud.spanner_v1.types.PartitionOptions]):
 Additional options that affect how many partitions are created. If a dict is provided, it must be of the same form as the protobuf message :class:`~google.cloud.spanner_v1.types.PartitionOptions` retry (Optional[google.api_core.retry.Retry]):
 A retry object used to retry requests. If ``None`` is specified, requests will not be retried. timeout (Optional[float]):
 The amount of time, in seconds, to wait for the request to complete. Note that if ``retry`` is specified, the timeout applies to each individual attempt. metadata (Optional[Sequence[Tuple[str, str]]]):
 Additional metadata that is provided to the method. Returns: A :class:`~google.cloud.spanner_v1.types.PartitionResponse` instance. Raises: google.api_core.exceptions.GoogleAPICallError: If the request failed for any reason. google.api_core.exceptions.RetryError: If the request failed due to a retryable error and retry attempts failed. ValueError: If the parameters are invalid. """

        # Wrap the transport method to add retry and timeout logic.
        if "partition_read" not in self._inner_api_calls:
            self._inner_api_calls[
                "partition_read"
            ] = google.api_core.gapic_v1.method.wrap_method(
                self.transport.partition_read,
                default_retry=self._method_configs["PartitionRead"].retry,
                default_timeout=self._method_configs["PartitionRead"].timeout,
                client_info=self._client_info,
            )

        request = spanner_pb2.PartitionReadRequest(
            session=session,
            table=table,
            key_set=key_set,
            transaction=transaction,
            index=index,
            columns=columns,
            partition_options=partition_options,
        )
        if metadata is None:
            metadata = []
        metadata = list(metadata)
        try:
            routing_header = [("session", session)]
        except AttributeError:
            pass
        else:
            routing_metadata = google.api_core.gapic_v1.routing_header.to_grpc_metadata(
                routing_header
            )
            metadata.append(routing_metadata)

        return self._inner_api_calls["partition_read"](
            request, retry=retry, timeout=timeout, metadata=metadata
        )