| 94 | } |
| 95 | |
| 96 | void StartProducing() { |
| 97 | if (finished_.is_finished()) { |
| 98 | finished_ = Future<>::MakeFinished( |
| 99 | Status::Invalid("StartProducing called after plan had already finished")); |
| 100 | return; |
| 101 | } |
| 102 | if (started_) { |
| 103 | finished_.MarkFinished( |
| 104 | Status::Invalid("StartProducing called on a plan that had already started.")); |
| 105 | return; |
| 106 | } |
| 107 | if (query_context_.exec_context()->executor() == nullptr) { |
| 108 | finished_.MarkFinished(Status::Invalid( |
| 109 | "An exec plan must have an executor for CPU tasks. To run without threads use " |
| 110 | "a SerialExecutor (the arrow::compute::DeclarationTo... methods should take " |
| 111 | "care of this for you and are an easier way to execute an ExecPlan.)")); |
| 112 | return; |
| 113 | } |
| 114 | if (query_context_.io_context()->executor() == nullptr) { |
| 115 | finished_.MarkFinished( |
| 116 | Status::Invalid("An exec plan must have an I/O executor for I/O tasks.")); |
| 117 | return; |
| 118 | } |
| 119 | |
| 120 | started_ = true; |
| 121 | |
| 122 | // We call StartProducing on each of the nodes. The source nodes should generally |
| 123 | // start scheduling some tasks during this call. |
| 124 | // |
| 125 | // If no source node schedules any tasks (e.g. they do all their word synchronously as |
| 126 | // part of StartProducing) then the plan may be finished before we return from this |
| 127 | // call. |
| 128 | auto scope = START_SCOPED_SPAN(span_, "ExecPlan", {{"plan", ToString()}}); |
| 129 | Future<> scheduler_finished = arrow::util::AsyncTaskScheduler::Make( |
| 130 | [this](arrow::util::AsyncTaskScheduler* async_scheduler) { |
| 131 | QueryContext* ctx = query_context(); |
| 132 | RETURN_NOT_OK(ctx->Init(async_scheduler)); |
| 133 | |
| 134 | #ifdef ARROW_WITH_OPENTELEMETRY |
| 135 | if (HasMetadata()) { |
| 136 | auto pairs = metadata().get()->sorted_pairs(); |
| 137 | opentelemetry::nostd::shared_ptr<opentelemetry::trace::Span> span = |
| 138 | ::arrow::internal::tracing::UnwrapSpan(span_.details.get()); |
| 139 | std::for_each(std::begin(pairs), std::end(pairs), |
| 140 | [span](const std::pair<std::string, std::string>& pair) { |
| 141 | span->SetAttribute(pair.first, pair.second); |
| 142 | }); |
| 143 | } |
| 144 | #endif |
| 145 | for (auto& n : nodes_) { |
| 146 | RETURN_NOT_OK(n->Init()); |
| 147 | } |
| 148 | |
| 149 | ctx->scheduler()->RegisterEnd(); |
| 150 | int num_threads = 1; |
| 151 | bool sync_execution = true; |
| 152 | if (auto executor = query_context()->exec_context()->executor()) { |
| 153 | num_threads = executor->GetCapacity(); |
nothing calls this directly
no test coverage detected