| 1129 | } |
| 1130 | |
| 1131 | Status Sorter::AddBatch(RowBatch* batch) { |
| 1132 | DCHECK(unsorted_run_ != nullptr); |
| 1133 | DCHECK(batch != nullptr); |
| 1134 | DCHECK(enable_spilling_); |
| 1135 | int num_processed = 0; |
| 1136 | int cur_batch_index = 0; |
| 1137 | while (cur_batch_index < batch->num_rows()) { |
| 1138 | RETURN_IF_ERROR(AddBatchNoSpill(batch, cur_batch_index, &num_processed)); |
| 1139 | |
| 1140 | cur_batch_index += num_processed; |
| 1141 | |
| 1142 | if (MustSortAndSpill(cur_batch_index, batch->num_rows())) { |
| 1143 | RETURN_IF_ERROR(state_->StartSpilling(mem_tracker_)); |
| 1144 | int64_t unsorted_run_bytes = unsorted_run_->TotalBytes(); |
| 1145 | |
| 1146 | if (inmem_run_max_pages_ == 0) { |
| 1147 | // The current run is full. Sort it and spill it. |
| 1148 | RETURN_IF_ERROR(SortCurrentInputRun()); |
| 1149 | sorted_runs_.push_back(unsorted_run_); |
| 1150 | unsorted_run_ = nullptr; |
| 1151 | RETURN_IF_ERROR(sorted_runs_.back()->UnpinAllPages()); |
| 1152 | } else { |
| 1153 | // The memory is full with miniruns. Sort, merge and spill them. |
| 1154 | RETURN_IF_ERROR(MergeAndSpill()); |
| 1155 | } |
| 1156 | |
| 1157 | // After we freed memory by spilling, initialize the next run. |
| 1158 | unsorted_run_ = |
| 1159 | run_pool_.Add(new Run(this, output_row_desc_->tuple_descriptors()[0], true)); |
| 1160 | RETURN_IF_ERROR(unsorted_run_->Init()); |
| 1161 | |
| 1162 | bool prev_enforcement = enforce_sort_run_bytes_limit_; |
| 1163 | CheckSortRunBytesLimitEnforcement(); |
| 1164 | if (!prev_enforcement && enforce_sort_run_bytes_limit_) { |
| 1165 | VLOG(3) << Substitute( |
| 1166 | "Enforcing sort_run_bytes_limit because reservation limit exceeded: " |
| 1167 | "prev_run_total_bytes=$0 sort_run_bytes_limit=$1", |
| 1168 | PrettyPrinter::PrintBytes(unsorted_run_bytes), |
| 1169 | PrettyPrinter::PrintBytes(GetSortRunBytesLimit())); |
| 1170 | } |
| 1171 | } |
| 1172 | } |
| 1173 | if (enforce_sort_run_bytes_limit_) TryLowerMemUpToSortRunBytesLimit(); |
| 1174 | // Clear any temporary allocations made while materializing the sort tuples. |
| 1175 | expr_results_pool_.Clear(); |
| 1176 | return Status::OK(); |
| 1177 | } |
| 1178 | |
| 1179 | void Sorter::CheckSortRunBytesLimitEnforcement() { |
| 1180 | enforce_sort_run_bytes_limit_ = GetSortRunBytesLimit() > 0 |
no test coverage detected