| 458 | } |
| 459 | |
| 460 | inline Status AnalyticEvalNode::TryAddRemainingResults(int64_t partition_idx, |
| 461 | int64_t prev_partition_idx) { |
| 462 | DCHECK_LT(prev_partition_idx, partition_idx); |
| 463 | // For PARTITION, RANGE, or ROWS with UNBOUNDED PRECEDING: add a result tuple for the |
| 464 | // remaining rows in the partition that do not have an associated result tuple yet. |
| 465 | if (fn_scope_ != ROWS || !window_.__isset.window_end) { |
| 466 | if (last_result_idx_ < partition_idx - 1) { |
| 467 | RETURN_IF_ERROR(AddResultTuple(partition_idx - 1)); |
| 468 | } |
| 469 | return Status::OK(); |
| 470 | } |
| 471 | |
| 472 | // lead() is re-written to a ROWS window with an end bound FOLLOWING. Any remaining |
| 473 | // results need the default value (set by Init()). If this is the case, the start bound |
| 474 | // is UNBOUNDED PRECEDING (DCHECK in Init()). |
| 475 | for (int i = 0; i < analytic_fn_evals_.size(); ++i) { |
| 476 | if (is_lead_fn_[i]) { |
| 477 | // Needs to call Finalize() to release resources. |
| 478 | analytic_fn_evals_[i]->Finalize(curr_tuple_, dummy_result_tuple_); |
| 479 | analytic_fn_evals_[i]->Init(curr_tuple_); |
| 480 | } |
| 481 | } |
| 482 | |
| 483 | // If the start bound is not UNBOUNDED PRECEDING and there are still rows in the |
| 484 | // partition for which we need to produce result tuples, we need to continue removing |
| 485 | // input tuples at the start of the window from each row that we're adding results for. |
| 486 | VLOG_ROW << id() << " TryAddRemainingResults prev_partition_idx=" << prev_partition_idx |
| 487 | << " " << DebugStateString(true); |
| 488 | for (int64_t next_result_idx = last_result_idx_ + 1; next_result_idx < partition_idx; |
| 489 | ++next_result_idx) { |
| 490 | if (window_tuples_.empty()) break; |
| 491 | if (next_result_idx + rows_start_offset_ > window_tuples_.front().first) { |
| 492 | DCHECK_EQ(next_result_idx + rows_start_offset_ - 1, window_tuples_.front().first); |
| 493 | // For every tuple that is removed from the window: Remove() from the evaluators |
| 494 | // and add the result tuple at the next index. |
| 495 | VLOG_ROW << id() << " Remove window_row_idx=" << window_tuples_.front().first |
| 496 | << " for result row at idx=" << next_result_idx; |
| 497 | TupleRow* remove_row = reinterpret_cast<TupleRow*>(&window_tuples_.front().second); |
| 498 | AggFnEvaluator::Remove(analytic_fn_evals_, remove_row, curr_tuple_); |
| 499 | window_tuples_.pop_front(); |
| 500 | } |
| 501 | RETURN_IF_ERROR(AddResultTuple(last_result_idx_ + 1)); |
| 502 | } |
| 503 | |
| 504 | // If there are still rows between the row with the last result (AddResultTuple() may |
| 505 | // have updated last_result_idx_) and the partition boundary, add the current results |
| 506 | // for the remaining rows with the same result tuple (curr_tuple_ is not modified). |
| 507 | if (last_result_idx_ < partition_idx - 1) { |
| 508 | RETURN_IF_ERROR(AddResultTuple(partition_idx - 1)); |
| 509 | } |
| 510 | return Status::OK(); |
| 511 | } |
| 512 | |
| 513 | inline Status AnalyticEvalNode::InitNextPartition(RuntimeState* state, |
| 514 | int64_t stream_idx) { |