| 105 | |
| 106 | template <typename TSortCursor> |
| 107 | void MergingSortedBlockInputStream::fetchNextBlock(const TSortCursor & current, SortingHeap<TSortCursor> & queue) |
| 108 | { |
| 109 | size_t order = current->order; |
| 110 | size_t size = cursors.size(); |
| 111 | |
| 112 | if (order >= size || &cursors[order] != current.impl) |
| 113 | throw Exception("Logical error in MergingSortedBlockInputStream", ErrorCodes::LOGICAL_ERROR); |
| 114 | |
| 115 | while (true) |
| 116 | { |
| 117 | source_blocks[order] = children[order]->read(); |
| 118 | |
| 119 | if (!source_blocks[order]) |
| 120 | { |
| 121 | queue.removeTop(); |
| 122 | break; |
| 123 | } |
| 124 | |
| 125 | if (source_blocks[order].rows()) |
| 126 | { |
| 127 | cursors[order].reset(source_blocks[order]); |
| 128 | queue.replaceTop(&cursors[order]); |
| 129 | break; |
| 130 | } |
| 131 | } |
| 132 | } |
| 133 | |
| 134 | |
| 135 | template <typename TSortingHeap> |