address CR.

Author: Rachelint

server/src/handlers/prom.rs

@@ -25,7 +25,7 @@ use query_engine::executor::{Executor as QueryExecutor, RecordBatchVec};
 use snafu::{ensure, Backtrace, OptionExt, ResultExt, Snafu};
 use warp::reject;
 
-use super::query::{QueryRequest, QueryType};
+use super::query::QueryRequest;
 use crate::{
     context::RequestContext, handlers, instance::InstanceRef,
     schema_config_provider::SchemaConfigProviderRef,
@@ -292,10 +292,7 @@ impl<Q: QueryExecutor + 'static> RemoteStorage for CeresDBStorage<Q> {
             TIMESTAMP_COLUMN
         );
 
-        let request = QueryRequest {
-            query_type: QueryType::Sql,
-            request: sql.into(),
-        };
+        let request = QueryRequest::Sql(sql.into());
         let result = handlers::query::handle_query(ctx, self.instance.clone(), request)
             .await
             .map_err(Box::new)

server/src/handlers/query.rs

@@ -107,15 +107,18 @@ impl From<Bytes> for Request {
 }
 
 #[derive(Debug)]
-pub enum QueryType {
-    Sql,
-    Influxql,
+pub enum QueryRequest {
+    Sql(Request),
+    // TODO: influxql include more parameters, we should add it in later.
+    Influxql(Request),
 }
-
-#[derive(Debug)]
-pub struct QueryRequest {
-    pub query_type: QueryType,
-    pub request: Request,
+impl QueryRequest {
+    fn query(&self) -> &str {
+        match self {
+            QueryRequest::Sql(request) => request.query.as_str(),
+            QueryRequest::Influxql(request) => request.query.as_str(),
+        }
+    }
 }
 
 pub async fn handle_query<Q: QueryExecutor + 'static>(
@@ -143,12 +146,8 @@ pub async fn handle_query<Q: QueryExecutor + 'static>(
     let frontend = Frontend::new(provider);
     let mut sql_ctx = SqlContext::new(request_id, deadline);
 
-    let QueryRequest {
-        request,
-        query_type,
-    } = query_request;
-    let plan = match query_type {
-        QueryType::Sql => {
+    let plan = match &query_request {
+        QueryRequest::Sql(request) => {
             // Parse sql, frontend error of invalid sql already contains sql
             // TODO(yingwen): Maybe move sql from frontend error to outer error
             let mut stmts = frontend
@@ -165,7 +164,7 @@ pub async fn handle_query<Q: QueryExecutor + 'static>(
                 stmts.len() == 1,
                 TooMuchStmt {
                     len: stmts.len(),
-                    query: request.query,
+                    query: &request.query,
                 }
             );
 
@@ -178,7 +177,7 @@ pub async fn handle_query<Q: QueryExecutor + 'static>(
                 })?
         }
 
-        QueryType::Influxql => {
+        QueryRequest::Influxql(request) => {
             let mut stmts = frontend
                 .parse_influxql(&mut sql_ctx, &request.query)
                 .context(ParseInfluxql)?;
@@ -191,7 +190,7 @@ pub async fn handle_query<Q: QueryExecutor + 'static>(
                 stmts.len() == 1,
                 TooMuchStmt {
                     len: stmts.len(),
-                    query: request.query,
+                    query: &request.query,
                 }
             );
 
@@ -204,7 +203,7 @@ pub async fn handle_query<Q: QueryExecutor + 'static>(
     };
 
     instance.limiter.try_limit(&plan).context(QueryBlock {
-        query: &request.query,
+        query: query_request.query(),
     })?;
 
     // Execute in interpreter
@@ -223,7 +222,7 @@ pub async fn handle_query<Q: QueryExecutor + 'static>(
         interpreter_factory
             .create(interpreter_ctx, plan)
             .context(InterpreterExec {
-                query: &request.query,
+                query: query_request.query(),
             })?;
 
     let output = if let Some(deadline) = deadline {
@@ -233,24 +232,24 @@ pub async fn handle_query<Q: QueryExecutor + 'static>(
         )
         .await
         .context(QueryTimeout {
-            query: &request.query,
+            query: query_request.query(),
         })
         .and_then(|v| {
             v.context(InterpreterExec {
-                query: &request.query,
+                query: query_request.query(),
             })
         })?
     } else {
         interpreter.execute().await.context(InterpreterExec {
-            query: &request.query,
+            query: query_request.query(),
         })?
     };
 
     info!(
         "Query handler finished, request_id:{}, cost:{}ms, request:{:?}",
         request_id,
         begin_instant.saturating_elapsed().as_millis(),
-        request
+        query_request
     );
 
     Ok(output)

server/src/http.rs

@@ -31,11 +31,7 @@ use crate::{
     consts,
     context::RequestContext,
     error_util,
-    handlers::{
-        self,
-        prom::CeresDBStorage,
-        query::{QueryType, Request},
-    },
+    handlers::{self, prom::CeresDBStorage, query::Request},
     instance::InstanceRef,
     metrics,
     schema_config_provider::SchemaConfigProviderRef,
@@ -187,10 +183,7 @@ impl<Q: QueryExecutor + 'static> Service<Q> {
             .and(self.with_context())
             .and(self.with_instance())
             .and_then(|req, ctx, instance| async move {
-                let req = QueryRequest {
-                    query_type: QueryType::Sql,
-                    request: req,
-                };
+                let req = QueryRequest::Sql(req);
                 let result = handlers::query::handle_query(&ctx, instance, req)
                     .await
                     .map(handlers::query::convert_output)
@@ -207,7 +200,9 @@ impl<Q: QueryExecutor + 'static> Service<Q> {
             })
     }
 
-    // POST /sql
+    // POST /influxql
+    // this request type is not what influxdb API expected, the one in influxdb:
+    // https://docs.influxdata.com/influxdb/v1.8/tools/api/#query-http-endpoint
     fn influxql(
         &self,
     ) -> impl Filter<Extract = (impl warp::Reply,), Error = warp::Rejection> + Clone {
@@ -223,12 +218,11 @@ impl<Q: QueryExecutor + 'static> Service<Q> {
             .and(self.with_context())
             .and(self.with_instance())
             .and_then(|req, ctx, instance| async move {
-                let req = QueryRequest {
-                    query_type: QueryType::Influxql,
-                    request: req,
-                };
+                let req = QueryRequest::Influxql(req);
                 let result = handlers::query::handle_query(&ctx, instance, req)
                     .await
+                    // TODO: the sql's `convert_output` function may be not suitable to influxql.
+                    // We should implement influxql's related function in later.
                     .map(handlers::query::convert_output)
                     .map_err(|e| {
                         // TODO(yingwen): Maybe truncate and print the sql

server/src/mysql/worker.rs

@@ -13,7 +13,7 @@ use crate::{
     context::RequestContext,
     handlers::{
         self,
-        query::{QueryRequest, QueryType, Request},
+        query::{QueryRequest, Request},
     },
     instance::Instance,
     mysql::{
@@ -113,10 +113,7 @@ where
     async fn do_query<'a>(&'a mut self, sql: &'a str) -> Result<Output> {
         let ctx = self.create_ctx()?;
         let req = Request::from(sql.to_string());
-        let req = QueryRequest {
-            query_type: QueryType::Sql,
-            request: req,
-        };
+        let req = QueryRequest::Sql(req);
         handlers::query::handle_query(&ctx, self.instance.clone(), req)
             .await
             .map_err(|e| {

sql/src/influxql/mod.rs

@@ -4,9 +4,8 @@
 
 pub mod planner;
 pub(crate) mod select;
-#[cfg(test)]
-pub mod test_util;
 pub(crate) mod util;
+
 pub mod error {
     use common_util::error::GenericError;
     use snafu::{Backtrace, Snafu};
@@ -36,7 +35,7 @@ pub mod error {
         RewriteNoCause { msg: String, backtrace: Backtrace },
 
         #[snafu(display(
-            "Failed to convert to sql statement, msg:{}..\nBacktrace:{}",
+            "Failed to convert to sql statement, msg:{}.\nBacktrace:{}",
             msg,
             backtrace
         ))]

sql/src/influxql/planner.rs

@@ -27,6 +27,11 @@ impl<'a, P: MetaProvider> Planner<'a, P> {
         Self { sql_planner }
     }
 
+    /// Build sql logical plan from [InfluxqlStatement].
+    ///
+    /// NOTICE: when building plan from influxql select statement,
+    /// the [InfluxqlStatement] will be converted to [SqlStatement] first,
+    /// and build plan then.
     pub fn statement_to_plan(self, stmt: InfluxqlStatement) -> Result<Plan> {
         match stmt {
             InfluxqlStatement::Select(stmt) => self.select_to_plan(*stmt),

sql/src/influxql/select/converter.rs

@@ -35,7 +35,7 @@ impl Converter {
         //  - offset
         //  - select body
         let limit = stmt.limit.map(|limit| {
-            let limit_n: u64 = *limit;
+            let limit_n = *limit;
             Expr::Value(Value::Number(limit_n.to_string(), false))
         });
 
@@ -75,7 +75,7 @@ impl Converter {
         ensure!(
             stmt.from.len() == 1,
             Unimplemented {
-                msg: "from multiple measurements",
+                msg: "select from multiple measurements",
             }
         );
         let measurement_name = match &stmt.from[0] {
@@ -90,7 +90,7 @@ impl Converter {
             },
             MeasurementSelection::Subquery(_) => {
                 return Unimplemented {
-                    msg: "from subquery",
+                    msg: "select from subquery",
                 }
                 .fail()
             }
@@ -252,32 +252,32 @@ fn call_to_sql_expr(scope: ExprScope, name: &str, args: &[InfluxqlExpr]) -> Resu
             })
             .collect::<Result<Vec<_>>>()?;
 
-        Ok(Expr::Function(Function {
+        return Ok(Expr::Function(Function {
             name,
             args,
             over: None,
             distinct: false,
             special: false,
-        }))
-    } else {
-        match scope {
-            ExprScope::Projection => Unimplemented {
-                msg: "aggregate and selector functions in projection list",
-            }
-            .fail(),
+        }));
+    }
 
-            ExprScope::Where => {
-                if name.eq_ignore_ascii_case("now") {
-                    Unimplemented {
-                        msg: "now() in where clause",
-                    }
-                    .fail()
-                } else {
-                    Convert {
-                        msg: format!("invalid function call in condition: {name}"),
-                    }
-                    .fail()
+    match scope {
+        ExprScope::Projection => Unimplemented {
+            msg: "aggregate and selector functions in projection list",
+        }
+        .fail(),
+
+        ExprScope::Where => {
+            if name.eq_ignore_ascii_case("now") {
+                Unimplemented {
+                    msg: "now() in where clause",
                 }
+                .fail()
+            } else {
+                Convert {
+                    msg: format!("invalid function call in condition: {name}"),
+                }
+                .fail()
             }
         }
     }
@@ -355,9 +355,7 @@ mod test {
     use sqlparser::ast::Statement as SqlStatement;
 
     use crate::{
-        ast::Statement,
-        influxql::{select::converter::Converter, test_util::parse_select},
-        parser::Parser,
+        ast::Statement, influxql::select::converter::Converter, parser::Parser, tests::parse_select,
     };
 
     #[test]

sql/src/influxql/select/mod.rs

@@ -1,6 +1,6 @@
 // Copyright 2023 CeresDB Project Authors. Licensed under Apache-2.0.
 
-//! Statement level select converting from influxql to sql
+//! Convert influxql to sql at statement level
 
 pub(crate) mod converter;
 pub(crate) mod rewriter;