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

databend_common_ast/parser/
stream.rs
