| 788 | } |
| 789 | |
| 790 | void PhjBuilder::CleanUpHashPartitions( |
| 791 | deque<unique_ptr<PhjBuilderPartition>>* output_partitions, RowBatch* batch) { |
| 792 | SCOPED_TIMER(profile()->total_time_counter()); |
| 793 | if (state_ == HashJoinState::REPARTITIONING_PROBE) { |
| 794 | // Finished repartitioning this partition. Discard before pushing more spilled |
| 795 | // partitions onto 'spilled_partitions_'. |
| 796 | DCHECK(!spilled_partitions_.empty()); |
| 797 | spilled_partitions_.pop_back(); |
| 798 | } |
| 799 | |
| 800 | for (int i = 0; i < PARTITION_FANOUT; ++i) { |
| 801 | unique_ptr<PhjBuilderPartition> partition = std::move(hash_partitions_[i]); |
| 802 | if (partition->IsClosed()) continue; |
| 803 | if (partition->is_spilled()) { |
| 804 | DCHECK(partition->hash_tbl() == nullptr) << DebugString(); |
| 805 | DCHECK_EQ(partition->build_rows()->BytesPinned(false), 0) |
| 806 | << "Build was fully unpinned in BuildHashTablesAndPrepareProbeStreams()"; |
| 807 | if (partition->num_spilled_probe_rows() == 0 |
| 808 | && !NeedToProcessUnmatchedBuildRows(join_op_) |
| 809 | && num_probe_threads_ == 1) { |
| 810 | COUNTER_ADD(num_hash_table_builds_skipped_, 1); |
| 811 | partition->Close(nullptr); |
| 812 | } else { |
| 813 | // For shared builds, always add the partition to keep the spilled partitions |
| 814 | // in sync across all the builders and join nodes. |
| 815 | spilled_partitions_.push_back(std::move(partition)); |
| 816 | } |
| 817 | } else if (NeedToProcessUnmatchedBuildRows(join_op_)) { |
| 818 | output_partitions->push_back(std::move(partition)); |
| 819 | } else { |
| 820 | // No more processing is required for this partition. |
| 821 | partition->Close(batch); |
| 822 | } |
| 823 | } |
| 824 | hash_partitions_.clear(); |
| 825 | } |
| 826 | |
| 827 | Status PhjBuilder::DoneProbingSinglePartition(BufferPool::ClientHandle* probe_client, |
| 828 | RuntimeProfile* probe_profile, |
nothing calls this directly
no test coverage detected