Method Process

cpp/src/arrow/acero/fetch_node.cc:161–188

  }

  Result<std::optional<util::SequencingQueue::Task>> Process(ExecBatch batch) override {
    if (finished_) {
      return std::nullopt;
    }
    FetchCounter::Page page = fetch_counter_.NextPage(batch);
    std::optional<util::SequencingQueue::Task> task_or_none;
    if (page.to_send > 0) {
      int new_index = out_batch_count_++;
      task_or_none = [this, to_send = page.to_send, to_skip = page.to_skip, new_index,
                      batch = std::move(batch)]() mutable {
        ExecBatch batch_to_send = std::move(batch);
        if (to_skip > 0 || to_send < batch_to_send.length) {
          batch_to_send = batch_to_send.Slice(to_skip, to_send);
        }
        batch_to_send.index = new_index;
        return output_->InputReceived(this, std::move(batch_to_send));
      };
    }
    // In the in_batch_counter_ case we've run out of data to process (count_ was
    // greater than the total # of non-skipped rows)  In the page.ended case we've
    // just hit our desired output count
    if (in_batch_counter_.Increment() || (page.ended && !finished_)) {
      finished_ = true;
      ARROW_RETURN_NOT_OK(inputs_[0]->StopProducing());
      ARROW_RETURN_NOT_OK(output_->InputFinished(this, out_batch_count_));
    }
    return task_or_none;
  }

  void Schedule(util::SequencingQueue::Task task) override {
    plan_->query_context()->ScheduleTask(std::move(task), "FetchNode::ProcessBatch");