

def execute(self, 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 = {
        ('', 'count'): 'COUNT(*)',
    }
    for (col, col_type) in list(field_types.items()):
        d = {
            
        }
        if self.assignment_func:
            d = self.assignment_func(col, col_type)
            if (d is None):
                d = self.get_default_exprs(col, col_type)
        else:
            d = self.get_default_exprs(col, col_type)
        exprs.update(d)
    exprs.update(self.extra_exprs)
    exprs = OrderedDict(exprs)
    exprs_str = ',\n        '.join([((((v + ' AS ') + k[0]) + '__') + k[1]) for (k, v) in exprs.items()])
    where_clause = ["{0} = '{1}'".format(k, v) for (k, v) in self.partition.items()]
    where_clause = ' AND\n        '.join(where_clause)
    sql = '\n        SELECT\n            {exprs_str}\n        FROM {self.table}\n        WHERE\n            {where_clause};\n        '.format(**locals())
    hook = PrestoHook(presto_conn_id=self.presto_conn_id)
    logging.info(('Executing SQL check: ' + sql))
    row = hook.get_first(hql=sql)
    logging.info(('Record: ' + str(table)))
    if (not row):
        raise AirflowException('The query returned None')
    part_json = json.dumps(self.partition, sort_keys=True)
    logging.info('Deleting rows from previous runs if they exist')
    mysql = MySqlHook(self.mysql_conn_id)
    sql = "\n        SELECT 1 FROM hive_stats\n        WHERE\n            table_name='{self.table}' AND\n            partition_repr='{part_json}' AND\n            dttm='{self.dttm}'\n        LIMIT 1;\n        ".format(**locals())
    if mysql.get_records(sql):
        sql = "\n            DELETE FROM hive_stats\n            WHERE\n                table_name='{self.table}' AND\n                partition_repr='{part_json}' AND\n                dttm='{self.dttm}';\n            ".format(**locals())
        mysql.run(sql)
    logging.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'])

dataset/ETHPy150Open airbnb/airflow/airflow/operators/hive_stats_operator.py HiveStatsCollectionOperator.execute/VarMisuse@313/456 `row`->`table`