| 44 | |
| 45 | public: |
| 46 | Status Init(QueryContext* ctx, JoinType join_type, size_t num_threads, |
| 47 | const HashJoinProjectionMaps* proj_map_left, |
| 48 | const HashJoinProjectionMaps* proj_map_right, |
| 49 | std::vector<JoinKeyCmp> key_cmp, Expression filter, |
| 50 | RegisterTaskGroupCallback register_task_group_callback, |
| 51 | StartTaskGroupCallback start_task_group_callback, |
| 52 | OutputBatchCallback output_batch_callback, |
| 53 | FinishedCallback finished_callback) override { |
| 54 | START_COMPUTE_SPAN(span_, "HashJoinBasicImpl", |
| 55 | {{"detail", filter.ToString()}, |
| 56 | {"join.kind", arrow::acero::ToString(join_type)}, |
| 57 | {"join.threads", static_cast<uint32_t>(num_threads)}}); |
| 58 | |
| 59 | num_threads_ = num_threads; |
| 60 | ctx_ = ctx; |
| 61 | join_type_ = join_type; |
| 62 | schema_[0] = proj_map_left; |
| 63 | schema_[1] = proj_map_right; |
| 64 | key_cmp_ = std::move(key_cmp); |
| 65 | filter_ = std::move(filter); |
| 66 | register_task_group_callback_ = std::move(register_task_group_callback); |
| 67 | start_task_group_callback_ = std::move(start_task_group_callback); |
| 68 | output_batch_callback_ = std::move(output_batch_callback); |
| 69 | finished_callback_ = std::move(finished_callback); |
| 70 | local_states_.resize(num_threads_); |
| 71 | |
| 72 | for (size_t i = 0; i < local_states_.size(); ++i) { |
| 73 | local_states_[i].is_initialized = false; |
| 74 | local_states_[i].is_has_match_initialized = false; |
| 75 | } |
| 76 | |
| 77 | dict_probe_.Init(num_threads_); |
| 78 | |
| 79 | has_hash_table_ = false; |
| 80 | num_batches_produced_.store(0); |
| 81 | cancelled_ = false; |
| 82 | |
| 83 | RegisterBuildHashTable(); |
| 84 | RegisterScanHashTable(); |
| 85 | return Status::OK(); |
| 86 | } |
| 87 | |
| 88 | void Abort(AbortContinuationImpl pos_abort_callback) override { |
| 89 | EVENT(span_, "Abort"); |