databend_common_ast/ast/statements/
copy.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::collections::BTreeMap;
16use std::collections::HashSet;
17use std::fmt;
18use std::fmt::Display;
19use std::fmt::Formatter;
20use std::str::FromStr;
21
22use derive_visitor::Drive;
23use derive_visitor::DriveMut;
24use itertools::Itertools;
25use percent_encoding::percent_decode_str;
26use url::Url;
27
28use crate::ParseError;
29use crate::Result;
30use crate::ast::Expr;
31use crate::ast::Hint;
32use crate::ast::Identifier;
33use crate::ast::Query;
34use crate::ast::SelectTarget;
35use crate::ast::With;
36use crate::ast::WithOptions;
37use crate::ast::quote::QuotedString;
38use crate::ast::write_comma_separated_list;
39use crate::ast::write_comma_separated_map;
40use crate::ast::write_comma_separated_string_list;
41use crate::ast::write_comma_separated_string_map;
42use crate::ast::write_dot_separated_list;
43
44/// CopyIntoTableStmt is the parsed statement of `COPY into <table> from <location>`.
45///
46/// ## Examples
47///
48/// ```sql
49/// COPY INTO table from s3://bucket/path/to/x.csv
50/// ```
51#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
52pub struct CopyIntoTableStmt {
53    pub with: Option<With>,
54    pub src: CopyIntoTableSource,
55    pub catalog: Option<Identifier>,
56    pub database: Option<Identifier>,
57    pub table: Identifier,
58    pub dst_columns: Option<Vec<Identifier>>,
59
60    pub hints: Option<Hint>,
61
62    pub file_format: FileFormatOptions,
63
64    // files to load
65    pub files: Option<Vec<String>>,
66    pub pattern: Option<LiteralStringOrVariable>,
67
68    pub options: CopyIntoTableOptions,
69}
70
71impl CopyIntoTableStmt {
72    pub fn apply_option(
73        &mut self,
74        opt: CopyIntoTableOption,
75    ) -> std::result::Result<(), &'static str> {
76        match opt {
77            CopyIntoTableOption::Files(v) => self.files = Some(v),
78            CopyIntoTableOption::Pattern(v) => self.pattern = Some(v),
79            CopyIntoTableOption::FileFormat(v) => self.file_format = v,
80            CopyIntoTableOption::SizeLimit(v) => self.options.size_limit = v,
81            CopyIntoTableOption::MaxFiles(v) => self.options.max_files = v,
82            CopyIntoTableOption::SplitSize(v) => self.options.split_size = v,
83            CopyIntoTableOption::Purge(v) => self.options.purge = v,
84            CopyIntoTableOption::Force(v) => self.options.force = v,
85            CopyIntoTableOption::DisableVariantCheck(v) => self.options.disable_variant_check = v,
86            CopyIntoTableOption::ReturnFailedOnly(v) => self.options.return_failed_only = v,
87            CopyIntoTableOption::OnError(v) => self.options.on_error = OnErrorMode::from_str(&v)?,
88            CopyIntoTableOption::ColumnMatchMode(v) => {
89                self.options.column_match_mode = Some(ColumnMatchMode::from_str(&v)?)
90            }
91        }
92        Ok(())
93    }
94}
95
96impl Display for CopyIntoTableStmt {
97    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
98        if let Some(cte) = &self.with {
99            write!(f, "WITH {} ", cte)?;
100        }
101        write!(f, "COPY")?;
102        if let Some(hints) = &self.hints {
103            write!(f, "{} ", hints)?;
104        }
105
106        write!(f, " INTO ")?;
107        write_dot_separated_list(
108            f,
109            self.catalog
110                .iter()
111                .chain(self.database.iter())
112                .chain(Some(&self.table)),
113        )?;
114
115        if let Some(columns) = &self.dst_columns {
116            write!(f, "({})", columns.iter().map(|c| c.to_string()).join(","))?;
117        }
118        write!(f, " FROM {}", self.src)?;
119
120        if let Some(files) = &self.files {
121            write!(f, " FILES = (")?;
122            write_comma_separated_string_list(f, files)?;
123            write!(f, " )")?;
124        }
125
126        if let Some(pattern) = &self.pattern {
127            write!(f, " PATTERN = {}", pattern)?;
128        }
129
130        if !self.file_format.is_empty() {
131            write!(f, " FILE_FORMAT = ({})", self.file_format)?;
132        }
133        write!(f, " {}", self.options)?;
134        Ok(())
135    }
136}
137
138#[derive(
139    serde::Serialize, serde::Deserialize, Debug, Clone, Default, PartialEq, Drive, DriveMut, Eq,
140)]
141pub struct CopyIntoTableOptions {
142    pub on_error: OnErrorMode,
143    pub max_files: usize,
144    pub force: bool,
145    pub purge: bool,
146    pub disable_variant_check: bool,
147    pub return_failed_only: bool,
148    pub column_match_mode: Option<ColumnMatchMode>,
149
150    // not used for now
151    pub size_limit: usize,
152    pub split_size: usize,
153    pub validation_mode: String,
154}
155
156impl CopyIntoTableOptions {
157    fn parse_uint(k: &str, v: &String) -> std::result::Result<usize, String> {
158        usize::from_str(v).map_err(|e| format!("can not parse {}={} as uint: {}", k, v, e))
159    }
160    fn parse_bool(k: &str, v: &String) -> std::result::Result<bool, String> {
161        bool::from_str(v).map_err(|e| format!("can not parse {}={} as bool: {}", k, v, e))
162    }
163
164    pub fn set_column_match_mode(&mut self, mode: ColumnMatchMode) {
165        self.column_match_mode = Some(mode);
166    }
167
168    pub fn apply(
169        &mut self,
170        opts: &BTreeMap<String, String>,
171        ignore_unknown: bool,
172    ) -> std::result::Result<(), String> {
173        if opts.is_empty() {
174            return Ok(());
175        }
176        for (k, v) in opts.iter() {
177            match k.as_str() {
178                "on_error" => {
179                    let on_error = OnErrorMode::from_str(v)?;
180                    self.on_error = on_error;
181                }
182                "column_match_mode" => {
183                    let column_match_mode = ColumnMatchMode::from_str(v)?;
184                    self.column_match_mode = Some(column_match_mode);
185                }
186                "size_limit" => {
187                    self.size_limit = Self::parse_uint(k, v)?;
188                }
189                "max_files" => {
190                    self.max_files = Self::parse_uint(k, v)?;
191                }
192                "split_size" => {
193                    self.split_size = Self::parse_uint(k, v)?;
194                }
195                "purge" => {
196                    self.purge = Self::parse_bool(k, v)?;
197                }
198                "disable_variant_check" => {
199                    self.disable_variant_check = Self::parse_bool(k, v)?;
200                }
201                "return_failed_only" => {
202                    self.return_failed_only = Self::parse_bool(k, v)?;
203                }
204                _ => {
205                    if !ignore_unknown {
206                        return Err(format!("Unknown stage copy option {}", k));
207                    }
208                }
209            }
210        }
211        Ok(())
212    }
213}
214
215impl Display for CopyIntoTableOptions {
216    fn fmt(&self, f: &mut Formatter) -> fmt::Result {
217        if !self.validation_mode.is_empty() {
218            write!(f, "VALIDATION_MODE = {}", self.validation_mode)?;
219        }
220
221        if self.size_limit != 0 {
222            write!(f, " SIZE_LIMIT = {}", self.size_limit)?;
223        }
224
225        if self.max_files != 0 {
226            write!(f, " MAX_FILES = {}", self.max_files)?;
227        }
228
229        if self.split_size != 0 {
230            write!(f, " SPLIT_SIZE = {}", self.split_size)?;
231        }
232
233        write!(f, " PURGE = {}", self.purge)?;
234        write!(f, " FORCE = {}", self.force)?;
235        write!(f, " DISABLE_VARIANT_CHECK = {}", self.disable_variant_check)?;
236        write!(f, " ON_ERROR = {}", self.on_error)?;
237        write!(f, " RETURN_FAILED_ONLY = {}", self.return_failed_only)?;
238        if let Some(mode) = &self.column_match_mode {
239            write!(f, " COLUMN_MATCH_MODE = {}", mode)?;
240        }
241        Ok(())
242    }
243}
244
245#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Drive, DriveMut, Eq)]
246pub struct CopyIntoLocationOptions {
247    pub single: bool,
248    pub max_file_size: usize,
249    pub detailed_output: bool,
250    pub use_raw_path: bool,
251    pub include_query_id: bool,
252    pub overwrite: bool,
253}
254
255#[derive(
256    serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Drive, DriveMut, Eq, Default,
257)]
258pub struct CopyIntoLocationOptionsRaw {
259    pub single: Option<bool>,
260    pub max_file_size: Option<usize>,
261    pub detailed_output: Option<bool>,
262    pub use_raw_path: Option<bool>,
263    pub include_query_id: Option<bool>,
264    pub overwrite: Option<bool>,
265}
266
267impl CopyIntoLocationOptionsRaw {
268    pub fn with_defaults(&self) -> CopyIntoLocationOptions {
269        let defaults = CopyIntoLocationOptions::default();
270        CopyIntoLocationOptions {
271            single: self.single.unwrap_or(defaults.single),
272            max_file_size: self.max_file_size.unwrap_or(defaults.max_file_size),
273            detailed_output: self.detailed_output.unwrap_or(defaults.detailed_output),
274            use_raw_path: self.use_raw_path.unwrap_or(defaults.use_raw_path),
275            include_query_id: self.include_query_id.unwrap_or(defaults.include_query_id),
276            overwrite: self.overwrite.unwrap_or(defaults.overwrite),
277        }
278    }
279}
280
281impl Default for CopyIntoLocationOptions {
282    fn default() -> Self {
283        Self {
284            single: Default::default(),
285            max_file_size: Default::default(),
286            detailed_output: false,
287            use_raw_path: false,
288            include_query_id: true,
289            overwrite: false,
290        }
291    }
292}
293
294/// CopyIntoLocationStmt is the parsed statement of `COPY into <location>  from <table> ...`
295#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
296pub struct CopyIntoLocationStmt {
297    pub with: Option<With>,
298    pub hints: Option<Hint>,
299    pub src: CopyIntoLocationSource,
300    pub dst: FileLocation,
301    pub partition_by: Option<Expr>,
302    pub file_format: FileFormatOptions,
303    pub options: CopyIntoLocationOptionsRaw,
304}
305
306impl Display for CopyIntoLocationStmt {
307    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
308        if let Some(cte) = &self.with {
309            write!(f, "WITH {} ", cte)?;
310        }
311        write!(f, "COPY")?;
312        if let Some(hints) = &self.hints {
313            write!(f, "{} ", hints)?;
314        }
315        write!(f, " INTO {}", self.dst)?;
316        write!(f, " FROM {}", self.src)?;
317        if let Some(partition_by) = &self.partition_by {
318            write!(f, " PARTITION BY ({partition_by})")?;
319        }
320
321        if !self.file_format.is_empty() {
322            write!(f, " FILE_FORMAT = ({})", self.file_format)?;
323        }
324        if let Some(single) = self.options.single {
325            write!(f, " SINGLE = {single}")?;
326        }
327        if let Some(max_file_size) = self.options.max_file_size {
328            write!(f, " MAX_FILE_SIZE = {max_file_size}")?;
329        }
330        if let Some(detailed_output) = self.options.detailed_output {
331            write!(f, " DETAILED_OUTPUT = {detailed_output}")?;
332        }
333        if let Some(include_query_id) = self.options.include_query_id {
334            write!(f, " INCLUDE_QUERY_ID = {include_query_id}")?;
335        }
336        if let Some(use_raw_path) = self.options.use_raw_path {
337            write!(f, " USE_RAW_PATH = {use_raw_path}")?;
338        }
339        if let Some(overwrite) = self.options.overwrite {
340            write!(f, " OVERWRITE = {overwrite}")?;
341        }
342
343        Ok(())
344    }
345}
346
347impl CopyIntoLocationStmt {
348    pub fn apply_option(&mut self, opt: CopyIntoLocationOption) {
349        match opt {
350            CopyIntoLocationOption::FileFormat(v) => self.file_format = v,
351            CopyIntoLocationOption::Single(v) => self.options.single = Some(v),
352            CopyIntoLocationOption::MaxFileSize(v) => self.options.max_file_size = Some(v),
353            CopyIntoLocationOption::DetailedOutput(v) => self.options.detailed_output = Some(v),
354            CopyIntoLocationOption::IncludeQueryID(v) => self.options.include_query_id = Some(v),
355            CopyIntoLocationOption::UseRawPath(v) => self.options.use_raw_path = Some(v),
356            CopyIntoLocationOption::OverWrite(v) => self.options.overwrite = Some(v),
357        }
358    }
359}
360
361#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
362pub enum CopyIntoTableSource {
363    Location(FileLocation),
364    /// Load with Transform
365    /// limited to `(SELECT ... FROM <location>)`
366    Query {
367        select_list: Vec<SelectTarget>,
368        from: FileLocation,
369        // no need to support this, for compatible only
370        alias_name: Option<Identifier>,
371    },
372}
373
374impl Display for CopyIntoTableSource {
375    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
376        match self {
377            CopyIntoTableSource::Location(location) => write!(f, "{location}"),
378            CopyIntoTableSource::Query {
379                select_list,
380                from,
381                alias_name,
382            } => {
383                write!(f, "(SELECT ")?;
384                write_comma_separated_list(f, select_list)?;
385                write!(f, " FROM {from}")?;
386                if let Some(a) = alias_name {
387                    write!(f, " AS {}", a.name)?;
388                }
389                write!(f, ")")?;
390                Ok(())
391            }
392        }
393    }
394}
395
396#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
397pub enum CopyIntoLocationSource {
398    Query(Box<Query>),
399    /// it will be rewritten as `(SELECT * FROM table)`
400    Table {
401        catalog: Option<Identifier>,
402        database: Option<Identifier>,
403        table: Identifier,
404        with_options: Option<WithOptions>,
405    },
406}
407
408impl Display for CopyIntoLocationSource {
409    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
410        match self {
411            CopyIntoLocationSource::Query(query) => {
412                write!(f, "({query})")
413            }
414            CopyIntoLocationSource::Table {
415                catalog,
416                database,
417                table,
418                with_options,
419            } => {
420                write_dot_separated_list(
421                    f,
422                    catalog.iter().chain(database.iter()).chain(Some(table)),
423                )?;
424                if let Some(with_options) = with_options {
425                    write!(f, " {with_options}")?;
426                }
427                Ok(())
428            }
429        }
430    }
431}
432
433#[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut)]
434pub struct Connection {
435    #[drive(skip)]
436    visited_keys: HashSet<String>,
437    pub conns: BTreeMap<String, String>,
438}
439
440impl Connection {
441    pub fn new(conns: BTreeMap<String, String>) -> Self {
442        Self {
443            visited_keys: HashSet::new(),
444            conns,
445        }
446    }
447
448    pub fn mask(&self) -> Self {
449        let mut conns = BTreeMap::new();
450        for (k, v) in &self.conns {
451            conns.insert(k.to_string(), mask_string(v, 3));
452        }
453        Self {
454            visited_keys: self.visited_keys.clone(),
455            conns,
456        }
457    }
458
459    pub fn get(&mut self, key: &str) -> Option<&String> {
460        self.visited_keys.insert(key.to_string());
461        self.conns.get(key)
462    }
463
464    pub fn check(&self) -> Result<()> {
465        let conn_keys = HashSet::from_iter(self.conns.keys().cloned());
466        let diffs: Vec<String> = conn_keys
467            .difference(&self.visited_keys)
468            .map(|x| x.to_string())
469            .collect();
470
471        if !diffs.is_empty() {
472            return Err(ParseError(
473                None,
474                format!(
475                    "connection params invalid: expected [{}], got [{}]",
476                    self.visited_keys
477                        .iter()
478                        .cloned()
479                        .collect::<Vec<_>>()
480                        .join(","),
481                    diffs.join(",")
482                ),
483            ));
484        }
485        Ok(())
486    }
487}
488
489impl Display for Connection {
490    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
491        if !self.conns.is_empty() {
492            write!(f, " CONNECTION = ( ")?;
493            write_comma_separated_string_map(f, &self.conns)?;
494            write!(f, " )")?;
495        }
496        Ok(())
497    }
498}
499
500/// Mask a string by "******", but keep `unmask_len` of suffix.
501fn mask_string(s: &str, unmask_len: usize) -> String {
502    if s.len() <= unmask_len {
503        s.to_string()
504    } else {
505        let mut ret = "******".to_string();
506        ret.push_str(&s[(s.len() - unmask_len)..]);
507        ret
508    }
509}
510
511/// UriLocation (a.k.a external location) can be used in `INTO` or `FROM`.
512///
513/// For examples: `'s3://example/path/to/dir' CONNECTION = (AWS_ACCESS_ID="admin" AWS_SECRET_KEY="admin")`
514#[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut)]
515pub struct UriLocation {
516    pub protocol: String,
517    pub name: String,
518    pub path: String,
519    pub connection: Connection,
520}
521
522impl UriLocation {
523    pub fn new(
524        protocol: String,
525        name: String,
526        path: String,
527        conns: BTreeMap<String, String>,
528    ) -> Self {
529        Self {
530            protocol,
531            name,
532            path,
533            connection: Connection::new(conns),
534        }
535    }
536
537    pub fn from_uri(uri: String, conns: BTreeMap<String, String>) -> Result<Self> {
538        // fs location is not a valid url, let's check it in advance.
539        if let Some(path) = uri.strip_prefix("fs://") {
540            if !path.starts_with('/') {
541                return Err(ParseError(
542                    None,
543                    format!("Invalid uri: {}. fs location must start with 'fs:///'", uri),
544                ));
545            }
546            return Ok(UriLocation::new(
547                "fs".to_string(),
548                "".to_string(),
549                path.to_string(),
550                BTreeMap::default(),
551            ));
552        }
553
554        let parsed =
555            Url::parse(&uri).map_err(|e| ParseError(None, format!("invalid uri {}", e)))?;
556
557        let protocol = parsed.scheme().to_string();
558
559        let name = parsed
560            .host_str()
561            .map(|hostname| {
562                if let Some(port) = parsed.port() {
563                    format!("{}:{}", hostname, port)
564                } else {
565                    hostname.to_string()
566                }
567            })
568            .ok_or_else(|| ParseError(None, "invalid uri".to_string()))?;
569
570        let path = if parsed.path().is_empty() {
571            "/".to_string()
572        } else {
573            percent_decode_str(parsed.path())
574                .decode_utf8_lossy()
575                .to_string()
576        };
577
578        Ok(Self {
579            protocol,
580            name,
581            path,
582            connection: Connection::new(conns),
583        })
584    }
585
586    pub fn mask(&self) -> Self {
587        Self {
588            connection: self.connection.mask(),
589            ..self.clone()
590        }
591    }
592}
593
594impl Display for UriLocation {
595    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
596        write!(f, "'{}://{}{}'", self.protocol, self.name, self.path)?;
597        write!(f, "{}", self.connection)?;
598        Ok(())
599    }
600}
601
602/// StageLocation (a.k.a internal and external stage) can be used
603/// in `INTO` or `FROM`.
604///
605/// For examples:
606///
607/// - internal stage: `@internal_stage/path/to/dir/`
608/// - external stage: `@s3_external_stage/path/to/dir/`
609///
610/// UriLocation (a.k.a external location) can be used in `INTO` or `FROM`.
611///
612/// For examples: `'s3://example/path/to/dir' CONNECTION = (AWS_ACCESS_ID="admin" AWS_SECRET_KEY="admin")`
613#[derive(Debug, Clone, PartialEq, Eq, Drive, DriveMut)]
614pub enum FileLocation {
615    Stage(String),
616    Uri(UriLocation),
617}
618
619impl Display for FileLocation {
620    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
621        match self {
622            FileLocation::Uri(loc) => {
623                write!(f, "{}", loc)
624            }
625            FileLocation::Stage(loc) => {
626                write!(f, "'@{}'", loc)
627            }
628        }
629    }
630}
631
632/// Used when we want to allow use variable for options etc.
633/// Other expr is not necessary, because
634/// 1. we can always create a variable that can be used directly.
635/// 2. columns can not be referred.
636///
637/// Can extend to all type of Literals if needed later.
638#[derive(Debug, Clone, PartialEq, Drive, DriveMut)]
639pub enum LiteralStringOrVariable {
640    Literal(String),
641    Variable(String),
642}
643
644impl Display for LiteralStringOrVariable {
645    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
646        match self {
647            LiteralStringOrVariable::Literal(s) => {
648                write!(f, "'{s}'")
649            }
650            LiteralStringOrVariable::Variable(s) => {
651                write!(f, "${s}")
652            }
653        }
654    }
655}
656
657pub enum CopyIntoTableOption {
658    Files(Vec<String>),
659    Pattern(LiteralStringOrVariable),
660    FileFormat(FileFormatOptions),
661    SizeLimit(usize),
662    MaxFiles(usize),
663    SplitSize(usize),
664    Purge(bool),
665    Force(bool),
666    DisableVariantCheck(bool),
667    ReturnFailedOnly(bool),
668    OnError(String),
669    ColumnMatchMode(String),
670}
671
672pub enum CopyIntoLocationOption {
673    FileFormat(FileFormatOptions),
674    MaxFileSize(usize),
675    Single(bool),
676    IncludeQueryID(bool),
677    UseRawPath(bool),
678    DetailedOutput(bool),
679    OverWrite(bool),
680}
681
682#[derive(Clone, Debug, PartialEq, Eq, Default, Drive, DriveMut)]
683pub struct FileFormatOptions {
684    pub options: BTreeMap<String, FileFormatValue>,
685}
686
687impl FileFormatOptions {
688    pub fn is_empty(&self) -> bool {
689        self.options.is_empty()
690    }
691}
692
693impl Display for FileFormatOptions {
694    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
695        write_comma_separated_map(f, &self.options)
696    }
697}
698
699#[derive(Clone, Debug, PartialEq, Eq, Drive, DriveMut)]
700pub enum FileFormatValue {
701    Keyword(String),
702    Bool(bool),
703    U64(u64),
704    String(String),
705    StringList(Vec<String>),
706}
707
708impl FileFormatValue {
709    pub fn to_meta_value(&self) -> String {
710        match self {
711            FileFormatValue::Keyword(v) => v.clone(),
712            FileFormatValue::Bool(v) => v.to_string(),
713            FileFormatValue::U64(v) => v.to_string(),
714            FileFormatValue::String(v) => v.clone(),
715            FileFormatValue::StringList(v) => serde_json::to_string(v).unwrap(),
716        }
717    }
718}
719
720impl Display for FileFormatValue {
721    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
722        match self {
723            FileFormatValue::Keyword(v) => write!(f, "{v}"),
724            FileFormatValue::Bool(v) => write!(f, "{v}"),
725            FileFormatValue::U64(v) => write!(f, "{v}"),
726            FileFormatValue::String(v) => {
727                write!(f, "{}", QuotedString(v, '\''))
728            }
729            FileFormatValue::StringList(v) => {
730                write!(f, "(")?;
731                for (i, s) in v.iter().enumerate() {
732                    if i > 0 {
733                        write!(f, ", ")?;
734                    }
735                    write!(f, "{}", QuotedString(s, '\''))?;
736                }
737                write!(f, ")")
738            }
739        }
740    }
741}
742
743#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Drive, DriveMut, Eq)]
744pub enum OnErrorMode {
745    Continue,
746    SkipFileNum(u64),
747    AbortNum(u64),
748}
749
750impl Default for OnErrorMode {
751    fn default() -> Self {
752        Self::AbortNum(1)
753    }
754}
755
756impl Display for OnErrorMode {
757    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
758        match self {
759            OnErrorMode::Continue => {
760                write!(f, "continue")
761            }
762            OnErrorMode::SkipFileNum(n) => {
763                if *n <= 1 {
764                    write!(f, "skipfile")
765                } else {
766                    write!(f, "skipfile_{}", n)
767                }
768            }
769            OnErrorMode::AbortNum(n) => {
770                if *n <= 1 {
771                    write!(f, "abort")
772                } else {
773                    write!(f, "abort_{}", n)
774                }
775            }
776        }
777    }
778}
779
780const ERROR_MODE_MSG: &str =
781    "OnError must one of {{ CONTINUE | SKIP_FILE | SKIP_FILE_<num> | ABORT | ABORT_<num> }}";
782impl FromStr for OnErrorMode {
783    type Err = &'static str;
784
785    fn from_str(s: &str) -> std::result::Result<Self, &'static str> {
786        match s.to_uppercase().as_str() {
787            "" | "ABORT" => Ok(OnErrorMode::AbortNum(1)),
788            "CONTINUE" => Ok(OnErrorMode::Continue),
789            "SKIP_FILE" => Ok(OnErrorMode::SkipFileNum(1)),
790            v => {
791                if v.starts_with("ABORT_") {
792                    let num_str = v.replace("ABORT_", "");
793                    let nums = num_str.parse::<u64>();
794                    match nums {
795                        Ok(n) if n < 1 => Err(ERROR_MODE_MSG),
796                        Ok(n) => Ok(OnErrorMode::AbortNum(n)),
797                        Err(_) => Err(ERROR_MODE_MSG),
798                    }
799                } else {
800                    let num_str = v.replace("SKIP_FILE_", "");
801                    let nums = num_str.parse::<u64>();
802                    match nums {
803                        Ok(n) if n < 1 => Err(ERROR_MODE_MSG),
804                        Ok(n) => Ok(OnErrorMode::SkipFileNum(n)),
805                        Err(_) => Err(ERROR_MODE_MSG),
806                    }
807                }
808            }
809        }
810    }
811}
812
813#[derive(serde::Serialize, serde::Deserialize, Debug, Clone, PartialEq, Drive, DriveMut, Eq)]
814pub enum ColumnMatchMode {
815    CaseSensitive,
816    CaseInsensitive,
817    Position,
818}
819
820impl Display for ColumnMatchMode {
821    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
822        match self {
823            ColumnMatchMode::CaseSensitive => write!(f, "CASE_SENSITIVE"),
824            ColumnMatchMode::CaseInsensitive => write!(f, "CASE_INSENSITIVE"),
825            ColumnMatchMode::Position => write!(f, "POSITION"),
826        }
827    }
828}
829
830const COLUMN_MATCH_MODE_MSG: &str =
831    "ColumnMatchMode must be one of {{ CASE_SENSITIVE | CASE_INSENSITIVE | POSITION }}";
832impl FromStr for ColumnMatchMode {
833    type Err = &'static str;
834
835    fn from_str(s: &str) -> std::result::Result<Self, &'static str> {
836        match s.to_uppercase().as_str() {
837            "CASE_SENSITIVE" => Ok(Self::CaseSensitive),
838            "CASE_INSENSITIVE" => Ok(Self::CaseInsensitive),
839            "POSITION" => Ok(Self::Position),
840            _ => Err(COLUMN_MATCH_MODE_MSG),
841        }
842    }
843}
databend_common_ast/ast/statements/copy.rs

databend_common_ast/ast/statements/
copy.rs
