feat: support rewrite basic raw query in influxql

Cargo.toml

@@ -84,6 +84,7 @@ log = "0.4"
 logger = { path = "components/logger" }
 lru = "0.7.6"
 interpreters = { path = "interpreters" }
+itertools = "0.10.5"
 meta_client = { path = "meta_client" }
 object_store = { path = "components/object_store" }
 parquet_ext = { path = "components/parquet_ext" }

server/src/grpc/storage_service/prom_query.rs

@@ -35,9 +35,12 @@ use crate::grpc::storage_service::{
 };
 
 fn is_table_not_found_error(e: &FrontendError) -> bool {
-    matches!(&e, FrontendError::CreatePlan { source }
-             if matches!(source, sql::planner::Error::BuildPromPlanError { source }
-                         if matches!(source, sql::promql::Error::TableNotFound { .. })))
+    matches!(&e, 
+        FrontendError::CreatePlan { 
+            source, 
+        .. }
+        if matches!(source, sql::planner::Error::BuildPromPlanError { source }
+                    if matches!(source, sql::promql::Error::TableNotFound { .. })))
 }
 
 pub async fn handle_query<Q>(

sql/Cargo.toml

@@ -27,9 +27,12 @@ datafusion-expr = { workspace = true }
 datafusion-proto = { workspace = true }
 df_operator = { workspace = true }
 hashbrown = { version = "0.12", features = ["raw"] }
+influxdb_influxql_parser = { git = "https://github.com/Rachelint/influxdb_iox.git", branch = "influxql-parser" }
+itertools = { workspace = true }
 log = { workspace = true }
 paste = { workspace = true }
 regex = "1"
+regex-syntax = "0.6.28"
 snafu = { workspace = true }
 sqlparser = { workspace = true }
 table_engine = { workspace = true }

sql/src/frontend.rs

@@ -7,6 +7,7 @@ use std::{sync::Arc, time::Instant};
 use ceresdbproto::{prometheus::Expr as PromExpr, storage::WriteTableRequest};
 use cluster::config::SchemaConfig;
 use common_types::request_id::RequestId;
+use influxdb_influxql_parser::statement::Statement as InfluxqlStatement;
 use snafu::{Backtrace, OptionExt, ResultExt, Snafu};
 use table_engine::table;
 
@@ -29,14 +30,24 @@ pub enum Error {
     },
 
     // TODO(yingwen): Should we store stmt here?
-    #[snafu(display("Failed to create plan, err:{}", source))]
-    CreatePlan { source: crate::planner::Error },
+    #[snafu(display("Failed to create plan, plan_type:{}, err:{}", plan_type, source))]
+    CreatePlan {
+        plan_type: String,
+        source: crate::planner::Error,
+    },
 
     #[snafu(display("Invalid prom request, err:{}", source))]
     InvalidPromRequest { source: crate::promql::Error },
 
     #[snafu(display("Expr is not found in prom request.\nBacktrace:\n{}", backtrace))]
     ExprNotFoundInPromRequest { backtrace: Backtrace },
+
+    // invalid sql is quite common, so we don't provide a backtrace now.
+    #[snafu(display("invalid influxql, influxql:{}, err:{}", influxql, parse_err))]
+    InvalidInfluxql {
+        influxql: String,
+        parse_err: influxdb_influxql_parser::common::ParseError,
+    },
 }
 
 define_result!(Error);
@@ -90,14 +101,31 @@ impl<P> Frontend<P> {
         let expr = expr.context(ExprNotFoundInPromRequest)?;
         Expr::try_from(expr).context(InvalidPromRequest)
     }
+
+    /// Parse the sql and returns the statements
+    pub fn parse_influxql(
+        &self,
+        _ctx: &mut Context,
+        influxql: &str,
+    ) -> Result<Vec<InfluxqlStatement>> {
+        match influxdb_influxql_parser::parse_statements(influxql) {
+            Ok(stmts) => Ok(stmts),
+            Err(e) => Err(Error::InvalidInfluxql {
+                influxql: influxql.to_string(),
+                parse_err: e,
+            }),
+        }
+    }
 }
 
 impl<P: MetaProvider> Frontend<P> {
     /// Create logical plan for the statement
     pub fn statement_to_plan(&self, ctx: &mut Context, stmt: Statement) -> Result<Plan> {
         let planner = Planner::new(&self.provider, ctx.request_id, ctx.read_parallelism);
 
-        planner.statement_to_plan(stmt).context(CreatePlan)
+        planner
+            .statement_to_plan(stmt)
+            .context(CreatePlan { plan_type: "sql" })
     }
 
     pub fn promql_expr_to_plan(
@@ -107,7 +135,21 @@ impl<P: MetaProvider> Frontend<P> {
     ) -> Result<(Plan, Arc<ColumnNames>)> {
         let planner = Planner::new(&self.provider, ctx.request_id, ctx.read_parallelism);
 
-        planner.promql_expr_to_plan(expr).context(CreatePlan)
+        planner.promql_expr_to_plan(expr).context(CreatePlan {
+            plan_type: "promql",
+        })
+    }
+
+    pub fn influxql_stmt_to_plan(
+        &self,
+        ctx: &mut Context,
+        stmt: InfluxqlStatement,
+    ) -> Result<Plan> {
+        let planner = Planner::new(&self.provider, ctx.request_id, ctx.read_parallelism);
+
+        planner.influxql_stmt_to_plan(stmt).context(CreatePlan {
+            plan_type: "influxql",
+        })
     }
 
     pub fn write_req_to_plan(
@@ -120,6 +162,8 @@ impl<P: MetaProvider> Frontend<P> {
 
         planner
             .write_req_to_plan(schema_config, write_table)
-            .context(CreatePlan)
+            .context(CreatePlan {
+                plan_type: "internal write",
+            })
     }
 }

sql/src/influxql/mod.rs

@@ -0,0 +1,47 @@
+// Copyright 2023 CeresDB Project Authors. Licensed under Apache-2.0.
+
+pub mod planner;
+mod stmt_rewriter;
+pub(crate) mod util;
+pub mod error {
+    use common_util::error::GenericError;
+    use snafu::Snafu;
+
+    #[derive(Debug, Snafu)]
+    #[snafu(visibility = "pub")]
+    pub enum Error {
+        #[snafu(display("Unimplemented influxql statement, msg: {}", msg))]
+        Unimplemented { msg: String },
+
+        #[snafu(display(
+            "Failed to rewrite influxql from statement with cause, msg:{}, source:{}",
+            msg,
+            source
+        ))]
+        RewriteFromWithCause { msg: String, source: GenericError },
+
+        #[snafu(display("Failed to rewrite influxql from statement no cause, msg:{}", msg))]
+        RewriteFromNoCause { msg: String },
+
+        #[snafu(display(
+            "Failed to rewrite influxql projection statement with cause, msg:{}, source: {}",
+            msg,
+            source
+        ))]
+        RewriteFieldsWithCause { msg: String, source: GenericError },
+
+        #[snafu(display(
+            "Failed to rewrite influxql projection statement no cause, msg: {}",
+            msg
+        ))]
+        RewriteFieldsNoCause { msg: String },
+
+        #[snafu(display("Failed to find table with case, source:{}", source))]
+        FindTableWithCause { source: GenericError },
+
+        #[snafu(display("Failed to find table no case, msg: {}", msg))]
+        FindTableNoCause { msg: String },
+    }
+
+    define_result!(Error);
+}

sql/src/influxql/planner.rs

@@ -0,0 +1,52 @@
+// Copyright 2023 CeresDB Project Authors. Licensed under Apache-2.0.
+
+use common_util::error::BoxError;
+use influxdb_influxql_parser::statement::Statement as InfluxqlStatement;
+use snafu::ResultExt;
+use table_engine::table::TableRef;
+
+use crate::{influxql::error::*, plan::Plan, provider::MetaProvider};
+
+#[allow(dead_code)]
+pub(crate) struct Planner<'a, P: MetaProvider> {
+    sql_planner: crate::planner::PlannerDelegate<'a, P>,
+}
+
+impl<'a, P: MetaProvider> Planner<'a, P> {
+    pub fn new(sql_planner: crate::planner::PlannerDelegate<'a, P>) -> Self {
+        Self { sql_planner }
+    }
+
+    pub fn statement_to_plan(&self, stmt: InfluxqlStatement) -> Result<Plan> {
+        match stmt {
+            InfluxqlStatement::Select(_) => todo!(),
+            InfluxqlStatement::CreateDatabase(_) => todo!(),
+            InfluxqlStatement::ShowDatabases(_) => todo!(),
+            InfluxqlStatement::ShowRetentionPolicies(_) => todo!(),
+            InfluxqlStatement::ShowTagKeys(_) => todo!(),
+            InfluxqlStatement::ShowTagValues(_) => todo!(),
+            InfluxqlStatement::ShowFieldKeys(_) => todo!(),
+            InfluxqlStatement::ShowMeasurements(_) => todo!(),
+            InfluxqlStatement::Delete(_) => todo!(),
+            InfluxqlStatement::DropMeasurement(_) => todo!(),
+            InfluxqlStatement::Explain(_) => todo!(),
+        }
+    }
+}
+
+pub trait MeasurementProvider {
+    fn measurement(&self, measurement_name: &str) -> Result<Option<TableRef>>;
+}
+
+pub(crate) struct MeasurementProviderImpl<'a, P: MetaProvider>(
+    crate::planner::PlannerDelegate<'a, P>,
+);
+
+impl<'a, P: MetaProvider> MeasurementProvider for MeasurementProviderImpl<'a, P> {
+    fn measurement(&self, measurement_name: &str) -> Result<Option<TableRef>> {
+        self.0
+            .find_table(measurement_name)
+            .box_err()
+            .context(FindTableWithCause)
+    }
+}