Method FinalizeBuild

be/src/exec/partitioned-hash-join-builder.cc:376–439

}

Status PhjBuilder::FinalizeBuild(RuntimeState* state) {
  int64_t num_build_rows = 0;
  for (const unique_ptr<PhjBuilderPartition>& partition : hash_partitions_) {
    num_build_rows += partition->build_rows()->num_rows();
    partition->build_rows()->DoneWriting();
  }

  if (num_build_rows > 0) {
    double largest_fraction = 0.0;
    for (const unique_ptr<PhjBuilderPartition>& partition : hash_partitions_) {
      largest_fraction = max(largest_fraction,
          partition->build_rows()->num_rows() / static_cast<double>(num_build_rows));
    }
    COUNTER_SET(largest_partition_percent_, static_cast<int64_t>(largest_fraction * 100));
  }

  if (VLOG_IS_ON(2)) {
    stringstream ss;
    ss << Substitute("PHJ(node_id=$0) partitioned(level=$1) $2 rows into:", join_node_id_,
        hash_partitions_[0]->level(), num_build_rows);
    for (int i = 0; i < hash_partitions_.size(); ++i) {
      PhjBuilderPartition* partition = hash_partitions_[i].get();
      double percent = num_build_rows == 0 ? 0.0 : partition->build_rows()->num_rows()
              * 100 / static_cast<double>(num_build_rows);
      ss << "  " << i << " " << (partition->is_spilled() ? "spilled" : "not spilled")
         << " (fraction=" << fixed << setprecision(2) << percent << "%)" << endl
         << "    #rows:" << partition->build_rows()->num_rows() << endl;
    }
    if (null_aware_partition_ != nullptr) {
      ss << " Null-aware partition: " << null_aware_partition_->DebugString();
    }
    VLOG(2) << ss.str();
  }

  if (ht_ctx_->level() == 0) {
    PublishRuntimeFilters(num_build_rows);
    non_empty_build_ |= (num_build_rows > 0);

    if (join_op_ == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN) {
      if (null_aware_partition_->is_spilled()) {
        // Free up memory for the hash tables of other partitions by unpinning the
        // last block of the null aware partition's stream.
        RETURN_IF_ERROR(null_aware_partition_->Spill(BufferedTupleStream::UNPIN_ALL));
      } else {
        // Invalidate the write iterator so we can safely do concurrent reads later.
        null_aware_partition_->build_rows()->DoneWriting();
      }
    }
  }

  HashJoinState next_state;
  if (state_ == HashJoinState::PARTITIONING_BUILD) {
    next_state = HashJoinState::PARTITIONING_PROBE;
  } else {
    DCHECK_ENUM_EQ(state_, HashJoinState::REPARTITIONING_BUILD);
    next_state = HashJoinState::REPARTITIONING_PROBE;
  }
  RETURN_IF_ERROR(BuildHashTablesAndReserveProbeBuffers(next_state));