| 139 | } |
| 140 | |
| 141 | static Record SortConsume(OpBase *opBase) { |
| 142 | OpSort *op = (OpSort *)opBase; |
| 143 | |
| 144 | if(!op->first) { |
| 145 | return _handoff(op); |
| 146 | } |
| 147 | // make sure consume will not be called on children again, as their depleted |
| 148 | op->first = false; |
| 149 | |
| 150 | Record r; |
| 151 | // if we're here, we don't have any records to return |
| 152 | // try to get records |
| 153 | OpBase *child = op->op.children[0]; |
| 154 | bool newData = false; |
| 155 | while((r = OpBase_Consume(child))) { |
| 156 | _accumulate(op, r); |
| 157 | newData = true; |
| 158 | } |
| 159 | if(!newData) return NULL; |
| 160 | |
| 161 | if(op->buffer) { |
| 162 | sort_r(op->buffer, array_len(op->buffer), sizeof(Record), |
| 163 | (heap_cmp)_buffer_elem_cmp, op); |
| 164 | } else { |
| 165 | // heap |
| 166 | int records_count = Heap_count(op->heap); |
| 167 | op->buffer = array_newlen(Record, records_count); |
| 168 | for(int i = records_count-1; i >= 0 ; i--) { |
| 169 | op->buffer[i] = Heap_poll(op->heap); |
| 170 | } |
| 171 | } |
| 172 | |
| 173 | // pass ordered records downward |
| 174 | return _handoff(op); |
| 175 | } |
| 176 | |
| 177 | // restart iterator |
| 178 | static OpResult SortReset(OpBase *ctx) { |
nothing calls this directly
no test coverage detected