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.

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