Method CodegenAddBatchStreamingImpl

be/src/exec/grouping-aggregator.cc:1195–1275

}

Status GroupingAggregatorConfig::CodegenAddBatchStreamingImpl(
    LlvmCodeGen* codegen, TPrefetchMode::type prefetch_mode) {
  DCHECK(is_streaming_preagg_);

  IRFunction::Type ir_fn = IRFunction::GROUPING_AGG_ADD_BATCH_STREAMING_IMPL;
  llvm::Function* add_batch_streaming_impl_fn = codegen->GetFunction(ir_fn, true);
  DCHECK(add_batch_streaming_impl_fn != nullptr);

  // Make agg_idx arg constant.
  llvm::Value* agg_idx_arg = codegen->GetArgument(add_batch_streaming_impl_fn, 2);
  agg_idx_arg->replaceAllUsesWith(codegen->GetI32Constant(agg_idx_));

  // Make needs_serialize arg constant so dead code can be optimised out.
  llvm::Value* needs_serialize_arg = codegen->GetArgument(add_batch_streaming_impl_fn, 3);
  needs_serialize_arg->replaceAllUsesWith(codegen->GetBoolConstant(needs_serialize_));

  // Replace prefetch_mode with constant so branches can be optimised out.
  llvm::Value* prefetch_mode_arg = codegen->GetArgument(add_batch_streaming_impl_fn, 4);
  prefetch_mode_arg->replaceAllUsesWith(codegen->GetI32Constant(prefetch_mode));

  llvm::Function* update_tuple_fn;
  RETURN_IF_ERROR(CodegenUpdateTuple(codegen, &update_tuple_fn));

  llvm::Function* copy_grouping_values_fn;
  RETURN_IF_ERROR(CodegenCopyGroupingValues(codegen, &copy_grouping_values_fn));

  // We only use the top-level hash function for streaming aggregations.
  llvm::Function* hash_fn;
  RETURN_IF_ERROR(
      HashTableCtx::CodegenHashRow(codegen, false, *hash_table_config_, &hash_fn));

  // Codegen HashTable::Equals
  llvm::Function* equals_fn;
  RETURN_IF_ERROR(
      HashTableCtx::CodegenEquals(codegen, true, *hash_table_config_, &equals_fn));

  // Codegen for evaluating input rows
  llvm::Function* eval_grouping_expr_fn;
  RETURN_IF_ERROR(HashTableCtx::CodegenEvalRow(
      codegen, false, *hash_table_config_, &eval_grouping_expr_fn));

  // Replace call sites
  int replaced = codegen->ReplaceCallSites(
      add_batch_streaming_impl_fn, update_tuple_fn, "UpdateTuple");
  DCHECK_REPLACE_COUNT(replaced, 2);

  replaced = codegen->ReplaceCallSites(
      add_batch_streaming_impl_fn, eval_grouping_expr_fn, "EvalProbeRow");
  DCHECK_REPLACE_COUNT(replaced, 1);

  replaced = codegen->ReplaceCallSites(add_batch_streaming_impl_fn, hash_fn, "HashRow");
  DCHECK_REPLACE_COUNT(replaced, 1);

  replaced = codegen->ReplaceCallSites(add_batch_streaming_impl_fn, equals_fn, "Equals");
  DCHECK_REPLACE_COUNT(replaced, 1);

  replaced = codegen->ReplaceCallSites(add_batch_streaming_impl_fn,
      copy_grouping_values_fn, "CopyGroupingValues");