Method StartNextScanRange

be/src/exec/hdfs-scan-node-base.cc:814–846

}

Status HdfsScanNodeBase::StartNextScanRange(const std::vector<FilterContext>& filter_ctxs,
    int64_t* reservation, ScanRange** scan_range) {
  DiskIoMgr* io_mgr = ExecEnv::GetInstance()->disk_io_mgr();
  bool needs_buffers;
  // Loop until we've got a scan range or run out of ranges.
  do {
    RETURN_IF_ERROR(GetNextScanRangeToRead(scan_range, &needs_buffers));
    if (*scan_range == nullptr) return Status::OK();
    if (filter_ctxs.size() > 0) {
      int64_t partition_id =
          static_cast<ScanRangeMetadata*>((*scan_range)->meta_data())->partition_id;
      if (!hdfs_table()->IsIcebergTable() &&
          !PartitionPassesFilters(partition_id, FilterStats::SPLITS_KEY, filter_ctxs)) {
        SkipScanRange(*scan_range);
        *scan_range = nullptr;
      }
    }
  } while (*scan_range == nullptr);
  if (needs_buffers) {
    // Check if we should increase our reservation to read this range more efficiently.
    // E.g. if we are scanning a large text file, we might want extra I/O buffers to
    // improve throughput. Note that if this is a columnar format like Parquet,
    // '*scan_range' is the small footer range only so we won't request an increase.
    int64_t ideal_scan_range_reservation =
        io_mgr->ComputeIdealBufferReservation((*scan_range)->bytes_to_read());
    *reservation = IncreaseReservationIncrementally(*reservation, ideal_scan_range_reservation);
    initial_range_ideal_reservation_stats_->UpdateCounter(ideal_scan_range_reservation);
    initial_range_actual_reservation_stats_->UpdateCounter(*reservation);
    RETURN_IF_ERROR(
        io_mgr->AllocateBuffersForRange(buffer_pool_client(), *scan_range, *reservation));
  }
  return Status::OK();
}

int64_t HdfsScanNodeBase::IncreaseReservationIncrementally(int64_t curr_reservation,
      int64_t ideal_reservation) {