Method Advance

cpp/src/arrow/acero/sorted_merge_node.cc:169–202

  bool Finished() const { return batches_processed_ == total_batches_; }

  void Advance(SingleRecordBatchSliceBuilder& builder) {
    // Advance the row until a new time is encountered or the record batch
    // ends. This will return a range of {-1, -1} and a nullptr if there is
    // no input
    bool active =
        (latest_ref_row_ > 0 /*short circuit the lock on the queue*/) || !queue_.Empty();

    if (!active) {
      return;
    }

    row_index_t start = latest_ref_row_;
    row_index_t end = latest_ref_row_;
    time_unit_t startTime = GetLatestTime();
    std::shared_ptr<arrow::RecordBatch> batch = queue_.Front();
    auto rows_in_batch = (row_index_t)batch->num_rows();

    while (GetLatestTime() == startTime) {
      end = ++latest_ref_row_;
      if (latest_ref_row_ >= rows_in_batch) {
        // hit the end of the batch, need to get the next batch if
        // possible.
        ++batches_processed_;
        latest_ref_row_ = 0;
        active &= !queue_.TryPop();
        if (active) {
          DCHECK_GT(queue_.Front()->num_rows(),
                    0);  // empty batches disallowed, sanity check
        }
        break;
      }
    }
    builder.AddEntry(batch, start, end);
  }

  arrow::Status Push(const std::shared_ptr<arrow::RecordBatch>& rb) {
    if (rb->num_rows() > 0) {