MCPcopy Create free account
hub / github.com/bytedance/bolt / addInput

Method addInput

bolt/exec/SortBuffer.cpp:145–210  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

143}
144
145void 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,

Callers

nothing calls this directly

Calls 11

wrapColumnsFunction · 0.85
getNumBatchesMethod · 0.80
getNumRowsMethod · 0.80
storeSingleRowIdMethod · 0.80
childAtMethod · 0.80
poolFunction · 0.50
sizeMethod · 0.45
newRowMethod · 0.45
kindMethod · 0.45
typeMethod · 0.45
addPayloadMethod · 0.45

Tested by

no test coverage detected