diff --git a/Cargo.lock b/Cargo.lock index af28ecf6a21..3167b27a6ee 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/Cargo.toml b/Cargo.toml index 8f221b2be1a..e738ff55ce7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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" diff --git a/src/pipeline/Cargo.toml b/src/pipeline/Cargo.toml index 6ce55591a17..32fa4eaa523 100644 --- a/src/pipeline/Cargo.toml +++ b/src/pipeline/Cargo.toml @@ -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 diff --git a/src/pipeline/src/etl.rs b/src/pipeline/src/etl.rs index 5d98dce4a81..676600d0a4c 100644 --- a/src/pipeline/src/etl.rs +++ b/src/pipeline/src/etl.rs @@ -170,6 +170,23 @@ pub fn json_array_to_map(val: Vec) -> Result val.into_iter().map(json_to_map).collect() } +pub fn simd_json_to_map(val: simd_json::OwnedValue) -> Result { + 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) -> Result> { + val.into_iter().map(simd_json_to_map).collect() +} + impl Pipeline { pub fn exec_mut(&self, val: &mut PipelineMap) -> Result { // process diff --git a/src/pipeline/src/etl/value.rs b/src/pipeline/src/etl/value.rs index ece852056a1..09faa787377 100644 --- a/src/pipeline/src/etl/value.rs +++ b/src/pipeline/src/etl/value.rs @@ -367,6 +367,37 @@ impl std::fmt::Display for Value { } } +impl TryFrom for Value { + type Error = Error; + + fn try_from(v: simd_json::value::OwnedValue) -> Result { + 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 for Value { type Error = Error; diff --git a/src/pipeline/src/lib.rs b/src/pipeline/src/lib.rs index 6c477ed7e59..857c321d337 100644 --- a/src/pipeline/src/lib.rs +++ b/src/pipeline/src/lib.rs @@ -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, diff --git a/src/servers/Cargo.toml b/src/servers/Cargo.toml index 329c97c27e7..abd3d105c01 100644 --- a/src/servers/Cargo.toml +++ b/src/servers/Cargo.toml @@ -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" diff --git a/src/servers/src/elasticsearch.rs b/src/servers/src/elasticsearch.rs index 60b9834b907..ba0e97acbfa 100644 --- a/src/servers/src/elasticsearch.rs +++ b/src/servers/src/elasticsearch.rs @@ -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::(&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(), ], }, ]), diff --git a/src/servers/src/http/event.rs b/src/servers/src/http/event.rs index 442f05802ac..bee41b80531 100644 --- a/src/servers/src/http/event.rs +++ b/src/servers/src/http/event.rs @@ -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, + pub values: Vec, } 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: Vec, pipeline: Arc, 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> { +) -> Result> { 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::>(), + _ => 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]); } } diff --git a/src/servers/src/interceptor.rs b/src/servers/src/interceptor.rs index b31b0893700..5a6e94723a1 100644 --- a/src/servers/src/interceptor.rs +++ b/src/servers/src/interceptor.rs @@ -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, + values: Vec, _query_ctx: QueryContextRef, - ) -> Result, Self::Error> { + ) -> Result, Self::Error> { Ok(values) } @@ -412,9 +412,9 @@ where fn pre_pipeline( &self, - values: Vec, + values: Vec, query_ctx: QueryContextRef, - ) -> Result, Self::Error> { + ) -> Result, Self::Error> { if let Some(this) = self { this.pre_pipeline(values, query_ctx) } else {