code: def execute_partitioned_dml(self, dml, params=None, param_types=None):
          
          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
docs: 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.