http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/testdata/workloads/functional-planner/queries/PlannerTest/inline-view-limit.test
----------------------------------------------------------------------
diff --git 
a/testdata/workloads/functional-planner/queries/PlannerTest/inline-view-limit.test
 
b/testdata/workloads/functional-planner/queries/PlannerTest/inline-view-limit.test
index 79f75b6..556ba65 100644
--- 
a/testdata/workloads/functional-planner/queries/PlannerTest/inline-view-limit.test
+++ 
b/testdata/workloads/functional-planner/queries/PlannerTest/inline-view-limit.test
@@ -1,10 +1,14 @@
 # predicate pushdown
 select * from (select * from functional.alltypessmall) a where id < 5
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN HDFS [functional.alltypessmall]
    partitions=4/4 files=4 size=6.32KB
    predicates: functional.alltypessmall.id < 5
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 01:EXCHANGE [UNPARTITIONED]
 |
 00:SCAN HDFS [functional.alltypessmall]
@@ -14,6 +18,8 @@ select * from (select * from functional.alltypessmall) a 
where id < 5
 # predicate pushdown is prevented in presence of limit clause
 select * from (select * from functional.alltypessmall limit 10) a where id < 5 
limit 5
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SELECT
 |  predicates: functional.alltypessmall.id < 5
 |  limit: 5
@@ -22,6 +28,8 @@ select * from (select * from functional.alltypessmall limit 
10) a where id < 5 l
    partitions=4/4 files=4 size=6.32KB
    limit: 10
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 01:SELECT
 |  predicates: functional.alltypessmall.id < 5
 |  limit: 5
@@ -39,6 +47,8 @@ select *
 from (select * from functional.alltypessmall order by id limit 10) a
 where id < 5 limit 5
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:SELECT
 |  predicates: id < 5
 |  limit: 5
@@ -49,6 +59,8 @@ where id < 5 limit 5
 00:SCAN HDFS [functional.alltypessmall]
    partitions=4/4 files=4 size=6.32KB
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 02:SELECT
 |  predicates: id < 5
 |  limit: 5
@@ -72,6 +84,8 @@ from functional.alltypes
     group by 1 order by 2 limit 5) a using (id)
 where a.id < 5 limit 5
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: functional.alltypes.id = id
 |  runtime filters: RF000 <- id
@@ -92,6 +106,8 @@ where a.id < 5 limit 5
    predicates: functional.alltypes.id < 5
    runtime filters: RF000 -> functional.alltypes.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 09:EXCHANGE [UNPARTITIONED]
 |  limit: 5
 |
@@ -135,6 +151,8 @@ from (
   limit 10) a
 where id < 5 limit 5
 ---- PLAN
+PLAN-ROOT SINK
+|
 03:SELECT
 |  predicates: a.id < 5
 |  limit: 5
@@ -151,6 +169,8 @@ where id < 5 limit 5
    partitions=4/4 files=4 size=6.32KB
    runtime filters: RF000 -> a.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 03:SELECT
 |  predicates: a.id < 5
 |  limit: 5
@@ -179,6 +199,8 @@ where id < 5
 order by id
 limit 5
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:TOP-N [LIMIT=5]
 |  order by: id ASC
 |
@@ -189,6 +211,8 @@ limit 5
    partitions=4/4 files=4 size=6.32KB
    limit: 10
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 02:TOP-N [LIMIT=5]
 |  order by: id ASC
 |
@@ -213,6 +237,8 @@ where id < 5
 order by id
 limit 5
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:TOP-N [LIMIT=5]
 |  order by: id ASC
 |
@@ -231,6 +257,8 @@ limit 5
    partitions=4/4 files=4 size=6.32KB
    runtime filters: RF000 -> a.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:TOP-N [LIMIT=5]
 |  order by: id ASC
 |
@@ -259,6 +287,8 @@ select *
 from functional.alltypes
   join (select id from functional.alltypessmall limit 10) a using (id)
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: functional.alltypes.id = id
 |  runtime filters: RF000 <- id
@@ -271,6 +301,8 @@ from functional.alltypes
    partitions=24/24 files=24 size=478.45KB
    runtime filters: RF000 -> functional.alltypes.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 05:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [INNER JOIN, BROADCAST]
@@ -296,6 +328,8 @@ select *
 from functional.alltypes
   join (select id from functional.alltypessmall order by id limit 10) a using 
(id)
 ---- PLAN
+PLAN-ROOT SINK
+|
 03:HASH JOIN [INNER JOIN]
 |  hash predicates: functional.alltypes.id = id
 |  runtime filters: RF000 <- id
@@ -310,6 +344,8 @@ from functional.alltypes
    partitions=24/24 files=24 size=478.45KB
    runtime filters: RF000 -> functional.alltypes.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 06:EXCHANGE [UNPARTITIONED]
 |
 03:HASH JOIN [INNER JOIN, BROADCAST]
@@ -344,6 +380,8 @@ where a.id < 5
 order by a.id
 limit 5
 ---- PLAN
+PLAN-ROOT SINK
+|
 05:TOP-N [LIMIT=5]
 |  order by: id ASC
 |
@@ -368,6 +406,8 @@ limit 5
    predicates: functional.alltypes.id < 5
    runtime filters: RF000 -> functional.alltypes.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 09:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: id ASC
 |  limit: 5
@@ -416,6 +456,8 @@ where a.id < 5
 order by a.id
 limit 5
 ---- PLAN
+PLAN-ROOT SINK
+|
 06:TOP-N [LIMIT=5]
 |  order by: id ASC
 |
@@ -442,6 +484,8 @@ limit 5
    predicates: functional.alltypes.id < 5
    runtime filters: RF000 -> functional.alltypes.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 10:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: id ASC
 |  limit: 5
@@ -486,6 +530,8 @@ select x.id from (
 order by x.id
 limit 100 offset 4
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:TOP-N [LIMIT=100 OFFSET=4]
 |  order by: id ASC
 |
@@ -495,6 +541,8 @@ limit 100 offset 4
 00:SCAN HDFS [functional.alltypesagg]
    partitions=11/11 files=11 size=814.73KB
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 02:TOP-N [LIMIT=100 OFFSET=4]
 |  order by: id ASC
 |
@@ -519,6 +567,8 @@ left outer join
 on (a.id = b.id)
 where a.id > 10 and b.id > 20
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [LEFT OUTER JOIN]
 |  hash predicates: id = id
 |  other predicates: id > 20
@@ -542,6 +592,8 @@ left outer join
 on (a.id = b.id)
 where a.id > 10 and b.id > 20
 ---- PLAN
+PLAN-ROOT SINK
+|
 03:HASH JOIN [RIGHT OUTER JOIN]
 |  hash predicates: id = id
 |  other predicates: id > 20
@@ -570,6 +622,8 @@ right outer join
 on (a.id = b.id)
 where a.id > 10 and b.id > 20
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT OUTER JOIN]
 |  hash predicates: id = id
 |  other predicates: id > 10
@@ -595,6 +649,8 @@ right outer join
 on (a.id = b.id)
 where a.id > 10 and b.id > 20
 ---- PLAN
+PLAN-ROOT SINK
+|
 03:HASH JOIN [RIGHT OUTER JOIN]
 |  hash predicates: id = id
 |  other predicates: id > 10
@@ -618,6 +674,8 @@ where a.id > 10 and b.id > 20
 # have explain_level=1
 select * from (select * from functional.alltypes limit 100) v where id < 10 
limit 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SELECT
 |  predicates: functional.alltypes.id < 10
 |  limit: 1

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/testdata/workloads/functional-planner/queries/PlannerTest/inline-view.test
----------------------------------------------------------------------
diff --git 
a/testdata/workloads/functional-planner/queries/PlannerTest/inline-view.test 
b/testdata/workloads/functional-planner/queries/PlannerTest/inline-view.test
index fe6ade8..13f6326 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/inline-view.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/inline-view.test
@@ -12,6 +12,8 @@ from (
 join functional.alltypes t2 on (t1.int_col = t2.int_col)
 where month = 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: t2.int_col = int_col
 |  runtime filters: RF000 <- int_col
@@ -31,6 +33,8 @@ where month = 1
    partitions=2/24 files=2 size=40.32KB
    runtime filters: RF000 -> t2.int_col
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 09:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, BROADCAST]
@@ -67,8 +71,12 @@ where month = 1
 # simple full scan subquery
 select * from (select y x from (select id y from 
functional_hbase.alltypessmall) a) b
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN HBASE [functional_hbase.alltypessmall]
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 01:EXCHANGE [UNPARTITIONED]
 |
 00:SCAN HBASE [functional_hbase.alltypessmall]
@@ -78,6 +86,8 @@ select * from (select t2.*
 from functional.testtbl t1 join functional.testtbl t2 using(id)
 where t1.zip = 94611) x
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: t1.id = t2.id
 |  runtime filters: RF000 <- t2.id
@@ -90,6 +100,8 @@ where t1.zip = 94611) x
    predicates: t1.zip = 94611
    runtime filters: RF000 -> t1.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [INNER JOIN, BROADCAST]
@@ -121,6 +133,8 @@ from
      and b.string_col = '15'
      and a.tinyint_col + b.tinyint_col < 15) x
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT OUTER JOIN]
 |  hash predicates: a.id = b.id, a.int_col = b.int_col
 |  other predicates: a.tinyint_col = 15, a.day >= 6, a.tinyint_col + 
b.tinyint_col < 15
@@ -145,6 +159,8 @@ NODE 1:
   HDFS SPLIT 
hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=3/090301.txt
 0:1620
   HDFS SPLIT 
hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=4/090401.txt
 0:1621
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 05:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [RIGHT OUTER JOIN, PARTITIONED]
@@ -168,6 +184,8 @@ NODE 1:
 # predicate pushdown
 select * from (select * from functional_hbase.alltypessmall) a where id < 5
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN HBASE [functional_hbase.alltypessmall]
    predicates: functional_hbase.alltypessmall.id < 5
 ====
@@ -188,6 +206,8 @@ and b.string_col = '15'
 and a.tinyint_col + b.tinyint_col < 15
 and b.id + 15 = 27
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT OUTER JOIN]
 |  hash predicates: id = id, int_col = int_col
 |  other predicates: tinyint_col = 15, day >= 6, tinyint_col + tinyint_col < 15
@@ -202,6 +222,8 @@ and b.id + 15 = 27
    predicates: functional.alltypesagg.tinyint_col = 15, 
functional.alltypesagg.id + 15 = 27
    runtime filters: RF000 -> id, RF001 -> int_col
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 05:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [RIGHT OUTER JOIN, PARTITIONED]
@@ -240,6 +262,8 @@ and a.tinyint_col = 15
 and b.string_col = '15'
 and a.tinyint_col + b.tinyint_col < 15
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT OUTER JOIN]
 |  hash predicates: id = id, int_col = int_col
 |  other predicates: tinyint_col = 15, tinyint_col + tinyint_col < 15
@@ -264,6 +288,8 @@ NODE 1:
   HDFS SPLIT 
hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=3/090301.txt
 0:1620
   HDFS SPLIT 
hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=4/090401.txt
 0:1621
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 05:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [RIGHT OUTER JOIN, PARTITIONED]
@@ -301,6 +327,8 @@ and x.float_col > 4.5
 and c.string_col < '7'
 and x.int_col + x.float_col + cast(c.string_col as float) < 1000
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: c.id = a.tinyint_col
 |  other predicates: a.int_col + b.float_col + CAST(c.string_col AS FLOAT) < 
1000
@@ -337,6 +365,8 @@ NODE 2:
   HDFS SPLIT 
hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=3/090301.txt
 0:1620
   HDFS SPLIT 
hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=4/090401.txt
 0:1621
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 08:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, PARTITIONED]
@@ -374,6 +404,8 @@ avg(tinyint_col)
 from (select * from functional.alltypesagg) a
 group by 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:AGGREGATE [FINALIZE]
 |  output: count(*), min(functional.alltypesagg.tinyint_col), 
max(functional.alltypesagg.tinyint_col), 
sum(functional.alltypesagg.tinyint_col), avg(functional.alltypesagg.tinyint_col)
 |  group by: functional.alltypesagg.tinyint_col
@@ -394,6 +426,8 @@ NODE 0:
   HDFS SPLIT 
hdfs://localhost:20500/test-warehouse/alltypesagg/year=2010/month=1/day=9/100109.txt
 0:76263
   HDFS SPLIT 
hdfs://localhost:20500/test-warehouse/alltypesagg/year=2010/month=1/day=__HIVE_DEFAULT_PARTITION__/000000_0
 0:72759
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:EXCHANGE [UNPARTITIONED]
 |
 03:AGGREGATE [FINALIZE]
@@ -417,6 +451,8 @@ from functional.alltypesagg
 group by 1
 ) a
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:AGGREGATE [FINALIZE]
 |  output: count(*), min(tinyint_col), max(tinyint_col), sum(tinyint_col), 
avg(tinyint_col)
 |  group by: tinyint_col
@@ -424,6 +460,8 @@ group by 1
 00:SCAN HDFS [functional.alltypesagg]
    partitions=11/11 files=11 size=814.73KB
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:EXCHANGE [UNPARTITIONED]
 |
 03:AGGREGATE [FINALIZE]
@@ -450,6 +488,8 @@ from
     limit 5
 ) y
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:TOP-N [LIMIT=5]
 |  order by: c2 ASC, c3 DESC
 |
@@ -459,6 +499,8 @@ from
 |
 00:SCAN HBASE [functional_hbase.alltypessmall]
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 05:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: c2 ASC, c3 DESC
 |  limit: 5
@@ -496,14 +538,20 @@ from (
 order by 2,1 desc
 limit 0
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:EMPTYSET
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 00:EMPTYSET
 ====
 # distinct *
 select distinct *
 from (select distinct * from functional.testtbl) x
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:AGGREGATE [FINALIZE]
 |  group by: functional.testtbl.id, functional.testtbl.name, 
functional.testtbl.zip
 |
@@ -513,6 +561,8 @@ from (select distinct * from functional.testtbl) x
 00:SCAN HDFS [functional.testtbl]
    partitions=1/1 files=0 size=0B
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 05:EXCHANGE [UNPARTITIONED]
 |
 02:AGGREGATE [FINALIZE]
@@ -533,6 +583,8 @@ from (select distinct * from functional.testtbl) x
 select distinct id, zip
 from (select distinct * from functional.testtbl) x
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:AGGREGATE [FINALIZE]
 |  group by: functional.testtbl.id, functional.testtbl.zip
 |
@@ -542,6 +594,8 @@ from (select distinct * from functional.testtbl) x
 00:SCAN HDFS [functional.testtbl]
    partitions=1/1 files=0 size=0B
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 07:EXCHANGE [UNPARTITIONED]
 |
 06:AGGREGATE [FINALIZE]
@@ -576,6 +630,8 @@ from (
 where c1 is not null
 and   c2 > 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:AGGREGATE [FINALIZE]
 |  output: count(*), avg(functional.alltypesagg.int_col)
 |  group by: functional.alltypesagg.int_col % 7
@@ -584,6 +640,8 @@ and   c2 > 10
 00:SCAN HDFS [functional.alltypesagg]
    partitions=11/11 files=11 size=814.73KB
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:EXCHANGE [UNPARTITIONED]
 |
 03:AGGREGATE [FINALIZE]
@@ -614,6 +672,8 @@ from (
      on (j.test_name = d.name)
 where j.test_id <= 1006
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT OUTER JOIN]
 |  hash predicates: b.name = a.test_name
 |  runtime filters: RF000 <- a.test_name
@@ -626,6 +686,8 @@ where j.test_id <= 1006
    partitions=1/1 files=1 size=171B
    runtime filters: RF000 -> b.name
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 05:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [RIGHT OUTER JOIN, PARTITIONED]
@@ -663,6 +725,8 @@ from functional.alltypessmall c
      on (x.tinyint_col = c.id)
 group by x.smallint_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 05:AGGREGATE [FINALIZE]
 |  output: count(b.id)
 |  group by: a.smallint_col
@@ -685,6 +749,8 @@ group by x.smallint_col
    partitions=11/11 files=11 size=814.73KB
    runtime filters: RF000 -> a.tinyint_col, RF001 -> a.smallint_col
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 11:EXCHANGE [UNPARTITIONED]
 |
 10:AGGREGATE [FINALIZE]
@@ -746,6 +812,8 @@ and x.float_col > 4.5
 and c.string_col < '7'
 and x.int_col + x.float_col + CAST(c.string_col AS FLOAT) < 1000
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: c.id = a.tinyint_col
 |  other predicates: a.int_col + b.float_col + CAST(c.string_col AS FLOAT) < 
1000
@@ -769,6 +837,8 @@ and x.int_col + x.float_col + CAST(c.string_col AS FLOAT) < 
1000
    predicates: c.string_col < '7'
    runtime filters: RF000 -> c.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 08:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, PARTITIONED]
@@ -812,6 +882,8 @@ from functional.alltypessmall c
   ) x on (x.smallint_col = c.id)
 group by x.smallint_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 06:AGGREGATE [FINALIZE]
 |  output: sum(count(a.id))
 |  group by: b.smallint_col
@@ -839,6 +911,8 @@ group by x.smallint_col
    partitions=4/4 files=4 size=6.32KB
    runtime filters: RF000 -> c.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 13:EXCHANGE [UNPARTITIONED]
 |
 12:AGGREGATE [FINALIZE]
@@ -887,9 +961,13 @@ group by x.smallint_col
 # Values statement in subqueries with predicate
 select * from (select y from (values((1 as y),(11))) a where y < 10) b
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:UNION
    constant-operands=1
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 00:UNION
    constant-operands=1
 ====
@@ -902,6 +980,8 @@ select * from
      (select tinyint_col from functional.alltypes)) a
    where y < 10) b
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:UNION
 |  constant-operands=1
 |
@@ -909,6 +989,8 @@ select * from
    partitions=24/24 files=24 size=478.45KB
    predicates: functional.alltypes.tinyint_col < 10
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 02:EXCHANGE [UNPARTITIONED]
 |
 00:UNION
@@ -922,9 +1004,13 @@ select * from
 select * from (select 1 as y union all select 2 union all select * from 
(select 11) a) b
 where y < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:UNION
    constant-operands=2
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 00:UNION
    constant-operands=2
 ====
@@ -933,11 +1019,15 @@ where y < 10
 select * from (values(1 as y) union all values(2) union all select * from 
(values(11)) a) b
 where y < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:UNION
 |  constant-operands=2
 |
 01:UNION
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 00:UNION
 |  constant-operands=2
 |
@@ -951,6 +1041,8 @@ inner join
 inner join
 (select 1 a, 3 b union all select 1 a, 3 b) z on z.b = y.b
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: b = b
 |
@@ -966,6 +1058,8 @@ inner join
 00:UNION
    constant-operands=2
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN, BROADCAST]
 |  hash predicates: b = b
 |
@@ -992,6 +1086,8 @@ left semi join
 inner join
 (select 1 a, 3 b union all select 1 a, 3 b) z on z.b = x.id + 2
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: x.id + 2 = b
 |  runtime filters: RF000 <- b
@@ -1010,6 +1106,8 @@ inner join
    partitions=4/4 files=4 size=6.32KB
    runtime filters: RF000 -> x.id + 2, RF001 -> x.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 07:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, BROADCAST]
@@ -1039,6 +1137,8 @@ select b.* from functional.decimal_tbl a left outer join
   (select d1, d1 + NULL IS NULL x from functional.decimal_tbl) b
 on (a.d1 = b.d1)
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [LEFT OUTER JOIN]
 |  hash predicates: a.d1 = d1
 |
@@ -1056,6 +1156,8 @@ where foo = 10
 group by foo
 limit 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:AGGREGATE [FINALIZE]
 |  output: sum(foo)
 |  group by: foo
@@ -1075,6 +1177,8 @@ select * from
    from functional.alltypestiny t1 left outer join functional.alltypes t2
    on t1.int_col = t2.int_col and t1.tinyint_col = t2.tinyint_col) t
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT OUTER JOIN]
 |  hash predicates: t2.int_col = t1.int_col, t2.tinyint_col = t1.tinyint_col
 |  runtime filters: RF000 <- t1.int_col, RF001 <- t1.tinyint_col
@@ -1096,6 +1200,8 @@ select 1 from
 inner join functional.alltypestiny c
 on (aid < bid and aid = c.id)
 ---- PLAN
+PLAN-ROOT SINK
+|
 06:HASH JOIN [INNER JOIN]
 |  hash predicates: a.id = c.id
 |  runtime filters: RF000 <- c.id
@@ -1131,6 +1237,8 @@ inner join
    on a.id = b.int_col) v
 on (t1.id = v.id and v.int_col is null and v.int_col < 10 and v.id < 10)
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: a.id = t1.id
 |  runtime filters: RF000 <- t1.id
@@ -1161,6 +1269,8 @@ select * from
      from functional.alltypestiny) iv
   ) ivv
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN HDFS [functional.alltypestiny]
    partitions=4/4 files=4 size=460B
 ====
@@ -1171,6 +1281,8 @@ select * from
      from functional.alltypestiny) iv
   ) ivv where bigint_col = bigint_col2
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN HDFS [functional.alltypestiny]
    partitions=4/4 files=4 size=460B
    predicates: bigint_col = bigint_col
@@ -1183,6 +1295,8 @@ select * from
      from functional.alltypestiny) iv
   ) ivv
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:AGGREGATE [FINALIZE]
 |  output: sum(bigint_col)
 |
@@ -1196,6 +1310,8 @@ select * from
      from functional.alltypestiny) iv
   ) ivv where s1 = s2
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:AGGREGATE [FINALIZE]
 |  output: sum(bigint_col)
 |  having: sum(bigint_col) = sum(bigint_col)

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/testdata/workloads/functional-planner/queries/PlannerTest/join-order.test
----------------------------------------------------------------------
diff --git 
a/testdata/workloads/functional-planner/queries/PlannerTest/join-order.test 
b/testdata/workloads/functional-planner/queries/PlannerTest/join-order.test
index e1951e0..49a0b87 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/join-order.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/join-order.test
@@ -23,6 +23,8 @@ order by
   o_orderdate
 limit 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 06:TOP-N [LIMIT=10]
 |  order by: round(sum(l_extendedprice * (1 - l_discount)), 5) DESC, 
o_orderdate ASC
 |
@@ -52,6 +54,8 @@ limit 10
    predicates: l_shipdate > '1995-03-15'
    runtime filters: RF001 -> l.l_orderkey
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 11:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: round(sum(l_extendedprice * (1 - l_discount)), 5) DESC, 
o_orderdate ASC
 |  limit: 10
@@ -120,6 +124,8 @@ order by
   o_orderdate
 limit 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 06:TOP-N [LIMIT=10]
 |  order by: round(sum(l_extendedprice * (1 - l_discount)), 5) DESC, 
o_orderdate ASC
 |
@@ -149,6 +155,8 @@ limit 10
    predicates: c.c_mktsegment = 'BUILDING'
    runtime filters: RF001 -> c.c_custkey
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 11:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: round(sum(l_extendedprice * (1 - l_discount)), 5) DESC, 
o_orderdate ASC
 |  limit: 10
@@ -219,6 +227,8 @@ order by
   revenue desc
 limit 100
 ---- PLAN
+PLAN-ROOT SINK
+|
 12:TOP-N [LIMIT=100]
 |  order by: round(sum(l_extendedprice * (1 - l_discount)), 5) DESC
 |
@@ -271,6 +281,8 @@ limit 100
    partitions=1/1 files=1 size=718.94MB
    runtime filters: RF003 -> l_suppkey, RF005 -> l_orderkey
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 20:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: round(sum(l_extendedprice * (1 - l_discount)), 5) DESC
 |  limit: 100
@@ -369,6 +381,8 @@ where
   and s.s_nationkey = n.n_nationkey
   and n.n_regionkey = r.r_regionkey
 ---- PLAN
+PLAN-ROOT SINK
+|
 08:HASH JOIN [INNER JOIN]
 |  hash predicates: n.n_regionkey = r.r_regionkey
 |  runtime filters: RF000 <- r.r_regionkey
@@ -405,6 +419,8 @@ where
    partitions=1/1 files=1 size=1.33MB
    runtime filters: RF001 -> s.s_nationkey, RF002 -> s.s_suppkey
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 13:EXCHANGE [UNPARTITIONED]
 |
 08:HASH JOIN [INNER JOIN, BROADCAST]
@@ -469,6 +485,8 @@ order by
   o_orderpriority
 limit 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:TOP-N [LIMIT=10]
 |  order by: o_orderpriority ASC
 |
@@ -489,6 +507,8 @@ limit 10
    predicates: l_commitdate < l_receiptdate
    runtime filters: RF000 -> l_orderkey
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 09:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: o_orderpriority ASC
 |  limit: 10
@@ -531,6 +551,8 @@ from tpch.orders
 group by o_orderpriority
 order by o_orderpriority limit 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:TOP-N [LIMIT=10]
 |  order by: o_orderpriority ASC
 |
@@ -547,6 +569,8 @@ order by o_orderpriority limit 10
 00:SCAN HDFS [tpch.orders]
    partitions=1/1 files=1 size=162.56MB
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 09:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: o_orderpriority ASC
 |  limit: 10
@@ -585,6 +609,8 @@ from tpch.orders
 group by o_orderpriority
 order by o_orderpriority limit 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:TOP-N [LIMIT=10]
 |  order by: o_orderpriority ASC
 |
@@ -603,6 +629,8 @@ order by o_orderpriority limit 10
    partitions=1/1 files=1 size=162.56MB
    runtime filters: RF000 -> o_orderkey
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 09:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: o_orderpriority ASC
 |  limit: 10
@@ -643,6 +671,8 @@ from tpch.customer
   join tpch.nation on (c_nationkey = n_nationkey)
 where n_name = 'x'
 ---- PLAN
+PLAN-ROOT SINK
+|
 05:AGGREGATE [FINALIZE]
 |  output: count(*)
 |
@@ -666,6 +696,8 @@ where n_name = 'x'
    partitions=1/1 files=1 size=162.56MB
    runtime filters: RF001 -> o_custkey
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 10:AGGREGATE [FINALIZE]
 |  output: count:merge(*)
 |
@@ -708,6 +740,8 @@ from tpch.customer
   join tpch.nation on (c_nationkey = n_nationkey)
 where n_name = 'x'
 ---- PLAN
+PLAN-ROOT SINK
+|
 05:AGGREGATE [FINALIZE]
 |  output: count(*)
 |
@@ -728,6 +762,8 @@ where n_name = 'x'
    partitions=1/1 files=1 size=23.08MB
    runtime filters: RF000 -> c_nationkey
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 09:AGGREGATE [FINALIZE]
 |  output: count:merge(*)
 |
@@ -765,6 +801,8 @@ select c.int_col from functional.alltypestiny a
 cross join functional.alltypestiny b
 cross join functional.alltypes c
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:NESTED LOOP JOIN [CROSS JOIN]
 |
 |--03:NESTED LOOP JOIN [CROSS JOIN]
@@ -788,6 +826,8 @@ right join functional.alltypesagg t4 on (t3.id = t4.id)
 inner join functional.alltypes t5 on (t4.id = t5.id)
 inner join functional.alltypestiny t6 on (t5.id = t6.id)
 ---- PLAN
+PLAN-ROOT SINK
+|
 11:AGGREGATE [FINALIZE]
 |  output: count(*)
 |
@@ -843,6 +883,8 @@ left semi join functional.alltypesagg t4 on (t3.id = t4.id)
 inner join functional.alltypes t5 on (t3.id = t5.id)
 right join functional.alltypestiny t6 on (t5.id = t6.id)
 ---- PLAN
+PLAN-ROOT SINK
+|
 13:AGGREGATE [FINALIZE]
 |  output: count(*)
 |
@@ -897,6 +939,8 @@ inner join functional.alltypessmall t4 on (t3.id = t4.id)
 left semi join functional.alltypes t5 on (t4.id = t5.id)
 inner join functional.alltypestiny t6 on (t3.id = t6.id)
 ---- PLAN
+PLAN-ROOT SINK
+|
 13:AGGREGATE [FINALIZE]
 |  output: count(*)
 |
@@ -952,6 +996,8 @@ inner join functional.alltypessmall t4 on (t3.id = t4.id)
 left anti join functional.alltypes t5 on (t4.id = t5.id)
 inner join functional.alltypestiny t6 on (t3.id = t6.id)
 ---- PLAN
+PLAN-ROOT SINK
+|
 13:AGGREGATE [FINALIZE]
 |  output: count(*)
 |
@@ -1012,6 +1058,8 @@ on (t2.id = t3.id)
 inner join functional.alltypestiny t4
 on (t3.id = t4.id)
 ---- PLAN
+PLAN-ROOT SINK
+|
 12:HASH JOIN [INNER JOIN]
 |  hash predicates: t3.id = t4.id
 |  runtime filters: RF000 <- t4.id
@@ -1075,6 +1123,8 @@ on (t2.id = t3.id)
 inner join functional.alltypestiny t4
 on (t3.id = t4.id)
 ---- PLAN
+PLAN-ROOT SINK
+|
 12:HASH JOIN [INNER JOIN]
 |  hash predicates: t3.id = t4.id
 |  runtime filters: RF000 <- t4.id
@@ -1137,6 +1187,8 @@ inner join functional.alltypestiny t4
 on (t2.id = t4.id)
 where t2.month = 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 12:HASH JOIN [INNER JOIN]
 |  hash predicates: b.id = t4.id
 |  runtime filters: RF000 <- t4.id
@@ -1201,6 +1253,8 @@ inner join functional.alltypestiny t4
 on (t2.id = t4.id)
 where t2.month = 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 12:HASH JOIN [INNER JOIN]
 |  hash predicates: b.id = t4.id
 |  runtime filters: RF000 <- t4.id
@@ -1264,6 +1318,8 @@ LEFT SEMI JOIN
 ON t4.bigint_col = `$a$3`.`$c$1`
 WHERE `$a$2`.`$c$1` > t4.id
 ---- PLAN
+PLAN-ROOT SINK
+|
 10:AGGREGATE [FINALIZE]
 |  output: sum(t4.tinyint_col)
 |
@@ -1310,6 +1366,8 @@ left outer join functional.alltypessmall b
 # both predicates should appear in the 'other predicates'
 where a.int_col = b.int_col and b.bigint_col < a.tinyint_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT OUTER JOIN]
 |  hash predicates: b.id = a.id
 |  other predicates: a.int_col = b.int_col, b.bigint_col < a.tinyint_col
@@ -1333,6 +1391,8 @@ right outer join functional.alltypes c
 where a.int_col = b.int_col and b.bigint_col < a.tinyint_col
   and b.tinyint_col = c.tinyint_col and b.bool_col != c.bool_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [LEFT OUTER JOIN]
 |  hash predicates: c.id = b.id
 |  other predicates: a.int_col = b.int_col, b.bool_col != c.bool_col, 
b.tinyint_col = c.tinyint_col, b.bigint_col < a.tinyint_col
@@ -1363,6 +1423,8 @@ right outer join functional.alltypes c
 # all predicates should appear in the 'other predicates'
 where b.tinyint_col = c.tinyint_col and b.bool_col != c.bool_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [LEFT OUTER JOIN]
 |  hash predicates: c.id = b.id
 |  other predicates: b.bool_col != c.bool_col, b.tinyint_col = c.tinyint_col
@@ -1393,6 +1455,8 @@ select count(1) from
 left outer join functional.alltypestiny t3
 on (t3.string_col = t1.string_col_1 and t3.date_string_col = t1.string_col_2)
 ---- PLAN
+PLAN-ROOT SINK
+|
 05:AGGREGATE [FINALIZE]
 |  output: count(1)
 |
@@ -1427,6 +1491,8 @@ where timestamp_col = now()) b
 on (a.id = b.id)
 and a.date_string_col = ''
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [LEFT SEMI JOIN]
 |  hash predicates: a.id = functional.alltypes.id
 |  runtime filters: RF000 <- functional.alltypes.id
@@ -1447,6 +1513,8 @@ where date_string_col = '') b
 on (a.id = b.id)
 and a.timestamp_col = now()
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT SEMI JOIN]
 |  hash predicates: functional.alltypes.id = a.id
 |  runtime filters: RF000 <- a.id
@@ -1468,6 +1536,8 @@ where timestamp_col <=> now()) b
 on (a.id = b.id)
 and a.date_string_col <=> ''
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [LEFT SEMI JOIN]
 |  hash predicates: a.id = functional.alltypes.id
 |  runtime filters: RF000 <- functional.alltypes.id
@@ -1488,6 +1558,8 @@ where date_string_col <=> '') b
 on (a.id = b.id)
 and a.timestamp_col <=> now()
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT SEMI JOIN]
 |  hash predicates: functional.alltypes.id = a.id
 |  runtime filters: RF000 <- a.id

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/testdata/workloads/functional-planner/queries/PlannerTest/joins.test
----------------------------------------------------------------------
diff --git 
a/testdata/workloads/functional-planner/queries/PlannerTest/joins.test 
b/testdata/workloads/functional-planner/queries/PlannerTest/joins.test
index 260ba21..ba1395a 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/joins.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/joins.test
@@ -2,6 +2,8 @@ select *
 from functional.testtbl t1 join functional.testtbl t2 using(id)
 where t1.zip = 94611
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: t1.id = t2.id
 |  runtime filters: RF000 <- t2.id
@@ -14,6 +16,8 @@ where t1.zip = 94611
    predicates: t1.zip = 94611
    runtime filters: RF000 -> t1.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [INNER JOIN, BROADCAST]
@@ -36,6 +40,8 @@ from functional.testtbl t1 left outer join functional.testtbl 
t2
 on (t1.id - 1 = t2.id + 1)
 where t1.zip = 94611
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [LEFT OUTER JOIN]
 |  hash predicates: t1.id - 1 = t2.id + 1
 |
@@ -46,6 +52,8 @@ where t1.zip = 94611
    partitions=1/1 files=0 size=0B
    predicates: t1.zip = 94611
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [LEFT OUTER JOIN, BROADCAST]
@@ -67,6 +75,8 @@ from (select * from functional.alltypestiny) t1
   join (select * from functional.alltypestiny) t2 on (t1.id = t2.id)
   join functional.alltypestiny t3 on (coalesce(t1.id, t2.id) = t3.id)
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: coalesce(functional.alltypestiny.id, 
functional.alltypestiny.id) = t3.id
 |  runtime filters: RF000 <- t3.id
@@ -86,6 +96,8 @@ from (select * from functional.alltypestiny) t1
    partitions=4/4 files=4 size=460B
    runtime filters: RF000 -> coalesce(functional.alltypestiny.id, 
functional.alltypestiny.id), RF001 -> functional.alltypestiny.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 07:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, BROADCAST]
@@ -122,6 +134,8 @@ and a.tinyint_col = 15
 and b.string_col = '15'
 and a.tinyint_col + b.tinyint_col < 15
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT OUTER JOIN]
 |  hash predicates: a.id = b.id, a.int_col = b.int_col
 |  other predicates: a.tinyint_col = 15, a.day >= 6, a.tinyint_col + 
b.tinyint_col < 15
@@ -136,6 +150,8 @@ and a.tinyint_col + b.tinyint_col < 15
    predicates: a.tinyint_col = 15
    runtime filters: RF000 -> a.id, RF001 -> a.int_col
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 05:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [RIGHT OUTER JOIN, PARTITIONED]
@@ -171,6 +187,8 @@ and a.tinyint_col + b.tinyint_col < 15
 and a.float_col - c.double_col < 0
 and (b.double_col * c.tinyint_col > 1000 or c.tinyint_col < 1000)
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [LEFT OUTER JOIN]
 |  hash predicates: c.id = a.id, c.string_col = b.string_col
 |  other predicates: a.tinyint_col = 15, b.string_col = '15', a.day >= 6, 
b.month > 2, a.float_col - c.double_col < 0, a.tinyint_col + b.tinyint_col < 
15, (b.double_col * c.tinyint_col > 1000 OR c.tinyint_col < 1000)
@@ -189,6 +207,8 @@ and (b.double_col * c.tinyint_col > 1000 or c.tinyint_col < 
1000)
 02:SCAN HDFS [functional.alltypesaggnonulls c]
    partitions=2/10 files=2 size=148.10KB
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 09:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [LEFT OUTER JOIN, PARTITIONED]
@@ -224,6 +244,8 @@ inner join
 (select 1 as x, id from functional.alltypessmall) b
 on a.x = b.x
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: 1 = 1
 |
@@ -237,6 +259,8 @@ on a.x = b.x
 select a.int_col, b.x from functional.alltypessmall a inner join
 (values(1 as int_col, 'a' as x), (1, 'b'), (2, 'c')) b on a.int_col = b.int_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: a.int_col = int_col
 |  runtime filters: RF000 <- int_col
@@ -248,6 +272,8 @@ select a.int_col, b.x from functional.alltypessmall a inner 
join
    partitions=4/4 files=4 size=6.32KB
    runtime filters: RF000 -> a.int_col
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [INNER JOIN, BROADCAST]
@@ -267,6 +293,8 @@ select a.int_col, b.x from functional.alltypessmall a inner 
join
 select *
 from functional.alltypesagg join functional_hbase.alltypessmall using (id, 
int_col)
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: functional.alltypesagg.id = 
functional_hbase.alltypessmall.id, functional.alltypesagg.int_col = 
functional_hbase.alltypessmall.int_col
 |  runtime filters: RF000 <- functional_hbase.alltypessmall.id, RF001 <- 
functional_hbase.alltypessmall.int_col
@@ -277,6 +305,8 @@ from functional.alltypesagg join 
functional_hbase.alltypessmall using (id, int_c
    partitions=11/11 files=11 size=814.73KB
    runtime filters: RF000 -> functional.alltypesagg.id, RF001 -> 
functional.alltypesagg.int_col
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [INNER JOIN, BROADCAST]
@@ -302,6 +332,8 @@ and b.tinyint_col = 5
 and b.tinyint_col > 123
 and a.tinyint_col + b.tinyint_col < 15
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: a.int_col = b.int_col, a.id = CAST(b.id AS INT)
 |  other predicates: a.tinyint_col + b.tinyint_col < 15
@@ -326,6 +358,8 @@ NODE 0:
 NODE 1:
   HBASE KEYRANGE port=16202 5:5\0
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [INNER JOIN, BROADCAST]
@@ -361,6 +395,8 @@ group by x.tinyint_col
 order by 2
 limit 5
 ---- PLAN
+PLAN-ROOT SINK
+|
 10:TOP-N [LIMIT=5]
 |  order by: count(x.day) ASC
 |
@@ -401,6 +437,8 @@ limit 5
    partitions=11/11 files=11 size=814.73KB
    runtime filters: RF000 -> d.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 10:TOP-N [LIMIT=5]
 |  order by: count(x.day) ASC
 |
@@ -456,6 +494,8 @@ limit 5
 # join without "other join conjuncts"
 select * from functional.alltypessmall a, functional.alltypessmall b where 
a.id = b.id limit 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: a.id = b.id
 |  runtime filters: RF000 <- b.id
@@ -468,6 +508,8 @@ select * from functional.alltypessmall a, 
functional.alltypessmall b where a.id
    partitions=4/4 files=4 size=6.32KB
    runtime filters: RF000 -> a.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 05:EXCHANGE [UNPARTITIONED]
 |  limit: 1
 |
@@ -493,6 +535,8 @@ select *
 from functional.testtbl t1, functional.testtbl t2, functional.testtbl t3
 where t1.id = t3.id and t2.id = t3.id
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: t1.id = t3.id
 |  runtime filters: RF000 <- t3.id
@@ -512,6 +556,8 @@ where t1.id = t3.id and t2.id = t3.id
    partitions=1/1 files=0 size=0B
    runtime filters: RF000 -> t1.id, RF001 -> t1.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 07:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, BROADCAST]
@@ -542,6 +588,8 @@ where t1.id = t3.id and t2.id = t3.id
 select * from functional.emptytable a inner join
 functional.alltypes b on a.f2 = b.int_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: b.int_col = a.f2
 |  runtime filters: RF000 <- a.f2
@@ -553,6 +601,8 @@ functional.alltypes b on a.f2 = b.int_col
    partitions=24/24 files=24 size=478.45KB
    runtime filters: RF000 -> b.int_col
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [INNER JOIN, BROADCAST]
@@ -572,6 +622,8 @@ functional.alltypes b on a.f2 = b.int_col
 select *
 from functional.testtbl t1 cross join functional.testtbl
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:NESTED LOOP JOIN [CROSS JOIN]
 |
 |--01:SCAN HDFS [functional.testtbl]
@@ -580,6 +632,8 @@ from functional.testtbl t1 cross join functional.testtbl
 00:SCAN HDFS [functional.testtbl t1]
    partitions=1/1 files=0 size=0B
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:EXCHANGE [UNPARTITIONED]
 |
 02:NESTED LOOP JOIN [CROSS JOIN, BROADCAST]
@@ -596,6 +650,8 @@ from functional.testtbl t1 cross join functional.testtbl
 select *
 from functional.testtbl t1 cross join functional.testtbl t2 where t1.id < t2.id
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:NESTED LOOP JOIN [INNER JOIN]
 |  predicates: t1.id < t2.id
 |
@@ -605,6 +661,8 @@ from functional.testtbl t1 cross join functional.testtbl t2 
where t1.id < t2.id
 00:SCAN HDFS [functional.testtbl t1]
    partitions=1/1 files=0 size=0B
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:EXCHANGE [UNPARTITIONED]
 |
 02:NESTED LOOP JOIN [INNER JOIN, BROADCAST]
@@ -626,6 +684,8 @@ on (a.id = b.id and b.int_col = a.int_col)
 inner join [shuffle] functional.alltypes c
 on (b.id = c.id and c.int_col = b.int_col)
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: b.id = c.id, b.int_col = c.int_col
 |  runtime filters: RF000 <- c.id, RF001 <- c.int_col
@@ -645,6 +705,8 @@ on (b.id = c.id and c.int_col = b.int_col)
    partitions=24/24 files=24 size=478.45KB
    runtime filters: RF000 -> a.id, RF001 -> a.int_col, RF002 -> a.id, RF003 -> 
a.int_col
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 08:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, PARTITIONED]
@@ -680,6 +742,8 @@ inner join [shuffle]
    from functional.alltypes group by int_col, bool_col) b
 on (a.int_col = b.int_col and b.bool_col = a.bool_col)
 ---- PLAN
+PLAN-ROOT SINK
+|
 03:HASH JOIN [INNER JOIN]
 |  hash predicates: a.bool_col = bool_col, a.int_col = int_col
 |  runtime filters: RF000 <- bool_col, RF001 <- int_col
@@ -695,6 +759,8 @@ on (a.int_col = b.int_col and b.bool_col = a.bool_col)
    partitions=24/24 files=24 size=478.45KB
    runtime filters: RF000 -> a.bool_col, RF001 -> a.int_col
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 07:EXCHANGE [UNPARTITIONED]
 |
 03:HASH JOIN [INNER JOIN, PARTITIONED]
@@ -730,6 +796,8 @@ inner join [shuffle]
    from functional.alltypes group by int_col, bool_col) c
 on (b.int_col = c.int_col and c.bool_col = b.bool_col)
 ---- PLAN
+PLAN-ROOT SINK
+|
 05:HASH JOIN [INNER JOIN]
 |  hash predicates: b.bool_col = a.bool_col, b.int_col = a.int_col
 |  runtime filters: RF000 <- a.bool_col, RF001 <- a.int_col
@@ -753,6 +821,8 @@ on (b.int_col = c.int_col and c.bool_col = b.bool_col)
    partitions=24/24 files=24 size=478.45KB
    runtime filters: RF000 -> b.bool_col, RF001 -> b.int_col
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 10:EXCHANGE [UNPARTITIONED]
 |
 05:HASH JOIN [INNER JOIN, PARTITIONED]
@@ -801,6 +871,8 @@ and b.string_col = a.string_col and b.date_string_col = 
a.string_col
 # redundant predicates to test minimal spanning tree of equivalent slots at a
 where a.tinyint_col = a.smallint_col and a.int_col = a.bigint_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: b.id = a.id, b.string_col = a.string_col
 |  runtime filters: RF000 <- a.id, RF001 <- a.string_col
@@ -826,6 +898,8 @@ and b.string_col = a.string_col and b.date_string_col = 
a.string_col
 # redundant predicates to test minimal spanning tree of equivalent slots at a
 where a.tinyint_col = a.smallint_col and a.int_col = a.bigint_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT OUTER JOIN]
 |  hash predicates: b.id = a.id, b.int_col = a.id, b.id = a.int_col, b.id = 
a.bigint_col, b.bigint_col = a.id, b.id = a.smallint_col, b.string_col = 
a.string_col, b.id = a.tinyint_col, b.date_string_col = a.string_col
 |  runtime filters: RF000 <- a.id, RF001 <- a.id, RF002 <- a.int_col, RF003 <- 
a.bigint_col, RF004 <- a.id, RF005 <- a.smallint_col, RF006 <- a.string_col, 
RF007 <- a.tinyint_col, RF008 <- a.string_col
@@ -848,6 +922,8 @@ inner join functional.alltypessmall c
 # redundant join predicates
 where a.id = c.id and a.int_col = c.int_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: c.id = b.id, c.int_col = b.int_col
 |  runtime filters: RF000 <- b.id, RF001 <- b.int_col
@@ -877,6 +953,8 @@ functional.alltypes b,
 functional.alltypessmall c
 where a.id = c.id and b.int_col = c.int_col and b.int_col = c.id
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: c.id = a.id
 |  runtime filters: RF000 <- a.id
@@ -907,6 +985,8 @@ select a.* from
    group by 1, 2) b
 on a.int_col = b.int_col and a.int_col = b.smallint_col and a.int_col = b.c
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: int_col = int_col
 |  runtime filters: RF000 <- int_col
@@ -946,6 +1026,8 @@ and t3.int_col = t2.bigint_col
 # already been established by 't3.int_col = t2.bigint_col'
 and t3.bigint_col = t2.bigint_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: t1.smallint_col = t3.int_col, t2.bigint_col = t3.int_col
 |  runtime filters: RF000 <- t3.int_col, RF001 <- t3.int_col
@@ -966,6 +1048,8 @@ and t3.bigint_col = t2.bigint_col
    partitions=11/11 files=11 size=814.73KB
    runtime filters: RF000 -> t2.smallint_col, RF001 -> t2.bigint_col, RF002 -> 
t2.smallint_col
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 08:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, PARTITIONED]
@@ -1006,6 +1090,8 @@ inner join
 on a.id = b.x and a.id = b.tinyint_col and
    a.int_col = b.y and a.int_col = b.bigint_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: a.id = id + id, a.int_col = int_col * int_col
 |  runtime filters: RF000 <- id + id, RF001 <- int_col * int_col
@@ -1031,6 +1117,8 @@ inner join
 on a.id = b.x and a.id = b.tinyint_col and
    a.int_col = b.y and a.int_col = b.bigint_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: a.id = id + id, a.int_col = int_col * int_col
 |  runtime filters: RF000 <- id + id, RF001 <- int_col * int_col
@@ -1061,6 +1149,8 @@ inner join
 on a.id = b.x and a.id = b.tinyint_col and
    a.int_col = b.y and a.int_col = b.bigint_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 05:HASH JOIN [INNER JOIN]
 |  hash predicates: a.id = id + id, a.int_col = int_col * int_col
 |  runtime filters: RF000 <- id + id, RF001 <- int_col * int_col
@@ -1098,6 +1188,8 @@ inner join
 on a.id = b.x and a.id = b.tinyint_col and
    a.int_col = b.y and a.int_col = b.bigint_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: a.id = tinyint_col, a.id = x, a.int_col = bigint_col, 
a.int_col = y
 |  runtime filters: RF000 <- tinyint_col, RF001 <- x, RF002 <- bigint_col, 
RF003 <- y
@@ -1123,6 +1215,8 @@ on t3.smallint_col = t1.tinyint_col
 inner join functional.alltypes t2
 on t2.string_col = t1.string_col and t3.date_string_col = t2.string_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: t1.string_col = t2.string_col
 |  runtime filters: RF000 <- t2.string_col
@@ -1149,6 +1243,8 @@ full outer join
 (select distinct bigint_col from functional.alltypestiny) b
 on (a.int_col = b.bigint_col)
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [FULL OUTER JOIN]
 |  hash predicates: int_col = bigint_col
 |
@@ -1164,6 +1260,8 @@ on (a.int_col = b.bigint_col)
 00:SCAN HDFS [functional.alltypessmall]
    partitions=4/4 files=4 size=6.32KB
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 09:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [FULL OUTER JOIN, PARTITIONED]
@@ -1206,6 +1304,8 @@ functional.alltypestiny b
 on a.id = b.id
 where b.id < 5
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 06:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, BROADCAST]
@@ -1246,6 +1346,8 @@ left outer join
 on a.id = b.id
 where b.id < 5
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 07:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [RIGHT OUTER JOIN, PARTITIONED]
@@ -1294,6 +1396,8 @@ full outer join
 on a.id = b.id
 where b.id < 5
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 09:EXCHANGE [UNPARTITIONED]
 |
 06:HASH JOIN [FULL OUTER JOIN, PARTITIONED]
@@ -1329,6 +1433,8 @@ where b.id < 5
 select straight_join count(*)
 from functional.decimal_tbl a join functional.decimal_tbl b on a.d1 = b.d5
 ---- PLAN
+PLAN-ROOT SINK
+|
 03:AGGREGATE [FINALIZE]
 |  output: count(*)
 |
@@ -1348,6 +1454,8 @@ select j.* from functional.jointbl j left semi join 
functional.dimtbl d
   on (j.test_id = d.id and j.test_zip < d.zip and d.name = 'Name2')
 where j.test_id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [LEFT SEMI JOIN]
 |  hash predicates: j.test_id = d.id
 |  other join predicates: j.test_zip < d.zip
@@ -1367,6 +1475,8 @@ select b.* from functional.alltypes a right semi join 
functional.alltypestiny b
   on (a.id = b.id and a.int_col < b.int_col and a.bool_col = false)
 where b.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT SEMI JOIN]
 |  hash predicates: a.id = b.id
 |  other join predicates: a.int_col < b.int_col
@@ -1386,6 +1496,8 @@ select j.* from functional.jointbl j left anti join 
functional.dimtbl d
   on (j.test_id = d.id and j.test_zip < d.zip and d.name = 'Name2')
 where j.test_id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [LEFT ANTI JOIN]
 |  hash predicates: j.test_id = d.id
 |  other join predicates: j.test_zip < d.zip
@@ -1403,6 +1515,8 @@ select count(*) from functional.JoinTbl j
 left anti join functional.DimTbl d on j.test_id = d.id
 inner join functional.JoinTbl k on j.test_id = k.test_id and j.alltypes_id = 
5000
 ---- PLAN
+PLAN-ROOT SINK
+|
 06:AGGREGATE [FINALIZE]
 |  output: count(*)
 |
@@ -1429,6 +1543,8 @@ select straight_join * from functional.alltypestiny a
 inner join [broadcast] functional.alltypes b on a.id = b.id
 inner join [shuffle] functional.alltypessmall c on b.id = c.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 08:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, PARTITIONED]
@@ -1461,6 +1577,8 @@ select /* +straight_join */ * from 
functional.alltypestiny a
 inner join /* +broadcast */ functional.alltypes b on a.id = b.id
 inner join /* +shuffle */ functional.alltypessmall c on b.id = c.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 08:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, PARTITIONED]
@@ -1499,6 +1617,8 @@ inner join
 -- +shuffle
 functional.alltypessmall c on b.id = c.id
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 08:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, PARTITIONED]
@@ -1536,6 +1656,8 @@ cross join functional.alltypes c
 # assigned to inverted cross join
 where c.id != b.id
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:NESTED LOOP JOIN [INNER JOIN]
 |  predicates: c.id != b.id
 |
@@ -1570,6 +1692,8 @@ on (v1.tinyint_col = v2.tinyint_col and
     v1.tinyint_col = v2.int_col and
     v1.tinyint_col = v2.bigint_col)
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 10:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, PARTITIONED]
@@ -1620,6 +1744,8 @@ on (v1.tinyint_col = v2.tinyint_col and
     v1.tinyint_col = v2.int_col and
     v1.tinyint_col = v2.bigint_col)
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 10:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [FULL OUTER JOIN, PARTITIONED]
@@ -1666,6 +1792,8 @@ inner join
 on (v1.string_col = v2.string_col and
     v1.bigint_col = v2.bigint_col)
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 10:EXCHANGE [UNPARTITIONED]
 |
 04:HASH JOIN [INNER JOIN, PARTITIONED]
@@ -1706,6 +1834,8 @@ left semi join functional.alltypes b
 # predicates are in reverse order of compatible group by exprs
 on (a.string_col = b.string_col and a.int_col = b.int_col)
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 07:EXCHANGE [UNPARTITIONED]
 |
 03:HASH JOIN [LEFT SEMI JOIN, PARTITIONED]
@@ -1738,6 +1868,8 @@ right semi join
 # predicates are in reverse order of compatible group by exprs
 on (a.string_col = b.string_col and a.int_col = b.int_col)
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 07:EXCHANGE [UNPARTITIONED]
 |
 03:HASH JOIN [RIGHT SEMI JOIN, PARTITIONED]
@@ -1777,6 +1909,8 @@ inner join
     where bigint_col = smallint_col and smallint_col = tinyint_col) c
 on (b.int_col = c.smallint_col and b.string_col = c.string_col)
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 16:EXCHANGE [UNPARTITIONED]
 |
 07:HASH JOIN [INNER JOIN, PARTITIONED]
@@ -1842,6 +1976,8 @@ where not exists (select *
 # Predicate on c (invisible side of anti join) is assigned to the scan node
                   and c.int_col < 10)
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [LEFT ANTI JOIN]
 |  hash predicates: a.id = c.id
 |  other join predicates: a.tinyint_col = 10, a.int_col = b.int_col
@@ -1875,6 +2011,8 @@ inner join functional.alltypesagg d
   on (a.tinyint_col = d.tinyint_col and a.int_col < 10)
 where a.float_col < b.float_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 08:HASH JOIN [INNER JOIN]
 |  hash predicates: d.tinyint_col = a.tinyint_col
 |  runtime filters: RF000 <- a.tinyint_col
@@ -1914,6 +2052,8 @@ inner join functional.alltypesagg d
   on b.tinyint_col > d.int_col or b.id != d.id
 where a.int_col = 10 and c.bigint_col = d.bigint_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 06:NESTED LOOP JOIN [INNER JOIN]
 |  predicates: a.id < b.id, b.tinyint_col > d.int_col OR b.id != d.id
 |
@@ -1938,6 +2078,8 @@ where a.int_col = 10 and c.bigint_col = d.bigint_col
    partitions=11/11 files=11 size=814.73KB
    runtime filters: RF000 -> d.bigint_col
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 11:EXCHANGE [UNPARTITIONED]
 |
 06:NESTED LOOP JOIN [INNER JOIN, BROADCAST]
@@ -1985,6 +2127,8 @@ inner join
 on vv1.bigint_col > vv2.bigint_col
 where vv1.tinyint_col < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 06:NESTED LOOP JOIN [INNER JOIN]
 |  predicates: a.bigint_col > c.bigint_col
 |
@@ -2021,6 +2165,8 @@ left anti join functional.alltypesagg e
   on c.string_col != e.string_col
 where a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 08:NESTED LOOP JOIN [RIGHT ANTI JOIN]
 |  join predicates: c.string_col != e.string_col
 |
@@ -2055,6 +2201,8 @@ left join (select coalesce(1, 10) as int_col
 from functional.alltypessmall) t2 on t1.id = t2.int_col
 where t2.int_col in (t2.int_col, 10);
 ---- PLAN
+PLAN-ROOT SINK
+|
 03:AGGREGATE [FINALIZE]
 |  output: count(id)
 |
@@ -2074,6 +2222,8 @@ select *
 from functional.testtbl t1 join functional.testtbl t2
 where t1.id <=> t2.id and t1.zip = 94611
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: t1.id IS NOT DISTINCT FROM t2.id
 |  runtime filters: RF000 <- t2.id
@@ -2090,6 +2240,8 @@ select *
 from functional.testtbl t1 join functional.testtbl t2
 where t1.id is not distinct from t2.id and t1.zip = 94611
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: t1.id IS NOT DISTINCT FROM t2.id
 |  runtime filters: RF000 <- t2.id
@@ -2106,6 +2258,8 @@ select *
 from functional.testtbl t1 join functional.testtbl t2
 where (t1.id IS DISTINCT FROM t2.id) and t1.zip = 94611
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:NESTED LOOP JOIN [INNER JOIN]
 |  predicates: (t1.id IS DISTINCT FROM t2.id)
 |
@@ -2122,6 +2276,8 @@ from (select * from functional.alltypestiny) t1
   join (select * from functional.alltypestiny) t2 on (t1.id is not distinct 
from t2.id)
   join functional.alltypestiny t3 on (coalesce(t1.id, t2.id) is not distinct 
from t3.id)
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [INNER JOIN]
 |  hash predicates: coalesce(functional.alltypestiny.id, 
functional.alltypestiny.id) IS NOT DISTINCT FROM t3.id
 |
@@ -2152,6 +2308,8 @@ and a.tinyint_col + b.tinyint_col < 15
 and a.float_col - c.double_col < 0
 and (b.double_col * c.tinyint_col > 1000 or c.tinyint_col < 1000)
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:HASH JOIN [LEFT OUTER JOIN]
 |  hash predicates: c.id = a.id, c.string_col IS NOT DISTINCT FROM b.string_col
 |  other predicates: a.tinyint_col = 15, b.string_col = '15', a.day >= 6, 
b.month > 2, a.float_col - c.double_col < 0, a.tinyint_col + b.tinyint_col < 
15, (b.double_col * c.tinyint_col > 1000 OR c.tinyint_col < 1000)
@@ -2173,6 +2331,8 @@ and (b.double_col * c.tinyint_col > 1000 or c.tinyint_col 
< 1000)
 select t1.d, t2.d from functional.nulltable t1, functional.nulltable t2
 where not(t1.d IS DISTINCT FROM t2.d)
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:NESTED LOOP JOIN [INNER JOIN]
 |  predicates: NOT (t1.d IS DISTINCT FROM t2.d)
 |
@@ -2187,6 +2347,8 @@ from functional.nulltable t1, functional.nulltable t2, 
functional.nulltable t3
 where t1.d IS DISTINCT FROM t2.d
 and t3.a != t2.g
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:NESTED LOOP JOIN [INNER JOIN]
 |  predicates: t3.a != t2.g
 |
@@ -2207,6 +2369,8 @@ and t3.a != t2.g
 # have explain_level=1
 select a.c_custkey as c_custkey from tpch.customer a, tpch.customer b limit 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:NESTED LOOP JOIN [CROSS JOIN]
 |  limit: 1
 |
@@ -2219,6 +2383,8 @@ select a.c_custkey as c_custkey from tpch.customer a, 
tpch.customer b limit 1
 select a.c_custkey as c_custkey from tpch.customer a left semi join 
tpch.customer b
 using (c_custkey) limit 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [LEFT SEMI JOIN]
 |  hash predicates: a.c_custkey = b.c_custkey
 |  runtime filters: RF000 <- b.c_custkey
@@ -2234,6 +2400,8 @@ using (c_custkey) limit 1
 select b.c_custkey as c_custkey from tpch.customer a right semi join 
tpch.customer b
 using (c_custkey) limit 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT SEMI JOIN]
 |  hash predicates: a.c_custkey = b.c_custkey
 |  runtime filters: RF000 <- b.c_custkey
@@ -2249,6 +2417,8 @@ using (c_custkey) limit 1
 select a.c_custkey as c_custkey from tpch.customer a left outer join 
tpch.customer b
 using (c_custkey) limit 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [LEFT OUTER JOIN]
 |  hash predicates: a.c_custkey = b.c_custkey
 |  limit: 1
@@ -2262,6 +2432,8 @@ using (c_custkey) limit 1
 select b.c_custkey as c_custkey from tpch.customer a right outer join 
tpch.customer b
 using (c_custkey) limit 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT OUTER JOIN]
 |  hash predicates: a.c_custkey = b.c_custkey
 |  runtime filters: RF000 <- b.c_custkey
@@ -2277,6 +2449,8 @@ using (c_custkey) limit 1
 select a.c_custkey as c_custkey from tpch.customer a full outer join 
tpch.customer b
 using (c_custkey) limit 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [FULL OUTER JOIN]
 |  hash predicates: a.c_custkey = b.c_custkey
 |  limit: 1
@@ -2290,6 +2464,8 @@ using (c_custkey) limit 1
 select a.c_custkey as c_custkey from tpch.customer a left anti join 
tpch.customer b
 using (c_custkey) limit 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [LEFT ANTI JOIN]
 |  hash predicates: a.c_custkey = b.c_custkey
 |  limit: 1
@@ -2303,6 +2479,8 @@ using (c_custkey) limit 1
 select b.c_custkey as c_custkey from tpch.customer a right anti join 
tpch.customer b
 using (c_custkey) limit 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 02:HASH JOIN [RIGHT ANTI JOIN]
 |  hash predicates: a.c_custkey = b.c_custkey
 |  limit: 1

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
----------------------------------------------------------------------
diff --git 
a/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
 
b/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
index 73acf7f..a187ed8 100644
--- 
a/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
+++ 
b/testdata/workloads/functional-planner/queries/PlannerTest/kudu-selectivity.test
@@ -1,12 +1,16 @@
 select * from functional_kudu.zipcode_incomes where id = '8600000US00601'
 ---- PLAN
 F00:PLAN FRAGMENT [UNPARTITIONED]
+  PLAN-ROOT SINK
+  |
   00:SCAN KUDU [functional_kudu.zipcode_incomes]
      kudu predicates: id = '8600000US00601'
      hosts=3 per-host-mem=unavailable
      tuple-ids=0 row-size=124B cardinality=1
 ---- DISTRIBUTEDPLAN
 F01:PLAN FRAGMENT [UNPARTITIONED]
+  PLAN-ROOT SINK
+  |
   01:EXCHANGE [UNPARTITIONED]
      hosts=3 per-host-mem=unavailable
      tuple-ids=0 row-size=124B cardinality=1
@@ -22,6 +26,8 @@ F00:PLAN FRAGMENT [RANDOM]
 select * from functional_kudu.zipcode_incomes where id != '1' and zip = '2'
 ---- PLAN
 F00:PLAN FRAGMENT [UNPARTITIONED]
+  PLAN-ROOT SINK
+  |
   00:SCAN KUDU [functional_kudu.zipcode_incomes]
      predicates: id != '1'
      kudu predicates: zip = '2'
@@ -29,6 +35,8 @@ F00:PLAN FRAGMENT [UNPARTITIONED]
      tuple-ids=0 row-size=124B cardinality=1
 ---- DISTRIBUTEDPLAN
 F01:PLAN FRAGMENT [UNPARTITIONED]
+  PLAN-ROOT SINK
+  |
   01:EXCHANGE [UNPARTITIONED]
      hosts=3 per-host-mem=unavailable
      tuple-ids=0 row-size=124B cardinality=1
@@ -44,12 +52,16 @@ F00:PLAN FRAGMENT [RANDOM]
 select * from functional_kudu.zipcode_incomes where id > '1' and zip > '2'
 ---- PLAN
 F00:PLAN FRAGMENT [UNPARTITIONED]
+  PLAN-ROOT SINK
+  |
   00:SCAN KUDU [functional_kudu.zipcode_incomes]
      kudu predicates: zip > '2', id > '1'
      hosts=3 per-host-mem=unavailable
      tuple-ids=0 row-size=124B cardinality=3317
 ---- DISTRIBUTEDPLAN
 F01:PLAN FRAGMENT [UNPARTITIONED]
+  PLAN-ROOT SINK
+  |
   01:EXCHANGE [UNPARTITIONED]
      hosts=3 per-host-mem=unavailable
      tuple-ids=0 row-size=124B cardinality=3317
@@ -64,12 +76,16 @@ F00:PLAN FRAGMENT [RANDOM]
 select * from functional_kudu.zipcode_incomes where id = '1' or id = '2'
 ---- PLAN
 F00:PLAN FRAGMENT [UNPARTITIONED]
+  PLAN-ROOT SINK
+  |
   00:SCAN KUDU [functional_kudu.zipcode_incomes]
      predicates: id = '1' OR id = '2'
      hosts=3 per-host-mem=unavailable
      tuple-ids=0 row-size=124B cardinality=2
 ---- DISTRIBUTEDPLAN
 F01:PLAN FRAGMENT [UNPARTITIONED]
+  PLAN-ROOT SINK
+  |
   01:EXCHANGE [UNPARTITIONED]
      hosts=3 per-host-mem=unavailable
      tuple-ids=0 row-size=124B cardinality=2

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
----------------------------------------------------------------------
diff --git 
a/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test 
b/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
index 565f3a3..06ce157 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
@@ -1,5 +1,7 @@
 select * from functional_kudu.testtbl
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN KUDU [functional_kudu.testtbl]
 ---- SCANRANGELOCATIONS
 NODE 0:
@@ -7,12 +9,16 @@ NODE 0:
   ScanToken{table=testtbl, range-partition: [(int64 id=1007), <end>)}
   ScanToken{table=testtbl, range-partition: [<start>, (int64 id=1003))}
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 01:EXCHANGE [UNPARTITIONED]
 |
 00:SCAN KUDU [functional_kudu.testtbl]
 ====
 select * from functional_kudu.testtbl where name = '10'
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN KUDU [functional_kudu.testtbl]
    kudu predicates: name = '10'
 ---- SCANRANGELOCATIONS
@@ -21,6 +27,8 @@ NODE 0:
   ScanToken{table=testtbl, range-partition: [(int64 id=1007), <end>)}
   ScanToken{table=testtbl, range-partition: [<start>, (int64 id=1003))}
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 01:EXCHANGE [UNPARTITIONED]
 |
 00:SCAN KUDU [functional_kudu.testtbl]
@@ -28,6 +36,8 @@ NODE 0:
 ====
 select * from functional_kudu.testtbl where name = NULL
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN KUDU [functional_kudu.testtbl]
    predicates: name = NULL
 ====
@@ -95,12 +105,16 @@ select * from functional_kudu.testtbl
 where id >= 10 and zip <= 5 and 20 >= id and 'foo' = name and zip >= 0 and 30 
>= zip
 and zip > 1 and zip < 50
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN KUDU [functional_kudu.testtbl]
    kudu predicates: id <= 20, zip <= 30, id >= 10, zip < 50, zip <= 5, zip > 
1, zip >= 0, name = 'foo'
 ---- SCANRANGELOCATIONS
 NODE 0:
   ScanToken{table=testtbl, range-partition: [<start>, (int64 id=1003))}
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 01:EXCHANGE [UNPARTITIONED]
 |
 00:SCAN KUDU [functional_kudu.testtbl]
@@ -110,6 +124,8 @@ NODE 0:
 select * from functional_kudu.testtbl
 where id < 10 + 30  and cast(sin(id) as boolean) = true and 20 * 3 >= id and 
10 * 10 + 3 > id
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN KUDU [functional_kudu.testtbl]
    predicates: CAST(sin(id) AS BOOLEAN) = TRUE
    kudu predicates: id <= 60, id < 40, id < 103
@@ -117,6 +133,8 @@ where id < 10 + 30  and cast(sin(id) as boolean) = true and 
20 * 3 >= id and 10
 NODE 0:
   ScanToken{table=testtbl, range-partition: [<start>, (int64 id=1003))}
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 01:EXCHANGE [UNPARTITIONED]
 |
 00:SCAN KUDU [functional_kudu.testtbl]
@@ -127,6 +145,8 @@ NODE 0:
 select * from functional_kudu.testtbl
 where cast(sin(id) as boolean) = true and name = 'a'
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN KUDU [functional_kudu.testtbl]
    predicates: CAST(sin(id) AS BOOLEAN) = TRUE
    kudu predicates: name = 'a'
@@ -136,6 +156,8 @@ NODE 0:
   ScanToken{table=testtbl, range-partition: [(int64 id=1007), <end>)}
   ScanToken{table=testtbl, range-partition: [<start>, (int64 id=1003))}
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 01:EXCHANGE [UNPARTITIONED]
 |
 00:SCAN KUDU [functional_kudu.testtbl]
@@ -147,6 +169,8 @@ NODE 0:
 select * from functional_kudu.testtbl
 where cast(sin(id) as boolean) = true and name is null
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN KUDU [functional_kudu.testtbl]
    predicates: name IS NULL, CAST(sin(id) AS BOOLEAN) = TRUE
 ---- SCANRANGELOCATIONS
@@ -155,6 +179,8 @@ NODE 0:
   ScanToken{table=testtbl, range-partition: [(int64 id=1007), <end>)}
   ScanToken{table=testtbl, range-partition: [<start>, (int64 id=1003))}
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 01:EXCHANGE [UNPARTITIONED]
 |
 00:SCAN KUDU [functional_kudu.testtbl]
@@ -163,6 +189,8 @@ NODE 0:
 # IMPALA-3856: KuduScanNode crash when pushing predicates including a cast
 select o_orderkey from tpch_kudu.orders where o_orderkey < 10.0 order by 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SORT
 |  order by: o_orderkey ASC
 |
@@ -174,6 +202,8 @@ select t.c from
   (select cast(o_orderdate as timestamp) c from tpch_kudu.orders where 
o_orderkey < 10) t
 where t.c <= cast('1995-01-01 00:00:00' as timestamp) order by c
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SORT
 |  order by: c ASC
 |
@@ -186,6 +216,8 @@ select count(*) from functional_kudu.alltypes
 where id < 1475059765 + 10
 and 1475059765 + 100 < id
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:AGGREGATE [FINALIZE]
 |  output: count(*)
 |

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/testdata/workloads/functional-planner/queries/PlannerTest/mem-limit-broadcast-join.test
----------------------------------------------------------------------
diff --git 
a/testdata/workloads/functional-planner/queries/PlannerTest/mem-limit-broadcast-join.test
 
b/testdata/workloads/functional-planner/queries/PlannerTest/mem-limit-broadcast-join.test
index 2e73767..689e78a 100644
--- 
a/testdata/workloads/functional-planner/queries/PlannerTest/mem-limit-broadcast-join.test
+++ 
b/testdata/workloads/functional-planner/queries/PlannerTest/mem-limit-broadcast-join.test
@@ -2,6 +2,8 @@ select * from tpch.nation n1
 join[broadcast]
 tpch.nation n2 on n1.n_regionkey=n2.n_regionkey
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 04:EXCHANGE [UNPARTITIONED]
 |
 02:HASH JOIN [INNER JOIN, BROADCAST]

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/testdata/workloads/functional-planner/queries/PlannerTest/nested-collections.test
----------------------------------------------------------------------
diff --git 
a/testdata/workloads/functional-planner/queries/PlannerTest/nested-collections.test
 
b/testdata/workloads/functional-planner/queries/PlannerTest/nested-collections.test
index 6270c11..da2e17f 100644
--- 
a/testdata/workloads/functional-planner/queries/PlannerTest/nested-collections.test
+++ 
b/testdata/workloads/functional-planner/queries/PlannerTest/nested-collections.test
@@ -2,6 +2,8 @@
 select * from functional.allcomplextypes.int_array_col
 where item > 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN HDFS [functional.allcomplextypes.int_array_col]
    partitions=0/0 files=0 size=0B
    predicates: item > 10
@@ -10,6 +12,8 @@ where item > 10
 select * from functional.allcomplextypes.int_map_col
 where key = 'test' and value < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 00:SCAN HDFS [functional.allcomplextypes.int_map_col]
    partitions=0/0 files=0 size=0B
    predicates: value < 10, key = 'test'
@@ -18,6 +22,8 @@ where key = 'test' and value < 10
 select count(f21) from 
functional.allcomplextypes.complex_nested_struct_col.f2.f12
 where key = 'test'
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:AGGREGATE [FINALIZE]
 |  output: count(f21)
 |
@@ -32,6 +38,8 @@ inner join functional.allcomplextypes.struct_map_col b
 on (a.f1 = b.f1)
 where a.f2 = 'test_a' and b.f2 = 'test_b'
 ---- PLAN
+PLAN-ROOT SINK
+|
 03:AGGREGATE [FINALIZE]
 |  output: count(*)
 |
@@ -58,6 +66,8 @@ select 1 from
 where c_nationkey = n_nationkey and s_nationkey = n_nationkey
   and c_comment = s_comment and n_comment = s_comment
 ---- PLAN
+PLAN-ROOT SINK
+|
 08:HASH JOIN [INNER JOIN]
 |  hash predicates: c_nationkey = n_nationkey, s_comment = n_comment
 |  runtime filters: RF000 <- n_nationkey, RF001 <- n_comment
@@ -90,6 +100,8 @@ where c_nationkey = n_nationkey and s_nationkey = n_nationkey
 select a.id, b.item from functional.allcomplextypes a cross join 
a.int_array_col b
 where a.id < 10 and b.item % 2 = 0
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [CROSS JOIN]
@@ -107,6 +119,8 @@ where a.id < 10 and b.item % 2 = 0
 select a.id from functional.allcomplextypes a left semi join a.int_array_col b
 where a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT SEMI JOIN]
@@ -123,6 +137,8 @@ where a.id < 10
 select b.item from functional.allcomplextypes a right semi join 
a.int_array_col b
 where b.item % 2 = 0
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [LEFT SEMI JOIN]
@@ -140,6 +156,8 @@ where b.item % 2 = 0
 select a.id from functional.allcomplextypes a left anti join a.int_array_col b
 where a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT ANTI JOIN]
@@ -157,6 +175,8 @@ select a.id from functional.allcomplextypes a
 left anti join (select * from a.int_array_col) v
 where a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT ANTI JOIN]
@@ -173,6 +193,8 @@ where a.id < 10
 select b.item from functional.allcomplextypes a right anti join 
a.int_array_col b
 where b.item % 2 = 0
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [LEFT ANTI JOIN]
@@ -189,6 +211,8 @@ where b.item % 2 = 0
 select a.id, b.item from functional.allcomplextypes a left outer join 
a.int_array_col b
 where a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT OUTER JOIN]
@@ -206,6 +230,8 @@ select a.id, v.item from functional.allcomplextypes a
 left outer join (select * from a.int_array_col) v
 where a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT OUTER JOIN]
@@ -222,6 +248,8 @@ where a.id < 10
 select a.id, b.item from functional.allcomplextypes a right outer join 
a.int_array_col b
 where b.item % 2 = 0
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [LEFT OUTER JOIN]
@@ -238,6 +266,8 @@ where b.item % 2 = 0
 select a.id, b.item from functional.allcomplextypes a full outer join 
a.int_array_col b
 where b.item % 2 = 0 and a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [FULL OUTER JOIN]
@@ -256,6 +286,8 @@ where b.item % 2 = 0 and a.id < 10
 select a.id, b.item from functional.allcomplextypes a, a.int_array_col b
 where a.id < 10 and b.item % 2 = 0 and a.id < b.item
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [INNER JOIN]
@@ -275,6 +307,8 @@ select a.id from functional.allcomplextypes a
   left semi join a.int_array_col b on (a.id < b.item and b.item % 2 = 0)
 where a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT SEMI JOIN]
@@ -294,6 +328,8 @@ select b.item from functional.allcomplextypes a
   right semi join a.int_array_col b on (a.id < b.item and a.id < 10)
 where b.item % 2 = 0
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [LEFT SEMI JOIN]
@@ -313,6 +349,8 @@ select a.id from functional.allcomplextypes a
   left anti join a.int_array_col b on (a.id < b.item and b.item % 2 = 0)
 where a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT ANTI JOIN]
@@ -332,6 +370,8 @@ select b.item from functional.allcomplextypes a
   right anti join a.int_array_col b on (a.id < b.item and a.id < 10)
 where b.item % 2 = 0
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [LEFT ANTI JOIN]
@@ -351,6 +391,8 @@ select a.id, b.item from functional.allcomplextypes a
   left outer join a.int_array_col b on (a.id < b.item and b.item % 2 = 0)
 where a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT OUTER JOIN]
@@ -370,6 +412,8 @@ select a.id, b.item from functional.allcomplextypes a
   right outer join a.int_array_col b on (a.id < b.item and a.id < 10)
 where b.item % 2 = 0
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [LEFT OUTER JOIN]
@@ -389,6 +433,8 @@ select a.id, b.item from functional.allcomplextypes a
   full outer join a.int_array_col b on (a.id < b.item and a.id < 10)
 where b.item % 2 = 0
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [FULL OUTER JOIN]
@@ -408,6 +454,8 @@ select a.id, b.f1, b.f2 from functional.allcomplextypes a
   inner join a.struct_array_col b
 where a.id < 10 and b.f1 % 2 = 0 and b.f1 = a.id and b.f1 < a.year
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [INNER JOIN]
@@ -429,6 +477,8 @@ select a.id from functional.allcomplextypes a
     on (b.f1 % 2 = 0 and b.f1 = a.id and b.f1 < a.year)
 where a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT SEMI JOIN]
@@ -449,6 +499,8 @@ select b.f1, b.f2 from functional.allcomplextypes a
     on (a.id < 10 and b.f1 = a.id and b.f1 < a.year)
 where b.f1 % 2 = 0
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [LEFT SEMI JOIN]
@@ -469,6 +521,8 @@ select a.id from functional.allcomplextypes a
     on (b.f1 % 2 = 0 and b.f1 = a.id and b.f1 < a.year)
 where a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT ANTI JOIN]
@@ -489,6 +543,8 @@ select b.f1, b.f2 from functional.allcomplextypes a
     on (a.id < 10 and b.f1 = a.id and b.f1 < a.year)
 where b.f1 % 2 = 0
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [LEFT ANTI JOIN]
@@ -509,6 +565,8 @@ select a.id from functional.allcomplextypes a
     on (b.f1 % 2 = 0 and b.f1 = a.id and b.f1 < a.year)
 where a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT OUTER JOIN]
@@ -529,6 +587,8 @@ select b.f1, b.f2 from functional.allcomplextypes a
     on (a.id < 10 and b.f1 = a.id and b.f1 < a.year)
 where b.f1 % 2 = 0
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [LEFT OUTER JOIN]
@@ -549,6 +609,8 @@ select b.f1, b.f2 from functional.allcomplextypes a
     on (b.f1 = a.id and b.f1 < a.year)
 where a.id < 10 and b.f1 % 2 = 0
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [FULL OUTER JOIN]
@@ -572,6 +634,8 @@ select 1 from functional.allcomplextypes a
   left outer join a.struct_array_col d on (a.month < 4 or d.f1 > 5)
   inner join a.struct_map_col e on (e.f1 = a.id and e.f2 = 'test')
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--10:HASH JOIN [INNER JOIN]
@@ -612,6 +676,8 @@ inner join functional.alltypes d on (b.id = d.id)
 inner join a.struct_array_col e
 where e.f1 < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 11:HASH JOIN [INNER JOIN]
 |  hash predicates: d.id = b.id
 |  runtime filters: RF000 <- b.id
@@ -656,6 +722,8 @@ inner join a.struct_array_col e
 right anti join functional.alltypessmall c on (b.int_col = c.int_col and e.f1 
< 10)
 inner join functional.alltypes d on (b.id = d.id)
 ---- PLAN
+PLAN-ROOT SINK
+|
 11:HASH JOIN [INNER JOIN]
 |  hash predicates: d.id = b.id
 |
@@ -697,6 +765,8 @@ inner join a.int_map_col d
 left semi join functional.alltypes e on (d.value = e.id)
 where b.item < 10 and c.int_col > 30
 ---- PLAN
+PLAN-ROOT SINK
+|
 12:HASH JOIN [RIGHT SEMI JOIN]
 |  hash predicates: e.id = d.value
 |  runtime filters: RF000 <- d.value
@@ -745,6 +815,8 @@ inner join a.int_map_col d
 right anti join functional.alltypestiny e on (d.value = e.id)
 where b.item < 10 and c.int_col > 30
 ---- PLAN
+PLAN-ROOT SINK
+|
 12:HASH JOIN [RIGHT ANTI JOIN]
 |  hash predicates: d.value = e.id
 |
@@ -785,6 +857,8 @@ from functional.allcomplextypes a,
   (select count(*) cnt from a.int_array_col) v
 where v.cnt < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--05:NESTED LOOP JOIN [CROSS JOIN]
@@ -806,6 +880,8 @@ from functional.allcomplextypes a,
   (select f1, count(*) cnt from a.struct_array_col group by f1) v
 where v.cnt < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--05:NESTED LOOP JOIN [CROSS JOIN]
@@ -828,6 +904,8 @@ from functional.allcomplextypes a,
   (select * from a.struct_array_col order by f1 limit 10) v
 where v.f2 = 'test'
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--06:NESTED LOOP JOIN [CROSS JOIN]
@@ -851,6 +929,8 @@ from functional.allcomplextypes a,
   (select key, row_number() over (order by key) rnum from a.int_map_col) v
 where v.key != 'bad'
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--07:NESTED LOOP JOIN [CROSS JOIN]
@@ -877,6 +957,8 @@ where v.key != 'bad'
 select a.id from functional.allcomplextypes a
 where id < (select avg(item) from a.int_array_col)
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--05:NESTED LOOP JOIN [RIGHT SEMI JOIN]
@@ -896,6 +978,8 @@ where id < (select avg(item) from a.int_array_col)
 select a.id from functional.allcomplextypes a
 where exists (select item from a.int_array_col where item > 100)
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT SEMI JOIN]
@@ -912,6 +996,8 @@ where exists (select item from a.int_array_col where item > 
100)
 select a.id from functional.allcomplextypes a
 where not exists (select item from a.int_array_col)
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT ANTI JOIN]
@@ -928,6 +1014,8 @@ where not exists (select item from a.int_array_col)
 select a.id from functional.allcomplextypes a
 where exists (select m.key from a.struct_map_col m where a.id < m.f1)
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT SEMI JOIN]
@@ -945,6 +1033,8 @@ where exists (select m.key from a.struct_map_col m where 
a.id < m.f1)
 select a.id from functional.allcomplextypes a
 where not exists (select c.f2 from a.struct_array_col c where a.id < c.f1)
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT ANTI JOIN]
@@ -964,6 +1054,8 @@ select a.id
 from functional.allcomplextypes a
 where id in (select b.item from a.int_array_col b where a.year < b.item)
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT SEMI JOIN]
@@ -984,6 +1076,8 @@ select a.id
 from functional.allcomplextypes a
 where id not in (select b.item from a.int_array_col b)
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:HASH JOIN [NULL AWARE LEFT ANTI JOIN]
@@ -1004,6 +1098,8 @@ select a.id
 from functional.allcomplextypes a
 where id not in (select b.item from a.int_array_col b where a.year < b.item)
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:HASH JOIN [NULL AWARE LEFT ANTI JOIN]
@@ -1028,6 +1124,8 @@ cross join
    from a.struct_array_col x inner join b.struct_map_col y
    on x.f1 = y.f1) v
 ---- PLAN
+PLAN-ROOT SINK
+|
 17:SUBPLAN
 |
 |--15:NESTED LOOP JOIN [CROSS JOIN]
@@ -1085,6 +1183,8 @@ cross join
    on x.f1 = y.f1) v
 where b.id = d.value
 ---- PLAN
+PLAN-ROOT SINK
+|
 17:SUBPLAN
 |
 |--15:NESTED LOOP JOIN [CROSS JOIN]
@@ -1145,6 +1245,8 @@ cross join
    from a.struct_array_col x inner join b.struct_map_col y
    on x.f1 = y.f1) v
 ---- PLAN
+PLAN-ROOT SINK
+|
 15:SUBPLAN
 |
 |--13:NESTED LOOP JOIN [CROSS JOIN]
@@ -1192,6 +1294,8 @@ from tpch_nested_parquet.customer c, c.c_orders o, 
o.o_lineitems
 where c_custkey < 10 and o_orderkey < 5 and l_linenumber < 3
 limit 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |  limit: 10
 |
@@ -1222,6 +1326,8 @@ cross join
   (select m1.key from a.map_map_col m1,
    (select m2.key from m1.value m2) v1) v2
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--08:NESTED LOOP JOIN [CROSS JOIN]
@@ -1249,6 +1355,8 @@ inner join functional.allcomplextypes b on (a.id = b.id)
 cross join (select count(*) c from a.int_map_col) v1
 cross join (select avg(item) a from b.int_array_col) v2
 ---- PLAN
+PLAN-ROOT SINK
+|
 10:SUBPLAN
 |
 |--08:NESTED LOOP JOIN [CROSS JOIN]
@@ -1288,6 +1396,8 @@ where c.c_custkey = o.o_orderkey and c.c_custkey = 
o.o_shippriority
 # redundant predicates
   and o.o_orderkey = l.l_partkey and o.o_shippriority = l.l_suppkey
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--08:NESTED LOOP JOIN [INNER JOIN]
@@ -1327,6 +1437,8 @@ where a.item between 10 and 20
   and v2.key = 'test2'
   and v2.x = 'test3'
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--12:NESTED LOOP JOIN [INNER JOIN]
@@ -1376,6 +1488,8 @@ where s.s_suppkey not in
  inner join s.s_partsupps ps3
    on ps2.ps_comment = ps3.ps_comment)
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--08:HASH JOIN [NULL AWARE LEFT ANTI JOIN]
@@ -1404,6 +1518,8 @@ left outer join functional.allcomplextypes t2 ON (t1.id = 
t2.id)
 # The subplan for this table ref must come after the outer join of t1 and t2.
 inner join t2.int_array_col
 ---- PLAN
+PLAN-ROOT SINK
+|
 06:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [CROSS JOIN]
@@ -1426,6 +1542,8 @@ select a from functional.allcomplextypes t,
   (select count(*) over(partition by key) a
    from t.int_map_col group by key) v
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--07:NESTED LOOP JOIN [CROSS JOIN]
@@ -1454,6 +1572,8 @@ left outer join functional.allcomplextypes t2 on (t1.id = 
t2.id)
 inner join t1.map_map_col m1
 inner join m1.value m2
 ---- PLAN
+PLAN-ROOT SINK
+|
 10:SUBPLAN
 |
 |--08:SUBPLAN
@@ -1494,6 +1614,8 @@ where c.c_custkey in
  left outer join c.c_orders o3 on o3.pos = o2.pos
  where c.c_custkey = o2.pos)
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--12:HASH JOIN [RIGHT SEMI JOIN]
@@ -1538,6 +1660,8 @@ where c.c_custkey in
  inner join c.c_orders o3 on o3.pos = o2.pos
  where c.c_custkey = o2.pos)
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--12:HASH JOIN [RIGHT SEMI JOIN]
@@ -1580,6 +1704,8 @@ where c.c_custkey in
  left outer join o2.o_lineitems l
  where c.c_custkey = o2.pos)
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--12:HASH JOIN [RIGHT SEMI JOIN]
@@ -1616,6 +1742,8 @@ where c.c_custkey in
 select straight_join id from functional.allcomplextypes t1 left outer join 
t1.int_array_col t2
 where t1.id = t2.pos and t1.int_struct_col.f1 = 10 and t2.item = 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:HASH JOIN [LEFT OUTER JOIN]
@@ -1637,6 +1765,8 @@ where t1.id = t2.pos and t1.int_struct_col.f1 = 10 and 
t2.item = 1
 select straight_join id from functional.allcomplextypes t1 right outer join 
t1.int_array_col t2
 where t1.id = t2.pos and t1.int_struct_col.f1 = 10 and t2.item = 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:HASH JOIN [RIGHT OUTER JOIN]
@@ -1658,6 +1788,8 @@ where t1.id = t2.pos and t1.int_struct_col.f1 = 10 and 
t2.item = 1
 select id from functional.allcomplextypes t1 full outer join t1.int_array_col 
t2
 where t1.id = t2.pos and t1.int_struct_col.f1 = 10 and t2.item = 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [FULL OUTER JOIN]
@@ -1679,6 +1811,8 @@ select id from functional.allcomplextypes t1 left outer 
join
   (select pos, item from t1.int_array_col t2) v
  where t1.id > v.pos and v.item = 1
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT OUTER JOIN]
@@ -1698,6 +1832,8 @@ left outer join c.c_orders o
 # Has an ordering dependency on c and o
 inner join o.o_lineitems
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--08:SUBPLAN
@@ -1724,6 +1860,8 @@ inner join c1.c_orders
 right outer join tpch_nested_parquet.customer c2
   on c1.c_custkey = c2.c_custkey
 ---- PLAN
+PLAN-ROOT SINK
+|
 06:HASH JOIN [RIGHT OUTER JOIN]
 |  hash predicates: c1.c_custkey = c2.c_custkey
 |  runtime filters: RF000 <- c2.c_custkey
@@ -1751,6 +1889,8 @@ full outer join tpch_nested_parquet.customer c2
 inner join c1.c_orders o1
 left semi join c2.c_orders o2
 ---- PLAN
+PLAN-ROOT SINK
+|
 08:SUBPLAN
 |
 |--06:NESTED LOOP JOIN [LEFT SEMI JOIN]
@@ -1778,6 +1918,8 @@ select * from tpch_nested_parquet.customer c
 left outer join c.c_orders o
 where o.o_orderkey is null and o.o_orderstatus <=> o_orderpriority
 ---- PLAN
+PLAN-ROOT SINK
+|
 01:SUBPLAN
 |
 |--04:NESTED LOOP JOIN [RIGHT OUTER JOIN]
@@ -1799,6 +1941,8 @@ left join t2.c_orders t4
 inner join tpch_nested_parquet.region t5 on t5.r_regionkey = t2.c_custkey
 left join t4.item.o_lineitems t6 on t6.item.l_returnflag = 
t4.item.o_orderpriority
 ---- PLAN
+PLAN-ROOT SINK
+|
 14:SUBPLAN
 |
 |--12:SUBPLAN

http://git-wip-us.apache.org/repos/asf/incubator-impala/blob/9f61397f/testdata/workloads/functional-planner/queries/PlannerTest/nested-loop-join.test
----------------------------------------------------------------------
diff --git 
a/testdata/workloads/functional-planner/queries/PlannerTest/nested-loop-join.test
 
b/testdata/workloads/functional-planner/queries/PlannerTest/nested-loop-join.test
index c7f9830..fecec7f 100644
--- 
a/testdata/workloads/functional-planner/queries/PlannerTest/nested-loop-join.test
+++ 
b/testdata/workloads/functional-planner/queries/PlannerTest/nested-loop-join.test
@@ -6,6 +6,8 @@ right outer join functional.alltypesagg c
   on a.smallint_col >= c.smallint_col
 where a.id < 10 and c.bigint_col = 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:NESTED LOOP JOIN [RIGHT OUTER JOIN]
 |  join predicates: a.smallint_col >= c.smallint_col
 |  predicates: a.id < 10
@@ -34,6 +36,8 @@ right semi join functional.alltypesagg d
   on c.tinyint_col < d.bigint_col
 where d.bigint_col < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 04:NESTED LOOP JOIN [RIGHT SEMI JOIN]
 |  join predicates: c.tinyint_col < d.bigint_col
 |
@@ -62,6 +66,8 @@ full outer join functional.alltypes d
   on c.int_col > d.int_col
 where a.bigint_col != c.bigint_col and a.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 06:NESTED LOOP JOIN [FULL OUTER JOIN]
 |  join predicates: c.int_col > d.int_col
 |  predicates: a.bigint_col != c.bigint_col, a.id < 10
@@ -93,6 +99,8 @@ from functional.alltypestiny a right anti join 
functional.alltypessmall b
   on a.id < b.id
 where b.int_col = 5
 ---- PLAN
+PLAN-ROOT SINK
+|
 03:AGGREGATE [FINALIZE]
 |  output: count(*)
 |
@@ -116,6 +124,8 @@ right semi join functional.alltypes d on c.tinyint_col < 
d.tinyint_col
 right anti join functional.alltypesnopart e on d.tinyint_col > e.tinyint_col
 where e.id < 10
 ---- PLAN
+PLAN-ROOT SINK
+|
 09:AGGREGATE [FINALIZE]
 |  output: count(*)
 |
@@ -158,6 +168,8 @@ right semi join functional.alltypes d on c.tinyint_col < 
d.tinyint_col
 right anti join functional.alltypesnopart e on d.tinyint_col > e.tinyint_col
 where e.id < 10
 ---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
 15:AGGREGATE [FINALIZE]
 |  output: count:merge(*)
 |

Reply via email to