Method IssueInitialRanges

be/src/exec/base-sequence-scanner.cc:47–78

#define RETURN_IF_FALSE(x) if (UNLIKELY(!(x))) return parse_status_

Status BaseSequenceScanner::IssueInitialRanges(HdfsScanNodeBase* scan_node,
    const vector<HdfsFileDesc*>& files) {
  DCHECK(!files.empty());
  // Issue just the header range for each file.  When the header is complete,
  // we'll issue the splits for that file.  Splits cannot be processed until the
  // header is parsed (the header object is then shared across splits for that file).
  vector<ScanRange*> header_ranges;
  for (int i = 0; i < files.size(); ++i) {
    ScanRangeMetadata* metadata =
        static_cast<ScanRangeMetadata*>(files[i]->splits[0]->meta_data());
    int64_t header_size = min<int64_t>(HEADER_SIZE, files[i]->file_length);
    // The header is almost always a remote read. Set the disk id to -1 and indicate
    // it is not cached.
    // TODO: add remote disk id and plumb that through to the io mgr.  It should have
    // 1 queue for each NIC as well?
    bool expected_local = false;
    int cache_options = !scan_node->IsDataCacheDisabled() ? BufferOpts::USE_DATA_CACHE :
        BufferOpts::NO_CACHING;
    ScanRange* header_range = scan_node->AllocateScanRange(files[i]->GetFileInfo(),
        header_size, 0, metadata->partition_id, -1, expected_local,
        BufferOpts(cache_options), metadata->original_split);
    ScanRangeMetadata* header_metadata =
            static_cast<ScanRangeMetadata*>(header_range->meta_data());
    header_metadata->is_sequence_header = true;
    header_ranges.push_back(header_range);
  }
  // When the header is parsed, we will issue more AddDiskIoRanges in
  // the scanner threads.
  scan_node->UpdateRemainingScanRangeSubmissions(header_ranges.size());
  RETURN_IF_ERROR(scan_node->AddDiskIoRanges(header_ranges, EnqueueLocation::TAIL));
  return Status::OK();
}

bool BaseSequenceScanner::FileFormatIsSequenceBased(THdfsFileFormat::type format) {
  return format == THdfsFileFormat::SEQUENCE_FILE ||