feat: implement FlowTaskMetadataManager

[WenyXu]

I hereby agree to the terms of the GreptimeDB CLA.

Refer to a related PR or issue link (optional)

#3664

What's changed and what's your intention?

1. Introduces FlowTaskInfoKey, layout: __flow_task/{catalog}/info/{flow_task_id}, which stores the metadata of the flow task.
2. Introduces FlowTaskNameKey, layout: __flow_task/{catalog}/name/{task_name}, which mapping task_name to task_id.
3. Introduces FlownodeTaskKey, layout: __flow_task/{catalog}/flownode/{flownode_id}/{flow_task_id}/{partition_id}, which mapping flownode_id to task_id.
4. Introduces TableTaskKey, layout: __table_task/{catalog}/source_table/{table_id}/{flownode_id}/{flow_task_id}/{partition_id}, which mapping table_id to node_id.
5. Introduces FlowTaskMetadataManager, and implements the create_flow_task_metadata method.

The whole picture will be like this:

__flow_task/
    {catalog}/
      info/
        {tsak_id}
    
      name/
        {task_name}
    
      flownode/
        {flownode_id}/
          {task_id}/
            {partition_id}

      source_table/
        {table_id}/
          {flownode_id}/
            {task_id}/
              {partition_id}

Checklist

• I have written the necessary rustdoc comments.
• I have added the necessary unit tests and integration tests.
• This PR does not require documentation updates.

[codecov]

Codecov Report

Attention: Patch coverage is 93.11552% with 59 lines in your changes are missing coverage. Please review.

Project coverage is 85.32%. Comparing base (20a933e) to head (726488e).
Report is 10 commits behind head on main.

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3766      +/-   ##
==========================================
- Coverage   85.55%   85.32%   -0.24%     
==========================================
  Files         947      952       +5     
  Lines      160481   161940    +1459     
==========================================
+ Hits       137307   138172     +865     
- Misses      23174    23768     +594     

[WenyXu]

Why is the flow task is directly put under "catalog"? Do you mean flow task can cross database but not catalog?

Yes, I think we should treat a catalog as a tenant. cc @discord9

[WenyXu]

PTAL @MichaelScofield @fengjiachun

[discord9]

Why is the flow task is directly put under "catalog"? Do you mean flow task can cross database but not catalog?

Yes, I think we should treat a catalog as a tenant. cc @discord9

It would make sense for one user to use multiple database as input, so task should be cross database but also belong to one catalog, so it's per user?

[MichaelScofield]

Can we reduce the depth of the key? I think we can end up with "flownode_id", put flow task id and partition id in the value.

[MichaelScofield]

I don't think it's good to place the variable "catalog" between two static prefixes. For example, better __flow_task/name/{catalog} than __flow_task/{catalog}/name. It can be a lot easier for listing, for example, all flow tasks' names.

[fengjiachun]

LGTM

[fengjiachun]

I don't think it's good to place the variable "catalog" between two static prefixes. For example, better __flow_task/name/{catalog} than __flow_task/{catalog}/name. It can be a lot easier for listing, for example, all flow tasks' names.

We need to be able to see all keys under a catalog, and the catalog also serves as an isolation, in other words: catalog1 cannot see catalog2.