refactor(batch): maintain serving vnode mapping in meta node (#10004)

proto/meta.proto

@@ -106,6 +106,10 @@ message FragmentParallelUnitMapping {
   common.ParallelUnitMapping mapping = 2;
 }
 
+message FragmentParallelUnitMappings {
+  repeated FragmentParallelUnitMapping mappings = 1;
+}
+
 // TODO: remove this when dashboard refactored.
 message ActorLocation {
   common.WorkerNode node = 1;
@@ -253,12 +257,15 @@ message MetaSnapshot {
   repeated catalog.Function functions = 15;
   repeated catalog.Connection connections = 17;
   repeated user.UserInfo users = 8;
+  // for streaming
   repeated FragmentParallelUnitMapping parallel_unit_mappings = 9;
   repeated common.WorkerNode nodes = 10;
   hummock.HummockSnapshot hummock_snapshot = 11;
   hummock.HummockVersion hummock_version = 12;
   backup_service.MetaBackupManifestId meta_backup_manifest_id = 14;
   hummock.WriteLimits hummock_write_limits = 16;
+  // for serving
+  repeated FragmentParallelUnitMapping serving_parallel_unit_mappings = 18;
 
   SnapshotVersion version = 13;
 }
@@ -293,6 +300,7 @@ message SubscribeResponse {
     catalog.Schema schema = 5;
     catalog.Function function = 6;
     user.UserInfo user = 11;
+    // for streaming
     FragmentParallelUnitMapping parallel_unit_mapping = 12;
     common.WorkerNode node = 13;
     hummock.HummockSnapshot hummock_snapshot = 14;
@@ -303,6 +311,7 @@ message SubscribeResponse {
     hummock.WriteLimits hummock_write_limits = 20;
     RelationGroup relation_group = 21;
     catalog.Connection connection = 22;
+    FragmentParallelUnitMappings serving_parallel_unit_mappings = 23;
   }
 }
 

risedev.yml

@@ -597,13 +597,13 @@ profile:
         exporter-port: 1225
         enable-tiered-cache: true
         role: serving
-        parallelism: 6
+        parallelism: 4
       - use: compute-node
         port: 5686
         exporter-port: 1226
         enable-tiered-cache: true
         role: serving
-        parallelism: 6
+        parallelism: 8
       - use: frontend
         port: 4565
         exporter-port: 2222

src/common/Cargo.toml

@@ -47,6 +47,7 @@ itertools = "0.10"
 itoa = "1.0"
 lru = { git = "https://github.com/risingwavelabs/lru-rs.git", rev = "cb2d7c7" }
 memcomparable = { version = "0.2", features = ["decimal"] }
+num-integer = "0.1"
 num-traits = "0.2"
 number_prefix = "0.4.0"
 parking_lot = "0.12"

src/common/common_service/src/observer_manager.rs

@@ -136,6 +136,7 @@ where
             Info::HummockSnapshot(_) => true,
             Info::MetaBackupManifestId(_) => true,
             Info::SystemParams(_) => true,
+            Info::ServingParallelUnitMappings(_) => true,
             Info::Snapshot(_) | Info::HummockWriteLimits(_) => unreachable!(),
         });
 

src/common/src/lib.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #![allow(rustdoc::private_intra_doc_links)]
+#![feature(drain_filter)]
 #![feature(trait_alias)]
 #![feature(binary_heap_drain_sorted)]
 #![feature(is_sorted)]
@@ -66,6 +67,7 @@ pub mod telemetry;
 pub mod metrics;
 pub mod test_utils;
 pub mod types;
+pub mod vnode_mapping;
 
 pub mod test_prelude {
     pub use super::array::{DataChunkTestExt, StreamChunkTestExt};

src/common/src/vnode_mapping/mod.rs

@@ -0,0 +1,15 @@
+// Copyright 2023 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod vnode_placement;