Panic when the number of partitions of the pipeline that throws the exception is inconsistent with the number of partitions output by the query

[yukkit]

Describe the bug

I am the developer of CnosDB, when we are doing integration testing, some SQL causes panic.

such as:

WITH gp AS (SELECT id FROM table1 GROUP BY id) 
SELECT
    COUNT(CAST(CAST(gp.id || '' AS TIMESTAMP) AS BIGINT)) 
FROM 
    gp;

To Reproduce

1. physical plan

Plan: ProjectionExec: expr=[COUNT(gp.id || Utf8(""))@0 as COUNT(gp.id || Utf8(""))]
  AggregateExec: mode=Final, gby=[], aggr=[COUNT(gp.id || Utf8(""))]
    CoalescePartitionsExec
      AggregateExec: mode=Partial, gby=[], aggr=[COUNT(gp.id || Utf8(""))]
        ProjectionExec: expr=[id@0 as id]
          AggregateExec: mode=FinalPartitioned, gby=[id@0 as id], aggr=[]
            CoalesceBatchesExec: target_batch_size=4096
              RepartitionExec: partitioning=Hash([Column { name: "id", index: 0 }], 4)
                AggregateExec: mode=Partial, gby=[id@0 as id], aggr=[]
                  RepartitionExec: partitioning=RoundRobinBatch(4)
                    MemoryExec: partitions=2, partition_sizes=[20, 20]

2. result

thread 'df-worker-1' panicked at 'index out of bounds: the len is 1 but the index is 2', datafusion/core/src/scheduler/task.rs:306:17
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
thread 'df-worker-2' panicked at 'index out of bounds: the len is 1 but the index is 1', datafusion/core/src/scheduler/task.rs:306:17
thread 'df-worker-3' panicked at 'index out of bounds: the len is 1 but the index is 3', datafusion/core/src/scheduler/task.rs:306:17
thread 'scheduler::tests::test_runtime_err' panicked at 'called `Result::unwrap()` on an `Err` value: ExternalError(ArrowError(ExternalError(ArrowError(CastError("Error parsing '2' as timestamp")))))', datafusion/core/src/scheduler/mod.rs:480:64
[2022-11-03T08:53:53Z ERROR datafusion::scheduler] worker 3 panicked with: index out of bounds: the len is 1 but the index is 3
[2022-11-03T08:53:53Z ERROR datafusion::scheduler] worker 2 panicked with: index out of bounds: the len is 1 but the index is 1

Expected behavior
no panic

Additional context

I am trying to fix this

[yukkit]

I use the Scheduler to execute execution plan.