feat(meta): ensure each command is applied to exactly one database

[wenym1]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Previously, we have a global streaming graph, and therefore the barrier command to the global barrier manager is applied to the whole streaming graph. However, as we are supporting isolation between multiple databases, if we have multiple databases sharing a same barrier command, the complexity will be greatly increased.

In this PR, we will change to ensure that each barrier command is applied to exactly one database. Currently we have the following command that may be applied to multiple databases according to its definition: Pause, Resume, and command that contains a HashMap<FragmentId, _> where the FragmentId can be across multiple databases, including RescheduleFragment, SourceSplitAssignment and Throttle.

For Pause and Resume, they are generated from either risectl or configuration change. For the ones generated from configuration change, we can associated the database id of the specific configuration change command to the Pause and Resume. For the ones generated from risectl, when handling these risectl command, we will list all active databases from catalog, and then send Pause or Resume one by one to these active databases.

For commands containing HashMap<FragmentId, _>, the strategy on the refactor will be to list all databases where the FragmentIds belong to, and then divide them into HashMap<DatabaseId, HashMap<FragmentId, _>>, and then we apply the command to each database one by one.

We have refactored the barrier command by the way. Previously, each barrier will be associated with a command. For normal barrier, we have Command::Plain(None). In this PR, such variant of Command is removed. Instead, we introduce an extra Command::Flush to handle the manually generated flush SQL. The database id of the Command::Flush will be the database of the current frontend client session. For periodic barrier, there won't be an associated command, and therefore each barrier will change to hold Option<Command> instead of Command, and it will carry Some(command) when the barrier is generated by a scheduled command, and None when the barrier is periodic command.

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added test labels as necessary. See details.
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
• My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
• All checks passed in ./risedev check (or alias, ./risedev c)
• My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

• My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

• My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

[hzxa21]

LGTM!