<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def execute_partitioned_dml(self, dml, params=None, param_types=None):
 """Execute a partitionable DML statement. :type dml: str :param dml: DML statement :type params: dict, {str -> column value} :param params: values for parameter replacement. Keys must match the names used in ``dml``. :type param_types: dict[str -> Union[dict, .types.Type]] :param param_types: (Optional) maps explicit types for one or more param values; required if parameters are passed. :rtype: int :returns: Count of rows affected by the DML statement. """

        if params is not None:
            if param_types is None:
                raise ValueError("Specify 'param_types' when passing 'params'.")
            params_pb = Struct(
                fields={key: _make_value_pb(value) for key, value in params.items()}
            )
        else:
            params_pb = None

        api = self.spanner_api

        txn_options = TransactionOptions(
            partitioned_dml=TransactionOptions.PartitionedDml()
        )

        metadata = _metadata_with_prefix(self.name)

        with SessionCheckout(self._pool) as session:

            txn = api.begin_transaction(session.name, txn_options, metadata=metadata)

            txn_selector = TransactionSelector(id=txn.id)

            restart = functools.partial(
                api.execute_streaming_sql,
                session.name,
                dml,
                transaction=txn_selector,
                params=params_pb,
                param_types=param_types,
                metadata=metadata,
            )

            iterator = _restart_on_unavailable(restart)

            result_set = StreamedResultSet(iterator)
            list(result_set)  # consume all partials

            return result_set.stats.row_count_lower_bound