feat: support integration tests for influxql

Cargo.toml

@@ -97,6 +97,7 @@ prost = "0.11"
 query_engine = { path = "query_engine" }
 rand = "0.7"
 remote_engine_client = { path = "remote_engine_client" }
+reqwest = { version = "0.11", default-features = false, features = ["rustls-tls", "json"] }
 router = { path = "router" }
 snafu = { version = "0.6.10", features = ["backtraces"] }
 serde = "1.0"

integration_tests/Cargo.toml

@@ -9,6 +9,8 @@ workspace = true
 anyhow = "1.0.58"
 async-trait = "0.1"
 ceresdb-client = "1.0"
+reqwest = { workspace = true }
+serde = { workspace = true }
 sql = { workspace = true }
 sqlness = "0.4.0"
 sqlparser = { workspace = true }

integration_tests/cases/env/local/influxql/basic.result

@@ -0,0 +1,49 @@
+DROP TABLE IF EXISTS `h2o_feet`;
+
+affected_rows: 0
+
+CREATE TABLE `h2o_feet` (
+    `time` timestamp NOT NULL,
+    `level_description` string TAG,
+    `location` string TAG,
+    `water_level` double NOT NULL,
+    timestamp KEY (time)) ENGINE = Analytic WITH (
+    enable_ttl = 'false'
+);
+
+affected_rows: 0
+
+INSERT INTO h2o_feet(time, level_description, location, water_level)
+    VALUES
+        (1439827200000, "between 6 and 9 feet", "coyote_creek", 8.12),
+        (1439827200000, "below 3 feet", "santa_monica", 2.064),
+        (1439827560000, "between 6 and 9 feet", "coyote_creek", 8.005),
+        (1439827560000, "below 3 feet", "santa_monica", 2.116),
+        (1439827620000, "between 6 and 9 feet", "coyote_creek", 7.887),
+        (1439827620000, "below 3 feet", "santa_monica", 2.028);
+
+affected_rows: 6
+
+SELECT * FROM "h2o_feet";
+
+tsid,time,level_description,location,water_level,
+UInt64(4483051411356144610),Timestamp(1439827200000),String("between 6 and 9 feet"),String("coyote_creek"),Double(8.12),
+UInt64(4483051411356144610),Timestamp(1439827560000),String("between 6 and 9 feet"),String("coyote_creek"),Double(8.005),
+UInt64(4483051411356144610),Timestamp(1439827620000),String("between 6 and 9 feet"),String("coyote_creek"),Double(7.887),
+UInt64(8247797837995683878),Timestamp(1439827200000),String("below 3 feet"),String("santa_monica"),Double(2.064),
+UInt64(8247797837995683878),Timestamp(1439827560000),String("below 3 feet"),String("santa_monica"),Double(2.116),
+UInt64(8247797837995683878),Timestamp(1439827620000),String("below 3 feet"),String("santa_monica"),Double(2.028),
+
+
+SELECT level_description, location, water_level FROM "h2o_feet" where location = 'santa_monica';
+
+level_description,location,water_level,
+String("below 3 feet"),String("santa_monica"),Double(2.064),
+String("below 3 feet"),String("santa_monica"),Double(2.116),
+String("below 3 feet"),String("santa_monica"),Double(2.028),
+
+
+DROP TABLE IF EXISTS `h2o_feet`;
+
+affected_rows: 0
+

integration_tests/cases/env/local/influxql/basic.sql

@@ -0,0 +1,35 @@
+DROP TABLE IF EXISTS `h2o_feet`;
+
+CREATE TABLE `h2o_feet` (
+    `time` timestamp NOT NULL,
+    `level_description` string TAG,
+    `location` string TAG,
+    `water_level` double NOT NULL,
+    timestamp KEY (time)) ENGINE = Analytic WITH (
+    enable_ttl = 'false'
+);
+
+-- Insert Records:
+-- ("2015-08-18T00:00:00Z", "between 6 and 9 feet", "coyote_creek", 8.12),
+-- ("2015-08-18T00:00:00Z", "below 3 feet", "santa_monica", 2.064),
+-- ("2015-08-18T00:06:00Z", "between 6 and 9 feet", "coyote_creek", 8.005),
+-- ("2015-08-18T00:06:00Z", "below 3 feet", "santa_monica", 2.116),
+-- ("2015-08-18T00:12:00Z", "between 6 and 9 feet", "coyote_creek", 7.887),
+-- ("2015-08-18T00:12:00Z", "below 3 feet", "santa_monica", 2.028);
+INSERT INTO h2o_feet(time, level_description, location, water_level)
+    VALUES
+        (1439827200000, "between 6 and 9 feet", "coyote_creek", 8.12),
+        (1439827200000, "below 3 feet", "santa_monica", 2.064),
+        (1439827560000, "between 6 and 9 feet", "coyote_creek", 8.005),
+        (1439827560000, "below 3 feet", "santa_monica", 2.116),
+        (1439827620000, "between 6 and 9 feet", "coyote_creek", 7.887),
+        (1439827620000, "below 3 feet", "santa_monica", 2.028);
+
+
+-- protocol=influxql
+SELECT * FROM "h2o_feet";
+
+-- protocol=influxql
+SELECT level_description, location, water_level FROM "h2o_feet" where location = 'santa_monica';
+
+DROP TABLE IF EXISTS `h2o_feet`;

integration_tests/src/database.rs

@@ -2,6 +2,7 @@
 
 use std::{
     borrow::Cow,
+    collections::HashMap,
     env,
     fmt::Display,
     fs::File,
@@ -16,6 +17,8 @@ use ceresdb_client::{
     model::sql_query::{display::CsvFormatter, Request},
     RpcContext,
 };
+use reqwest::ClientBuilder;
+use serde::Serialize;
 use sql::{
     ast::{Statement, TableName},
     parser::Parser,
@@ -35,15 +38,82 @@ pub enum DeployMode {
     Cluster,
 }
 
+// Used to access CeresDB by http service.
+#[derive(Clone)]
+struct HttpClient {
+    client: reqwest::Client,
+    endpoint: String,
+}
+
+#[derive(Clone, Serialize)]
+struct InfluxQLRequest {
+    query: String,
+}
+
+impl HttpClient {
+    fn new(endpoint: String) -> Self {
+        let client = ClientBuilder::new()
+            .build()
+            .expect("should succeed to build http client");
+        Self { client, endpoint }
+    }
+}
+
 pub struct CeresDB {
     server_process: Option<Child>,
     db_client: Arc<dyn DbClient>,
+    // FIXME: Currently, the new protocol does not support by the dbclient but is exposed by http
+    // service. And remove this client when the new protocol is supported by the dbclient.
+    http_client: Option<HttpClient>,
+}
+
+#[derive(Debug, Clone, Copy)]
+enum Protocol {
+    Sql,
+    InfluxQL,
+}
+
+impl TryFrom<&str> for Protocol {
+    type Error = String;
+
+    fn try_from(s: &str) -> Result<Self, Self::Error> {
+        let protocol = match s {
+            "influxql" => Protocol::InfluxQL,
+            "sql" => Protocol::Sql,
+            _ => return Err(format!("unknown protocol:{s}")),
+        };
+
+        Ok(protocol)
+    }
+}
+
+struct ProtocolParser;
+
+impl ProtocolParser {
+    fn parse_from_ctx(&self, ctx: &HashMap<String, String>) -> Result<Protocol, String> {
+        ctx.get("protocol")
+            .map(|s| Protocol::try_from(s.as_str()))
+            .unwrap_or(Ok(Protocol::Sql))
+    }
 }
 
 #[async_trait]
 impl Database for CeresDB {
-    async fn query(&self, _context: QueryContext, query: String) -> Box<dyn Display> {
-        Self::execute(query, self.db_client.clone()).await
+    async fn query(&self, context: QueryContext, query: String) -> Box<dyn Display> {
+        let protocol = ProtocolParser
+            .parse_from_ctx(&context.context)
+            .expect("parse protocol");
+
+        match protocol {
+            Protocol::Sql => Self::execute_sql(query, self.db_client.clone()).await,
+            Protocol::InfluxQL => {
+                let http_client = self
+                    .http_client
+                    .clone()
+                    .expect("http client is not initialized for execute influxql");
+                Self::execute_influxql(query, http_client).await
+            }
+        }
     }
 }
 
@@ -66,10 +136,11 @@ impl CeresDB {
                 let endpoint = env::var(SERVER_ENDPOINT_ENV).unwrap_or_else(|_| {
                     panic!("Cannot read server endpoint from env {SERVER_ENDPOINT_ENV:?}")
                 });
-                let db_client = Builder::new(endpoint, Mode::Proxy).build();
+                let db_client = Builder::new(endpoint.clone(), Mode::Proxy).build();
                 CeresDB {
                     server_process: Some(server_process),
                     db_client,
+                    http_client: Some(HttpClient::new(endpoint)),
                 }
             }
             DeployMode::Cluster => {
@@ -83,6 +154,7 @@ impl CeresDB {
                 CeresDB {
                     server_process: None,
                     db_client,
+                    http_client: None,
                 }
             }
         }
@@ -101,7 +173,24 @@ impl CeresDB {
         }
     }
 
-    async fn execute(query: String, client: Arc<dyn DbClient>) -> Box<dyn Display> {
+    async fn execute_influxql(query: String, http_client: HttpClient) -> Box<dyn Display> {
+        let url = format!("http://{}/influxql", http_client.endpoint);
+        let query_request = InfluxQLRequest { query };
+        let resp = http_client
+            .client
+            .post(url)
+            .json(&query_request)
+            .send()
+            .await
+            .unwrap();
+        let query_res = match resp.text().await {
+            Ok(text) => text,
+            Err(e) => format!("Failed to do influxql query, err:{e:?}"),
+        };
+        Box::new(query_res)
+    }
+
+    async fn execute_sql(query: String, client: Arc<dyn DbClient>) -> Box<dyn Display> {
         let query_ctx = RpcContext {
             database: Some("public".to_string()),
             timeout: None,

meta_client/Cargo.toml

@@ -19,6 +19,7 @@ futures = { workspace = true }
 log = { workspace = true }
 prost = { workspace = true }
 rand = { workspace = true }
+reqwest = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 snafu = { workspace = true }