feat(query): Standardize Nested Type String Quoting and Variant Serialization for JSON Compatibility

src/query/formats/Cargo.toml

@@ -34,6 +34,7 @@ match-template = { workspace = true }
 micromarshal = { workspace = true }
 num = { workspace = true }
 num-traits = { workspace = true }
+serde = { workspace = true }
 serde_json = { workspace = true }
 
 [dev-dependencies]

src/query/formats/src/field_decoder/nested.rs

@@ -56,6 +56,9 @@ use databend_common_io::parse_bitmap;
 use databend_common_io::parse_bytes_to_ewkb;
 use jsonb::parse_owned_jsonb_with_buf;
 use lexical_core::FromLexical;
+use serde::Deserialize;
+use serde_json::Deserializer;
+use serde_json::value::RawValue;
 
 use crate::FileFormatOptionsExt;
 use crate::InputCommonSettings;
@@ -209,7 +212,7 @@ impl NestedValues {
         column: &mut StringColumnBuilder,
         reader: &mut Cursor<R>,
     ) -> Result<()> {
-        reader.read_quoted_text(&mut column.row_buffer, b'\'')?;
+        self.read_string_inner(reader, &mut column.row_buffer)?;
         column.commit_row();
         Ok(())
     }
@@ -220,7 +223,7 @@ impl NestedValues {
         reader: &mut Cursor<R>,
     ) -> Result<()> {
         let mut buf = Vec::new();
-        reader.read_quoted_text(&mut buf, b'\'')?;
+        self.read_string_inner(reader, &mut buf)?;
         let decoded = decode_binary(&buf, self.common_settings.binary_format)?;
         column.put_slice(&decoded);
         column.commit_row();
@@ -232,7 +235,10 @@ impl NestedValues {
         reader: &mut Cursor<R>,
         out_buf: &mut Vec<u8>,
     ) -> Result<()> {
-        reader.read_quoted_text(out_buf, b'\'')?;
+        if reader.read_quoted_text(out_buf, b'"').is_err() {
+            // Read single quoted text, compatible with previous implementations
+            reader.read_quoted_text(out_buf, b'\'')?;
+        }
         Ok(())
     }
 
@@ -320,8 +326,13 @@ impl NestedValues {
         column: &mut BinaryColumnBuilder,
         reader: &mut Cursor<R>,
     ) -> Result<()> {
-        let mut buf = Vec::new();
-        self.read_string_inner(reader, &mut buf)?;
+        let buf = if let Ok(val) = self.read_json(reader) {
+            val.as_bytes().to_vec()
+        } else {
+            let mut buf = Vec::new();
+            reader.read_quoted_text(&mut buf, b'\'')?;
+            buf
+        };
         match parse_owned_jsonb_with_buf(&buf, &mut column.data) {
             Ok(_) => {
                 column.commit_row();
@@ -343,7 +354,12 @@ impl NestedValues {
         reader: &mut Cursor<R>,
     ) -> Result<()> {
         let mut buf = Vec::new();
-        self.read_string_inner(reader, &mut buf)?;
+        if reader.read_quoted_text(&mut buf, b'"').is_err()
+            && reader.read_quoted_text(&mut buf, b'\'').is_err()
+        {
+            let val = self.read_json(reader)?;
+            buf = val.as_bytes().to_vec();
+        }
         let geom = parse_bytes_to_ewkb(&buf, None)?;
         column.put_slice(geom.as_bytes());
         column.commit_row();
@@ -356,13 +372,27 @@ impl NestedValues {
         reader: &mut Cursor<R>,
     ) -> Result<()> {
         let mut buf = Vec::new();
-        self.read_string_inner(reader, &mut buf)?;
+        if reader.read_quoted_text(&mut buf, b'"').is_err()
+            && reader.read_quoted_text(&mut buf, b'\'').is_err()
+        {
+            let val = self.read_json(reader)?;
+            buf = val.as_bytes().to_vec();
+        }
         let geog = geography_from_ewkt_bytes(&buf)?;
         column.put_slice(geog.as_bytes());
         column.commit_row();
         Ok(())
     }
 
+    fn read_json<R: AsRef<[u8]>>(&self, reader: &mut Cursor<R>) -> Result<String> {
+        let start = reader.position() as usize;
+        let data = reader.get_ref().as_ref();
+        let mut deserializer = Deserializer::from_slice(&data[start..]);
+        let raw: Box<RawValue> = Box::<RawValue>::deserialize(&mut deserializer)?;
+        reader.set_position((start + raw.get().len()) as u64);
+        Ok(raw.to_string())
+    }
+
     fn read_nullable<R: AsRef<[u8]>>(
         &self,
         column: &mut NullableColumnBuilder<AnyType>,

src/query/formats/src/field_encoder/csv.rs

@@ -93,7 +93,8 @@ impl FieldEncoderCSV {
                     binary_format: params.binary_format,
                     geometry_format: params.geometry_format,
                 },
-                quote_char: 0, // not used
+                escape_char: 0, // not used
+                quote_char: 0,  // not used
             },
             string_formatter: StringFormatter::Csv {
                 quote_char: params.quote.as_bytes()[0],
@@ -116,7 +117,8 @@ impl FieldEncoderCSV {
                     binary_format: Default::default(),
                     geometry_format: Default::default(),
                 },
-                quote_char: 0, // not used
+                escape_char: 0, // not used
+                quote_char: 0,  // not used
             },
             string_formatter: StringFormatter::Tsv {
                 record_delimiter: params.field_delimiter.as_bytes().to_vec()[0],

src/query/formats/src/field_encoder/helpers/escape.rs

@@ -44,15 +44,15 @@ static ESCAPE: [u8; 256] = [
     __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, __, // F
 ];
 
-pub fn write_quoted_string(bytes: &[u8], buf: &mut Vec<u8>, quote: u8) {
+pub fn write_quoted_string(bytes: &[u8], buf: &mut Vec<u8>, escape: u8, quote: u8) {
     let mut start = 0;
 
     for (i, &byte) in bytes.iter().enumerate() {
         if byte == quote {
             if start < i {
                 buf.extend_from_slice(&bytes[start..i]);
             }
-            buf.push(quote);
+            buf.push(escape);
             buf.push(quote);
             start = i + 1;
         }

src/query/formats/src/field_encoder/json.rs

@@ -50,6 +50,7 @@ impl FieldEncoderJSON {
                     binary_format: Default::default(),
                     geometry_format: Default::default(),
                 },
+                escape_char: 0,
                 quote_char: 0,
             },
             quote_denormals: false,

src/query/formats/src/field_encoder/values.rs

@@ -61,6 +61,7 @@ use crate::field_encoder::helpers::write_quoted_string;
 
 pub struct FieldEncoderValues {
     pub common_settings: OutputCommonSettings,
+    pub escape_char: u8,
     pub quote_char: u8,
 }
 
@@ -78,7 +79,8 @@ impl FieldEncoderValues {
                 binary_format: Default::default(),
                 geometry_format: Default::default(),
             },
-            quote_char: b'\'',
+            escape_char: b'"',
+            quote_char: b'"',
         }
     }
 
@@ -99,7 +101,8 @@ impl FieldEncoderValues {
                 binary_format: Default::default(),
                 geometry_format,
             },
-            quote_char: b'\'',
+            escape_char: b'\\',
+            quote_char: b'"',
         }
     }
 
@@ -124,7 +127,8 @@ impl FieldEncoderValues {
                 binary_format: Default::default(),
                 geometry_format,
             },
-            quote_char: b'\'',
+            escape_char: b'\\',
+            quote_char: b'"',
         }
     }
 
@@ -163,7 +167,7 @@ impl FieldEncoderValues {
             Column::Timestamp(c) => self.write_timestamp(c, row_index, out_buf, in_nested),
             Column::TimestampTz(c) => self.write_timestamp_tz(c, row_index, out_buf, in_nested),
             Column::Bitmap(b) => self.write_bitmap(b, row_index, out_buf, in_nested),
-            Column::Variant(c) => self.write_variant(c, row_index, out_buf, in_nested),
+            Column::Variant(c) => self.write_variant(c, row_index, out_buf),
             Column::Geometry(c) => self.write_geometry(c, row_index, out_buf, in_nested),
             Column::Geography(c) => self.write_geography(c, row_index, out_buf, in_nested),
 
@@ -186,12 +190,13 @@ impl FieldEncoderValues {
             // so we do not expect the scalar literal to be used in sql.
             // it is better to keep it simple: minimal escape.
             // it make result easier to decode csv, tsv and http handler result.
-            write_quoted_string(in_buf, out_buf, self.quote_char);
+            write_quoted_string(in_buf, out_buf, self.escape_char, self.quote_char);
             out_buf.push(self.quote_char);
         } else {
             out_buf.extend_from_slice(in_buf);
         }
     }
+
     fn write_bool(&self, column: &Bitmap, row_index: usize, out_buf: &mut Vec<u8>) {
         let v = if column.get_bit(row_index) {
             &self.common_settings().true_bytes
@@ -328,16 +333,10 @@ impl FieldEncoderValues {
         self.write_string_inner(bitmap_result, out_buf, in_nested);
     }
 
-    fn write_variant(
-        &self,
-        column: &BinaryColumn,
-        row_index: usize,
-        out_buf: &mut Vec<u8>,
-        in_nested: bool,
-    ) {
+    fn write_variant(&self, column: &BinaryColumn, row_index: usize, out_buf: &mut Vec<u8>) {
         let v = unsafe { column.index_unchecked(row_index) };
         let s = RawJsonb::new(v).to_string();
-        self.write_string_inner(s.as_bytes(), out_buf, in_nested);
+        out_buf.extend_from_slice(s.as_bytes());
     }
 
     fn write_geometry(
@@ -360,7 +359,14 @@ impl FieldEncoderValues {
             })
             .unwrap_or_else(|_| v.to_vec());
 
-        self.write_string_inner(&s, out_buf, in_nested);
+        match self.common_settings().geometry_format {
+            GeometryDataType::GEOJSON => {
+                out_buf.extend_from_slice(&s);
+            }
+            _ => {
+                self.write_string_inner(&s, out_buf, in_nested);
+            }
+        }
     }
 
     fn write_geography(
@@ -383,7 +389,14 @@ impl FieldEncoderValues {
             })
             .unwrap_or_else(|_| v.0.to_vec());
 
-        self.write_string_inner(&s, out_buf, in_nested);
+        match self.common_settings().geometry_format {
+            GeometryDataType::GEOJSON => {
+                out_buf.extend_from_slice(&s);
+            }
+            _ => {
+                self.write_string_inner(&s, out_buf, in_nested);
+            }
+        }
     }
 
     fn write_array(&self, column: &ArrayColumn<AnyType>, row_index: usize, out_buf: &mut Vec<u8>) {

src/tests/sqlsmith/src/sql_gen/dml.rs

@@ -579,7 +579,8 @@ impl<'a, R: Rng + 'a> SqlGenerator<'a, R> {
                         binary_format: Default::default(),
                         geometry_format: Default::default(),
                     },
-                    quote_char: b'\'',
+                    escape_char: b'\\',
+                    quote_char: b'"',
                 };
 
                 for i in 0..row_count {

tests/nox/noxfile.py

@@ -2,7 +2,7 @@
 import os
 
 
-PYTHON_DRIVER = ["0.28.1", "0.28.2"]
+PYTHON_DRIVER = ["0.33.1"]
 
 
 @nox.session

tests/nox/python_client/test_data_type.py

@@ -1,4 +1,4 @@
-from datetime import datetime, date, timedelta
+from datetime import datetime, date, timedelta, timezone
 from decimal import Decimal
 from databend_driver import BlockingDatabendClient
 
@@ -37,6 +37,6 @@ def test_data_types():
 
     # Tuple
     row = conn.query_row("select (10, '20', to_datetime('2024-04-16 12:34:56.789'))")
-    assert row.values() == ((10, "20", datetime(2024, 4, 16, 12, 34, 56, 789000)),), (
+    assert row.values() == ((10, "20", datetime(2024, 4, 16, 12, 34, 56, 789000, tzinfo=timezone.utc)),), (
         f"Tuple: {row.values()}"
     )

tests/nox/python_client/test_local.py

@@ -1,5 +1,5 @@
 from databend_driver import BlockingDatabendClient
-from datetime import datetime, date
+from datetime import datetime, date, timezone
 
 from .utils import DATABEND_DSL
 
@@ -31,9 +31,9 @@ def test_insert():
     rows = conn.query_iter("SELECT * FROM test")
     ret = [row.values() for row in rows]
     expected = [
-        (-1, 1, 1.0, "'", None, date(2011, 3, 6), datetime(2011, 3, 6, 6, 20)),
-        (-2, 2, 2.0, '"', "", date(2012, 5, 31), datetime(2012, 5, 31, 11, 20)),
-        (-3, 3, 3.0, "\\", "NULL", date(2016, 4, 4), datetime(2016, 4, 4, 11, 30)),
+        (-1, 1, 1.0, "'", None, date(2011, 3, 6), datetime(2011, 3, 6, 6, 20, tzinfo=timezone.utc)),
+        (-2, 2, 2.0, '"', "", date(2012, 5, 31), datetime(2012, 5, 31, 11, 20, tzinfo=timezone.utc)),
+        (-3, 3, 3.0, '\\', "NULL", date(2016, 4, 4), datetime(2016, 4, 4, 11, 30, tzinfo=timezone.utc))
     ]
     assert ret == expected, f"ret: {ret}"
 
@@ -66,9 +66,9 @@ def test_stream_load():
     rows = conn.query_iter("SELECT * FROM test")
     ret = [row.values() for row in rows]
     expected = [
-        (-1, 1, 1.0, "'", None, date(2011, 3, 6), datetime(2011, 3, 6, 6, 20)),
-        (-2, 2, 2.0, '"', None, date(2012, 5, 31), datetime(2012, 5, 31, 11, 20)),
-        (-3, 3, 3.0, "\\", "NULL", date(2016, 4, 4), datetime(2016, 4, 4, 11, 30)),
+        (-1, 1, 1.0, "'", None, date(2011, 3, 6), datetime(2011, 3, 6, 6, 20, tzinfo=timezone.utc)),
+        (-2, 2, 2.0, '"', None, date(2012, 5, 31), datetime(2012, 5, 31, 11, 20, tzinfo=timezone.utc)),
+        (-3, 3, 3.0, "\\", "NULL", date(2016, 4, 4), datetime(2016, 4, 4, 11, 30, tzinfo=timezone.utc)),
     ]
     assert ret == expected, f"ret: {ret}"
 
@@ -111,8 +111,8 @@ def run_load_file(load_method):
     rows = conn.query_iter("SELECT * FROM test")
     ret = [row.values() for row in rows]
     expected = [
-        (-1, 1, 1.0, "'", None, date(2011, 3, 6), datetime(2011, 3, 6, 6, 20)),
-        (-2, 2, 2.0, '"', None, date(2012, 5, 31), datetime(2012, 5, 31, 11, 20)),
-        (-3, 3, 3.0, "\\", "NULL", date(2016, 4, 4), datetime(2016, 4, 4, 11, 30)),
+        (-1, 1, 1.0, "'", None, date(2011, 3, 6), datetime(2011, 3, 6, 6, 20, tzinfo=timezone.utc)),
+        (-2, 2, 2.0, '"', None, date(2012, 5, 31), datetime(2012, 5, 31, 11, 20, tzinfo=timezone.utc)),
+        (-3, 3, 3.0, "\\", "NULL", date(2016, 4, 4), datetime(2016, 4, 4, 11, 30, tzinfo=timezone.utc)),
     ]
     assert ret == expected, f"{load_method} ret: {ret}"

tests/sqllogictests/suites/base/03_common/03_0003_select_group_by.test

@@ -280,9 +280,9 @@ DROP TABLE t_str
 query TTT rowsort
 SELECT ( null, to_hour(to_timestamp(3501857592331)), number::Date) from numbers(3) group by all
 ----
-(NULL,18,'1970-01-01')
-(NULL,18,'1970-01-02')
-(NULL,18,'1970-01-03')
+(NULL,18,"1970-01-01")
+(NULL,18,"1970-01-02")
+(NULL,18,"1970-01-03")
 
 query TTT rowsort
 SELECT TRY_CAST('1900-12-30 12:00:00' AS TIMESTAMP) AS "TEMP(Test)(4058757556)(0)",