testdata/workloads/functional-planner/queries/PlannerTest/topn.test - impala - Git at Google

 select name, zip
 from functional.testtbl
 order by name
 limit 1
 ---- PLAN
 PLAN-ROOT SINK
 |
 01:TOP-N [LIMIT=1]
 |  order by: name ASC
 |  row-size=16B cardinality=0
 |
 00:SCAN HDFS [functional.testtbl]
    partitions=1/1 files=0 size=0B
    row-size=16B cardinality=0
 ---- DISTRIBUTEDPLAN
 PLAN-ROOT SINK
 |
 02:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: name ASC
 |  limit: 1
 |
 01:TOP-N [LIMIT=1]
 |  order by: name ASC
 |  row-size=16B cardinality=0
 |
 00:SCAN HDFS [functional.testtbl]
    partitions=1/1 files=0 size=0B
    row-size=16B cardinality=0
 ====
 select zip, count(*)
 from functional.testtbl
 where name like 'm%'
 group by 1
 order by 2 desc
 limit 10
 ---- PLAN
 PLAN-ROOT SINK
 |
 02:TOP-N [LIMIT=10]
 |  order by: count(*) DESC
 |  row-size=12B cardinality=0
 |
 01:AGGREGATE [FINALIZE]
 |  output: count(*)
 |  group by: zip
 |  row-size=12B cardinality=0
 |
 00:SCAN HDFS [functional.testtbl]
    partitions=1/1 files=0 size=0B
    predicates: name LIKE 'm%'
    row-size=16B cardinality=0
 ---- DISTRIBUTEDPLAN
 PLAN-ROOT SINK
 |
 05:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: count(*) DESC
 |  limit: 10
 |
 02:TOP-N [LIMIT=10]
 |  order by: count(*) DESC
 |  row-size=12B cardinality=0
 |
 04:AGGREGATE [FINALIZE]
 |  output: count:merge(*)
 |  group by: zip
 |  row-size=12B cardinality=0
 |
 03:EXCHANGE [HASH(zip)]
 |
 01:AGGREGATE [STREAMING]
 |  output: count(*)
 |  group by: zip
 |  row-size=12B cardinality=0
 |
 00:SCAN HDFS [functional.testtbl]
    partitions=1/1 files=0 size=0B
    predicates: name LIKE 'm%'
    row-size=16B cardinality=0
 ====
 select int_col, sum(float_col)
 from functional_hbase.alltypessmall
 where id < 5
 group by 1
 order by 2
 limit 4
 ---- PLAN
 PLAN-ROOT SINK
 |
 02:TOP-N [LIMIT=4]
 |  order by: sum(float_col) ASC
 |  row-size=12B cardinality=4
 |
 01:AGGREGATE [FINALIZE]
 |  output: sum(float_col)
 |  group by: int_col
 |  row-size=12B cardinality=5
 |
 00:SCAN HBASE [functional_hbase.alltypessmall]
    predicates: id < 5
    row-size=12B cardinality=5
 ---- DISTRIBUTEDPLAN
 PLAN-ROOT SINK
 |
 05:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: sum(float_col) ASC
 |  limit: 4
 |
 02:TOP-N [LIMIT=4]
 |  order by: sum(float_col) ASC
 |  row-size=12B cardinality=4
 |
 04:AGGREGATE [FINALIZE]
 |  output: sum:merge(float_col)
 |  group by: int_col
 |  row-size=12B cardinality=5
 |
 03:EXCHANGE [HASH(int_col)]
 |
 01:AGGREGATE [STREAMING]
 |  output: sum(float_col)
 |  group by: int_col
 |  row-size=12B cardinality=5
 |
 00:SCAN HBASE [functional_hbase.alltypessmall]
    predicates: id < 5
    row-size=12B cardinality=5
 ====
 select int_col, sum(float_col), min(float_col)
 from functional_hbase.alltypessmall
 group by 1
 order by 2,3 desc
 limit 0
 ---- PLAN
 PLAN-ROOT SINK
 |
 00:EMPTYSET
 ---- DISTRIBUTEDPLAN
 PLAN-ROOT SINK
 |
 00:EMPTYSET
 ====
 # Test correct identification of the implicit aliasing of int_col in the select
 # list to t1.int_col;
 # t2.int_col is not null references enough cols of t2 to prevent a broadcast join
 select t1.int_col
 from functional.alltypessmall t1, functional.alltypessmall t2
 where t1.id = t2.id and t2.int_col is not null
 order by int_col
 limit 2
 ---- PLAN
 PLAN-ROOT SINK
 |
 03:TOP-N [LIMIT=2]
 |  order by: int_col ASC
 |  row-size=4B cardinality=2
 |
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: t1.id = t2.id
 |  runtime filters: RF000 <- t2.id
 |  row-size=16B cardinality=10
 |
 |--01:SCAN HDFS [functional.alltypessmall t2]
 |     partitions=4/4 files=4 size=6.32KB
 |     predicates: t2.int_col IS NOT NULL
 |     row-size=8B cardinality=10
 |
 00:SCAN HDFS [functional.alltypessmall t1]
    partitions=4/4 files=4 size=6.32KB
    runtime filters: RF000 -> t1.id
    row-size=8B cardinality=100
 ---- DISTRIBUTEDPLAN
 PLAN-ROOT SINK
 |
 05:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: int_col ASC
 |  limit: 2
 |
 03:TOP-N [LIMIT=2]
 |  order by: int_col ASC
 |  row-size=4B cardinality=2
 |
 02:HASH JOIN [INNER JOIN, BROADCAST]
 |  hash predicates: t1.id = t2.id
 |  runtime filters: RF000 <- t2.id
 |  row-size=16B cardinality=10
 |
 |--04:EXCHANGE [BROADCAST]
 |  |
 |  01:SCAN HDFS [functional.alltypessmall t2]
 |     partitions=4/4 files=4 size=6.32KB
 |     predicates: t2.int_col IS NOT NULL
 |     row-size=8B cardinality=10
 |
 00:SCAN HDFS [functional.alltypessmall t1]
    partitions=4/4 files=4 size=6.32KB
    runtime filters: RF000 -> t1.id
    row-size=8B cardinality=100
 ====
 # Test that the top-n is on int_col and not on the id column
 select int_col as id from functional.alltypessmall order by id limit 2
 ---- PLAN
 PLAN-ROOT SINK
 |
 01:TOP-N [LIMIT=2]
 |  order by: int_col ASC
 |  row-size=4B cardinality=2
 |
 00:SCAN HDFS [functional.alltypessmall]
    partitions=4/4 files=4 size=6.32KB
    row-size=4B cardinality=100
 ---- DISTRIBUTEDPLAN
 PLAN-ROOT SINK
 |
 02:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: int_col ASC
 |  limit: 2
 |
 01:TOP-N [LIMIT=2]
 |  order by: int_col ASC
 |  row-size=4B cardinality=2
 |
 00:SCAN HDFS [functional.alltypessmall]
    partitions=4/4 files=4 size=6.32KB
    row-size=4B cardinality=100
 ====
 # Test that the top-n is on id and not on int_col
 select int_col as id from functional.alltypessmall order by functional.alltypessmall.id limit 2
 ---- PLAN
 PLAN-ROOT SINK
 |
 01:TOP-N [LIMIT=2]
 |  order by: id ASC
 |  row-size=8B cardinality=2
 |
 00:SCAN HDFS [functional.alltypessmall]
    partitions=4/4 files=4 size=6.32KB
    row-size=8B cardinality=100
 ---- DISTRIBUTEDPLAN
 PLAN-ROOT SINK
 |
 02:MERGING-EXCHANGE [UNPARTITIONED]
 |  order by: id ASC
 |  limit: 2
 |
 01:TOP-N [LIMIT=2]
 |  order by: id ASC
 |  row-size=8B cardinality=2
 |
 00:SCAN HDFS [functional.alltypessmall]
    partitions=4/4 files=4 size=6.32KB
    row-size=8B cardinality=100
 ====
 # Test that the limit/offset is propagated correctly to child nodes. The TOP-N node
 # should have the limit/offset specified in the query. Child sort nodes should have a
 # limit equal to the parent's (limit + offset) and offset equal to 0.
 select t1.int_col from functional.alltypessmall t1, functional.alltypessmall t2
 where t1.id = t2.id and t2.int_col is not null
 order by int_col
 limit 10 offset 5
 ---- PLAN
 PLAN-ROOT SINK
 |
 03:TOP-N [LIMIT=10 OFFSET=5]
 |  order by: int_col ASC
 |  row-size=4B cardinality=10
 |
 02:HASH JOIN [INNER JOIN]
 |  hash predicates: t1.id = t2.id
 |  runtime filters: RF000 <- t2.id
 |  row-size=16B cardinality=10
 |
 |--01:SCAN HDFS [functional.alltypessmall t2]
 |     partitions=4/4 files=4 size=6.32KB
 |     predicates: t2.int_col IS NOT NULL
 |     row-size=8B cardinality=10
 |
 00:SCAN HDFS [functional.alltypessmall t1]
    partitions=4/4 files=4 size=6.32KB
    runtime filters: RF000 -> t1.id
    row-size=8B cardinality=100
 ---- SCANRANGELOCATIONS
 NODE 0:
   HDFS SPLIT hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=1/090101.txt 0:1610
   HDFS SPLIT hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=2/090201.txt 0:1621
   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
 NODE 1:
   HDFS SPLIT hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=1/090101.txt 0:1610
   HDFS SPLIT hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=2/090201.txt 0:1621
   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:MERGING-EXCHANGE [UNPARTITIONED]
 |  offset: 5
 |  order by: int_col ASC
 |  limit: 10
 |
 03:TOP-N [LIMIT=15]
 |  order by: int_col ASC
 |  row-size=4B cardinality=10
 |
 02:HASH JOIN [INNER JOIN, BROADCAST]
 |  hash predicates: t1.id = t2.id
 |  runtime filters: RF000 <- t2.id
 |  row-size=16B cardinality=10
 |
 |--04:EXCHANGE [BROADCAST]
 |  |
 |  01:SCAN HDFS [functional.alltypessmall t2]
 |     partitions=4/4 files=4 size=6.32KB
 |     predicates: t2.int_col IS NOT NULL
 |     row-size=8B cardinality=10
 |
 00:SCAN HDFS [functional.alltypessmall t1]
    partitions=4/4 files=4 size=6.32KB
    runtime filters: RF000 -> t1.id
    row-size=8B cardinality=100
 ====
 # test distributed top-n over a union (IMPALA-831)
 select int_col, bigint_col from
   (select * from functional.alltypes
    union all
    select * from functional.alltypessmall) t
 order by int_col desc limit 10 offset 5
 ---- PLAN
 PLAN-ROOT SINK
 |
 03:TOP-N [LIMIT=10 OFFSET=5]
 |  order by: int_col DESC
 |  row-size=12B cardinality=10
 |
 00:UNION
 |  pass-through-operands: all
 |  row-size=12B cardinality=7.40K
 |
 |--02:SCAN HDFS [functional.alltypessmall]
 |     partitions=4/4 files=4 size=6.32KB
 |     row-size=12B cardinality=100
 |
 01:SCAN HDFS [functional.alltypes]
    partitions=24/24 files=24 size=478.45KB
    row-size=12B cardinality=7.30K
 ---- DISTRIBUTEDPLAN
 PLAN-ROOT SINK
 |
 04:MERGING-EXCHANGE [UNPARTITIONED]
 |  offset: 5
 |  order by: int_col DESC
 |  limit: 10
 |
 03:TOP-N [LIMIT=15]
 |  order by: int_col DESC
 |  row-size=12B cardinality=15
 |
 00:UNION
 |  pass-through-operands: all
 |  row-size=12B cardinality=7.40K
 |
 |--02:SCAN HDFS [functional.alltypessmall]
 |     partitions=4/4 files=4 size=6.32KB
 |     row-size=12B cardinality=100
 |
 01:SCAN HDFS [functional.alltypes]
    partitions=24/24 files=24 size=478.45KB
    row-size=12B cardinality=7.30K
 ====
 # test distributed top-n over a union distinct (IMPALA-831)
 select int_col, bigint_col from
   (select * from functional.alltypes
    union distinct
    select * from functional.alltypessmall) t
 order by int_col desc limit 10 offset 5
 ---- PLAN
 PLAN-ROOT SINK
 |
 04:TOP-N [LIMIT=10 OFFSET=5]
 |  order by: int_col DESC
 |  row-size=12B cardinality=10
 |
 03:AGGREGATE [FINALIZE]
 |  group by: id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col, year, month
 |  row-size=89B cardinality=7.40K
 |
 00:UNION
 |  pass-through-operands: all
 |  row-size=89B cardinality=7.40K
 |
 |--02:SCAN HDFS [functional.alltypessmall]
 |     partitions=4/4 files=4 size=6.32KB
 |     row-size=89B cardinality=100
 |
 01:SCAN HDFS [functional.alltypes]
    partitions=24/24 files=24 size=478.45KB
    row-size=89B cardinality=7.30K
 ---- DISTRIBUTEDPLAN
 PLAN-ROOT SINK
 |
 07:MERGING-EXCHANGE [UNPARTITIONED]
 |  offset: 5
 |  order by: int_col DESC
 |  limit: 10
 |
 04:TOP-N [LIMIT=15]
 |  order by: int_col DESC
 |  row-size=12B cardinality=15
 |
 06:AGGREGATE [FINALIZE]
 |  group by: id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col, year, month
 |  row-size=89B cardinality=7.40K
 |
 05:EXCHANGE [HASH(id,bool_col,tinyint_col,smallint_col,int_col,bigint_col,float_col,double_col,date_string_col,string_col,timestamp_col,year,month)]
 |
 03:AGGREGATE [STREAMING]
 |  group by: id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col, year, month
 |  row-size=89B cardinality=7.40K
 |
 00:UNION
 |  pass-through-operands: all
 |  row-size=89B cardinality=7.40K
 |
 |--02:SCAN HDFS [functional.alltypessmall]
 |     partitions=4/4 files=4 size=6.32KB
 |     row-size=89B cardinality=100
 |
 01:SCAN HDFS [functional.alltypes]
    partitions=24/24 files=24 size=478.45KB
    row-size=89B cardinality=7.30K
 ====
 # test that top-n is not placed below an unpartitioned exchange with a limit
 select * from (select * from functional.alltypes limit 10) t
 order by int_col limit 10
 ---- PLAN
 PLAN-ROOT SINK
 |
 01:TOP-N [LIMIT=10]
 |  order by: int_col ASC
 |  row-size=89B cardinality=10
 |
 00:SCAN HDFS [functional.alltypes]
    partitions=24/24 files=24 size=478.45KB
    limit: 10
    row-size=89B cardinality=10
 ---- DISTRIBUTEDPLAN
 PLAN-ROOT SINK
 |
 01:TOP-N [LIMIT=10]
 |  order by: int_col ASC
 |  row-size=89B cardinality=10
 |
 02:EXCHANGE [UNPARTITIONED]
 |  limit: 10
 |
 00:SCAN HDFS [functional.alltypes]
    partitions=24/24 files=24 size=478.45KB
    limit: 10
    row-size=89B cardinality=10
 ====
 # test that top-n is not placed below an unpartitioned exchange with a limit
 select * from
   (select * from functional.alltypes
    union all
    (select * from functional.alltypessmall) limit 10) t
 order by int_col limit 10
 ---- PLAN
 PLAN-ROOT SINK
 |
 03:TOP-N [LIMIT=10]
 |  order by: int_col ASC
 |  row-size=89B cardinality=10
 |
 00:UNION
 |  pass-through-operands: all
 |  limit: 10
 |  row-size=89B cardinality=10
 |
 |--02:SCAN HDFS [functional.alltypessmall]
 |     partitions=4/4 files=4 size=6.32KB
 |     row-size=89B cardinality=100
 |
 01:SCAN HDFS [functional.alltypes]
    partitions=24/24 files=24 size=478.45KB
    row-size=89B cardinality=7.30K
 ---- DISTRIBUTEDPLAN
 PLAN-ROOT SINK
 |
 03:TOP-N [LIMIT=10]
 |  order by: int_col ASC
 |  row-size=89B cardinality=10
 |
 04:EXCHANGE [UNPARTITIONED]
 |  limit: 10
 |
 00:UNION
 |  pass-through-operands: all
 |  limit: 10
 |  row-size=89B cardinality=10
 |
 |--02:SCAN HDFS [functional.alltypessmall]
 |     partitions=4/4 files=4 size=6.32KB
 |     row-size=89B cardinality=100
 |
 01:SCAN HDFS [functional.alltypes]
    partitions=24/24 files=24 size=478.45KB
    row-size=89B cardinality=7.30K
 ====
	select name, zip
	from functional.testtbl
	order by name
	limit 1
	---- PLAN
	PLAN-ROOT SINK
	\|
	01:TOP-N [LIMIT=1]
	\| order by: name ASC
	\| row-size=16B cardinality=0
	\|
	00:SCAN HDFS [functional.testtbl]
	partitions=1/1 files=0 size=0B
	row-size=16B cardinality=0
	---- DISTRIBUTEDPLAN
	PLAN-ROOT SINK
	\|
	02:MERGING-EXCHANGE [UNPARTITIONED]
	\| order by: name ASC
	\| limit: 1
	\|
	01:TOP-N [LIMIT=1]
	\| order by: name ASC
	\| row-size=16B cardinality=0
	\|
	00:SCAN HDFS [functional.testtbl]
	partitions=1/1 files=0 size=0B
	row-size=16B cardinality=0
	====
	select zip, count(*)
	from functional.testtbl
	where name like 'm%'
	group by 1
	order by 2 desc
	limit 10
	---- PLAN
	PLAN-ROOT SINK
	\|
	02:TOP-N [LIMIT=10]
	\| order by: count(*) DESC
	\| row-size=12B cardinality=0
	\|
	01:AGGREGATE [FINALIZE]
	\| output: count(*)
	\| group by: zip
	\| row-size=12B cardinality=0
	\|
	00:SCAN HDFS [functional.testtbl]
	partitions=1/1 files=0 size=0B
	predicates: name LIKE 'm%'
	row-size=16B cardinality=0
	---- DISTRIBUTEDPLAN
	PLAN-ROOT SINK
	\|
	05:MERGING-EXCHANGE [UNPARTITIONED]
	\| order by: count(*) DESC
	\| limit: 10
	\|
	02:TOP-N [LIMIT=10]
	\| order by: count(*) DESC
	\| row-size=12B cardinality=0
	\|
	04:AGGREGATE [FINALIZE]
	\| output: count:merge(*)
	\| group by: zip
	\| row-size=12B cardinality=0
	\|
	03:EXCHANGE [HASH(zip)]
	\|
	01:AGGREGATE [STREAMING]
	\| output: count(*)
	\| group by: zip
	\| row-size=12B cardinality=0
	\|
	00:SCAN HDFS [functional.testtbl]
	partitions=1/1 files=0 size=0B
	predicates: name LIKE 'm%'
	row-size=16B cardinality=0
	====
	select int_col, sum(float_col)
	from functional_hbase.alltypessmall
	where id < 5
	group by 1
	order by 2
	limit 4
	---- PLAN
	PLAN-ROOT SINK
	\|
	02:TOP-N [LIMIT=4]
	\| order by: sum(float_col) ASC
	\| row-size=12B cardinality=4
	\|
	01:AGGREGATE [FINALIZE]
	\| output: sum(float_col)
	\| group by: int_col
	\| row-size=12B cardinality=5
	\|
	00:SCAN HBASE [functional_hbase.alltypessmall]
	predicates: id < 5
	row-size=12B cardinality=5
	---- DISTRIBUTEDPLAN
	PLAN-ROOT SINK
	\|
	05:MERGING-EXCHANGE [UNPARTITIONED]
	\| order by: sum(float_col) ASC
	\| limit: 4
	\|
	02:TOP-N [LIMIT=4]
	\| order by: sum(float_col) ASC
	\| row-size=12B cardinality=4
	\|
	04:AGGREGATE [FINALIZE]
	\| output: sum:merge(float_col)
	\| group by: int_col
	\| row-size=12B cardinality=5
	\|
	03:EXCHANGE [HASH(int_col)]
	\|
	01:AGGREGATE [STREAMING]
	\| output: sum(float_col)
	\| group by: int_col
	\| row-size=12B cardinality=5
	\|
	00:SCAN HBASE [functional_hbase.alltypessmall]
	predicates: id < 5
	row-size=12B cardinality=5
	====
	select int_col, sum(float_col), min(float_col)
	from functional_hbase.alltypessmall
	group by 1
	order by 2,3 desc
	limit 0
	---- PLAN
	PLAN-ROOT SINK
	\|
	00:EMPTYSET
	---- DISTRIBUTEDPLAN
	PLAN-ROOT SINK
	\|
	00:EMPTYSET
	====
	# Test correct identification of the implicit aliasing of int_col in the select
	# list to t1.int_col;
	# t2.int_col is not null references enough cols of t2 to prevent a broadcast join
	select t1.int_col
	from functional.alltypessmall t1, functional.alltypessmall t2
	where t1.id = t2.id and t2.int_col is not null
	order by int_col
	limit 2
	---- PLAN
	PLAN-ROOT SINK
	\|
	03:TOP-N [LIMIT=2]
	\| order by: int_col ASC
	\| row-size=4B cardinality=2
	\|
	02:HASH JOIN [INNER JOIN]
	\| hash predicates: t1.id = t2.id
	\| runtime filters: RF000 <- t2.id
	\| row-size=16B cardinality=10
	\|
	\|--01:SCAN HDFS [functional.alltypessmall t2]
	\| partitions=4/4 files=4 size=6.32KB
	\| predicates: t2.int_col IS NOT NULL
	\| row-size=8B cardinality=10
	\|
	00:SCAN HDFS [functional.alltypessmall t1]
	partitions=4/4 files=4 size=6.32KB
	runtime filters: RF000 -> t1.id
	row-size=8B cardinality=100
	---- DISTRIBUTEDPLAN
	PLAN-ROOT SINK
	\|
	05:MERGING-EXCHANGE [UNPARTITIONED]
	\| order by: int_col ASC
	\| limit: 2
	\|
	03:TOP-N [LIMIT=2]
	\| order by: int_col ASC
	\| row-size=4B cardinality=2
	\|
	02:HASH JOIN [INNER JOIN, BROADCAST]
	\| hash predicates: t1.id = t2.id
	\| runtime filters: RF000 <- t2.id
	\| row-size=16B cardinality=10
	\|
	\|--04:EXCHANGE [BROADCAST]
	\| \|
	\| 01:SCAN HDFS [functional.alltypessmall t2]
	\| partitions=4/4 files=4 size=6.32KB
	\| predicates: t2.int_col IS NOT NULL
	\| row-size=8B cardinality=10
	\|
	00:SCAN HDFS [functional.alltypessmall t1]
	partitions=4/4 files=4 size=6.32KB
	runtime filters: RF000 -> t1.id
	row-size=8B cardinality=100
	====
	# Test that the top-n is on int_col and not on the id column
	select int_col as id from functional.alltypessmall order by id limit 2
	---- PLAN
	PLAN-ROOT SINK
	\|
	01:TOP-N [LIMIT=2]
	\| order by: int_col ASC
	\| row-size=4B cardinality=2
	\|
	00:SCAN HDFS [functional.alltypessmall]
	partitions=4/4 files=4 size=6.32KB
	row-size=4B cardinality=100
	---- DISTRIBUTEDPLAN
	PLAN-ROOT SINK
	\|
	02:MERGING-EXCHANGE [UNPARTITIONED]
	\| order by: int_col ASC
	\| limit: 2
	\|
	01:TOP-N [LIMIT=2]
	\| order by: int_col ASC
	\| row-size=4B cardinality=2
	\|
	00:SCAN HDFS [functional.alltypessmall]
	partitions=4/4 files=4 size=6.32KB
	row-size=4B cardinality=100
	====
	# Test that the top-n is on id and not on int_col
	select int_col as id from functional.alltypessmall order by functional.alltypessmall.id limit 2
	---- PLAN
	PLAN-ROOT SINK
	\|
	01:TOP-N [LIMIT=2]
	\| order by: id ASC
	\| row-size=8B cardinality=2
	\|
	00:SCAN HDFS [functional.alltypessmall]
	partitions=4/4 files=4 size=6.32KB
	row-size=8B cardinality=100
	---- DISTRIBUTEDPLAN
	PLAN-ROOT SINK
	\|
	02:MERGING-EXCHANGE [UNPARTITIONED]
	\| order by: id ASC
	\| limit: 2
	\|
	01:TOP-N [LIMIT=2]
	\| order by: id ASC
	\| row-size=8B cardinality=2
	\|
	00:SCAN HDFS [functional.alltypessmall]
	partitions=4/4 files=4 size=6.32KB
	row-size=8B cardinality=100
	====
	# Test that the limit/offset is propagated correctly to child nodes. The TOP-N node
	# should have the limit/offset specified in the query. Child sort nodes should have a
	# limit equal to the parent's (limit + offset) and offset equal to 0.
	select t1.int_col from functional.alltypessmall t1, functional.alltypessmall t2
	where t1.id = t2.id and t2.int_col is not null
	order by int_col
	limit 10 offset 5
	---- PLAN
	PLAN-ROOT SINK
	\|
	03:TOP-N [LIMIT=10 OFFSET=5]
	\| order by: int_col ASC
	\| row-size=4B cardinality=10
	\|
	02:HASH JOIN [INNER JOIN]
	\| hash predicates: t1.id = t2.id
	\| runtime filters: RF000 <- t2.id
	\| row-size=16B cardinality=10
	\|
	\|--01:SCAN HDFS [functional.alltypessmall t2]
	\| partitions=4/4 files=4 size=6.32KB
	\| predicates: t2.int_col IS NOT NULL
	\| row-size=8B cardinality=10
	\|
	00:SCAN HDFS [functional.alltypessmall t1]
	partitions=4/4 files=4 size=6.32KB
	runtime filters: RF000 -> t1.id
	row-size=8B cardinality=100
	---- SCANRANGELOCATIONS
	NODE 0:
	HDFS SPLIT hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=1/090101.txt 0:1610
	HDFS SPLIT hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=2/090201.txt 0:1621
	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
	NODE 1:
	HDFS SPLIT hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=1/090101.txt 0:1610
	HDFS SPLIT hdfs://localhost:20500/test-warehouse/alltypessmall/year=2009/month=2/090201.txt 0:1621
	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:MERGING-EXCHANGE [UNPARTITIONED]
	\| offset: 5
	\| order by: int_col ASC
	\| limit: 10
	\|
	03:TOP-N [LIMIT=15]
	\| order by: int_col ASC
	\| row-size=4B cardinality=10
	\|
	02:HASH JOIN [INNER JOIN, BROADCAST]
	\| hash predicates: t1.id = t2.id
	\| runtime filters: RF000 <- t2.id
	\| row-size=16B cardinality=10
	\|
	\|--04:EXCHANGE [BROADCAST]
	\| \|
	\| 01:SCAN HDFS [functional.alltypessmall t2]
	\| partitions=4/4 files=4 size=6.32KB
	\| predicates: t2.int_col IS NOT NULL
	\| row-size=8B cardinality=10
	\|
	00:SCAN HDFS [functional.alltypessmall t1]
	partitions=4/4 files=4 size=6.32KB
	runtime filters: RF000 -> t1.id
	row-size=8B cardinality=100
	====
	# test distributed top-n over a union (IMPALA-831)
	select int_col, bigint_col from
	(select * from functional.alltypes
	union all
	select * from functional.alltypessmall) t
	order by int_col desc limit 10 offset 5
	---- PLAN
	PLAN-ROOT SINK
	\|
	03:TOP-N [LIMIT=10 OFFSET=5]
	\| order by: int_col DESC
	\| row-size=12B cardinality=10
	\|
	00:UNION
	\| pass-through-operands: all
	\| row-size=12B cardinality=7.40K
	\|
	\|--02:SCAN HDFS [functional.alltypessmall]
	\| partitions=4/4 files=4 size=6.32KB
	\| row-size=12B cardinality=100
	\|
	01:SCAN HDFS [functional.alltypes]
	partitions=24/24 files=24 size=478.45KB
	row-size=12B cardinality=7.30K
	---- DISTRIBUTEDPLAN
	PLAN-ROOT SINK
	\|
	04:MERGING-EXCHANGE [UNPARTITIONED]
	\| offset: 5
	\| order by: int_col DESC
	\| limit: 10
	\|
	03:TOP-N [LIMIT=15]
	\| order by: int_col DESC
	\| row-size=12B cardinality=15
	\|
	00:UNION
	\| pass-through-operands: all
	\| row-size=12B cardinality=7.40K
	\|
	\|--02:SCAN HDFS [functional.alltypessmall]
	\| partitions=4/4 files=4 size=6.32KB
	\| row-size=12B cardinality=100
	\|
	01:SCAN HDFS [functional.alltypes]
	partitions=24/24 files=24 size=478.45KB
	row-size=12B cardinality=7.30K
	====
	# test distributed top-n over a union distinct (IMPALA-831)
	select int_col, bigint_col from
	(select * from functional.alltypes
	union distinct
	select * from functional.alltypessmall) t
	order by int_col desc limit 10 offset 5
	---- PLAN
	PLAN-ROOT SINK
	\|
	04:TOP-N [LIMIT=10 OFFSET=5]
	\| order by: int_col DESC
	\| row-size=12B cardinality=10
	\|
	03:AGGREGATE [FINALIZE]
	\| group by: id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col, year, month
	\| row-size=89B cardinality=7.40K
	\|
	00:UNION
	\| pass-through-operands: all
	\| row-size=89B cardinality=7.40K
	\|
	\|--02:SCAN HDFS [functional.alltypessmall]
	\| partitions=4/4 files=4 size=6.32KB
	\| row-size=89B cardinality=100
	\|
	01:SCAN HDFS [functional.alltypes]
	partitions=24/24 files=24 size=478.45KB
	row-size=89B cardinality=7.30K
	---- DISTRIBUTEDPLAN
	PLAN-ROOT SINK
	\|
	07:MERGING-EXCHANGE [UNPARTITIONED]
	\| offset: 5
	\| order by: int_col DESC
	\| limit: 10
	\|
	04:TOP-N [LIMIT=15]
	\| order by: int_col DESC
	\| row-size=12B cardinality=15
	\|
	06:AGGREGATE [FINALIZE]
	\| group by: id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col, year, month
	\| row-size=89B cardinality=7.40K
	\|
	05:EXCHANGE [HASH(id,bool_col,tinyint_col,smallint_col,int_col,bigint_col,float_col,double_col,date_string_col,string_col,timestamp_col,year,month)]
	\|
	03:AGGREGATE [STREAMING]
	\| group by: id, bool_col, tinyint_col, smallint_col, int_col, bigint_col, float_col, double_col, date_string_col, string_col, timestamp_col, year, month
	\| row-size=89B cardinality=7.40K
	\|
	00:UNION
	\| pass-through-operands: all
	\| row-size=89B cardinality=7.40K
	\|
	\|--02:SCAN HDFS [functional.alltypessmall]
	\| partitions=4/4 files=4 size=6.32KB
	\| row-size=89B cardinality=100
	\|
	01:SCAN HDFS [functional.alltypes]
	partitions=24/24 files=24 size=478.45KB
	row-size=89B cardinality=7.30K
	====
	# test that top-n is not placed below an unpartitioned exchange with a limit
	select * from (select * from functional.alltypes limit 10) t
	order by int_col limit 10
	---- PLAN
	PLAN-ROOT SINK
	\|
	01:TOP-N [LIMIT=10]
	\| order by: int_col ASC
	\| row-size=89B cardinality=10
	\|
	00:SCAN HDFS [functional.alltypes]
	partitions=24/24 files=24 size=478.45KB
	limit: 10
	row-size=89B cardinality=10
	---- DISTRIBUTEDPLAN
	PLAN-ROOT SINK
	\|
	01:TOP-N [LIMIT=10]
	\| order by: int_col ASC
	\| row-size=89B cardinality=10
	\|
	02:EXCHANGE [UNPARTITIONED]
	\| limit: 10
	\|
	00:SCAN HDFS [functional.alltypes]
	partitions=24/24 files=24 size=478.45KB
	limit: 10
	row-size=89B cardinality=10
	====
	# test that top-n is not placed below an unpartitioned exchange with a limit
	select * from
	(select * from functional.alltypes
	union all
	(select * from functional.alltypessmall) limit 10) t
	order by int_col limit 10
	---- PLAN
	PLAN-ROOT SINK
	\|
	03:TOP-N [LIMIT=10]
	\| order by: int_col ASC
	\| row-size=89B cardinality=10
	\|
	00:UNION
	\| pass-through-operands: all
	\| limit: 10
	\| row-size=89B cardinality=10
	\|
	\|--02:SCAN HDFS [functional.alltypessmall]
	\| partitions=4/4 files=4 size=6.32KB
	\| row-size=89B cardinality=100
	\|
	01:SCAN HDFS [functional.alltypes]
	partitions=24/24 files=24 size=478.45KB
	row-size=89B cardinality=7.30K
	---- DISTRIBUTEDPLAN
	PLAN-ROOT SINK
	\|
	03:TOP-N [LIMIT=10]
	\| order by: int_col ASC
	\| row-size=89B cardinality=10
	\|
	04:EXCHANGE [UNPARTITIONED]
	\| limit: 10
	\|
	00:UNION
	\| pass-through-operands: all
	\| limit: 10
	\| row-size=89B cardinality=10
	\|
	\|--02:SCAN HDFS [functional.alltypessmall]
	\| partitions=4/4 files=4 size=6.32KB
	\| row-size=89B cardinality=100
	\|
	01:SCAN HDFS [functional.alltypes]
	partitions=24/24 files=24 size=478.45KB
	row-size=89B cardinality=7.30K
	====