feat(query): unify pipeline for all inputs with format.

src/query/formats/src/lib.rs

@@ -27,4 +27,5 @@ mod output_format_values;
 
 pub use format::InputFormat;
 pub use format::InputState;
+pub use format_diagnostic::verbose_string;
 pub use format_factory::FormatFactory;

src/query/pipeline/sources/Cargo.toml

@@ -9,19 +9,29 @@ doctest = false
 test = false
 
 [dependencies]
+async-channel = "1.7.1"
+common-arrow = { path = "../../../common/arrow" }
 common-base = { path = "../../../common/base" }
 common-catalog = { path = "../../catalog" }
 common-datablocks = { path = "../../datablocks" }
+common-datavalues = { path = "../../datavalues" }
 common-exception = { path = "../../../common/exception" }
 common-formats = { path = "../../formats" }
 common-io = { path = "../../../common/io" }
 common-meta-types = { path = "../../../meta/types" }
 common-pipeline-core = { path = "../core" }
+common-settings = { path = "../../settings" }
 common-storage = { path = "../../../common/storage" }
 common-streams = { path = "../../streams" }
 
 async-trait = { version = "0.1.0", package = "async-trait-fn" }
+bstr = "0.2.17"
+crossbeam-channel = "0.5.6"
+csv-core = "0.1.10"
 futures = "0.3.21"
 futures-util = "0.3.21"
 opendal = { version = "0.17.1", features = ["layers-retry", "compress"] }
 parking_lot = "0.12.1"
+serde_json = "1.0.81"
+similar-asserts = "1.2.0"
+tracing = "0.1.35"

src/query/pipeline/sources/src/processors/sources/input_formats/delimiter.rs

@@ -0,0 +1,57 @@
+//  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 common_exception::ErrorCode;
+use common_exception::Result;
+
+pub enum RecordDelimiter {
+    Crlf,
+    Any(u8),
+}
+
+impl RecordDelimiter {
+    pub fn end(&self) -> u8 {
+        match self {
+            RecordDelimiter::Crlf => b'\n',
+            RecordDelimiter::Any(b) => *b,
+        }
+    }
+}
+
+impl TryFrom<&str> for RecordDelimiter {
+    type Error = ErrorCode;
+    fn try_from(s: &str) -> Result<Self> {
+        Self::try_from(s.as_bytes())
+    }
+}
+
+impl TryFrom<&[u8]> for RecordDelimiter {
+    type Error = ErrorCode;
+    fn try_from(s: &[u8]) -> Result<Self> {
+        match s.len() {
+            1 => Ok(RecordDelimiter::Any(s[0])),
+            2 if s.eq(b"\r\n") => Ok(RecordDelimiter::Crlf),
+            _ => Err(ErrorCode::InvalidArgument(format!(
+                "bad RecordDelimiter: '{:?}'",
+                s
+            ))),
+        }
+    }
+}
+
+impl Default for RecordDelimiter {
+    fn default() -> Self {
+        RecordDelimiter::Crlf
+    }
+}