| 380 | } |
| 381 | |
| 382 | arrow::Status InputFinished(arrow::acero::ExecNode* input, int total_batches) override { |
| 383 | ARROW_DCHECK(std_has(inputs_, input)); |
| 384 | { |
| 385 | std::lock_guard<std::mutex> guard(gate); |
| 386 | ARROW_DCHECK(std_has(inputs_, input)); |
| 387 | size_t k = std_find(inputs_, input) - inputs_.begin(); |
| 388 | state.at(k)->set_total_batches(total_batches); |
| 389 | } |
| 390 | // Trigger a final process call for stragglers |
| 391 | PushTask(kNewTask); |
| 392 | |
| 393 | return Status::OK(); |
| 394 | } |
| 395 | |
| 396 | arrow::Status StartProducing() override { |
| 397 | ARROW_ASSIGN_OR_RAISE(process_task, plan_->query_context()->BeginExternalTask( |
no test coverage detected