Method AddRow

be/src/exec/analytic-eval-node.cc:348–381

}

inline Status AnalyticEvalNode::AddRow(int64_t stream_idx, TupleRow* row) {
  if (fn_scope_ != ROWS || !window_.__isset.window_start ||
      stream_idx - rows_start_offset_ >= curr_partition_idx_) {
    VLOG_ROW << id() << " Update idx=" << stream_idx;
    AggFnEvaluator::Add(analytic_fn_evals_, row, curr_tuple_);
    if (window_.__isset.window_start) {
      VLOG_ROW << id() << " Adding tuple to window at idx=" << stream_idx;
      Tuple* tuple = row->GetTuple(0)->DeepCopy(
          *child(0)->row_desc()->tuple_descriptors()[0], curr_tuple_pool_.get());
      window_tuples_.emplace_back(stream_idx, tuple);
    }
  }

  Status status;
  // Buffer the entire input row to be returned later with the analytic eval results.
  if (UNLIKELY(!input_stream_->AddRow(row, &status))) {
    // AddRow returns false if an error occurs (available via status()) or there is
    // not enough memory (status() is OK). If there isn't enough memory, we unpin
    // the stream and continue writing/reading in unpinned mode.
    // TODO: Consider re-pinning later if the output stream is fully consumed.
    RETURN_IF_ERROR(status);
    RETURN_IF_ERROR(state_->StartSpilling(mem_tracker()));
    RETURN_IF_ERROR(
        input_stream_->UnpinStream(BufferedTupleStream::UNPIN_ALL_EXCEPT_CURRENT));
    VLOG_FILE << id() << " Unpin input stream while adding row idx=" << stream_idx;
    if (!input_stream_->AddRow(row, &status)) {
      // Rows should be added in unpinned mode unless an error occurs.
      RETURN_IF_ERROR(status);
      DCHECK(false);
    }
  }
  DCHECK(status.ok());
  return Status::OK();
}

Status AnalyticEvalNode::AddResultTuple(int64_t stream_idx) {
  VLOG_ROW << id() << " AddResultTuple idx=" << stream_idx;