Method InsertTupleWithTieHandling

be/src/exec/topn-node-ir.cc:75–125

}

int TopNNode::Heap::InsertTupleWithTieHandling(
    const TupleRowComparator& cmp, TopNNode* node, Tuple* materialized_tuple) {
  DCHECK(include_ties());
  DCHECK_EQ(capacity_, priority_queue_.Size())
        << "Ties only need special handling when heap is at capacity";
  const TupleDescriptor& tuple_desc = *node->output_tuple_desc_;
  Tuple* top_tuple = priority_queue_.Top();
  // If we need to retain ties with the current head, the logic is more complex - we
  // have a logical heap in indices [0, heap_capacity()) of 'priority_queue_' plus
  // some number of tuples in 'overflowed_ties_' that are equal to priority_queue_.Top()
  // according to 'cmp'.
  int cmp_result = cmp.Compare(materialized_tuple, top_tuple);
  if (cmp_result == 0) {
    // This is a duplicate of the current min, we need to include it as a tie with min.
    Tuple* insert_tuple =
        reinterpret_cast<Tuple*>(node->tuple_pool_->Allocate(node->tuple_byte_size()));
    materialized_tuple->DeepCopy(insert_tuple, tuple_desc, node->tuple_pool_.get());
    overflowed_ties_.push_back(insert_tuple);
    return 0;
  } else if (cmp_result > 0) {
    // Tuple does not belong in the heap.
    return 1;
  } else {
    // 'materialized_tuple' needs to be added. Figure out which other tuples, if any,
    // need to be removed from the heap.
    DCHECK_LT(cmp_result, 0);
    // Pop off the head.
    priority_queue_.Pop();

    // Check if 'top_tuple' (the tuple we just popped off) is tied with the new head.
    if (heap_capacity() > 1 && cmp.Compare(top_tuple, priority_queue_.Top()) == 0) {
      // The new top is still tied with the tuples in 'overflowed_ties_' so we must keep
      // it. The previous top becomes another overflowed tuple.
      overflowed_ties_.push_back(top_tuple);
      Tuple* insert_tuple =
          reinterpret_cast<Tuple*>(node->tuple_pool_->Allocate(node->tuple_byte_size()));
      materialized_tuple->DeepCopy(insert_tuple, tuple_desc, node->tuple_pool_.get());
      // Push the new tuple onto the heap, but retain the tied tuple.
      priority_queue_.Push(insert_tuple);
      return 0;
    } else {
      // No tuples tied with 'top_tuple' are left.
      // Reuse the fixed-length memory of 'top_tuple' to reduce allocations.
      int64_t num_rows_replaced = overflowed_ties_.size() + 1;
      overflowed_ties_.clear();
      materialized_tuple->DeepCopy(top_tuple, tuple_desc, node->tuple_pool_.get());
      priority_queue_.Push(top_tuple);
      return num_rows_replaced;
    }
  }
}

void TopNNode::InsertBatchPartitioned(RuntimeState* state, RowBatch* batch) {
  DCHECK(is_partitioned());