refactor(meta): split meta into smaller crates (step 1) (#12924)

Cargo.toml

@@ -7,6 +7,7 @@ members = [
   "src/cmd_all",
   "src/common",
   "src/common/common_service",
+  "src/common/heap_profiling",
   "src/compute",
   "src/connector",
   "src/ctl",
@@ -18,6 +19,8 @@ members = [
   "src/java_binding",
   "src/jni_core",
   "src/meta",
+  "src/meta/node",
+  "src/meta/service",
   "src/meta/src/model_v2/migration",
   "src/object_store",
   "src/prost",
@@ -139,6 +142,8 @@ risingwave_hummock_sdk = { path = "./src/storage/hummock_sdk" }
 risingwave_hummock_test = { path = "./src/storage/hummock_test" }
 risingwave_hummock_trace = { path = "./src/storage/hummock_trace" }
 risingwave_meta = { path = "./src/meta" }
+risingwave_meta_service = { path = "./src/meta/service" }
+risingwave_meta_node = { path = "./src/meta/node" }
 risingwave_object_store = { path = "./src/object_store" }
 risingwave_pb = { path = "./src/prost" }
 risingwave_rpc_client = { path = "./src/rpc_client" }

src/batch/Cargo.toml

@@ -17,7 +17,6 @@ normal = ["workspace-hack"]
 anyhow = "1"
 assert_matches = "1"
 async-recursion = "1"
-async-stream = "0.3.5"
 async-trait = "0.1"
 either = "1"
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
@@ -53,7 +52,6 @@ tokio-metrics = "0.3.0"
 tokio-stream = "0.1"
 tonic = { workspace = true }
 tracing = "0.1"
-uuid = "1"
 
 [target.'cfg(enable_task_local_alloc)'.dependencies]
 task_stats_alloc = { path = "../utils/task_stats_alloc" }

src/batch/src/lib.rs

@@ -15,7 +15,6 @@
 #![expect(dead_code)]
 #![allow(clippy::derive_partial_eq_without_eq)]
 #![feature(trait_alias)]
-#![feature(binary_heap_drain_sorted)]
 #![feature(exact_size_is_empty)]
 #![feature(type_alias_impl_trait)]
 #![cfg_attr(coverage, feature(no_coverage))]

src/cmd/Cargo.toml

@@ -27,7 +27,7 @@ risingwave_compute = { workspace = true }
 risingwave_ctl = { workspace = true }
 risingwave_expr_impl = { workspace = true }
 risingwave_frontend = { workspace = true }
-risingwave_meta = { workspace = true }
+risingwave_meta_node = { workspace = true }
 risingwave_rt = { workspace = true }
 tikv-jemallocator = { workspace = true, features = [
     "unprefixed_malloc_on_supported_platforms",

src/cmd/src/lib.rs

@@ -16,7 +16,7 @@ use risingwave_compactor::CompactorOpts;
 use risingwave_compute::ComputeNodeOpts;
 use risingwave_ctl::CliOpts as CtlOpts;
 use risingwave_frontend::FrontendOpts;
-use risingwave_meta::MetaNodeOpts;
+use risingwave_meta_node::MetaNodeOpts;
 use risingwave_rt::{init_risingwave_logger, main_okk, LoggerSettings};
 
 /// Define the `main` function for a component.
@@ -48,7 +48,7 @@ pub fn compute(opts: ComputeNodeOpts) {
 
 pub fn meta(opts: MetaNodeOpts) {
     init_risingwave_logger(LoggerSettings::new("meta"));
-    main_okk(risingwave_meta::start(opts));
+    main_okk(risingwave_meta_node::start(opts));
 }
 
 pub fn frontend(opts: FrontendOpts) {

src/cmd_all/Cargo.toml

@@ -31,7 +31,7 @@ risingwave_compute = { workspace = true }
 risingwave_ctl = { workspace = true }
 risingwave_expr_impl = { workspace = true }
 risingwave_frontend = { workspace = true }
-risingwave_meta = { workspace = true }
+risingwave_meta_node = { workspace = true }
 risingwave_rt = { workspace = true }
 shell-words = "1.1.0"
 strum = "0.25"

src/cmd_all/src/README.md

@@ -26,5 +26,5 @@ You may run and reference the [demo script](../scripts/e2e-full-standalone-demo.
 
 Standalone mode simply passes the options to the corresponding node, and starts them in the same process.
 
-For example `--meta-opts` is parsed, and then Meta Node's entrypoint, `risingwave_meta::start`, is called with the parsed options.
+For example `--meta-opts` is parsed, and then Meta Node's entrypoint, `risingwave_meta_node::start`, is called with the parsed options.
 If any option is missing, the corresponding node will not be started.

src/cmd_all/src/bin/risingwave.rs

@@ -25,7 +25,7 @@ use risingwave_compactor::CompactorOpts;
 use risingwave_compute::ComputeNodeOpts;
 use risingwave_ctl::CliOpts as CtlOpts;
 use risingwave_frontend::FrontendOpts;
-use risingwave_meta::MetaNodeOpts;
+use risingwave_meta_node::MetaNodeOpts;
 use strum::IntoEnumIterator;
 use strum_macros::{Display, EnumIter, EnumString, IntoStaticStr};
 use tracing::Level;

src/cmd_all/src/playground.rs

@@ -159,9 +159,9 @@ pub async fn playground(opts: PlaygroundOpts) -> Result<()> {
             RisingWaveService::Meta(mut opts) => {
                 opts.insert(0, "meta-node".into());
                 tracing::info!("starting meta-node thread with cli args: {:?}", opts);
-                let opts = risingwave_meta::MetaNodeOpts::parse_from(opts);
+                let opts = risingwave_meta_node::MetaNodeOpts::parse_from(opts);
                 let _meta_handle = tokio::spawn(async move {
-                    risingwave_meta::start(opts).await;
+                    risingwave_meta_node::start(opts).await;
                     tracing::warn!("meta is stopped, shutdown all nodes");
                     // As a playground, it's fine to just kill everything.
                     if idle_exit {