Modi Tamam created SPARK-22898:
----------------------------------
Summary: collect_set aggregation on bucketed table causes an
exchange stage
Key: SPARK-22898
URL: https://issues.apache.org/jira/browse/SPARK-22898
Project: Spark
Issue Type: Bug
Components: Spark Core
Affects Versions: 2.3.0
Reporter: Modi Tamam
I'm using Spark-2.2. I'm POCing Spark's bucketing. I've created a bucketed
table, here's the desc formatted my_bucketed_tbl output:
+--------------------+--------------------+-------+
| col_nam| data_type|comment|
+--------------------+--------------------+-------+
| bundle| string| null|
| ifa| string| null|
| date_| date| null|
| hour| int| null|
| | | |
|# Detailed Table ...| | |
| Database| default| |
| Table| my_bucketed_tbl|
| Owner| zeppelin| |
| Created|Thu Dec 21 13:43:...| |
| Last Access|Thu Jan 01 00:00:...| |
| Type| EXTERNAL| |
| Provider| orc| |
| Num Buckets| 16| |
| Bucket Columns| [`ifa`]| |
| Sort Columns| [`ifa`]| |
| Table Properties|[transient_lastDd...| |
| Location|hdfs:/user/hive/w...| |
| Serde Library|org.apache.hadoop...| |
| InputFormat|org.apache.hadoop...| |
| OutputFormat|org.apache.hadoop...| |
| Storage Properties|[serialization.fo...| |
+--------------------+--------------------+-------+
When I'm executing an explain of a group by query, I can see that we've spared
the exchange phase :
{code:java}
sql("select ifa,max(bundle) from my_bucketed_tbl group by ifa").explain
== Physical Plan ==
SortAggregate(key=[ifa#932], functions=[max(bundle#920)])
+- SortAggregate(key=[ifa#932], functions=[partial_max(bundle#920)])
+- *Sort [ifa#932 ASC NULLS FIRST], false, 0
+- *FileScan orc default.level_1[bundle#920,ifa#932] Batched: false,
Format: ORC, Location:
InMemoryFileIndex[hdfs://ip-10-44-9-73.ec2.internal:8020/user/hive/warehouse/level_1/date_=2017-1...,
PartitionFilters: [], PushedFilters: [], ReadSchema:
struct<bundle:string,ifa:string>
{code}
But, when I replace Spark's max function with collect_set, I can see that the
execution plan is the same as a non-bucketed table, means, the exchange phase
is not spared :
{code:java}
sql("select ifa,collect_set(bundle) from my_bucketed_tbl group by ifa").explain
== Physical Plan ==
ObjectHashAggregate(keys=[ifa#1010], functions=[collect_set(bundle#998, 0, 0)])
+- Exchange hashpartitioning(ifa#1010, 200)
+- ObjectHashAggregate(keys=[ifa#1010],
functions=[partial_collect_set(bundle#998, 0, 0)])
+- *FileScan orc default.level_1[bundle#998,ifa#1010] Batched: false,
Format: ORC, Location:
InMemoryFileIndex[hdfs://ip-10-44-9-73.ec2.internal:8020/user/hive/warehouse/level_1/date_=2017-1...,
PartitionFilters: [], PushedFilters: [], ReadSchema:
struct<bundle:string,ifa:string
{code}
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
