Exchange on WINDOW operation when the data is already partitioned and exchanged

user_14744348 0
Viewed 37

I have the query below which is EXCHANGING data to perform row_number() over a COLUMN on which the data is already exchanged in both the data sets

the table crime_parquet_table is a Bucketed table on the column Incident_Number with 30 buckets

explain
select
*
from
(
select  *,
row_number() over(partition by Incident_Number order by Incident_Number) as rnk
from
    (
        select  *
    from
        (
        select /*+ REPARTITION(30,Incident_Number) */ * 
        from austin_crime_data_new
        union all
        select *
        from crime_parquet_table
        ) tmp
    )t
) d
where rnk = 1

below is the explain plan. I believe the +- Exchange (7) is not necessary. Let me know if you guys think otherwise and the reason for the same.

== Physical Plan ==
* Filter (10)
+- Window (9)
   +- * Sort (8)
      +- Exchange (7)
         +- Union (6)
            :- Exchange (3)
            :  +- * Project (2)
            :     +- Scan csv  (1)
            +- * ColumnarToRow (5)
               +- Scan parquet default.crime_parquet_table (4)
apache-spark-sql apache-spark
1 Answers

I got this worked using the below alternative approach as union all doesn't look at this as a special case . it just treats this as any other tables

So first I saved my intermediate data also as a bucketed table

create table crime_parquet_table2
using parquet
CLUSTERED BY (Incident_Number) INTO 30 BUCKETS
location 'path2'
select /*+ REPARTITION(30,Incident_Number) */ * 
from austin_crime_data_new

Now I create a table using LIKE with both paths ( each have bucketed data)

CREATE TABLE crime_data like crime_parquet_table2
location 'path1,path2'
;

Now the sql on this new table gives the expected plan

explain
select *
from
(
select * ,
row_number() over(partition by Incident_Number order by Incident_Number) as rnk
from crime_data
) 
where rnk = 1

PLAN

== Physical Plan ==
* Filter (5)
+- Window (4)
   +- * Sort (3)
      +- * ColumnarToRow (2)
         +- Scan parquet default.crime_data (1)

So the extra work is persisting the intermediate data to a storage but shuffle is eliminated

user_147443480
Related