Method NextLineItemBatch

cpp/src/arrow/acero/tpch_node.cc:1388–1465

  }

  Result<std::optional<ExecBatch>> NextLineItemBatch(size_t thread_index) {
    ThreadLocalData& tld = thread_local_data_[thread_index];
    ExecBatch queued;
    bool from_queue = false;
    {
      std::lock_guard<std::mutex> lock(lineitem_output_queue_mutex_);
      if (!lineitem_output_queue_.empty()) {
        queued = std::move(lineitem_output_queue_.front());
        lineitem_output_queue_.pop();
        from_queue = true;
      }
    }
    tld.first_batch_offset = 0;
    if (from_queue) {
      ARROW_DCHECK(queued.length <= batch_size_);
      tld.first_batch_offset = queued.length;
      if (queued.length == batch_size_) return queued;
    }
    {
      std::lock_guard<std::mutex> lock(orders_output_queue_mutex_);
      if (orders_rows_generated_ == orders_rows_to_generate_) {
        if (from_queue) return queued;
        return std::nullopt;
      }

      tld.orderkey_start = orders_rows_generated_;
      tld.orders_to_generate =
          std::min(batch_size_, orders_rows_to_generate_ - orders_rows_generated_);
      orders_rows_generated_ += tld.orders_to_generate;
      orders_batches_generated_.fetch_add(1);
      RETURN_NOT_OK(GenerateRowCounts(thread_index));
      lineitem_batches_generated_.fetch_add(
          static_cast<int64_t>(tld.lineitem.size() - from_queue));
      ARROW_DCHECK(orders_rows_generated_ <= orders_rows_to_generate_);
    }
    tld.orders.resize(ORDERS::kNumCols);
    std::fill(tld.orders.begin(), tld.orders.end(), Datum());
    tld.generated_lineitem.reset();
    if (from_queue) {
      for (size_t i = 0; i < lineitem_cols_.size(); i++)
        if (tld.lineitem[0][lineitem_cols_[i]].kind() == Datum::NONE)
          tld.lineitem[0][lineitem_cols_[i]] = std::move(queued[i]);
    }

    for (int col : orders_cols_) RETURN_NOT_OK(kOrdersGenerators[col](thread_index));
    for (int col : lineitem_cols_) RETURN_NOT_OK(kLineitemGenerators[col](thread_index));

    if (!orders_cols_.empty()) {
      std::vector<Datum> orders_result(orders_cols_.size());
      for (size_t i = 0; i < orders_cols_.size(); i++) {
        int col_idx = orders_cols_[i];
        orders_result[i] = tld.orders[col_idx];
      }
      ARROW_ASSIGN_OR_RAISE(ExecBatch orders_batch,
                            ExecBatch::Make(std::move(orders_result)));
      {
        std::lock_guard<std::mutex> lock(orders_output_queue_mutex_);
        orders_output_queue_.emplace(std::move(orders_batch));