| 484 | } |
| 485 | |
| 486 | Status PartitionedHashJoinNode::ProcessProbeBatch(RowBatch* out_batch) { |
| 487 | DCHECK_ENUM_EQ(probe_state_, ProbeState::PROBING_IN_BATCH); |
| 488 | DCHECK_NE(probe_batch_pos_, -1); |
| 489 | // Putting SCOPED_TIMER in the IR version of ProcessProbeBatch() causes weird exception |
| 490 | // handling IR in the xcompiled function, so call it here instead. |
| 491 | int rows_added = 0; |
| 492 | Status status; |
| 493 | TPrefetchMode::type prefetch_mode = runtime_state_->query_options().prefetch_mode; |
| 494 | SCOPED_TIMER(probe_timer_); |
| 495 | |
| 496 | PartitionedHashJoinPlanNode::ProcessProbeBatchFn process_probe_batch_fn; |
| 497 | if (ht_ctx_->level() == 0) { |
| 498 | process_probe_batch_fn = process_probe_batch_fn_level0_.load(); |
| 499 | } else { |
| 500 | process_probe_batch_fn = process_probe_batch_fn_.load(); |
| 501 | } |
| 502 | |
| 503 | if (process_probe_batch_fn != nullptr) { |
| 504 | rows_added = process_probe_batch_fn( |
| 505 | this, prefetch_mode, out_batch, ht_ctx_.get(), &status); |
| 506 | } else { |
| 507 | rows_added = ProcessProbeBatch( |
| 508 | join_op_, prefetch_mode, out_batch, ht_ctx_.get(), &status); |
| 509 | } |
| 510 | |
| 511 | if (UNLIKELY(rows_added < 0)) { |
| 512 | DCHECK(!status.ok()); |
| 513 | return status; |
| 514 | } |
| 515 | DCHECK(status.ok()); |
| 516 | out_batch->CommitRows(rows_added); |
| 517 | return Status::OK(); |
| 518 | } |
| 519 | |
| 520 | int PartitionedHashJoinNode::ProcessProbeBatch( |
| 521 | const TJoinOp::type join_op, TPrefetchMode::type prefetch_mode, |