| 384 | } |
| 385 | |
| 386 | LocalPartition::LocalPartition( |
| 387 | int32_t operatorId, |
| 388 | DriverCtx* ctx, |
| 389 | const std::shared_ptr<const core::LocalPartitionNode>& planNode) |
| 390 | : Operator( |
| 391 | ctx, |
| 392 | planNode->outputType(), |
| 393 | operatorId, |
| 394 | planNode->id(), |
| 395 | "LocalPartition"), |
| 396 | queues_{ |
| 397 | ctx->task->getLocalExchangeQueues(ctx->splitGroupId, planNode->id())}, |
| 398 | numPartitions_{queues_.size()}, |
| 399 | partitionFunction_( |
| 400 | numPartitions_ == 1 |
| 401 | ? nullptr |
| 402 | : planNode->partitionFunctionSpec().create(numPartitions_)) { |
| 403 | BOLT_CHECK(numPartitions_ == 1 || partitionFunction_ != nullptr); |
| 404 | |
| 405 | for (auto& queue : queues_) { |
| 406 | queue->addProducer(); |
| 407 | } |
| 408 | } |
| 409 | |
| 410 | LocalPartition::LocalPartition( |
| 411 | int32_t operatorId, |
nothing calls this directly
no test coverage detected