Function FinalizePartitionedColumnStats

be/src/exec/incr-stats-util.cc:219–367

namespace impala {

void FinalizePartitionedColumnStats(const TTableSchema& col_stats_schema,
    const vector<TPartitionStats>& existing_part_stats,
    const vector<vector<string>>& expected_partitions, const TRowSet& rowset,
    int32_t num_partition_cols, TAlterTableUpdateStatsParams* params) {
  // The rowset should have the following schema: for every column in the source table,
  // seven columns are produced, one row per partition.
  // <ndv buckets>, <num nulls>, <max width>, <avg width>, <count rows>,
  // <num trues>, <num falses>, <low value>, <high value>
  static const int COLUMNS_PER_STAT = 9;

  const int num_cols =
      (col_stats_schema.columns.size() - num_partition_cols) / COLUMNS_PER_STAT;
  unordered_set<vector<string>> seen_partitions;
  vector<PerColumnStats> stats(num_cols);

  if (rowset.rows.size() > 0) {
    DCHECK_GE(rowset.rows[0].colVals.size(), COLUMNS_PER_STAT);
    params->__isset.partition_stats = true;
    for (const TRow& col_stats_row: rowset.rows) {
      // The last few columns are partition columns that the results are grouped by, and
      // so uniquely identify the partition that these stats belong to.
      vector<string> partition_key_vals;
      partition_key_vals.reserve(col_stats_row.colVals.size());
      for (int j = num_cols * COLUMNS_PER_STAT; j < col_stats_row.colVals.size(); ++j) {
        stringstream ss;
        PrintTColumnValue(col_stats_row.colVals[j], &ss);
        partition_key_vals.push_back(ss.str());
      }
      seen_partitions.insert(partition_key_vals);

      TPartitionStats* part_stat = &params->partition_stats[partition_key_vals];
      part_stat->__isset.intermediate_col_stats = true;
      for (int i = 0; i < num_cols * COLUMNS_PER_STAT; i += COLUMNS_PER_STAT) {
        PerColumnStats* stat = &stats[i / COLUMNS_PER_STAT];
        const string& ndv = col_stats_row.colVals[i].stringVal.value;
        int64_t num_rows = col_stats_row.colVals[i + 4].i64Val.value;
        double avg_width = col_stats_row.colVals[i + 3].doubleVal.value;
        int32_t max_width = col_stats_row.colVals[i + 2].i32Val.value;
        int64_t num_nulls = col_stats_row.colVals[i + 1].i64Val.value;
        int64_t num_trues = col_stats_row.colVals[i + 5].i64Val.value;
        int64_t num_falses = col_stats_row.colVals[i + 6].i64Val.value;
        TColumnValueHive low_value = col_stats_row.colVals[i + 7];
        TColumnValueHive high_value = col_stats_row.colVals[i + 8];

        impala::TColumnValue low_value_impala =
            ConvertToTColumnValue(col_stats_schema.columns[i + 7], low_value);
        impala::TColumnValue high_value_impala =
            ConvertToTColumnValue(col_stats_schema.columns[i + 8], high_value);

        VLOG(3) << "Updated statistics for column=["
                << col_stats_schema.columns[i].columnName << "]," << " statistics={"
                << ndv << "," << num_rows << "，" << avg_width << "," << num_trues
                << "," << max_width << "," << num_nulls << "," << num_falses
                << PrintTColumnValue(low_value_impala) << ","
                << PrintTColumnValue(high_value_impala) << "}";
        stat->Update(ndv, num_rows, avg_width, max_width, num_nulls, num_trues,
            num_falses, low_value_impala, high_value_impala);