| 987 | } |
| 988 | |
| 989 | Status PartitionedHashJoinNode::PrepareForPartitionedProbe() { |
| 990 | DCHECK(builder_->state() == HashJoinState::PARTITIONING_PROBE |
| 991 | || builder_->state() == HashJoinState::REPARTITIONING_PROBE) |
| 992 | << builder_->DebugString(); |
| 993 | DCHECK_EQ(PARTITION_FANOUT, build_hash_partitions_.hash_partitions->size()); |
| 994 | DCHECK(probe_hash_partitions_.empty()); |
| 995 | // Initialize the probe partitions, providing them with probe streams. The reservation |
| 996 | // for the probe streams was obtained from 'builder_' when BeginInitialProbe() |
| 997 | // or BeginSpilledProbe() was called. |
| 998 | vector<unique_ptr<BufferedTupleStream>> probe_streams; |
| 999 | if (input_partition_ != nullptr) { |
| 1000 | DCHECK_ENUM_EQ(builder_->state(), HashJoinState::REPARTITIONING_PROBE); |
| 1001 | // This is a spilled partition - we need to read the probe rows. Memory was reserved |
| 1002 | // in RepartitionBuildInput() for the input stream's read buffer. |
| 1003 | RETURN_IF_ERROR(input_partition_->PrepareForRead()); |
| 1004 | } |
| 1005 | |
| 1006 | bool have_spilled_hash_partitions; |
| 1007 | RETURN_IF_ERROR(CreateProbeHashPartitions(&have_spilled_hash_partitions)); |
| 1008 | |
| 1009 | // Unpin null-aware probe streams if any partitions spilled: we don't want to waste |
| 1010 | // memory pinning the probe streams that might be needed to process spilled partitions. |
| 1011 | if (join_op_ == TJoinOp::NULL_AWARE_LEFT_ANTI_JOIN |
| 1012 | && (have_spilled_hash_partitions |
| 1013 | || builder_->null_aware_partition()->is_spilled())) { |
| 1014 | RETURN_IF_ERROR( |
| 1015 | null_probe_rows_->UnpinStream(BufferedTupleStream::UNPIN_ALL_EXCEPT_CURRENT)); |
| 1016 | RETURN_IF_ERROR(null_aware_probe_partition_->probe_rows()->UnpinStream( |
| 1017 | BufferedTupleStream::UNPIN_ALL_EXCEPT_CURRENT)); |
| 1018 | } |
| 1019 | |
| 1020 | // Initialize the hash_tbl_ caching array. |
| 1021 | for (int i = 0; i < PARTITION_FANOUT; ++i) { |
| 1022 | hash_tbls_[i] = (*build_hash_partitions_.hash_partitions)[i]->hash_tbl(); |
| 1023 | } |
| 1024 | |
| 1025 | // Validate the state of the partitions. |
| 1026 | for (int i = 0; i < PARTITION_FANOUT; ++i) { |
| 1027 | PhjBuilderPartition* build_partition = |
| 1028 | (*build_hash_partitions_.hash_partitions)[i].get(); |
| 1029 | ProbePartition* probe_partition = probe_hash_partitions_[i].get(); |
| 1030 | if (build_partition->IsClosed()) { |
| 1031 | DCHECK(hash_tbls_[i] == nullptr); |
| 1032 | DCHECK(probe_partition == nullptr); |
| 1033 | } else if (build_partition->is_spilled()) { |
| 1034 | DCHECK(hash_tbls_[i] == nullptr); |
| 1035 | DCHECK(probe_partition != nullptr); |
| 1036 | } else { |
| 1037 | DCHECK(hash_tbls_[i] != nullptr); |
| 1038 | DCHECK(probe_partition == nullptr); |
| 1039 | } |
| 1040 | } |
| 1041 | return Status::OK(); |
| 1042 | } |
| 1043 | |
| 1044 | Status PartitionedHashJoinNode::CreateProbeHashPartitions( |
| 1045 | bool* have_spilled_hash_partitions) { |
nothing calls this directly
no test coverage detected