Redundant repartition after explicit repartition

[linhr]

When I was reviewing #2183 I noticed something that worth looking into (unrelated to the changes in the PR).

>>> spark.range(6).repartition(3).filter(F.col("id") % 2 == 0).explain()
== Physical Plan ==
ProjectionExec: expr=[#0@0 as id]
  FilterExec: #0@0 % 2 = 0
    RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=3
      ExplicitRepartitionExec: partitioning=RoundRobinBatch(3), input_partitions=1
        RangeExec

I feel the RepartitionExec node should be removed.

[nis12ram]

@linhr I have a feeling it is related to some physical optimization rule of FilterExec that does this at the EnforceDistribution stage.

spark.range(6).repartition(3).filter(F.col("id") % 2 == 0)

spark.range(6).filter(F.col("id") % 2 == 0)

[linhr]

Yeah EnforceDistribution would inject the repartition depending on the characteristics of physical nodes (e.g., the benefits_from_input_partitioning implementation for FilterExec).

The question now is how we turn off this behavior, or whether we should add another optimizer rule to remove redundant partitioning.