| 350 | } |
| 351 | |
| 352 | arrow::Status InputReceived(arrow::acero::ExecNode* input, |
| 353 | arrow::ExecBatch batch) override { |
| 354 | ARROW_DCHECK(std_has(inputs_, input)); |
| 355 | const size_t index = std_find(inputs_, input) - inputs_.begin(); |
| 356 | ARROW_ASSIGN_OR_RAISE(std::shared_ptr<RecordBatch> rb, |
| 357 | batch.ToRecordBatch(output_schema_)); |
| 358 | |
| 359 | // Push into the queue. Note that we don't need to lock since |
| 360 | // InputState's ConcurrentQueue manages locking |
| 361 | input_counter[index] += rb->num_rows(); |
| 362 | ARROW_RETURN_NOT_OK(state[index]->Push(rb)); |
| 363 | PushTask(kNewTask); |
| 364 | return Status::OK(); |
| 365 | } |
| 366 | |
| 367 | void PushTask(bool ok) { |
| 368 | #ifdef ARROW_ENABLE_THREADING |
no test coverage detected