implement influxql stmt rewriter.

Cargo.toml

@@ -114,6 +114,7 @@ tokio = { version = "1.25", features = ["full"] }
 wal = { path = "wal" }
 message_queue = { path = "components/message_queue" }
 zstd = { version = "0.12", default-features = false }
+itertools = "0.10.5"
 
 [workspace.dependencies.ceresdbproto]
 git = "https://github.com/CeresDB/ceresdbproto.git"

sql/Cargo.toml

@@ -34,6 +34,7 @@ sqlparser = { workspace = true }
 table_engine = { workspace = true }
 influxdb_influxql_parser = {git = "https://github.com/Rachelint/influxdb_iox.git", branch = "influxql-parser"}
 regex-syntax = "0.6.28"
+itertools = { workspace = true }
 
 [dev-dependencies]
 common_types = { workspace = true, features = ["test"] }

sql/src/influxql/mod.rs

@@ -11,8 +11,21 @@ pub mod error {
         #[snafu(display("Unimplemented influxql statement, msg: {}", msg))]
         Unimplemented { msg: String },
 
-        #[snafu(display("Unimplemented influxql statement, source: {}", source))]
-        RewriteStmtWithCause { source: GenericError },
+        #[snafu(display(
+            "Rewrite influxql from clause with cause, msg:{}, source:{}",
+            msg,
+            source
+        ))]
+        RewriteFromWithCause { msg: String, source: GenericError },
+
+        #[snafu(display("Rewrite influxql from clause no cause, msg:{}", msg))]
+        RewriteFromNoCause { msg: String },
+
+        #[snafu(display("Unimplemented influxql select fields with cause, source: {}", source))]
+        RewriteFieldsWithCause { msg: String, source: GenericError },
+
+        #[snafu(display("Unimplemented influxql select fields no cause, msg: {}", msg))]
+        RewriteFieldsNoCause { msg: String },
     }
 
     define_result!(Error);

sql/src/influxql/planner.rs

@@ -1,5 +1,6 @@
 use influxdb_influxql_parser::statement::Statement as InfluxqlStatement;
 
+use super::stmt_rewriter::StmtRewriter;
 use crate::{influxql::error::*, plan::Plan, provider::MetaProvider};
 
 pub(crate) struct Planner<'a, P: MetaProvider> {
@@ -28,6 +29,9 @@ impl<'a, P: MetaProvider> Planner<'a, P> {
     }
 
     fn rewrite_stmt(&self, stmt: InfluxqlStatement) -> Result<InfluxqlStatement> {
+        let mut stmt = stmt;
+        let stmt_rewriter = StmtRewriter::new(&self.sql_planner);
+        // stmt_rewriter.rewrite_from(&mut stmt)?;
         todo!()
     }
 }