| 924 | } |
| 925 | |
| 926 | Status PartitionedHashJoinNode::OutputNullAwareProbeRows( |
| 927 | RuntimeState* state, RowBatch* out_batch, bool* done) { |
| 928 | DCHECK_ENUM_EQ(probe_state_, ProbeState::OUTPUTTING_NULL_AWARE); |
| 929 | DCHECK(null_aware_probe_partition_ != nullptr); |
| 930 | *done = false; |
| 931 | ScalarExprEvaluator* const* join_conjunct_evals = other_join_conjunct_evals_.data(); |
| 932 | int num_join_conjuncts = other_join_conjuncts_.size(); |
| 933 | DCHECK(probe_batch_ != nullptr); |
| 934 | |
| 935 | BufferedTupleStream* probe_stream = null_aware_probe_partition_->probe_rows(); |
| 936 | if (probe_batch_pos_ == probe_batch_->num_rows()) { |
| 937 | probe_batch_pos_ = 0; |
| 938 | probe_batch_->TransferResourceOwnership(out_batch); |
| 939 | if (out_batch->AtCapacity()) return Status::OK(); |
| 940 | |
| 941 | // Get the next probe batch. |
| 942 | bool eos; |
| 943 | RETURN_IF_ERROR(probe_stream->GetNext(probe_batch_.get(), &eos)); |
| 944 | |
| 945 | if (probe_batch_->num_rows() == 0 && eos) { |
| 946 | RETURN_IF_ERROR( |
| 947 | EvaluateNullProbe(state, builder_->null_aware_partition()->build_rows())); |
| 948 | RETURN_IF_ERROR(PrepareNullAwareNullProbe()); |
| 949 | *done = true; |
| 950 | return Status::OK(); |
| 951 | } |
| 952 | } |
| 953 | |
| 954 | RowBatch null_build_batch(&build_row_desc(), state->batch_size(), mem_tracker()); |
| 955 | // For each probe row, iterate over all the build rows and check for rows |
| 956 | // that did not have any matches. |
| 957 | for (; probe_batch_pos_ < probe_batch_->num_rows(); ++probe_batch_pos_) { |
| 958 | if (out_batch->AtCapacity()) break; |
| 959 | TupleRow* probe_row = probe_batch_->GetRow(probe_batch_pos_); |
| 960 | bool matched = false; |
| 961 | BufferedTupleStream* null_build_stream = |
| 962 | builder_->null_aware_partition()->build_rows(); |
| 963 | DCHECK(null_build_stream->is_pinned()); |
| 964 | BufferedTupleStream::ReadIterator build_itr; |
| 965 | RETURN_IF_ERROR(null_build_stream->PrepareForPinnedRead(&build_itr)); |
| 966 | bool eos; |
| 967 | do { |
| 968 | RETURN_IF_ERROR(null_build_stream->GetNext(&build_itr, &null_build_batch, &eos)); |
| 969 | FOREACH_ROW(&null_build_batch, 0, iter) { |
| 970 | CreateOutputRow(semi_join_staging_row_, probe_row, iter.Get()); |
| 971 | if (ExecNode::EvalConjuncts( |
| 972 | join_conjunct_evals, num_join_conjuncts, semi_join_staging_row_)) { |
| 973 | matched = true; |
| 974 | break; |
| 975 | } |
| 976 | } |
| 977 | null_build_batch.Reset(); |
| 978 | RETURN_IF_CANCELLED(state); |
| 979 | } while (!matched && !eos); |
| 980 | if (!matched) { |
| 981 | TupleRow* out_row = out_batch->GetRow(out_batch->AddRow()); |
| 982 | out_batch->CopyRow(probe_row, out_row); |
| 983 | out_batch->CommitLastRow(); |
nothing calls this directly
no test coverage detected