feat: add snmalloc flags as options to SCQ

Cargo.toml

@@ -1,7 +1,8 @@
 [workspace]
 members = [
     "examples",
-    "src/bin",
+    "src/bin/cli",
+    "src/bin/func",
     "src/driver",
     "src/lambda",
 ]

examples/lambda/partitioned_plan.rs

@@ -402,7 +402,7 @@ mod tests {
         assert!(subplan.contains(r#"execution_plan":"memory_exec"#));
     }
 
-    fn quick_init<'a>(sql: &'a str) -> LambdaDag {
+    fn quick_init(sql: &str) -> LambdaDag {
         let schema = Arc::new(Schema::new(vec![
             Field::new("c1", DataType::Int64, false),
             Field::new("c2", DataType::Float64, false),
@@ -442,7 +442,7 @@ mod tests {
 
         let mut ctx = ExecutionContext::new();
         // batch? Only support 1 RecordBatch now.
-        let provider = MemTable::try_new(schema.clone(), vec![vec![batch.clone()]]).unwrap();
+        let provider = MemTable::try_new(schema, vec![vec![batch]]).unwrap();
         ctx.register_table("test_table", Box::new(provider));
 
         let logical_plan = ctx.create_logical_plan(sql).unwrap();

src/bin/Cargo.toml → src/bin/cli/Cargo.toml

@@ -20,7 +20,7 @@ ctrlc = "3.1.1"
 
 lazy_static = "1.4.0"
 rustyline = { version = "7.0", optional = true }
-scq_lambda = { path = "../lambda" }
+scq_lambda = { path = "../../lambda" }
 sqlparser = { version = "0.7.0", features = [ "json_example" ] }
 tokio = { version = "0.2", features = [ "macros" ] }
 

src/bin/func/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "lambda-operators"
+version = "0.1.0"
+description = "The generic lambda functions for cloud environment execution."
+authors = [ "Gang Liao <[email protected]>" ]
+license = "Apache-2.0"
+keywords = [ "AWS", "pre-compiled", "lambda", "S3" ]
+edition = "2018"
+
+[features]
+snmalloc = [ "snmalloc-rs" ]
+
+[dependencies]
+aws_lambda_events = "0.4"
+daggy = "0.7.0"
+datafusion = { git = "https://github.com/DSLAM-UMD/arrow", branch = "scq", features = [ "simd" ] }
+lambda = { git = "https://github.com/awslabs/aws-lambda-rust-runtime/", branch = "master" }
+scq_lambda = { path = "../../lambda" }
+serde_json = "1.0"
+snmalloc-rs = { version = "0.2", optional = true, features = [ "cache-friendly" ] }
+tokio = { version = "0.2", features = [ "macros" ] }
+
+[[bin]]
+name = "scq_ops"
+path = "src/ops.rs"
+
+[[bin]]
+name = "scq_kinesis"
+path = "src/kinesis.rs"

src/driver/src/funcgen/bin/kinesis.rs → src/bin/func/src/kinesis.rs

@@ -29,6 +29,10 @@ use scq_lambda::{exec_plan, init_plan};
 
 type Error = Box<dyn std::error::Error + Sync + Send + 'static>;
 
+#[cfg(feature = "snmalloc")]
+#[global_allocator]
+static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
+
 #[tokio::main]
 async fn main() -> Result<(), Error> {
     lambda::run(handler_fn(handler)).await?;

src/driver/src/funcgen/bin/ops.rs → src/bin/func/src/ops.rs

@@ -27,6 +27,10 @@ use scq_lambda::{exec_plan, init_plan};
 
 type Error = Box<dyn std::error::Error + Sync + Send + 'static>;
 
+#[cfg(feature = "snmalloc")]
+#[global_allocator]
+static ALLOC: snmalloc_rs::SnMalloc = snmalloc_rs::SnMalloc;
+
 #[tokio::main]
 async fn main() -> Result<(), Error> {
     lambda::run(handler_fn(handler)).await?;

src/driver/Cargo.toml

@@ -29,11 +29,3 @@ url = "2.0"
 name = "scq_driver"
 path = "src/lib.rs"
 crate-type = [ "lib" ]
-
-[[bin]]
-name = "lambda_ops"
-path = "src/funcgen/bin/ops.rs"
-
-[[bin]]
-name = "lambda_kinesis"
-path = "src/funcgen/bin/kinesis.rs"

src/driver/src/funcgen/mod.rs

@@ -14,5 +14,4 @@
 
 //! Convert the physical plan into lambda functions for deployment.
 
-pub mod bin;
 pub mod dag;