refactor(expr): generate build-from-prost with procedural macros (ris…

…ingwavelabs#8499)

Signed-off-by: Runji Wang <wangrunji0408@163.com>

Cargo.toml

@@ -11,6 +11,7 @@ members = [
   "src/connector",
   "src/ctl",
   "src/expr",
+  "src/expr/macro",
   "src/frontend",
   "src/frontend/planner_test",
   "src/java_binding",

Makefile.toml

@@ -622,12 +622,13 @@ tar xf ${TARGET_PATH} -C "${PREFIX_BIN}/connector-node"
 category = "RiseDev - Build in simulation mode"
 description = "Build in simulation mode"
 dependencies = ["warn-on-missing-tools"]
-env = { RUSTFLAGS = "-Ctarget-cpu=native --cfg tokio_unstable --cfg madsim", RUSTDOCFLAGS = "--cfg madsim", CARGO_TARGET_DIR = "target/sim" }
+env = { CARGO_TARGET_DIR = "target/sim" }
 script = """
 #!/usr/bin/env bash
 set -e
 
 cargo build \
+  --config "target.'cfg(all())'.rustflags = ['--cfg=madsim']" \
   -p risingwave_batch \
   -p risingwave_common \
   -p risingwave_compute \
@@ -646,12 +647,13 @@ cargo build \
 category = "RiseDev - Deterministic Simulation Test"
 description = "Run unit tests in deterministic simulation mode"
 dependencies = ["warn-on-missing-tools"]
-env = { RUSTFLAGS = "-Ctarget-cpu=native --cfg tokio_unstable --cfg madsim", RUSTDOCFLAGS = "--cfg madsim", CARGO_TARGET_DIR = "target/sim" }
+env = { CARGO_TARGET_DIR = "target/sim" }
 script = """
 #!/usr/bin/env bash
 set -e
 
 cargo nextest run \
+  --config "target.'cfg(all())'.rustflags = ['--cfg=madsim']" \
   -p risingwave_batch \
   -p risingwave_common \
   -p risingwave_compute \
@@ -670,12 +672,13 @@ cargo nextest run \
 category = "RiseDev - Simulation scaling tests"
 description = "Run integration scaling tests in deterministic simulation mode"
 dependencies = ["warn-on-missing-tools"]
-env = { RUSTFLAGS = "-Ctarget-cpu=native --cfg tokio_unstable --cfg madsim", RUSTDOCFLAGS = "--cfg madsim", CARGO_TARGET_DIR = "target/sim" }
+env = { CARGO_TARGET_DIR = "target/sim" }
 script = """
 #!/usr/bin/env bash
 set -e
 
 cargo nextest run \
+  --config "target.'cfg(all())'.rustflags = ['--cfg=madsim']" \
   -p risingwave_simulation \
   "$@"
 """
@@ -684,12 +687,13 @@ cargo nextest run \
 category = "RiseDev - Simulation scaling tests"
 description = "Archive integration scaling tests in deterministic simulation mode"
 dependencies = ["warn-on-missing-tools"]
-env = { RUSTFLAGS = "-Ctarget-cpu=native --cfg tokio_unstable --cfg madsim", RUSTDOCFLAGS = "--cfg madsim", CARGO_TARGET_DIR = "target/sim" }
+env = { CARGO_TARGET_DIR = "target/sim" }
 script = """
 #!/usr/bin/env bash
 set -e
 
 cargo nextest archive \
+  --config "target.'cfg(all())'.rustflags = ['--cfg=madsim']" \
   -p risingwave_simulation \
   --archive-file scale-test.tar.zst \
   "$@"
@@ -699,48 +703,58 @@ cargo nextest archive \
 category = "RiseDev - Deterministic Simulation End-to-end Test"
 description = "Run cargo check in deterministic simulation mode"
 dependencies = ["warn-on-missing-tools"]
-env = { RUSTFLAGS = "-Ctarget-cpu=native --cfg tokio_unstable --cfg madsim", RUSTDOCFLAGS = "--cfg madsim", CARGO_TARGET_DIR = "target/sim" }
+env = { CARGO_TARGET_DIR = "target/sim" }
 script = """
 #!/usr/bin/env bash
 set -e
 
-cargo check -p risingwave_simulation --all-targets "$@"
+cargo check \
+  --config "target.'cfg(all())'.rustflags = ['--cfg=madsim']" \
+  -p risingwave_simulation --all-targets "$@"
 """
 
 [tasks.sslt]
 category = "RiseDev - Deterministic Simulation End-to-end Test"
 description = "Run e2e tests in deterministic simulation mode"
 dependencies = ["warn-on-missing-tools"]
-env = { RUSTFLAGS = "-Ctarget-cpu=native --cfg tokio_unstable --cfg madsim", RUSTDOCFLAGS = "--cfg madsim", CARGO_TARGET_DIR = "target/sim" }
+env = { CARGO_TARGET_DIR = "target/sim" }
 script = """
 #!/usr/bin/env bash
 set -e
 
-cargo run -p risingwave_simulation "$@"
+cargo run \
+  --config "target.'cfg(all())'.rustflags = ['--cfg=madsim']" \
+  -p risingwave_simulation "$@"
 """
 
 [tasks.sslt-build-all]
 category = "RiseDev - Deterministic Simulation End-to-end Test"
 description = "Build deterministic simulation runner and tests"
 dependencies = ["warn-on-missing-tools"]
-env = { RUSTFLAGS = "-Ctarget-cpu=native --cfg tokio_unstable --cfg madsim", RUSTDOCFLAGS = "--cfg madsim", CARGO_TARGET_DIR = "target/sim" }
+env = { CARGO_TARGET_DIR = "target/sim" }
 script = """
 #!/usr/bin/env bash
 set -e
 
-cargo build -p risingwave_simulation --tests "$@"
+cargo build \
+  --config "target.'cfg(all())'.rustflags = ['--cfg=madsim']" \
+  -p risingwave_simulation \
+  --tests "$@"
 """
 
 [tasks.sslt-cov]
 category = "RiseDev - Deterministic Simulation End-to-end Test"
 description = "Run e2e tests in deterministic simulation mode and report code coverage"
 dependencies = ["warn-on-missing-tools"]
-env = { RUSTFLAGS = "-Ctarget-cpu=native --cfg tokio_unstable --cfg madsim", RUSTDOCFLAGS = "--cfg madsim", CARGO_TARGET_DIR = "target/sim-cov" }
+env = { CARGO_TARGET_DIR = "target/sim-cov" }
 script = """
 #!/usr/bin/env bash
 set -e
 
-cargo llvm-cov run -p risingwave_simulation --html "$@"
+cargo llvm-cov run \
+  --config "target.'cfg(all())'.rustflags = ['--cfg=madsim']" \
+  -p risingwave_simulation \
+  --html "$@"
 """
 
 [tasks.check-java]

src/batch/src/executor/join/hash_join.rs

@@ -1799,8 +1799,8 @@ mod tests {
     use risingwave_common::test_prelude::DataChunkTestExt;
     use risingwave_common::types::DataType;
     use risingwave_common::util::iter_util::ZipEqDebug;
-    use risingwave_expr::expr::{new_binary_expr, BoxedExpression, InputRefExpression};
-    use risingwave_pb::expr::expr_node::Type;
+    use risingwave_expr::expr::{build, BoxedExpression, Expression, InputRefExpression};
+    use risingwave_pb::expr::expr_node::PbType;
 
     use super::{
         ChunkedData, HashJoinExecutor, JoinType, LeftNonEquiJoinState, RightNonEquiJoinState, RowId,
@@ -1985,13 +1985,13 @@ mod tests {
         }
 
         fn create_cond() -> BoxedExpression {
-            let left_expr = InputRefExpression::new(DataType::Float32, 1);
-            let right_expr = InputRefExpression::new(DataType::Float64, 3);
-            new_binary_expr(
-                Type::LessThan,
+            build(
+                PbType::LessThan,
                 DataType::Boolean,
-                Box::new(left_expr),
-                Box::new(right_expr),
+                vec![
+                    InputRefExpression::new(DataType::Float32, 1).boxed(),
+                    InputRefExpression::new(DataType::Float64, 3).boxed(),
+                ],
             )
             .unwrap()
         }

src/batch/src/executor/join/local_lookup_join.rs

@@ -462,9 +462,9 @@ mod tests {
     use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
     use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
     use risingwave_expr::expr::{
-        new_binary_expr, BoxedExpression, InputRefExpression, LiteralExpression,
+        build, BoxedExpression, Expression, InputRefExpression, LiteralExpression,
     };
-    use risingwave_pb::expr::expr_node::Type;
+    use risingwave_pb::expr::expr_node::PbType;
 
     use super::LocalLookupJoinExecutorArgs;
     use crate::executor::join::JoinType;
@@ -676,14 +676,13 @@ mod tests {
         );
 
         let condition = Some(
-            new_binary_expr(
-                Type::LessThan,
+            build(
+                PbType::LessThan,
                 DataType::Boolean,
-                Box::new(LiteralExpression::new(
-                    DataType::Int32,
-                    Some(ScalarImpl::Int32(5)),
-                )),
-                Box::new(InputRefExpression::new(DataType::Float32, 3)),
+                vec![
+                    LiteralExpression::new(DataType::Int32, Some(ScalarImpl::Int32(5))).boxed(),
+                    InputRefExpression::new(DataType::Float32, 3).boxed(),
+                ],
             )
             .unwrap(),
         );
@@ -705,14 +704,13 @@ mod tests {
         );
 
         let condition = Some(
-            new_binary_expr(
-                Type::LessThan,
+            build(
+                PbType::LessThan,
                 DataType::Boolean,
-                Box::new(LiteralExpression::new(
-                    DataType::Int32,
-                    Some(ScalarImpl::Int32(5)),
-                )),
-                Box::new(InputRefExpression::new(DataType::Float32, 3)),
+                vec![
+                    LiteralExpression::new(DataType::Int32, Some(ScalarImpl::Int32(5))).boxed(),
+                    InputRefExpression::new(DataType::Float32, 3).boxed(),
+                ],
             )
             .unwrap(),
         );
@@ -730,14 +728,13 @@ mod tests {
         );
 
         let condition = Some(
-            new_binary_expr(
-                Type::LessThan,
+            build(
+                PbType::LessThan,
                 DataType::Boolean,
-                Box::new(LiteralExpression::new(
-                    DataType::Int32,
-                    Some(ScalarImpl::Int32(5)),
-                )),
-                Box::new(InputRefExpression::new(DataType::Float32, 3)),
+                vec![
+                    LiteralExpression::new(DataType::Int32, Some(ScalarImpl::Int32(5))).boxed(),
+                    InputRefExpression::new(DataType::Float32, 3).boxed(),
+                ],
             )
             .unwrap(),
         );
@@ -756,14 +753,13 @@ mod tests {
         );
 
         let condition = Some(
-            new_binary_expr(
-                Type::LessThan,
+            build(
+                PbType::LessThan,
                 DataType::Boolean,
-                Box::new(LiteralExpression::new(
-                    DataType::Int32,
-                    Some(ScalarImpl::Int32(5)),
-                )),
-                Box::new(InputRefExpression::new(DataType::Float32, 3)),
+                vec![
+                    LiteralExpression::new(DataType::Int32, Some(ScalarImpl::Int32(5))).boxed(),
+                    InputRefExpression::new(DataType::Float32, 3).boxed(),
+                ],
             )
             .unwrap(),
         );

src/batch/src/executor/join/nested_loop_join.rs

@@ -470,8 +470,8 @@ mod tests {
     use risingwave_common::array::*;
     use risingwave_common::catalog::{Field, Schema};
     use risingwave_common::types::DataType;
-    use risingwave_expr::expr::{new_binary_expr, InputRefExpression};
-    use risingwave_pb::expr::expr_node::Type;
+    use risingwave_expr::expr::{build, InputRefExpression};
+    use risingwave_pb::expr::expr_node::PbType;
 
     use crate::executor::join::nested_loop_join::NestedLoopJoinExecutor;
     use crate::executor::join::JoinType;
@@ -587,11 +587,13 @@ mod tests {
             };
 
             Box::new(NestedLoopJoinExecutor::new(
-                new_binary_expr(
-                    Type::Equal,
+                build(
+                    PbType::Equal,
                     DataType::Boolean,
-                    Box::new(InputRefExpression::new(DataType::Int32, 0)),
-                    Box::new(InputRefExpression::new(DataType::Int32, 2)),
+                    vec![
+                        Box::new(InputRefExpression::new(DataType::Int32, 0)),
+                        Box::new(InputRefExpression::new(DataType::Int32, 2)),
+                    ],
                 )
                 .unwrap(),
                 join_type,

src/common/src/array/jsonb_array.rs

@@ -274,6 +274,26 @@ impl JsonbRef<'_> {
     }
 }
 
+impl FromIterator<Option<JsonbVal>> for JsonbArray {
+    fn from_iter<I: IntoIterator<Item = Option<JsonbVal>>>(iter: I) -> Self {
+        let iter = iter.into_iter();
+        let mut builder = <Self as Array>::Builder::new(iter.size_hint().0);
+        for i in iter {
+            match i {
+                Some(x) => builder.append(Some(x.as_scalar_ref())),
+                None => builder.append(None),
+            }
+        }
+        builder.finish()
+    }
+}
+
+impl FromIterator<JsonbVal> for JsonbArray {
+    fn from_iter<I: IntoIterator<Item = JsonbVal>>(iter: I) -> Self {
+        iter.into_iter().map(Some).collect()
+    }
+}
+
 #[derive(Debug)]
 pub struct JsonbArrayBuilder {
     bitmap: BitmapBuilder,

src/common/src/array/list_array.rs

@@ -156,7 +156,7 @@ pub struct ListArray {
     bitmap: Bitmap,
     pub(super) offsets: Vec<u32>,
     pub(super) value: Box<ArrayImpl>,
-    value_type: DataType,
+    pub(super) value_type: DataType,
 }
 
 impl Array for ListArray {

src/common/src/array/mod.rs

@@ -59,13 +59,13 @@ pub use list_array::{ListArray, ListArrayBuilder, ListRef, ListValue};
 use paste::paste;
 pub use primitive_array::{PrimitiveArray, PrimitiveArrayBuilder, PrimitiveArrayItemType};
 use risingwave_pb::data::{PbArray, PbArrayType};
+pub use serial_array::{Serial, SerialArray, SerialArrayBuilder};
 pub use stream_chunk::{Op, StreamChunk, StreamChunkTestExt};
 pub use struct_array::{StructArray, StructArrayBuilder, StructRef, StructValue};
 pub use utf8_array::*;
 pub use vis::{Vis, VisRef};
 
 pub use self::error::ArrayError;
-use crate::array::serial_array::{Serial, SerialArray, SerialArrayBuilder};
 use crate::buffer::Bitmap;
 use crate::types::*;
 use crate::util::iter_util::ZipEqFast;