Chore/update deps

tesseract-clickhouse/Cargo.toml

@@ -5,15 +5,16 @@ name = "tesseract-clickhouse"
 version = "0.1.0"
 
 [dependencies]
-clickhouse-rs = "0.1.20"
-failure = "0.1.2"
-futures = "0.1.25"
-itertools = "0.8.0"
+async-trait = "0.1.42"
+clickhouse-rs = "1.0.0-alpha.1"
+anyhow = "1"
+futures = "0.3.13"
+itertools = "0.10.0"
 log = "0.4.3"
 regex = "1"
 
 [dependencies.tesseract-core]
 path = "../tesseract-core"
 
 [dev-dependencies]
-tokio = "0.1.22"
+tokio = { version = "1.0", features = ["rt", "macros"] }

tesseract-clickhouse/src/df.rs

@@ -1,6 +1,6 @@
 //! Convert clickhouse Block to tesseract_core::DataFrame
 
-use failure::{Error, bail};
+use anyhow::{Error, bail};
 
 use clickhouse_rs::types::{Block, ColumnType, Complex, Simple, SqlType};
 use tesseract_core::{DataFrame, Column, ColumnData};

tesseract-clickhouse/src/lib.rs

@@ -1,7 +1,7 @@
+use anyhow::Error;
+use async_trait::async_trait;
 use clickhouse_rs::Pool;
-use clickhouse_rs::types::{Options, Simple, Complex, Block};
-use failure::{Error, format_err};
-use futures::{future, Future, Stream};
+use clickhouse_rs::types::Options;
 use log::*;
 use std::time::{Duration, Instant};
 use tesseract_core::{Backend, DataFrame, QueryIr};
@@ -11,7 +11,7 @@ use regex::Regex;
 mod df;
 mod sql;
 
-use self::df::{block_to_df};
+use self::df::block_to_df;
 use self::sql::clickhouse_sql;
 
 // Ping timeout in millis
@@ -47,42 +47,37 @@ impl Clickhouse {
     }
 }
 
+#[async_trait]
 impl Backend for Clickhouse {
-    fn exec_sql(&self, sql: String) -> Box<dyn Future<Item=DataFrame, Error=Error>> {
+    async fn exec_sql(&self, sql: String) -> Result<DataFrame, Error> {
         let time_start = Instant::now();
 
-        let fut = self.pool
-            .get_handle()
-            .and_then(move |c| c.query(&sql[..]).fetch_all())
-            .from_err()
-            .and_then(move |(_, block): (_, Block<Complex>)| {
-                let timing = time_start.elapsed();
-                info!("Time for sql execution: {}.{:03}", timing.as_secs(), timing.subsec_millis());
-                //debug!("Block: {:?}", block);
+        let mut client = self.pool.get_handle().await?;
+        let block = client.query(&sql[..]).fetch_all().await?;
+        let timing = time_start.elapsed();
+        info!("Time for sql execution: {}.{:03}", timing.as_secs(), timing.subsec_millis());
+        //debug!("Block: {:?}", block);
 
-                Ok(block_to_df(block)?)
-            });
-
-        Box::new(fut)
+        Ok(block_to_df(block)?)
     }
 
-    fn exec_sql_stream(&self, sql: String) -> Box<dyn Stream<Item=Result<DataFrame, Error>, Error=Error>> {
-        let fut_stream = self.pool
-            .get_handle()
-            .and_then(move |c| {
-                future::ok(
-                    c.query(&sql[..])
-                        .stream_blocks()
-                        .map(move |block: Block<Simple>| {
-                            block_to_df(block)
-                        })
-                )
-            })
-            .flatten_stream()
-            .map_err(|err| format_err!("{}", err));
-
-        Box::new(fut_stream)
-    }
+    //fn exec_sql_stream(&self, sql: String) -> Box<dyn Stream<Item=Result<DataFrame, Error>, Error=Error>> {
+    //    let fut_stream = self.pool
+    //        .get_handle()
+    //        .and_then(move |c| {
+    //            future::ok(
+    //                c.query(&sql[..])
+    //                    .stream_blocks()
+    //                    .map(move |block: Block<Simple>| {
+    //                        block_to_df(block)
+    //                    })
+    //            )
+    //        })
+    //        .flatten_stream()
+    //        .map_err(|err| format_err!("{}", err));
+
+    //    Box::new(fut_stream)
+    //}
 
     // https://users.rust-lang.org/t/solved-is-it-possible-to-clone-a-boxed-trait-object/1714/4
     fn box_clone(&self) -> Box<dyn Backend + Send + Sync> {

tesseract-clickhouse/tests/integration.rs

@@ -3,43 +3,26 @@ extern crate tokio;
 
 // Basic testing structured taken from
 // clickhouse-rs database tests by suharev7.
-use tokio::prelude::*;
-use std::{
-    env
-};
-use clickhouse_rs::{
-    types::Block, ClientHandle, Pool,
-};
-use clickhouse_rs::types::Complex;
+use std::env;
+use clickhouse_rs::Pool;
 
 fn database_url() -> String {
     let tmp = env::var("TESSERACT_DATABASE_URL").unwrap_or_else(|_| "tcp://localhost:9000?compression=lz4".into());
     tmp.replace("clickhouse://", "tcp://")
 }
 
-fn run<F, T, U>(future: F) -> Result<T, U>
-where
-    F: Future<Item = T, Error = U> + Send + 'static,
-    T: Send + 'static,
-    U: Send + 'static,
-{
-    let mut runtime = tokio::runtime::Runtime::new().unwrap();
-    let result = runtime.block_on(future);
-    runtime.shutdown_on_idle().wait().unwrap();
-    result
-}
-
-#[test]
-fn test_ping() {
+#[tokio::test]
+async fn test_ping() {
     // This test is meant as a sanity check
     // to ensure the docker provisioning process worked
     let pool = Pool::new(database_url());
-    let done = pool.get_handle().and_then(ClientHandle::ping).map(|_| ());
-    run(done).unwrap()
+    println!("{:?}", pool);
+    let mut client = pool.get_handle().await.unwrap();
+    client.ping().await.unwrap();
 }
 
-#[test]
-fn test_query() {
+#[tokio::test]
+async fn test_query() {
     #[derive(Debug, Clone, PartialEq)]
     pub struct RowResult {
         pub month_name: String,
@@ -49,29 +32,18 @@ fn test_query() {
     // to ensure the SQL ingestion worked
     let pool = Pool::new(database_url());
     let sql = "SELECT month_name FROM tesseract_webshop_time;";
-    let fut = pool.get_handle()
-        .and_then(move |c| {
-            c.query(&sql).fetch_all()
-        })
-        .and_then(move |(_, block): (_, Block<Complex>)| {
-            let schema_vec: Vec<RowResult> = block.rows().map(|row| {
-                RowResult {
-                    month_name: row.get("month_name").expect("missing month_name"),
-                }
-            }).collect();
-            Ok(schema_vec)
-        });
+    let mut client = pool.get_handle().await.unwrap();
+    let block = client.query(&sql).fetch_all().await.unwrap();
 
-    let res = run(fut);
-    let status = match res {
-        Ok(val) => {
-            println!("Value={:?}", val);
-            val.len() == 12
-        }
-        Err(err) => {
-            println!("Query failed. Error={:?}", err);
-            false
-        }
-    };
-    assert!(status);
-}
+    let schema_vec: Vec<RowResult> = block.rows()
+        .map(|row| {
+            RowResult {
+                month_name: row.get("month_name").expect("missing month_name"),
+            }
+        }).collect();
+
+    assert_eq!(
+        schema_vec.len(),
+        12
+    );
+}

tesseract-core/Cargo.toml

@@ -5,10 +5,11 @@ name = "tesseract-core"
 version = "0.1.0"
 
 [dependencies]
+anyhow = "1"
+async-trait = "0.1.42"
 bytes = "0.4.12"
 csv = "1.0.2"
-failure = "0.1.3"
-futures = "0.1.25"
+futures = "0.3.13"
 itertools = "0.7.11"
 log = "0.4.3"
 serde = "1.0.71"

tesseract-core/src/backend.rs

@@ -1,22 +1,23 @@
-use failure::Error;
-use futures::{Future, Stream};
+use anyhow::Error;
+use async_trait::async_trait;
 
 use crate::dataframe::DataFrame;
 use crate::query_ir::QueryIr;
 use crate::sql;
 
 
+#[async_trait]
 pub trait Backend {
     /// Takes in a SQL string, outputs a DataFrame, which will go on to be formatted into the
     /// desired query output format.
-    fn exec_sql(&self, sql: String) -> Box<dyn Future<Item=DataFrame, Error=Error>>;
+    async fn exec_sql(&self, sql: String) -> Result<DataFrame, Error>;
 
-    /// Takes in a SQL string, outputs a stream of
-    /// DataFrames, which will go on to be formatted into the
-    /// desired query output format.
-    fn exec_sql_stream(&self, sql: String) -> Box<dyn Stream<Item=Result<DataFrame, Error>, Error=Error>> {
-        unimplemented!()
-    }
+    ///// Takes in a SQL string, outputs a stream of
+    ///// DataFrames, which will go on to be formatted into the
+    ///// desired query output format.
+    //fn exec_sql_stream(&self, sql: String) -> Box<dyn Stream<Item=Result<DataFrame, Error>>> {
+    //    unimplemented!()
+    //}
 
     fn box_clone(&self) -> Box<dyn Backend + Send + Sync>;
 

tesseract-core/src/dataframe.rs

@@ -1,4 +1,4 @@
-use failure::{Error, format_err};
+use anyhow::{Error, format_err};
 
 
 #[derive(Debug)]

tesseract-core/src/format.rs

@@ -1,5 +1,5 @@
 use csv;
-use failure::{Error, format_err};
+use anyhow::{Error, format_err};
 use indexmap::IndexMap;
 use serde::Serializer;
 use serde::ser::{SerializeSeq};

tesseract-core/src/format_stream.rs

@@ -1,3 +1,5 @@
+/// Currently dead code, until module gets switched back on.
+
 use bytes::Bytes;
 use csv;
 use failure::{Error, format_err};

tesseract-core/src/lib.rs

@@ -2,13 +2,16 @@ mod backend;
 mod dataframe;
 mod sql;
 pub mod format;
-pub mod format_stream;
+
+// fix Stream later; it's kind of a pain for now.
+//pub mod format_stream;
+
 pub mod names;
 pub mod schema;
 pub mod query;
 pub mod query_ir;
 
-use failure::{Error, format_err, bail};
+use anyhow::{Error, format_err, bail};
 use log::*;
 use serde_xml_rs as serde_xml;
 use serde_xml::from_reader;

tesseract-core/src/names.rs

@@ -20,7 +20,7 @@
 // - Dimension.Level
 // etc.
 
-use failure::{Error, bail, format_err, ensure};
+use anyhow::{Error, bail, format_err, ensure};
 use serde_derive::{Deserialize, Serialize};
 use std::fmt;
 use std::str::FromStr;

tesseract-core/src/query.rs

@@ -1,6 +1,6 @@
 use itertools::join;
 
-use failure::{Error, format_err, bail};
+use anyhow::{Error, format_err, bail};
 use std::str::FromStr;
 
 use crate::names::{

tesseract-core/src/schema.rs

@@ -1,6 +1,6 @@
 use serde_derive::{Serialize, Deserialize};
 use std::convert::From;
-use failure::{Error, format_err};
+use anyhow::{Error, format_err};
 
 pub mod aggregator;
 pub mod metadata;

tesseract-server/Cargo.toml

@@ -7,11 +7,18 @@ edition = "2018"
 name = "tesseract-olap"
 version = "0.14.13"
 
+[package.metadata.deb]
+maintainer = "Walther Chen <[email protected]>"
+extended-description = "Olap engine for serving web applications"
+depends = "$auto"
+section = "Web Servers"
+priority = "optional"
+assets = [ [ "../target/release/tesseract-olap", "usr/bin/", "755",], [ "pkg/deb/tesseract-olap.service", "/etc/systemd/system/tesseract-olap.service", "644",],]
+maintainer-scripts = "pkg/deb/maintainer-scripts"
+
 [dependencies]
-actix = "0.7.7"
+anyhow = "1"
 dotenv = "0.13.0"
-failure = "0.1.2"
-futures = "0.1.25"
 lazy_static = "1.2.0"
 log = "0.4.3"
 pretty_env_logger = "0.2.4"
@@ -20,14 +27,16 @@ serde_derive = "1.0.70"
 serde_json = "1.0.33"
 serde_qs = "0.4.1"
 structopt = "0.2.13"
+thiserror = "1"
 mime = "0.3.13"
 url = "2.1.0"
-jsonwebtoken = "6"
+jsonwebtoken = "7"
 r2d2_redis = "0.13.0"
+futures = "0.3.13"
 
 [dependencies.actix-web]
-version = "0.7.18"
-features = [ "ssl",]
+version = "4.0.0-beta.3"
+features = [ "rustls",]
 
 [dependencies.tesseract-clickhouse]
 path = "../tesseract-clickhouse"
@@ -40,12 +49,3 @@ path = "../tesseract-postgres"
 
 [dependencies.tesseract-core]
 path = "../tesseract-core"
-
-[package.metadata.deb]
-maintainer = "Walther Chen <[email protected]>"
-extended-description = "Olap engine for serving web applications"
-depends = "$auto"
-section = "Web Servers"
-priority = "optional"
-assets = [ [ "../target/release/tesseract-olap", "usr/bin/", "755",], [ "pkg/deb/tesseract-olap.service", "/etc/systemd/system/tesseract-olap.service", "644",],]
-maintainer-scripts = "pkg/deb/maintainer-scripts"