Method InitNextPartition

be/src/exec/analytic-eval-node.cc:513–600

}

inline Status AnalyticEvalNode::InitNextPartition(RuntimeState* state,
    int64_t stream_idx) {
  VLOG_FILE << id() << " InitNextPartition idx=" << stream_idx;
  DCHECK_LT(curr_partition_idx_, stream_idx);
  int64_t prev_partition_stream_idx = curr_partition_idx_;
  curr_partition_idx_ = stream_idx;

  // If the window has an end bound preceding the current row, we will have output tuples
  // for rows beyond the previous partition, so they should be removed.  Because
  // result_tuples_ is a sparse structure, the last result tuple of the previous
  // partition may have been added to result_tuples_ with a stream index equal to or
  // beyond curr_partition_idx_. So the last entry in result_tuples_ with a stream index
  // >= curr_partition_idx_ is the last result tuple of the previous partition.  Adding
  // the last result tuple to result_tuples_ with a stream index curr_partition_idx_ - 1
  // ensures that all rows in the previous partition have corresponding analytic results.
  Tuple* prev_partition_last_result_tuple = nullptr;
  while (!result_tuples_.empty() && result_tuples_.back().first >= curr_partition_idx_) {
    DCHECK(fn_scope_ == ROWS && window_.__isset.window_end &&
        window_.window_end.type == TAnalyticWindowBoundaryType::PRECEDING);
    VLOG_ROW << id() << " Removing result past partition idx: "
             << result_tuples_.back().first;
    prev_partition_last_result_tuple = result_tuples_.back().second;
    result_tuples_.pop_back();
  }
  if (prev_partition_last_result_tuple != nullptr) {
    if (result_tuples_.empty() ||
        result_tuples_.back().first < curr_partition_idx_ - 1) {
      // prev_partition_last_result_tuple was the last result tuple in the partition, add
      // it back with the index of the last row in the partition so that all output rows
      // in this partition get the correct value.
      result_tuples_.emplace_back(
          curr_partition_idx_ - 1, prev_partition_last_result_tuple);
    }
    DCHECK(!result_tuples_.empty());
    last_result_idx_ = result_tuples_.back().first;
    VLOG_ROW << id() << " After removing results past partition: "
             << DebugStateString(true);
    DCHECK_EQ(last_result_idx_, curr_partition_idx_ - 1);
    DCHECK_LE(input_stream_->rows_returned(), last_result_idx_);
  }
  DCHECK(result_tuples_.empty() || (last_result_idx_ == result_tuples_.back().first));

  if (fn_scope_ == ROWS && stream_idx > 0 && (!window_.__isset.window_end ||
        window_.window_end.type == TAnalyticWindowBoundaryType::FOLLOWING)) {
    RETURN_IF_ERROR(TryAddRemainingResults(stream_idx, prev_partition_stream_idx));
  }
  window_tuples_.clear();

  VLOG_ROW << id() << " Reset curr_tuple";
  // Call finalize to release resources; result is not needed but the dst tuple must be
  // a tuple described by result_tuple_desc_.
  DCHECK(curr_tuple_init_);
  AggFnEvaluator::Finalize(analytic_fn_evals_, curr_tuple_, dummy_result_tuple_);
  // Re-initialize curr_tuple_.
  // 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_);
  // Check for errors in AggFnEvaluator::Init().