Method SampledNdvFinalize

be/src/exprs/aggregate-functions-ir.cc:3177–3282

}

BigIntVal AggregateFunctions::SampledNdvFinalize(FunctionContext* ctx,
    const StringVal& src) {
  SampledNdvState* state = reinterpret_cast<SampledNdvState*>(src.ptr);

  // Generate 'num_points' data points with x=row_count and y=ndv_estimate. These points
  // are used to fit a function for the NDV growth and estimate the real NDV.
  constexpr int num_points =
      SampledNdvState::NUM_HLL_BUCKETS * SampledNdvState::NUM_HLL_BUCKETS;
  int64_t counts[num_points] = { 0 };
  int64_t ndvs[num_points] = { 0 };

  int64_t min_ndv = numeric_limits<int64_t>::max();
  int64_t min_count = numeric_limits<int64_t>::max();
  // We have a fixed number of HLL intermediates to generate data points. Any unique
  // subset of intermediates can be combined to create a new data point. It was
  // empirically determined that 'num_data' points is typically sufficient and there are
  // diminishing returns from generating additional data points.
  // The generation method below was chosen for its simplicity. It successively merges
  // buckets in a rolling window of size NUM_HLL_BUCKETS. Repeating the last data point
  // where all buckets are merged biases the curve fitting to hit that data point which
  // makes sense because that's likely the most accurate one. The number of data points
  // are sufficient for reasonable accuracy.
  int pidx = 0;
  for (int i = 0; i < SampledNdvState::NUM_HLL_BUCKETS; ++i) {
    uint8_t merged_hll_data[DEFAULT_HLL_LEN];
    memset(merged_hll_data, 0, DEFAULT_HLL_LEN);
    StringVal merged_hll(merged_hll_data, DEFAULT_HLL_LEN);
    int64_t merged_count = 0;
    for (int j = 0; j < SampledNdvState::NUM_HLL_BUCKETS; ++j) {
      int bucket_idx = (i + j) % SampledNdvState::NUM_HLL_BUCKETS;
      merged_count += state->buckets[bucket_idx].row_count;
      counts[pidx] = merged_count;
      StringVal hll = StringVal(state->buckets[bucket_idx].hll, DEFAULT_HLL_LEN);
      HllMerge(ctx, hll, &merged_hll);
      ndvs[pidx] = HllFinalEstimate(merged_hll.ptr);
      ++pidx;
    }
    min_count = std::min(min_count, state->buckets[i].row_count);
    min_ndv = std::min(min_ndv, ndvs[i * SampledNdvState::NUM_HLL_BUCKETS]);
  }
  // Based on the point-generation method above the last elements represent the data
  // point where all buckets are merged.
  int64_t max_count = counts[num_points - 1];
  int64_t max_ndv = ndvs[num_points - 1];

  // Scale all values to [0,1] since some objective functions require it (e.g., Sigmoid).
  double count_scale = max_count - min_count;
  double ndv_scale = max_ndv - min_ndv;
  if (count_scale == 0) count_scale = 1.0;
  if (ndv_scale == 0) ndv_scale = 1.0;
  double scaled_counts[num_points];
  double scaled_ndvs[num_points];
  for (int i = 0; i < num_points; ++i) {
    scaled_counts[i] = counts[i] / count_scale;
    scaled_ndvs[i] = ndvs[i] / ndv_scale;
  }

  // List of objective functions. Curve fitting will select the best values for the