| 178 | |
| 179 | template <bool HAS_VAR_LEN_SLOTS, bool INITIAL_RUN> |
| 180 | Status Sorter::Run::AddBatchInternal( |
| 181 | RowBatch* batch, int start_index, int* num_processed, bool* allocation_failed) { |
| 182 | DCHECK(!is_finalized_); |
| 183 | DCHECK(!fixed_len_pages_.empty()); |
| 184 | DCHECK_EQ(HAS_VAR_LEN_SLOTS, has_var_len_slots_); |
| 185 | DCHECK_EQ(INITIAL_RUN, initial_run_); |
| 186 | |
| 187 | *num_processed = 0; |
| 188 | Page* cur_fixed_len_page = &fixed_len_pages_.back(); |
| 189 | |
| 190 | if (!INITIAL_RUN) { |
| 191 | // For intermediate merges, the input row is the sort tuple. |
| 192 | DCHECK_EQ(batch->row_desc()->tuple_descriptors().size(), 1); |
| 193 | DCHECK_EQ(batch->row_desc()->tuple_descriptors()[0], sort_tuple_desc_); |
| 194 | } |
| 195 | |
| 196 | /// Keep initial unsorted runs pinned in memory so we can sort them. |
| 197 | const AddPageMode add_mode = INITIAL_RUN ? KEEP_PREV_PINNED : UNPIN_PREV; |
| 198 | |
| 199 | // Input rows are copied/materialized into tuples allocated in fixed_len_pages_. |
| 200 | // The variable length column data are copied into pages stored in var_len_pages_. |
| 201 | // Input row processing is split into two loops. |
| 202 | // The inner loop processes as many input rows as will fit in cur_fixed_len_page. |
| 203 | // The outer loop allocates a new page for fixed-len data if the input batch is |
| 204 | // not exhausted. |
| 205 | |
| 206 | // cur_input_index is the index into the input 'batch' of the current input row being |
| 207 | // processed. |
| 208 | int cur_input_index = start_index; |
| 209 | vector<StringValue*> string_values; |
| 210 | vector<pair<CollectionValue*, int64_t>> coll_values_and_sizes; |
| 211 | string_values.reserve(sort_tuple_desc_->string_slots().size()); |
| 212 | coll_values_and_sizes.reserve(sort_tuple_desc_->collection_slots().size()); |
| 213 | while (cur_input_index < batch->num_rows()) { |
| 214 | // tuples_remaining is the number of tuples to copy/materialize into |
| 215 | // cur_fixed_len_page. |
| 216 | int tuples_remaining = cur_fixed_len_page->BytesRemaining() / sort_tuple_size_; |
| 217 | tuples_remaining = min(batch->num_rows() - cur_input_index, tuples_remaining); |
| 218 | |
| 219 | for (int i = 0; i < tuples_remaining; ++i) { |
| 220 | int total_var_len = 0; |
| 221 | TupleRow* input_row = batch->GetRow(cur_input_index); |
| 222 | Tuple* new_tuple = |
| 223 | reinterpret_cast<Tuple*>(cur_fixed_len_page->AllocateBytes(sort_tuple_size_)); |
| 224 | if (INITIAL_RUN) { |
| 225 | new_tuple->MaterializeExprs<HAS_VAR_LEN_SLOTS, true>(input_row, |
| 226 | *sort_tuple_desc_, sorter_->sort_tuple_expr_evals_, nullptr, |
| 227 | &string_values, &coll_values_and_sizes, &total_var_len); |
| 228 | if (total_var_len > sorter_->page_len_) { |
| 229 | int64_t max_row_size = sorter_->state_->query_options().max_row_size; |
| 230 | return Status(TErrorCode::MAX_ROW_SIZE, |
| 231 | PrettyPrinter::Print(total_var_len, TUnit::BYTES), sorter_->node_label_, |
| 232 | PrettyPrinter::Print(max_row_size, TUnit::BYTES)); |
| 233 | } |
| 234 | } else { |
| 235 | memcpy(new_tuple, input_row->GetTuple(0), sort_tuple_size_); |
| 236 | if (HAS_VAR_LEN_SLOTS) { |
| 237 | CollectNonNullNonSmallVarSlots(new_tuple, *sort_tuple_desc_, &string_values, |
nothing calls this directly
no test coverage detected