| 60 | "partitioned top-n operator."); |
| 61 | |
| 62 | Status TopNPlanNode::Init(const TPlanNode& tnode, FragmentState* state) { |
| 63 | const TSortInfo& tsort_info = tnode.sort_node.sort_info; |
| 64 | RETURN_IF_ERROR(PlanNode::Init(tnode, state)); |
| 65 | RETURN_IF_ERROR(ScalarExpr::Create( |
| 66 | tsort_info.ordering_exprs, *row_descriptor_, state, &ordering_exprs_)); |
| 67 | DCHECK(tsort_info.__isset.sort_tuple_slot_exprs); |
| 68 | output_tuple_desc_ = row_descriptor_->tuple_descriptors()[0]; |
| 69 | RETURN_IF_ERROR(ScalarExpr::Create(tsort_info.sort_tuple_slot_exprs, |
| 70 | *children_[0]->row_descriptor_, state, &output_tuple_exprs_)); |
| 71 | ordering_comparator_config_ = |
| 72 | state->obj_pool()->Add(new TupleRowComparatorConfig(tsort_info, ordering_exprs_)); |
| 73 | if (is_partitioned()) { |
| 74 | DCHECK(tnode.sort_node.__isset.partition_exprs); |
| 75 | RETURN_IF_ERROR(ScalarExpr::Create( |
| 76 | tnode.sort_node.partition_exprs, *row_descriptor_, state, &partition_exprs_)); |
| 77 | |
| 78 | // We need a TSortInfo for internal use in the sorted map. Initialize with |
| 79 | // arbitrary parameters. |
| 80 | TSortInfo* tpartition_sort_info = state->obj_pool()->Add(new TSortInfo); |
| 81 | tpartition_sort_info->sorting_order = TSortingOrder::LEXICAL; |
| 82 | tpartition_sort_info->is_asc_order.resize(partition_exprs_.size(), true); |
| 83 | tpartition_sort_info->nulls_first.resize(partition_exprs_.size(), false); |
| 84 | partition_comparator_config_ = state->obj_pool()->Add( |
| 85 | new TupleRowComparatorConfig(*tpartition_sort_info, partition_exprs_)); |
| 86 | |
| 87 | DCHECK(tnode.sort_node.__isset.intra_partition_sort_info); |
| 88 | const TSortInfo& intra_part_sort_info = tnode.sort_node.intra_partition_sort_info; |
| 89 | // Set up the intra-partition comparator. |
| 90 | RETURN_IF_ERROR(ScalarExpr::Create(intra_part_sort_info.ordering_exprs, |
| 91 | *row_descriptor_, state, &intra_partition_ordering_exprs_)); |
| 92 | intra_partition_comparator_config_ = |
| 93 | state->obj_pool()->Add(new TupleRowComparatorConfig( |
| 94 | intra_part_sort_info, intra_partition_ordering_exprs_)); |
| 95 | |
| 96 | // Construct SlotRefs that simply copy the output tuple to itself. |
| 97 | for (const SlotDescriptor* slot_desc : output_tuple_desc_->slots()) { |
| 98 | SlotRef* slot_ref = state->obj_pool()->Add(SlotRef::TypeSafeCreate(slot_desc)); |
| 99 | noop_tuple_exprs_.push_back(slot_ref); |
| 100 | RETURN_IF_ERROR(slot_ref->Init(*row_descriptor_, true, state)); |
| 101 | } |
| 102 | } |
| 103 | DCHECK_EQ(conjuncts_.size(), 0) << "TopNNode should never have predicates to evaluate."; |
| 104 | state->CheckAndAddCodegenDisabledMessage(codegen_status_msgs_); |
| 105 | return Status::OK(); |
| 106 | } |
| 107 | |
| 108 | void TopNPlanNode::Close() { |
| 109 | ScalarExpr::Close(ordering_exprs_); |