refactor(expr): separate function implementations into a new crate (#…

…12485)

Signed-off-by: Runji Wang <[email protected]>

Cargo.toml

@@ -10,7 +10,8 @@ members = [
   "src/compute",
   "src/connector",
   "src/ctl",
-  "src/expr",
+  "src/expr/core",
+  "src/expr/impl",
   "src/expr/macro",
   "src/frontend",
   "src/frontend/planner_test",
@@ -125,7 +126,8 @@ risingwave_compactor = { path = "./src/storage/compactor" }
 risingwave_compute = { path = "./src/compute" }
 risingwave_ctl = { path = "./src/ctl" }
 risingwave_connector = { path = "./src/connector" }
-risingwave_expr = { path = "./src/expr" }
+risingwave_expr = { path = "./src/expr/core" }
+risingwave_expr_impl = { path = "./src/expr/impl" }
 risingwave_frontend = { path = "./src/frontend" }
 risingwave_hummock_sdk = { path = "./src/storage/hummock_sdk" }
 risingwave_hummock_test = { path = "./src/storage/hummock_test" }

src/batch/Cargo.toml

@@ -64,6 +64,7 @@ workspace-hack = { path = "../workspace-hack" }
 [dev-dependencies]
 criterion = { workspace = true, features = ["async_tokio", "async"] }
 rand = "0.8"
+risingwave_expr_impl = { workspace = true }
 tempfile = "3"
 
 [target.'cfg(unix)'.dev-dependencies]

src/batch/benches/hash_agg.rs

@@ -22,7 +22,7 @@ use risingwave_common::catalog::{Field, Schema};
 use risingwave_common::memory::MemoryContext;
 use risingwave_common::types::DataType;
 use risingwave_common::{enable_jemalloc_on_unix, hash};
-use risingwave_expr::agg::{AggCall, AggKind};
+use risingwave_expr::aggregate::{AggCall, AggKind};
 use risingwave_pb::expr::{PbAggCall, PbInputRef};
 use tokio::runtime::Runtime;
 use utils::{create_input, execute_executor};

src/batch/src/executor/aggregation/distinct.rs

@@ -20,7 +20,7 @@ use risingwave_common::buffer::BitmapBuilder;
 use risingwave_common::estimate_size::EstimateSize;
 use risingwave_common::row::{OwnedRow, Row};
 use risingwave_common::types::{DataType, Datum};
-use risingwave_expr::agg::{
+use risingwave_expr::aggregate::{
     AggStateDyn, AggregateFunction, AggregateState, BoxedAggregateFunction,
 };
 use risingwave_expr::Result;
@@ -112,7 +112,7 @@ mod tests {
     use risingwave_common::array::StreamChunk;
     use risingwave_common::test_prelude::StreamChunkTestExt;
     use risingwave_common::types::{Datum, Decimal};
-    use risingwave_expr::agg::AggCall;
+    use risingwave_expr::aggregate::AggCall;
 
     use super::super::build;
 

src/batch/src/executor/aggregation/filter.rs

@@ -17,7 +17,7 @@ use std::sync::Arc;
 
 use risingwave_common::array::StreamChunk;
 use risingwave_common::types::{DataType, Datum};
-use risingwave_expr::agg::{AggregateFunction, AggregateState, BoxedAggregateFunction};
+use risingwave_expr::aggregate::{AggregateFunction, AggregateState, BoxedAggregateFunction};
 use risingwave_expr::expr::Expression;
 use risingwave_expr::Result;
 
@@ -74,7 +74,7 @@ impl AggregateFunction for Filter {
 #[cfg(test)]
 mod tests {
     use risingwave_common::test_prelude::StreamChunkTestExt;
-    use risingwave_expr::agg::{build_append_only, AggCall};
+    use risingwave_expr::aggregate::{build_append_only, AggCall};
     use risingwave_expr::expr::{build_from_pretty, Expression, LiteralExpression};
 
     use super::*;

src/batch/src/executor/aggregation/mod.rs

@@ -20,7 +20,7 @@ mod filter;
 mod orderby;
 mod projection;
 
-use risingwave_expr::agg::{self, AggCall, BoxedAggregateFunction};
+use risingwave_expr::aggregate::{build_append_only, AggCall, BoxedAggregateFunction};
 use risingwave_expr::Result;
 
 use self::distinct::Distinct;
@@ -30,7 +30,7 @@ use self::projection::Projection;
 
 /// Build an `BoxedAggregateFunction` from `AggCall`.
 pub fn build(agg: &AggCall) -> Result<BoxedAggregateFunction> {
-    let mut aggregator = agg::build_append_only(agg)?;
+    let mut aggregator = build_append_only(agg)?;
 
     if agg.distinct {
         aggregator = Box::new(Distinct::new(aggregator));

src/batch/src/executor/aggregation/orderby.rs

@@ -22,7 +22,7 @@ use risingwave_common::types::{DataType, Datum};
 use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
 use risingwave_common::util::memcmp_encoding;
 use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
-use risingwave_expr::agg::{
+use risingwave_expr::aggregate::{
     AggStateDyn, AggregateFunction, AggregateState, BoxedAggregateFunction,
 };
 use risingwave_expr::{ExprError, Result};
@@ -151,7 +151,7 @@ impl AggregateFunction for ProjectionOrderBy {
 mod tests {
     use risingwave_common::array::{ListValue, StreamChunk};
     use risingwave_common::test_prelude::StreamChunkTestExt;
-    use risingwave_expr::agg::AggCall;
+    use risingwave_expr::aggregate::AggCall;
 
     use super::super::build;
 

src/batch/src/executor/aggregation/projection.rs

@@ -16,7 +16,7 @@ use std::ops::Range;
 
 use risingwave_common::array::StreamChunk;
 use risingwave_common::types::{DataType, Datum};
-use risingwave_expr::agg::{AggregateFunction, AggregateState, BoxedAggregateFunction};
+use risingwave_expr::aggregate::{AggregateFunction, AggregateState, BoxedAggregateFunction};
 use risingwave_expr::Result;
 
 pub struct Projection {

src/batch/src/executor/hash_agg.rs

@@ -24,7 +24,7 @@ use risingwave_common::hash::{HashKey, HashKeyDispatcher, PrecomputedBuildHasher
 use risingwave_common::memory::MemoryContext;
 use risingwave_common::types::DataType;
 use risingwave_common::util::iter_util::ZipEqFast;
-use risingwave_expr::agg::{AggCall, AggregateState, BoxedAggregateFunction};
+use risingwave_expr::aggregate::{AggCall, AggregateState, BoxedAggregateFunction};
 use risingwave_pb::batch_plan::plan_node::NodeBody;
 use risingwave_pb::batch_plan::HashAggNode;
 

src/batch/src/executor/hop_window.rs

@@ -239,7 +239,7 @@ mod tests {
             6 2 ^10:42:00
             7 1 ^10:51:00
             8 3 ^11:02:00"
-                .replace('^', "2022-2-2T"),
+                .replace('^', "2022-02-02T"),
         );
         let mut mock_executor = MockExecutor::new(schema.clone());
         mock_executor.add(chunk);
@@ -326,7 +326,7 @@ mod tests {
                 6 2 ^10:42:00 ^10:14:00 ^10:44:00
                 7 1 ^10:51:00 ^10:29:00 ^10:59:00
                 8 3 ^11:02:00 ^10:44:00 ^11:14:00"
-                    .replace('^', "2022-2-2T"),
+                    .replace('^', "2022-02-02T"),
             )
         );
         assert_eq!(
@@ -341,7 +341,7 @@ mod tests {
                 6 2 ^10:42:00 ^10:14:00 ^10:44:00
                 7 1 ^10:51:00 ^10:29:00 ^10:59:00
                 8 3 ^11:02:00 ^10:44:00 ^11:14:00"
-                    .replace('^', "2022-2-2T"),
+                    .replace('^', "2022-02-02T"),
             )
         );
     }
@@ -371,7 +371,7 @@ mod tests {
                 6 2 ^10:42:00 ^10:15:00 ^10:45:00
                 7 1 ^10:51:00 ^10:30:00 ^11:00:00
                 8 3 ^11:02:00 ^10:45:00 ^11:15:00"
-                    .replace('^', "2022-2-2T"),
+                    .replace('^', "2022-02-02T"),
             )
         );
 
@@ -388,7 +388,7 @@ mod tests {
                 6 2 ^10:42:00 ^10:30:00 ^11:00:00
                 7 1 ^10:51:00 ^10:45:00 ^11:15:00
                 8 3 ^11:02:00 ^11:00:00 ^11:30:00"
-                    .replace('^', "2022-2-2T"),
+                    .replace('^', "2022-02-02T"),
             )
         );
     }
@@ -415,7 +415,7 @@ mod tests {
                    2 ^10:15:00 ^10:45:00 ^10:42:00
                    1 ^10:30:00 ^11:00:00 ^10:51:00
                    3 ^10:45:00 ^11:15:00 ^11:02:00"
-                    .replace('^', "2022-2-2T"),
+                    .replace('^', "2022-02-02T"),
             )
         );
 
@@ -432,7 +432,7 @@ mod tests {
                   2 ^10:30:00 ^11:00:00 ^10:42:00
                   1 ^10:45:00 ^11:15:00 ^10:51:00
                   3 ^11:00:00 ^11:30:00 ^11:02:00"
-                    .replace('^', "2022-2-2T"),
+                    .replace('^', "2022-02-02T"),
             )
         );
     }

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

@@ -692,7 +692,7 @@ mod tests {
              2 5.5 2 5.5
              2 8.4 2 5.5",
         );
-        let condition = build_from_pretty("(less_than:boolean (cast:float4 5:int4) $3:float4)");
+        let condition = build_from_pretty("(less_than:boolean 5:float4 $3:float4)");
 
         do_test(JoinType::Inner, Some(condition), false, expected).await;
     }
@@ -709,7 +709,7 @@ mod tests {
              5 9.1 . .
              . .   . .",
         );
-        let condition = build_from_pretty("(less_than:boolean (cast:float4 5:int4) $3:float4)");
+        let condition = build_from_pretty("(less_than:boolean 5:float4 $3:float4)");
 
         do_test(JoinType::LeftOuter, Some(condition), false, expected).await;
     }
@@ -722,7 +722,7 @@ mod tests {
              2 5.5
              2 8.4",
         );
-        let condition = build_from_pretty("(less_than:boolean (cast:float4 5:int4) $3:float4)");
+        let condition = build_from_pretty("(less_than:boolean 5:float4 $3:float4)");
 
         do_test(JoinType::LeftSemi, Some(condition), false, expected).await;
     }
@@ -736,7 +736,7 @@ mod tests {
             5 9.1
             . .",
         );
-        let condition = build_from_pretty("(less_than:boolean (cast:float4 5:int4) $3:float4)");
+        let condition = build_from_pretty("(less_than:boolean 5:float4 $3:float4)");
 
         do_test(JoinType::LeftAnti, Some(condition), false, expected).await;
     }

src/batch/src/executor/sort_agg.rs

@@ -20,7 +20,7 @@ use risingwave_common::array::{Array, ArrayBuilderImpl, ArrayImpl, DataChunk, St
 use risingwave_common::catalog::{Field, Schema};
 use risingwave_common::error::{Result, RwError};
 use risingwave_common::util::iter_util::ZipEqFast;
-use risingwave_expr::agg::{AggCall, AggregateState, BoxedAggregateFunction};
+use risingwave_expr::aggregate::{AggCall, AggregateState, BoxedAggregateFunction};
 use risingwave_expr::expr::{build_from_prost, BoxedExpression};
 use risingwave_pb::batch_plan::plan_node::NodeBody;
 

src/batch/src/lib.rs

@@ -48,3 +48,6 @@ pub mod task;
 extern crate tracing;
 #[macro_use]
 extern crate risingwave_common;
+
+#[cfg(test)]
+risingwave_expr_impl::enable!();

src/cmd/Cargo.toml

@@ -25,6 +25,7 @@ risingwave_common = { workspace = true }
 risingwave_compactor = { workspace = true }
 risingwave_compute = { workspace = true }
 risingwave_ctl = { workspace = true }
+risingwave_expr_impl = { workspace = true }
 risingwave_frontend = { workspace = true }
 risingwave_meta = { workspace = true }
 risingwave_rt = { workspace = true }