Moving PipelineFixer above all rules to use ExecutionPlan APIs

[metesynnada]

Which issue does this PR close?

Closes #5878.

Rationale for this change

Since some strongly dependent optimizer rules affect each other, rule ordering becomes more important. PipelineFixer (maybe more rules in the future) can change the ExecutionPlan at a level, and the new ExecutionPlan can have a different set of flags (maybe ordering, distribution, or more).

I suggest making the executor changer rules above the rules that fill the sort, distribution, etc.

If the sources are also sorted, we are looking for keeping the order information without adding additional SortExec. However, current planner results

[
    "SymmetricHashJoinExec: join_type=Full, on=[(Column { name: \"a2\", index: 1 }, Column { name: \"a2\", index: 1 })], filter=BinaryExpr { left: BinaryExpr { left: CastExpr { expr: Column { name: \"a1\", index: 0 }, cast_type: Int64, cast_options: CastOptions { safe: false } }, op: Gt, right: BinaryExpr { left: CastExpr { expr: Column { name: \"a1\", index: 1 }, cast_type: Int64, cast_options: CastOptions { safe: false } }, op: Plus, right: Literal { value: Int64(3) } } }, op: And, right: BinaryExpr { left: CastExpr { expr: Column { name: \"a1\", index: 0 }, cast_type: Int64, cast_options: CastOptions { safe: false } }, op: Lt, right: BinaryExpr { left: CastExpr { expr: Column { name: \"a1\", index: 1 }, cast_type: Int64, cast_options: CastOptions { safe: false } }, op: Plus, right: Literal { value: Int64(10) } } } }",
    "  CoalesceBatchesExec: target_batch_size=8192",
    "    RepartitionExec: partitioning=Hash([Column { name: \"a2\", index: 1 }], 8), input_partitions=8",
    "      RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
    "        CsvExec: files={1 group: [[private/var/folders/rf/dhj0s83d57l2_m51k2dmd_ch0000gn/T/.tmpcRbDJD/left.csv]]}, has_header=false, limit=None, projection=[a1, a2]",
    "  CoalesceBatchesExec: target_batch_size=8192",
    "    RepartitionExec: partitioning=Hash([Column { name: \"a2\", index: 1 }], 8), input_partitions=8",
    "      RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
    "        CsvExec: files={1 group: [[private/var/folders/rf/dhj0s83d57l2_m51k2dmd_ch0000gn/T/.tmpcRbDJD/right.csv]]}, has_header=false, limit=None, projection=[a1, a2]",
]

and, unfortunately, two consecutive RepartitionExec removes the order information. To prevent this, SHJ particularly sets benefits_from_input_partitioning to false, however, it is ineffective since RepartitionExec::RounRobin is added before HashJoin -> SymmetricHashJoin change.

If we move the Repartition rule below the PipelineFixer, we are able to use SymmetricHashJoinExec 's benefits_from_input_partitioning API effectively.

[
    "SymmetricHashJoinExec: join_type=Full, on=[(Column { name: \"a2\", index: 1 }, Column { name: \"a2\", index: 1 })], filter=BinaryExpr { left: BinaryExpr { left: CastExpr { expr: Column { name: \"a1\", index: 0 }, cast_type: Int64, cast_options: CastOptions { safe: false } }, op: Gt, right: BinaryExpr { left: CastExpr { expr: Column { name: \"a1\", index: 1 }, cast_type: Int64, cast_options: CastOptions { safe: false } }, op: Plus, right: Literal { value: Int64(3) } } }, op: And, right: BinaryExpr { left: CastExpr { expr: Column { name: \"a1\", index: 0 }, cast_type: Int64, cast_options: CastOptions { safe: false } }, op: Lt, right: BinaryExpr { left: CastExpr { expr: Column { name: \"a1\", index: 1 }, cast_type: Int64, cast_options: CastOptions { safe: false } }, op: Plus, right: Literal { value: Int64(10) } } } }",
    "  CoalesceBatchesExec: target_batch_size=8192",
    "    RepartitionExec: partitioning=Hash([Column { name: \"a2\", index: 1 }], 8), input_partitions=1",
    "      CsvExec: files={1 group: [[private/var/folders/rf/dhj0s83d57l2_m51k2dmd_ch0000gn/T/.tmpdTwdrk/left.csv]]}, has_header=false, limit=None, projection=[a1, a2]",
    "  CoalesceBatchesExec: target_batch_size=8192",
    "    RepartitionExec: partitioning=Hash([Column { name: \"a2\", index: 1 }], 8), input_partitions=1",
    "      CsvExec: files={1 group: [[private/var/folders/rf/dhj0s83d57l2_m51k2dmd_ch0000gn/T/.tmpdTwdrk/right.csv]]}, has_header=false, limit=None, projection=[a1, a2]",
]

What changes are included in this PR?

Optimizer reorders to use the benefits_from_input_partitioning API. If we call PipelineFixer above (almost) everything, we can leverage the changed executor APIs in the optimizer.

Are these changes tested?

Yes.

Are there any user-facing changes?

No.

[alamb]

makes sense to me -- thank you @metesynnada