| 1141 | } |
| 1142 | |
| 1143 | Status PartitionedHashJoinNode::DoneProbing(RuntimeState* state, RowBatch* batch) { |
| 1144 | DCHECK_ENUM_EQ(probe_state_, ProbeState::PROBING_END_BATCH); |
| 1145 | DCHECK_EQ(probe_batch_pos_, -1); |
| 1146 | DCHECK(output_build_partitions_.empty()); |
| 1147 | // At this point all the rows have been read from the probe side for all partitions in |
| 1148 | // hash_partitions_. |
| 1149 | VLOG(2) << "Probe Side Consumed\n" << NodeDebugString(); |
| 1150 | // Clean up input partition first to free up probe reservation before calling |
| 1151 | // DoneProbing*(). |
| 1152 | if (input_partition_ != nullptr) { |
| 1153 | input_partition_->Close(batch); |
| 1154 | input_partition_.reset(); |
| 1155 | } |
| 1156 | if (builder_->state() == HashJoinState::PROBING_SPILLED_PARTITION) { |
| 1157 | // Need to clean up single in-memory build partition instead of hash partitions. |
| 1158 | DCHECK(build_hash_partitions_.hash_partitions == nullptr); |
| 1159 | RETURN_IF_ERROR( |
| 1160 | builder_->DoneProbingSinglePartition(buffer_pool_client(), runtime_profile(), |
| 1161 | &output_build_partitions_, IsLeftSemiJoin(join_op_) ? nullptr : batch)); |
| 1162 | } else { |
| 1163 | // Walk the partitions that had hash tables built for the probe phase and either |
| 1164 | // close them or move them to 'spilled_partitions_'. |
| 1165 | DCHECK_EQ(build_hash_partitions_.hash_partitions->size(), PARTITION_FANOUT); |
| 1166 | DCHECK_EQ(probe_hash_partitions_.size(), PARTITION_FANOUT); |
| 1167 | int64_t num_spilled_probe_rows[PARTITION_FANOUT] = {0}; |
| 1168 | for (int i = 0; i < PARTITION_FANOUT; ++i) { |
| 1169 | ProbePartition* probe_partition = probe_hash_partitions_[i].get(); |
| 1170 | PhjBuilderPartition* build_partition = |
| 1171 | (*build_hash_partitions_.hash_partitions)[i].get(); |
| 1172 | if (probe_partition == nullptr) { |
| 1173 | // Partition was not spilled. |
| 1174 | if (join_op_ == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN) { |
| 1175 | // For NAAJ, we need to try to match the NULL probe rows with this build |
| 1176 | // partition before we are done with it. |
| 1177 | if (!build_partition->IsClosed()) { |
| 1178 | RETURN_IF_ERROR(EvaluateNullProbe(state, build_partition->build_rows())); |
| 1179 | } |
| 1180 | } |
| 1181 | } else if (probe_partition->probe_rows()->num_rows() != 0 |
| 1182 | || NeedToProcessUnmatchedBuildRows(join_op_) |
| 1183 | || builder_->num_probe_threads() > 1) { |
| 1184 | num_spilled_probe_rows[i] = probe_partition->probe_rows()->num_rows(); |
| 1185 | // Unpin the probe stream to free up more memory. We need to free all memory so we |
| 1186 | // can recurse the algorithm and create new hash partitions from spilled |
| 1187 | // partitions. |
| 1188 | RETURN_IF_ERROR( |
| 1189 | probe_partition->probe_rows()->UnpinStream(BufferedTupleStream::UNPIN_ALL)); |
| 1190 | spilled_partitions_.emplace( |
| 1191 | build_partition->id(), std::move(probe_hash_partitions_[i])); |
| 1192 | } else { |
| 1193 | // There's no more processing to do for this partition, and since there were no |
| 1194 | // probe rows we didn't return any rows that reference memory from these |
| 1195 | // partitions, so just free the resources. |
| 1196 | // Avoid doing this for shared builds so that all probe threads have the same |
| 1197 | // number of partitions, which simplifies logic. |
| 1198 | probe_partition->Close(nullptr); |
| 1199 | } |
| 1200 | } |
nothing calls this directly
no test coverage detected