databend_common_ast/ast/statements/
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 std::fmt::Display;
16use std::fmt::Formatter;
17
18use derive_visitor::Drive;
19use derive_visitor::DriveMut;
20
21use crate::ast::quote::QuotedString;
22use crate::ast::write_dot_separated_list;
23use crate::ast::CreateOption;
24use crate::ast::Identifier;
25use crate::ast::ShowLimit;
26use crate::ast::TimeTravelPoint;
27
28#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
29pub struct CreateStreamStmt {
30    pub create_option: CreateOption,
31    pub catalog: Option<Identifier>,
32    pub database: Option<Identifier>,
33    pub stream: Identifier,
34    pub table_database: Option<Identifier>,
35    pub table: Identifier,
36    pub travel_point: Option<TimeTravelPoint>,
37    pub append_only: bool,
38    pub comment: Option<String>,
39}
40
41impl Display for CreateStreamStmt {
42    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
43        write!(f, "CREATE ")?;
44        if let CreateOption::CreateOrReplace = self.create_option {
45            write!(f, "OR REPLACE ")?;
46        }
47        write!(f, "STREAM ")?;
48        if let CreateOption::CreateIfNotExists = self.create_option {
49            write!(f, "IF NOT EXISTS ")?;
50        }
51        write_dot_separated_list(
52            f,
53            self.catalog
54                .iter()
55                .chain(self.database.iter())
56                .chain(Some(&self.stream)),
57        )?;
58        write!(f, " ON TABLE ")?;
59        write_dot_separated_list(f, self.table_database.iter().chain(Some(&self.table)))?;
60        if let Some(travel_point) = &self.travel_point {
61            write!(f, " AT {}", travel_point)?;
62        }
63        if !self.append_only {
64            write!(f, " APPEND_ONLY = false")?;
65        }
66        if let Some(comment) = &self.comment {
67            write!(f, " COMMENT = {}", QuotedString(comment, '\''))?;
68        }
69        Ok(())
70    }
71}
72
73#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
74pub struct DropStreamStmt {
75    pub if_exists: bool,
76    pub catalog: Option<Identifier>,
77    pub database: Option<Identifier>,
78    pub stream: Identifier,
79}
80
81impl Display for DropStreamStmt {
82    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
83        write!(f, "DROP STREAM ")?;
84        if self.if_exists {
85            write!(f, "IF EXISTS ")?;
86        }
87        write_dot_separated_list(
88            f,
89            self.catalog
90                .iter()
91                .chain(self.database.iter())
92                .chain(Some(&self.stream)),
93        )
94    }
95}
96
97#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
98pub struct ShowStreamsStmt {
99    pub catalog: Option<Identifier>,
100    pub database: Option<Identifier>,
101    pub full: bool,
102    pub limit: Option<ShowLimit>,
103}
104
105impl Display for ShowStreamsStmt {
106    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
107        write!(f, "SHOW ")?;
108        if self.full {
109            write!(f, "FULL ")?;
110        }
111        write!(f, "STREAMS")?;
112        if let Some(database) = &self.database {
113            write!(f, " FROM ")?;
114            if let Some(catalog) = &self.catalog {
115                write!(f, "{catalog}.",)?;
116            }
117            write!(f, "{database}")?;
118        }
119        if let Some(limit) = &self.limit {
120            write!(f, " {limit}")?;
121        }
122
123        Ok(())
124    }
125}
126
127#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
128pub struct DescribeStreamStmt {
129    pub catalog: Option<Identifier>,
130    pub database: Option<Identifier>,
131    pub stream: Identifier,
132}
133
134impl Display for DescribeStreamStmt {
135    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
136        write!(f, "DESCRIBE STREAM ")?;
137        write_dot_separated_list(
138            f,
139            self.catalog
140                .iter()
141                .chain(self.database.iter().chain(Some(&self.stream))),
142        )
143    }
144}
databend_common_ast/ast/statements/stream.rs

databend_common_ast/ast/statements/
stream.rs
