Hello Aman Sinha, Qifan Chen, Shant Hovsepian, David Rorke, Impala Public
Jenkins,
I'd like you to reexamine a change. Please visit
http://gerrit.cloudera.org:8080/16242
to look at the new patch set (#27).
Change subject: IMPALA-9979: part 2: partitioned top-n
......................................................................
IMPALA-9979: part 2: partitioned top-n
Planner changes:
---------------
The planner now identifies predicates that can be converted into
limits in a partitioned or unpartitioned top-n with the following
method:
* Push down predicates that reference analytic tuple into inline view.
These will be evaluated after the analytic plan for the inline
SelectStmt is generated.
* Identify predicates that reference the analytic tuple and could
be converted to limits.
* If they can be applied to the last sort group of the analytic
plan, and the windows are all compatible, then the lowest
limit gets converted into a limit in the top N.
* Otherwise generate a select node with the conjuncts. We add
logic to merge SELECT nodes to avoid generating duplicates
from inside and outside the inline view.
* The pushed predicate is still added to the SELECT node
because it is necessary for correctness for predicates
like '=' to filter additional rows and also the limit
pushdown optimization looks for analytic predicates
there, so retaining all predicates simplifies that.
The selectivity of the predicate is adjusted so that
cardinality estimates remain accurate.
The optimization can be disabled by setting
ANALYTIC_RANK_PUSHDOWN_THRESHOLD=0. By default it is
only enabled for limits of 1000 or less, because the
in-memory Top-N may perform significantly worse than
a full sort for large heaps (since updating the heap
for every input row ends up being more expensive than
doing a traditional sort). We could probably optimize
this more with better tuning so that it can gracefully
fall back to doing the full sort at runtime.
rank() and row_number() are handled. rank() needs support in
the TopN node to include ties for the last place, which is
also added in this patch.
If predicates are trivially false, we generate empty nodes.
The interacts with the limit pushdwon optimization. The limit
pushdown optimization is applied after the partitioned top-n
is generated, and can sometimes result in more optimal plans,
so it is generalized to handle pushing into partitioned top-n
nodes.
Backend changes:
---------------
The top-n node in the backend is augmented to handle both
the partitioning (for which we use a std::map and a
comparator based on the partition exprs) and the tie-handling
semantics required by rank() predicates. The partitioned
top-n node has a soft limit of 64MB on the size of the
in-memory heaps and can spill with use of an embedded Sorter.
The current implementation tries to evict heaps that are
less effective at filtering rows.
We currently use the partitioned top-n node to implement
rank() pushdown in all cases because of the tie-handling
support. We also cannot use the merging exchange for
rank() because the limit does not handle ties in the same way,
so we need to generate a hash exchange with a partitioned
top-n node on top of the exchange.
Limitations:
-----------
There are several possible extensions to this that we did not do:
* dense_rank() is not supported because it would require additional
backend support - IMPALA-10014.
* ntile() is not supported because it would require additional
backend support - IMPALA-10174.
* Only one predicate per analytic is pushed.
* Redundant rank()/row_number() predicates are not merged,
only the lowest is chosen.
* Lower bounds are not converted into OFFSET.
* The analytic operator cannot be eliminated even if the analytic
expression was only used in the predicate.
* This doesn't push predicates into UNION - IMPALA-10013
* Always false predicates don't result in empty plan - IMPALA-10015
Tests:
-----
* Planner tests - added tests that exercise the interesting code
paths added in planning.
- Predicate ordering in SELECT nodes changed in a couple of cases
because some predicates were pushed into the inline views.
* Modified SORT targeted perf tests to avoid conversion to Top-N
* Added targeted perf test for partitioned top-n.
* End-to-end tests
- Unpartitioned Top-N end-to-end tests
- Basic partitioning and duplicate handling tests on functional
- Similar basic tests on larger inputs from TPC-DS and with
larger partition counts.
- I inspected the results and also ran the same tests with
analytic_rank_pushdown_threshold=0 to confirm that the
results were the same as with the full sort.
- Fallback to spilling sort.
Perf:
-----
Added a targeted benchmark that goes from ~2s to ~1s with
mt_dop=8 on TPC-H 30 on my desktop.
Change-Id: Ic638af9495981d889a4cb7455a71e8be0eb1a8e5
---
M be/src/codegen/gen_ir_descriptions.py
M be/src/exec/exec-node.cc
M be/src/exec/topn-node-ir.cc
M be/src/exec/topn-node.cc
M be/src/exec/topn-node.h
M be/src/exprs/slot-ref.h
M be/src/service/query-options-test.cc
M be/src/service/query-options.cc
M be/src/service/query-options.h
M be/src/util/priority-queue.h
M be/src/util/tuple-row-compare.h
M common/thrift/ImpalaInternalService.thrift
M common/thrift/ImpalaService.thrift
M common/thrift/PlanNodes.thrift
M fe/src/main/java/org/apache/impala/analysis/Expr.java
M fe/src/main/java/org/apache/impala/analysis/SlotRef.java
M fe/src/main/java/org/apache/impala/analysis/SortInfo.java
M fe/src/main/java/org/apache/impala/planner/AnalyticEvalNode.java
M fe/src/main/java/org/apache/impala/planner/AnalyticPlanner.java
M fe/src/main/java/org/apache/impala/planner/DistributedPlanner.java
M fe/src/main/java/org/apache/impala/planner/PlanNode.java
M fe/src/main/java/org/apache/impala/planner/SelectNode.java
M fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
M fe/src/main/java/org/apache/impala/planner/SortNode.java
M fe/src/test/java/org/apache/impala/planner/CardinalityTest.java
M fe/src/test/java/org/apache/impala/planner/PlannerTest.java
M testdata/workloads/functional-planner/queries/PlannerTest/analytic-fns.test
A
testdata/workloads/functional-planner/queries/PlannerTest/analytic-rank-pushdown.test
A
testdata/workloads/functional-planner/queries/PlannerTest/limit-pushdown-partitioned-top-n.test
M
testdata/workloads/functional-planner/queries/PlannerTest/resource-requirements.test
M testdata/workloads/functional-planner/queries/PlannerTest/tpcds/tpcds-q44.test
M testdata/workloads/functional-planner/queries/PlannerTest/tpcds/tpcds-q67.test
M testdata/workloads/functional-planner/queries/PlannerTest/tpcds/tpcds-q70.test
A
testdata/workloads/functional-query/queries/QueryTest/analytic-fns-tpcds-partitioned-topn.test
A testdata/workloads/functional-query/queries/QueryTest/partitioned-top-n.test
M testdata/workloads/functional-query/queries/QueryTest/spilling.test
M testdata/workloads/functional-query/queries/QueryTest/top-n.test
M testdata/workloads/targeted-perf/queries/primitive_orderby_all.test
M testdata/workloads/targeted-perf/queries/primitive_orderby_bigint.test
M
testdata/workloads/targeted-perf/queries/primitive_orderby_bigint_expression.test
A testdata/workloads/targeted-perf/queries/primitive_top-n_partitioned.test
M tests/experiments/test_targeted_perf.py
M tests/query_test/test_analytic_tpcds.py
M tests/query_test/test_queries.py
44 files changed, 5,258 insertions(+), 386 deletions(-)
git pull ssh://gerrit.cloudera.org:29418/Impala-ASF refs/changes/42/16242/27
--
To view, visit http://gerrit.cloudera.org:8080/16242
To unsubscribe, visit http://gerrit.cloudera.org:8080/settings
Gerrit-Project: Impala-ASF
Gerrit-Branch: master
Gerrit-MessageType: newpatchset
Gerrit-Change-Id: Ic638af9495981d889a4cb7455a71e8be0eb1a8e5
Gerrit-Change-Number: 16242
Gerrit-PatchSet: 27
Gerrit-Owner: Tim Armstrong <tarmstr...@cloudera.com>
Gerrit-Reviewer: Aman Sinha <amsi...@cloudera.com>
Gerrit-Reviewer: David Rorke <dro...@cloudera.com>
Gerrit-Reviewer: Impala Public Jenkins <impala-public-jenk...@cloudera.com>
Gerrit-Reviewer: Qifan Chen <qc...@cloudera.com>
Gerrit-Reviewer: Shant Hovsepian <sh...@cloudera.com>
Gerrit-Reviewer: Tim Armstrong <tarmstr...@cloudera.com>
