refactor(batch): split executors into a separate crate (#19742)

Signed-off-by: Bugen Zhao <i@bugenzhao.com>

Cargo.toml

@@ -3,6 +3,7 @@ cargo-features = ["profile-rustflags"]
 [workspace]
 members = [
     "src/batch",
+    "src/batch/executors",
     "src/bench",
     "src/cmd",
     "src/cmd_all",
@@ -205,6 +206,7 @@ tracing-opentelemetry = "0.25"
 rand = { version = "0.8", features = ["small_rng"] }
 risingwave_backup = { path = "./src/storage/backup" }
 risingwave_batch = { path = "./src/batch" }
+risingwave_batch_executors = { path = "./src/batch/executors" }
 risingwave_cmd = { path = "./src/cmd" }
 risingwave_common = { path = "./src/common" }
 risingwave_common_service = { path = "./src/common/common_service" }

src/batch/Cargo.toml

@@ -15,18 +15,12 @@ normal = ["workspace-hack"]
 
 [dependencies]
 anyhow = "1"
-assert_matches = "1"
 async-recursion = "1"
 async-trait = "0.1"
-bytes = "1"
-chrono = "0.4"
 either = "1"
-foyer = { workspace = true }
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 futures-async-stream = { workspace = true }
 futures-util = "0.3"
-hashbrown = { workspace = true }
-hytra = "0.1.2"
 iceberg = { workspace = true }
 itertools = { workspace = true }
 linkme = { workspace = true }
@@ -35,21 +29,16 @@ mysql_async = { workspace = true }
 opendal = { workspace = true }
 parking_lot = { workspace = true }
 parquet = { workspace = true }
-paste = "1"
 prometheus = { version = "0.13", features = ["process"] }
-prost = { workspace = true }
 rand = { workspace = true }
 risingwave_common = { workspace = true }
-risingwave_common_estimate_size = { workspace = true }
 risingwave_connector = { workspace = true }
 risingwave_dml = { workspace = true }
 risingwave_expr = { workspace = true }
 risingwave_hummock_sdk = { workspace = true }
 risingwave_pb = { workspace = true }
 risingwave_rpc_client = { workspace = true }
 risingwave_storage = { workspace = true }
-rust_decimal = "1"
-rw_futures_util = { workspace = true }
 scopeguard = "1"
 serde_json = "1"
 thiserror = "1"
@@ -63,57 +52,21 @@ tokio = { version = "0.2", package = "madsim-tokio", features = [
     "signal",
     "fs",
 ] }
-tokio-metrics = "0.3.0"
 tokio-postgres = "0.7"
 tokio-stream = { workspace = true }
-tokio-util = { workspace = true }
 tonic = { workspace = true }
 tracing = "0.1"
 twox-hash = "1"
-uuid = { version = "1", features = ["v4"] }
 
 [target.'cfg(not(madsim))'.dependencies]
 workspace-hack = { path = "../workspace-hack" }
 
 [dev-dependencies]
 criterion = { workspace = true, features = ["async_tokio", "async"] }
 rand = { workspace = true }
-risingwave_expr_impl = { workspace = true }
 risingwave_hummock_sdk = { workspace = true }
 tempfile = "3"
 tikv-jemallocator = { workspace = true }
 
-[[bench]]
-name = "filter"
-harness = false
-
-[[bench]]
-name = "nested_loop_join"
-harness = false
-
-[[bench]]
-name = "hash_join"
-harness = false
-
-[[bench]]
-name = "sort"
-harness = false
-
-[[bench]]
-name = "top_n"
-harness = false
-
-[[bench]]
-name = "hash_agg"
-harness = false
-
-[[bench]]
-name = "expand"
-harness = false
-
-[[bench]]
-name = "limit"
-harness = false
-
 [lints]
 workspace = true

src/batch/executors/Cargo.toml

@@ -0,0 +1,105 @@
+[package]
+name = "risingwave_batch_executors"
+version = { workspace = true }
+edition = { workspace = true }
+homepage = { workspace = true }
+keywords = { workspace = true }
+license = { workspace = true }
+repository = { workspace = true }
+
+[package.metadata.cargo-machete]
+ignored = ["workspace-hack"]
+
+[package.metadata.cargo-udeps.ignore]
+normal = ["workspace-hack"]
+
+[dependencies]
+anyhow = "1"
+assert_matches = "1"
+async-trait = "0.1"
+bytes = "1"
+either = "1"
+foyer = { workspace = true }
+futures = { version = "0.3", default-features = false, features = ["alloc"] }
+futures-async-stream = { workspace = true }
+futures-util = "0.3"
+hashbrown = { workspace = true }
+iceberg = { workspace = true }
+itertools = { workspace = true }
+linkme = { workspace = true }
+mysql_async = { workspace = true }
+opendal = { workspace = true }
+prometheus = { version = "0.13", features = ["process"] }
+rand = { workspace = true }
+risingwave_batch = { workspace = true }
+risingwave_common = { workspace = true }
+risingwave_common_estimate_size = { workspace = true }
+risingwave_connector = { workspace = true }
+risingwave_dml = { workspace = true }
+risingwave_expr = { workspace = true }
+risingwave_hummock_sdk = { workspace = true }
+risingwave_pb = { workspace = true }
+risingwave_rpc_client = { workspace = true }
+risingwave_storage = { workspace = true }
+rw_futures_util = { workspace = true }
+scopeguard = "1"
+thiserror-ext = { workspace = true }
+tokio = { version = "0.2", package = "madsim-tokio", features = [
+    "rt",
+    "rt-multi-thread",
+    "sync",
+    "macros",
+    "time",
+    "signal",
+    "fs",
+] }
+tokio-postgres = "0.7"
+tokio-stream = { workspace = true }
+tracing = "0.1"
+uuid = { version = "1", features = ["v4"] }
+
+[target.'cfg(not(madsim))'.dependencies]
+workspace-hack = { path = "../../workspace-hack" }
+
+[dev-dependencies]
+criterion = { workspace = true, features = ["async_tokio", "async"] }
+rand = { workspace = true }
+risingwave_expr_impl = { workspace = true }
+risingwave_hummock_sdk = { workspace = true }
+tempfile = "3"
+tikv-jemallocator = { workspace = true }
+
+[[bench]]
+name = "filter"
+harness = false
+
+[[bench]]
+name = "nested_loop_join"
+harness = false
+
+[[bench]]
+name = "hash_join"
+harness = false
+
+[[bench]]
+name = "sort"
+harness = false
+
+[[bench]]
+name = "top_n"
+harness = false
+
+[[bench]]
+name = "hash_agg"
+harness = false
+
+[[bench]]
+name = "expand"
+harness = false
+
+[[bench]]
+name = "limit"
+harness = false
+
+[lints]
+workspace = true

src/batch/benches/README.md → src/batch/executors/benches/README.md

@@ -7,28 +7,28 @@ We use [criterion](https://bheisler.github.io/criterion.rs/book/index.html) micr
 Run all benchmarks
 
 ```bash
-cargo bench -p risingwave_batch
+cargo bench -p risingwave_batch_executors
 ```
 
 Run a specific benchmark
 
 ```bash
-cargo bench -p risingwave_batch -- <filter>
+cargo bench -p risingwave_batch_executors -- <filter>
 ```
 
 where `<filter>` is a regular expression matching the benchmark ID, e.g.,
 `top_n.rs` uses `BenchmarkId::new("TopNExecutor", params)` , so we can run TopN benchmarks with
 
 ```bash
 # All TopN benchmarks
-cargo bench -p risingwave_batch -- TopN
+cargo bench -p risingwave_batch_executors -- TopN
 # One specific setting of TopN benchmarks
-cargo bench -p risingwave_batch -- "TopNExecutor/2048\(single_column: true\)"
+cargo bench -p risingwave_batch_executors -- "TopNExecutor/2048\(single_column: true\)"
 ```
 
-> *Note*: `-p risingwave_batch` can be omitted if you are in the `src/batch` directory.
+> *Note*: `-p risingwave_batch_executors` can be omitted if you are in the `src/batch/executors` directory.
 
 ## Add new Benchmarks
 
-* Add benchmark target to `src/batch/Cargo.toml`
-* Implement benchmarks in `src/batch/benches`, referring to existing ones
+* Add benchmark target to `src/batch/executors/Cargo.toml`
+* Implement benchmarks in `src/batch/executors/benches`, referring to existing ones