feat: remove the skip_bad_records parameter

Signed-off-by: StandingMan <jmtangcs@gmail.com>

Author: Standing-Man

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

@@ -37,7 +37,6 @@ use crate::share_buffer::SharedBuffer;
 pub struct CsvFormat {
     pub has_header: bool,
     pub header: bool,
-    pub skip_bad_records: bool,
     pub continue_on_error: bool,
     pub delimiter: u8,
     pub schema_infer_max_record: Option<usize>,
@@ -95,15 +94,6 @@ impl TryFrom<&HashMap<String, String>> for CsvFormat {
                 .build()
             })?;
         }
-        if let Some(skip_bad_records) = value.get(file_format::FORMAT_SKIP_BAD_RECORDS) {
-            format.skip_bad_records = skip_bad_records.parse().map_err(|_| {
-                error::ParseFormatSnafu {
-                    key: file_format::FORMAT_SKIP_BAD_RECORDS,
-                    value: skip_bad_records,
-                }
-                .build()
-            })?;
-        }
         if let Some(continue_on_error) = value.get(file_format::FORMAT_CONTINUE_ON_ERROR) {
             format.continue_on_error = continue_on_error.parse().map_err(|_| {
                 error::ParseFormatSnafu {
@@ -131,7 +121,6 @@ impl Default for CsvFormat {
         Self {
             has_header: true,
             header: false,
-            skip_bad_records: false,
             continue_on_error: true,
             delimiter: b',',
             schema_infer_max_record: Some(file_format::DEFAULT_SCHEMA_INFER_MAX_RECORD),
@@ -342,7 +331,6 @@ mod tests {
                 has_header: false,
                 continue_on_error: true,
                 header: false,
-                skip_bad_records: false,
                 timestamp_format: None,
                 time_format: None,
                 date_format: None

src/operator/src/statement/copy_table_from.rs

@@ -437,17 +437,7 @@ impl StatementExecutor {
             let mut pending = vec![];
 
             while let Some(r) = stream.next().await {
-                let record_batch = match r.context(error::ReadDfRecordBatchSnafu) {
-                    Ok(batch) => batch,
-                    Err(err) => {
-                        if let Some(csv_format) = csv_format_opt
-                            && csv_format.skip_bad_records
-                        {
-                            continue;
-                        }
-                        return Err(err);
-                    }
-                };
+                let record_batch = r.context(error::ReadDfRecordBatchSnafu)?;
                 let vectors =
                     Helper::try_into_vectors(record_batch.columns()).context(IntoVectorsSnafu)?;