| 352 | } |
| 353 | |
| 354 | void HashProbe::maybeSetupSpillInput( |
| 355 | const std::optional<SpillPartitionId>& restoredPartitionId, |
| 356 | const SpillPartitionIdSet& spillPartitionIds, |
| 357 | SpillOffsetToBitsSet offsetToJoinBits) { |
| 358 | if (!reuseSpillReader_) { |
| 359 | BOLT_CHECK_NULL(spillInputReader_); |
| 360 | } |
| 361 | |
| 362 | // If 'restoredPartitionId' is not null, then 'table_' is built from the |
| 363 | // spilled build data. Create an unsorted reader to read the probe inputs from |
| 364 | // the corresponding spilled probe partition on disk. |
| 365 | if (restoredPartitionId.has_value()) { |
| 366 | if (restoredPartitionId->isSubRangePartiton()) { |
| 367 | setupSpillRestorForRangePartition(restoredPartitionId); |
| 368 | } else { |
| 369 | auto iter = spillPartitionSet_.find(restoredPartitionId.value()); |
| 370 | BOLT_CHECK(iter != spillPartitionSet_.end()); |
| 371 | BOLT_CHECK(reuseSpillReader_ == false); |
| 372 | auto partition = std::move(iter->second); |
| 373 | BOLT_CHECK_EQ(partition->id(), restoredPartitionId.value()); |
| 374 | spillInputReader_ = partition->createUnorderedReader( |
| 375 | pool(), spillConfig_->spillUringEnabled); |
| 376 | spillPartitionSet_.erase(iter); |
| 377 | probeRangePartition_ = false; |
| 378 | } |
| 379 | } |
| 380 | |
| 381 | BOLT_CHECK_NULL(spiller_); |
| 382 | spillInputPartitionIds_ = spillPartitionIds; |
| 383 | if (spillInputPartitionIds_.empty()) { |
| 384 | return; |
| 385 | } |
| 386 | |
| 387 | // If 'spillInputPartitionIds_' is not empty, then we set up a spiller to |
| 388 | // spill the incoming probe inputs. |
| 389 | const auto& spillConfig = spillConfig_.value(); |
| 390 | uint8_t bitOffset = spillInputPartitionIds_.begin()->partitionBitOffset(); |
| 391 | auto search = offsetToJoinBits->find(bitOffset); |
| 392 | BOLT_CHECK(search != offsetToJoinBits->end()); |
| 393 | LOG(INFO) << __FUNCTION__ |
| 394 | << ": setupSpiller startBit = " << (uint32_t)bitOffset |
| 395 | << ", partitionBits = " << (uint32_t)search->second; |
| 396 | auto* tmpConfig = const_cast<common::SpillConfig*>(&spillConfig); |
| 397 | operatorCtx_->adjustSpillCompressionKind(tmpConfig); |
| 398 | spiller_ = std::make_unique<Spiller>( |
| 399 | Spiller::Type::kHashJoinProbe, |
| 400 | probeType_, |
| 401 | HashBitRange(bitOffset, bitOffset + search->second), |
| 402 | &spillConfig, |
| 403 | spillConfig.maxFileSize); |
| 404 | spiller_->setSpillConfig(&spillConfig); |
| 405 | |
| 406 | // Set the spill partitions to the corresponding ones at the build side. The |
| 407 | // hash probe operator itself won't trigger any spilling. |
| 408 | spiller_->setPartitionsSpilled(toPartitionNumSet(spillInputPartitionIds_)); |
| 409 | |
| 410 | spillHashFunction_ = std::make_unique<HashPartitionFunction>( |
| 411 | spiller_->hashBits(), probeType_, keyChannels_); |
nothing calls this directly
no test coverage detected