Method InsertMaterializedTuple

be/src/exec/topn-node-ir.cc:157–188

}

void TopNNode::Heap::InsertMaterializedTuple(
    TopNNode* node, Tuple* materialized_tuple) {
  DCHECK(node->is_partitioned());
  const TupleDescriptor& tuple_desc = *node->output_tuple_desc_;
  Tuple* insert_tuple = nullptr;
  if (priority_queue_.Size() < heap_capacity()) {
    // Add all tuples until we hit capacity.
    insert_tuple =
        reinterpret_cast<Tuple*>(node->tuple_pool_->Allocate(node->tuple_byte_size()));
    materialized_tuple->DeepCopy(insert_tuple, tuple_desc, node->tuple_pool_.get());
    priority_queue_.Push(insert_tuple);
    return;
  }

  // We're at capacity - compare to the first row in the priority queue to see if
  // we need to insert this row into the heap.
  DCHECK(!priority_queue_.Empty());
  Tuple* top_tuple = priority_queue_.Top();
  if (!include_ties()) {
    ++num_tuples_discarded_; // One of the tuples will be discarded.
    int cmp_result =
        node->intra_partition_order_cmp_->Compare(materialized_tuple, top_tuple);
    if (cmp_result >= 0) return;
    // Pop off the old head, and replace with the new tuple. Reuse the fixed-length
    // memory of 'top_tuple' to reduce allocations.
    materialized_tuple->DeepCopy(top_tuple, tuple_desc, node->tuple_pool_.get());
    priority_queue_.HeapifyFromTop();
    return;
  }
  num_tuples_discarded_ += InsertTupleWithTieHandling(
      *node->intra_partition_order_cmp_, node, materialized_tuple);
}