CollapseRepartition physical optimizer

[Jefffrey]

Which issue does this PR close?

Closes #41

Rationale for this change

Add repartition elimination as suggested by #41

What changes are included in this PR?

New physical optimizer to eliminate adjacent RepartitionExec nodes

Are there any user-facing changes?

[Jefffrey]

Unsure if this is what is meant by #41, but I gave it a shot

There's probably more to the optimization that could be done, such as dealing with cases like having a Projection sandwiched between two repartition nodes, or cases like this:

HashJoinExec: mode=Partitioned, join_type=Inner, on=[(Column { name: "int_col", index: 0 }, Column { name: "int_col3", index: 0 })]                                                                                                    
 *RepartitionExec: partitioning=Hash([Column { name: "int_col", index: 0 }], 12)                                                                                                                                                       
    HashJoinExec: mode=Partitioned, join_type=Inner, on=[(Column { name: "int_col", index: 0 }, Column { name: "int_col2", index: 0 })]                                                                                                
     *RepartitionExec: partitioning=Hash([Column { name: "int_col", index: 0 }], 12)                                                                                                                                                   
        ProjectionExec: expr=[int_col@2 as int_col, double_col@3 as double_col, CAST(date_string_col@4 AS Utf8) as alltypes_plain.date_string_col]                                                                                     
          FilterExec: id@0 > 1 AND CAST(tinyint_col@1 AS Float64) < double_col@3                                                                                                                                                       
            ParquetExec: limit=None, partitions=[home/jeffrey/Code/arrow-datafusion/parquet-testing/data/alltypes_plain.parquet], predicate=id_max@0 > 1 AND true, projection=[id, tinyint_col, int_col, double_col, date_string_col]  
      RepartitionExec: partitioning=Hash([Column { name: "int_col2", index: 0 }], 12)                                                                                                                                                  
        ProjectionExec: expr=[int_col@0 as int_col2]                                                                                                                                                                                   
          ProjectionExec: expr=[int_col@2 as int_col, double_col@3 as double_col, CAST(date_string_col@4 AS Utf8) as alltypes_plain.date_string_col]                                                                                   
            FilterExec: id@0 > 1 AND CAST(tinyint_col@1 AS Float64) < double_col@3                                                                                                                                                     
              ParquetExec: limit=None, partitions=[home/jeffrey/Code/arrow-datafusion/parquet-testing/data/alltypes_plain.parquet], predicate=id_max@0 > 1 AND true, projection=[id, tinyint_col, int_col, double_col, date_string_col]
  RepartitionExec: partitioning=Hash([Column { name: "int_col3", index: 0 }], 12)                                                                                                                                                      
    ProjectionExec: expr=[int_col@0 as int_col3]                                                                                                                                                                                       
      ProjectionExec: expr=[int_col@2 as int_col, double_col@3 as double_col, CAST(date_string_col@4 AS Utf8) as alltypes_plain.date_string_col]                                                                                       
        FilterExec: id@0 > 1 AND CAST(tinyint_col@1 AS Float64) < double_col@3                                                                                                                                                         
          ParquetExec: limit=None, partitions=[home/jeffrey/Code/arrow-datafusion/parquet-testing/data/alltypes_plain.parquet], predicate=id_max@0 > 1 AND true, projection=[id, tinyint_col, int_col, double_col, date_string_col]

Where could potentially collapse the two marked RepartitionExec nodes? Unsure if that is correct.

Appreciate any feedback on this

[mingmwang]

There is a general implementation in below PR, please take a look.
#3855
I working on split this PR to small ones now.

[Dandandan]

@Jefffrey the physical nodes have information on them (output_partitioning) that enables retrieving the child partitioning and seeing that an extra repartitioning will not change it, doing it in a more generic way wether there is a filter, projection, aggregate, join in between.
See PR from @mingmwang

[Jefffrey]

I see, so should #41 be linked to #3855 as well then, if I'm understanding correctly?