feat(format): implement format trait

common/datavalues/src/types/deserializations/boolean.rs

@@ -49,6 +49,14 @@ impl TypeDeserializer for BooleanDeserializer {
         Ok(())
     }
 
+    fn de_json(&mut self, value: &serde_json::Value, _format: &FormatSettings) -> Result<()> {
+        match value {
+            serde_json::Value::Bool(v) => self.builder.append_value(*v),
+            _ => return Err(ErrorCode::BadBytes("Incorrect boolean value")),
+        }
+        Ok(())
+    }
+
     fn de_whole_text(&mut self, reader: &[u8], _format: &FormatSettings) -> Result<()> {
         if reader.eq_ignore_ascii_case(b"true") {
             self.builder.append_value(true);
@@ -77,14 +85,6 @@ impl TypeDeserializer for BooleanDeserializer {
         Ok(())
     }
 
-    fn de_json(&mut self, value: &serde_json::Value, _format: &FormatSettings) -> Result<()> {
-        match value {
-            serde_json::Value::Bool(v) => self.builder.append_value(*v),
-            _ => return Err(ErrorCode::BadBytes("Incorrect boolean value")),
-        }
-        Ok(())
-    }
-
     fn append_data_value(&mut self, value: DataValue, _format: &FormatSettings) -> Result<()> {
         self.builder.append_value(value.as_bool()?);
         Ok(())

common/datavalues/src/types/deserializations/number.rs

@@ -71,6 +71,10 @@ where
         }
     }
 
+    fn de_null(&mut self, _format: &FormatSettings) -> bool {
+        false
+    }
+
     fn de_whole_text(&mut self, reader: &[u8], _format: &FormatSettings) -> Result<()> {
         let mut reader = BufferReader::new(reader);
         let v: T = if !T::FLOATING {
@@ -98,8 +102,30 @@ where
         Ok(())
     }
 
-    fn de_null(&mut self, _format: &FormatSettings) -> bool {
-        false
+    fn de_text_csv<R: BufferRead>(
+        &mut self,
+        reader: &mut CheckpointReader<R>,
+        settings: &FormatSettings,
+    ) -> Result<()> {
+        let maybe_quote = reader.ignore(|f| f == b'\'' || f == b'"')?;
+
+        if maybe_quote && reader.ignore(|f| f == b'\'' || f == b'"')? && settings.empty_as_default {
+            self.de_default(settings);
+            return Ok(());
+        }
+
+        let v: T = if !T::FLOATING {
+            reader.read_int_text()
+        } else {
+            reader.read_float_text()
+        }?;
+
+        if maybe_quote {
+            reader.must_ignore(|f| f == b'\'' || f == b'"')?;
+        }
+
+        self.builder.append_value(v);
+        Ok(())
     }
 
     fn append_data_value(&mut self, value: DataValue, _format: &FormatSettings) -> Result<()> {

common/datavalues/src/types/deserializations/string.rs

@@ -79,29 +79,48 @@ impl TypeDeserializer for StringDeserializer {
         }
     }
 
-    fn de_text_quoted<R: BufferRead>(
+    fn de_whole_text(&mut self, reader: &[u8], _format: &FormatSettings) -> Result<()> {
+        self.builder.append_value(reader);
+        Ok(())
+    }
+
+    fn de_text<R: BufferRead>(
         &mut self,
         reader: &mut CheckpointReader<R>,
         _format: &FormatSettings,
     ) -> Result<()> {
         self.buffer.clear();
-        reader.read_quoted_text(&mut self.buffer, b'\'')?;
+        reader.read_escaped_string_text(&mut self.buffer)?;
         self.builder.append_value(self.buffer.as_slice());
         Ok(())
     }
 
-    fn de_whole_text(&mut self, reader: &[u8], _format: &FormatSettings) -> Result<()> {
-        self.builder.append_value(reader);
+    fn de_text_quoted<R: BufferRead>(
+        &mut self,
+        reader: &mut CheckpointReader<R>,
+        _format: &FormatSettings,
+    ) -> Result<()> {
+        self.buffer.clear();
+        reader.read_quoted_text(&mut self.buffer, b'\'')?;
+        self.builder.append_value(self.buffer.as_slice());
         Ok(())
     }
 
-    fn de_text<R: BufferRead>(
+    fn de_text_csv<R: BufferRead>(
         &mut self,
         reader: &mut CheckpointReader<R>,
-        _format: &FormatSettings,
+        format: &FormatSettings,
     ) -> Result<()> {
         self.buffer.clear();
-        reader.read_escaped_string_text(&mut self.buffer)?;
+        if reader.ignore_byte(b'"')? {
+            reader.keep_read(&mut self.buffer, |b| b != b'"')?;
+            reader.must_ignore_byte(b'"')?;
+        } else if format.field_delimiter.is_empty() {
+            reader.keep_read(&mut self.buffer, |b| b != b',')?;
+        } else {
+            reader.keep_read(&mut self.buffer, |b| b != format.field_delimiter[0])?;
+        }
+
         self.builder.append_value(self.buffer.as_slice());
         Ok(())
     }

common/datavalues/src/types/deserializations/variant.rs

@@ -75,6 +75,12 @@ impl TypeDeserializer for VariantDeserializer {
         Ok(())
     }
 
+    fn de_whole_text(&mut self, reader: &[u8], _format: &FormatSettings) -> Result<()> {
+        let val = serde_json::from_slice(reader)?;
+        self.builder.append_value(val);
+        Ok(())
+    }
+
     fn de_text<R: BufferRead>(
         &mut self,
         reader: &mut CheckpointReader<R>,
@@ -87,19 +93,27 @@ impl TypeDeserializer for VariantDeserializer {
         Ok(())
     }
 
-    fn de_whole_text(&mut self, reader: &[u8], _format: &FormatSettings) -> Result<()> {
-        let val = serde_json::from_slice(reader)?;
+    fn de_text_quoted<R: BufferRead>(
+        &mut self,
+        reader: &mut CheckpointReader<R>,
+        _format: &FormatSettings,
+    ) -> Result<()> {
+        self.buffer.clear();
+        reader.read_quoted_text(&mut self.buffer, b'\'')?;
+
+        let val = serde_json::from_slice(self.buffer.as_slice())?;
+
         self.builder.append_value(val);
         Ok(())
     }
 
-    fn de_text_quoted<R: BufferRead>(
+    fn de_text_csv<R: BufferRead>(
         &mut self,
         reader: &mut CheckpointReader<R>,
         _format: &FormatSettings,
     ) -> Result<()> {
         self.buffer.clear();
-        reader.read_quoted_text(&mut self.buffer, b'\'')?;
+        reader.read_quoted_text(&mut self.buffer, b'"')?;
 
         let val = serde_json::from_slice(self.buffer.as_slice())?;
         self.builder.append_value(val);

common/exception/src/exception_code.rs

@@ -139,6 +139,8 @@ build_exceptions! {
     // Network error codes.
     NetworkRequestError(1073),
 
+    UnknownFormat(1074),
+
     // Tenant error codes.
     TenantIsEmpty(1101),
     IndexOutOfBounds(1102),

common/io/src/buffer/buffer_read_ext.rs

@@ -24,6 +24,7 @@ pub trait BufferReadExt: BufferRead {
     fn ignore_bytes(&mut self, bs: &[u8]) -> Result<bool>;
     fn ignore_insensitive_bytes(&mut self, bs: &[u8]) -> Result<bool>;
     fn ignore_white_spaces(&mut self) -> Result<bool>;
+    fn ignore_white_spaces_and_byte(&mut self, b: u8) -> Result<bool>;
     fn until(&mut self, delim: u8, buf: &mut Vec<u8>) -> Result<usize>;
 
     fn keep_read(&mut self, buf: &mut Vec<u8>, f: impl Fn(u8) -> bool) -> Result<usize>;
@@ -55,6 +56,11 @@ pub trait BufferReadExt: BufferRead {
         Ok(())
     }
 
+    fn eof(&mut self) -> Result<bool> {
+        let buffer = self.fill_buf()?;
+        Ok(buffer.is_empty())
+    }
+
     fn must_eof(&mut self) -> Result<()> {
         let buffer = self.fill_buf()?;
         if !buffer.is_empty() {
@@ -78,9 +84,21 @@ pub trait BufferReadExt: BufferRead {
 
     fn must_ignore_byte(&mut self, b: u8) -> Result<()> {
         if !self.ignore_byte(b)? {
+            let buf = self.fill_buf()?;
             return Err(std::io::Error::new(
                 ErrorKind::InvalidData,
-                format!("Expected to have char {}", b as char),
+                format!("Expected to have char {}, {}", b as char, buf[0] as char),
+            ));
+        }
+        Ok(())
+    }
+
+    fn must_ignore_white_spaces_and_byte(&mut self, b: u8) -> Result<()> {
+        if !self.ignore_white_spaces_and_byte(b)? {
+            let buf = self.fill_buf()?;
+            return Err(std::io::Error::new(
+                ErrorKind::InvalidData,
+                format!("Expected to have char {}, {}", b as char, buf[0] as char),
             ));
         }
         Ok(())
@@ -108,7 +126,7 @@ pub trait BufferReadExt: BufferRead {
 }
 
 impl<R> BufferReadExt for R
-where R: BufferRead
+    where R: BufferRead
 {
     fn ignores(&mut self, f: impl Fn(u8) -> bool) -> Result<usize> {
         let mut bytes = 0;
@@ -172,6 +190,17 @@ where R: BufferRead
         Ok(cnt > 0)
     }
 
+    fn ignore_white_spaces_and_byte(&mut self, b: u8) -> Result<bool> {
+        self.ignores(|c: u8| c == b' ')?;
+
+        if self.ignore_byte(b)? {
+            self.ignores(|c: u8| c == b' ')?;
+            return Ok(true);
+        }
+
+        Ok(false)
+    }
+
     fn until(&mut self, delim: u8, buf: &mut Vec<u8>) -> Result<usize> {
         self.read_until(delim, buf)
     }

query/src/formats/format.rs

@@ -0,0 +1,36 @@
+// Copyright 2022 Datafuse Labs.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::any::Any;
+
+use common_datablocks::DataBlock;
+use common_exception::Result;
+
+pub trait InputState: Send {
+    fn as_any(&mut self) -> &mut dyn Any;
+}
+
+pub trait InputFormat: Send {
+    fn support_parallel(&self) -> bool {
+        false
+    }
+
+    fn create_state(&self) -> Box<dyn InputState>;
+
+    fn deserialize_data(&self, state: &mut Box<dyn InputState>) -> Result<DataBlock>;
+
+    fn read_buf(&self, buf: &[u8], state: &mut Box<dyn InputState>) -> Result<usize>;
+
+    fn skip_header(&self, buf: &[u8], state: &mut Box<dyn InputState>) -> Result<usize>;
+}