Method SpillReadFileBase

bolt/exec/SpillFile.cpp:591–630

}

SpillReadFileBase::SpillReadFileBase(
    const SpillFileInfo& fileInfo,
    memory::MemoryPool* pool,
    const bool spillUringEnabled)
    : id_(fileInfo.id),
      path_(fileInfo.path),
      size_(fileInfo.size),
      type_(fileInfo.type),
      sortingKeys_(fileInfo.sortingKeys),
      compressionKind_(fileInfo.compressionKind),
      readOptions_{kDefaultUseLosslessTimestamp, compressionKind_},
      serdeKind_(fileInfo.serdeKind),
      serde_(
          serdeKind_.has_value() ? getNamedVectorSerde(*serdeKind_) : nullptr),
      spillUringEnabled_(spillUringEnabled),
      pool_(pool) {
  constexpr uint64_t kMaxReadBufferSize =
      (1 << 20) - AlignedBuffer::kPaddedSize; // 1MB - padding.
  auto fs = filesystems::getFileSystem(path_, nullptr);
  std::unique_ptr<ReadFile> file;
#ifdef IO_URING_SUPPORTED
  file = spillUringEnabled_ ? fs->openAsyncFileForRead(path_)
                            : fs->openFileForRead(path_);
#else
  file = fs->openFileForRead(path_);
#endif

  // For io_uring enabled spill read, maintain two read buffers, one for storing
  // current batch, one for prefetching. If io_uring is disabled or fails to
  // init, just use one buffer for blocking pread.
  int bufferLength = (spillUringEnabled && file->uringEnabled()) ? 2 : 1;
  std::vector<BufferPtr> readBuffers;
  readBuffers.resize(bufferLength);
  for (int i = 0; i < bufferLength; i++) {
    readBuffers[i] = AlignedBuffer::allocate<char>(kMaxReadBufferSize, pool_);
  }

  input_ = std::make_unique<SpillInputStream>(
      std::move(file), readBuffers, spillUringEnabled);
}

bool SpillReadFile::nextBatch(RowVectorPtr& rowVector) {
  if (input_->atEnd()) {