Method MoveWriteRangesHelper

be/src/runtime/tmp-file-mgr.cc:2281–2311

}

Status TmpFileBufferPool::MoveWriteRangesHelper(DiskFile* disk_file,
    vector<TmpFileMgr::WriteDoneCallback>* write_callbacks, bool is_cancelled) {
  Status status = Status::OK();
  auto write_ranges_it = write_ranges_to_add_.find(disk_file);
  if (write_ranges_it != write_ranges_to_add_.end()) {
    auto write_range_it = write_ranges_it->second.begin();
    while (write_range_it != write_ranges_it->second.end()) {
      auto range = *write_range_it;
      DCHECK(range != nullptr);
      if (status.ok() && !is_cancelled) {
        status = range->io_ctx()->AddWriteRange(range);
      } else {
        write_callbacks->push_back(range->callback());
        if (is_cancelled && range->offset() == 0) {
          // If is_cancelled is set, try to remove the range from the write_ranges list.
          // If the range hasn't been popped, it must still be in the write_ranges list.
          if (cur_write_range_ != range) {
            auto key_range_it = write_ranges_iterator_.find(range);
            DCHECK(key_range_it != write_ranges_iterator_.end());
            DCHECK_EQ(*(key_range_it->second.first), range);
            write_ranges_.erase(key_range_it->second.first);
            write_ranges_iterator_.erase(key_range_it);
          }
        }
      }
      write_range_it = write_ranges_it->second.erase(write_range_it);
    }
    write_ranges_to_add_.erase(write_ranges_it);
  }
  return status;
}

Status TmpFileBufferPool::EnqueueWriteRange(io::WriteRange* range, TmpFile* tmp_file) {
  Status status = Status::OK();