Impala 2.12.0与3.4.0版本的compute stats兼容问题

  // 3.4.0  ColumnStats.java
  public void validate(Type colType) {
    // avgSize_ and avgSerializedSize_ must be set together.
    Preconditions.checkState(avgSize_ >= 0 == avgSerializedSize_ >= 0, this);

    // Values must be either valid or -1.
    Preconditions.checkState(avgSize_ == -1 || avgSize_ >= 0, this);
    Preconditions.checkState(avgSerializedSize_ == -1 || avgSerializedSize_ >= 0, this);
    Preconditions.checkState(maxSize_ == -1 || maxSize_ >= 0, this);
    Preconditions.checkState(numDistinctValues_ == -1 || numDistinctValues_ >= 0, this);
    Preconditions.checkState(numNulls_ == -1 || numNulls_ >= 0, this);
    if (colType != null && colType.isFixedLengthType()) {
      Preconditions.checkState(avgSize_ == colType.getSlotSize(), this);
      Preconditions.checkState(avgSerializedSize_ == colType.getSlotSize(), this);
      Preconditions.checkState(maxSize_ == colType.getSlotSize(), this);
    }
  }

HdfsTable.load()
-HdfsTable.loadAllPartitions()
--HdfsTable.createPartition() 分区表会循环调用这个函数
---HdfsPartition.ctor() public
----HdfsPartition.ctor() private
-----HdfsPartition.extractAndCompressPartStats()
------PartitionStatsUtil.partStatsBytesFromParameters()
------HdfsPartition.setPartitionStatsBytes() 使用上面函数的返回结果作为输入参数

    // PartitionStatsUtil.java
    public static final String INCREMENTAL_STATS_NUM_CHUNKS =
      "impala_intermediate_stats_num_chunks";
   // 省略后续代码
  /**
   * Reconstructs the intermediate stats from chunks and returns the corresponding
   * byte array. The output byte array is deflate-compressed. Sets hasIncrStats to
   * 'true' if the partition stats contain intermediate col stats.
   */
  public static byte[] partStatsBytesFromParameters(
      Map<String, String> hmsParameters, Reference<Boolean> hasIncrStats) throws
      ImpalaException {
    if (hmsParameters == null) return null;
    String numChunksStr = hmsParameters.get(INCREMENTAL_STATS_NUM_CHUNKS);
    if (numChunksStr == null) return null;
    int numChunks = Integer.parseInt(numChunksStr);
    if (numChunks == 0) return null;
   // 省略后续代码

  // Binary representation of the TPartitionStats for this partition. Populated
  // when the partition is loaded and updated using setPartitionStatsBytes().
  private byte[] partitionStats_;

  // True if partitionStats_ has intermediate_col_stats populated.
  private boolean hasIncrementalStats_ ;
  // 省略后续代码
  public void setPartitionStatsBytes(byte[] partitionStats, boolean hasIncrStats) {
    if (hasIncrStats) Preconditions.checkNotNull(partitionStats);
    partitionStats_ = partitionStats;
    hasIncrementalStats_ = hasIncrStats;
  }

ExecDdl(catalog-server.cc)
-ExecDdl(catalog.cc)
--execDdl(JniCatalog.java)
---CatalogOpExecutor.execDdlRequest
----CatalogOpExecutor.alterTable

CatalogService.TDdlExecRequest
-alter_table_params: JniCatalog.TAlterTableParams
--update_stats_params: JniCatalog.TAlterTableUpdateStatsParams
---partition_stats: map<list<string>, CatalogObjects.TPartitionStats>
----CatalogObjects.TPartitionStats
-----intermediate_col_stats: map<string, TIntermediateColumnStats>
------CatalogObjects.TIntermediateColumnStats
---column_stats: map<string, CatalogObjects.TColumnStats>
----CatalogObjects.TColumnStats

// Per-partition statistics
struct TPartitionStats {
  // so would interfere with the non-incremental stats path
  1: required TTableStats stats

  // Intermediate state for incremental statistics, one entry per column name.
  2: optional map<string, TIntermediateColumnStats> intermediate_col_stats
}

// Intermediate state for the computation of per-column stats. Impala can aggregate these
// structures together to produce final stats for a column.
struct TIntermediateColumnStats {
  // One byte for each bucket of the NDV HLL computation
  1: optional binary intermediate_ndv

  // If true, intermediate_ndv is RLE-compressed
  2: optional bool is_ndv_encoded

  // Number of nulls seen so far (or -1 if nulls are not counted)
  3: optional i64 num_nulls

  // The maximum width, in bytes, of the column
  4: optional i32 max_width

  // The average width (in bytes) of the column
  5: optional double avg_width

  // The number of rows counted, needed to compute NDVs from intermediate_ndv
  6: optional i64 num_rows
}

  // catalog-op-executor.cc ExecComputeStats()
  // 其中FinalizePartitionedColumnStats方法就是用来构造intermediate_col_stats的相关信息
  // col_stats_schema and col_stats_data will be empty if there was no column stats query.
  if (!col_stats_schema.columns.empty()) {
    if (compute_stats_params.is_incremental) {
      RuntimeProfile::Counter* incremental_finalize_timer =
          ADD_TIMER(profile_, "FinalizeIncrementalStatsTimer");
      SCOPED_TIMER(incremental_finalize_timer);
      FinalizePartitionedColumnStats(col_stats_schema,
          compute_stats_params.existing_part_stats,
          compute_stats_params.expected_partitions,
          col_stats_data, compute_stats_params.num_partition_cols, &update_stats_params);
    } else {
      SetColumnStats(col_stats_schema, col_stats_data, &update_stats_params);
    }
  }

CatalogOpExecutor.alterTable
-CatalogOpExecutor.alterTableUpdateStats
--CatalogOpExecutor.alterTableUpdateStatsInner
---CatalogOpExecutor.updatePartitionStats
----PartitionStatsUtil.partStatsToPartition
-----HdfsPartition.setPartitionStatsBytes
---CatalogOpExecutor.bulkAlterPartitions
----HdfsPartition.toHmsPartition
-----PartitionStatsUtil.partStatsToParams

executeAndWait(impala-beeswax-server.cc)
-Execute(impala-server.cc)
--ExecuteInternal(impala-server.cc)
---Exec(client-request-stats.cc)
----ExecDdlRequest(client-request-stats.cc)
-----ExecAsync(child-query.cc)
------ExecChildQueries(child-query.cc)
-------ExecAndFetch(child-query.cc)
-Wait(client-request-state.cc)
--WaitInternal(client-request-state.cc)
---UpdateTableAndColumnStats(client-request-state.cc)
----ExecComputeStats(catalog-op-executor.cc)
-----SetTableStats(catalog-op-executor.cc)
-----FinalizePartitionedColumnStats(incr-stats-util.cc)
------Update(incr-stats-util.cc)

Frontend.TExecRequest
-catalog_op_request: Frontend.TCatalogOpRequest
--ddl_params: CatalogService.TDdlExecRequest
---compute_stats_params: JniCatalog.TComputeStatsParams
----existing_part_stats: list<CatalogObjects.TPartitionStats>
-----CatalogObjects.TPartitionStats
------intermediate_col_stats: map<string, TIntermediateColumnStats>
-------CatalogObjects.TIntermediateColumnStats

  // The list of valid partition statistics that can be used in an incremental computation
  // without themselves being recomputed. Populated in analyze().
  private final List<TPartitionStats> validPartStats_ = new ArrayList<>();

  // ComputeStatsStmt.toThrift()
  public TComputeStatsParams toThrift() {
    TComputeStatsParams params = new TComputeStatsParams();
    params.setTable_name(new TTableName(table_.getDb().getName(), table_.getName()));
    params.setTbl_stats_query(tableStatsQueryStr_);
    if (columnStatsQueryStr_ != null) {
      params.setCol_stats_query(columnStatsQueryStr_);
    } else {
      params.setCol_stats_queryIsSet(false);
    }
    params.setIs_incremental(isIncremental_);
    params.setExisting_part_stats(validPartStats_);
    params.setExpect_all_partitions(expectAllPartitions_);
    // 省略后续代码

  // Updates all aggregate statistics with a new set of measurements.
  void Update(const string& ndv, int64_t num_new_rows, double new_avg_width,
      int32_t max_new_width, int64_t num_new_nulls) {
    DCHECK_EQ(intermediate_ndv.size(), ndv.size()) << "Incompatible intermediate NDVs";
    DCHECK_GE(num_new_rows, 0);
    DCHECK_GE(max_new_width, 0);
    DCHECK_GE(new_avg_width, 0);
    DCHECK_GE(num_new_nulls, 0);
    for (int j = 0; j < ndv.size(); ++j) {
      intermediate_ndv[j] = ::max(intermediate_ndv[j], ndv[j]);
    }
    num_nulls += num_new_nulls;
    max_width = ::max(max_width, max_new_width);
    avg_width += (new_avg_width * num_new_rows);
    num_rows += num_new_rows;
  }

CREATE TABLE stats_test (id INT, name STRING)
PARTITIONED BY (day STRING)
STORED AS PARQUET;
insert into stats_test partition(day='2020-01-01') values(1,'Jack');
insert into stats_test partition(day='2020-01-02') values(1,'Jack');
compute incremental stats stats_test;

compute incremental stats stats_test partition(day='2020-01-01');