mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-28 22:56:19 +00:00
perf: introduce simd_json for parsing ndjson (#5794)
* perf: introduce simd_json for parsing ndjson * fix: some tests * fix: some tests * fix: es test case * chore: use `as_bytes_mut()` * chore: remove unnecessary `to_string` * chore: add safety comment
This commit is contained in:
shuiyisong
Generated
+67
-13
@@ -4091,6 +4091,15 @@ dependencies = [
|
||||
"miniz_oxide",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "float-cmp"
|
||||
version = "0.10.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b09cf3155332e944990140d967ff5eceb70df778b34f77d8075db46e4704e6d8"
|
||||
dependencies = [
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "float_eq"
|
||||
version = "1.0.1"
|
||||
@@ -4626,9 +4635,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "73fea8450eea4bac3940448fb7ae50d91f034f941199fcd9d909a5a07aa455f0"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"js-sys",
|
||||
"libc",
|
||||
"r-efi",
|
||||
"wasi 0.14.2+wasi-0.2.4",
|
||||
"wasm-bindgen",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -4750,6 +4761,16 @@ dependencies = [
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "halfbrown"
|
||||
version = "0.3.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "aa2c385c6df70fd180bbb673d93039dbd2cd34e41d782600bdf6e1ca7bce39aa"
|
||||
dependencies = [
|
||||
"hashbrown 0.15.2",
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hash32"
|
||||
version = "0.3.1"
|
||||
@@ -5815,10 +5836,11 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "js-sys"
|
||||
version = "0.3.70"
|
||||
version = "0.3.77"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1868808506b929d7b0cfa8f75951347aa71bb21144b7791bae35d9bccfcfe37a"
|
||||
checksum = "1cfaf33c695fc6e08064efbc1f72ec937429614f25eef83af942d0e227c3a28f"
|
||||
dependencies = [
|
||||
"once_cell",
|
||||
"wasm-bindgen",
|
||||
]
|
||||
|
||||
@@ -8385,6 +8407,7 @@ dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
"session",
|
||||
"simd-json",
|
||||
"snafu 0.8.5",
|
||||
"sql",
|
||||
"table",
|
||||
@@ -10592,6 +10615,7 @@ dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
"session",
|
||||
"simd-json",
|
||||
"snafu 0.8.5",
|
||||
"snap",
|
||||
"socket2",
|
||||
@@ -10730,6 +10754,21 @@ dependencies = [
|
||||
"wide",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "simd-json"
|
||||
version = "0.15.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "10b5602e4f1f7d358956f94cac1eff59220f34cf9e26d49f5fde5acef851cbed"
|
||||
dependencies = [
|
||||
"getrandom 0.3.2",
|
||||
"halfbrown",
|
||||
"ref-cast",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"simdutf8",
|
||||
"value-trait",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "simdutf8"
|
||||
version = "0.1.5"
|
||||
@@ -13194,6 +13233,18 @@ version = "0.1.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "830b7e5d4d90034032940e4ace0d9a9a057e7a45cd94e6c007832e39edb82f6d"
|
||||
|
||||
[[package]]
|
||||
name = "value-trait"
|
||||
version = "0.11.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0508fce11ad19e0aab49ce20b6bec7f8f82902ded31df1c9fc61b90f0eb396b8"
|
||||
dependencies = [
|
||||
"float-cmp",
|
||||
"halfbrown",
|
||||
"itoa",
|
||||
"ryu",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "variadics"
|
||||
version = "0.0.4"
|
||||
@@ -13284,24 +13335,24 @@ checksum = "b8dad83b4f25e74f184f64c43b150b91efe7647395b42289f38e50566d82855b"
|
||||
|
||||
[[package]]
|
||||
name = "wasm-bindgen"
|
||||
version = "0.2.93"
|
||||
version = "0.2.100"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a82edfc16a6c469f5f44dc7b571814045d60404b55a0ee849f9bcfa2e63dd9b5"
|
||||
checksum = "1edc8929d7499fc4e8f0be2262a241556cfc54a0bea223790e71446f2aab1ef5"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"once_cell",
|
||||
"rustversion",
|
||||
"wasm-bindgen-macro",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "wasm-bindgen-backend"
|
||||
version = "0.2.93"
|
||||
version = "0.2.100"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9de396da306523044d3302746f1208fa71d7532227f15e347e2d93e4145dd77b"
|
||||
checksum = "2f0a0651a5c2bc21487bde11ee802ccaf4c51935d0d3d42a6101f98161700bc6"
|
||||
dependencies = [
|
||||
"bumpalo",
|
||||
"log",
|
||||
"once_cell",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.96",
|
||||
@@ -13322,9 +13373,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "wasm-bindgen-macro"
|
||||
version = "0.2.93"
|
||||
version = "0.2.100"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "585c4c91a46b072c92e908d99cb1dcdf95c5218eeb6f3bf1efa991ee7a68cccf"
|
||||
checksum = "7fe63fc6d09ed3792bd0897b314f53de8e16568c2b3f7982f468c0bf9bd0b407"
|
||||
dependencies = [
|
||||
"quote",
|
||||
"wasm-bindgen-macro-support",
|
||||
@@ -13332,9 +13383,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "wasm-bindgen-macro-support"
|
||||
version = "0.2.93"
|
||||
version = "0.2.100"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "afc340c74d9005395cf9dd098506f7f44e38f2b4a21c6aaacf9a105ea5e1e836"
|
||||
checksum = "8ae87ea40c9f689fc23f209965b6fb8a99ad69aeeb0231408be24920604395de"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
@@ -13345,9 +13396,12 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "wasm-bindgen-shared"
|
||||
version = "0.2.93"
|
||||
version = "0.2.100"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c62a0a307cb4a311d3a07867860911ca130c3494e8c2719593806c08bc5d0484"
|
||||
checksum = "1a05d73b933a847d6cccdda8f838a22ff101ad9bf93e33684f39c1f5f0eece3d"
|
||||
dependencies = [
|
||||
"unicode-ident",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "wasm-streams"
|
||||
|
||||
@@ -185,6 +185,7 @@ serde = { version = "1.0", features = ["derive"] }
|
||||
serde_json = { version = "1.0", features = ["float_roundtrip"] }
|
||||
serde_with = "3"
|
||||
shadow-rs = "1.1"
|
||||
simd-json = "0.15"
|
||||
similar-asserts = "1.6.0"
|
||||
smallvec = { version = "1", features = ["serde"] }
|
||||
snafu = "0.8"
|
||||
|
||||
@@ -53,6 +53,7 @@ query.workspace = true
|
||||
regex.workspace = true
|
||||
serde_json.workspace = true
|
||||
session.workspace = true
|
||||
simd-json.workspace = true
|
||||
snafu.workspace = true
|
||||
sql.workspace = true
|
||||
table.workspace = true
|
||||
|
||||
@@ -170,6 +170,23 @@ pub fn json_array_to_map(val: Vec<serde_json::Value>) -> Result<Vec<PipelineMap>
|
||||
val.into_iter().map(json_to_map).collect()
|
||||
}
|
||||
|
||||
pub fn simd_json_to_map(val: simd_json::OwnedValue) -> Result<PipelineMap> {
|
||||
match val {
|
||||
simd_json::OwnedValue::Object(map) => {
|
||||
let mut intermediate_state = PipelineMap::new();
|
||||
for (k, v) in map.into_iter() {
|
||||
intermediate_state.insert(k, Value::try_from(v)?);
|
||||
}
|
||||
Ok(intermediate_state)
|
||||
}
|
||||
_ => PrepareValueMustBeObjectSnafu.fail(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn simd_json_array_to_map(val: Vec<simd_json::OwnedValue>) -> Result<Vec<PipelineMap>> {
|
||||
val.into_iter().map(simd_json_to_map).collect()
|
||||
}
|
||||
|
||||
impl Pipeline {
|
||||
pub fn exec_mut(&self, val: &mut PipelineMap) -> Result<PipelineExecOutput> {
|
||||
// process
|
||||
|
||||
@@ -367,6 +367,37 @@ impl std::fmt::Display for Value {
|
||||
}
|
||||
}
|
||||
|
||||
impl TryFrom<simd_json::value::OwnedValue> for Value {
|
||||
type Error = Error;
|
||||
|
||||
fn try_from(v: simd_json::value::OwnedValue) -> Result<Self> {
|
||||
match v {
|
||||
simd_json::value::OwnedValue::Static(v) => match v {
|
||||
simd_json::value::StaticNode::Null => Ok(Value::Null),
|
||||
simd_json::value::StaticNode::Bool(v) => Ok(Value::Boolean(v)),
|
||||
simd_json::value::StaticNode::I64(v) => Ok(Value::Int64(v)),
|
||||
simd_json::value::StaticNode::U64(v) => Ok(Value::Uint64(v)),
|
||||
simd_json::value::StaticNode::F64(v) => Ok(Value::Float64(v)),
|
||||
},
|
||||
simd_json::OwnedValue::String(s) => Ok(Value::String(s)),
|
||||
simd_json::OwnedValue::Array(values) => {
|
||||
let mut re = Vec::with_capacity(values.len());
|
||||
for v in values.into_iter() {
|
||||
re.push(Value::try_from(v)?);
|
||||
}
|
||||
Ok(Value::Array(Array { values: re }))
|
||||
}
|
||||
simd_json::OwnedValue::Object(map) => {
|
||||
let mut values = PipelineMap::new();
|
||||
for (k, v) in map.into_iter() {
|
||||
values.insert(k, Value::try_from(v)?);
|
||||
}
|
||||
Ok(Value::Map(Map { values }))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl TryFrom<serde_json::Value> for Value {
|
||||
type Error = Error;
|
||||
|
||||
|
||||
@@ -25,8 +25,8 @@ pub use etl::transform::transformer::identity_pipeline;
|
||||
pub use etl::transform::GreptimeTransformer;
|
||||
pub use etl::value::{Array, Map, Value};
|
||||
pub use etl::{
|
||||
json_array_to_map, json_to_map, parse, Content, DispatchedTo, Pipeline, PipelineExecOutput,
|
||||
PipelineMap,
|
||||
json_array_to_map, json_to_map, parse, simd_json_array_to_map, simd_json_to_map, Content,
|
||||
DispatchedTo, Pipeline, PipelineExecOutput, PipelineMap,
|
||||
};
|
||||
pub use manager::{
|
||||
pipeline_operator, table, util, IdentityTimeIndex, PipelineDefinition, PipelineInfo,
|
||||
|
||||
@@ -77,6 +77,7 @@ notify.workspace = true
|
||||
object-pool = "0.5"
|
||||
once_cell.workspace = true
|
||||
openmetrics-parser = "0.4"
|
||||
simd-json.workspace = true
|
||||
socket2 = "0.5"
|
||||
# use crates.io version after current revision is merged in next release
|
||||
# opensrv-mysql = "0.7.0"
|
||||
|
||||
@@ -30,7 +30,7 @@ use session::context::{Channel, QueryContext};
|
||||
use snafu::{ensure, ResultExt};
|
||||
|
||||
use crate::error::{
|
||||
status_code_to_http_status, InvalidElasticsearchInputSnafu, ParseJsonSnafu,
|
||||
status_code_to_http_status, InvalidElasticsearchInputSnafu, ParseJsonSnafu, PipelineSnafu,
|
||||
Result as ServersResult,
|
||||
};
|
||||
use crate::http::event::{ingest_logs_inner, LogIngestRequest, LogIngesterQueryParams, LogState};
|
||||
@@ -330,6 +330,7 @@ fn parse_bulk_request(
|
||||
}
|
||||
);
|
||||
|
||||
let log_value = pipeline::json_to_map(log_value).context(PipelineSnafu)?;
|
||||
requests.push(LogIngestRequest {
|
||||
table: index.unwrap_or_else(|| index_from_url.as_ref().unwrap().clone()),
|
||||
values: vec![log_value],
|
||||
@@ -371,8 +372,8 @@ fn get_log_value_from_msg_field(mut v: Value, msg_field: &str) -> Value {
|
||||
match message {
|
||||
Value::String(s) => match serde_json::from_str::<Value>(&s) {
|
||||
Ok(s) => s,
|
||||
// If the message is not a valid JSON, just use the original message as the log value.
|
||||
Err(_) => Value::String(s),
|
||||
// If the message is not a valid JSON, return a map with the original message key and value.
|
||||
Err(_) => json!({msg_field: s}),
|
||||
},
|
||||
// If the message is not a string, just use the original message as the log value.
|
||||
_ => message,
|
||||
@@ -404,12 +405,12 @@ mod tests {
|
||||
LogIngestRequest {
|
||||
table: "test".to_string(),
|
||||
values: vec![
|
||||
json!({"foo1": "foo1_value", "bar1": "bar1_value"}),
|
||||
pipeline::json_to_map(json!({"foo1": "foo1_value", "bar1": "bar1_value"})).unwrap(),
|
||||
],
|
||||
},
|
||||
LogIngestRequest {
|
||||
table: "test".to_string(),
|
||||
values: vec![json!({"foo2": "foo2_value", "bar2": "bar2_value"})],
|
||||
values: vec![pipeline::json_to_map(json!({"foo2": "foo2_value", "bar2": "bar2_value"})).unwrap()],
|
||||
},
|
||||
]),
|
||||
),
|
||||
@@ -426,11 +427,11 @@ mod tests {
|
||||
Ok(vec![
|
||||
LogIngestRequest {
|
||||
table: "test".to_string(),
|
||||
values: vec![json!({"foo1": "foo1_value", "bar1": "bar1_value"})],
|
||||
values: vec![pipeline::json_to_map(json!({"foo1": "foo1_value", "bar1": "bar1_value"})).unwrap()],
|
||||
},
|
||||
LogIngestRequest {
|
||||
table: "logs".to_string(),
|
||||
values: vec![json!({"foo2": "foo2_value", "bar2": "bar2_value"})],
|
||||
values: vec![pipeline::json_to_map(json!({"foo2": "foo2_value", "bar2": "bar2_value"})).unwrap()],
|
||||
},
|
||||
]),
|
||||
),
|
||||
@@ -447,11 +448,11 @@ mod tests {
|
||||
Ok(vec![
|
||||
LogIngestRequest {
|
||||
table: "test".to_string(),
|
||||
values: vec![json!({"foo1": "foo1_value", "bar1": "bar1_value"})],
|
||||
values: vec![pipeline::json_to_map(json!({"foo1": "foo1_value", "bar1": "bar1_value"})).unwrap()],
|
||||
},
|
||||
LogIngestRequest {
|
||||
table: "logs".to_string(),
|
||||
values: vec![json!({"foo2": "foo2_value", "bar2": "bar2_value"})],
|
||||
values: vec![pipeline::json_to_map(json!({"foo2": "foo2_value", "bar2": "bar2_value"})).unwrap()],
|
||||
},
|
||||
]),
|
||||
),
|
||||
@@ -467,7 +468,7 @@ mod tests {
|
||||
Ok(vec![
|
||||
LogIngestRequest {
|
||||
table: "test".to_string(),
|
||||
values: vec![json!({"foo1": "foo1_value", "bar1": "bar1_value"})],
|
||||
values: vec![pipeline::json_to_map(json!({"foo1": "foo1_value", "bar1": "bar1_value"})).unwrap()],
|
||||
},
|
||||
]),
|
||||
),
|
||||
@@ -484,11 +485,11 @@ mod tests {
|
||||
Ok(vec![
|
||||
LogIngestRequest {
|
||||
table: "test".to_string(),
|
||||
values: vec![json!({"foo1": "foo1_value", "bar1": "bar1_value"})],
|
||||
values: vec![pipeline::json_to_map(json!({"foo1": "foo1_value", "bar1": "bar1_value"})).unwrap()],
|
||||
},
|
||||
LogIngestRequest {
|
||||
table: "test".to_string(),
|
||||
values: vec![json!({"foo2": "foo2_value", "bar2": "bar2_value"})],
|
||||
values: vec![pipeline::json_to_map(json!({"foo2": "foo2_value", "bar2": "bar2_value"})).unwrap()],
|
||||
},
|
||||
]),
|
||||
),
|
||||
@@ -506,13 +507,13 @@ mod tests {
|
||||
LogIngestRequest {
|
||||
table: "logs-generic-default".to_string(),
|
||||
values: vec![
|
||||
json!("172.16.0.1 - - [25/May/2024:20:19:37 +0000] \"GET /contact HTTP/1.1\" 404 162 \"-\" \"Mozilla/5.0 (iPhone; CPU iPhone OS 14_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0 Mobile/15E148 Safari/604.1\""),
|
||||
pipeline::json_to_map(json!({"message": "172.16.0.1 - - [25/May/2024:20:19:37 +0000] \"GET /contact HTTP/1.1\" 404 162 \"-\" \"Mozilla/5.0 (iPhone; CPU iPhone OS 14_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0 Mobile/15E148 Safari/604.1\""})).unwrap(),
|
||||
],
|
||||
},
|
||||
LogIngestRequest {
|
||||
table: "logs-generic-default".to_string(),
|
||||
values: vec![
|
||||
json!("10.0.0.1 - - [25/May/2024:20:18:37 +0000] \"GET /images/logo.png HTTP/1.1\" 304 0 \"-\" \"Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:89.0) Gecko/20100101 Firefox/89.0\""),
|
||||
pipeline::json_to_map(json!({"message": "10.0.0.1 - - [25/May/2024:20:18:37 +0000] \"GET /images/logo.png HTTP/1.1\" 304 0 \"-\" \"Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:89.0) Gecko/20100101 Firefox/89.0\""})).unwrap(),
|
||||
],
|
||||
},
|
||||
]),
|
||||
|
||||
@@ -33,7 +33,7 @@ use datatypes::value::column_data_to_json;
|
||||
use headers::ContentType;
|
||||
use lazy_static::lazy_static;
|
||||
use pipeline::util::to_pipeline_version;
|
||||
use pipeline::{GreptimePipelineParams, PipelineDefinition};
|
||||
use pipeline::{GreptimePipelineParams, PipelineDefinition, PipelineMap};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::{json, Deserializer, Map, Value};
|
||||
use session::context::{Channel, QueryContext, QueryContextRef};
|
||||
@@ -99,12 +99,12 @@ pub struct LogIngesterQueryParams {
|
||||
|
||||
/// LogIngestRequest is the internal request for log ingestion. The raw log input can be transformed into multiple LogIngestRequests.
|
||||
/// Multiple LogIngestRequests will be ingested into the same database with the same pipeline.
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
#[derive(Debug, PartialEq)]
|
||||
pub(crate) struct LogIngestRequest {
|
||||
/// The table where the log data will be written to.
|
||||
pub table: String,
|
||||
/// The log data to be ingested.
|
||||
pub values: Vec<Value>,
|
||||
pub values: Vec<PipelineMap>,
|
||||
}
|
||||
|
||||
pub struct PipelineContent(String);
|
||||
@@ -281,7 +281,7 @@ fn transform_ndjson_array_factory(
|
||||
|
||||
/// Dryrun pipeline with given data
|
||||
async fn dryrun_pipeline_inner(
|
||||
value: Vec<Value>,
|
||||
value: Vec<PipelineMap>,
|
||||
pipeline: Arc<pipeline::Pipeline>,
|
||||
pipeline_handler: PipelineHandlerRef,
|
||||
query_ctx: &QueryContextRef,
|
||||
@@ -292,7 +292,7 @@ async fn dryrun_pipeline_inner(
|
||||
&pipeline_handler,
|
||||
&PipelineDefinition::Resolved(pipeline),
|
||||
¶ms,
|
||||
pipeline::json_array_to_map(value).context(PipelineSnafu)?,
|
||||
value,
|
||||
"dry_run".to_owned(),
|
||||
query_ctx,
|
||||
true,
|
||||
@@ -449,7 +449,7 @@ pub async fn pipeline_dryrun(
|
||||
|
||||
match check_pipeline_dryrun_params_valid(&payload) {
|
||||
Some(params) => {
|
||||
let data = params.data;
|
||||
let data = pipeline::json_array_to_map(params.data).context(PipelineSnafu)?;
|
||||
|
||||
check_data_valid(data.len())?;
|
||||
|
||||
@@ -580,16 +580,20 @@ fn extract_pipeline_value_by_content_type(
|
||||
content_type: ContentType,
|
||||
payload: Bytes,
|
||||
ignore_errors: bool,
|
||||
) -> Result<Vec<Value>> {
|
||||
) -> Result<Vec<PipelineMap>> {
|
||||
Ok(match content_type {
|
||||
ct if ct == *JSON_CONTENT_TYPE => transform_ndjson_array_factory(
|
||||
Deserializer::from_slice(&payload).into_iter(),
|
||||
ignore_errors,
|
||||
)?,
|
||||
ct if ct == *JSON_CONTENT_TYPE => {
|
||||
// `simd_json` have not support stream and ndjson, see https://github.com/simd-lite/simd-json/issues/349
|
||||
pipeline::json_array_to_map(transform_ndjson_array_factory(
|
||||
Deserializer::from_slice(&payload).into_iter(),
|
||||
ignore_errors,
|
||||
)?)
|
||||
.context(PipelineSnafu)?
|
||||
}
|
||||
ct if ct == *NDJSON_CONTENT_TYPE => {
|
||||
let mut result = Vec::with_capacity(1000);
|
||||
for (index, line) in payload.lines().enumerate() {
|
||||
let line = match line {
|
||||
let mut line = match line {
|
||||
Ok(line) if !line.is_empty() => line,
|
||||
Ok(_) => continue, // Skip empty lines
|
||||
Err(_) if ignore_errors => continue,
|
||||
@@ -602,7 +606,10 @@ fn extract_pipeline_value_by_content_type(
|
||||
}
|
||||
};
|
||||
|
||||
if let Ok(v) = serde_json::from_str(&line) {
|
||||
// simd_json, according to description, only de-escapes string at character level,
|
||||
// like any other json parser. So it should be safe here.
|
||||
if let Ok(v) = simd_json::to_owned_value(unsafe { line.as_bytes_mut() }) {
|
||||
let v = pipeline::simd_json_to_map(v).context(PipelineSnafu)?;
|
||||
result.push(v);
|
||||
} else if !ignore_errors {
|
||||
warn!("invalid JSON at index: {}, content: {:?}", index, line);
|
||||
@@ -617,8 +624,13 @@ fn extract_pipeline_value_by_content_type(
|
||||
ct if ct == *TEXT_CONTENT_TYPE || ct == *TEXT_UTF8_CONTENT_TYPE => payload
|
||||
.lines()
|
||||
.filter_map(|line| line.ok().filter(|line| !line.is_empty()))
|
||||
.map(|line| json!({"message": line}))
|
||||
.collect(),
|
||||
.map(|line| {
|
||||
let mut map = PipelineMap::new();
|
||||
map.insert("message".to_string(), pipeline::Value::String(line));
|
||||
map
|
||||
})
|
||||
.collect::<Vec<_>>(),
|
||||
|
||||
_ => UnsupportedContentTypeSnafu { content_type }.fail()?,
|
||||
})
|
||||
}
|
||||
@@ -646,7 +658,7 @@ pub(crate) async fn ingest_logs_inner(
|
||||
&state,
|
||||
&pipeline,
|
||||
&pipeline_params,
|
||||
pipeline::json_array_to_map(request.values).context(PipelineSnafu)?,
|
||||
request.values,
|
||||
request.table,
|
||||
&query_ctx,
|
||||
true,
|
||||
@@ -754,14 +766,19 @@ mod tests {
|
||||
let fail_rest =
|
||||
extract_pipeline_value_by_content_type(ContentType::json(), payload.clone(), true);
|
||||
assert!(fail_rest.is_ok());
|
||||
assert_eq!(fail_rest.unwrap(), vec![json!({"a": 1})]);
|
||||
assert_eq!(
|
||||
fail_rest.unwrap(),
|
||||
pipeline::json_array_to_map(vec![json!({"a": 1})]).unwrap()
|
||||
);
|
||||
|
||||
let fail_only_wrong =
|
||||
extract_pipeline_value_by_content_type(NDJSON_CONTENT_TYPE.clone(), payload, true);
|
||||
assert!(fail_only_wrong.is_ok());
|
||||
assert_eq!(
|
||||
fail_only_wrong.unwrap(),
|
||||
vec![json!({"a": 1}), json!({"c": 1})]
|
||||
);
|
||||
|
||||
let mut map1 = PipelineMap::new();
|
||||
map1.insert("a".to_string(), pipeline::Value::Uint64(1));
|
||||
let mut map2 = PipelineMap::new();
|
||||
map2.insert("c".to_string(), pipeline::Value::Uint64(1));
|
||||
assert_eq!(fail_only_wrong.unwrap(), vec![map1, map2]);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -23,8 +23,8 @@ use common_error::ext::ErrorExt;
|
||||
use common_query::Output;
|
||||
use datafusion_expr::LogicalPlan;
|
||||
use log_query::LogQuery;
|
||||
use pipeline::PipelineMap;
|
||||
use query::parser::PromQuery;
|
||||
use serde_json::Value;
|
||||
use session::context::QueryContextRef;
|
||||
use sql::statements::statement::Statement;
|
||||
|
||||
@@ -385,9 +385,9 @@ pub trait LogIngestInterceptor {
|
||||
/// Called before pipeline execution.
|
||||
fn pre_pipeline(
|
||||
&self,
|
||||
values: Vec<Value>,
|
||||
values: Vec<PipelineMap>,
|
||||
_query_ctx: QueryContextRef,
|
||||
) -> Result<Vec<Value>, Self::Error> {
|
||||
) -> Result<Vec<PipelineMap>, Self::Error> {
|
||||
Ok(values)
|
||||
}
|
||||
|
||||
@@ -412,9 +412,9 @@ where
|
||||
|
||||
fn pre_pipeline(
|
||||
&self,
|
||||
values: Vec<Value>,
|
||||
values: Vec<PipelineMap>,
|
||||
query_ctx: QueryContextRef,
|
||||
) -> Result<Vec<Value>, Self::Error> {
|
||||
) -> Result<Vec<PipelineMap>, Self::Error> {
|
||||
if let Some(this) = self {
|
||||
this.pre_pipeline(values, query_ctx)
|
||||
} else {
|
||||
|
||||
Reference in New Issue
Block a user