Method Codegen

be/src/runtime/krpc-data-stream-sender.cc:1054–1098

}

void KrpcDataStreamSenderConfig::Codegen(FragmentState* state) {
  LlvmCodeGen* codegen = state->codegen();
  DCHECK(codegen != nullptr);
  const string sender_name = PartitionTypeName() + " Sender";
  if (partition_type_ != TPartitionType::HASH_PARTITIONED) {
    const string& msg = Substitute("not $0",
        partition_type_ == TPartitionType::KUDU ? "supported" : "needed");
    codegen_status_msgs_.emplace_back(
        FragmentState::GenerateCodegenMsg(false, msg, sender_name));
    return;
  }

  llvm::Function* hash_row_fn;
  Status codegen_status = CodegenHashRow(codegen, &hash_row_fn);
  if (codegen_status.ok()) {
    llvm::Function* hash_and_add_rows_fn =
        codegen->GetFunction(IRFunction::KRPC_DSS_HASH_AND_ADD_ROWS, true);
    DCHECK(hash_and_add_rows_fn != nullptr);

    int num_replaced;
    // Replace GetNumChannels() with a constant.
    num_replaced = codegen->ReplaceCallSitesWithValue(hash_and_add_rows_fn,
        codegen->GetI32Constant(num_channels_), "GetNumChannels");
    DCHECK_EQ(num_replaced, 1);

    num_replaced = codegen->ReplaceCallSitesWithValue(hash_and_add_rows_fn,
        codegen->GetI32Constant(input_row_desc_->num_tuples_no_inline()),
        "num_tuples_no_inline");
    DCHECK_EQ(num_replaced, 1);

    // Replace HashRow() with the handcrafted IR function.
    num_replaced = codegen->ReplaceCallSites(hash_and_add_rows_fn,
        hash_row_fn, KrpcDataStreamSender::HASH_ROW_SYMBOL);
    DCHECK_EQ(num_replaced, 1);

    hash_and_add_rows_fn = codegen->FinalizeFunction(hash_and_add_rows_fn);
    if (hash_and_add_rows_fn == nullptr) {
      codegen_status =
          Status("Codegen'd HashAndAddRows() failed verification. See log");
    } else {
      codegen->AddFunctionToJit(hash_and_add_rows_fn, &hash_and_add_rows_fn_);
    }
  }
  AddCodegenStatus(codegen_status, sender_name);
}

uint64_t KrpcDataStreamSender::HashRow(TupleRow* row, uint64_t seed) {
  uint64_t hash_val = seed;