| 555 | } |
| 556 | |
| 557 | Status PartitionedHashJoinNode::GetNext( |
| 558 | RuntimeState* state, RowBatch* out_batch, bool* eos) { |
| 559 | SCOPED_TIMER(runtime_profile_->total_time_counter()); |
| 560 | ScopedGetNextEventAdder ea(this, eos); |
| 561 | RETURN_IF_ERROR(ExecDebugAction(TExecNodePhase::GETNEXT, state)); |
| 562 | DCHECK(!out_batch->AtCapacity()); |
| 563 | |
| 564 | Status status = Status::OK(); |
| 565 | *eos = false; |
| 566 | // Save the number of rows in case GetNext() is called with a non-empty batch, |
| 567 | // which can happen in a subplan. |
| 568 | int num_rows_before = out_batch->num_rows(); |
| 569 | |
| 570 | // This loop executes the 'probe_state_' state machine until either a full batch is |
| 571 | // produced, resources are attached to 'out_batch' that require flushing, or eos |
| 572 | // is reached (i.e. all rows are returned). The next call into GetNext() will resume |
| 573 | // execution of the state machine where the current call into GetNext() left off. |
| 574 | // See the definition of ProbeState for description of the state machine and states. |
| 575 | do { |
| 576 | DCHECK(status.ok()); |
| 577 | DCHECK(builder_->state() != HashJoinState::PARTITIONING_BUILD) |
| 578 | << "Should not be in GetNext() " << static_cast<int>(builder_->state()); |
| 579 | RETURN_IF_CANCELLED(state); |
| 580 | RETURN_IF_ERROR(QueryMaintenance(state)); |
| 581 | switch (probe_state_) { |
| 582 | case ProbeState::PROBING_IN_BATCH: { |
| 583 | // Finish processing rows in the current probe batch. |
| 584 | RETURN_IF_ERROR(ProcessProbeBatch(out_batch)); |
| 585 | DCHECK(out_batch->AtCapacity() || probe_batch_pos_ == probe_batch_->num_rows() |
| 586 | || ht_ctx_->expr_values_cache()->AtEnd()); |
| 587 | if (probe_batch_pos_ == probe_batch_->num_rows() |
| 588 | && current_probe_row_ == nullptr) { |
| 589 | probe_state_ = ProbeState::PROBING_END_BATCH; |
| 590 | } |
| 591 | break; |
| 592 | } |
| 593 | case ProbeState::PROBING_END_BATCH: { |
| 594 | // Try to get the next row batch from the current probe input. |
| 595 | bool probe_eos; |
| 596 | RETURN_IF_ERROR(NextProbeRowBatch(state, out_batch, &probe_eos)); |
| 597 | if (probe_batch_pos_ == 0) { |
| 598 | // Got a batch, need to process it. |
| 599 | probe_state_ = ProbeState::PROBING_IN_BATCH; |
| 600 | } else if (probe_eos) { |
| 601 | DCHECK_EQ(probe_batch_pos_, -1); |
| 602 | if (UseSeparateBuild(state->query_options()) |
| 603 | && !flushed_unattachable_build_buffers_ && ReturnsBuildData(join_op_)) { |
| 604 | // Can't attach build-side data because it may be referenced by multiple |
| 605 | // finstances. Note that this makes the batch AtCapacity(), so we will exit |
| 606 | // the loop below. |
| 607 | // TODO: IMPALA-9411: implement shared ownership of buffers to avoid this. |
| 608 | flushed_unattachable_build_buffers_ = true; |
| 609 | out_batch->MarkNeedsDeepCopy(); |
| 610 | } else { |
| 611 | // Finished processing all the probe rows for the current hash partitions. |
| 612 | // There may be some partitions that need to outpt their unmatched build rows. |
| 613 | RETURN_IF_ERROR(DoneProbing(state, out_batch)); |
| 614 | probe_state_ = output_build_partitions_.empty() ? |
no test coverage detected