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

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",

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/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;

src/common/src/types/mod.rs

@@ -20,6 +20,7 @@ use bytes::{Buf, BufMut, Bytes};
 use num_traits::Float;
 use parse_display::{Display, FromStr};
 use postgres_types::FromSql;
+use risingwave_pb::data::data_type::PbTypeName;
 use risingwave_pb::data::PbDataType;
 use serde::{Deserialize, Serialize};
 
@@ -35,7 +36,6 @@ use std::fmt::Debug;
 use std::str::{FromStr, Utf8Error};
 
 pub use native_type::*;
-use risingwave_pb::data::data_type::TypeName;
 pub use scalar_impl::*;
 pub use successor::*;
 pub mod chrono_wrapper;
@@ -210,32 +210,57 @@ pub fn unnested_list_type(datatype: DataType) -> DataType {
 impl From<&PbDataType> for DataType {
     fn from(proto: &PbDataType) -> DataType {
         match proto.get_type_name().expect("missing type field") {
-            TypeName::Int16 => DataType::Int16,
-            TypeName::Int32 => DataType::Int32,
-            TypeName::Int64 => DataType::Int64,
-            TypeName::Serial => DataType::Serial,
-            TypeName::Float => DataType::Float32,
-            TypeName::Double => DataType::Float64,
-            TypeName::Boolean => DataType::Boolean,
-            TypeName::Varchar => DataType::Varchar,
-            TypeName::Date => DataType::Date,
-            TypeName::Time => DataType::Time,
-            TypeName::Timestamp => DataType::Timestamp,
-            TypeName::Timestamptz => DataType::Timestamptz,
-            TypeName::Decimal => DataType::Decimal,
-            TypeName::Interval => DataType::Interval,
-            TypeName::Bytea => DataType::Bytea,
-            TypeName::Jsonb => DataType::Jsonb,
-            TypeName::Struct => {
+            PbTypeName::Int16 => DataType::Int16,
+            PbTypeName::Int32 => DataType::Int32,
+            PbTypeName::Int64 => DataType::Int64,
+            PbTypeName::Serial => DataType::Serial,
+            PbTypeName::Float => DataType::Float32,
+            PbTypeName::Double => DataType::Float64,
+            PbTypeName::Boolean => DataType::Boolean,
+            PbTypeName::Varchar => DataType::Varchar,
+            PbTypeName::Date => DataType::Date,
+            PbTypeName::Time => DataType::Time,
+            PbTypeName::Timestamp => DataType::Timestamp,
+            PbTypeName::Timestamptz => DataType::Timestamptz,
+            PbTypeName::Decimal => DataType::Decimal,
+            PbTypeName::Interval => DataType::Interval,
+            PbTypeName::Bytea => DataType::Bytea,
+            PbTypeName::Jsonb => DataType::Jsonb,
+            PbTypeName::Struct => {
                 let fields: Vec<DataType> = proto.field_type.iter().map(|f| f.into()).collect_vec();
                 let field_names: Vec<String> = proto.field_names.iter().cloned().collect_vec();
                 DataType::new_struct(fields, field_names)
             }
-            TypeName::List => DataType::List {
+            PbTypeName::List => DataType::List {
                 // The first (and only) item is the list element type.
                 datatype: Box::new((&proto.field_type[0]).into()),
             },
-            TypeName::TypeUnspecified => unreachable!(),
+            PbTypeName::TypeUnspecified => unreachable!(),
+        }
+    }
+}
+
+impl From<DataTypeName> for PbTypeName {
+    fn from(type_name: DataTypeName) -> Self {
+        match type_name {
+            DataTypeName::Boolean => PbTypeName::Boolean,
+            DataTypeName::Int16 => PbTypeName::Int16,
+            DataTypeName::Int32 => PbTypeName::Int32,
+            DataTypeName::Int64 => PbTypeName::Int64,
+            DataTypeName::Serial => PbTypeName::Serial,
+            DataTypeName::Float32 => PbTypeName::Float,
+            DataTypeName::Float64 => PbTypeName::Double,
+            DataTypeName::Varchar => PbTypeName::Varchar,
+            DataTypeName::Date => PbTypeName::Date,
+            DataTypeName::Timestamp => PbTypeName::Timestamp,
+            DataTypeName::Timestamptz => PbTypeName::Timestamptz,
+            DataTypeName::Time => PbTypeName::Time,
+            DataTypeName::Interval => PbTypeName::Interval,
+            DataTypeName::Decimal => PbTypeName::Decimal,
+            DataTypeName::Bytea => PbTypeName::Bytea,
+            DataTypeName::Jsonb => PbTypeName::Jsonb,
+            DataTypeName::Struct => PbTypeName::Struct,
+            DataTypeName::List => PbTypeName::List,
         }
     }
 }
@@ -273,26 +298,26 @@ impl DataType {
         }
     }
 
-    pub fn prost_type_name(&self) -> TypeName {
+    pub fn prost_type_name(&self) -> PbTypeName {
         match self {
-            DataType::Int16 => TypeName::Int16,
-            DataType::Int32 => TypeName::Int32,
-            DataType::Int64 => TypeName::Int64,
-            DataType::Serial => TypeName::Serial,
-            DataType::Float32 => TypeName::Float,
-            DataType::Float64 => TypeName::Double,
-            DataType::Boolean => TypeName::Boolean,
-            DataType::Varchar => TypeName::Varchar,
-            DataType::Date => TypeName::Date,
-            DataType::Time => TypeName::Time,
-            DataType::Timestamp => TypeName::Timestamp,
-            DataType::Timestamptz => TypeName::Timestamptz,
-            DataType::Decimal => TypeName::Decimal,
-            DataType::Interval => TypeName::Interval,
-            DataType::Jsonb => TypeName::Jsonb,
-            DataType::Struct { .. } => TypeName::Struct,
-            DataType::List { .. } => TypeName::List,
-            DataType::Bytea => TypeName::Bytea,
+            DataType::Int16 => PbTypeName::Int16,
+            DataType::Int32 => PbTypeName::Int32,
+            DataType::Int64 => PbTypeName::Int64,
+            DataType::Serial => PbTypeName::Serial,
+            DataType::Float32 => PbTypeName::Float,
+            DataType::Float64 => PbTypeName::Double,
+            DataType::Boolean => PbTypeName::Boolean,
+            DataType::Varchar => PbTypeName::Varchar,
+            DataType::Date => PbTypeName::Date,
+            DataType::Time => PbTypeName::Time,
+            DataType::Timestamp => PbTypeName::Timestamp,
+            DataType::Timestamptz => PbTypeName::Timestamptz,
+            DataType::Decimal => PbTypeName::Decimal,
+            DataType::Interval => PbTypeName::Interval,
+            DataType::Jsonb => PbTypeName::Jsonb,
+            DataType::Struct { .. } => PbTypeName::Struct,
+            DataType::List { .. } => PbTypeName::List,
+            DataType::Bytea => PbTypeName::Bytea,
         }
     }
 

src/expr/Cargo.toml

@@ -22,6 +22,7 @@ arrow-schema = "34"
 async-trait = "0.1"
 chrono = { version = "0.4", default-features = false, features = ["clock", "std"] }
 chrono-tz = { version = "0.7", features = ["case-insensitive"] }
+ctor = "0.1"
 dyn-clone = "1"
 either = "1"
 futures-util = "0.3"
@@ -33,12 +34,14 @@ parse-display = "0.6"
 paste = "1"
 regex = "1"
 risingwave_common = { path = "../common" }
+risingwave_expr_macro = { path = "macro" }
 risingwave_pb = { path = "../prost" }
 risingwave_udf = { path = "../udf" }
 speedate = "0.7.0"
 static_assertions = "1"
 thiserror = "1"
 tokio = { version = "0.2", package = "madsim-tokio", features = ["rt", "rt-multi-thread", "sync", "macros", "time", "signal"] }
+tracing = "0.1"
 
 [target.'cfg(not(madsim))'.dependencies]
 workspace-hack = { path = "../workspace-hack" }