Method Open

be/src/exec/analytic-eval-node.cc:199–258

}

Status AnalyticEvalNode::Open(RuntimeState* state) {
  SCOPED_TIMER(runtime_profile_->total_time_counter());
  ScopedOpenEventAdder ea(this);
  RETURN_IF_ERROR(ExecNode::Open(state));
  RETURN_IF_CANCELLED(state);
  RETURN_IF_ERROR(QueryMaintenance(state));
  RETURN_IF_ERROR(child(0)->Open(state));

  // Claim reservation after the child has been opened to reduce the peak reservation
  // requirement.
  if (!buffer_pool_client()->is_registered()) {
    RETURN_IF_ERROR(ClaimBufferReservation(state));
  }
  DCHECK(input_stream_ == nullptr);
  input_stream_.reset(new BufferedTupleStream(state, child(0)->row_desc(),
      buffer_pool_client(), resource_profile_.spillable_buffer_size,
      resource_profile_.spillable_buffer_size));
  RETURN_IF_ERROR(input_stream_->Init(label(), true));
  bool success;
  RETURN_IF_ERROR(input_stream_->PrepareForReadWrite(true, &success));
  DCHECK(success) << "Had reservation: " << buffer_pool_client()->DebugString();

  for (int i = 0; i < analytic_fn_evals_.size(); ++i) {
    RETURN_IF_ERROR(analytic_fn_evals_[i]->Open(state));
    FunctionContext* agg_fn_ctx = analytic_fn_evals_[i]->agg_fn_ctx();
    if (!has_first_val_null_offset_ &&
        "first_value_rewrite" == analytic_fns_[i]->fn_name() &&
        agg_fn_ctx->GetNumArgs() == 2) {
      DCHECK(!has_first_val_null_offset_);
      first_val_null_offset_ =
          reinterpret_cast<BigIntVal*>(agg_fn_ctx->GetConstantArg(1))->val;
      VLOG_FILE << id() << " FIRST_VAL rewrite null offset: " << first_val_null_offset_;
      has_first_val_null_offset_ = true;
    }
  }

  if (partition_by_eq_expr_eval_ != nullptr) {
    RETURN_IF_ERROR(partition_by_eq_expr_eval_->Open(state));
  }

  if (order_by_eq_expr_eval_ != nullptr) {
    RETURN_IF_ERROR(order_by_eq_expr_eval_->Open(state));
  }

  // Initialize the tuple that was allocated in Prepare().
  // TODO: zeroing out curr_tuple_ shouldn't be strictly necessary.
  curr_tuple_->Init(intermediate_tuple_desc_->byte_size());
  AggFnEvaluator::Init(analytic_fn_evals_, curr_tuple_);
  curr_tuple_init_ = true;
  // Check for failures during AggFnEvaluator::Init().
  RETURN_IF_ERROR(state->GetQueryStatus());

  // Initialize state for the first partition.
  RETURN_IF_ERROR(InitNextPartition(state, 0));
  if (curr_child_batch_ == nullptr) {
    curr_child_batch_.reset(
        new RowBatch(child(0)->row_desc(), state->batch_size(), mem_tracker()));
  }