fix: rename transform to transmute (#385)

* fix: rename transform to transmute * fix: enable test in ci
flock-lab · Dec 26, 2021 · 2f11574 · 2f11574
1 parent e612513
commit 2f11574
Show file tree

Hide file tree

Showing 30 changed files with 135 additions and 139 deletions.
diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
@@ -39,8 +39,11 @@ jobs:
         with:
           filters: |
             src:
-              - 'bench/**'
-              - 'src/**'
+              - 'benchmarks/**'
+              - 'flock/**'
+              - 'flock-cli/**'
+              - 'flock-function/**'
+              - 'playground/**'
       - name: build and tests
         if: steps.changes.outputs.src == 'true'
         run: |

diff --git a/benchmarks/nexmark/main.rs b/benchmarks/nexmark/main.rs
@@ -336,6 +336,7 @@ pub fn nexmark_query(query_number: usize) -> String {
 mod tests {
     use super::*;
     use arrow::util::pretty::pretty_format_batches;
+    use flock::transmute::event_bytes_to_batch;
 
     #[tokio::test]
     async fn nexmark_sql_queries() -> Result<()> {
@@ -370,14 +371,16 @@ mod tests {
 
             flock_ctx
                 .feed_data_sources(&[
-                    vec![NEXMarkSource::to_batch(&event.bids, NEXMARK_BID.clone())],
-                    vec![NEXMarkSource::to_batch(
+                    vec![event_bytes_to_batch(&event.bids, NEXMARK_BID.clone(), 1024)],
+                    vec![event_bytes_to_batch(
                         &event.persons,
                         NEXMARK_PERSON.clone(),
+                        1024,
                     )],
-                    vec![NEXMarkSource::to_batch(
+                    vec![event_bytes_to_batch(
                         &event.auctions,
                         NEXMARK_AUCTION.clone(),
+                        1024,
                     )],
                 ])
                 .await?;

diff --git a/benchmarks/ysb/main.rs b/benchmarks/ysb/main.rs
@@ -214,6 +214,7 @@ fn ysb_query() -> String {
 mod tests {
     use super::*;
     use arrow::util::pretty::pretty_format_batches;
+    use flock::transmute::event_bytes_to_batch;
 
     #[tokio::test]
     async fn ysb_sql_query() -> Result<()> {
@@ -239,8 +240,12 @@ mod tests {
 
         flock_ctx
             .feed_data_sources(&[
-                vec![YSBSource::to_batch(&event.ad_events, YSB_AD_EVENT.clone())],
-                vec![YSBSource::to_batch(&campaigns, YSB_CAMPAIGN.clone())],
+                vec![event_bytes_to_batch(
+                    &event.ad_events,
+                    YSB_AD_EVENT.clone(),
+                    1024,
+                )],
+                vec![event_bytes_to_batch(&campaigns, YSB_CAMPAIGN.clone(), 1024)],
             ])
             .await?;
 

diff --git a/flock/src/datasink/mod.rs b/flock/src/datasink/mod.rs
@@ -18,7 +18,7 @@ use crate::config::FLOCK_CONF;
 use crate::encoding::Encoding;
 use crate::error::{FlockError, Result};
 use crate::runtime::payload::DataFrame;
-use crate::runtime::transform::*;
+use crate::transmute::*;
 use arrow::datatypes::Schema;
 use arrow::record_batch::RecordBatch;
 use arrow_flight::utils::flight_data_from_arrow_batch;

diff --git a/flock/src/datasource/nexmark/nexmark.rs b/flock/src/datasource/nexmark/nexmark.rs
@@ -24,15 +24,13 @@ use crate::error::FlockError;
 use crate::error::Result;
 use crate::runtime::payload::{Payload, Uuid};
 use crate::runtime::query::StreamWindow;
-use crate::runtime::transform::*;
+use crate::transmute::*;
 use arrow::datatypes::SchemaRef;
-use arrow::json;
 use arrow::record_batch::RecordBatch;
 use lazy_static::lazy_static;
 use log::info;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
-use std::io::BufReader;
 use std::sync::{Arc, Mutex};
 use std::thread;
 
@@ -181,28 +179,16 @@ impl DataStream for NEXMarkStream {
         };
         let (r1, r2) = match query_number.expect("Query number is not set.") {
             0 | 1 | 2 | 5 | 7 => (
-                NEXMarkSource::to_batch_v2(&event.bids, NEXMARK_BID.clone(), granule_size * 2),
+                event_bytes_to_batch(&event.bids, NEXMARK_BID.clone(), granule_size * 2),
                 vec![],
             ),
             3 | 8 => (
-                NEXMarkSource::to_batch_v2(
-                    &event.persons,
-                    NEXMARK_PERSON.clone(),
-                    granule_size / 5,
-                ),
-                NEXMarkSource::to_batch_v2(
-                    &event.auctions,
-                    NEXMARK_AUCTION.clone(),
-                    granule_size / 5,
-                ),
+                event_bytes_to_batch(&event.persons, NEXMARK_PERSON.clone(), granule_size / 5),
+                event_bytes_to_batch(&event.auctions, NEXMARK_AUCTION.clone(), granule_size / 5),
             ),
             4 | 6 | 9 => (
-                NEXMarkSource::to_batch_v2(
-                    &event.auctions,
-                    NEXMARK_AUCTION.clone(),
-                    granule_size / 8,
-                ),
-                NEXMarkSource::to_batch_v2(&event.bids, NEXMARK_BID.clone(), granule_size * 2),
+                event_bytes_to_batch(&event.auctions, NEXMARK_AUCTION.clone(), granule_size / 8),
+                event_bytes_to_batch(&event.bids, NEXMARK_BID.clone(), granule_size * 2),
             ),
             _ => unimplemented!(),
         };
@@ -262,22 +248,23 @@ impl DataStream for NEXMarkStream {
             time, persons_num, auctions_num, bids_num
         );
 
+        let batch_size = *FLOCK_SYNC_GRANULE_SIZE;
         match query_number.expect("Query number is not set.") {
             0 | 1 | 2 | 5 | 7 => Ok(to_payload(
-                &NEXMarkSource::to_batch(&event.bids, NEXMARK_BID.clone()),
+                &event_bytes_to_batch(&event.bids, NEXMARK_BID.clone(), batch_size),
                 &[],
                 uuid,
                 sync,
             )),
             3 | 8 => Ok(to_payload(
-                &NEXMarkSource::to_batch(&event.persons, NEXMARK_PERSON.clone()),
-                &NEXMarkSource::to_batch(&event.auctions, NEXMARK_AUCTION.clone()),
+                &event_bytes_to_batch(&event.persons, NEXMARK_PERSON.clone(), batch_size),
+                &event_bytes_to_batch(&event.auctions, NEXMARK_AUCTION.clone(), batch_size),
                 uuid,
                 sync,
             )),
             4 | 6 | 9 => Ok(to_payload(
-                &NEXMarkSource::to_batch(&event.auctions, NEXMARK_AUCTION.clone()),
-                &NEXMarkSource::to_batch(&event.bids, NEXMARK_BID.clone()),
+                &event_bytes_to_batch(&event.auctions, NEXMARK_AUCTION.clone(), batch_size),
+                &event_bytes_to_batch(&event.bids, NEXMARK_BID.clone(), batch_size),
                 uuid,
                 sync,
             )),
@@ -399,28 +386,6 @@ impl NEXMarkSource {
         Ok(std::mem::take(&mut events))
     }
 
-    /// Converts NEXMarkSource events to record batches in Arrow.
-    pub fn to_batch(events: &[u8], schema: SchemaRef) -> Vec<RecordBatch> {
-        let batch_size = *FLOCK_SYNC_GRANULE_SIZE;
-        let mut reader = json::Reader::new(BufReader::new(events), schema, batch_size, None);
-
-        let mut batches = vec![];
-        while let Some(batch) = reader.next().unwrap() {
-            batches.push(batch);
-        }
-        batches
-    }
-
-    /// Converts NEXMarkSource events to record batches in Arrow.
-    pub fn to_batch_v2(events: &[u8], schema: SchemaRef, batch_size: usize) -> Vec<RecordBatch> {
-        let mut reader = json::Reader::new(BufReader::new(events), schema, batch_size, None);
-        let mut batches = vec![];
-        while let Some(batch) = reader.next().unwrap() {
-            batches.push(batch);
-        }
-        batches
-    }
-
     /// Counts the number of events. (for testing)
     pub fn count_events(&self, events: &NEXMarkStream) -> usize {
         let threads: usize = self.config.get_as_or("threads", 100);
@@ -509,18 +474,20 @@ mod test {
         // decompression and deserialization
         let de_events: NEXMarkEvent = serde_json::from_value(values).unwrap();
 
+        let batch_size = *FLOCK_SYNC_GRANULE_SIZE;
+
         let person_schema = Arc::new(Person::schema());
-        let batches = NEXMarkSource::to_batch(&de_events.persons, person_schema);
+        let batches = event_bytes_to_batch(&de_events.persons, person_schema, batch_size);
         let formatted = arrow::util::pretty::pretty_format_batches(&batches).unwrap();
         println!("{}", formatted);
 
         let auction_schema = Arc::new(Auction::schema());
-        let batches = NEXMarkSource::to_batch(&de_events.auctions, auction_schema);
+        let batches = event_bytes_to_batch(&de_events.auctions, auction_schema, batch_size);
         let formatted = arrow::util::pretty::pretty_format_batches(&batches).unwrap();
         println!("{}", formatted);
 
         let bid_schema = Arc::new(Bid::schema());
-        let batches = NEXMarkSource::to_batch(&de_events.bids, bid_schema);
+        let batches = event_bytes_to_batch(&de_events.bids, bid_schema, batch_size);
         let formatted = arrow::util::pretty::pretty_format_batches(&batches).unwrap();
         println!("{}", formatted);
 

diff --git a/flock/src/datasource/nexmark/queries/q0.rs b/flock/src/datasource/nexmark/queries/q0.rs
@@ -22,6 +22,7 @@ mod tests {
     use crate::error::Result;
     use crate::runtime::executor::plan::physical_plan;
     use crate::runtime::query::StreamWindow;
+    use crate::transmute::event_bytes_to_batch;
     use datafusion::datasource::MemTable;
     use datafusion::physical_plan::collect;
     use std::sync::Arc;
@@ -42,7 +43,7 @@ mod tests {
             // events to record batches
             let bm = events.bids.get(&Epoch::new(i)).unwrap();
             let (bids, _) = bm.get(&0).unwrap();
-            let batches = NEXMarkSource::to_batch(bids, schema.clone());
+            let batches = event_bytes_to_batch(bids, schema.clone(), 1024);
 
             // register memory table
             let mut ctx = datafusion::execution::context::ExecutionContext::new();

diff --git a/flock/src/datasource/nexmark/queries/q1.rs b/flock/src/datasource/nexmark/queries/q1.rs
@@ -22,6 +22,7 @@ mod tests {
     use crate::error::Result;
     use crate::runtime::executor::plan::physical_plan;
     use crate::runtime::query::StreamWindow;
+    use crate::transmute::event_bytes_to_batch;
     use datafusion::datasource::MemTable;
     use datafusion::physical_plan::collect;
     use std::sync::Arc;
@@ -43,7 +44,7 @@ mod tests {
             // events to record batches
             let bm = events.bids.get(&Epoch::new(i)).unwrap();
             let (bids, _) = bm.get(&0).unwrap();
-            let batches = NEXMarkSource::to_batch(bids, schema.clone());
+            let batches = event_bytes_to_batch(bids, schema.clone(), 1024);
 
             // register memory table
             let mut ctx = datafusion::execution::context::ExecutionContext::new();

diff --git a/flock/src/datasource/nexmark/queries/q10.rs b/flock/src/datasource/nexmark/queries/q10.rs
@@ -22,6 +22,7 @@ mod tests {
     use crate::error::Result;
     use crate::runtime::executor::plan::physical_plan;
     use crate::runtime::query::StreamWindow;
+    use crate::transmute::event_bytes_to_batch;
     use datafusion::datasource::MemTable;
     use datafusion::physical_plan::collect;
     use indoc::indoc;
@@ -50,7 +51,7 @@ mod tests {
             // events to record batches
             let bm = events.bids.get(&Epoch::new(i)).unwrap();
             let (bids, _) = bm.get(&0).unwrap();
-            let batches = NEXMarkSource::to_batch(bids, schema.clone());
+            let batches = event_bytes_to_batch(bids, schema.clone(), 1024);
 
             // register memory table
             let mut ctx = datafusion::execution::context::ExecutionContext::new();

diff --git a/flock/src/datasource/nexmark/queries/q11.rs b/flock/src/datasource/nexmark/queries/q11.rs
@@ -22,7 +22,7 @@ mod tests {
     use crate::error::{FlockError, Result};
     use crate::runtime::executor::plan::physical_plan;
     use crate::runtime::query::{Schedule, StreamWindow};
-    use crate::runtime::transform::*;
+    use crate::transmute::*;
     use arrow::array::{Int32Array, TimestampMillisecondArray, UInt64Array};
     use arrow::record_batch::RecordBatch;
     use arrow::util::pretty::pretty_format_batches;
@@ -74,7 +74,7 @@ mod tests {
         for i in 0..seconds {
             let bm = events.bids.get(&Epoch::new(i)).unwrap();
             let (bids, _) = bm.get(&0).unwrap();
-            let mut batches = vec![NEXMarkSource::to_batch(bids, schema.clone())];
+            let mut batches = vec![event_bytes_to_batch(bids, schema.clone(), 1024)];
 
             // register memory tables
             let mut ctx = datafusion::execution::context::ExecutionContext::new();

diff --git a/flock/src/datasource/nexmark/queries/q13.rs b/flock/src/datasource/nexmark/queries/q13.rs
@@ -22,6 +22,7 @@ mod tests {
     use crate::error::Result;
     use crate::runtime::executor::plan::physical_plan;
     use crate::runtime::query::StreamWindow;
+    use crate::transmute::event_bytes_to_batch;
     use arrow::datatypes::{DataType, Field, Schema};
     use arrow::util::pretty::pretty_format_batches;
     use datafusion::datasource::MemTable;
@@ -105,7 +106,7 @@ mod tests {
             // events to record batches
             let bm = events.bids.get(&Epoch::new(i)).unwrap();
             let (bids, _) = bm.get(&0).unwrap();
-            let bids_batches = NEXMarkSource::to_batch(bids, bid_schema.clone());
+            let bids_batches = event_bytes_to_batch(bids, bid_schema.clone(), 1024);
 
             // register memory tables
             let bid_table = MemTable::try_new(bid_schema.clone(), vec![bids_batches])?;

diff --git a/flock/src/datasource/nexmark/queries/q2.rs b/flock/src/datasource/nexmark/queries/q2.rs
@@ -22,6 +22,7 @@ mod tests {
     use crate::error::Result;
     use crate::runtime::executor::plan::physical_plan;
     use crate::runtime::query::StreamWindow;
+    use crate::transmute::event_bytes_to_batch;
     use datafusion::datasource::MemTable;
     use datafusion::physical_plan::collect;
     use std::sync::Arc;
@@ -43,7 +44,7 @@ mod tests {
             // events to record batches
             let bm = events.bids.get(&Epoch::new(i)).unwrap();
             let (bids, _) = bm.get(&0).unwrap();
-            let batches = NEXMarkSource::to_batch(bids, schema.clone());
+            let batches = event_bytes_to_batch(bids, schema.clone(), 1024);
 
             // register memory table
             let mut ctx = datafusion::execution::context::ExecutionContext::new();

diff --git a/flock/src/datasource/nexmark/queries/q3.rs b/flock/src/datasource/nexmark/queries/q3.rs
@@ -22,6 +22,7 @@ mod tests {
     use crate::error::Result;
     use crate::runtime::executor::plan::physical_plan;
     use crate::runtime::query::StreamWindow;
+    use crate::transmute::event_bytes_to_batch;
     use datafusion::datasource::MemTable;
     use datafusion::physical_plan::collect;
     use indoc::indoc;
@@ -65,11 +66,11 @@ mod tests {
             // events to record batches
             let am = events.auctions.get(&Epoch::new(i)).unwrap();
             let (auctions, _) = am.get(&0).unwrap();
-            let auctions_batches = NEXMarkSource::to_batch(auctions, auction_schema.clone());
+            let auctions_batches = event_bytes_to_batch(auctions, auction_schema.clone(), 1024);
 
             let pm = events.persons.get(&Epoch::new(i)).unwrap();
             let (persons, _) = pm.get(&0).unwrap();
-            let person_batches = NEXMarkSource::to_batch(persons, person_schema.clone());
+            let person_batches = event_bytes_to_batch(persons, person_schema.clone(), 1024);
 
             // register memory tables
             let mut ctx = datafusion::execution::context::ExecutionContext::new();

diff --git a/flock/src/datasource/nexmark/queries/q4.rs b/flock/src/datasource/nexmark/queries/q4.rs
@@ -22,6 +22,7 @@ mod tests {
     use crate::error::Result;
     use crate::runtime::executor::plan::physical_plan;
     use crate::runtime::query::StreamWindow;
+    use crate::transmute::event_bytes_to_batch;
     use datafusion::datasource::MemTable;
     use datafusion::physical_plan::collect;
     use indoc::indoc;
@@ -65,11 +66,11 @@ mod tests {
             // events to record batches
             let am = events.auctions.get(&Epoch::new(i)).unwrap();
             let (auctions, _) = am.get(&0).unwrap();
-            let auctions_batches = NEXMarkSource::to_batch(auctions, auction_schema.clone());
+            let auctions_batches = event_bytes_to_batch(auctions, auction_schema.clone(), 1024);
 
             let bm = events.bids.get(&Epoch::new(i)).unwrap();
             let (bids, _) = bm.get(&0).unwrap();
-            let bids_batches = NEXMarkSource::to_batch(bids, bid_schema.clone());
+            let bids_batches = event_bytes_to_batch(bids, bid_schema.clone(), 1024);
 
             // register memory tables
             let mut ctx = datafusion::execution::context::ExecutionContext::new();

diff --git a/flock/src/datasource/nexmark/queries/q5.rs b/flock/src/datasource/nexmark/queries/q5.rs
@@ -22,6 +22,7 @@ mod tests {
     use crate::error::{FlockError, Result};
     use crate::runtime::executor::plan::physical_plan;
     use crate::runtime::query::StreamWindow;
+    use crate::transmute::event_bytes_to_batch;
     use datafusion::datasource::MemTable;
     use datafusion::physical_plan::collect;
     use indoc::indoc;
@@ -90,7 +91,7 @@ mod tests {
                 }
                 let bm = events.bids.get(&Epoch::new(j)).unwrap();
                 let (bids, _) = bm.get(&0).unwrap();
-                bids_batches.push(NEXMarkSource::to_batch(bids, bid_schema.clone()));
+                bids_batches.push(event_bytes_to_batch(bids, bid_schema.clone(), 1024));
             }
 
             let old_batches = bids_batches.clone();

diff --git a/flock/src/datasource/nexmark/queries/q6.rs b/flock/src/datasource/nexmark/queries/q6.rs
@@ -22,6 +22,7 @@ mod tests {
     use crate::error::Result;
     use crate::runtime::executor::plan::physical_plan;
     use crate::runtime::query::StreamWindow;
+    use crate::transmute::event_bytes_to_batch;
     use arrow::array::UInt64Array;
     use arrow::record_batch::RecordBatch;
     use datafusion::datasource::MemTable;
@@ -95,11 +96,11 @@ mod tests {
             // events to record batches
             let am = events.auctions.get(&Epoch::new(i)).unwrap();
             let (auctions, _) = am.get(&0).unwrap();
-            let auctions_batches = NEXMarkSource::to_batch(auctions, auction_schema.clone());
+            let auctions_batches = event_bytes_to_batch(auctions, auction_schema.clone(), 1024);
 
             let bm = events.bids.get(&Epoch::new(i)).unwrap();
             let (bids, _) = bm.get(&0).unwrap();
-            let bids_batches = NEXMarkSource::to_batch(bids, bid_schema.clone());
+            let bids_batches = event_bytes_to_batch(bids, bid_schema.clone(), 1024);
 
             // register memory tables
             let mut ctx = datafusion::execution::context::ExecutionContext::new();