class_docstr: Gather partition statistics and insert them into MySQL.
  
  Statistics are gathered with a dynamically generated Presto query and
  inserted with this format. Stats overwrite themselves if you rerun the
  same date/partition.
  
  .. code-block:: sql
  
      CREATE TABLE hive_stats (
          ds VARCHAR(16),
          table_name VARCHAR(500),
          metric VARCHAR(200),
          value BIGINT
      );
  
  :param metastore_conn_id: Reference to the
      :ref:`Hive Metastore connection id <howto/connection:hive_metastore>`.
  :param table: the source table, in the format ``database.table_name``. (templated)
  :param partition: the source partition. (templated)
  :param extra_exprs: dict of expression to run against the table where
      keys are metric names and values are Presto compatible expressions
  :param excluded_columns: list of columns to exclude, consider
      excluding blobs, large json columns, ...
  :param assignment_func: a function that receives a column name and
      a type, and returns a dict of metric names and an Presto expressions.
      If None is returned, the global defaults are applied. If an
      empty dictionary is returned, no stats are computed for that
      column.
code: def execute(self, context: Context) -> None:
      metastore = HiveMetastoreHook(
          metastore_conn_id=self.metastore_conn_id
      )
      table = metastore.get_table(table_name=self.table)
      field_types = {col.name: col.type for col in table.sd.cols}
  
      exprs: Any = {("", "count"): "COUNT(*)"}
      for col, col_type in list(field_types.items()):
          if self.assignment_func:
              assign_exprs = self.assignment_func(col, col_type)
              if assign_exprs is None:
                  assign_exprs = self.get_default_exprs(col, col_type)
          else:
              assign_exprs = self.get_default_exprs(col, col_type)
          exprs.update(assign_exprs)
      exprs.update(self.extra_exprs)
      exprs_str = ",\n        ".join(
          f"{v} AS {k[0]}__{k[1]}" for k, v in exprs.items()
      )
  
      where_clause_ = [
          f"{k} = '{v}'" for k, v in self.partition.items()
      ]
      where_clause = " AND\n        ".join(where_clause_)
      sql = (
          f"SELECT {exprs_str} FROM {self.table} WHERE {where_clause};"
      )
  
      presto = PrestoHook(presto_conn_id=self.presto_conn_id)
      self.log.info("Executing SQL check: %s", sql)
      row = presto.get_first(sql)
      self.log.info("Record: %s", row)
      if not row:
          raise AirflowException("The query returned None")
  
      part_json = json.dumps(self.partition, sort_keys=True)
  
      self.log.info("Deleting rows from previous runs if they exist")
      mysql = MySqlHook(self.mysql_conn_id)
      sql = f"""
          SELECT 1 FROM hive_stats
          WHERE
              table_name='{self.table}' AND
              partition_repr='{part_json}' AND
              dttm='{self.dttm}'
          LIMIT 1;
          """
      if mysql.get_records(sql):
          sql = f"""
              DELETE FROM hive_stats
              WHERE
                  table_name='{self.table}' AND
                  partition_repr='{part_json}' AND
                  dttm='{self.dttm}';
              """
          mysql.run(sql)
  
      self.log.info("Pivoting and loading cells into the Airflow db")
      rows = [
          (self.ds, self.dttm, self.table, part_json)
          + (r[0][0], r[0][1], r[1])
          for r in zip(exprs, row)
      ]
      mysql.insert_rows(
          table="hive_stats",
          rows=rows,
          target_fields=[
              "ds",
              "dttm",
              "table_name",
              "partition_repr",
              "col",
              "metric",
              "value",
          ],
      )
  
file_path: airflow/airflow/providers/apache/hive/operators/hive_stats.py
imports: from __future__ import annotations
  import json
  import warnings
  from typing import TYPE_CHECKING, Any, Callable, Sequence
  from airflow.exceptions import AirflowException
  from airflow.models import BaseOperator
  from airflow.providers.apache.hive.hooks.hive import HiveMetastoreHook
  from airflow.providers.mysql.hooks.mysql import MySqlHook
  from airflow.providers.presto.hooks.presto import PrestoHook
label_desc: Use image node_id 3 for calling the HiveStatsCollectionOperator obj's underlying member method code with example usage: obj.execute(context) without return types