| 631 | } |
| 632 | |
| 633 | Status ScanHashTable_exec_task(size_t thread_index, int64_t task_id) { |
| 634 | if (cancelled_) { |
| 635 | return Status::Cancelled("Hash join cancelled"); |
| 636 | } |
| 637 | |
| 638 | int32_t start_row_id = static_cast<int32_t>(hash_table_scan_unit_ * task_id); |
| 639 | int32_t end_row_id = |
| 640 | static_cast<int32_t>(std::min(static_cast<int64_t>(hash_table_keys_.num_rows()), |
| 641 | hash_table_scan_unit_ * (task_id + 1))); |
| 642 | |
| 643 | ThreadLocalState& local_state = local_states_[thread_index]; |
| 644 | RETURN_NOT_OK(InitLocalStateIfNeeded(thread_index)); |
| 645 | |
| 646 | std::vector<int32_t>& id_left = local_state.no_match; |
| 647 | std::vector<int32_t>& id_right = local_state.match; |
| 648 | id_left.clear(); |
| 649 | id_right.clear(); |
| 650 | bool use_left = false; |
| 651 | |
| 652 | bool match_search_value = (join_type_ == JoinType::RIGHT_SEMI); |
| 653 | for (int32_t row_id = start_row_id; row_id < end_row_id; ++row_id) { |
| 654 | if (bit_util::GetBit(has_match_.data(), row_id) == match_search_value) { |
| 655 | id_right.push_back(row_id); |
| 656 | } |
| 657 | } |
| 658 | |
| 659 | if (id_right.empty()) { |
| 660 | return Status::OK(); |
| 661 | } |
| 662 | |
| 663 | if (join_type_ != JoinType::RIGHT_SEMI && join_type_ != JoinType::RIGHT_ANTI) { |
| 664 | use_left = true; |
| 665 | id_left.resize(id_right.size()); |
| 666 | for (size_t i = 0; i < id_left.size(); ++i) { |
| 667 | id_left[i] = RowEncoder::kRowIdForNulls(); |
| 668 | } |
| 669 | } |
| 670 | |
| 671 | RETURN_NOT_OK( |
| 672 | ProbeBatch_OutputOne(thread_index, static_cast<int64_t>(id_right.size()), |
| 673 | use_left ? id_left.data() : nullptr, id_right.data())); |
| 674 | return Status::OK(); |
| 675 | } |
| 676 | |
| 677 | Status ScanHashTable_on_finished(size_t thread_index) { |
| 678 | if (cancelled_) { |