Method ProcessScanRangesAndInitSharedState

be/src/exec/hdfs-scan-node-base.cc:245–414

}

Status HdfsScanPlanNode::ProcessScanRangesAndInitSharedState(FragmentState* state) {
  // Initialize the template tuple pool.
  using namespace org::apache::impala::fb;
  shared_state_.template_pool_.reset(new MemPool(state->query_mem_tracker()));
  auto& template_tuple_map_ = shared_state_.partition_template_tuple_map_;
  ObjectPool* obj_pool = shared_state_.obj_pool();
  auto& file_descs = shared_state_.file_descs_;
  HdfsFsCache::HdfsFsMap fs_cache;
  int num_ranges_missing_volume_id = 0;
  int64_t total_splits = 0;
  const vector<const PlanFragmentInstanceCtxPB*>& instance_ctx_pbs =
      state->instance_ctx_pbs();
  auto instance_ctxs = state->instance_ctxs();
  DCHECK_EQ(instance_ctxs.size(), instance_ctx_pbs.size());
  for (int i = 0; i < instance_ctxs.size(); ++i) {
    auto ctx = instance_ctx_pbs[i];
    auto instance_ctx = instance_ctxs[i];
    auto ranges = ctx->per_node_scan_ranges().find(tnode_->node_id);
    if (ranges == ctx->per_node_scan_ranges().end()) continue;
    for (const ScanRangeParamsPB& params : ranges->second.scan_ranges()) {
      DCHECK(params.scan_range().has_hdfs_file_split());
      const HdfsFileSplitPB& split = params.scan_range().hdfs_file_split();
      const org::apache::impala::fb::FbSplitFileMetadata* file_metadata = nullptr;
      if (params.scan_range().has_file_metadata()) {
        file_metadata =
            flatbuffers::GetRoot<org::apache::impala::fb::FbSplitFileMetadata>(
                params.scan_range().file_metadata().c_str());
      }
      HdfsPartitionDescriptor* partition_desc =
          hdfs_table_->GetPartition(split.partition_id());
      if (template_tuple_map_.find(split.partition_id()) == template_tuple_map_.end()) {
        template_tuple_map_[split.partition_id()] =
            InitTemplateTuple(partition_desc->partition_key_value_evals(),
                shared_state_.template_pool_.get());
      }
      // Convert the ScanRangeParamsPB into per-file DiskIO::ScanRange objects and
      // populate partition_ids_, file_descs_, and per_type_files_.
      if (partition_desc == nullptr) {
        // TODO: this should be a DCHECK but we sometimes hit it. It's likely IMPALA-1702.
        LOG(ERROR) << "Bad table descriptor! table_id=" << hdfs_table_->id()
                   << " partition_id=" << split.partition_id() << "\n"
                   << state->fragment()
                   << state->fragment_ctx().DebugString();
        return Status("Query encountered invalid metadata, likely due to IMPALA-1702."
                      " Try rerunning the query.");
      }

      filesystem::path file_path;
      if (hdfs_table_->IsIcebergTable() && split.relative_path().empty()) {
        file_path.append(split.absolute_path(), filesystem::path::codecvt());
      } else {
        file_path.append(partition_desc->location(), filesystem::path::codecvt())
            .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);