Method BeginSpilledProbe

be/src/exec/partitioned-hash-join-node.cc:449–484

}

Status PartitionedHashJoinNode::BeginSpilledProbe() {
  VLOG(2) << "BeginSpilledProbe\n" << NodeDebugString();
  DCHECK(input_partition_ == nullptr);
  DCHECK(build_hash_partitions_.hash_partitions == nullptr);
  DCHECK(probe_hash_partitions_.empty());
  DCHECK(!spilled_partitions_.empty());

  PhjBuilderPartition* build_input_partition;
  bool repartitioned;
  RETURN_IF_ERROR(builder_->BeginSpilledProbe(buffer_pool_client(), runtime_profile(),
      &repartitioned, &build_input_partition, &build_hash_partitions_));

  auto it = spilled_partitions_.find(build_input_partition->id());
  DCHECK(it != spilled_partitions_.end())
      << "All spilled build partitions must have a corresponding probe partition";
  input_partition_ = std::move(it->second);
  spilled_partitions_.erase(it);
  DCHECK_EQ(build_input_partition, input_partition_->build_partition());
  DCHECK_EQ(input_partition_->probe_rows()->BytesPinned(false), 0) << NodeDebugString();

  ht_ctx_->set_level(build_input_partition->level() + (repartitioned ? 1 : 0));
  if (!repartitioned && build_input_partition->hash_tbl() == nullptr) {
    // Build skipped the hash table build, which can only happen if there are no probe
    // rows.
    DCHECK_EQ(0, input_partition_->probe_rows()->num_rows())
        << build_input_partition->DebugString() << endl
        << input_partition_->probe_rows()->DebugString();
    return Status::OK();
  } else if (repartitioned) {
    RETURN_IF_ERROR(PrepareForPartitionedProbe());
  } else {
    RETURN_IF_ERROR(PrepareForUnpartitionedProbe());
  }
  COUNTER_ADD(num_probe_rows_partitioned_, input_partition_->probe_rows()->num_rows());
  return Status::OK();
}

Status PartitionedHashJoinNode::ProcessProbeBatch(RowBatch* out_batch) {
  DCHECK_ENUM_EQ(probe_state_, ProbeState::PROBING_IN_BATCH);