databend_common_ast/parser/
dynamic_table.rs

1// Copyright 2021 Datafuse Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use nom::Parser;
16use nom_rule::rule;
17
18use crate::ast::ClusterOption;
19use crate::ast::ClusterType;
20use crate::ast::CreateDynamicTableStmt;
21use crate::ast::InitializeMode;
22use crate::ast::RefreshMode;
23use crate::ast::Statement;
24use crate::ast::TargetLag;
25use crate::ast::WarehouseOptions;
26use crate::parser::Input;
27use crate::parser::common::IResult;
28use crate::parser::common::comma_separated_list1;
29use crate::parser::common::dot_separated_idents_1_to_3;
30use crate::parser::common::map_res;
31use crate::parser::common::*;
32use crate::parser::expr::expr;
33use crate::parser::expr::literal_u64;
34use crate::parser::query::query;
35use crate::parser::statement::cluster_type;
36use crate::parser::statement::create_table_source;
37use crate::parser::statement::parse_create_option;
38use crate::parser::statement::table_option;
39use crate::parser::statement::task_warehouse_option;
40use crate::parser::token::TokenKind::*;
41
42pub fn dynamic_table(i: Input) -> IResult<Statement> {
43    rule!(
44        #create_dynamic_table : "`CREATE [OR REPLACE] [TRANSIENT] DYNAMIC TABLE [ IF NOT EXISTS ] [<database>.]<table> [<source>]
45  [ CLUSTER BY <expr> ]
46  TARGET_LAG = { <num> { SECOND | MINUTE | HOUR | DAY } | DOWNSTREAM}
47  [ { WAREHOUSE = <string> } ]
48  [ REFRESH_MODE = { AUTO | FULL | INCREMENTAL } ]
49  [ INITIALIZE = { ON_CREATE | ON_SCHEDULE } ]
50  [ COMMENT = '<string_literal>' ]
51AS
52  <sql>`"
53    ).parse(i)
54}
55
56fn create_dynamic_table(i: Input) -> IResult<Statement> {
57    map_res(
58        rule! {
59            CREATE ~ ( OR ~ ^REPLACE )? ~ TRANSIENT? ~ DYNAMIC ~ TABLE ~ ( IF ~ ^NOT ~ ^EXISTS )?
60            ~ #dot_separated_idents_1_to_3
61            ~ #create_table_source?
62            ~ ( CLUSTER ~ ^BY ~ ( #cluster_type )? ~ ^"(" ~ ^#comma_separated_list1(expr) ~ ^")" )?
63            ~ #dynamic_table_options
64            ~ (#table_option)?
65            ~ (AS ~ ^#query)
66        },
67        |(
68            _,
69            opt_or_replace,
70            opt_transient,
71            _,
72            _,
73            opt_if_not_exists,
74            (catalog, database, table),
75            source,
76            opt_cluster_by,
77            (target_lag, warehouse_opts, refresh_mode_opt, initialize_opt),
78            opt_table_options,
79            (_, query),
80        )| {
81            let create_option =
82                parse_create_option(opt_or_replace.is_some(), opt_if_not_exists.is_some())?;
83            Ok(Statement::CreateDynamicTable(CreateDynamicTableStmt {
84                create_option,
85                transient: opt_transient.is_some(),
86                catalog,
87                database,
88                table,
89                source,
90                cluster_by: opt_cluster_by.map(|(_, _, typ, _, cluster_exprs, _)| ClusterOption {
91                    cluster_type: typ.unwrap_or(ClusterType::Linear),
92                    cluster_exprs,
93                }),
94                target_lag,
95                warehouse_opts,
96                refresh_mode: refresh_mode_opt.unwrap_or(RefreshMode::Auto),
97                initialize: initialize_opt.unwrap_or(InitializeMode::OnCreate),
98                table_options: opt_table_options.unwrap_or_default(),
99                as_query: Box::new(query),
100            }))
101        },
102    )(i)
103}
104
105fn dynamic_table_options(
106    i: Input,
107) -> IResult<(
108    TargetLag,
109    WarehouseOptions,
110    Option<RefreshMode>,
111    Option<InitializeMode>,
112)> {
113    let target_lag = map(
114        rule! {
115            TARGET_LAG ~ "=" ~ #target_lag
116        },
117        |(_, _, target_lag)| target_lag,
118    );
119
120    let refresh_mode = alt((
121        value(RefreshMode::Auto, rule! { AUTO }),
122        value(RefreshMode::Full, rule! { FULL }),
123        value(RefreshMode::Incremental, rule! { INCREMENTAL }),
124    ));
125    let refresh_mode_opt = map(
126        rule! {
127            (REFRESH_MODE ~ "=" ~ #refresh_mode)?
128        },
129        |v| v.map(|v| v.2),
130    );
131
132    let initialize_mode = alt((
133        value(InitializeMode::OnCreate, rule! { ON_CREATE }),
134        value(InitializeMode::OnSchedule, rule! { ON_SCHEDULE }),
135    ));
136    let initialize_opt = map(
137        rule! {
138            (INITIALIZE ~ "=" ~ #initialize_mode)?
139        },
140        |v| v.map(|v| v.2),
141    );
142
143    permutation((
144        target_lag,
145        task_warehouse_option,
146        refresh_mode_opt,
147        initialize_opt,
148    ))
149    .parse(i)
150}
151
152fn target_lag(i: Input) -> IResult<TargetLag> {
153    let interval_sec = map(
154        rule! {
155             #literal_u64 ~ SECOND
156        },
157        |(secs, _)| TargetLag::IntervalSecs(secs),
158    );
159    let interval_min = map(
160        rule! {
161             #literal_u64 ~ MINUTE
162        },
163        |(mins, _)| TargetLag::IntervalSecs(mins * 60),
164    );
165    let interval_hour = map(
166        rule! {
167             #literal_u64 ~ HOUR
168        },
169        |(hours, _)| TargetLag::IntervalSecs(hours * 60 * 60),
170    );
171    let interval_day = map(
172        rule! {
173             #literal_u64 ~ DAY
174        },
175        |(days, _)| TargetLag::IntervalSecs(days * 60 * 60 * 24),
176    );
177    let downstream = map(
178        rule! {
179            DOWNSTREAM
180        },
181        |_| TargetLag::Downstream,
182    );
183    rule!(
184        #interval_sec
185        | #interval_min
186        | #interval_hour
187        | #interval_day
188        | #downstream
189    )
190    .parse(i)
191}
databend_common_ast/parser/dynamic_table.rs

databend_common_ast/parser/
dynamic_table.rs
