Method Open

be/src/exec/hdfs-scanner.cc:93–140

}

Status HdfsScanner::Open(ScannerContext* context) {
  context_ = context;
  RETURN_IF_ERROR(ValidateSlotDescriptors());
  file_metadata_utils_.SetFile(state_, scan_node_->GetFileDesc(
      context->partition_descriptor()->id(), context->GetStream()->filename()));
  stream_ = context->GetStream();

  // Clone the scan node's conjuncts map. The cloned evaluators must be closed by the
  // caller.
  for (const auto& entry: scan_node_->conjuncts_map()) {
    RETURN_IF_ERROR(ScalarExprEvaluator::Clone(&obj_pool_, scan_node_->runtime_state(),
        expr_perm_pool_.get(), context_->expr_results_pool(), entry.second,
        &conjunct_evals_map_[entry.first]));
  }
  DCHECK(conjunct_evals_map_.find(scan_node_->tuple_desc()->id()) !=
         conjunct_evals_map_.end());
  conjunct_evals_ = &conjunct_evals_map_[scan_node_->tuple_desc()->id()];

  // Set up the scan node's dictionary filtering conjuncts map.
  if (scan_node_->thrift_dict_filter_conjuncts_map() != nullptr) {
    for (auto& entry : *(scan_node_->thrift_dict_filter_conjuncts_map())) {
      SlotDescriptor* slot_desc = state_->desc_tbl().GetSlotDescriptor(entry.first);
      TupleId tuple_id = (slot_desc->type().IsCollectionType() ?
          slot_desc->children_tuple_descriptor()->id() :
          slot_desc->parent()->id());
      auto conjunct_evals_it = conjunct_evals_map_.find(tuple_id);
      DCHECK(conjunct_evals_it != conjunct_evals_map_.end());
      const vector<ScalarExprEvaluator*>& conjunct_evals = conjunct_evals_it->second;

      // Convert this slot's list of conjunct indices into a list of pointers
      // into conjunct_evals_.
      for (int conjunct_idx : entry.second) {
        DCHECK_LT(conjunct_idx, conjunct_evals.size());
        DCHECK((conjunct_evals)[conjunct_idx] != nullptr);
        dict_filter_map_[entry.first].push_back((conjunct_evals)[conjunct_idx]);
      }
    }
  }

  std::map<const SlotId, const SlotDescriptor*> slot_descs_written;
  template_tuple_ = file_metadata_utils_.CreateTemplateTuple(
      context_->partition_descriptor()->id(), template_tuple_pool_.get(),
      &slot_descs_written);
  template_tuple_map_[scan_node_->tuple_desc()] = template_tuple_;

  decompress_timer_ = ADD_TIMER(scan_node_->runtime_profile(), "DecompressionTime");
  return Status::OK();
}

Status HdfsScanner::ProcessSplit() {
  DCHECK(scan_node_->HasRowBatchQueue());