Impala 3.4 SQL查询之ScanRange详解（四）

class ScanRange : public RequestRange {
    struct SubRange {
    int64_t offset;
    int64_t length;
  };
  
  DiskIoMgr* io_mgr_ = nullptr;
  RequestContext* reader_ = nullptr;
  bool read_in_flight_ = false;
  int64_t bytes_read_ = 0;
  std::vector<SubRange> sub_ranges_;
  ......
}

// hdfs-scan-node-base.h
/// This is a pair for partition ID and filename
typedef pair<int64_t, std::string> PartitionFileKey;

/// partition_id, File path => file descriptor (which includes the file's splits)
typedef std::unordered_map<PartitionFileKey, HdfsFileDesc*, pair_hash> FileDescMap;
FileDescMap file_descs_;

struct HdfsFileDesc {
  hdfsFS fs;
  std::string filename;
  int64_t file_length;
  int64_t mtime
  THdfsCompression::type file_compression;
  bool is_erasure_coded;
  std::vector<io::ScanRange*> splits;
};

  for (const TScanRangeParams& params: *scan_range_params_) {
    const THdfsFileSplit& split = params.scan_range.hdfs_file_split;
    partition_ids_.insert(split.partition_id);
    HdfsPartitionDescriptor* partition_desc =
        hdfs_table_->GetPartition(split.partition_id);

    filesystem::path file_path(partition_desc->location());
    file_path.append(split.relative_path, filesystem::path::codecvt());
    const string& native_file_path = file_path.native();

    auto file_desc_map_key = make_pair(partition_desc->id(), native_file_path);
    HdfsFileDesc* file_desc = NULL;
    FileDescMap::iterator file_desc_it = file_descs_.find(file_desc_map_key);
    if (file_desc_it == file_descs_.end()) {
      // Add new file_desc to file_descs_ and per_type_files_
      file_descs_[file_desc_map_key] = file_desc;
      // 省略其余代码
      file_desc = runtime_state_->obj_pool()->Add(new HdfsFileDesc(native_file_path));
      per_type_files_[partition_desc->file_format()].push_back(file_desc);
    } else {
      // File already processed
      file_desc = file_desc_it->second;
    }

    file_desc->splits.push_back(
        AllocateScanRange(file_desc->fs, file_desc->filename.c_str(), split.length,
            split.offset, split.partition_id, params.volume_id, expected_local,
            file_desc->is_erasure_coded, file_desc->mtime, BufferOpts(cache_options)));
  }

// hdfs-scan-node-base.h
  /// File format => file descriptors.
  typedef std::map<THdfsFileFormat::type, std::vector<HdfsFileDesc*>>
    FileFormatsMap;
  FileFormatsMap per_type_files_;

  /// Queue of ranges that have not started being read.  This list is exclusive
  /// with in_flight_ranges.
  InternalQueue<ScanRange> unstarted_scan_ranges_;

ExecFInstance(query-state.cc):697
-Exec(fragment-instance-state.cc):98
--ExecInternal(fragment-instance-state.cc):383
---GetNext(hdfs-scan-node.cc):91
----IssueInitialScanRanges(hdfs-scan-node-base.cc):636
-----IssueInitialRanges(hdfs-parquet-scanner.cc):82
------IssueFooterRanges(hdfs-scanner.cc):837
-------AddDiskIoRanges(hdfs-scan-node.cc):212
--------AddScanRanges(request-context.cc):404
---------AddRangeToDisk(request-context.cc):357
----------unstarted_scan_ranges()->Enqueue
---------AddRangeToDisk(request-context.cc):362
----------num_unstarted_scan_ranges_.Add(1)
---------AddRangeToDisk(request-context.cc):366
----------next_scan_range_to_start()=null ScheduleContext(request-context.cc)
---------AddRangeToDisk(request-context.cc):379
----------num_remaining_ranges_++

    //这里FOOTER_SIZE是一个常量，为1024*100
    int64_t footer_size = min(FOOTER_SIZE, files[i]->file_length);
    int64_t footer_start = files[i]->file_length - footer_size;

    ScanRange* footer_split = FindFooterSplit(files[i]);

    for (int j = 0; j < files[i]->splits.size(); ++j) {
      ScanRange* split = files[i]->splits[j];

      if (!scan_node->IsZeroSlotTableScan() || footer_split == split) {
        ScanRangeMetadata* split_metadata =
            static_cast<ScanRangeMetadata*>(split->meta_data());
        ScanRange* footer_range;
        if (footer_split != nullptr) {
          footer_range = scan_node->AllocateScanRange(files[i]->fs,
              files[i]->filename.c_str(), footer_size, footer_start,
              split_metadata->partition_id, footer_split->disk_id(),
              footer_split->expected_local(), files[i]->is_erasure_coded, files[i]->mtime,
              BufferOpts(footer_split->cache_options()), split);
        }
        footer_ranges.push_back(footer_range);
    }
  // The threads that process the footer will also do the scan.
  if (footer_ranges.size() > 0) {
    RETURN_IF_ERROR(scan_node->AddDiskIoRanges(footer_ranges, EnqueueLocation::TAIL));
  }
  return Status::OK();
}

// HdfsScanner::IssueFooterRanges()
      // If there are no materialized slots (such as count(*) over the table), we can
      // get the result with the file metadata alone and don't need to read any row
      // groups. We only want a single node to process the file footer in this case,
      // which is the node with the footer split.  If it's not a count(*), we create a
      // footer range for the split always.
      if (!scan_node->IsZeroSlotTableScan() || footer_split == split) {

RequestRange* RequestContext::GetNextRequestRange(int disk_id) {
  PerDiskState* request_disk_state = &disk_states_[disk_id];
  unique_lock<mutex> request_lock(lock_);

  if (request_disk_state->next_scan_range_to_start() == nullptr &&
      !request_disk_state->unstarted_scan_ranges()->empty()) {
    ScanRange* new_range = request_disk_state->unstarted_scan_ranges()->Dequeue();
    num_unstarted_scan_ranges_.Add(-1);
    ready_to_start_ranges_.Enqueue(new_range);
    request_disk_state->set_next_scan_range_to_start(new_range);
  }

  if (request_disk_state->in_flight_ranges()->empty()) {
    request_disk_state->DecrementDiskThread(request_lock, this);
    return nullptr;
  }

  RequestRange* range = request_disk_state->in_flight_ranges()->Dequeue();

  request_disk_state->ScheduleContext(request_lock, this, disk_id);
  return range;
}

// RequestContext::GetNextUnstartedRange()
*range = ready_to_start_ranges_.Dequeue();
int disk_id = (*range)->disk_id();
disk_states_[disk_id].set_next_scan_range_to_start(nullptr);