databend_common_ast/ast/statements/
pipe.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::CopyIntoTableStmt;
22
23#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
24pub struct CreatePipeStmt {
25    pub if_not_exists: bool,
26    pub name: String,
27    pub auto_ingest: bool,
28    pub comments: String,
29    pub copy_stmt: CopyIntoTableStmt,
30}
31
32impl Display for CreatePipeStmt {
33    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
34        write!(f, "CREATE PIPE")?;
35        if self.if_not_exists {
36            write!(f, " IF NOT EXISTS")?;
37        }
38        write!(f, " {}", self.name)?;
39
40        if self.auto_ingest {
41            write!(f, " AUTO_INGEST = TRUE")?;
42        }
43
44        if !self.comments.is_empty() {
45            write!(f, " COMMENTS = '{}'", self.comments)?;
46        }
47
48        write!(f, " AS {}", self.copy_stmt)?;
49        Ok(())
50    }
51}
52
53#[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut)]
54pub struct DropPipeStmt {
55    pub if_exists: bool,
56    pub name: String,
57}
58
59impl Display for DropPipeStmt {
60    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
61        write!(f, "DROP PIPE")?;
62        if self.if_exists {
63            write!(f, " IF EXISTS")?;
64        }
65        write!(f, " {}", self.name)
66    }
67}
68
69#[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut)]
70pub struct DescribePipeStmt {
71    pub name: String,
72}
73
74impl Display for DescribePipeStmt {
75    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
76        write!(f, "DESCRIBE PIPE {}", self.name)
77    }
78}
79
80#[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut)]
81pub struct AlterPipeStmt {
82    pub if_exists: bool,
83    pub name: String,
84    pub options: AlterPipeOptions,
85}
86
87#[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut)]
88pub enum AlterPipeOptions {
89    Set {
90        execution_paused: Option<bool>,
91        comments: Option<String>,
92    },
93    Refresh {
94        prefix: Option<String>,
95        modified_after: Option<String>,
96    },
97}
98
99impl Display for AlterPipeOptions {
100    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
101        match self {
102            AlterPipeOptions::Set {
103                execution_paused,
104                comments,
105            } => {
106                if let Some(execution_paused) = execution_paused {
107                    write!(f, " SET PIPE_EXECUTION_PAUSED = {}", execution_paused)?;
108                }
109                if let Some(comments) = comments {
110                    write!(f, " SET COMMENTS = '{}'", comments)?;
111                }
112                Ok(())
113            }
114            AlterPipeOptions::Refresh {
115                prefix,
116                modified_after,
117            } => {
118                write!(f, " REFRESH")?;
119                if let Some(prefix) = prefix {
120                    write!(f, " PREFIX = '{}'", prefix)?;
121                }
122                if let Some(modified_after) = modified_after {
123                    write!(f, " MODIFIED_AFTER = '{}'", modified_after)?;
124                }
125                Ok(())
126            }
127        }
128    }
129}
130
131impl Display for AlterPipeStmt {
132    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
133        write!(f, "ALTER PIPE")?;
134        if self.if_exists {
135            write!(f, " IF EXISTS")?;
136        }
137        write!(f, " {}", self.name)?;
138        write!(f, "{}", self.options)?;
139        Ok(())
140    }
141}
databend_common_ast/ast/statements/pipe.rs

databend_common_ast/ast/statements/
pipe.rs
