Method Prepare

be/src/exec/hdfs-scan-node-base.cc:498–546

HdfsScanNodeBase::~HdfsScanNodeBase() {}

Status HdfsScanNodeBase::Prepare(RuntimeState* state) {
  SCOPED_TIMER(runtime_profile_->total_time_counter());
  RETURN_IF_ERROR(ScanNode::Prepare(state));
  AddBytesReadCounters();

  // Prepare collection conjuncts
  for (const auto& entry: conjuncts_map_) {
    TupleDescriptor* tuple_desc = state->desc_tbl().GetTupleDescriptor(entry.first);
    // conjuncts_ are already prepared in ExecNode::Prepare(), don't try to prepare again
    if (tuple_desc == tuple_desc_) {
      conjunct_evals_map_[entry.first] = conjunct_evals();
    } else {
      DCHECK(conjunct_evals_map_[entry.first].empty());
      RETURN_IF_ERROR(ScalarExprEvaluator::Create(entry.second, state, pool_,
          expr_perm_pool(), expr_results_pool(), &conjunct_evals_map_[entry.first]));
    }
  }

  // Prepare stats statistics conjuncts.
  if (stats_tuple_id_ != -1) {
    RETURN_IF_ERROR(ScalarExprEvaluator::Create(stats_conjuncts_, state, pool_,
        expr_perm_pool(), expr_results_pool(), &stats_conjunct_evals_));
  }

  // Check if reservation was enough to allocate at least one buffer. The
  // reservation calculation in HdfsScanNode.java should guarantee this.
  // Hitting this error indicates a misconfiguration or bug.
  int64_t min_buffer_size = ExecEnv::GetInstance()->disk_io_mgr()->min_buffer_size();
  if (scan_range_params_->size() > 0
      && resource_profile_.min_reservation < min_buffer_size) {
    return Status(TErrorCode::INTERNAL_ERROR,
      Substitute("HDFS scan min reservation $0 must be >= min buffer size $1",
       resource_profile_.min_reservation, min_buffer_size));
  }

  // One-time initialization of state that is constant across scan ranges
  iceberg_partition_filtering_pool_.reset(new MemPool(mem_tracker()));
  runtime_profile()->AddInfoString("Table Name", hdfs_table_->fully_qualified_name());

  if (HasRowBatchQueue()) {
    // Add per volume stats to the runtime profile for Non MT scan node.
    PerVolumeStats per_volume_stats;
    stringstream str;
    UpdateHdfsSplitStats(*scan_range_params_, &per_volume_stats);
    PrintHdfsSplitStats(per_volume_stats, &str);
    runtime_profile()->AddInfoString(HDFS_SPLIT_STATS_DESC, str.str());
  }
  return Status::OK();
}

void HdfsScanPlanNode::Codegen(FragmentState* state) {
  DCHECK(state->ShouldCodegen());