Method BuildHashTable

be/src/exec/partitioned-hash-join-builder.cc:1218–1303

}

Status PhjBuilderPartition::BuildHashTable(bool* built) {
  SCOPED_TIMER(parent_->build_hash_table_timer_);
  DCHECK(build_rows_ != nullptr);
  *built = false;

  // Before building the hash table, we need to pin the rows in memory.
  RETURN_IF_ERROR(build_rows_->PinStream(built));
  if (!*built) return Status::OK();

  RuntimeState* state = parent_->runtime_state_;
  HashTableCtx* ctx = parent_->ht_ctx_.get();
  ctx->set_level(level()); // Set the hash function for building the hash table.
  RowBatch batch(parent_->row_desc_, state->batch_size(), parent_->mem_tracker());
  vector<BufferedTupleStream::FlatRowPtr> flat_rows;
  bool eos = false;

  // Allocate the partition-local hash table. Initialize the number of buckets based on
  // the number of build rows (the number of rows is known at this point). This assumes
  // there are no duplicates which can be wrong. However, the upside in the common case
  // (few/no duplicates) is large and the downside when there are is low (a bit more
  // memory; the bucket memory is small compared to the memory needed for all the build
  // side allocations).
  // One corner case is if the stream contains tuples with zero footprint (no materialized
  // slots). If the tuples occupy no space, this implies all rows will be duplicates, so
  // create a small hash table, IMPALA-2256.
  //
  // TODO: Try to allocate the hash table before pinning the stream to avoid needlessly
  // reading all of the spilled rows from disk when we won't succeed anyway.
  int64_t estimated_num_buckets = HashTable::EstimateNumBuckets(build_rows()->num_rows());
  hash_tbl_.reset(HashTable::Create(parent_->ht_allocator_.get(),
      true /* store_duplicates */, parent_->row_desc_->tuple_descriptors().size(),
      build_rows(), 1 << (32 - PhjBuilder::NUM_PARTITIONING_BITS),
      estimated_num_buckets));
  bool success;
  Status status = hash_tbl_->Init(&success);
  if (!status.ok() || !success) goto not_built;
  status = build_rows_->PrepareForRead(false, &success);
  if (!status.ok()) goto not_built;
  DCHECK(success) << "Stream was already pinned.";

  do {
    status = build_rows_->GetNext(&batch, &eos, &flat_rows);
    if (!status.ok()) goto not_built;
    DCHECK_EQ(batch.num_rows(), flat_rows.size());
    DCHECK_LE(batch.num_rows(), hash_tbl_->EmptyBuckets());
    TPrefetchMode::type prefetch_mode = state->query_options().prefetch_mode;

    InsertBatchFn insert_batch_fn;
    if (level() == 0) {
      insert_batch_fn = parent_->insert_batch_fn_level0_.load();
    } else {
      insert_batch_fn = parent_->insert_batch_fn_.load();
    }

    if (insert_batch_fn != nullptr) {
      if (UNLIKELY(
            !insert_batch_fn(this, prefetch_mode, ctx, &batch, flat_rows, &status))) {
        goto not_built;