Method EvaluateNullProbe

be/src/exec/partitioned-hash-join-node.cc:1094–1141

}

Status PartitionedHashJoinNode::EvaluateNullProbe(
    RuntimeState* state, BufferedTupleStream* build) {
  DCHECK(build->is_pinned());
  if (null_probe_rows_ == nullptr || null_probe_rows_->num_rows() == 0) {
    return Status::OK();
  }
  DCHECK_EQ(null_probe_rows_->num_rows(), matched_null_probe_.size());
  bool got_read_buffer;
  RETURN_IF_ERROR(null_probe_rows_->PrepareForRead(false, &got_read_buffer));
  DCHECK(got_read_buffer) << "Probe stream should always have a read or write iterator";

  ScalarExprEvaluator* const* join_conjunct_evals = other_join_conjunct_evals_.data();
  int num_join_conjuncts = other_join_conjuncts_.size();
  RowBatch probe_batch(&probe_row_desc(), runtime_state_->batch_size(), mem_tracker());
  RowBatch build_batch(&build_row_desc(), state->batch_size(), mem_tracker());

  // For each probe row, iterate over all rows in the build table.
  SCOPED_TIMER(null_aware_eval_timer_);
  int64_t probe_row_idx = 0;
  bool probe_stream_eos = false;
  while (!probe_stream_eos) {
    RETURN_IF_ERROR(null_probe_rows_->GetNext(&probe_batch, &probe_stream_eos));
    for (int i = 0; i < probe_batch.num_rows(); ++i, ++probe_row_idx) {
      // This loop may run for a long time. Check for cancellation.
      RETURN_IF_CANCELLED(state);
      if (matched_null_probe_[probe_row_idx]) continue;
      BufferedTupleStream::ReadIterator build_itr;
      RETURN_IF_ERROR(build->PrepareForPinnedRead(&build_itr));
      bool build_eos;
      do {
        RETURN_IF_ERROR(build->GetNext(&build_itr, &build_batch, &build_eos));
        FOREACH_ROW(&build_batch, 0, iter) {
          CreateOutputRow(semi_join_staging_row_, probe_batch.GetRow(i), iter.Get());
          if (ExecNode::EvalConjuncts(
              join_conjunct_evals, num_join_conjuncts, semi_join_staging_row_)) {
            matched_null_probe_[probe_row_idx] = true;
            break;
          }
        }
        build_batch.Reset();
        RETURN_IF_CANCELLED(state);
      } while (!matched_null_probe_[probe_row_idx] && !build_eos);
    }
    probe_batch.Reset();
  }
  DCHECK_EQ(probe_row_idx, null_probe_rows_->num_rows());
  return Status::OK();
}

Status PartitionedHashJoinNode::DoneProbing(RuntimeState* state, RowBatch* batch) {
  DCHECK_ENUM_EQ(probe_state_, ProbeState::PROBING_END_BATCH);