| 128 | } |
| 129 | |
| 130 | void Partitioner::executeInternal(ExecutionContext* context) { |
| 131 | const auto relOffsetVector = resultSet->getValueVector(info.relOffsetDataPos); |
| 132 | while (children[0]->getNextTuple(context)) { |
| 133 | DASSERT(dataInfo.columnEvaluators.size() >= 1); |
| 134 | const auto numRels = relOffsetVector->state->getSelVector().getSelSize(); |
| 135 | evaluateExpressions(numRels); |
| 136 | auto currentRelOffset = sharedState->relTable->reserveRelOffsets(numRels); |
| 137 | for (auto i = 0u; i < numRels; i++) { |
| 138 | const auto pos = relOffsetVector->state->getSelVector()[i]; |
| 139 | relOffsetVector->setValue<offset_t>(pos, currentRelOffset++); |
| 140 | } |
| 141 | for (auto partitioningIdx = 0u; partitioningIdx < info.infos.size(); partitioningIdx++) { |
| 142 | auto& partitionInfo = info.infos[partitioningIdx]; |
| 143 | auto keyVector = dataInfo.columnEvaluators[partitionInfo.keyIdx]->resultVector; |
| 144 | partitionIdxes->state = keyVector->state; |
| 145 | partitionInfo.partitionerFunc(keyVector.get(), partitionIdxes.get()); |
| 146 | auto chunkToCopyFrom = constructDataChunk(keyVector->state); |
| 147 | copyDataToPartitions(*MemoryManager::Get(*context->clientContext), partitioningIdx, |
| 148 | chunkToCopyFrom); |
| 149 | } |
| 150 | } |
| 151 | sharedState->merge(localState->partitioningBuffers); |
| 152 | } |
| 153 | |
| 154 | void Partitioner::evaluateExpressions(uint64_t numRels) const { |
| 155 | for (auto i = 0u; i < dataInfo.evaluateTypes.size(); ++i) { |
nothing calls this directly
no test coverage detected