Split ArrowFlightReadExec node placement for distributed planning

[gabotechs]

Splits the distributed planning in two steps:

1. A first step that places the appropriate ArrowFlightReadExec nodes in "pending" mode through the plan. These will serve as placeholders, just puts a logical network boundary in the appropriate places so that the second step can build stages out of that.
2. A second step that inspects the plan looking for placed ArrowFlightReadExec nodes in "pending" mode, puts them in "ready" mode and wraps the plan in stages.

For example:

Given the following query:

SELECT count(*), "RainToday" FROM weather GROUP BY "RainToday" ORDER BY count(*)

and the following plan

        ProjectionExec: expr=[count(*)@0 as count(*), RainToday@1 as RainToday]
          SortPreservingMergeExec: [count(Int64(1))@2 ASC NULLS LAST]
            SortExec: expr=[count(Int64(1))@2 ASC NULLS LAST], preserve_partitioning=[true]
              ProjectionExec: expr=[count(Int64(1))@1 as count(*), RainToday@0 as RainToday, count(Int64(1))@1 as count(Int64(1))]
                AggregateExec: mode=FinalPartitioned, gby=[RainToday@0 as RainToday], aggr=[count(Int64(1))]
                  CoalesceBatchesExec: target_batch_size=8192
                    RepartitionExec: partitioning=Hash([RainToday@0], 4), input_partitions=4
                      RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
                        AggregateExec: mode=Partial, gby=[RainToday@0 as RainToday], aggr=[count(Int64(1))]
                          DataSourceExec: file_groups={1 group: [[/testdata/weather.parquet]]}, projection=[RainToday], file_type=parquet

1. Places the appropriate ArrowFlightReadExec nodes

        ProjectionExec: expr=[count(*)@0 as count(*), RainToday@1 as RainToday]
          SortPreservingMergeExec: [count(Int64(1))@2 ASC NULLS LAST]
            SortExec: expr=[count(Int64(1))@2 ASC NULLS LAST], preserve_partitioning=[true]
              ProjectionExec: expr=[count(Int64(1))@1 as count(*), RainToday@0 as RainToday, count(Int64(1))@1 as count(Int64(1))]
                AggregateExec: mode=FinalPartitioned, gby=[RainToday@0 as RainToday], aggr=[count(Int64(1))]
                  CoalesceBatchesExec: target_batch_size=8192
this ->             ArrowFlightReadExec
                      RepartitionExec: partitioning=Hash([RainToday@0], 4), input_partitions=2
this ->                 PartitionIsolatorExec [providing upto 2 partitions]
this ->                   ArrowFlightReadExec
                            RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
this ->                       PartitionIsolatorExec [providing upto 2 partitions]
                                AggregateExec: mode=Partial, gby=[RainToday@0 as RainToday], aggr=[count(Int64(1))]
                                  DataSourceExec: file_groups={1 group: [[/testdata/weather.parquet]]}, projection=[RainToday], file_type=parquet

2. Distributes the plan into stages

        ┌───── Stage 3   Task: partitions: 0,unassigned]
        │partitions [out:1  <-- in:1  ] ProjectionExec: expr=[count(*)@0 as count(*), RainToday@1 as RainToday]
        │partitions [out:1  <-- in:4  ]   SortPreservingMergeExec: [count(Int64(1))@2 ASC NULLS LAST]
        │partitions [out:4  <-- in:4  ]     SortExec: expr=[count(Int64(1))@2 ASC NULLS LAST], preserve_partitioning=[true]
        │partitions [out:4  <-- in:4  ]       ProjectionExec: expr=[count(Int64(1))@1 as count(*), RainToday@0 as RainToday, count(Int64(1))@1 as count(Int64(1))]
        │partitions [out:4  <-- in:4  ]         AggregateExec: mode=FinalPartitioned, gby=[RainToday@0 as RainToday], aggr=[count(Int64(1))]
        │partitions [out:4  <-- in:4  ]           CoalesceBatchesExec: target_batch_size=8192
        │partitions [out:4            ]             ArrowFlightReadExec: Stage 2  
        │
        └──────────────────────────────────────────────────
          ┌───── Stage 2   Task: partitions: 0,1,unassigned],Task: partitions: 2,3,unassigned]
          │partitions [out:4  <-- in:2  ] RepartitionExec: partitioning=Hash([RainToday@0], 4), input_partitions=2
          │partitions [out:2  <-- in:4  ]   PartitionIsolatorExec [providing upto 2 partitions]
          │partitions [out:4            ]     ArrowFlightReadExec: Stage 1  
          │
          └──────────────────────────────────────────────────
            ┌───── Stage 1   Task: partitions: 0,1,unassigned],Task: partitions: 2,3,unassigned]
            │partitions [out:4  <-- in:2  ] RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=2
            │partitions [out:2  <-- in:1  ]   PartitionIsolatorExec [providing upto 2 partitions]
            │partitions [out:1  <-- in:1  ]     AggregateExec: mode=Partial, gby=[RainToday@0 as RainToday], aggr=[count(Int64(1))]
            │partitions [out:1            ]       DataSourceExec: file_groups={1 group: [[/testdata/weather.parquet]]}, projection=[RainToday], file_type=parquet
            │
            └──────────────────────────────────────────────────

The benefit is that now users are able to skip the first step, and choose to place the ArrowFlightReadExec nodes wherever they want, and just reuse the 2 step to convert their plans into stages.

[gabotechs]

I think the depth here was wrong in the previous implementation, the new depth looks correct