multithreaded working

vayu-common/Cargo.toml

@@ -8,3 +8,5 @@ edition = "2021"
 [dependencies]
 datafusion = { version = "36.0.0", path = "../arrow-datafusion/datafusion/core"}
 arrow = {}
+crossbeam-utils = "0.8.19"
+crossbeam-skiplist = "0.1.3"

vayu-common/src/lib.rs

@@ -2,6 +2,8 @@ use arrow::record_batch::RecordBatch;
 use datafusion::common::Result;
 use datafusion::physical_plan::{ExecutionPlan, SendableRecordBatchStream};
 use std::sync::Arc;
+pub mod store;
+
 pub trait PhysicalOperator {
     fn name(&self) -> String;
 }

vayu-common/src/store.rs

@@ -0,0 +1,96 @@
+use arrow::array::RecordBatch;
+use crossbeam_skiplist::SkipMap;
+use crossbeam_utils::thread::scope;
+use datafusion::physical_plan::joins::hash_join::JoinLeftData;
+
+use core::panic;
+use std::collections::HashMap;
+use std::hash::Hash;
+use std::sync::Arc;
+pub enum Blob {
+    // RecordBatchBlob(Vec<RecordBatch>),
+    HashMapBlob(JoinLeftData),
+}
+
+impl Blob {
+    pub fn get_map(self) -> JoinLeftData {
+        match self {
+            Blob::HashMapBlob(m) => m,
+            _ => panic!("error"),
+        }
+    }
+    // pub fn get_records(self) -> Vec<RecordBatch> {
+    //     match self {
+    //         Blob::RecordBatchBlob(records) => records,
+    //         _ => panic!("error"),
+    //     }
+    // }
+    // pub fn append_records(&mut self, batches: Vec<RecordBatch>) {
+    //     match self {
+    //         Blob::RecordBatchBlob(records) => {
+    //             // TODO: check if schema is same
+    //             records.extend(batches)
+    //         }
+    //         _ => panic!("error"),
+    //     }
+    // }
+}
+
+// right now this is typedef of HashMap<i32, Blob>,
+// but we may need something else in near future
+
+pub struct Store {
+    store: HashMap<i32, Blob>,
+}
+impl Store {
+    pub fn new() -> Store {
+        Store {
+            store: HashMap::new(),
+        }
+    }
+    pub fn insert(&mut self, key: i32, value: Blob) {
+        self.store.insert(key, value);
+    }
+    // pub fn append(&mut self, key: i32, value: Vec<RecordBatch>) {
+    //     let blob = self.remove(key);
+    //     let mut blob = match blob {
+    //         Some(r) => r,
+    //         None => Blob::RecordBatchBlob(Vec::new()),
+    //     };
+    //     blob.append_records(value);
+    //     self.store.insert(key, blob);
+    // }
+    pub fn remove(&mut self, key: i32) -> Option<Blob> {
+        self.store.remove(&key)
+        // let x = self.store.remove(&key).unwrap().value();
+        // Some(x)
+    }
+}
+pub struct Store1 {
+    store: SkipMap<i32, i32>,
+}
+impl Store1 {
+    pub fn new() -> Self {
+        Store1 {
+            store: SkipMap::new(),
+        }
+    }
+    pub fn insert(&mut self, key: i32, value: i32) {
+        self.store.insert(key, value);
+    }
+    // pub fn append(&mut self, key: i32, value: Vec<RecordBatch>) {
+    //     let blob = self.remove(key);
+    //     let mut blob = match blob {
+    //         Some(r) => r,
+    //         None => Blob::RecordBatchBlob(Vec::new()),
+    //     };
+    //     blob.append_records(value);
+    //     self.store.insert(key, blob);
+    // }
+    // pub fn remove(&mut self, key: i32) -> &Option<Arc<Blob>> {
+    //     let value = self.store.remove(&key).unwrap().value();
+    //     Some(value.un)
+    //     // let x = self.store.remove(&key).unwrap().value();
+    //     // Some(x)
+    // }
+}

vayu/Cargo.toml

@@ -12,3 +12,7 @@ futures = { version = "0.3.28" }
 datafusion = { version = "36.0.0", path = "../arrow-datafusion/datafusion/core"}
 vayu-common = { version = "0.1.0", path = "../vayu-common"}
 ahash = "0.8.11"
+crossbeam-skiplist = "0.1.3"
+crossbeam-utils = "0.8.19"
+leapfrog =  {version = "0.3.0", features = ["stable_alloc"]}
+

vayu/src/lib.rs

@@ -4,18 +4,24 @@ use vayu_common::DatafusionPipelineWithData;
 use vayu_common::VayuPipeline;
 pub mod operators;
 pub mod pipeline;
+use std::sync::{Arc, Mutex};
+
 pub mod sinks;
-pub mod store;
-use crate::store::Store;
+use vayu_common::store::Store;
 pub mod df2vayu;
 pub struct VayuExecutionEngine {
+    // this is per node store
     pub store: Store,
+    // this is global store
+    pub global_store: Arc<Mutex<Store>>,
+    // Note: only one of them will survive lets see which
 }
 
 impl VayuExecutionEngine {
-    pub fn new() -> VayuExecutionEngine {
+    pub fn new(global_store: Arc<Mutex<Store>>) -> VayuExecutionEngine {
         VayuExecutionEngine {
             store: Store::new(),
+            global_store,
         }
     }
     pub fn sink(&mut self, sink: vayu_common::SchedulerSinkType, result: Vec<RecordBatch>) {
@@ -33,16 +39,25 @@ impl VayuExecutionEngine {
             // }
             vayu_common::SchedulerSinkType::BuildAndStoreHashMap(uuid, join_node) => {
                 let mut sink = sinks::HashMapSink::new(uuid, join_node);
-                let map = sink.build_map(result);
+                let hashmap = sink.build_map(result);
                 println!("BuildAndStoreHashMap storing in uuid {uuid}");
-                self.store.insert(uuid, map.unwrap());
+
+                // old store
+                // self.store.insert(uuid, hashmap.unwrap());
+                // new store
+                let mut map = self.global_store.lock().unwrap();
+                map.insert(uuid, hashmap.unwrap());
             }
         };
     }
     pub fn execute(&mut self, pipeline: DatafusionPipelineWithData) {
         let data = pipeline.data;
         let sink = pipeline.pipeline.sink;
-        let mut pipeline: VayuPipeline = df2vayu::df2vayu(pipeline.pipeline.plan, &mut self.store);
+
+        let mut store = self.global_store.lock().unwrap();
+        let mut pipeline: VayuPipeline = df2vayu::df2vayu(pipeline.pipeline.plan, &mut store);
+        drop(store);
+
         pipeline.sink = sink;
 
         self.execute_internal(pipeline, data);

vayu/src/pipeline.rs

@@ -1,22 +1 @@
-use crate::operators::filter::FilterOperator;
-use crate::operators::join::HashProbeOperator;
-use crate::operators::projection::ProjectionOperator;
-use crate::pipeline;
-use crate::store::Store;
-use arrow::array::BooleanBufferBuilder;
-use core::panic;
-use datafusion::arrow::array::RecordBatch;
-use datafusion::datasource::physical_plan::CsvExec;
-use datafusion::error::Result;
-use datafusion::execution::context::SessionContext;
-use datafusion::execution::SendableRecordBatchStream;
-use datafusion::physical_plan::coalesce_batches::CoalesceBatchesExec;
-use datafusion::physical_plan::filter::FilterExec;
-use datafusion::physical_plan::joins::hash_join::BuildSide;
-use datafusion::physical_plan::joins::hash_join::BuildSideReadyState;
-use datafusion::physical_plan::joins::HashJoinExec;
-use datafusion::physical_plan::projection::ProjectionExec;
-use datafusion::physical_plan::repartition::RepartitionExec;
-use datafusion::physical_plan::ExecutionPlan;
-use datafusion::prelude::SessionConfig;
-use std::sync::Arc;
+

vayu/src/sinks.rs

@@ -1,10 +1,10 @@
 use datafusion::physical_plan::joins::hash_join;
 use datafusion::physical_plan::ExecutionPlan;
 
-use crate::store::Blob;
 use crate::RecordBatch;
 use ahash::RandomState;
 use arrow::datatypes::Schema;
+use vayu_common::store::Blob;
 
 use datafusion::execution::memory_pool::MemoryConsumer;
 use datafusion::physical_expr::PhysicalExprRef;

vayuDB/Cargo.toml

@@ -12,3 +12,7 @@ vayu-common = { version = "0.1.0", path = "../vayu-common"}
 futures = { version = "0.3.28" }
 heapless = "0.8.0"
 crossbeam-channel = "0.5"
+crossbeam-skiplist = "0.1.3"
+crossbeam-utils = "0.8.19"
+leapfrog =  {version = "0.3.0", features = ["stable_alloc"]}
+lockfree = "0.5.1"