You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by as...@apache.org on 2021/06/24 14:56:20 UTC
[impala] 04/04: IMPALA-10755: Fix migration of analytic predicate
to inline view.
This is an automated email from the ASF dual-hosted git repository.
asherman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 89131e70ff649545c7192f760ac7f8d2abcb6e37
Author: Aman Sinha <am...@cloudera.com>
AuthorDate: Mon Jun 21 23:11:37 2021 -0700
IMPALA-10755: Fix migration of analytic predicate to inline view.
As part of IMPALA-9979 we made changes to push down predicates
that reference analytic tuple into the inline view. In cases where
both sides of a predicate are slot references (for example,
a = MAX(b) where MAX(b) is an analytic function), it may not be
safe to push it into the inline view since the two sides may be
referencing separate tuples.
This patch fixes the behavior by skipping such predicates such
that they will be left unassigned and will subsequently get
assigned to a SELECT node above the analytic operator.
Testing:
- Added planner tests for analytic predicates ensuring that
analytic predicates are present in the SELECT node.
- Added run time tests for the same using TPC-H and
verified correctness.
Change-Id: Ib5cad3d408ee3695cafb35f66a4f19b4e8d0529e
Reviewed-on: http://gerrit.cloudera.org:8080/17615
Reviewed-by: Aman Sinha <am...@cloudera.com>
Tested-by: Aman Sinha <am...@cloudera.com>
---
.../apache/impala/planner/SingleNodePlanner.java | 13 ++-
.../queries/PlannerTest/analytic-fns.test | 119 ++++++++++++++++++++-
testdata/workloads/tpch/queries/analytic-fns.test | 62 +++++++++++
tests/query_test/test_queries.py | 16 +++
4 files changed, 205 insertions(+), 5 deletions(-)
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 3c25722..c5c9e28 100644
--- a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
@@ -1392,10 +1392,17 @@ public class SingleNodePlanner {
// to be evaluated earlier in the plan (which could produce incorrect results).
TupleDescriptor analyticTuple = selectStmt.getAnalyticInfo().getOutputTupleDesc();
List<Expr> analyticPreds = new ArrayList<>();
- for (int i = 0; i < conjuncts.size(); ++i) {
- Expr pred = conjuncts.get(i);
+ List<TupleId> tupleIds = new ArrayList<>();
+ for (Expr pred : conjuncts) {
Expr viewPred = pred.substitute(inlineViewRef.getSmap(), analyzer, false);
- if (viewPred.referencesTuple(analyticTuple.getId())) {
+ tupleIds.clear();
+ viewPred.getIds(tupleIds, null);
+ // Only consider analytic conjuncts which are referencing the analytic tuple
+ // and skip the ones which are referencing more than 1 tuple.
+ // Example: a = MAX(b) (where MAX(b) is an analytic function). In this case,
+ // the left side may not be referencing analytic tuple and it would not
+ // be safe to migrate such predicates into the inline view.
+ if (viewPred.referencesTuple(analyticTuple.getId()) && tupleIds.size() <= 1) {
analyticPreds.add(pred);
}
}
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/analytic-fns.test b/testdata/workloads/functional-planner/queries/PlannerTest/analytic-fns.test
index 481a38f..a01b401 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/analytic-fns.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/analytic-fns.test
@@ -1554,7 +1554,7 @@ where
PLAN-ROOT SINK
|
07:SELECT
-| predicates: avg(int_col) < 5, count(int_col) < 3, max(int_col) < 2, min(int_col) < 1, sum(int_col) < 4, bigint_col > 10, max(int_col) != count(int_col), min(int_col) != avg(int_col), min(int_col) != count(int_col), count(int_col) < bigint_col + 3, sum(int_col) < bigint_col + 4, avg(int_col) < bigint_col + 5, max(int_col) < bigint_col + 2, min(int_col) < bigint_col + 1
+| predicates: avg(int_col) < 5, count(int_col) < 3, max(int_col) < 2, min(int_col) < 1, sum(int_col) < 4, bigint_col > 10, max(int_col) != count(int_col), min(int_col) != avg(int_col), min(int_col) != count(int_col), count(int_col) < bigint_col + 3, sum(int_col) < bigint_col + 4, min(int_col) < bigint_col + 1, max(int_col) < bigint_col + 2, avg(int_col) < bigint_col + 5
| row-size=49B cardinality=73
|
06:ANALYTIC
@@ -1596,7 +1596,7 @@ PLAN-ROOT SINK
PLAN-ROOT SINK
|
07:SELECT
-| predicates: avg(int_col) < 5, count(int_col) < 3, max(int_col) < 2, min(int_col) < 1, sum(int_col) < 4, bigint_col > 10, max(int_col) != count(int_col), min(int_col) != avg(int_col), min(int_col) != count(int_col), count(int_col) < bigint_col + 3, sum(int_col) < bigint_col + 4, avg(int_col) < bigint_col + 5, max(int_col) < bigint_col + 2, min(int_col) < bigint_col + 1
+| predicates: avg(int_col) < 5, count(int_col) < 3, max(int_col) < 2, min(int_col) < 1, sum(int_col) < 4, bigint_col > 10, max(int_col) != count(int_col), min(int_col) != avg(int_col), min(int_col) != count(int_col), count(int_col) < bigint_col + 3, sum(int_col) < bigint_col + 4, min(int_col) < bigint_col + 1, max(int_col) < bigint_col + 2, avg(int_col) < bigint_col + 5
| row-size=49B cardinality=73
|
06:ANALYTIC
@@ -3381,3 +3381,118 @@ PLAN-ROOT SINK
runtime filters: RF000 -> t1.id
row-size=4B cardinality=7.30K
====
+# IMPALA-10755: Predicate on analytic function should
+# be evaluated in a SELECT node
+select *
+ FROM (
+ SELECT
+ l_quantity, l_linenumber,
+ nr_pvp,
+ max(nr_pvp) over (partition BY l_quantity) max_nr_pvp
+ FROM (
+ SELECT
+ l_quantity, l_linenumber,
+ count(1) AS nr_pvp
+ FROM
+ tpch.lineitem, tpch.partsupp
+ where l_partkey = ps_partkey
+ and ps_suppkey = 10 and l_quantity in (1, 2)
+ GROUP BY l_quantity, l_linenumber
+ ) a
+ ) b
+ WHERE nr_pvp = max_nr_pvp
+---- PLAN
+PLAN-ROOT SINK
+|
+06:SELECT
+| predicates: count(*) = max(nr_pvp)
+| row-size=28B cardinality=36
+|
+05:ANALYTIC
+| functions: max(count(*))
+| partition by: l_quantity
+| row-size=28B cardinality=357
+|
+04:SORT
+| order by: l_quantity ASC NULLS LAST
+| row-size=20B cardinality=357
+|
+03:AGGREGATE [FINALIZE]
+| output: count(*)
+| group by: l_quantity, l_linenumber
+| row-size=20B cardinality=357
+|
+02:HASH JOIN [INNER JOIN]
+| hash predicates: l_partkey = ps_partkey
+| runtime filters: RF000 <- ps_partkey
+| row-size=36B cardinality=2.45K
+|
+|--01:SCAN HDFS [tpch.partsupp]
+| HDFS partitions=1/1 files=1 size=112.71MB
+| predicates: ps_suppkey = 10
+| row-size=16B cardinality=82
+|
+00:SCAN HDFS [tpch.lineitem]
+ HDFS partitions=1/1 files=1 size=718.94MB
+ predicates: l_quantity IN (1, 2)
+ runtime filters: RF000 -> l_partkey
+ row-size=20B cardinality=235.34K
+====
+# IMPALA-10755: Equality predicate where both
+# sides refer to analytic function.
+select *
+ FROM (
+ SELECT
+ l_quantity, l_linenumber,
+ nr_pvp,
+ min(nr_pvp) over (partition BY l_quantity) min_nr_pvp,
+ max(nr_pvp) over (partition BY l_quantity) max_nr_pvp
+ FROM (
+ SELECT
+ l_quantity, l_linenumber,
+ count(1) AS nr_pvp
+ FROM
+ tpch.lineitem, tpch.partsupp
+ where l_partkey = ps_partkey
+ and ps_suppkey = 10 and l_quantity in (1, 2)
+ GROUP BY l_quantity, l_linenumber
+ ) a
+ ) b
+ WHERE min_nr_pvp = max_nr_pvp
+---- PLAN
+PLAN-ROOT SINK
+|
+06:SELECT
+| predicates: min(nr_pvp) = max(nr_pvp)
+| row-size=36B cardinality=36
+|
+05:ANALYTIC
+| functions: min(count(*)), max(count(*))
+| partition by: l_quantity
+| row-size=36B cardinality=357
+|
+04:SORT
+| order by: l_quantity ASC NULLS LAST
+| row-size=20B cardinality=357
+|
+03:AGGREGATE [FINALIZE]
+| output: count(*)
+| group by: l_quantity, l_linenumber
+| row-size=20B cardinality=357
+|
+02:HASH JOIN [INNER JOIN]
+| hash predicates: l_partkey = ps_partkey
+| runtime filters: RF000 <- ps_partkey
+| row-size=36B cardinality=2.45K
+|
+|--01:SCAN HDFS [tpch.partsupp]
+| HDFS partitions=1/1 files=1 size=112.71MB
+| predicates: ps_suppkey = 10
+| row-size=16B cardinality=82
+|
+00:SCAN HDFS [tpch.lineitem]
+ HDFS partitions=1/1 files=1 size=718.94MB
+ predicates: l_quantity IN (1, 2)
+ runtime filters: RF000 -> l_partkey
+ row-size=20B cardinality=235.34K
+====
diff --git a/testdata/workloads/tpch/queries/analytic-fns.test b/testdata/workloads/tpch/queries/analytic-fns.test
new file mode 100644
index 0000000..583aa65
--- /dev/null
+++ b/testdata/workloads/tpch/queries/analytic-fns.test
@@ -0,0 +1,62 @@
+====
+---- QUERY
+# IMPALA-10755: Predicate on analytic function should
+# be evaluated in a SELECT node
+select *
+ FROM (
+ SELECT
+ l_quantity, l_linenumber,
+ nr_pvp,
+ max(nr_pvp) over (partition BY l_quantity) max_nr_pvp
+ FROM (
+ SELECT
+ l_quantity, l_linenumber,
+ count(1) AS nr_pvp
+ FROM
+ tpch.lineitem, tpch.partsupp
+ where l_partkey = ps_partkey
+ and ps_suppkey = 10 and l_quantity in (1, 2)
+ GROUP BY l_quantity, l_linenumber
+ ) a
+ ) b
+ WHERE nr_pvp = max_nr_pvp
+---- RESULTS
+1.00,2,12,12
+2.00,1,12,12
+2.00,2,12,12
+---- TYPES
+DECIMAL, INT, BIGINT, BIGINT
+====
+---- QUERY
+# IMPALA-10755: Predicate on analytic function should
+# be evaluated in a SELECT node
+select *
+ FROM (
+ SELECT
+ l_quantity, l_linenumber,
+ nr_pvp,
+ avg(nr_pvp) over (partition BY l_quantity) avg_nr_pvp,
+ max(nr_pvp) over (partition BY l_quantity) max_nr_pvp
+ FROM (
+ SELECT
+ l_quantity, l_linenumber,
+ count(1) AS nr_pvp
+ FROM
+ tpch.lineitem, tpch.partsupp
+ where l_partkey = ps_partkey
+ and ps_suppkey = 10 and l_quantity in (1, 2)
+ GROUP BY l_quantity, l_linenumber
+ ) a
+ ) b
+ WHERE cast(avg_nr_pvp as bigint) = max_nr_pvp - 5
+---- RESULTS
+1.00,1,11,7,12
+1.00,2,12,7,12
+1.00,3,10,7,12
+1.00,4,5,7,12
+1.00,5,4,7,12
+1.00,6,5,7,12
+1.00,7,2,7,12
+---- TYPES
+DECIMAL, INT, BIGINT, DOUBLE, BIGINT
+====
diff --git a/tests/query_test/test_queries.py b/tests/query_test/test_queries.py
index e6a844a..afd7354 100644
--- a/tests/query_test/test_queries.py
+++ b/tests/query_test/test_queries.py
@@ -363,3 +363,19 @@ class TestTopNReclaimQuery(ImpalaTestSuite):
# Tuple pool is expected to be reclaimed for this query
for n in num_of_times_tuple_pool_reclaimed:
assert int(n) > 0
+
+
+class TestAnalyticFnsTpch(ImpalaTestSuite):
+
+ @classmethod
+ def get_workload(cls):
+ return 'tpch'
+
+ @classmethod
+ def add_test_dimensions(cls):
+ super(TestAnalyticFnsTpch, cls).add_test_dimensions()
+ cls.ImpalaTestMatrix.add_constraint(lambda v:
+ v.get_value('table_format').file_format in ['parquet'])
+
+ def test_analytic_predicate(self, vector):
+ self.run_test_case('analytic-fns', vector)