databend_common_ast/parser/
stream.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::CreateStreamStmt;
19use crate::ast::DescribeStreamStmt;
20use crate::ast::DropStreamStmt;
21use crate::ast::ShowStreamsStmt;
22use crate::ast::Statement;
23use crate::parser::Input;
24use crate::parser::common::IResult;
25use crate::parser::common::dot_separated_idents_1_to_2;
26use crate::parser::common::dot_separated_idents_1_to_3;
27use crate::parser::common::map_res;
28use crate::parser::common::*;
29use crate::parser::expr::literal_bool;
30use crate::parser::expr::literal_string;
31use crate::parser::query::travel_point;
32use crate::parser::statement::parse_create_option;
33use crate::parser::statement::show_limit;
34use crate::parser::token::TokenKind::*;
35
36pub fn create_stream(i: Input) -> IResult<Statement> {
37    map_res(
38        rule! {
39            CREATE ~ ( OR ~ ^REPLACE )? ~ STREAM ~ ( IF ~ ^NOT ~ ^EXISTS )?
40            ~ #dot_separated_idents_1_to_3
41            ~ ON ~ TABLE ~ #dot_separated_idents_1_to_2
42            ~ ( AT ~ ^#travel_point )?
43            ~ ( APPEND_ONLY ~ "=" ~ #literal_bool )?
44            ~ ( COMMENT ~ "=" ~ #literal_string )?
45        },
46        |(
47            _,
48            opt_or_replace,
49            _,
50            opt_if_not_exists,
51            (catalog, database, stream),
52            _,
53            _,
54            (table_database, table),
55            opt_travel_point,
56            opt_append_only,
57            opt_comment,
58        )| {
59            let create_option =
60                parse_create_option(opt_or_replace.is_some(), opt_if_not_exists.is_some())?;
61            Ok(Statement::CreateStream(CreateStreamStmt {
62                create_option,
63                catalog,
64                database,
65                stream,
66                table_database,
67                table,
68                travel_point: opt_travel_point.map(|p| p.1),
69                append_only: opt_append_only
70                    .map(|(_, _, append_only)| append_only)
71                    .unwrap_or(true),
72                comment: opt_comment.map(|(_, _, comment)| comment),
73            }))
74        },
75    )(i)
76}
77
78pub fn drop_stream(i: Input) -> IResult<Statement> {
79    map(
80        rule! {
81            DROP ~ STREAM ~ ( IF ~ ^EXISTS )? ~ #dot_separated_idents_1_to_3
82        },
83        |(_, _, opt_if_exists, (catalog, database, stream))| {
84            Statement::DropStream(DropStreamStmt {
85                if_exists: opt_if_exists.is_some(),
86                catalog,
87                database,
88                stream,
89            })
90        },
91    )
92    .parse(i)
93}
94
95pub fn show_streams(i: Input) -> IResult<Statement> {
96    map(
97        rule! {
98            SHOW ~ FULL? ~ STREAMS ~ ( ( FROM | IN ) ~ #dot_separated_idents_1_to_2 )? ~ #show_limit?
99        },
100        |(_, opt_full, _, ctl_db, limit)| {
101            let (catalog, database) = match ctl_db {
102                Some((_, (Some(c), d))) => (Some(c), Some(d)),
103                Some((_, (None, d))) => (None, Some(d)),
104                _ => (None, None),
105            };
106            Statement::ShowStreams(ShowStreamsStmt {
107                catalog,
108                database,
109                full: opt_full.is_some(),
110                limit,
111            })
112        },
113    ).parse(i)
114}
115
116pub fn describe_stream(i: Input) -> IResult<Statement> {
117    map(
118        rule! {
119            ( DESC | DESCRIBE ) ~ STREAM ~ #dot_separated_idents_1_to_3
120        },
121        |(_, _, (catalog, database, stream))| {
122            Statement::DescribeStream(DescribeStreamStmt {
123                catalog,
124                database,
125                stream,
126            })
127        },
128    )
129    .parse(i)
130}
databend_common_ast/parser/stream.rs

databend_common_ast/parser/
stream.rs
