| 125 | } |
| 126 | |
| 127 | void TopNNode::InsertBatchPartitioned(RuntimeState* state, RowBatch* batch) { |
| 128 | DCHECK(is_partitioned()); |
| 129 | // Insert all of the rows in the batch into the per-partition heaps. The soft memory |
| 130 | // limit will be checked later, in case this batch put us over the limit. |
| 131 | FOREACH_ROW(batch, 0, iter) { |
| 132 | tmp_tuple_->MaterializeExprs<false, true>( |
| 133 | iter.Get(), *output_tuple_desc_, output_tuple_expr_evals_, nullptr); |
| 134 | // TODO: IMPALA-10228: the comparator won't get inlined by codegen here. |
| 135 | auto it = partition_heaps_.find(tmp_tuple_); |
| 136 | Heap* new_heap = nullptr; |
| 137 | Heap* heap; |
| 138 | if (it == partition_heaps_.end()) { |
| 139 | // Allocate the heap here, but insert in into partition_heaps_ later once we've |
| 140 | // initialized the tuple that will be the key. |
| 141 | new_heap = |
| 142 | new Heap(*intra_partition_order_cmp_, per_partition_limit(), include_ties()); |
| 143 | heap = new_heap; |
| 144 | COUNTER_ADD(in_mem_heap_created_counter_, 1); |
| 145 | } else { |
| 146 | heap = it->second.get(); |
| 147 | } |
| 148 | heap->InsertMaterializedTuple(this, tmp_tuple_); |
| 149 | if (new_heap != nullptr) { |
| 150 | DCHECK_GT(new_heap->num_tuples(), 0); |
| 151 | // Add the new heap with the first tuple as the key. |
| 152 | partition_heaps_.emplace(new_heap->top(), unique_ptr<Heap>(new_heap)); |
| 153 | } |
| 154 | } |
| 155 | } |
| 156 | |
| 157 | void TopNNode::Heap::InsertMaterializedTuple( |
| 158 | TopNNode* node, Tuple* materialized_tuple) { |
nothing calls this directly
no test coverage detected