Method ProcessChildBatch

be/src/exec/analytic-eval-node.cc:638–726

}

Status AnalyticEvalNode::ProcessChildBatch(RuntimeState* state) {
  // TODO: DCHECK input is sorted (even just first row vs prev_input_tuple_)
  VLOG_FILE << id() << " ProcessChildBatch: " << DebugStateString()
            << " input batch size:" << curr_child_batch_->num_rows()
            << " tuple pool size:" << curr_tuple_pool_->total_allocated_bytes();
  SCOPED_TIMER(evaluation_timer_);

  // BufferedTupleStream::num_rows() returns the total number of rows that have been
  // inserted into the stream (it does not decrease when we read rows), so the index of
  // the next input row that will be inserted will be the current size of the stream.
  int64_t stream_idx = input_stream_->num_rows();

  // The very first row in the stream is handled specially because there is no previous
  // row to compare and we cannot rely on PrevRowCompare() returning true even for the
  // same row pointers if there are NaN values.
  int batch_idx = 0;
  if (UNLIKELY(stream_idx == 0 && curr_child_batch_->num_rows() > 0)) {
    TupleRow* row = curr_child_batch_->GetRow(0);
    RETURN_IF_ERROR(AddRow(0, row));
    RETURN_IF_ERROR(TryAddResultTupleForCurrRow(0));
    prev_input_tuple_ = row->GetTuple(0);
    ++batch_idx;
    ++stream_idx;
  }
  Tuple* child_tuple_cmp_row_tuples[2] = {nullptr, nullptr};
  TupleRow* child_tuple_cmp_row =
      reinterpret_cast<TupleRow*>(child_tuple_cmp_row_tuples);
  for (; batch_idx < curr_child_batch_->num_rows(); ++batch_idx, ++stream_idx) {
    TupleRow* row = curr_child_batch_->GetRow(batch_idx);
    if (has_partition_or_order_by_expr_eval()) {
      // Only set the tuples in child_tuple_cmp_row if there are partition exprs or
      // order by exprs that require comparing the current and previous rows. If there
      // aren't partition or order by exprs (i.e. empty OVER() clause), there was no
      // sort and there could be nullable tuples (whereas the sort node does not produce
      // them), see IMPALA-1562.
      child_tuple_cmp_row->SetTuple(0, prev_input_tuple_);
      child_tuple_cmp_row->SetTuple(1, row->GetTuple(0));
    }
    TryRemoveRowsBeforeWindow(stream_idx);

    // Every row is compared against the previous row to determine if (a) the row
    // starts a new partition or (b) the row does not share the same values for the
    // ordering exprs. When either of these occurs, the analytic_fn_evals_ are
    // finalized and the result tuple is added to result_tuples_ so that it may be
    // added to output rows in GetNextOutputBatch(). When a new partition is found
    // (a), a new, empty result tuple is created and initialized over the
    // analytic_fn_evals_. If the row has different values for the ordering
    // exprs (b), then a new tuple is created but copied from curr_tuple_ because the
    // original is used for one or more previous row(s) but the incremental state still
    // applies to the current row.
    bool next_partition = false;
    if (partition_by_eq_expr_eval_ != nullptr) {
      // partition_by_eq_expr_eval_ checks equality over the predicate exprs
      next_partition = !PrevRowCompare(partition_by_eq_expr_eval_, child_tuple_cmp_row);
    }
    RETURN_IF_ERROR(TryAddResultTupleForPrevRow(
          child_tuple_cmp_row, next_partition, stream_idx));
    if (next_partition) RETURN_IF_ERROR(InitNextPartition(state, stream_idx));