| 446 | } |
| 447 | |
| 448 | std::unique_ptr<TreeOfLosers<RowBasedSpillMergeStream>> |
| 449 | SpillPartition::createRowBasedOrderedReader( |
| 450 | memory::MemoryPool* pool, |
| 451 | RowContainer* const rows, |
| 452 | bool canJit, |
| 453 | bool spillUringEnabled) { |
| 454 | #ifdef ENABLE_BOLT_JIT |
| 455 | bolt::jit::CompiledModuleSP jitModule; |
| 456 | if (rows != nullptr && canJit && RowContainer::JITable(rows->keyTypes())) { |
| 457 | // Extract compare flags from sorting keys |
| 458 | std::vector<CompareFlags> cmpFlags; |
| 459 | for (const auto& sortKey : files_[0].sortingKeys) { |
| 460 | cmpFlags.push_back(sortKey.second); |
| 461 | } |
| 462 | |
| 463 | if (cmpFlags.empty()) { |
| 464 | cmpFlags.resize(rows->keyTypes().size(), CompareFlags()); |
| 465 | } |
| 466 | jitModule = std::get<0>(rows->codegenCompare( |
| 467 | rows->keyTypes(), |
| 468 | cmpFlags, |
| 469 | bytedance::bolt::jit::CmpType::CMP_SPILL, |
| 470 | true)); |
| 471 | LOG(INFO) << "JIT enabled for row based spill ordered reader!"; |
| 472 | } |
| 473 | #endif |
| 474 | |
| 475 | std::vector<std::unique_ptr<RowBasedSpillMergeStream>> streams; |
| 476 | streams.reserve(files_.size()); |
| 477 | for (auto& fileInfo : files_) { |
| 478 | BOLT_CHECK(fileInfo.rowInfo.has_value()); |
| 479 | streams.push_back(RowBasedFileSpillMergeStream::create( |
| 480 | RowBasedSpillReadFile::create(fileInfo, pool, spillUringEnabled) |
| 481 | #ifdef ENABLE_BOLT_JIT |
| 482 | , |
| 483 | jitModule |
| 484 | #endif |
| 485 | )); |
| 486 | } |
| 487 | files_.clear(); |
| 488 | // Check if the partition is empty or not. |
| 489 | if (FOLLY_UNLIKELY(streams.empty())) { |
| 490 | return nullptr; |
| 491 | } |
| 492 | return std::make_unique<TreeOfLosers<RowBasedSpillMergeStream>>( |
| 493 | std::move(streams)); |
| 494 | } |
no test coverage detected