fix: distinguish quoted and unquoted empty and null string when loading CSV (#19207)

* feat: CSV support options QUOTED_EMPTY_FIELD_AS and ALLOW_QUOTED_NULLS

* feat: unload CSV support NULL_DISPLAY.

* feat: `EMPTY_FIELD_AS=String` try set NULL for non-string columns.

* ci: update tpcds logic test.

* ci: update logic test.

* skip compat test case

Author: youngsofun

Cargo.lock

@@ -197,7 +197,7 @@ cookie = "0.18.1"
 crc32fast = "1.3.2"
 cron = "0.12.0"
 crossbeam-channel = "0.5.6"
-csv-core = "0.1.11"
+csv-core = "0.1.13"
 ctor = "0.2"
 ctrlc = { version = "3.2.3", features = ["termination"] }
 dashmap = { version = "6.1.0", features = ["serde"] }
@@ -572,6 +572,7 @@ async-backtrace = { git = "https://github.com/datafuse-extras/async-backtrace.gi
 async-recursion = { git = "https://github.com/datafuse-extras/async-recursion.git", rev = "a353334" }
 backtrace = { git = "https://github.com/rust-lang/backtrace-rs.git", rev = "72265be" }
 color-eyre = { git = "https://github.com/eyre-rs/eyre.git", rev = "e5d92c3" }
+csv-core = { git = "https://github.com/youngsofun/rust-csv.git", rev = "346f18c" }
 databend-base = { git = "https://github.com/databendlabs/databend-base", tag = "v0.1.0" }
 deltalake = { git = "https://github.com/delta-io/delta-rs", rev = "9954bff" }
 jsonb = { git = "https://github.com/databendlabs/jsonb", rev = "1868abf" }

Cargo.toml

@@ -44,6 +44,8 @@ const OPT_ESCAPE: &str = "escape";
 const OPT_QUOTE: &str = "quote";
 const OPT_ROW_TAG: &str = "row_tag";
 const OPT_ERROR_ON_COLUMN_COUNT_MISMATCH: &str = "error_on_column_count_mismatch";
+const OPT_ALLOW_QUOTED_NULLS: &str = "allow_quoted_nulls";
+const OPT_QUOTED_EMPTY_FIELD_AS: &str = "quoted_empty_field_as";
 const MISSING_FIELD_AS: &str = "missing_field_as";
 const NULL_FIELD_AS: &str = "null_field_as";
 const NULL_IF: &str = "null_if";
@@ -162,7 +164,10 @@ impl FileFormatParams {
     pub fn need_field_default(&self) -> bool {
         match self {
             FileFormatParams::Parquet(v) => v.missing_field_as == NullAs::FieldDefault,
-            FileFormatParams::Csv(v) => v.empty_field_as == EmptyFieldAs::FieldDefault,
+            FileFormatParams::Csv(v) => {
+                v.empty_field_as == EmptyFieldAs::FieldDefault
+                    || v.quoted_empty_field_as == EmptyFieldAs::FieldDefault
+            }
             FileFormatParams::NdJson(v) => {
                 v.null_field_as == NullAs::FieldDefault
                     || v.missing_field_as == NullAs::FieldDefault
@@ -257,6 +262,14 @@ impl FileFormatParams {
                     OPT_ERROR_ON_COLUMN_COUNT_MISMATCH,
                     default.error_on_column_count_mismatch,
                 )?;
+                let allow_quoted_nulls =
+                    reader.take_bool(OPT_ALLOW_QUOTED_NULLS, default.allow_quoted_nulls)?;
+                let quoted_empty_field_as = reader
+                    .options
+                    .remove(OPT_QUOTED_EMPTY_FIELD_AS)
+                    .map(|s| EmptyFieldAs::from_str(&s))
+                    .transpose()?
+                    .unwrap_or(default.quoted_empty_field_as.clone());
                 let output_header = reader.take_bool(OPT_OUTPUT_HEADER, default.output_header)?;
                 FileFormatParams::Csv(CsvFileFormatParams {
                     compression,
@@ -268,6 +281,8 @@ impl FileFormatParams {
                     escape,
                     quote,
                     error_on_column_count_mismatch,
+                    allow_quoted_nulls,
+                    quoted_empty_field_as,
                     empty_field_as,
                     binary_format,
                     output_header,
@@ -457,16 +472,20 @@ pub struct CsvFileFormatParams {
     pub quote: String,
     pub error_on_column_count_mismatch: bool,
 
+    // load only options
+    pub allow_quoted_nulls: bool,
+    pub empty_field_as: EmptyFieldAs,
+    pub quoted_empty_field_as: EmptyFieldAs,
+
     // header
     pub headers: u64,
     pub output_header: bool,
 
-    // field
-    pub binary_format: BinaryFormat,
-    pub null_display: String,
+    // field encoding/decoding
     pub nan_display: String,
-    pub empty_field_as: EmptyFieldAs,
+    pub binary_format: BinaryFormat,
     pub geometry_format: GeometryDataType,
+    pub null_display: String,
 }
 
 impl Default for CsvFileFormatParams {
@@ -476,15 +495,17 @@ impl Default for CsvFileFormatParams {
             headers: 0,
             field_delimiter: ",".to_string(),
             record_delimiter: "\n".to_string(),
-            null_display: NULL_BYTES_ESCAPE.to_string(),
-            nan_display: "NaN".to_string(),
             escape: "".to_string(),
             quote: "\"".to_string(),
             error_on_column_count_mismatch: true,
+            allow_quoted_nulls: false,
             empty_field_as: Default::default(),
+            quoted_empty_field_as: EmptyFieldAs::String,
             output_header: false,
             binary_format: Default::default(),
             geometry_format: GeometryDataType::default(),
+            nan_display: "NaN".to_string(),
+            null_display: NULL_BYTES_ESCAPE.to_string(),
         }
     }
 }
@@ -850,7 +871,7 @@ impl Display for FileFormatParams {
                      FIELD_DELIMITER = '{}' RECORD_DELIMITER = '{}' QUOTE = '{}' ESCAPE = '{}' \
                      SKIP_HEADER= {} OUTPUT_HEADER= {} \
                      NULL_DISPLAY = '{}' NAN_DISPLAY = '{}' EMPTY_FIELD_AS = {} BINARY_FORMAT = {} \
-                     ERROR_ON_COLUMN_COUNT_MISMATCH = {}",
+                     ERROR_ON_COLUMN_COUNT_MISMATCH = {} ALLOW_QUOTED_NULLS = {} QUOTED_EMPTY_FIELD_AS = {}",
                     params.compression,
                     escape_string(&params.field_delimiter),
                     escape_string(&params.record_delimiter),
@@ -863,6 +884,8 @@ impl Display for FileFormatParams {
                     params.empty_field_as,
                     params.binary_format,
                     params.error_on_column_count_mismatch,
+                    params.allow_quoted_nulls,
+                    params.quoted_empty_field_as,
                 )
             }
             FileFormatParams::Tsv(params) => {

src/meta/app/src/principal/file_format.rs

@@ -481,11 +481,7 @@ impl FromToProto for mt::principal::CsvFileFormatParams {
                 Incompatible::new(format!("invalid StageFileCompression: {}", p.compression))
             })?,
         )?;
-        let null_display = if p.null_display.is_empty() {
-            "\\N".to_string()
-        } else {
-            p.null_display
-        };
+        let null_display = p.null_display.unwrap_or_else(|| "\\N".to_string());
 
         let empty_field_as = p
             .empty_field_as
@@ -506,6 +502,12 @@ impl FromToProto for mt::principal::CsvFileFormatParams {
             .map_err(|e| Incompatible::new(format!("{:?}", e)))?
             .unwrap_or_default();
 
+        let quoted_empty_field_as = if let Some(value) = p.quoted_empty_field_as {
+            EmptyFieldAs::from_str(&value).map_err(|e| Incompatible::new(format!("{:?}", e)))?
+        } else {
+            EmptyFieldAs::String
+        };
+
         Ok(Self {
             compression,
             headers: p.headers,
@@ -516,7 +518,9 @@ impl FromToProto for mt::principal::CsvFileFormatParams {
             nan_display: p.nan_display,
             null_display,
             error_on_column_count_mismatch: !p.allow_column_count_mismatch,
+            allow_quoted_nulls: p.allow_quoted_nulls,
             empty_field_as,
+            quoted_empty_field_as,
             binary_format,
             output_header: p.output_header,
             geometry_format,
@@ -536,8 +540,10 @@ impl FromToProto for mt::principal::CsvFileFormatParams {
             quote: self.quote.clone(),
             escape: self.escape.clone(),
             nan_display: self.nan_display.clone(),
-            null_display: self.null_display.clone(),
+            null_display: Some(self.null_display.clone()),
             allow_column_count_mismatch: !self.error_on_column_count_mismatch,
+            allow_quoted_nulls: self.allow_quoted_nulls,
+            quoted_empty_field_as: Some(self.quoted_empty_field_as.to_string()),
             empty_field_as: Some(self.empty_field_as.to_string()),
             binary_format: Some(self.binary_format.to_string()),
             output_header: self.output_header,

src/meta/proto-conv/src/file_format_from_to_protobuf_impl.rs

@@ -193,6 +193,7 @@ const META_CHANGE_LOG: &[(u64, &str)] = &[
     (161, "2025-12-04: Add: Grant/OwnershipRowAccessPolicyObject and row access policy privileges"),
     (162, "2025-12-16: Add: tag.proto"),
     (163, "2025-12-31: Add: SnapshotRef"),
+    (164, "2026-01-22: Update: user.proto/CsvFileFormatParams add allow_quoted_nulls and quoted_empty_field_as, change null_display to optional", ),
     // Dear developer:
     //      If you're gonna add a new metadata version, you'll have to add a test for it.
     //      You could just copy an existing test file(e.g., `../tests/it/v024_table_meta.rs`)

src/meta/proto-conv/src/util.rs

@@ -155,3 +155,4 @@ mod v160_udf_update_on;
 mod v161_grant_object_row_access_policy;
 mod v162_tag;
 mod v163_snapshot_ref;
+mod v164_csv_format_params;

src/meta/proto-conv/tests/it/main.rs

@@ -14,6 +14,7 @@
 
 use databend_common_meta_app as mt;
 use databend_common_meta_app::principal::CsvFileFormatParams;
+use databend_common_meta_app::principal::EmptyFieldAs;
 use databend_common_meta_app::principal::NullAs;
 use databend_common_meta_app::principal::StageFileCompression;
 use databend_common_meta_app::principal::TsvFileFormatParams;
@@ -52,7 +53,9 @@ fn test_decode_v32_csv_file_format_params() -> anyhow::Result<()> {
             escape: "\\".to_string(),
             quote: "\'".to_string(),
             error_on_column_count_mismatch: true,
+            allow_quoted_nulls: false,
             empty_field_as: Default::default(),
+            quoted_empty_field_as: EmptyFieldAs::String,
             binary_format: Default::default(),
             geometry_format: Default::default(),
         })

src/meta/proto-conv/tests/it/v032_file_format_params.rs

@@ -14,6 +14,7 @@
 
 use databend_common_meta_app as mt;
 use databend_common_meta_app::principal::CsvFileFormatParams;
+use databend_common_meta_app::principal::EmptyFieldAs;
 use databend_common_meta_app::principal::StageFileCompression;
 use fastrace::func_name;
 
@@ -46,7 +47,9 @@ fn test_decode_v53_csv_file_format_params() -> anyhow::Result<()> {
             escape: "\\".to_string(),
             quote: "\'".to_string(),
             error_on_column_count_mismatch: true,
+            allow_quoted_nulls: false,
             empty_field_as: Default::default(),
+            quoted_empty_field_as: EmptyFieldAs::String,
             binary_format: Default::default(),
             geometry_format: Default::default(),
         })

src/meta/proto-conv/tests/it/v053_csv_format_params.rs

@@ -14,6 +14,7 @@
 
 use databend_common_meta_app as mt;
 use databend_common_meta_app::principal::CsvFileFormatParams;
+use databend_common_meta_app::principal::EmptyFieldAs;
 use databend_common_meta_app::principal::StageFileCompression;
 use fastrace::func_name;
 
@@ -46,7 +47,9 @@ fn test_decode_v59_csv_file_format_params() -> anyhow::Result<()> {
             escape: "\\".to_string(),
             quote: "\'".to_string(),
             error_on_column_count_mismatch: false,
+            allow_quoted_nulls: false,
             empty_field_as: Default::default(),
+            quoted_empty_field_as: EmptyFieldAs::String,
             binary_format: Default::default(),
             geometry_format: Default::default(),
         })

src/meta/proto-conv/tests/it/v059_csv_format_params.rs

@@ -48,7 +48,9 @@ fn test_decode_v72_csv_file_format_params() -> anyhow::Result<()> {
             escape: "|".to_string(),
             quote: "\'".to_string(),
             error_on_column_count_mismatch: false,
+            allow_quoted_nulls: false,
             empty_field_as: EmptyFieldAs::String,
+            quoted_empty_field_as: EmptyFieldAs::String,
             binary_format: Default::default(),
             geometry_format: Default::default(),
         })