Method IssueInitialScanRanges

be/src/exec/hdfs-scan-node-base.cc:709–774

}

Status HdfsScanNodeBase::IssueInitialScanRanges(RuntimeState* state) {
  DCHECK(!initial_ranges_issued_.Load());
  initial_ranges_issued_.Store(true);
  // We want to decrement this remaining_scan_range_submissions in all cases.
  auto remaining_scan_range_submissions_trigger =
    MakeScopeExitTrigger([&](){ UpdateRemainingScanRangeSubmissions(-1); });

  // No need to issue ranges with limit 0.
  if (ReachedLimitShared()) {
    DCHECK_EQ(limit_, 0);
    return Status::OK();
  }

  if (filter_ctxs_.size() > 0) WaitForRuntimeFilters();
  // Apply dynamic partition-pruning per-file.
  HdfsFileDesc::FileFormatsMap matching_per_type_files;
  std::vector<HdfsFileDesc*>* file_list =
      shared_state_->GetFilesForIssuingScanRangesForInstance(
          runtime_state_->instance_ctx().fragment_instance_id);
  if (file_list == nullptr) return Status::OK();
  for (HdfsFileDesc* file : *file_list) {
    if (FilePassesFilterPredicates(state, file, filter_ctxs_)) {
      matching_per_type_files[file->file_format].push_back(file);
    } else {
      SkipFile(file->file_format, file);
    }
  }

  // Issue initial ranges for all file types. Only call functions for file types that
  // actually exist - trying to add empty lists of ranges can result in spurious
  // CANCELLED errors - see IMPALA-6564.
  for (auto& entry : matching_per_type_files) {
    if (entry.second.empty()) continue;
    // Randomize the order this node processes the files. We want to do this to avoid
    // issuing remote reads to the same DN from different impalads. In file formats such
    // as avro/seq/rc (i.e. splittable with a header), every node first reads the header.
    // If every node goes through the files in the same order, all the remote reads are
    // for the same file meaning a few DN serves a lot of remote reads at the same time.
    random_shuffle(entry.second.begin(), entry.second.end());
    switch (entry.first) {
      case THdfsFileFormat::PARQUET:
        RETURN_IF_ERROR(HdfsParquetScanner::IssueInitialRanges(this, entry.second));
        break;
      case THdfsFileFormat::TEXT:
        RETURN_IF_ERROR(HdfsTextScanner::IssueInitialRanges(this, entry.second));
        break;
      case THdfsFileFormat::SEQUENCE_FILE:
      case THdfsFileFormat::RC_FILE:
      case THdfsFileFormat::AVRO:
        RETURN_IF_ERROR(BaseSequenceScanner::IssueInitialRanges(this, entry.second));
        break;
      case THdfsFileFormat::ORC:
        RETURN_IF_ERROR(HdfsOrcScanner::IssueInitialRanges(this, entry.second));
        break;
      case THdfsFileFormat::JSON:
        RETURN_IF_ERROR(HdfsJsonScanner::IssueInitialRanges(this, entry.second));
        break;
      default: