frontend: Incorrect use of block_on() on plan_fragmenter

[ZENOTME]

Describe the bug

In plan_fragmenter, for getting the source information(it's a async interface), we use block_on.

risingwave/src/frontend/src/scheduler/plan_fragmenter.rs

Line 623 in 733847f

let mut enumerator = block_on(SplitEnumeratorImpl::create(property))?;

But that block_on will cause the woker in runtime blocked. In deterministic simulation, the runtime seems only run in a single thread, so that will cause whole program be blocked. For the block reason, we can refer:
tokio-rs/tokio-compat#23 (comment)

Solution:

1. change BatchPlanFragmenter.split into a async. These solution seems impossible, because PlanRef is hard to satisfy Send,Sync trait. (BatchPlanFragmenter.split need to statisfy thread safety if we change it to a async function). Lots of thing need to modify if we take it...
2. delay get source information util we schedule the fragment. (The schedule work is async so we can directly use the async operation in it)

   risingwave/src/frontend/src/handler/query.rs

   Line 150 in 733847f

   local_execute(session.clone(), query, query_snapshot).await?,

   
   This solution seems also impossible. We need the source information to get the parallelism, parallelism needed for creating fragment.
   

   risingwave/src/frontend/src/scheduler/plan_fragmenter.rs

   Line 595 in 733847f

   let child_exchange_info = node.distribution().to_prost(builder.parallelism, self);

3. I think it's a available solution. Including split information in SourceCatalog. And that we don't need async operation to get source information(including parallelism)
   For the timestamp limit, we can delay it util we schedule the fragment.
   

   risingwave/src/frontend/src/scheduler/plan_fragmenter.rs

   Line 635 in 733847f

   kafka_enumerator.list_splits_batch(timestamp_bound.0, timestamp_bound.1),

To Reproduce

No response

Expected behavior

No response

Additional context

No response

[ZENOTME]

@liurenjie1024 @tabVersion PTAL

[liurenjie1024]

Solution 3 sounds great, there is a problems to resolve for this approach:

1. The partitions of source may change(e.g. user may add/remove partition of kafka), so we a job in meta to refresh this and send it to frontend. I think the refresh is already done?

[ZENOTME]

For solution 3, it has the problem: for some source like s3, source information may be appropriate to store in catalog.

So I'm going to investiagte can we split plan_fragmenter into two phase(sync and async phase) to solve this problem.

[ZENOTME]

Solved by #7581