Method createRowBasedOrderedReaderWithLength

bolt/exec/Spill.cpp:493–536

  }
  return std::make_unique<TreeOfLosers<RowBasedSpillMergeStream>>(
      std::move(streams));
}

std::unique_ptr<TreeOfLosers<RowBasedSpillMergeStream>>
SpillPartition::createRowBasedOrderedReaderWithLength(
    memory::MemoryPool* pool,
    RowContainer* const rows,
    bool canJit,
    bool spillUringEnabled) {
#ifdef ENABLE_BOLT_JIT
  bolt::jit::CompiledModuleSP jitModule;
  if (rows != nullptr && canJit && RowContainer::JITable(rows->keyTypes())) {
    // Extract compare flags from sorting keys
    std::vector<CompareFlags> cmpFlags;
    for (const auto& sortKey : files_[0].sortingKeys) {
      cmpFlags.push_back(sortKey.second);
    }

    if (cmpFlags.empty()) {
      cmpFlags.resize(rows->keyTypes().size(), CompareFlags());
    }
    jitModule = std::get<0>(rows->codegenCompare(
        rows->keyTypes(),
        cmpFlags,
        bytedance::bolt::jit::CmpType::CMP_SPILL,
        true));
    LOG(INFO) << "JIT enabled for row based spill ordered reader!";
  }
#endif

  std::vector<std::unique_ptr<RowBasedSpillMergeStream>> streams;
  streams.reserve(files_.size());
  for (auto& fileInfo : files_) {
    BOLT_CHECK(fileInfo.rowInfo.has_value());
    streams.push_back(RowBasedFileSpillMergeStream::createWithLength(
        RowBasedSpillReadFile::create(fileInfo, pool, spillUringEnabled)
#ifdef ENABLE_BOLT_JIT
            ,
        jitModule
#endif
        ));
  }
  files_.clear();
  // Check if the partition is empty or not.
  if (FOLLY_UNLIKELY(streams.empty())) {
    return nullptr;
  }