Method SliceAndDeliverMorsel

cpp/src/arrow/acero/source_node.cc:122–164

  [[noreturn]] Status InputFinished(ExecNode*, int) override { NoInputs(); }

  void SliceAndDeliverMorsel(const ExecBatch& morsel) {
    bool use_legacy_batching = plan_->query_context()->options().use_legacy_batching;
    int64_t morsel_length = static_cast<int64_t>(morsel.length);
    int initial_batch_index = batch_count_;
    if (use_legacy_batching || morsel_length == 0) {
      // For various reasons (e.g. ARROW-13982) we pass empty batches
      // through
      batch_count_++;
    } else {
      int num_batches =
          static_cast<int>(bit_util::CeilDiv(morsel_length, ExecPlan::kMaxBatchSize));
      batch_count_ += num_batches;
    }
    plan_->query_context()->ScheduleTask(
        [this, morsel_length, use_legacy_batching, initial_batch_index, morsel,
         has_ordering = !ordering_.is_unordered()]() {
          int64_t offset = 0;
          int batch_index = initial_batch_index;
          do {
            int64_t batch_size =
                std::min<int64_t>(morsel_length - offset, ExecPlan::kMaxBatchSize);
            // In order for the legacy batching model to work we must
            // not slice batches from the source
            if (use_legacy_batching) {
              batch_size = morsel_length;
            }
            ExecBatch batch = morsel.Slice(offset, batch_size);
            UnalignedBufferHandling unaligned_buffer_handling =
                plan_->query_context()->options().unaligned_buffer_handling.value_or(
                    GetDefaultUnalignedBufferHandling());
            ARROW_RETURN_NOT_OK(
                HandleUnalignedBuffers(&batch, unaligned_buffer_handling));
            if (has_ordering) {
              batch.index = batch_index;
            }
            offset += batch_size;
            batch_index++;
            ARROW_RETURN_NOT_OK(output_->InputReceived(this, std::move(batch)));
          } while (offset < morsel.length);
          return Status::OK();
        },
        "SourceNode::ProcessMorsel");
  }

  Status StartProducing() override {
    NoteStartProducing(ToStringExtra());