| 143 | } |
| 144 | |
| 145 | void SortBuffer::addInput(const VectorPtr& input) { |
| 146 | BOLT_CHECK(!noMoreInput_); |
| 147 | ensureInputFits(input); |
| 148 | |
| 149 | SelectivityVector allRows(input->size()); |
| 150 | std::vector<char*> rows(input->size()); |
| 151 | for (int row = 0; row < input->size(); ++row) { |
| 152 | rows[row] = data_->newRow(); |
| 153 | } |
| 154 | auto* inputRow = input->as<RowVector>(); |
| 155 | MicrosecondTimer timer(&sortColToRowTimeUs_); |
| 156 | if (hybridSortEnabled_) { |
| 157 | // Get batch/row info before processing |
| 158 | auto batchId = hybridData_->getNumBatches(); // for scattered mode |
| 159 | auto baseRow = hybridData_->getNumRows(); // for coalesced mode |
| 160 | for (int row = 0; row < input->size(); ++row) { |
| 161 | // Store RowId - encoding depends on mode |
| 162 | uint64_t encodedId; |
| 163 | if (scatteredMode_) { |
| 164 | // Scattered mode: rowId = (batchId << 32) | rowInBatch |
| 165 | // driverId stored in top 8 bits (always 0 for Sort) |
| 166 | BOLT_CHECK_LT(batchId, (1u << 24)); |
| 167 | encodedId = (static_cast<uint64_t>(0) << 56) | |
| 168 | ((static_cast<uint64_t>(batchId) << 32) | row); |
| 169 | } else { |
| 170 | // Coalesced mode: rowId = global row index |
| 171 | encodedId = (static_cast<uint64_t>(0) << 56) | |
| 172 | (static_cast<uint64_t>(row + baseRow) & ((1ULL << 56) - 1)); |
| 173 | } |
| 174 | data_->storeSingleRowId(encodedId, rows[row]); |
| 175 | } |
| 176 | // Store key columns |
| 177 | for (const auto& columnProjection : keyColumnMap_) { |
| 178 | DecodedVector decoded( |
| 179 | *inputRow->childAt(columnProjection.outputChannel), allRows); |
| 180 | auto kind = |
| 181 | inputRow->childAt(columnProjection.outputChannel)->type()->kind(); |
| 182 | BOLT_DYNAMIC_TYPE_DISPATCH( |
| 183 | data_->storeColumn, |
| 184 | kind, |
| 185 | decoded, |
| 186 | input->size(), |
| 187 | rows, |
| 188 | columnProjection.inputChannel); |
| 189 | } |
| 190 | auto payloadInput = wrapColumns( |
| 191 | input->as<RowVector>(), payloadChannels_, payloadTypes_, pool()); |
| 192 | hybridData_->addPayload(std::move(payloadInput)); |
| 193 | } else { |
| 194 | for (const auto& columnProjection : columnMap_) { |
| 195 | DecodedVector decoded( |
| 196 | *inputRow->childAt(columnProjection.outputChannel), allRows); |
| 197 | auto kind = |
| 198 | inputRow->childAt(columnProjection.outputChannel)->type()->kind(); |
| 199 | BOLT_DYNAMIC_TYPE_DISPATCH( |
| 200 | data_->storeColumn, |
| 201 | kind, |
| 202 | decoded, |
nothing calls this directly
no test coverage detected