Method NextProbeRow

be/src/exec/partitioned-hash-join-node-ir.cc:266–349

template<int const JoinOp>
bool IR_ALWAYS_INLINE PartitionedHashJoinNode::NextProbeRow(
    HashTableCtx* ht_ctx, RowBatch::Iterator* probe_batch_iterator,
    int* remaining_capacity, Status* status) {
  HashTableCtx::ExprValuesCache* expr_vals_cache = ht_ctx->expr_values_cache();
  while (!expr_vals_cache->AtEnd()) {
    // Establish current_probe_row_ and find its corresponding partition.
    DCHECK(!probe_batch_iterator->AtEnd());
    current_probe_row_ = probe_batch_iterator->Get();
    matched_probe_ = false;

    // True if the current row should be skipped for probing.
    bool skip_row = false;

    // The hash of the expressions results for the current probe row.
    uint32_t hash = expr_vals_cache->CurExprValuesHash();
    // Hoist the followings out of the else statement below to speed up non-null case.
    const uint32_t partition_idx = hash >> (32 - NUM_PARTITIONING_BITS);
    HashTable* hash_tbl = hash_tbls_[partition_idx];

    // Fetch the hash and expr values' nullness for this row.
    if (expr_vals_cache->IsRowNull()) {
      if (JoinOp == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN
          && build_hash_partitions_.non_empty_build) {
        const int num_other_join_conjuncts = other_join_conjunct_evals_.size();
        // For NAAJ, we need to treat NULLs on the probe carefully. The logic is:
        // 1. No build rows -> Return this row. The check for 'non_empty_build_'
        //    is for this case.
        // 2. Has build rows & no other join predicates, skip row.
        // 3. Has build rows & other join predicates, we need to evaluate against all
        // build rows. First evaluate it against this partition, and if there is not
        // a match, save it to evaluate against other partitions later. If there
        // is a match, the row is skipped.
        if (num_other_join_conjuncts == 0) {
          // Condition 2 above.
          skip_row = true;
        } else {
          // Condition 3 above.
          if (UNLIKELY(
                  !AppendProbeRow(null_probe_rows_.get(), current_probe_row_, status))) {
            DCHECK(!status->ok());
            return false;
          }
          matched_null_probe_.push_back(false);
          skip_row = true;
        }
      }
    } else {
      // The build partition is in memory. Return this row for probing.
      if (LIKELY(hash_tbl != NULL)) {
        hash_tbl_iterator_ = hash_tbl->FindProbeRow(ht_ctx);
      } else {
        // The build partition is either empty or spilled.
        PhjBuilderPartition* build_partition =
            (*build_hash_partitions_.hash_partitions)[partition_idx].get();
        ProbePartition* probe_partition = probe_hash_partitions_[partition_idx].get();
        DCHECK((build_partition->IsClosed() && probe_partition == NULL)
            || (build_partition->is_spilled() && probe_partition != NULL));