| 61 | using strings::Substitute; |
| 62 | |
| 63 | Status PartitionedHashJoinPlanNode::Init( |
| 64 | const TPlanNode& tnode, FragmentState* state) { |
| 65 | RETURN_IF_ERROR(BlockingJoinPlanNode::Init(tnode, state)); |
| 66 | DCHECK(tnode.__isset.join_node); |
| 67 | DCHECK(tnode.join_node.__isset.hash_join_node); |
| 68 | const vector<TEqJoinCondition>& eq_join_conjuncts = |
| 69 | tnode.join_node.hash_join_node.eq_join_conjuncts; |
| 70 | for (const TEqJoinCondition& eq_join_conjunct : eq_join_conjuncts) { |
| 71 | ScalarExpr* probe_expr; |
| 72 | RETURN_IF_ERROR(ScalarExpr::Create( |
| 73 | eq_join_conjunct.left, probe_row_desc(), state, &probe_expr)); |
| 74 | probe_exprs_.push_back(probe_expr); |
| 75 | ScalarExpr* build_expr; |
| 76 | RETURN_IF_ERROR(ScalarExpr::Create( |
| 77 | eq_join_conjunct.right, build_row_desc(), state, &build_expr)); |
| 78 | build_exprs_.push_back(build_expr); |
| 79 | is_not_distinct_from_.push_back(eq_join_conjunct.is_not_distinct_from); |
| 80 | } |
| 81 | |
| 82 | // other_join_conjuncts_ are evaluated in the context of rows assembled from all build |
| 83 | // and probe tuples; full_row_desc is not necessarily the same as the output row desc, |
| 84 | // e.g., because semi joins only return the build xor probe tuples |
| 85 | RowDescriptor full_row_desc(probe_row_desc(), build_row_desc()); |
| 86 | RETURN_IF_ERROR(ScalarExpr::Create(tnode.join_node.hash_join_node.other_join_conjuncts, |
| 87 | full_row_desc, state, &other_join_conjuncts_)); |
| 88 | DCHECK(tnode.join_node.join_op != TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN |
| 89 | || eq_join_conjuncts.size() == 1); |
| 90 | hash_seed_ = tnode.join_node.hash_join_node.hash_seed; |
| 91 | |
| 92 | hash_table_config_ = state->obj_pool()->Add(new HashTableConfig(build_exprs_, |
| 93 | probe_exprs_, PhjBuilder::HashTableStoresNulls(join_op_, is_not_distinct_from_), |
| 94 | is_not_distinct_from_)); |
| 95 | |
| 96 | // TODO: IMPALA-12265: create the config only if it is necessary |
| 97 | RETURN_IF_ERROR( |
| 98 | PhjBuilderConfig::CreateConfig(state, tnode_->node_id, tnode_->join_node.join_op, |
| 99 | &build_row_desc(), eq_join_conjuncts, tnode_->runtime_filters, |
| 100 | tnode_->join_node.hash_join_node.hash_seed, &phj_builder_config_)); |
| 101 | state->CheckAndAddCodegenDisabledMessage(codegen_status_msgs_); |
| 102 | return Status::OK(); |
| 103 | } |
| 104 | |
| 105 | void PartitionedHashJoinPlanNode::Close() { |
| 106 | ScalarExpr::Close(probe_exprs_); |