<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def create_read_session( self, table_reference, parent, table_modifiers=None, requested_streams=None, read_options=None, format_=None, retry=google.api_core.gapic_v1.method.DEFAULT, timeout=google.api_core.gapic_v1.method.DEFAULT, metadata=None, ):
 """ Creates a new read session. A read session divides the contents of a BigQuery table into one or more streams, which can then be used to read data from the table. The read session also specifies properties of the data to be read, such as a list of columns or a push-down filter describing the rows to be returned. A particular row can be read by at most one stream. When the caller has reached the end of each stream in the session, then all the data in the table has been read. Read sessions automatically expire 24 hours after they are created and do not require manual clean-up by the caller. Example: Args: table_reference (Union[dict, ~google.cloud.bigquery_storage_v1beta1.types.TableReference]):
 Required. Reference to the table to read. If a dict is provided, it must be of the same form as the protobuf message :class:`~google.cloud.bigquery_storage_v1beta1.types.TableReference` parent (str):
 Required. String of the form ``projects/{project_id}`` indicating the project this ReadSession is associated with. This is the project that will be billed for usage. table_modifiers (Union[dict, ~google.cloud.bigquery_storage_v1beta1.types.TableModifiers]):
 Optional. Any modifiers to the Table (e.g. snapshot timestamp). If a dict is provided, it must be of the same form as the protobuf message :class:`~google.cloud.bigquery_storage_v1beta1.types.TableModifiers` requested_streams (int):
 Optional. Initial number of streams. If unset or 0, we will provide a value of streams so as to produce reasonable throughput. Must be non-negative. The number of streams may be lower than the requested number, depending on the amount parallelism that is reasonable for the table and the maximum amount of parallelism allowed by the system. Streams must be read starting from offset 0. read_options (Union[dict, ~google.cloud.bigquery_storage_v1beta1.types.TableReadOptions]):
 Optional. Read options for this session (e.g. column selection, filters). If a dict is provided, it must be of the same form as the protobuf message :class:`~google.cloud.bigquery_storage_v1beta1.types.TableReadOptions` format_ (~google.cloud.bigquery_storage_v1beta1.types.DataFormat):
 Data output format. Currently default to Avro. 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.bigquery_storage_v1beta1.types.ReadSession` 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 "create_read_session" not in self._inner_api_calls:
            self._inner_api_calls[
                "create_read_session"
            ] = google.api_core.gapic_v1.method.wrap_method(
                self.transport.create_read_session,
                default_retry=self._method_configs["CreateReadSession"].retry,
                default_timeout=self._method_configs["CreateReadSession"].timeout,
                client_info=self._client_info,
            )

        request = storage_pb2.CreateReadSessionRequest(
            table_reference=table_reference,
            parent=parent,
            table_modifiers=table_modifiers,
            requested_streams=requested_streams,
            read_options=read_options,
            format=format_,
        )
        if metadata is None:
            metadata = []
        metadata = list(metadata)
        try:
            routing_header = [
                ("table_reference.project_id", table_reference.project_id),
                ("table_reference.dataset_id", table_reference.dataset_id),
            ]
        except AttributeError:
            pass
        else:
            routing_metadata = google.api_core.gapic_v1.routing_header.to_grpc_metadata(  # pragma: no cover
                routing_header
            )
            metadata.append(routing_metadata)  # pragma: no cover

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