feat: add optional parameters for copy table from command

Signed-off-by: Alan Tang <[email protected]>

Author: Standing-Man

src/common/datasource/src/file_format.rs

@@ -52,6 +52,8 @@ pub const FORMAT_COMPRESSION_TYPE: &str = "compression_type";
 pub const FORMAT_DELIMITER: &str = "delimiter";
 pub const FORMAT_SCHEMA_INFER_MAX_RECORD: &str = "schema_infer_max_record";
 pub const FORMAT_HAS_HEADER: &str = "has_header";
+pub const FORMAT_HEADER: &str = "header";
+pub const FORMAT_SKIP_BAD_RECORDS: &str = "skip_bad_records";
 pub const FORMAT_TYPE: &str = "format";
 pub const FILE_PATTERN: &str = "pattern";
 pub const TIMESTAMP_FORMAT: &str = "timestamp_format";

src/common/datasource/src/file_format/csv.rs

@@ -36,6 +36,8 @@ use crate::share_buffer::SharedBuffer;
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct CsvFormat {
     pub has_header: bool,
+    pub header: bool,
+    pub skip_bad_records: bool,
     pub delimiter: u8,
     pub schema_infer_max_record: Option<usize>,
     pub compression_type: CompressionType,
@@ -83,6 +85,24 @@ impl TryFrom<&HashMap<String, String>> for CsvFormat {
                 .build()
             })?;
         };
+        if let Some(header) = value.get(file_format::FORMAT_HEADER) {
+            format.header = header.parse().map_err(|_| {
+                error::ParseFormatSnafu {
+                    key: file_format::FORMAT_HEADER,
+                    value: header,
+                }
+                .build()
+            })?;
+        }
+        if let Some(skip_bad_records) = value.get(file_format::FORMAT_SKIP_BAD_RECORDS) {
+            format.header = skip_bad_records.parse().map_err(|_| {
+                error::ParseFormatSnafu {
+                    key: file_format::FORMAT_SKIP_BAD_RECORDS,
+                    value: skip_bad_records,
+                }
+                .build()
+            })?;
+        }
         if let Some(timestamp_format) = value.get(file_format::TIMESTAMP_FORMAT) {
             format.timestamp_format = Some(timestamp_format.clone());
         }
@@ -100,6 +120,8 @@ impl Default for CsvFormat {
     fn default() -> Self {
         Self {
             has_header: true,
+            header: true,
+            skip_bad_records: false,
             delimiter: b',',
             schema_infer_max_record: Some(file_format::DEFAULT_SCHEMA_INFER_MAX_RECORD),
             compression_type: CompressionType::Uncompressed,
@@ -291,6 +313,8 @@ mod tests {
                 schema_infer_max_record: Some(2000),
                 delimiter: b'\t',
                 has_header: false,
+                header: true,
+                skip_bad_records: false,
                 timestamp_format: None,
                 time_format: None,
                 date_format: None