22
33//! Influxql planner
44
5- use common_util:: error:: { BoxError , GenericResult } ;
6- use influxdb_influxql_parser:: {
7- select:: SelectStatement , statement:: Statement as InfluxqlStatement ,
8- } ;
5+ use std:: sync:: Arc ;
6+
7+ use common_util:: error:: BoxError ;
8+ use influxql_logical_planner:: planner:: InfluxQLToLogicalPlan ;
9+ use influxql_parser:: statement:: Statement as InfluxqlStatement ;
910use snafu:: ResultExt ;
10- use sqlparser:: ast:: Statement as SqlStatement ;
11- use table_engine:: table:: TableRef ;
1211
1312use crate :: {
14- influxql:: select:: { converter:: Converter , rewriter:: Rewriter } ,
15- plan:: Plan ,
16- planner:: { BuildInfluxqlPlan , Result } ,
17- provider:: MetaProvider ,
13+ influxql:: { error:: * , provider:: InfluxSchemaProviderImpl } ,
14+ plan:: { Plan , QueryPlan } ,
15+ provider:: { ContextProviderAdapter , MetaProvider } ,
1816} ;
1917
20- #[ allow( dead_code) ]
18+ const CERESDB_MEASUREMENT_COLUMN_NAME : & str = "ceresdb::measurement" ;
19+
2120pub ( crate ) struct Planner < ' a , P : MetaProvider > {
22- sql_planner : crate :: planner :: PlannerDelegate < ' a , P > ,
21+ context_provider : ContextProviderAdapter < ' a , P > ,
2322}
2423
2524impl < ' a , P : MetaProvider > Planner < ' a , P > {
26- pub fn new ( sql_planner : crate :: planner :: PlannerDelegate < ' a , P > ) -> Self {
27- Self { sql_planner }
25+ pub fn new ( context_provider : ContextProviderAdapter < ' a , P > ) -> Self {
26+ Self { context_provider }
2827 }
2928
3029 /// Build sql logical plan from [InfluxqlStatement].
@@ -33,51 +32,48 @@ impl<'a, P: MetaProvider> Planner<'a, P> {
3332/// the [InfluxqlStatement] will be converted to [SqlStatement] first,
3433/// and build plan then.
3534pub fn statement_to_plan ( self , stmt : InfluxqlStatement ) -> Result < Plan > {
36- match stmt {
37- InfluxqlStatement :: Select ( stmt) => self . select_to_plan ( * stmt) ,
38- InfluxqlStatement :: CreateDatabase ( _) => todo ! ( ) ,
39- InfluxqlStatement :: ShowDatabases ( _) => todo ! ( ) ,
40- InfluxqlStatement :: ShowRetentionPolicies ( _) => todo ! ( ) ,
41- InfluxqlStatement :: ShowTagKeys ( _) => todo ! ( ) ,
42- InfluxqlStatement :: ShowTagValues ( _) => todo ! ( ) ,
43- InfluxqlStatement :: ShowFieldKeys ( _) => todo ! ( ) ,
44- InfluxqlStatement :: ShowMeasurements ( _) => todo ! ( ) ,
45- InfluxqlStatement :: Delete ( _) => todo ! ( ) ,
46- InfluxqlStatement :: DropMeasurement ( _) => todo ! ( ) ,
47- InfluxqlStatement :: Explain ( _) => todo ! ( ) ,
35+ match & stmt {
36+ InfluxqlStatement :: Select ( _) => self . select_to_plan ( stmt) ,
37+ InfluxqlStatement :: CreateDatabase ( _)
38+ | InfluxqlStatement :: ShowDatabases ( _)
39+ | InfluxqlStatement :: ShowRetentionPolicies ( _)
40+ | InfluxqlStatement :: ShowTagKeys ( _)
41+ | InfluxqlStatement :: ShowTagValues ( _)
42+ | InfluxqlStatement :: ShowFieldKeys ( _)
43+ | InfluxqlStatement :: ShowMeasurements ( _)
44+ | InfluxqlStatement :: Delete ( _)
45+ | InfluxqlStatement :: DropMeasurement ( _)
46+ | InfluxqlStatement :: Explain ( _) => Unimplemented {
47+ stmt : stmt. to_string ( ) ,
48+ }
49+ . fail ( ) ,
4850 }
4951 }
5052
51- pub fn select_to_plan ( self , stmt : SelectStatement ) -> Result < Plan > {
52- let mut stmt = stmt;
53- let provider_impl = MeasurementProviderImpl ( & self . sql_planner ) ;
54- let rewriter = Rewriter :: new ( & provider_impl) ;
55- rewriter
56- . rewrite ( & mut stmt)
57- . box_err ( )
58- . context ( BuildInfluxqlPlan ) ?;
59-
60- let sql_stmt = SqlStatement :: Query ( Box :: new (
61- Converter :: convert ( stmt)
62- . box_err ( )
63- . context ( BuildInfluxqlPlan ) ?,
64- ) ) ;
53+ pub fn select_to_plan ( self , stmt : InfluxqlStatement ) -> Result < Plan > {
54+ let influx_schema_provider = InfluxSchemaProviderImpl {
55+ context_provider : & self . context_provider ,
56+ } ;
57+ let influxql_logical_planner = InfluxQLToLogicalPlan :: new (
58+ & influx_schema_provider,
59+ CERESDB_MEASUREMENT_COLUMN_NAME . to_string ( ) ,
60+ ) ;
6561
66- self . sql_planner
67- . sql_statement_to_plan ( sql_stmt )
62+ let df_plan = influxql_logical_planner
63+ . statement_to_plan ( stmt )
6864 . box_err ( )
69- . context ( BuildInfluxqlPlan )
70- }
71- }
72-
73- pub trait MeasurementProvider {
74- fn measurement ( & self , measurement_name : & str ) -> GenericResult < Option < TableRef > > ;
75- }
76-
77- struct MeasurementProviderImpl < ' a , P : MetaProvider > ( & ' a crate :: planner:: PlannerDelegate < ' a , P > ) ;
65+ . context ( BuildPlan {
66+ msg : "build df plan for influxql select statement" ,
67+ } ) ?;
68+ let tables = Arc :: new (
69+ self . context_provider
70+ . try_into_container ( )
71+ . box_err ( )
72+ . context ( BuildPlan {
73+ msg : "get tables from df plan of select" ,
74+ } ) ?,
75+ ) ;
7876
79- impl < ' a , P : MetaProvider > MeasurementProvider for MeasurementProviderImpl < ' a , P > {
80- fn measurement ( & self , measurement_name : & str ) -> GenericResult < Option < TableRef > > {
81- self . 0 . find_table ( measurement_name) . box_err ( )
77+ Ok ( Plan :: Query ( QueryPlan { df_plan, tables } ) )
8278 }
8379}
0 commit comments