feat: add parallelism column in rw_fragments (#12901)

risingwavelabs · Oct 17, 2023 · 1f151cd · 1f151cd
1 parent e713fb9
commit 1f151cd
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 0 deletions.
diff --git a/proto/meta.proto b/proto/meta.proto
@@ -221,6 +221,7 @@ message ListFragmentDistributionResponse {
     repeated uint32 state_table_ids = 4;
     repeated uint32 upstream_fragment_ids = 5;
     uint32 fragment_type_mask = 6;
+    uint32 parallelism = 7;
   }
   repeated FragmentDistribution distributions = 1;
 }

diff --git a/src/frontend/src/catalog/system_catalog/rw_catalog/rw_fragments.rs b/src/frontend/src/catalog/system_catalog/rw_catalog/rw_fragments.rs
@@ -35,6 +35,7 @@ pub static RW_FRAGMENTS_COLUMNS: LazyLock<Vec<SystemCatalogColumnsDef<'_>>> = La
             "upstream_fragment_ids",
         ),
         (DataType::List(Box::new(DataType::Varchar)), "flags"),
+        (DataType::Int32, "parallelism"),
     ]
 });
 
@@ -93,6 +94,7 @@ impl SysCatalogReaderImpl {
                             .map(|t| Some(ScalarImpl::Utf8(t.into())))
                             .collect_vec(),
                     ))),
+                    Some(ScalarImpl::Int32(distribution.parallelism as i32)),
                 ])
             })
             .collect_vec())

diff --git a/src/meta/src/rpc/service/stream_service.rs b/src/meta/src/rpc/service/stream_service.rs
@@ -207,6 +207,7 @@ impl StreamManagerService for StreamServiceImpl {
                             state_table_ids: fragment.state_table_ids.clone(),
                             upstream_fragment_ids: fragment.upstream_fragment_ids.clone(),
                             fragment_type_mask: fragment.fragment_type_mask,
+                            parallelism: fragment.actors.len() as _,
                         }
                     })
                 })