You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2021/01/23 10:15:42 UTC
[impala] 03/04: IMPALA-10296: Fix analytic limit pushdown when
predicates are present
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 1ada739e81416a6f49f2e8bb287a007252371444
Author: Tim Armstrong <ta...@cloudera.com>
AuthorDate: Tue Jan 12 11:21:12 2021 -0800
IMPALA-10296: Fix analytic limit pushdown when predicates are present
This fixes the analytic push down optimization for the case where
the ORDER BY expressions are compatible with the partitioning of the
analytic *and* there is a rank() or row_number() predicate.
In this case the rows returned are going to come from the first partitions,
i.e. if the limit is 100, if we go through the partitions in order until
the row count adds up to 100, then we know that the rows must come from
those partitions.
The problem is that predicates can discard rows from the partitions,
meaning that a limit naively pushed down to the top-n will filter
out rows that could be returned from the query.
We can avoid the problem in the case where the partition limit >=
order by limit, however.
In this case the relevant set of partitions is the set of partitions
that include the first <limit> rows, since the top-level limit
generally kicks in before the per-partition limit. The only twist
is that the orderings may be different within a partition, so we
need to make sure to include all of the rows in the final partition.
The solution implemented in this patch is to increase the pushed
down limit so that it is always guaranteed to include all of the
rows in the final partition to be returned. E.g. if you had a
row_number() <= 100 predicate and limit 100, if you pushed down
limit 200, then you'd be guaranteed to capture all of the rows
in the final partition. One case we need to handle is that,
in the case of a rank() predicate, we can have more than that
number of rows in the partition because of ties.
This patch implements tie handling in the backend (I took most
of that implementation from my in-progress partitioned top-n patch,
with the intention of rebasing that onto this patch).
This also adds a check against TOPN_BYTES_LIMIT so that
the limit can't be increased to an arbitarily large value.
Testing:
* Add new planner test with negative case where it's rejected
because the transformation is incorrect.
* Update other planner tests to reflect new limit calculation
+ tie handling required for correctness.
* Add planner test for very high rank predicate that overflows int32
* Add planner test that checks TOPN_BYTES_LIMIT handling
* Add planner test that checks that dense_rank() can't be pushed.
* Existing planner tests already have adequate coverage for predicates
: <=, <, = and row_number().
* Add some end-to-end tests that repro bugs that fall under the jira
* Add an end-to-end test on TPC-H with more data to exercise the
tie-handling logic in the execnode more.
Perf:
Ran TPC-DS q67 with mt_dop=1 on a single node, confirmed there was
no measurable change in performance as a result of this patchset.
Ran TPC-H scale 30 on a single node, no significant perf change.
Ran a targeted query to check for regressions in the top-n node.
The elapsed time for this targeted query did not change:
use tpch30_parquet;
set mt_dop=1;
select l_extendedprice from lineitem
order by 1 limit 100
Change-Id: I801d7799b0d649c73d2dd1703729a9b58a662509
Reviewed-on: http://gerrit.cloudera.org:8080/16942
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/topn-node-ir.cc | 94 +++-
be/src/exec/topn-node.cc | 143 +++---
be/src/exec/topn-node.h | 141 ++++--
be/src/util/tuple-row-compare.h | 6 +
common/thrift/PlanNodes.thrift | 6 +
.../org/apache/impala/analysis/AnalyticExpr.java | 10 +-
.../apache/impala/planner/AnalyticEvalNode.java | 180 +++++---
.../apache/impala/planner/DistributedPlanner.java | 69 +--
.../apache/impala/planner/SingleNodePlanner.java | 48 +-
.../java/org/apache/impala/planner/SortNode.java | 101 ++++-
.../PlannerTest/limit-pushdown-analytic.test | 493 +++++++++++++++++----
.../queries/PlannerTest/resource-requirements.test | 33 +-
.../queries/PlannerTest/tpcds/tpcds-q67.test | 43 +-
.../queries/limit-pushdown-analytic.test | 96 ++++
.../tpch/queries/limit-pushdown-analytic.test | 149 +++++++
tests/query_test/test_limit_pushdown_analytic.py | 24 +-
16 files changed, 1278 insertions(+), 358 deletions(-)
diff --git a/be/src/exec/topn-node-ir.cc b/be/src/exec/topn-node-ir.cc
index 74ec297..d64b2f0 100644
--- a/be/src/exec/topn-node-ir.cc
+++ b/be/src/exec/topn-node-ir.cc
@@ -25,39 +25,97 @@ void TopNNode::InsertBatch(RowBatch* batch) {
// are hoisted out of the loop as possible (either via code changes or __restrict__)
// annotations.
FOREACH_ROW(batch, 0, iter) {
- bool replaced_existing_row = heap_->InsertTupleRow(this, iter.Get());
- if (replaced_existing_row) ++rows_to_reclaim_;
+ int num_to_reclaim = heap_->InsertTupleRow(this, iter.Get());
+ // Using a branch here instead of adding the value directly improves
+ // top-n performance by a couple of %.
+ if (num_to_reclaim != 0) rows_to_reclaim_ += num_to_reclaim;
}
}
-bool TopNNode::Heap::InsertTupleRow(TopNNode* node, TupleRow* input_row) {
+int TopNNode::Heap::InsertTupleRow(TopNNode* node, TupleRow* input_row) {
const TupleDescriptor& tuple_desc = *node->output_tuple_desc_;
- bool replaced_existing_row = false;
- Tuple* insert_tuple = nullptr;
if (priority_queue_.Size() < heap_capacity()) {
- // Add all tuples until we hit capacity.
- insert_tuple = reinterpret_cast<Tuple*>(
+ // Add all tuples (including any ties) until we hit capacity.
+ Tuple* insert_tuple = reinterpret_cast<Tuple*>(
node->tuple_pool_->Allocate(node->tuple_byte_size()));
insert_tuple->MaterializeExprs<false, false>(input_row, tuple_desc,
node->output_tuple_expr_evals_, node->tuple_pool_.get());
priority_queue_.Push(insert_tuple);
+ return 0;
+ }
+
+ // We're at capacity - compare to the first row in the priority queue to see if
+ // we need to insert this row into the queue.
+ DCHECK(!priority_queue_.Empty());
+ Tuple* top_tuple = priority_queue_.Top();
+ node->tmp_tuple_->MaterializeExprs<false, true>(input_row, tuple_desc,
+ node->output_tuple_expr_evals_, nullptr);
+ if (include_ties()) {
+ return InsertTupleWithTieHandling(node, node->tmp_tuple_);
} else {
- // We're at capacity - compare to the first row in the priority queue to see if
- // we need to insert this row into the queue.
- DCHECK(!priority_queue_.Empty());
- Tuple* top_tuple = priority_queue_.Top();
- node->tmp_tuple_->MaterializeExprs<false, true>(input_row, tuple_desc,
- node->output_tuple_expr_evals_, nullptr);
- // If the new tuple is less than the top, need to swap it with the top and re-heapify.
if (node->tuple_row_less_than_->Less(node->tmp_tuple_, top_tuple)) {
- // Deep copy the new tuple into 'top_tuple' to reuse the fixed-length memory
- // of 'top_tuple'. The top element in the heap points at 'top_tuple'.
+ // Pop off the old head, and replace with the new tuple. Deep copy into 'top_tuple'
+ // to reuse the fixed-length memory of 'top_tuple'.
node->tmp_tuple_->DeepCopy(top_tuple, tuple_desc, node->tuple_pool_.get());
// Re-heapify from the top element and down.
priority_queue_.HeapifyFromTop();
- replaced_existing_row = true;
+ return 1;
+ }
+ return 0;
+ }
+}
+
+int TopNNode::Heap::InsertTupleWithTieHandling(
+ 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 'intra_partition_tuple_row_less_than_'.
+ int cmp_result = node->tuple_row_less_than_->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 &&
+ node->tuple_row_less_than_->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;
}
}
- return replaced_existing_row;
}
+
diff --git a/be/src/exec/topn-node.cc b/be/src/exec/topn-node.cc
index fb69621..1ea16e8 100644
--- a/be/src/exec/topn-node.cc
+++ b/be/src/exec/topn-node.cc
@@ -111,11 +111,11 @@ void TopNPlanNode::Codegen(FragmentState* state) {
codegen->GetFunction(IRFunction::TOPN_NODE_INSERT_BATCH, true);
DCHECK(insert_batch_fn != NULL);
- // Generate two MaterializeExprs() functions, one using tuple_pool_ and
- // one with no pool.
+ // Generate two MaterializeExprs() functions, one with no pool that
+ // does a shallow copy and one with 'tuple_pool_' that does a deep copy of the data.
DCHECK(output_tuple_desc_ != NULL);
- llvm::Function* materialize_exprs_tuple_pool_fn;
- llvm::Function* materialize_exprs_no_pool_fn;
+ llvm::Function* materialize_exprs_tuple_pool_fn = nullptr;
+ llvm::Function* materialize_exprs_no_pool_fn = nullptr;
codegen_status = Tuple::CodegenMaterializeExprs(codegen, false,
*output_tuple_desc_, output_tuple_exprs_,
@@ -125,37 +125,45 @@ void TopNPlanNode::Codegen(FragmentState* state) {
codegen_status = Tuple::CodegenMaterializeExprs(codegen, false,
*output_tuple_desc_, output_tuple_exprs_,
false, &materialize_exprs_no_pool_fn);
+ }
- if (codegen_status.ok()) {
- int replaced = codegen->ReplaceCallSites(insert_batch_fn,
- materialize_exprs_tuple_pool_fn, Tuple::MATERIALIZE_EXPRS_SYMBOL);
- DCHECK_REPLACE_COUNT(replaced, 1) << LlvmCodeGen::Print(insert_batch_fn);
-
- // The total number of calls to TupleRowComparator::Less(Tuple*, Tuple*)
- // is 3 in PriorityQueue and 1 in TopNNode::Heap::InsertTupleRow().
- // Each Less(Tuple*, Tuple*) indirectly calls TupleRowComparator::Compare()
- // once.
- replaced = codegen->ReplaceCallSites(insert_batch_fn,
- compare_fn, TupleRowComparator::COMPARE_SYMBOL);
- DCHECK_REPLACE_COUNT(replaced, 4) << LlvmCodeGen::Print(insert_batch_fn);
-
- replaced = codegen->ReplaceCallSites(insert_batch_fn,
- materialize_exprs_no_pool_fn, Tuple::MATERIALIZE_EXPRS_NULL_POOL_SYMBOL);
- DCHECK_REPLACE_COUNT(replaced, 1) << LlvmCodeGen::Print(insert_batch_fn);
-
- replaced = codegen->ReplaceCallSitesWithValue(insert_batch_fn,
- codegen->GetI64Constant(tnode_->limit + offset()), "heap_capacity");
- DCHECK_REPLACE_COUNT(replaced, 1) << LlvmCodeGen::Print(insert_batch_fn);
-
- int tuple_byte_size = output_tuple_desc_->byte_size();
- replaced = codegen->ReplaceCallSitesWithValue(insert_batch_fn,
- codegen->GetI32Constant(tuple_byte_size), "tuple_byte_size");
- DCHECK_REPLACE_COUNT(replaced, 1);
-
- insert_batch_fn = codegen->FinalizeFunction(insert_batch_fn);
- DCHECK(insert_batch_fn != NULL);
- codegen->AddFunctionToJit(insert_batch_fn, &codegend_insert_batch_fn_);
- }
+ if (codegen_status.ok()) {
+ int replaced;
+ replaced = codegen->ReplaceCallSites(insert_batch_fn,
+ materialize_exprs_tuple_pool_fn, Tuple::MATERIALIZE_EXPRS_SYMBOL);
+ DCHECK_REPLACE_COUNT(replaced, 1) << LlvmCodeGen::Print(insert_batch_fn);
+
+ replaced = codegen->ReplaceCallSites(insert_batch_fn,
+ materialize_exprs_no_pool_fn, Tuple::MATERIALIZE_EXPRS_NULL_POOL_SYMBOL);
+ DCHECK_REPLACE_COUNT(replaced, 1) << LlvmCodeGen::Print(insert_batch_fn);
+
+ // The total number of calls to tuple_row_less_than_->Compare() is 3 in
+ // PriorityQueue (called from 2 places), 1 in TopNNode::Heap::InsertTupleRow()
+ // and 3 in TopNNode::Heap::InsertTupleWithTieHandling
+ // Each tuple_row_less_than_->Less(Tuple*, Tuple*) indirectly calls Compare() once.
+ replaced = codegen->ReplaceCallSites(insert_batch_fn,
+ compare_fn, TupleRowComparator::COMPARE_SYMBOL);
+ DCHECK_REPLACE_COUNT(replaced, 10) << LlvmCodeGen::Print(insert_batch_fn);
+
+ replaced = codegen->ReplaceCallSitesWithValue(insert_batch_fn,
+ codegen->GetI64Constant(heap_capacity()), "heap_capacity");
+ DCHECK_REPLACE_COUNT(replaced, 2)
+ << LlvmCodeGen::Print(insert_batch_fn);
+
+ replaced = codegen->ReplaceCallSitesWithBoolConst(
+ insert_batch_fn, include_ties(), "include_ties");
+ DCHECK_REPLACE_COUNT(replaced, 1)
+ << LlvmCodeGen::Print(insert_batch_fn);
+
+ int tuple_byte_size = output_tuple_desc_->byte_size();
+ replaced = codegen->ReplaceCallSitesWithValue(insert_batch_fn,
+ codegen->GetI32Constant(tuple_byte_size), "tuple_byte_size");
+ DCHECK_REPLACE_COUNT(replaced, 3)
+ << LlvmCodeGen::Print(insert_batch_fn);
+
+ insert_batch_fn = codegen->FinalizeFunction(insert_batch_fn);
+ DCHECK(insert_batch_fn != NULL);
+ codegen->AddFunctionToJit(insert_batch_fn, &codegend_insert_batch_fn_);
}
}
AddCodegenStatus(codegen_status);
@@ -171,7 +179,9 @@ Status TopNNode::Open(RuntimeState* state) {
RETURN_IF_CANCELLED(state);
RETURN_IF_ERROR(QueryMaintenance(state));
- heap_.reset(new Heap(*tuple_row_less_than_, limit_ + offset_));
+ const TopNPlanNode& pnode = static_cast<const TopNPlanNode&>(plan_node_);
+ heap_.reset(
+ new Heap(*tuple_row_less_than_, pnode.heap_capacity(), pnode.include_ties()));
// Allocate memory for a temporary tuple.
tmp_tuple_ = reinterpret_cast<Tuple*>(
@@ -180,7 +190,7 @@ Status TopNNode::Open(RuntimeState* state) {
RETURN_IF_ERROR(child(0)->Open(state));
// Limit of 0, no need to fetch anything from children.
- if (limit_ != 0) {
+ if (pnode.heap_capacity() != 0) {
RowBatch batch(child(0)->row_desc(), state->batch_size(), mem_tracker());
bool eos;
do {
@@ -188,14 +198,14 @@ Status TopNNode::Open(RuntimeState* state) {
RETURN_IF_ERROR(child(0)->GetNext(state, &batch, &eos));
{
SCOPED_TIMER(insert_batch_timer_);
- const TopNPlanNode::InsertBatchFn insert_batch_fn
- = codegend_insert_batch_fn_.load();
+ TopNPlanNode::InsertBatchFn insert_batch_fn = codegend_insert_batch_fn_.load();
if (insert_batch_fn != nullptr) {
insert_batch_fn(this, &batch);
} else {
InsertBatch(&batch);
}
- if (rows_to_reclaim_ > 2 * (limit_ + offset_)) {
+ DCHECK(heap_->DCheckConsistency());
+ if (rows_to_reclaim_ > 2 * pnode.heap_capacity()) {
RETURN_IF_ERROR(ReclaimTuplePool(state));
COUNTER_ADD(tuple_pool_reclaim_counter_, 1);
}
@@ -204,8 +214,6 @@ Status TopNNode::Open(RuntimeState* state) {
RETURN_IF_ERROR(QueryMaintenance(state));
} while (!eos);
}
- // TODO: this wouldn't be valid for partitioned.
- DCHECK_LE(heap_->num_tuples(), limit_ + offset_);
PrepareForOutput();
// Unless we are inside a subplan expecting to call Open()/GetNext() on the child
@@ -248,6 +256,7 @@ Status TopNNode::GetNext(RuntimeState* state, RowBatch* row_batch, bool* eos) {
Status TopNNode::Reset(RuntimeState* state, RowBatch* row_batch) {
heap_->Reset();
+ tmp_tuple_ = nullptr;
num_rows_skipped_ = 0;
// Transfer ownership of tuple data to output batch.
row_batch->tuple_data_pool()->AcquireData(tuple_pool_.get(), false);
@@ -266,7 +275,7 @@ void TopNNode::Close(RuntimeState* state) {
}
void TopNNode::PrepareForOutput() {
- // TODO: this will need to iterate through partitions.
+ DCHECK(heap_->DCheckConsistency());
heap_->PrepareForOutput(*this, &sorted_top_n_);
get_next_iter_ = sorted_top_n_.begin();
}
@@ -274,17 +283,35 @@ void TopNNode::PrepareForOutput() {
void TopNNode::Heap::PrepareForOutput(
const TopNNode& RESTRICT node, vector<Tuple*>* sorted_top_n) RESTRICT {
// Reverse the order of the tuples in the priority queue
- sorted_top_n->resize(priority_queue_.Size());
+ sorted_top_n->resize(num_tuples());
int64_t index = sorted_top_n->size() - 1;
- while (priority_queue_.Size() > 0) {
+ /// Any ties with the min will be the last elements in 'sorted_top_n'.
+ while (!overflowed_ties_.empty()) {
+ (*sorted_top_n)[index] = overflowed_ties_.back();
+ overflowed_ties_.pop_back();
+ --index;
+ }
+ while (!priority_queue_.Empty()) {
(*sorted_top_n)[index] = priority_queue_.Pop();
--index;
}
}
+bool TopNNode::Heap::DCheckConsistency() {
+ DCHECK_LE(num_tuples(), capacity_ + overflowed_ties_.size())
+ << num_tuples() << " > " << capacity_ << " + " << overflowed_ties_.size();
+ if (!overflowed_ties_.empty()) {
+ DCHECK(include_ties_);
+ DCHECK_EQ(capacity_, priority_queue_.Size())
+ << "Ties should only be present if heap is at capacity";
+ }
+ return true;
+}
+
Status TopNNode::ReclaimTuplePool(RuntimeState* state) {
unique_ptr<MemPool> temp_pool(new MemPool(mem_tracker()));
RETURN_IF_ERROR(heap_->RematerializeTuples(this, state, temp_pool.get()));
+ DCHECK(heap_->DCheckConsistency());
rows_to_reclaim_ = 0;
tmp_tuple_ = reinterpret_cast<Tuple*>(temp_pool->TryAllocate(
@@ -314,34 +341,44 @@ void TopNNode::DebugString(int indentation_level, stringstream* out) const {
*out << ")";
}
-TopNNode::Heap::Heap(const TupleRowComparator& c, int64_t capacity)
- : capacity_(capacity), priority_queue_(c) {
- priority_queue_.Reserve(capacity);
-}
+TopNNode::Heap::Heap(const TupleRowComparator& c, int64_t capacity, bool include_ties) :
+ capacity_(capacity), include_ties_(include_ties), priority_queue_(c) {}
void TopNNode::Heap::Reset() {
priority_queue_.Clear();
+ overflowed_ties_.clear();
}
void TopNNode::Heap::Close() {
priority_queue_.Clear();
+ overflowed_ties_.clear();
}
-Status TopNNode::Heap::RematerializeTuples(TopNNode* node,
- RuntimeState* state, MemPool* new_pool) {
+template <class T>
+Status TopNNode::Heap::RematerializeTuplesHelper(TopNNode* node,
+ RuntimeState* state, MemPool* new_pool, T begin_it, T end_it) {
const TupleDescriptor& tuple_desc = *node->output_tuple_desc_;
int tuple_size = tuple_desc.byte_size();
- for (int i = 0; i < priority_queue_.Size(); i++) {
+ for (T it = begin_it; it != end_it; ++it) {
Tuple* insert_tuple = reinterpret_cast<Tuple*>(new_pool->TryAllocate(tuple_size));
if (UNLIKELY(insert_tuple == nullptr)) {
return new_pool->mem_tracker()->MemLimitExceeded(state,
"Failed to allocate memory in TopNNode::ReclaimTuplePool.", tuple_size);
}
- priority_queue_[i]->DeepCopy(insert_tuple, tuple_desc, new_pool);
- priority_queue_[i] = insert_tuple;
+ (*it)->DeepCopy(insert_tuple, tuple_desc, new_pool);
+ *it = insert_tuple;
}
return Status::OK();
}
+Status TopNNode::Heap::RematerializeTuples(TopNNode* node,
+ RuntimeState* state, MemPool* new_pool) {
+ RETURN_IF_ERROR(RematerializeTuplesHelper(
+ node, state, new_pool, priority_queue_.Begin(), priority_queue_.End()));
+ RETURN_IF_ERROR(RematerializeTuplesHelper(
+ node, state, new_pool, overflowed_ties_.begin(), overflowed_ties_.end()));
+ return Status::OK();
+}
+
template class impala::PriorityQueue<Tuple*, TupleRowComparator>;
template class impala::PriorityQueueIterator<Tuple*, TupleRowComparator>;
diff --git a/be/src/exec/topn-node.h b/be/src/exec/topn-node.h
index 4a6de25..9f033c2 100644
--- a/be/src/exec/topn-node.h
+++ b/be/src/exec/topn-node.h
@@ -46,6 +46,16 @@ class TopNPlanNode : public PlanNode {
}
~TopNPlanNode(){}
+ /// Returns the per-heap capacity used for the Heap objects in this node.
+ int64_t heap_capacity() const {
+ int64_t limit = include_ties() ? tnode_->sort_node.limit_with_ties : tnode_->limit;
+ return limit + offset();
+ }
+
+ bool include_ties() const {
+ return tnode_->sort_node.include_ties;
+ }
+
/// Ordering expressions used for tuple comparison.
std::vector<ScalarExpr*> ordering_exprs_;
@@ -55,7 +65,7 @@ class TopNPlanNode : public PlanNode {
/// Materialization exprs for the output tuple and their evaluators.
std::vector<ScalarExpr*> output_tuple_exprs_;
- /// Config used to create a TupleRowComparator instance.
+ /// Config used to create a TupleRowComparator instance for 'ordering_exprs_'.
TupleRowComparatorConfig* row_comparator_config_ = nullptr;
/// Codegened version of TopNNode::InsertBatch().
@@ -63,12 +73,37 @@ class TopNPlanNode : public PlanNode {
CodegenFnPtr<InsertBatchFn> codegend_insert_batch_fn_;
};
-/// Node for in-memory TopN (ORDER BY ... LIMIT)
-/// This handles the case where the result fits in memory.
-/// This node will materialize its input rows into a new tuple using the expressions
-/// in sort_tuple_slot_exprs_ in its sort_exec_exprs_ member.
-/// TopN is implemented by storing rows in a priority queue.
-
+/// Node for in-memory TopN operator that sorts input tuples and applies a limit such
+/// that only the Top N tuples according to the sort order are returned by the operator.
+/// This node materializes its input rows into a new row format comprised of a single
+/// tuple using the output_tuple_exprs_.
+///
+/// In-memory priority queues, represented as binary heaps, are used to compute the Top N
+/// efficiently. Maintaining in-memory heaps with the current top rows allows discarding
+/// rows that are not in the top N as soon as possible, minimizing the memory requirements
+/// and processing time of the operator.
+///
+/// TODO: currently we only support a single top-n heap per operator. IMPALA-9979 will
+/// add a partitioned mode.
+///
+/// Unpartitioned TopN Implementation Details
+/// =========================================
+/// Unpartitioned mode uses a single in-memory priority queue and does not spill results
+/// to disk. Memory consumption is bounded by the limit. After the input is consumed,
+/// rows can be directly outputted from the priority queue by calling
+/// Heap::PrepareForOutput().
+///
+/// Memory Management
+/// =================
+/// In-memory heaps are backed by 'tuple_pool_' - all tuples in the heaps must reference
+/// only memory allocated from this pool. To reclaim memory from tuples that have been
+/// evicted from the heaps, the in-memory heaps must be re-materialized with a new
+/// MemPool - see ReclaimTuplePool(). In some cases the fixed-length portion of a
+/// tuple can be reused to avoid the need to reclaim all the time.
+///
+/// In unpartitioned mode, reclamation is triggered by 'rows_to_reclaim_' hitting a
+/// threshold, which indicates that enough unused memory may have accumulated to
+/// be worth reclaiming.
class TopNNode : public ExecNode {
public:
TopNNode(ObjectPool* pool, const TopNPlanNode& pnode, const DescriptorTbl& descs);
@@ -87,18 +122,24 @@ class TopNNode : public ExecNode {
friend class TupleLessThan;
- /// Inserts all the rows in 'batch' into the queue.
+ bool include_ties() const {
+ const TopNPlanNode& pnode = static_cast<const TopNPlanNode&>(plan_node_);
+ return pnode.include_ties();
+ }
+
+ /// Inserts all the input rows in 'batch' into 'heap_'.
void InsertBatch(RowBatch* batch);
- /// Prepare to output all of the rows in this operator in sorted order. Initializes
- /// 'sorted_top_n_' and 'get_next_iterator_'.
+ /// Prepare to start outputting rows. Called after consuming all rows from the child.
+ /// Collects all output rows in 'sorted_top_n_' and initializes 'get_next_iter_' to
+ /// point to the first row.
void PrepareForOutput();
- // Re-materialize all tuples that reference 'tuple_pool_' and release 'tuple_pool_',
- // replacing it with a new pool.
+ /// Re-materialize all tuples that reference 'tuple_pool_' and release 'tuple_pool_',
+ /// replacing it with a new pool.
Status ReclaimTuplePool(RuntimeState* state);
- IR_NO_INLINE int tuple_byte_size() const {
+ IR_NO_INLINE int tuple_byte_size() const noexcept {
return output_tuple_desc_->byte_size();
}
@@ -112,17 +153,13 @@ class TopNNode : public ExecNode {
/// Cached descriptor for the materialized tuple.
TupleDescriptor* const output_tuple_desc_;
- /// Comparator for priority_queue_.
+ /// Comparator for ordering tuples globally.
std::unique_ptr<TupleRowComparator> tuple_row_less_than_;
- /// After computing the TopN in the priority_queue, pop them and put them in this vector
+ /// Temporary staging vector for sorted tuples extracted from a Heap via
+ /// Heap::PrepareForOutput().
std::vector<Tuple*> sorted_top_n_;
- /// Tuple allocated once from tuple_pool_ and reused in InsertTupleRow to
- /// materialize input tuples if necessary. After materialization, tmp_tuple_ may be
- /// copied into the tuple pool and inserted into the priority queue.
- Tuple* tmp_tuple_ = nullptr;
-
/// Stores everything referenced in priority_queue_.
std::unique_ptr<MemPool> tuple_pool_;
@@ -133,10 +170,10 @@ class TopNNode : public ExecNode {
/// was used to create this instance.
const CodegenFnPtr<TopNPlanNode::InsertBatchFn>& codegend_insert_batch_fn_;
- /// Timer for time spent in InsertBatch() function (or codegen'd version)
+ /// Timer for time spent in InsertBatch() function (or codegen'd version).
RuntimeProfile::Counter* insert_batch_timer_;
- /// Number of rows to be reclaimed since tuple_pool_ was last created/reclaimed
+ /// Number of rows to be reclaimed since tuple_pool_ was last created/reclaimed.
int64_t rows_to_reclaim_;
/// Number of times tuple pool memory was reclaimed
@@ -145,10 +182,15 @@ class TopNNode : public ExecNode {
/////////////////////////////////////////
/// BEGIN: Members that must be Reset()
- /// A heap containing up to 'limit_' + 'offset_' rows.
+ /// Tuple allocated once from tuple_pool_ and reused in InsertTupleRow to
+ /// materialize input tuples if necessary. After materialization, tmp_tuple_ may be
+ /// copied into the tuple pool and inserted into the priority queue.
+ Tuple* tmp_tuple_ = nullptr;
+
+ // Single heap used as the main heap in unpartitioned Top-N.
std::unique_ptr<Heap> heap_;
- /// Number of rows skipped. Used for adhering to offset_.
+ /// Number of rows skipped. Used for adhering to offset_ in unpartitioned Top-N.
int64_t num_rows_skipped_;
/// END: Members that must be Reset()
@@ -159,7 +201,7 @@ class TopNNode : public ExecNode {
/// up to 'capacity' tuples.
class TopNNode::Heap {
public:
- Heap(const TupleRowComparator& c, int64_t capacity);
+ Heap(const TupleRowComparator& c, int64_t capacity, bool include_ties);
void Reset();
void Close();
@@ -168,8 +210,8 @@ class TopNNode::Heap {
/// copy of 'tuple_row', which it stores in 'tuple_pool'. Always inlined in IR into
/// TopNNode::InsertBatch() because codegen relies on this for substituting exprs
/// in the body of TopNNode.
- /// Returns true if a previous row was replaced.
- bool IR_ALWAYS_INLINE InsertTupleRow(TopNNode* node, TupleRow* input_row);
+ /// Returns the number of materialized tuples discarded that may need to be reclaimed.
+ int IR_ALWAYS_INLINE InsertTupleRow(TopNNode* node, TupleRow* input_row);
/// Copy the elements in the priority queue into a new tuple pool, and release
/// the previous pool.
@@ -180,26 +222,57 @@ class TopNNode::Heap {
void PrepareForOutput(
const TopNNode& RESTRICT node, std::vector<Tuple*>* sorted_top_n) RESTRICT;
+ /// Can be called to invoke DCHECKs if the heap is in an inconsistent state.
+ /// Returns a bool so it can be wrapped in a DCHECK() macro.
+ bool DCheckConsistency();
+
/// Returns number of tuples currently in heap.
- int64_t num_tuples() const { return priority_queue_.Size(); }
+ int64_t num_tuples() const { return priority_queue_.Size() + overflowed_ties_.size(); }
- IR_NO_INLINE int64_t heap_capacity() const { return capacity_; }
+ IR_NO_INLINE int64_t heap_capacity() const noexcept { return capacity_; }
+
+ IR_NO_INLINE bool include_ties() const noexcept { return include_ties_; }
private:
+ /// Helper for RematerializeTuples() that materializes the tuples in a container in the
+ /// range (begin_it, end_it].
+ template <class T>
+ Status RematerializeTuplesHelper(TopNNode* node, RuntimeState* state,
+ MemPool* new_pool, T begin_it, T end_it);
+
+ /// Helper to insert tuple row into a full priority queue with tie handling. This
+ /// should not be called until the heap is at capacity and tie handling is needed.
+ /// 'materialized_tuple' must be materialized into the output row format, i.e.
+ /// output_tuple_desc_. Returns the number of materialized tuples discarded as a result
+ /// of this function.
+ /// Always inlined in IR because codegen relies on this for substituting exprs in the
+ /// body of the function.
+ int IR_ALWAYS_INLINE InsertTupleWithTieHandling(
+ TopNNode* node, Tuple* materialized_tuple);
+
+
/// Limit on capacity of 'priority_queue_'. If inserting a tuple into the queue
/// would exceed this, a tuple is popped off the queue.
const int64_t capacity_;
+ /// If true, the heap may include more than 'capacity_' if multiple tuples are
+ /// tied to be the head of the heap.
+ const bool include_ties_;
+
/////////////////////////////////////////
/// BEGIN: Members that must be Reset()
- /// The priority queue (represented by a vector and modified using
- /// push_heap()/pop_heap() to maintain ordered heap invariants) will never have more
- /// elements in it than the LIMIT + OFFSET. The order of the queue is the opposite of
- /// what the ORDER BY clause specifies, such that the top of the queue is the last
- /// sorted element.
+ /// The priority queue is represented by a vector and modified using push_heap()/
+ /// pop_heap() to maintain ordered heap invariants. It has up to 'capacity_' elements
+ /// in the heap. The order of the queue is the opposite of what the ORDER BY clause
+ /// specifies, such that the head of the queue (i.e. index 0) is the last sorted
+ /// element.
PriorityQueue<Tuple*, TupleRowComparator> priority_queue_;
+ /// Tuples tied with the head of the priority queue in excess of the heap capacity.
+ /// Only used when 'include_ties_' is true.
+ std::vector<Tuple*> overflowed_ties_;
+
/// END: Members that must be Reset()
/////////////////////////////////////////
};
diff --git a/be/src/util/tuple-row-compare.h b/be/src/util/tuple-row-compare.h
index 25b9f4d..bf982c4 100644
--- a/be/src/util/tuple-row-compare.h
+++ b/be/src/util/tuple-row-compare.h
@@ -113,6 +113,12 @@ class TupleRowComparator {
return Compare(nullptr, nullptr, lhs, rhs);
}
+ int ALWAYS_INLINE Compare(const Tuple* lhs, const Tuple* rhs) const {
+ TupleRow* lhs_row = reinterpret_cast<TupleRow*>(&lhs);
+ TupleRow* rhs_row = reinterpret_cast<TupleRow*>(&rhs);
+ return Compare(lhs_row, rhs_row);
+ }
+
/// Returns true if lhs is strictly less than rhs.
/// All exprs (ordering_exprs_lhs_ and ordering_exprs_rhs_) must have been prepared
/// and opened before calling this.
diff --git a/common/thrift/PlanNodes.thrift b/common/thrift/PlanNodes.thrift
index 6987e8e..e6a9bb5 100644
--- a/common/thrift/PlanNodes.thrift
+++ b/common/thrift/PlanNodes.thrift
@@ -485,6 +485,12 @@ struct TSortNode {
// Estimated bytes of input that will go into this sort node across all backends.
// -1 if such estimate is unavailable.
4: optional i64 estimated_full_input_size
+
+ // Whether to include ties for the last place in the Top-N
+ 5: optional bool include_ties
+
+ // If include_ties is true, the limit including ties.
+ 6: optional i64 limit_with_ties
}
enum TAnalyticWindowType {
diff --git a/fe/src/main/java/org/apache/impala/analysis/AnalyticExpr.java b/fe/src/main/java/org/apache/impala/analysis/AnalyticExpr.java
index 1fac321..c61bcbe 100644
--- a/fe/src/main/java/org/apache/impala/analysis/AnalyticExpr.java
+++ b/fe/src/main/java/org/apache/impala/analysis/AnalyticExpr.java
@@ -17,7 +17,6 @@
package org.apache.impala.analysis;
-import java.math.BigInteger;
import java.util.ArrayList;
import java.util.List;
@@ -74,15 +73,16 @@ public class AnalyticExpr extends Expr {
// SQL string of this AnalyticExpr before standardization. Returned in toSqlImpl().
private String sqlString_;
+ public static String DENSERANK = "dense_rank";
+ public static String RANK = "rank";
+ public static String ROWNUMBER = "row_number";
+
private static String LEAD = "lead";
private static String LAG = "lag";
private static String FIRST_VALUE = "first_value";
private static String LAST_VALUE = "last_value";
private static String FIRST_VALUE_IGNORE_NULLS = "first_value_ignore_nulls";
private static String LAST_VALUE_IGNORE_NULLS = "last_value_ignore_nulls";
- private static String RANK = "rank";
- private static String DENSERANK = "dense_rank";
- private static String ROWNUMBER = "row_number";
private static String MIN = "min";
private static String MAX = "max";
private static String PERCENT_RANK = "percent_rank";
@@ -190,7 +190,7 @@ public class AnalyticExpr extends Expr {
&& ((AggregateFunction) fn).isAnalyticFn();
}
- private static boolean isAnalyticFn(Function fn, String fnName) {
+ public static boolean isAnalyticFn(Function fn, String fnName) {
return isAnalyticFn(fn) && fn.functionName().equals(fnName);
}
diff --git a/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java b/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java
index 458f228..70d5cba 100644
--- a/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java
@@ -38,6 +38,7 @@ import org.apache.impala.analysis.SlotRef;
import org.apache.impala.analysis.SortInfo;
import org.apache.impala.analysis.TupleDescriptor;
import org.apache.impala.analysis.TupleId;
+import org.apache.impala.catalog.Function;
import org.apache.impala.common.Pair;
import org.apache.impala.thrift.TAnalyticNode;
import org.apache.impala.thrift.TExplainLevel;
@@ -221,7 +222,7 @@ public class AnalyticEvalNode extends PlanNode {
// compare elements[i]
Expr lhs = elements.get(i);
- Preconditions.checkState(lhs.isBound(inputTid));
+ Preconditions.checkState(lhs.isBound(inputTid), lhs.debugString());
Expr rhs = lhs.substitute(bufferedSmap, analyzer, false);
Expr bothNull = new CompoundPredicate(
@@ -362,6 +363,19 @@ public class AnalyticEvalNode extends PlanNode {
.setSpillableBufferBytes(bufferSize).setMaxRowBufferBytes(bufferSize).build();
}
+ public static class LimitPushdownInfo {
+ public LimitPushdownInfo(boolean includeTies, int additionalLimit) {
+ this.includeTies = includeTies;
+ this.additionalLimit = additionalLimit;
+ }
+
+ // Whether ties need to be returned from the top-n operator.
+ public final boolean includeTies;
+
+ // Amount that needs to be added to limit pushed down.
+ public final int additionalLimit;
+ }
+
/**
* Check if it is safe to push down limit to the Sort node of this AnalyticEval.
* Qualifying checks:
@@ -371,33 +385,33 @@ public class AnalyticEvalNode extends PlanNode {
* - If there is a predicate on the analytic function (provided through the
* selectNode), the predicate's eligibility is checked (see further below)
* @param sortInfo The sort info from the outer sort node
+ * @param sortInputSmap the input smap that can be applied to the sort exprs from
+ * sortInfo to get exprs referencing its input tuple.
* @param selectNode The selection node with predicates on analytic function.
* This can be null if no such predicate is present.
* @param limit Limit value from the outer sort node
* @param analyticNodeSort The analytic sort associated with this analytic node
- * @param sortExprsForPartitioning A placeholder list supplied by caller that is
- * populated with the sort exprs of the analytic sort that will be later used
- * for hash partitioning of the distributed TopN.
* @param analyzer analyzer instance
- * @return True if limit pushdown into analytic sort is safe, False if not
+ * @return parameters for limit pushdown into analytic sort if allowed, null if not
*/
- public boolean isLimitPushdownSafe(SortInfo sortInfo, SelectNode selectNode,
- long limit, SortNode analyticNodeSort, List<Expr> sortExprsForPartitioning,
- Analyzer analyzer) {
- if (analyticFnCalls_.size() != 1) return false;
+ public LimitPushdownInfo checkForLimitPushdown(SortInfo sortInfo,
+ ExprSubstitutionMap sortInputSmap, SelectNode selectNode, long limit,
+ SortNode analyticNodeSort, Analyzer analyzer) {
+ if (analyticFnCalls_.size() != 1) return null;
Expr expr = analyticFnCalls_.get(0);
if (!(expr instanceof FunctionCallExpr) ||
(!AnalyticExpr.isRankingFn(((FunctionCallExpr) expr).getFn()))) {
- return false;
+ return null;
}
+
List<Expr> analyticSortSortExprs = analyticNodeSort.getSortInfo().getSortExprs();
// In the mapping below, we use the original sort exprs that the sortInfo was
// created with, not the sort exprs that got mapped in SortInfo.createSortTupleInfo().
- // This allows us to substitute it using this node's output smap.
+ // This allows us to substitute it so that it references the analytic sort tuple.
List<Expr> origSortExprs = sortInfo != null ? sortInfo.getOrigSortExprs() :
new ArrayList<>();
- List<Expr> sortExprs = Expr.substituteList(origSortExprs, getOutputSmap(),
+ List<Expr> sortExprs = Expr.substituteList(origSortExprs, sortInputSmap,
analyzer, false);
// Also use substituted partition exprs such that they can be compared with the
// sort exprs
@@ -408,58 +422,51 @@ public class AnalyticEvalNode extends PlanNode {
// the limit to the sort below if there is no selection node or if
// the predicate in the selection node is eligible
if (selectNode == null) {
- return true;
+ return new LimitPushdownInfo(false, 0);
}
- Pair<Boolean, Double> status =
- isPredEligibleForLimitPushdown(selectNode.getConjuncts(), limit);
- if (status.first) {
- sortExprsForPartitioning.addAll(analyticSortSortExprs);
+ Pair<LimitPushdownInfo, Double> status =
+ checkPredEligibleForLimitPushdown(selectNode.getConjuncts(), limit);
+ if (status.first != null) {
selectNode.setSelectivity(status.second);
- return true;
+ return status.first;
}
- return false;
+ return null;
}
Preconditions.checkArgument(analyticSortSortExprs.size() >= pbExprs.size());
// Check if pby exprs are a prefix of the top level sort exprs
- // TODO: also check if subsequent expressions match. Need to check ASC and NULLS FIRST
- // compatibility more explicitly in the case.
- if (sortExprs.size() == 0) {
- sortExprsForPartitioning.addAll(pbExprs);
- } else {
+ if (sortExprs.size() > 0) {
if (!analyticSortExprsArePrefix(
sortInfo, sortExprs, analyticNodeSort.getSortInfo(), pbExprs)) {
- return false;
+ return null;
}
-
- // get the corresponding sort exprs from the analytic sort
- // since that's what will eventually be used for hash partitioning
- sortExprsForPartitioning.addAll(analyticSortSortExprs.subList(0, pbExprs.size()));
}
-
// check that the window frame is UNBOUNDED PRECEDING to CURRENT ROW
if (!(analyticWindow_.getLeftBoundary().getType() ==
AnalyticWindow.BoundaryType.UNBOUNDED_PRECEDING
&& analyticWindow_.getRightBoundary().getType() ==
AnalyticWindow.BoundaryType.CURRENT_ROW)) {
- return false;
+ return null;
}
if (selectNode == null) {
// Limit pushdown is valid if the pre-analytic sort puts rows into the same order
// as sortExprs. We check the prefix match, since extra analytic sort exprs do not
// affect the compatibility of the ordering with sortExprs.
- return analyticSortExprsArePrefix(sortInfo, sortExprs,
- analyticNodeSort.getSortInfo(), analyticSortSortExprs);
+ if (!analyticSortExprsArePrefix(sortInfo, sortExprs,
+ analyticNodeSort.getSortInfo(), analyticSortSortExprs)) {
+ return null;
+ }
+ return new LimitPushdownInfo(false, 0);
} else {
- Pair<Boolean, Double> status =
- isPredEligibleForLimitPushdown(selectNode.getConjuncts(), limit);
- if (status.first) {
+ Pair<LimitPushdownInfo, Double> status =
+ checkPredEligibleForLimitPushdown(selectNode.getConjuncts(), limit);
+ if (status.first != null) {
selectNode.setSelectivity(status.second);
- return true;
+ return status.first;
}
}
- return false;
+ return null;
}
/**
@@ -492,16 +499,47 @@ public class AnalyticEvalNode extends PlanNode {
/**
* Check eligibility of a predicate (provided as list of conjuncts) for limit
* pushdown optimization.
- * TODO: IMPALA-10296: this check is not strict enough to allow limit pushdown to
- * be safe in all circumstances.
+ *
+ * Pushing down the limit past the predicate is not, in general, possible because
+ * the predicate can filter out an arbitrary number of rows flowing returned from
+ * the analytic sort. However, if the predicate is a rank() or row_number() analytic
+ * predicate, in some cases we can determine an upper bound on the number of rows
+ * returned from the analytic sort that will end up in the output of the final top-n.
+ *
+ * The following conditions must be met:
+ * 1. The final top-n sort order must match the analytic partition sort order.
+ * The order within the partition doesn't matter for the purposes of the pushdown
+ * decision.
+ * 2. A single rank() or row_number() predicate evaluated by the analytic operator
+ * between the final top-n and analytic sort.
+ * 3. The max rows returned from each analytic partition - P - must be >= the limit
+ * being pushed down - N. (Note: ties in rank() means that P is not a strict limit
+ * on rows returned from each partition, but extra rows don't invalidate the
+ * argument below).
+ *
+ * In this case the rows returned from the final sort are going to come from the
+ * first analytic partitions because of #1. I.e. if we go through the partitions
+ * in order, until the row count adds up to N, then that is the set of partitions
+ * that can possibly end up in the output of the final top-n.
+ * If P >= N, we will never hit a partition limit before we hit N.
+ *
+ * If P < N, we could hit a partition limit and filter out an arbitrary number of
+ * rows before we find the next row to be returned, therefore the optimization
+ * can't be applied.
+ *
+ * The ordering within a partition in the analytic sort may be different from the
+ * final order, so we need to make sure to include all of the relevant rows in the
+ * final partition, which we can do by pushing down a limit of P + N, and including
+ * ties if the predicate is rank().
+ *
* @param conjuncts list of conjuncts from the predicate
- * @param limit limit from outer sort
- * @return a Pair whose first value is True if the conjuncts+limit allows pushdown,
- * False otherwise. Second value is the predicate's estimated selectivity
+ * @param limit limit from final sort
+ * @return a Pair whose first value is non-null if the conjuncts+limit allows pushdown,
+ * null otherwise. Second value is the predicate's estimated selectivity
*/
- private Pair<Boolean, Double> isPredEligibleForLimitPushdown(List<Expr> conjuncts,
- long limit) {
- Pair<Boolean, Double> falseStatus = new Pair<>(false, -1.0);
+ private Pair<LimitPushdownInfo, Double> checkPredEligibleForLimitPushdown(
+ List<Expr> conjuncts, long limit) {
+ Pair<LimitPushdownInfo, Double> falseStatus = new Pair<>(null, -1.0);
// Currently, single conjuncts are supported. In the future, multiple conjuncts
// involving a range e.g 'col >= 10 AND col <= 20' could potentially be supported
if (conjuncts.size() > 1) return falseStatus;
@@ -512,7 +550,8 @@ public class AnalyticEvalNode extends PlanNode {
Expr rhs = pred.getChild(1);
// Lhs of the binary predicate must be a ranking function.
// Also, it must be bound to the output tuple of this analytic eval node
- if (!(lhs instanceof SlotRef)) {
+ if (!(lhs instanceof SlotRef) ||
+ !lhs.isBound(outputTupleDesc_.getId())) {
return falseStatus;
}
List<Expr> lhsSourceExprs = ((SlotRef) lhs).getDesc().getSourceExprs();
@@ -520,11 +559,25 @@ public class AnalyticEvalNode extends PlanNode {
!(lhsSourceExprs.get(0) instanceof AnalyticExpr)) {
return falseStatus;
}
- if (!(AnalyticExpr.isRankingFn(((AnalyticExpr) lhsSourceExprs.
- get(0)).getFnCall().getFn()))
- || !lhs.isBound(outputTupleDesc_.getId())) {
+
+ boolean includeTies;
+ Function fn = ((AnalyticExpr) lhsSourceExprs.get(0)).getFnCall().getFn();
+ if (AnalyticExpr.isAnalyticFn(fn, AnalyticExpr.RANK)) {
+ // RANK() assigns equal values to rows that compare equal. Thus if there are
+ // ties for the max RANK() value to be returned, they all pass the filter.
+ // E.g. if the predicate is RANK() <= 3, and the values we are ordering by are
+ // 1, 42, 99, 99, 100, then the RANK() values are 1, 2, 3, 3, 5 and we need
+ // to return the top 4 rows:
+ // 1, 42, 99, 99. We do not need special handling for ties except at the limit
+ // value. E.g. if the ordering values were 1, 1, 42, 99, 100, then the RANK()
+ // values would be 1, 1, 3, 4, 5 and we only return the top 3.
+ includeTies = true;
+ } else if (AnalyticExpr.isAnalyticFn(fn, AnalyticExpr.ROWNUMBER)) {
+ includeTies = false;
+ } else {
return falseStatus;
}
+
// Restrict the pushdown for =, <, <= predicates because these ensure the
// qualifying rows are fully 'contained' within the LIMIT value. Other
// types of predicates would select rows that fall outside the LIMIT range.
@@ -533,12 +586,24 @@ public class AnalyticEvalNode extends PlanNode {
pred.getOp() == BinaryPredicate.Operator.LE)) {
return falseStatus;
}
- // Rhs of the predicate must be a numeric literal and its value
- // must be less than or equal to the limit.
- if (!(rhs instanceof NumericLiteral) ||
- ((NumericLiteral)rhs).getLongValue() > limit) {
- return falseStatus;
- }
+ // Rhs of the predicate must be a numeric literal we need to add the value
+ // to the limit to get correct results when the ordering expressions don't
+ // exactly match between the analytic sort and the parent sort.
+ if (!(rhs instanceof NumericLiteral)) return falseStatus;
+ long analyticLimit = ((NumericLiteral)rhs).getLongValue();
+ if (pred.getOp() == BinaryPredicate.Operator.LT) analyticLimit--;
+
+ // Equality predicates can filter out an arbitrary number of rows from
+ // each analytic partition, so it is not safe to push the limit down.
+ if (pred.getOp() == BinaryPredicate.Operator.EQ && limit > 1) return falseStatus;
+
+ // See method comment for explanation of why the analytic partition limit
+ // must be >= the pushed down limit.
+ if (analyticLimit < limit) return falseStatus;
+
+ // Avoid overflow of limit.
+ if (analyticLimit + (long)limit > Integer.MAX_VALUE) return falseStatus;
+
double selectivity = Expr.DEFAULT_SELECTIVITY;
// Since the predicate is qualified for limit pushdown, estimate its selectivity.
// For EQ conditions, leave it as the default. For LT and LE, assume all of the
@@ -547,7 +612,8 @@ public class AnalyticEvalNode extends PlanNode {
pred.getOp() == BinaryPredicate.Operator.LE) {
selectivity = 1.0;
}
- return new Pair<Boolean, Double>(true, selectivity);
+ return new Pair<LimitPushdownInfo, Double>(
+ new LimitPushdownInfo(includeTies, (int)analyticLimit), selectivity);
}
}
diff --git a/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java b/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
index f284e32..9f7456d 100644
--- a/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
@@ -1089,19 +1089,19 @@ public class DistributedPlanner {
// Make sure the childFragment's output is partitioned as required by the sortNode.
DataPartition sortPartition = sortNode.getInputPartition();
if (!childFragment.getDataPartition().equals(sortPartition)) {
- if (sortNode.hasLimit() && sortNode.isTypeTopN()) {
+ if (sortNode.isTypeTopN()) {
lowerTopN = sortNode;
childFragment.addPlanRoot(lowerTopN);
+ // Update partitioning exprs to reference sort tuple.
+ sortPartition.substitute(
+ sortNode.getSortInfo().getOutputSmap(), ctx_.getRootAnalyzer());
addedLowerTopN = true;
- DataPartition hashPartition =
- DataPartition.hashPartitioned(lowerTopN.getPartitioningExprs());
// When creating the analytic fragment, pass in a flag to unset the limit
// on the partition exchange. This ensures that the exchange does not
// prematurely stop sending rows in case there's a downstream operator
// that has a LIMIT - for instance a Sort with LIMIT after the
// Analytic operator.
- analyticFragment = createParentFragment(childFragment, hashPartition,
- true);
+ analyticFragment = createParentFragment(childFragment, sortPartition, true);
} else {
analyticFragment = createParentFragment(childFragment, sortPartition);
}
@@ -1109,11 +1109,12 @@ public class DistributedPlanner {
}
if (addedLowerTopN) {
// Create the upper TopN node
- SortNode upperTopN = SortNode.createTopNSortNode(ctx_.getNextNodeId(),
- childFragment.getPlanRoot(), lowerTopN.getSortInfo(), sortNode.getOffset());
+ SortNode upperTopN = SortNode.createTopNSortNode(ctx_.getQueryOptions(),
+ ctx_.getNextNodeId(), childFragment.getPlanRoot(),
+ lowerTopN.getSortInfo(), sortNode.getOffset(), lowerTopN.getSortLimit(),
+ lowerTopN.isIncludeTies());
upperTopN.setIsAnalyticSort(true);
upperTopN.init(ctx_.getRootAnalyzer());
- upperTopN.setLimit(lowerTopN.getLimit());
// connect this to the analytic eval node
analyticNode.setChild(0, upperTopN);
analyticFragment.addPlanRoot(upperTopN);
@@ -1137,33 +1138,47 @@ public class DistributedPlanner {
childFragment.addPlanRoot(node);
if (!childFragment.isPartitioned()) return childFragment;
- // Remember original offset and limit.
- boolean hasLimit = node.hasLimit();
- long limit = node.getLimit();
- long offset = node.getOffset();
-
// Create a new fragment for a sort-merging exchange.
PlanFragment mergeFragment =
createParentFragment(childFragment, DataPartition.UNPARTITIONED);
ExchangeNode exchNode = (ExchangeNode) mergeFragment.getPlanRoot();
-
- // Set limit, offset and merge parameters in the exchange node.
- exchNode.unsetLimit();
- if (hasLimit) exchNode.setLimit(limit);
- exchNode.setMergeInfo(node.getSortInfo(), offset);
-
- // Child nodes should not process the offset. If there is a limit,
- // the child nodes need only return (offset + limit) rows.
SortNode childSortNode = (SortNode) childFragment.getPlanRoot();
- Preconditions.checkState(node == childSortNode);
- if (hasLimit) {
- childSortNode.unsetLimit();
- childSortNode.setLimit(PlanNode.checkedAdd(limit, offset));
+ if (node.isIncludeTies()) {
+ Preconditions.checkState(node.getOffset() == 0,
+ "Tie handling with offset not supported");
+ // TopN that returns ties needs special handling because ties are not handled
+ // correctly by the ExchangeNode limit. We need to generate a top-n on top
+ // of the exchange to correctly merge the input.
+ SortNode parentSortNode = SortNode.createTopNSortNode(
+ ctx_.getQueryOptions(), ctx_.getNextNodeId(), exchNode,
+ childSortNode.getSortInfo(), 0, node.getSortLimit(),
+ childSortNode.isIncludeTies());
+ parentSortNode.init(ctx_.getRootAnalyzer());
+ mergeFragment.addPlanRoot(parentSortNode);
+ } else {
+ // Remember original offset and limit.
+ boolean hasLimit = node.hasLimit();
+ long limit = node.getLimit();
+ long offset = node.getOffset();
+
+ // Set limit, offset and merge parameters in the exchange node.
+ exchNode.unsetLimit();
+ if (hasLimit) exchNode.setLimit(limit);
+ exchNode.setMergeInfo(node.getSortInfo(), offset);
+
+ // Child nodes should not process the offset. If there is a limit,
+ // the child nodes need only return (offset + limit) rows.
+ Preconditions.checkState(node == childSortNode);
+ if (hasLimit) {
+ childSortNode.unsetLimit();
+ childSortNode.setLimit(PlanNode.checkedAdd(limit, offset));
+ }
+ childSortNode.setOffset(0);
}
- childSortNode.setOffset(0);
+
+
childSortNode.computeStats(ctx_.getRootAnalyzer());
exchNode.computeStats(ctx_.getRootAnalyzer());
-
return mergeFragment;
}
}
diff --git a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
index 2a85d9e..a556d80 100644
--- a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
@@ -64,8 +64,8 @@ import org.apache.impala.catalog.FeDataSourceTable;
import org.apache.impala.catalog.FeFsPartition;
import org.apache.impala.catalog.FeFsTable;
import org.apache.impala.catalog.FeHBaseTable;
-import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.catalog.FeIcebergTable;
+import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.HdfsFileFormat;
import org.apache.impala.catalog.ScalarType;
@@ -75,6 +75,7 @@ import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.InternalException;
import org.apache.impala.common.NotImplementedException;
import org.apache.impala.common.Pair;
+import org.apache.impala.planner.AnalyticEvalNode.LimitPushdownInfo;
import org.apache.impala.planner.JoinNode.DistributionMode;
import org.apache.impala.thrift.TQueryOptions;
import org.apache.impala.util.AcidUtils;
@@ -336,35 +337,20 @@ public class SingleNodePlanner {
long limit, long offset, boolean hasLimit, boolean disableTopN)
throws ImpalaException {
SortNode sortNode;
- long topNBytesLimit = ctx_.getQueryOptions().topn_bytes_limit;
if (hasLimit && offset == 0) {
checkAndApplyLimitPushdown(root, sortInfo, limit, analyzer);
}
if (hasLimit && !disableTopN) {
- if (topNBytesLimit <= 0) {
- sortNode =
- SortNode.createTopNSortNode(ctx_.getNextNodeId(), root, sortInfo, offset);
- } else {
- long topNCardinality = PlanNode.capCardinalityAtLimit(root.cardinality_, limit);
- long estimatedTopNMaterializedSize =
- sortInfo.estimateTopNMaterializedSize(topNCardinality, offset);
-
- if (estimatedTopNMaterializedSize < topNBytesLimit) {
- sortNode =
- SortNode.createTopNSortNode(ctx_.getNextNodeId(), root, sortInfo, offset);
- } else {
- sortNode =
- SortNode.createTotalSortNode(ctx_.getNextNodeId(), root, sortInfo, offset);
- }
- }
+ sortNode = SortNode.createTopNSortNode(ctx_.getQueryOptions(),
+ ctx_.getNextNodeId(), root, sortInfo, offset, limit, false);
} else {
sortNode =
SortNode.createTotalSortNode(ctx_.getNextNodeId(), root, sortInfo, offset);
+ sortNode.setLimit(limit);
}
Preconditions.checkState(sortNode.hasValidStats());
- sortNode.setLimit(limit);
sortNode.init(analyzer);
return sortNode;
@@ -376,10 +362,9 @@ public class SingleNodePlanner {
*/
private void checkAndApplyLimitPushdown(PlanNode root, SortInfo sortInfo, long limit,
Analyzer analyzer) {
- boolean pushdownLimit = false;
+ LimitPushdownInfo pushdownLimit = null;
AnalyticEvalNode analyticNode = null;
List<PlanNode> intermediateNodes = new ArrayList<>();
- List<Expr> partitioningExprs = new ArrayList<>();
SortNode analyticNodeSort = null;
PlanNode descendant = findDescendantAnalyticNode(root, intermediateNodes);
if (descendant != null && intermediateNodes.size() <= 1) {
@@ -391,24 +376,30 @@ public class SingleNodePlanner {
return;
}
analyticNodeSort = (SortNode) analyticNode.getChild(0);
+ // Don't attempt conversion if there is already a limit or offset.
+ if (analyticNodeSort.hasLimit() || analyticNodeSort.hasOffset()) {
+ return;
+ }
int numNodes = intermediateNodes.size();
if (numNodes > 1 ||
(numNodes == 1 && !(intermediateNodes.get(0) instanceof SelectNode))) {
- pushdownLimit = false;
+ pushdownLimit = null;
} else if (numNodes == 0) {
- pushdownLimit = analyticNode.isLimitPushdownSafe(sortInfo, null,
- limit, analyticNodeSort, partitioningExprs, ctx_.getRootAnalyzer());
+ pushdownLimit = analyticNode.checkForLimitPushdown(sortInfo,
+ root.getOutputSmap(), null, limit, analyticNodeSort, ctx_.getRootAnalyzer());
} else {
SelectNode selectNode = (SelectNode) intermediateNodes.get(0);
- pushdownLimit = analyticNode.isLimitPushdownSafe(sortInfo, selectNode,
- limit, analyticNodeSort, partitioningExprs, ctx_.getRootAnalyzer());
+ pushdownLimit = analyticNode.checkForLimitPushdown(sortInfo,
+ root.getOutputSmap(), selectNode, limit, analyticNodeSort,
+ ctx_.getRootAnalyzer());
}
}
- if (pushdownLimit) {
+ if (pushdownLimit != null) {
Preconditions.checkArgument(analyticNode != null);
Preconditions.checkArgument(analyticNode.getChild(0) instanceof SortNode);
- analyticNodeSort.convertToTopN(limit, partitioningExprs, analyzer);
+ analyticNodeSort.tryConvertToTopN(limit + pushdownLimit.additionalLimit,
+ analyzer, pushdownLimit.includeTies);
// after the limit is pushed down, update stats for the analytic eval node
// and intermediate nodes
analyticNode.computeStats(analyzer);
@@ -1997,7 +1988,6 @@ public class SingleNodePlanner {
Expr.getIds(allJoinConjuncts, null, allSlotIds);
List<TupleId> joinInputTupleIds = joinInput.getTupleIds();
List<Expr> distinctExprs = new ArrayList<>();
- double estDistinctTupleSize = 0;
for (SlotDescriptor slot : analyzer.getSlotDescs(allSlotIds)) {
if (joinInputTupleIds.contains(slot.getParent().getId())) {
distinctExprs.add(new SlotRef(slot));
diff --git a/fe/src/main/java/org/apache/impala/planner/SortNode.java b/fe/src/main/java/org/apache/impala/planner/SortNode.java
index ff78ab1..bfcebd5 100644
--- a/fe/src/main/java/org/apache/impala/planner/SortNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/SortNode.java
@@ -68,9 +68,6 @@ public class SortNode extends PlanNode {
// analytic eval node otherwise null
private AnalyticEvalNode analyticEvalNode_;
- // set only for the analytic sort node
- private List<Expr> partitioningExprs_;
-
// info_.sortTupleSlotExprs_ substituted with the outputSmap_ for materialized slots
// in init().
private List<Expr> resolvedTupleExprs_;
@@ -78,6 +75,14 @@ public class SortNode extends PlanNode {
// The offset of the first row to return.
protected long offset_;
+ // Whether to include ties for the last place in the Top-N values.
+ // Only supported if type_ is TOPN.
+ protected boolean includeTies_;
+
+ // If includeTies_ is true, this is the limit used. We cannot use the default 'limit_'
+ // because the plan node may return more than this number of rows.
+ protected long limitWithTies_;
+
// The type of sort. Determines the exec node used in the BE.
private TSortType type_;
@@ -90,15 +95,30 @@ public class SortNode extends PlanNode {
*/
public static SortNode createPartialSortNode(
PlanNodeId id, PlanNode input, SortInfo info) {
- return new SortNode(id, input, info, 0, TSortType.PARTIAL);
+ return new SortNode(id, input, info, 0, -1, false, TSortType.PARTIAL);
}
/**
- * Creates a new SortNode with a limit that is executed with TopNNode in the BE.
+ * Creates a new SortNode with a limit that is executed with either TopNNode in the BE
+ * or SortNode in the backend dependent on TOPN_BYTES_LIMIT.
*/
- public static SortNode createTopNSortNode(
- PlanNodeId id, PlanNode input, SortInfo info, long offset) {
- return new SortNode(id, input, info, offset, TSortType.TOPN);
+ public static SortNode createTopNSortNode(TQueryOptions queryOptions,
+ PlanNodeId id, PlanNode input, SortInfo info, long offset, long limit,
+ boolean includeTies) {
+ long topNBytesLimit = queryOptions.topn_bytes_limit;
+ long topNCardinality = capCardinalityAtLimit(input.cardinality_, limit);
+ long estimatedTopNMaterializedSize =
+ info.estimateTopNMaterializedSize(topNCardinality, offset);
+
+ SortNode result;
+ if (topNBytesLimit <= 0 || estimatedTopNMaterializedSize < topNBytesLimit
+ || includeTies) {
+ result = new SortNode(id, input, info, offset, limit, includeTies, TSortType.TOPN);
+ } else {
+ result = SortNode.createTotalSortNode(id, input, info, offset);
+ result.setLimit(limit);
+ }
+ return result;
}
/**
@@ -106,22 +126,34 @@ public class SortNode extends PlanNode {
*/
public static SortNode createTotalSortNode(
PlanNodeId id, PlanNode input, SortInfo info, long offset) {
- return new SortNode(id, input, info, offset, TSortType.TOTAL);
+ return new SortNode(id, input, info, offset, -1, false, TSortType.TOTAL);
}
private SortNode(
- PlanNodeId id, PlanNode input, SortInfo info, long offset, TSortType type) {
+ PlanNodeId id, PlanNode input, SortInfo info, long offset, long limit,
+ boolean includeTies, TSortType type) {
super(id, info.getSortTupleDescriptor().getId().asList(), getDisplayName(type));
+ Preconditions.checkState(offset >= 0);
+ Preconditions.checkArgument(type != TSortType.TOPN || limit >= 0);
info_ = info;
children_.add(input);
offset_ = offset;
+ includeTies_ = includeTies;
+ limitWithTies_ = includeTies ? limit : -1;
type_ = type;
+ if (!includeTies) setLimit(limit);
}
public long getOffset() { return offset_; }
public void setOffset(long offset) { offset_ = offset; }
public boolean hasOffset() { return offset_ > 0; }
public boolean isTypeTopN() { return type_ == TSortType.TOPN; }
+ public boolean isIncludeTies() { return includeTies_; }
+ // Get the limit that applies to the sort, including ties or not.
+ public long getSortLimit() {
+ return includeTies_ ? limitWithTies_ : limit_;
+ }
+
public SortInfo getSortInfo() { return info_; }
public void setInputPartition(DataPartition inputPartition) {
inputPartition_ = inputPartition;
@@ -134,20 +166,33 @@ public class SortNode extends PlanNode {
/**
* Under special cases, the planner may decide to convert a total sort into a
- * TopN sort with limit
+ * TopN sort with limit. This does the conversion to top-n if the converted
+ * node would pass the TOPN_BYTES_LIMIT check. Otherwise does not modify this node.
*/
- public void convertToTopN(long limit, List<Expr> partitioningExprs,
- Analyzer analyzer) {
+ public void tryConvertToTopN(long limit, Analyzer analyzer, boolean includeTies) {
Preconditions.checkArgument(type_ == TSortType.TOTAL);
+ Preconditions.checkState(!hasLimit());
+ Preconditions.checkState(!hasOffset());
+ long topNBytesLimit = analyzer.getQueryOptions().topn_bytes_limit;
+ long topNCardinality = capCardinalityAtLimit(getChild(0).cardinality_, limit);
+ long estimatedTopNMaterializedSize =
+ info_.estimateTopNMaterializedSize(topNCardinality, offset_);
+
+ if (topNBytesLimit > 0 && estimatedTopNMaterializedSize >= topNBytesLimit) {
+ return;
+ }
type_ = TSortType.TOPN;
displayName_ = getDisplayName(type_);
- setLimit(limit);
- partitioningExprs_ = partitioningExprs;
+ includeTies_ = includeTies;
+ if (includeTies) {
+ unsetLimit();
+ limitWithTies_ = limit;
+ } else {
+ setLimit(limit);
+ }
computeStats(analyzer);
}
- public List<Expr> getPartitioningExprs() { return partitioningExprs_ ; }
-
@Override
public boolean allowPartitioned() {
if (isAnalyticSort_ && hasLimit()) return true;
@@ -203,7 +248,12 @@ public class SortNode extends PlanNode {
@Override
protected void computeStats(Analyzer analyzer) {
super.computeStats(analyzer);
- cardinality_ = capCardinalityAtLimit(getChild(0).cardinality_);
+ if (includeTies_) {
+ cardinality_ = capCardinalityAtLimit(getChild(0).cardinality_, limitWithTies_);
+ } else {
+ cardinality_ = capCardinalityAtLimit(getChild(0).cardinality_);
+ }
+
if (LOG.isTraceEnabled()) {
LOG.trace("stats Sort: cardinality=" + Long.toString(cardinality_));
}
@@ -221,13 +271,19 @@ public class SortNode extends PlanNode {
.add("is_asc", "[" + Joiner.on(" ").join(strings) + "]")
.add("nulls_first", "[" + Joiner.on(" ").join(info_.getNullsFirst()) + "]")
.add("offset_", offset_)
+ .add("includeTies_", includeTies_)
+ .add("limitWithTies_", limitWithTies_)
.addValue(super.debugString())
.toString();
}
@Override
protected void toThrift(TPlanNode msg) {
- Preconditions.checkState(!isTypeTopN() || hasLimit(), "Top-N must have limit");
+ Preconditions.checkState(!isTypeTopN() || hasLimit() ||
+ (includeTies_ && limitWithTies_ >= 0), "Top-N must have limit",
+ debugString());
+ Preconditions.checkState(!includeTies_ || (!hasLimit() && limitWithTies_ >= 0),
+ "Top-N with tie handling must set limitWithTies_ only");
Preconditions.checkState(offset_ >= 0);
msg.node_type = TPlanNodeType.SORT_NODE;
TSortInfo sort_info = new TSortInfo(Expr.treesToThrift(info_.getSortExprs()),
@@ -239,6 +295,10 @@ public class SortNode extends PlanNode {
TSortNode sort_node = new TSortNode(sort_info, type_);
sort_node.setOffset(offset_);
sort_node.setEstimated_full_input_size(estimatedFullInputSize_);
+ sort_node.setInclude_ties(includeTies_);
+ if (includeTies_) {
+ sort_node.setLimit_with_ties(limitWithTies_);
+ }
msg.sort_node = sort_node;
}
@@ -253,6 +313,9 @@ public class SortNode extends PlanNode {
output.append(getSortingOrderExplainString(info_.getSortExprs(),
info_.getIsAscOrder(), info_.getNullsFirstParams(), info_.getSortingOrder(),
info_.getNumLexicalKeysInZOrder()));
+ if (includeTies_) {
+ output.append(detailPrefix + "limit with ties: " + limitWithTies_ + "\n");
+ }
}
if (detailLevel.ordinal() >= TExplainLevel.EXTENDED.ordinal()) {
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/limit-pushdown-analytic.test b/testdata/workloads/functional-planner/queries/PlannerTest/limit-pushdown-analytic.test
index 137243b..a6cdaa4 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/limit-pushdown-analytic.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/limit-pushdown-analytic.test
@@ -16,18 +16,19 @@ PLAN-ROOT SINK
|
03:SELECT
| predicates: rank() <= 10
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
02:ANALYTIC
| functions: rank()
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -45,59 +46,61 @@ PLAN-ROOT SINK
|
03:SELECT
| predicates: rank() <= 10
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
02:ANALYTIC
| functions: rank()
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-06:TOP-N [LIMIT=10]
+06:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
05:EXCHANGE [HASH(int_col)]
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
row-size=14B cardinality=11.00K
====
-# row_number() predicate on equality instead of range.
+# row_number() predicate
# Limit pushdown into analytic sort should be applied.
select * from (
select int_col, bigint_col, smallint_col,
row_number() over (partition by int_col order by smallint_col desc) rk
from functional.alltypesagg) dt
-where rk = 5
+where rk <= 10
order by int_col, bigint_col, smallint_col, rk
-limit 10
+limit 5
---- PLAN
PLAN-ROOT SINK
|
-04:TOP-N [LIMIT=10]
+04:TOP-N [LIMIT=5]
| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk ASC
-| row-size=22B cardinality=1
+| row-size=22B cardinality=5
|
03:SELECT
-| predicates: row_number() = 5
-| row-size=22B cardinality=1
+| predicates: row_number() <= 10
+| row-size=22B cardinality=15
|
02:ANALYTIC
| functions: row_number()
| partition by: int_col
| order by: smallint_col DESC
| window: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=15
|
-01:TOP-N [LIMIT=10]
+01:TOP-N [LIMIT=15]
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| row-size=14B cardinality=15
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -107,32 +110,32 @@ PLAN-ROOT SINK
|
07:MERGING-EXCHANGE [UNPARTITIONED]
| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk ASC
-| limit: 10
+| limit: 5
|
-04:TOP-N [LIMIT=10]
+04:TOP-N [LIMIT=5]
| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk ASC
-| row-size=22B cardinality=1
+| row-size=22B cardinality=5
|
03:SELECT
-| predicates: row_number() = 5
-| row-size=22B cardinality=1
+| predicates: row_number() <= 10
+| row-size=22B cardinality=15
|
02:ANALYTIC
| functions: row_number()
| partition by: int_col
| order by: smallint_col DESC
| window: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=15
|
-06:TOP-N [LIMIT=10]
+06:TOP-N [LIMIT=15]
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| row-size=14B cardinality=15
|
05:EXCHANGE [HASH(int_col)]
|
-01:TOP-N [LIMIT=10]
+01:TOP-N [LIMIT=15]
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| row-size=14B cardinality=15
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -157,18 +160,19 @@ PLAN-ROOT SINK
|
03:SELECT
| predicates: rank() <= 10
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
02:ANALYTIC
| functions: rank()
| partition by: int_col, bigint_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, bigint_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -186,24 +190,26 @@ PLAN-ROOT SINK
|
03:SELECT
| predicates: rank() <= 10
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
02:ANALYTIC
| functions: rank()
| partition by: int_col, bigint_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-06:TOP-N [LIMIT=10]
+06:TOP-N
| order by: int_col ASC NULLS LAST, bigint_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
05:EXCHANGE [HASH(int_col,bigint_col)]
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, bigint_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -289,11 +295,12 @@ PLAN-ROOT SINK
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -314,17 +321,19 @@ PLAN-ROOT SINK
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-05:TOP-N [LIMIT=10]
+05:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
-04:EXCHANGE [HASH(int_col,smallint_col)]
+04:EXCHANGE [HASH(int_col)]
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -336,7 +345,7 @@ select * from (
select int_col, bigint_col, smallint_col,
rank() over (partition by int_col order by smallint_col desc) rk
from functional.alltypesagg) dt
-where 10 > rk
+where 11 > rk
order by int_col, bigint_col, smallint_col, rk
limit 10
---- PLAN
@@ -347,19 +356,20 @@ PLAN-ROOT SINK
| row-size=22B cardinality=10
|
03:SELECT
-| predicates: rank() < 10
-| row-size=22B cardinality=10
+| predicates: rank() < 11
+| row-size=22B cardinality=20
|
02:ANALYTIC
| functions: rank()
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -376,25 +386,93 @@ PLAN-ROOT SINK
| row-size=22B cardinality=10
|
03:SELECT
-| predicates: rank() < 10
-| row-size=22B cardinality=10
+| predicates: rank() < 11
+| row-size=22B cardinality=20
|
02:ANALYTIC
| functions: rank()
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-06:TOP-N [LIMIT=10]
+06:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
05:EXCHANGE [HASH(int_col)]
|
-01:TOP-N [LIMIT=10]
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
+|
+00:SCAN HDFS [functional.alltypesagg]
+ HDFS partitions=11/11 files=11 size=814.73KB
+ row-size=14B cardinality=11.00K
+====
+# IMPALA-10296: Limit pushdown should not be applied
+# when the limit on rank() is smaller than the limit on the order by.
+select * from (
+ select int_col, bigint_col, smallint_col,
+ rank() over (partition by int_col order by smallint_col desc) rk
+ from functional.alltypesagg) dt
+where rk < 10
+order by int_col, bigint_col, smallint_col, rk
+limit 10
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=10]
+| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk ASC
+| row-size=22B cardinality=10
+|
+03:SELECT
+| predicates: rank() < 10
+| row-size=22B cardinality=1.10K
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: int_col
+| order by: smallint_col DESC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=22B cardinality=11.00K
+|
+01:SORT
+| order by: int_col ASC NULLS LAST, smallint_col DESC
+| row-size=14B cardinality=11.00K
+|
+00:SCAN HDFS [functional.alltypesagg]
+ HDFS partitions=11/11 files=11 size=814.73KB
+ row-size=14B cardinality=11.00K
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+06:MERGING-EXCHANGE [UNPARTITIONED]
+| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk ASC
+| limit: 10
+|
+04:TOP-N [LIMIT=10]
+| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk ASC
+| row-size=22B cardinality=10
+|
+03:SELECT
+| predicates: rank() < 10
+| row-size=22B cardinality=1.10K
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: int_col
+| order by: smallint_col DESC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=22B cardinality=11.00K
+|
+01:SORT
+| order by: int_col ASC NULLS LAST, smallint_col DESC
+| row-size=14B cardinality=11.00K
+|
+05:EXCHANGE [HASH(int_col)]
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -420,18 +498,19 @@ PLAN-ROOT SINK
|
04:SELECT
| predicates: rank() <= 10
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
03:ANALYTIC
| functions: rank()
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=10
+| row-size=22B cardinality=20
|
-02:TOP-N [LIMIT=10]
+02:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=10
+| limit with ties: 20
+| row-size=14B cardinality=20
|
01:AGGREGATE [FINALIZE]
| group by: int_col, bigint_col, smallint_col
@@ -493,18 +572,19 @@ PLAN-ROOT SINK
|
03:SELECT
| predicates: rank() <= 20
-| row-size=22B cardinality=1.10K
+| row-size=22B cardinality=30
|
02:ANALYTIC
| functions: rank()
| partition by: int_col
| order by: smallint_col DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
-| row-size=22B cardinality=11.00K
+| row-size=22B cardinality=30
|
-01:SORT
+01:TOP-N
| order by: int_col ASC NULLS LAST, smallint_col DESC
-| row-size=14B cardinality=11.00K
+| limit with ties: 30
+| row-size=14B cardinality=30
|
00:SCAN HDFS [functional.alltypesagg]
HDFS partitions=11/11 files=11 size=814.73KB
@@ -658,26 +738,74 @@ PLAN-ROOT SINK
row-size=14B cardinality=11.00K
====
# 2 ranking functions with different partition-by exprs. Predicate is on
-# the dense_rank() function. If the dense_rank() appears as the top
-# level analytic function, the limit pushdown can be applied. Otherwise,
-# it should not.
+# the row_number() function. If the row_number() appears as the top
+# level analytic function, like in this example, the limit pushdown can be
+# applied. Otherwise, it should not.
select * from (
select int_col, bigint_col, smallint_col,
- dense_rank() over (partition by bigint_col order by smallint_col desc) rk2,
+ rank() over (partition by int_col order by smallint_col desc) rk1,
+ row_number() over (partition by bigint_col order by smallint_col desc) rk2
+ from functional.alltypesagg) dt
+where rk2 <= 10
+order by bigint_col, int_col, smallint_col, rk1, rk2
+limit 10
+---- PLAN
+PLAN-ROOT SINK
+|
+06:TOP-N [LIMIT=10]
+| order by: bigint_col ASC, int_col ASC, smallint_col ASC, rk1 ASC, rk2 ASC
+| row-size=30B cardinality=10
+|
+05:SELECT
+| predicates: row_number() <= 10
+| row-size=30B cardinality=20
+|
+04:ANALYTIC
+| functions: row_number()
+| partition by: bigint_col
+| order by: smallint_col DESC
+| window: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=30B cardinality=20
+|
+03:TOP-N [LIMIT=20]
+| order by: bigint_col ASC NULLS LAST, smallint_col DESC
+| row-size=22B cardinality=20
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: int_col
+| order by: smallint_col DESC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=22B cardinality=11.00K
+|
+01:SORT
+| order by: int_col ASC NULLS LAST, smallint_col DESC
+| row-size=14B cardinality=11.00K
+|
+00:SCAN HDFS [functional.alltypesagg]
+ HDFS partitions=11/11 files=11 size=814.73KB
+ row-size=14B cardinality=11.00K
+====
+# 2 ranking functions with different partition-by exprs. Predicate is on
+# the row_number() function. row_number() is not the top analytic function
+# here, so limit pushdown cannot be applied.
+select * from (
+ select int_col, bigint_col, smallint_col,
+ row_number() over (partition by bigint_col order by smallint_col desc) rk2,
rank() over (partition by int_col order by smallint_col desc) rk1
from functional.alltypesagg) dt
where rk2 <= 10
-order by int_col, bigint_col, smallint_col, rk1, rk2
+order by bigint_col, int_col, smallint_col, rk1, rk2
limit 10
---- PLAN
PLAN-ROOT SINK
|
06:TOP-N [LIMIT=10]
-| order by: int_col ASC, bigint_col ASC, smallint_col ASC, rk1 ASC, rk2 ASC
+| order by: bigint_col ASC, int_col ASC, smallint_col ASC, rk1 ASC, rk2 ASC
| row-size=30B cardinality=10
|
05:SELECT
-| predicates: dense_rank() <= 10
+| predicates: row_number() <= 10
| row-size=30B cardinality=1.10K
|
04:ANALYTIC
@@ -692,10 +820,10 @@ PLAN-ROOT SINK
| row-size=22B cardinality=11.00K
|
02:ANALYTIC
-| functions: dense_rank()
+| functions: row_number()
| partition by: bigint_col
| order by: smallint_col DESC
-| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| window: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
| row-size=22B cardinality=11.00K
|
01:SORT
@@ -939,3 +1067,210 @@ PLAN-ROOT SINK
HDFS partitions=4/24 files=4 size=34.10KB
row-size=24B cardinality=2.17K
====
+# Very high rank() predicate is not pushed down because the limit would not fit in
+# a 32-bit integer.
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by string_col) rnk
+ from functional.alltypestiny) v
+where rnk <= 9999999999
+order by tinyint_col, string_col desc, id desc
+limit 6
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=6]
+| order by: tinyint_col ASC, string_col DESC, id DESC
+| row-size=26B cardinality=1
+|
+03:SELECT
+| predicates: rank() <= 9999999999
+| row-size=26B cardinality=1
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: tinyint_col
+| order by: string_col ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=26B cardinality=8
+|
+01:SORT
+| order by: tinyint_col ASC NULLS LAST, string_col ASC
+| row-size=18B cardinality=8
+|
+00:SCAN HDFS [functional.alltypestiny]
+ HDFS partitions=4/4 files=4 size=460B
+ row-size=18B cardinality=8
+====
+# Very high rank() predicate is pushed down because the limit fits in
+# a 32-bit integer and the top-n bytes limit does not kick in.
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by string_col) rnk
+ from functional.alltypestiny) v
+where rnk <= 999999999
+order by tinyint_col, string_col desc, id desc
+limit 6
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=6]
+| order by: tinyint_col ASC, string_col DESC, id DESC
+| row-size=26B cardinality=6
+|
+03:SELECT
+| predicates: rank() <= 999999999
+| row-size=26B cardinality=8
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: tinyint_col
+| order by: string_col ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=26B cardinality=8
+|
+01:TOP-N
+| order by: tinyint_col ASC NULLS LAST, string_col ASC
+| limit with ties: 1000000005
+| row-size=18B cardinality=8
+|
+00:SCAN HDFS [functional.alltypestiny]
+ HDFS partitions=4/4 files=4 size=460B
+ row-size=18B cardinality=8
+====
+# rank() predicate is not pushed down because TOPN_BYTES_LIMIT prevents conversion
+# to top-n node.
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by string_col) rnk
+ from functional.alltypesagg) v
+where rnk <= 9999
+order by tinyint_col, string_col desc, id desc
+limit 6
+---- QUERYOPTIONS
+topn_bytes_limit=1000
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=6]
+| order by: tinyint_col ASC, string_col DESC, id DESC
+| row-size=28B cardinality=6
+|
+03:SELECT
+| predicates: rank() <= 9999
+| row-size=28B cardinality=11.00K
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: tinyint_col
+| order by: string_col ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=28B cardinality=11.00K
+|
+01:SORT
+| order by: tinyint_col ASC NULLS LAST, string_col ASC
+| row-size=20B cardinality=11.00K
+|
+00:SCAN HDFS [functional.alltypesagg]
+ HDFS partitions=11/11 files=11 size=814.73KB
+ row-size=20B cardinality=11.00K
+====
+# Limit cannot be pushed down past a dense_rank() predicate.
+select tinyint_col, string_col, id, rnk from (
+ select *, dense_rank() over (partition by tinyint_col order by string_col) rnk
+ from functional.alltypestiny) v
+where rnk <= 10
+order by tinyint_col, string_col desc, id desc
+limit 10
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=10]
+| order by: tinyint_col ASC, string_col DESC, id DESC
+| row-size=26B cardinality=1
+|
+03:SELECT
+| predicates: dense_rank() <= 10
+| row-size=26B cardinality=1
+|
+02:ANALYTIC
+| functions: dense_rank()
+| partition by: tinyint_col
+| order by: string_col ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=26B cardinality=8
+|
+01:SORT
+| order by: tinyint_col ASC NULLS LAST, string_col ASC
+| row-size=18B cardinality=8
+|
+00:SCAN HDFS [functional.alltypestiny]
+ HDFS partitions=4/4 files=4 size=460B
+ row-size=18B cardinality=8
+====
+# Limit cannot be pushed down past = predicates, except for comparison to 1 where
+# it is equivalent to <= 1
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by id) rnk
+ from functional.alltypestiny) v
+where rnk = 2
+order by tinyint_col, string_col desc, id desc
+limit 2
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=2]
+| order by: tinyint_col ASC, string_col DESC, id DESC
+| row-size=26B cardinality=2
+|
+03:SELECT
+| predicates: rank() = 2
+| row-size=26B cardinality=8
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: tinyint_col
+| order by: id ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=26B cardinality=8
+|
+01:SORT
+| order by: tinyint_col ASC NULLS LAST, id ASC
+| row-size=18B cardinality=8
+|
+00:SCAN HDFS [functional.alltypestiny]
+ HDFS partitions=4/4 files=4 size=460B
+ row-size=18B cardinality=8
+====
+# Limit cannot be pushed down past = predicates, except for comparison to 1 where
+# it is equivalent to <= 1
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by id) rnk
+ from functional.alltypestiny) v
+where rnk = 1
+order by tinyint_col, string_col desc, id desc
+limit 1
+---- PLAN
+PLAN-ROOT SINK
+|
+04:TOP-N [LIMIT=1]
+| order by: tinyint_col ASC, string_col DESC, id DESC
+| row-size=26B cardinality=1
+|
+03:SELECT
+| predicates: rank() = 1
+| row-size=26B cardinality=1
+|
+02:ANALYTIC
+| functions: rank()
+| partition by: tinyint_col
+| order by: id ASC
+| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
+| row-size=26B cardinality=2
+|
+01:TOP-N
+| order by: tinyint_col ASC NULLS LAST, id ASC
+| limit with ties: 2
+| row-size=18B cardinality=2
+|
+00:SCAN HDFS [functional.alltypestiny]
+ HDFS partitions=4/4 files=4 size=460B
+ row-size=18B cardinality=8
+====
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/resource-requirements.test b/testdata/workloads/functional-planner/queries/PlannerTest/resource-requirements.test
index 9d97d5d..30159a5 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/resource-requirements.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/resource-requirements.test
@@ -6122,7 +6122,7 @@ PLAN-ROOT SINK
03:SELECT
| predicates: rank() <= CAST(100 AS BIGINT)
| mem-estimate=0B mem-reservation=0B thread-reservation=0
-| tuple-ids=8,7 row-size=32B cardinality=100
+| tuple-ids=8,7 row-size=32B cardinality=200
| in pipelines: 01(GETNEXT)
|
02:ANALYTIC
@@ -6131,13 +6131,14 @@ PLAN-ROOT SINK
| order by: l_orderkey DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0
-| tuple-ids=8,7 row-size=32B cardinality=100
+| tuple-ids=8,7 row-size=32B cardinality=200
| in pipelines: 01(GETNEXT)
|
-01:TOP-N [LIMIT=100]
+01:TOP-N
| order by: l_partkey ASC NULLS LAST, l_orderkey DESC
-| mem-estimate=2.34KB mem-reservation=0B thread-reservation=0
-| tuple-ids=8 row-size=24B cardinality=100
+| limit with ties: 200
+| mem-estimate=4.69KB mem-reservation=0B thread-reservation=0
+| tuple-ids=8 row-size=24B cardinality=200
| in pipelines: 01(GETNEXT), 00(OPEN)
|
00:SCAN HDFS [tpch_parquet.lineitem]
@@ -6181,7 +6182,7 @@ Per-Host Resources: mem-estimate=4.02MB mem-reservation=4.00MB thread-reservatio
03:SELECT
| predicates: rank() <= CAST(100 AS BIGINT)
| mem-estimate=0B mem-reservation=0B thread-reservation=0
-| tuple-ids=8,7 row-size=32B cardinality=100
+| tuple-ids=8,7 row-size=32B cardinality=200
| in pipelines: 06(GETNEXT)
|
02:ANALYTIC
@@ -6190,26 +6191,28 @@ Per-Host Resources: mem-estimate=4.02MB mem-reservation=4.00MB thread-reservatio
| order by: l_orderkey DESC
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0
-| tuple-ids=8,7 row-size=32B cardinality=100
+| tuple-ids=8,7 row-size=32B cardinality=200
| in pipelines: 06(GETNEXT)
|
-06:TOP-N [LIMIT=100]
+06:TOP-N
| order by: l_partkey ASC NULLS LAST, l_orderkey DESC
-| mem-estimate=2.34KB mem-reservation=0B thread-reservation=0
-| tuple-ids=8 row-size=24B cardinality=100
+| limit with ties: 200
+| mem-estimate=4.69KB mem-reservation=0B thread-reservation=0
+| tuple-ids=8 row-size=24B cardinality=200
| in pipelines: 06(GETNEXT), 01(OPEN)
|
05:EXCHANGE [HASH(l_partkey)]
-| mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
-| tuple-ids=8 row-size=24B cardinality=100
+| mem-estimate=17.97KB mem-reservation=0B thread-reservation=0
+| tuple-ids=8 row-size=24B cardinality=200
| in pipelines: 01(GETNEXT)
|
F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=80.00MB mem-reservation=16.00MB thread-reservation=2
-01:TOP-N [LIMIT=100]
+01:TOP-N
| order by: l_partkey ASC NULLS LAST, l_orderkey DESC
-| mem-estimate=2.34KB mem-reservation=0B thread-reservation=0
-| tuple-ids=8 row-size=24B cardinality=100
+| limit with ties: 200
+| mem-estimate=4.69KB mem-reservation=0B thread-reservation=0
+| tuple-ids=8 row-size=24B cardinality=200
| in pipelines: 01(GETNEXT), 00(OPEN)
|
00:SCAN HDFS [tpch_parquet.lineitem, RANDOM]
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds/tpcds-q67.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds/tpcds-q67.test
index ae96817..bc8328f 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds/tpcds-q67.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds/tpcds-q67.test
@@ -58,7 +58,7 @@ PLAN-ROOT SINK
11:SELECT
| predicates: rank() <= CAST(100 AS BIGINT)
| mem-estimate=0B mem-reservation=0B thread-reservation=0
-| tuple-ids=25,24 row-size=100B cardinality=100
+| tuple-ids=25,24 row-size=100B cardinality=200
| in pipelines: 09(GETNEXT)
|
10:ANALYTIC
@@ -67,13 +67,14 @@ PLAN-ROOT SINK
| order by: aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 6 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 7 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 8 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 9 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 10 THEN su [...]
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0
-| tuple-ids=25,24 row-size=100B cardinality=100
+| tuple-ids=25,24 row-size=100B cardinality=200
| in pipelines: 09(GETNEXT)
|
-09:TOP-N [LIMIT=100]
+09:TOP-N
| order by: CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END ASC NULLS LAST, aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quan [...]
-| mem-estimate=8.98KB mem-reservation=0B thread-reservation=0
-| tuple-ids=25 row-size=92B cardinality=100
+| limit with ties: 200
+| mem-estimate=17.97KB mem-reservation=0B thread-reservation=0
+| tuple-ids=25 row-size=92B cardinality=200
| in pipelines: 09(GETNEXT), 08(OPEN)
|
08:AGGREGATE [FINALIZE]
@@ -173,10 +174,10 @@ PLAN-ROOT SINK
| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpcds_parquet.store_sales]
- HDFS partitions=1824/1824 files=1824 size=200.95MB
+ HDFS partitions=1824/1824 files=1824 size=196.95MB
runtime filters: RF000[bloom] -> ss_item_sk, RF002[bloom] -> ss_store_sk, RF004[bloom] -> ss_sold_date_sk
stored statistics:
- table: rows=2.88M size=200.95MB
+ table: rows=2.88M size=196.95MB
partitions: 1824/1824 rows=2.88M
columns: all
extrapolated-rows=disabled max-scan-range-rows=130.09K
@@ -200,7 +201,7 @@ PLAN-ROOT SINK
| in pipelines: 12(GETNEXT)
|
F05:PLAN FRAGMENT [HASH(CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END)] hosts=3 instances=3
-Per-Host Resources: mem-estimate=4.04MB mem-reservation=4.00MB thread-reservation=1
+Per-Host Resources: mem-estimate=4.08MB mem-reservation=4.00MB thread-reservation=1
12:TOP-N [LIMIT=100]
| order by: i_category ASC, i_class ASC, i_brand ASC, i_product_name ASC, d_year ASC, d_qoy ASC, d_moy ASC, s_store_id ASC, sumsales ASC, rk ASC
| mem-estimate=9.38KB mem-reservation=0B thread-reservation=0
@@ -210,7 +211,7 @@ Per-Host Resources: mem-estimate=4.04MB mem-reservation=4.00MB thread-reservatio
11:SELECT
| predicates: rank() <= CAST(100 AS BIGINT)
| mem-estimate=0B mem-reservation=0B thread-reservation=0
-| tuple-ids=25,24 row-size=100B cardinality=100
+| tuple-ids=25,24 row-size=100B cardinality=200
| in pipelines: 19(GETNEXT)
|
10:ANALYTIC
@@ -219,26 +220,28 @@ Per-Host Resources: mem-estimate=4.04MB mem-reservation=4.00MB thread-reservatio
| order by: aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 6 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 7 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 8 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 9 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 10 THEN su [...]
| window: RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0
-| tuple-ids=25,24 row-size=100B cardinality=100
+| tuple-ids=25,24 row-size=100B cardinality=200
| in pipelines: 19(GETNEXT)
|
-19:TOP-N [LIMIT=100]
+19:TOP-N
| order by: CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END ASC NULLS LAST, aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quan [...]
-| mem-estimate=8.98KB mem-reservation=0B thread-reservation=0
-| tuple-ids=25 row-size=92B cardinality=100
+| limit with ties: 200
+| mem-estimate=17.97KB mem-reservation=0B thread-reservation=0
+| tuple-ids=25 row-size=92B cardinality=200
| in pipelines: 19(GETNEXT), 09(OPEN)
|
18:EXCHANGE [HASH(CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END)]
-| mem-estimate=31.12KB mem-reservation=0B thread-reservation=0
-| tuple-ids=25 row-size=92B cardinality=100
+| mem-estimate=62.24KB mem-reservation=0B thread-reservation=0
+| tuple-ids=25 row-size=92B cardinality=200
| in pipelines: 09(GETNEXT)
|
F04:PLAN FRAGMENT [HASH(CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN murmur_hash(i_category) WHEN 5 THEN murmur_hash(i_category) WHEN 6 THEN murmur_hash(i_category) WHEN 7 THEN murmur_hash(i_category) WHEN 8 THEN murmur_hash(i_category) WHEN 9 THEN murmur_hash(i_category) WHEN 10 THEN murmur_hash(i_category) WHEN 11 THEN murmur_hash(i_category) WHEN 12 THEN murmur_hash(NULL) END,CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN murmur_hash(i_class) WHEN 5 THEN murmur_hash(i_class) WHE [...]
Per-Host Resources: mem-estimate=43.90GB mem-reservation=244.75MB thread-reservation=1
-09:TOP-N [LIMIT=100]
+09:TOP-N
| order by: CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN i_category WHEN 5 THEN i_category WHEN 6 THEN i_category WHEN 7 THEN i_category WHEN 8 THEN i_category WHEN 9 THEN i_category WHEN 10 THEN i_category WHEN 11 THEN i_category WHEN 12 THEN NULL END ASC NULLS LAST, aggif(valid_tid(4,5,6,7,8,9,10,11,12) IN (4, 5, 6, 7, 8, 9, 10, 11, 12), CASE valid_tid(4,5,6,7,8,9,10,11,12) WHEN 4 THEN sum(coalesce(ss_sales_price * ss_quantity, 0)) WHEN 5 THEN sum(coalesce(ss_sales_price * ss_quan [...]
-| mem-estimate=8.98KB mem-reservation=0B thread-reservation=0
-| tuple-ids=25 row-size=92B cardinality=100
+| limit with ties: 200
+| mem-estimate=17.97KB mem-reservation=0B thread-reservation=0
+| tuple-ids=25 row-size=92B cardinality=200
| in pipelines: 09(GETNEXT), 08(OPEN)
|
08:AGGREGATE [FINALIZE]
@@ -398,10 +401,10 @@ Per-Host Resources: mem-estimate=8.96GB mem-reservation=221.75MB thread-reservat
| in pipelines: 01(GETNEXT)
|
00:SCAN HDFS [tpcds_parquet.store_sales, RANDOM]
- HDFS partitions=1824/1824 files=1824 size=200.95MB
+ HDFS partitions=1824/1824 files=1824 size=196.95MB
runtime filters: RF000[bloom] -> ss_item_sk, RF002[bloom] -> ss_store_sk, RF004[bloom] -> ss_sold_date_sk
stored statistics:
- table: rows=2.88M size=200.95MB
+ table: rows=2.88M size=196.95MB
partitions: 1824/1824 rows=2.88M
columns: all
extrapolated-rows=disabled max-scan-range-rows=130.09K
diff --git a/testdata/workloads/functional-query/queries/limit-pushdown-analytic.test b/testdata/workloads/functional-query/queries/limit-pushdown-analytic.test
new file mode 100644
index 0000000..0554932
--- /dev/null
+++ b/testdata/workloads/functional-query/queries/limit-pushdown-analytic.test
@@ -0,0 +1,96 @@
+====
+---- QUERY
+# IMPALA-10296: this query returns correct results even before
+# IMPALA-10296 because the limit is not hit. Reducing the limit
+# should simply truncate the results because the ordering is total.
+select tinyint_col, id from (
+ select *, rank() over (partition by tinyint_col order by id) rnk
+ from alltypestiny) v
+where rnk < 4
+order by tinyint_col, id desc
+limit 10
+---- TYPES
+TINYINT,INT
+---- RESULTS
+0,4
+0,2
+0,0
+1,5
+1,3
+1,1
+====
+---- QUERY
+# IMPALA-10296: it is not safe to push the limit down past the analytic.
+# This query reproduced incorrect results prior to the IMPALA-10296 fix.
+select tinyint_col, id from (
+ select *, rank() over (partition by tinyint_col order by id) rnk
+ from alltypestiny) v
+where rnk < 4
+order by tinyint_col, id desc
+limit 5
+---- TYPES
+TINYINT,INT
+---- RESULTS
+0,4
+0,2
+0,0
+1,5
+1,3
+====
+---- QUERY
+# IMPALA-10296: this query returns correct results even before
+# IMPALA-10296 because the limit is not hit. Reducing the limit
+# should simply truncate the results because the ordering is total.
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by string_col) rnk
+ from alltypestiny) v
+where rnk <= 5
+order by tinyint_col, string_col desc, id desc
+limit 10
+---- TYPES
+TINYINT,STRING,INT,BIGINT
+---- RESULTS
+0,'0',6,1
+0,'0',4,1
+0,'0',2,1
+0,'0',0,1
+1,'1',7,1
+1,'1',5,1
+1,'1',3,1
+1,'1',1,1
+====
+---- QUERY
+# IMPALA-10296: the limit can be pushed past the analytic operator,
+# but we need to increase the limit and include ties to guarantee
+# correct results.
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by string_col) rnk
+ from alltypestiny) v
+where rnk <= 5
+order by tinyint_col, string_col desc, id desc
+limit 5
+---- TYPES
+TINYINT,STRING,INT,BIGINT
+---- RESULTS
+0,'0',6,1
+0,'0',4,1
+0,'0',2,1
+0,'0',0,1
+1,'1',7,1
+====
+---- QUERY
+# IMPALA-10296: cannot push limit through an equality comparison to
+# rank the predicate can filter out an arbitrary number of rows returned
+# from the top-n sort.
+select tinyint_col, string_col, id, rnk from (
+ select *, rank() over (partition by tinyint_col order by id) rnk
+ from alltypestiny) v
+where rnk = 2
+order by tinyint_col, string_col desc, id desc
+limit 2
+---- TYPES
+TINYINT,STRING,INT,BIGINT
+---- RESULTS
+0,'0',2,2
+1,'1',3,2
+====
diff --git a/testdata/workloads/tpch/queries/limit-pushdown-analytic.test b/testdata/workloads/tpch/queries/limit-pushdown-analytic.test
index 4bd8b8b..5dc1635 100644
--- a/testdata/workloads/tpch/queries/limit-pushdown-analytic.test
+++ b/testdata/workloads/tpch/queries/limit-pushdown-analytic.test
@@ -89,3 +89,152 @@ limit 3
---- TYPES
INT, INT, BIGINT
====
+---- QUERY
+# IMPALA-10296: the limit pushdown through rank() requires tie handling in
+# the top-n. This query is designed to exercise that by using an analytic
+# order by with low cardinality and therefore many ties. The query returns
+# values from two analytic partitions.
+select l_orderkey, l_partkey, l_suppkey, l_linenumber, l_shipmode, rnk from (
+ select *, rank() over (partition by l_partkey order by l_shipmode) rnk
+ from lineitem) v
+where rnk <= 50
+order by l_partkey, l_orderkey, l_suppkey, l_linenumber, l_shipmode
+limit 50
+---- RESULTS
+504065,1,2502,6,'RAIL',15
+545153,1,2,4,'SHIP',24
+599361,1,5002,7,'AIR',1
+1139363,1,2,1,'FOB',8
+1225698,1,2,5,'FOB',8
+1695463,1,5002,5,'FOB',8
+1829732,1,2,3,'RAIL',15
+1876199,1,2502,3,'AIR',1
+2265089,1,2502,4,'RAIL',15
+2301222,1,2,5,'SHIP',24
+2485411,1,2502,6,'REG AIR',21
+2535332,1,5002,4,'MAIL',12
+2603011,1,5002,2,'MAIL',12
+2698307,1,2,3,'RAIL',15
+3438019,1,2,6,'RAIL',15
+3700610,1,5002,2,'AIR',1
+3743746,1,5002,6,'SHIP',24
+4076803,1,2,4,'SHIP',24
+4168578,1,5002,3,'TRUCK',30
+4343170,1,7502,2,'SHIP',24
+4422049,1,7502,5,'AIR',1
+4450401,1,7502,1,'MAIL',12
+4464900,1,2502,1,'AIR',1
+4810753,1,5002,5,'REG AIR',21
+4912929,1,5002,5,'TRUCK',30
+5120486,1,2,1,'SHIP',24
+5352710,1,7502,2,'AIR',1
+5362629,1,5002,3,'AIR',1
+5618244,1,7502,5,'REG AIR',21
+5862498,1,2,6,'RAIL',15
+5926723,1,2,2,'FOB',8
+16135,2,5003,1,'RAIL',11
+269286,2,3,2,'RAIL',11
+437670,2,3,5,'MAIL',6
+698981,2,3,1,'TRUCK',25
+864452,2,7503,6,'TRUCK',25
+957444,2,7503,2,'TRUCK',25
+976196,2,7503,2,'RAIL',11
+1140614,2,5003,2,'SHIP',22
+1204321,2,2503,3,'MAIL',6
+1339139,2,5003,5,'REG AIR',19
+1394727,2,2503,3,'AIR',1
+2189347,2,7503,3,'TRUCK',25
+2664550,2,7503,4,'AIR',1
+2774018,2,2503,1,'RAIL',11
+2838083,2,5003,5,'SHIP',22
+2859844,2,5003,2,'AIR',1
+2974054,2,2503,5,'REG AIR',19
+3204707,2,2503,1,'MAIL',6
+3815428,2,7503,3,'TRUCK',25
+---- TYPES
+BIGINT, BIGINT, BIGINT, INT, STRING, BIGINT
+====
+---- QUERY
+# IMPALA-10296: the same query as above with a higher limit. The limit pushdown
+# optimization should not be applied and the results should be the same as above,
+# but with additional rows.
+select l_orderkey, l_partkey, l_suppkey, l_linenumber, l_shipmode, rnk from (
+ select *, rank() over (partition by l_partkey order by l_shipmode) rnk
+ from lineitem) v
+where rnk <= 50
+order by l_partkey, l_orderkey, l_suppkey, l_linenumber, l_shipmode
+limit 70
+---- RESULTS
+504065,1,2502,6,'RAIL',15
+545153,1,2,4,'SHIP',24
+599361,1,5002,7,'AIR',1
+1139363,1,2,1,'FOB',8
+1225698,1,2,5,'FOB',8
+1695463,1,5002,5,'FOB',8
+1829732,1,2,3,'RAIL',15
+1876199,1,2502,3,'AIR',1
+2265089,1,2502,4,'RAIL',15
+2301222,1,2,5,'SHIP',24
+2485411,1,2502,6,'REG AIR',21
+2535332,1,5002,4,'MAIL',12
+2603011,1,5002,2,'MAIL',12
+2698307,1,2,3,'RAIL',15
+3438019,1,2,6,'RAIL',15
+3700610,1,5002,2,'AIR',1
+3743746,1,5002,6,'SHIP',24
+4076803,1,2,4,'SHIP',24
+4168578,1,5002,3,'TRUCK',30
+4343170,1,7502,2,'SHIP',24
+4422049,1,7502,5,'AIR',1
+4450401,1,7502,1,'MAIL',12
+4464900,1,2502,1,'AIR',1
+4810753,1,5002,5,'REG AIR',21
+4912929,1,5002,5,'TRUCK',30
+5120486,1,2,1,'SHIP',24
+5352710,1,7502,2,'AIR',1
+5362629,1,5002,3,'AIR',1
+5618244,1,7502,5,'REG AIR',21
+5862498,1,2,6,'RAIL',15
+5926723,1,2,2,'FOB',8
+16135,2,5003,1,'RAIL',11
+269286,2,3,2,'RAIL',11
+437670,2,3,5,'MAIL',6
+698981,2,3,1,'TRUCK',25
+864452,2,7503,6,'TRUCK',25
+957444,2,7503,2,'TRUCK',25
+976196,2,7503,2,'RAIL',11
+1140614,2,5003,2,'SHIP',22
+1204321,2,2503,3,'MAIL',6
+1339139,2,5003,5,'REG AIR',19
+1394727,2,2503,3,'AIR',1
+2189347,2,7503,3,'TRUCK',25
+2664550,2,7503,4,'AIR',1
+2774018,2,2503,1,'RAIL',11
+2838083,2,5003,5,'SHIP',22
+2859844,2,5003,2,'AIR',1
+2974054,2,2503,5,'REG AIR',19
+3204707,2,2503,1,'MAIL',6
+3815428,2,7503,3,'TRUCK',25
+3978178,2,2503,4,'REG AIR',19
+4052995,2,3,4,'MAIL',6
+4212039,2,3,2,'FOB',4
+4291943,2,5003,5,'MAIL',6
+4325446,2,2503,6,'SHIP',22
+4563589,2,2503,2,'TRUCK',25
+4701030,2,7503,1,'RAIL',11
+4849414,2,2503,4,'TRUCK',25
+5000327,2,5003,3,'RAIL',11
+5240129,2,7503,4,'RAIL',11
+5399651,2,3,1,'TRUCK',25
+5435686,2,7503,4,'FOB',4
+5879491,2,3,4,'RAIL',11
+367685,3,7504,1,'RAIL',20
+480000,3,5004,5,'MAIL',15
+601123,3,2504,3,'AIR',1
+789220,3,7504,5,'TRUCK',34
+810723,3,4,1,'AIR',1
+951425,3,2504,6,'RAIL',20
+993152,3,4,1,'SHIP',31
+---- TYPES
+BIGINT, BIGINT, BIGINT, INT, STRING, BIGINT
+====
diff --git a/tests/query_test/test_limit_pushdown_analytic.py b/tests/query_test/test_limit_pushdown_analytic.py
index dca9c6b..caac68b 100644
--- a/tests/query_test/test_limit_pushdown_analytic.py
+++ b/tests/query_test/test_limit_pushdown_analytic.py
@@ -19,9 +19,11 @@
# of ranking functions
from tests.common.impala_test_suite import ImpalaTestSuite
+from tests.common.test_dimensions import (create_single_exec_option_dimension,
+ extend_exec_option_dimension)
-class TestLimitPushdownAnalytic(ImpalaTestSuite):
+class TestLimitPushdownAnalyticTpch(ImpalaTestSuite):
@classmethod
def get_workload(cls):
@@ -29,7 +31,25 @@ class TestLimitPushdownAnalytic(ImpalaTestSuite):
@classmethod
def add_test_dimensions(cls):
- super(TestLimitPushdownAnalytic, cls).add_test_dimensions()
+ super(TestLimitPushdownAnalyticTpch, cls).add_test_dimensions()
+ cls.ImpalaTestMatrix.add_constraint(lambda v:
+ v.get_value('table_format').file_format in ['parquet'])
+
+ def test_limit_pushdown_analytic(self, vector):
+ self.run_test_case('limit-pushdown-analytic', vector)
+
+
+class TestLimitPushdownAnalyticFunctional(ImpalaTestSuite):
+ @classmethod
+ def get_workload(cls):
+ return 'functional-query'
+
+ @classmethod
+ def add_test_dimensions(cls):
+ super(TestLimitPushdownAnalyticFunctional, cls).add_test_dimensions()
+ cls.ImpalaTestMatrix.add_dimension(create_single_exec_option_dimension())
+ # Also run with num_nodes=1 because it's easier to reproduce IMPALA-10296.
+ extend_exec_option_dimension(cls, 'num_nodes', 1)
cls.ImpalaTestMatrix.add_constraint(lambda v:
v.get_value('table_format').file_format in ['parquet'])