diff --git a/Cargo.lock b/Cargo.lock index 2badf7c6605..7a8436b67b4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11840,6 +11840,7 @@ dependencies = [ "greptime-proto", "humantime", "itertools 0.14.0", + "jsonb", "lazy_static", "log-query", "meter-core", diff --git a/src/common/function/src/scalars/json/json_get.rs b/src/common/function/src/scalars/json/json_get.rs index fcce7c9ee1c..ddddf11f010 100644 --- a/src/common/function/src/scalars/json/json_get.rs +++ b/src/common/function/src/scalars/json/json_get.rs @@ -12,7 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::borrow::Cow; use std::sync::Arc; use arrow::array::{ArrayRef, BinaryViewArray, new_null_array}; @@ -28,23 +27,37 @@ use datafusion_expr::{ColumnarValue, ScalarFunctionArgs, Signature, Volatility}; use datatypes::extension::json::is_json2_extension_type; use datatypes::vectors::json::array::JsonArray; use derive_more::Display; +use jsonb::jsonpath::{JsonPath, Path, parse_json_path}; use crate::function::{Function, extract_args}; use crate::helper; -fn get_json_by_path(json: &[u8], path: &str) -> Option> { - let json_path = jsonb::jsonpath::parse_json_path(path.as_bytes()); - match json_path { - Ok(json_path) => { - let mut sub_jsonb = Vec::new(); - let mut sub_offsets = Vec::new(); - match jsonb::get_by_path(json, json_path, &mut sub_jsonb, &mut sub_offsets) { - Ok(_) => Some(sub_jsonb), - Err(_) => None, - } - } - _ => None, +/// Parses the JSONPath expression accepted by `json_get`. +/// Keeps the legacy empty path as access to the root value. +pub fn parse_json_get_path(path: &str) -> std::result::Result, jsonb::Error> { + if path.is_empty() { + return Ok(JsonPath { + paths: vec![Path::Root], + }); } + // TODO(LFC): Fix jsonb's unterminated quoted-field panic (e.g. `$."a`) and + // update the dependency; malformed paths must return InvalidJsonPath. + parse_json_path(path.as_bytes()) +} + +fn json_object_path(path: &str) -> Result>> { + Ok(parse_json_get_path(path) + .map_err(|e| exec_datafusion_err!("Invalid JSONPath {path:?}: {e}"))? + .paths + .into_iter() + .filter_map(|segment| match segment { + Path::Root => None, + Path::DotField(name) | Path::ColonField(name) | Path::ObjectField(name) => { + Some(Some(name.into_owned())) + } + _ => Some(None), + }) + .collect()) } trait JsonGetResultBuilder { @@ -55,15 +68,10 @@ trait JsonGetResultBuilder { fn build(&mut self) -> ArrayRef; } -fn result_builder( - len: usize, - with_type: &DataType, - is_json2: bool, -) -> Result> { +fn result_builder(len: usize, with_type: &DataType) -> Result> { let builder = match with_type { DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => Box::new(StringResultBuilder { inner: StringViewBuilder::with_capacity(len), - is_json2, }) as Box, DataType::Int64 => Box::new(IntResultBuilder(Int64Builder::with_capacity(len))), @@ -79,19 +87,11 @@ fn result_builder( // TODO: refactor this to StringLikeArrayBuilder from Arrow 57 struct StringResultBuilder { inner: StringViewBuilder, - is_json2: bool, } impl JsonGetResultBuilder for StringResultBuilder { fn append_value(&mut self, value: &[u8]) -> Result<()> { - // Scalar casts stay unquoted and map JSON null to SQL NULL; only containers - // use `to_string` to preserve their JSON representation. - let value = if self.is_json2 && (jsonb::is_array(value) || jsonb::is_object(value)) { - Some(jsonb::to_string(value)) - } else { - jsonb::to_str(value).ok() - }; - self.inner.append_option(value); + self.inner.append_option(jsonb::to_str(value).ok()); Ok(()) } @@ -104,6 +104,21 @@ impl JsonGetResultBuilder for StringResultBuilder { } } +impl JsonGetResultBuilder for BinaryViewBuilder { + fn append_value(&mut self, value: &[u8]) -> Result<()> { + BinaryViewBuilder::append_value(self, value); + Ok(()) + } + + fn append_null(&mut self) { + BinaryViewBuilder::append_null(self); + } + + fn build(&mut self) -> ArrayRef { + Arc::new(self.finish()) + } +} + #[derive(Default, Display, Debug)] #[display("{}", Self::NAME.to_ascii_uppercase())] pub struct JsonGetString(JsonGetWithType); @@ -272,28 +287,31 @@ fn jsonb_get( path: &str, builder: &mut dyn JsonGetResultBuilder, ) -> Result<()> { + let json_path = parse_json_get_path(path) + .map_err(|e| exec_datafusion_err!("Invalid JSONPath {path:?}: {e}"))?; let size = jsons.len(); for i in 0..size { - let json = jsons.is_valid(i).then(|| jsons.value(i)); - let result = match json { - Some(json) => get_json_by_path(json, path), - _ => None, - }; - if let Some(v) = result { - builder.append_value(&v)?; - } else { + if jsons.is_null(i) { builder.append_null(); + continue; + } + let mut value = Vec::new(); + let mut offsets = Vec::new(); + jsonb::get_by_path(jsons.value(i), json_path.clone(), &mut value, &mut offsets) + .map_err(|e| exec_datafusion_err!("Failed to evaluate JSONPath {path:?}: {e}"))?; + if value.is_empty() { + builder.append_null(); + } else { + builder.append_value(&value)?; } } Ok(()) } fn json_struct_get(array: &ArrayRef, path: &str, with_type: &DataType) -> Result { - let segments = path - .trim_start_matches("$") - .split('.') - .filter(|segment| !segment.is_empty()) - .collect::>(); + let Some(segments) = json_object_path(path)? else { + return exec_err!("JSONPath {path:?} is not supported for Struct input"); + }; let mut curr = array.clone(); @@ -308,7 +326,7 @@ fn json_struct_get(array: &ArrayRef, path: &str, with_type: &DataType) -> Result let Some(json) = curr.as_struct_opt() else { return exec_err!("unknown JSON array datatype: {}", curr.data_type()); }; - let Some(sub_json) = json.column_by_name(segment) else { + let Some(sub_json) = json.column_by_name(segment.as_ref()) else { return Ok(new_null_array(with_type, array.len())); }; curr = sub_json.clone(); @@ -327,11 +345,11 @@ fn json_struct_get(array: &ArrayRef, path: &str, with_type: &DataType) -> Result /// /// For example, path `["a", "b"]` with an `Int64` leaf produces /// `Struct>`. -fn nested_projection_type(path: &[&str], leaf_type: &DataType) -> DataType { +fn nested_projection_type(path: &[String], leaf_type: &DataType) -> DataType { path.iter() .rev() .fold(leaf_type.clone(), |data_type, name| { - DataType::Struct(vec![Arc::new(Field::new(*name, data_type, true))].into()) + DataType::Struct(vec![Arc::new(Field::new(name.as_str(), data_type, true))].into()) }) } @@ -422,19 +440,15 @@ impl Function for JsonGetWithType { let arg0 = compute::cast(&arg0, &DataType::BinaryView)?; let is_json2 = args.arg_fields.first().is_some_and(is_json2_extension_type); - if is_json2 && path.trim_start_matches('$').split('.').all(str::is_empty) { - JsonArray::from(&arg0) + if is_json2 { + let mut builder = BinaryViewBuilder::with_capacity(len); + jsonb_get(arg0.as_binary_view(), path, &mut builder)?; + JsonArray::from(&builder.build()) .project_to(&with_type) - .map_err(|e| exec_datafusion_err!("{e:?}"))? + .map_err(|e| exec_datafusion_err!("{e}"))? } else { - let jsons = arg0.as_binary_view(); - let path = if is_json2 && !path.starts_with('$') { - Cow::Owned(format!("$.{path}")) - } else { - Cow::Borrowed(path) - }; - let mut builder = result_builder(len, &with_type, is_json2)?; - jsonb_get(jsons, &path, builder.as_mut())?; + let mut builder = result_builder(len, &with_type)?; + jsonb_get(arg0.as_binary_view(), path, builder.as_mut())?; builder.build() } } @@ -536,6 +550,95 @@ mod tests { use super::*; + #[test] + fn test_parse_json_get_path_expression() -> std::result::Result<(), Box> + { + for expression in [ + "$.l[0]", + "$.l[*]", + "$.o.*", + "$.l[0 to 2]", + "$.l ? (@.a == 1)", + ] { + let parsed = parse_json_get_path(expression).map_err(|e| e.to_string())?; + assert_eq!( + parsed, + parse_json_path(expression.as_bytes()).map_err(|e| e.to_string())? + ); + assert_eq!(json_object_path(expression)?, None); + } + Ok(()) + } + + #[test] + fn test_json_object_path() -> Result<()> { + for path in ["a.b", "$.a.b", r#"$ ."a"."b""#, r#"["a"]["b"]"#] { + assert_eq!(json_object_path(path)?, Some(vec!["a".into(), "b".into()])); + } + for (path, field) in [ + (r#"$."a.b""#, "a.b"), + (r#"$."a\"b""#, "a\"b"), + (r#"$."a\\b""#, "a\\b"), + (r#"["a.b"]"#, "a.b"), + (r#"["a\"b"]"#, "a\"b"), + (r#"["a\\b"]"#, "a\\b"), + (r#"["a[0]"]"#, "a[0]"), + ] { + assert_eq!(json_object_path(path)?, Some(vec![field.into()]), "{path}"); + } + for path in ["a[0]", "a[*]", "a.*", "$ ? (@.a == 1)"] { + assert_eq!(json_object_path(path)?, None, "{path}"); + } + assert!(json_object_path("a[").is_err()); + assert_eq!(json_object_path("$")?, Some(vec![])); + Ok(()) + } + + #[test] + fn test_parse_json_get_path_quoted_keys() { + for path in [ + r#"$."a.b""#, + r#""a.b""#, + r#""a\"b""#, + r#""a\\b""#, + r#"$["a\"b"]"#, + r#"["a\\b"]"#, + r#"$.a."""#, + r#"$."""#, + r#"$[""]"#, + ] { + assert_eq!( + parse_json_get_path(path), + parse_json_path(path.as_bytes()), + "{path}", + ); + } + } + + #[test] + fn test_json_struct_get_object_paths() -> std::result::Result<(), Box> { + for path in [ + "payload.code", + r#"$."payload"."code""#, + r#"["payload"]["code"]"#, + ] { + let result = json_struct_get(&test_json_struct(), path, &DataType::Int64)?; + assert_eq!(result.as_primitive::().value(0), 404); + } + let nested: ArrayRef = Arc::new(StructArray::from(vec![( + Arc::new(Field::new("a.b", DataType::Binary, true)), + Arc::new(BinaryArray::from_iter_values([parse_string_to_jsonb( + r#"{"c.d": 7}"#, + )?])) as ArrayRef, + )])); + for path in [r#"$."a.b"."c.d""#, r#"$["a.b"]["c.d"]"#] { + let result = json_struct_get(&nested, path, &DataType::Int64)?; + assert!(!result.is_null(0), "{path}"); + assert_eq!(result.as_primitive::().value(0), 7, "{path}"); + } + Ok(()) + } + /// Create a JSON object like this (as a one element struct array for testing): /// /// ```JSON @@ -623,6 +726,49 @@ mod tests { } } + #[test] + fn test_json_get_invalid_path() -> std::result::Result<(), Box> { + let json = parse_string_to_jsonb(r#"{"a": 1}"#)?; + let arrays: Vec = vec![ + Arc::new(BinaryArray::from_iter_values([json])), + Arc::new(BinaryArray::from(vec![None::<&[u8]>])), + Arc::new(BinaryArray::from(Vec::>::new())), + test_json_struct(), + ]; + for array in arrays { + for is_json2 in [false, true] { + for path in ["$.a[", "..", "$$"] { + let args = ScalarFunctionArgs { + args: vec![ + ColumnarValue::Array(array.clone()), + ColumnarValue::Scalar(path.into()), + ], + arg_fields: vec![test_json_field(&array, is_json2)], + number_rows: array.len(), + return_field: Arc::new(Field::new("x", DataType::Int64, true)), + config_options: Arc::new(Default::default()), + }; + let err = JsonGetInt::default().invoke_with_args(args).unwrap_err(); + assert!(err.to_string().contains("Invalid JSONPath"), "{err}"); + } + } + } + Ok(()) + } + + #[test] + fn test_jsonb_get_evaluation_error() -> std::result::Result<(), Box> { + let json = parse_string_to_jsonb(r#"{"a": 1}"#)?; + let truncated = BinaryViewArray::from_iter_values([&json[..1]]); + let mut builder = IntResultBuilder(Int64Builder::new()); + let err = jsonb_get(&truncated, "$.a", &mut builder).unwrap_err(); + assert!( + err.to_string().contains("Failed to evaluate JSONPath"), + "{err}" + ); + Ok(()) + } + #[test] fn test_json_get_int() { let json_get_int = JsonGetInt::default(); @@ -650,8 +796,8 @@ mod tests { ("$.payload.code", Some(404)), ("$.payload.success", Some(0)), ("$.payload.result.time_cost", Some(1)), - ("$.payload.not-exists", None), - ("$.not-exists", None), + ("$.payload.not_exists", None), + ("$.not_exists", None), ("$", None), ]; @@ -719,8 +865,8 @@ mod tests { ("$.payload.code", Some(404.0)), ("$.payload.success", Some(0.0)), ("$.payload.result.time_cost", Some(1.234)), - ("$.payload.not-exists", None), - ("$.not-exists", None), + ("$.payload.not_exists", None), + ("$.not_exists", None), ("$", None), ]; @@ -788,8 +934,8 @@ mod tests { ("$.payload.code", Some(true)), ("$.payload.success", Some(false)), ("$.payload.result.time_cost", Some(true)), - ("$.payload.not-exists", None), - ("$.not-exists", None), + ("$.payload.not_exists", None), + ("$.not_exists", None), ("$", None), ]; @@ -860,9 +1006,9 @@ mod tests { "$.payload.success", "$.payload.result", "$.payload.result.error", - "$.payload.result.not-exists", - "$.payload.not-exists", - "$.not-exists", + "$.payload.result.not_exists", + "$.payload.not_exists", + "$.not_exists", "$", ]; let expects = [ @@ -996,9 +1142,9 @@ mod tests { "$.payload.success", "$.payload.result", "$.payload.result.error", - "$.payload.result.not-exists", - "$.payload.not-exists", - "$.not-exists", + "$.payload.result.not_exists", + "$.payload.not_exists", + "$.not_exists", "$", ]; let expects = [ @@ -1178,4 +1324,82 @@ mod tests { assert_eq!(actual, *expect); } } + + #[test] + fn test_json_get_json2_root() -> std::result::Result<(), Box> { + for (input, expected) in [ + ("42", ScalarValue::Int64(Some(42))), + ("42", ScalarValue::Int32(Some(42))), + ("18446744073709551615", ScalarValue::UInt64(Some(u64::MAX))), + ("1.5", ScalarValue::Float64(Some(1.5))), + ("true", ScalarValue::Boolean(Some(true))), + (r#""text""#, ScalarValue::Utf8View(Some("text".into()))), + ("null", ScalarValue::Utf8View(None)), + ( + r#"{"a":1}"#, + ScalarValue::Utf8View(Some(r#"{"a":1}"#.into())), + ), + ("[1,2]", ScalarValue::Utf8View(Some("[1,2]".into()))), + ] { + let json = parse_string_to_jsonb(input)?; + let array: ArrayRef = Arc::new(BinaryArray::from_iter_values([json])); + for path in ["", "$", " $ "] { + let args = ScalarFunctionArgs { + args: vec![ + ColumnarValue::Array(array.clone()), + ColumnarValue::Scalar(path.into()), + ColumnarValue::Scalar(ScalarValue::try_new_null(&expected.data_type())?), + ], + arg_fields: vec![test_json_field(&array, true)], + number_rows: 1, + return_field: Arc::new(Field::new("x", expected.data_type(), true)), + config_options: Arc::new(Default::default()), + }; + let result = JsonGetWithType::default() + .invoke_with_args(args)? + .to_array(1)?; + assert_eq!( + ScalarValue::try_from_array(&result, 0)?, + expected, + "{path:?}: {input}" + ); + } + } + Ok(()) + } + + #[test] + fn test_json_get_json2_quoted_root_field() -> std::result::Result<(), Box> + { + for key in ["a.b", "a\"b", "a\\b"] { + let input = json!({key: 42}).to_string(); + let value = jsonb::parse_value(input.as_bytes()).map_err(|e| e.to_string())?; + let json = Arc::new(BinaryArray::from_iter_values([value.to_vec()])) as ArrayRef; + let args = ScalarFunctionArgs { + args: vec![ + ColumnarValue::Array(json.clone()), + ColumnarValue::Scalar(ScalarValue::Utf8(Some(format!( + "[{}]", + serde_json::to_string(key)? + )))), + ColumnarValue::Scalar(ScalarValue::Int64(None)), + ], + arg_fields: vec![ + test_json_field(&json, true), + Arc::new(Field::new("path", DataType::Utf8, false)), + Arc::new(Field::new("with_type", DataType::Int64, true)), + ], + number_rows: 1, + return_field: Arc::new(Field::new("x", DataType::Int64, true)), + config_options: Arc::new(Default::default()), + }; + + let result = JsonGetWithType::default() + .invoke_with_args(args) + .and_then(|x| x.to_array(1))?; + assert_eq!(result.as_primitive::().value(0), 42); + assert!(!result.is_null(0)); + } + Ok(()) + } } diff --git a/src/datatypes/src/types/json_type.rs b/src/datatypes/src/types/json_type.rs index 77e67d9c6a7..337ac415187 100644 --- a/src/datatypes/src/types/json_type.rs +++ b/src/datatypes/src/types/json_type.rs @@ -14,7 +14,6 @@ use std::collections::BTreeMap; use std::fmt::{Debug, Display, Formatter}; -use std::str::FromStr; use std::sync::{Arc, LazyLock}; use arrow::datatypes::DataType as ArrowDataType; @@ -22,13 +21,10 @@ use arrow_schema::{Field, Fields}; use common_base::bytes::Bytes; use regex::{Captures, Regex}; use serde::{Deserialize, Serialize}; -use snafu::ResultExt; use crate::Error; use crate::data_type::DataType; -use crate::error::{ - DeserializeSnafu, InvalidJsonSnafu, InvalidJsonbSnafu, Result, UnsupportedArrowTypeSnafu, -}; +use crate::error::{InvalidJsonSnafu, InvalidJsonbSnafu, Result, UnsupportedArrowTypeSnafu}; use crate::prelude::ConcreteDataType; use crate::scalars::ScalarVectorBuilder; use crate::type_id::LogicalTypeId; @@ -408,8 +404,7 @@ pub fn jsonb_to_string(val: &[u8]) -> Result { /// Converts a json type value to serde_json::Value pub fn jsonb_to_serde_json(val: &[u8]) -> Result { - let json_string = jsonb_to_string(val)?; - serde_json::Value::from_str(&json_string).context(DeserializeSnafu { json: json_string }) + jsonb::to_serde_json(val).map_err(|error| InvalidJsonbSnafu { error }.build()) } /// Normalizes a JSON string by converting Rust-style Unicode escape sequences to JSON-compatible format. @@ -474,6 +469,16 @@ pub fn parse_string_to_jsonb(s: &str) -> Result> { #[cfg(test)] mod tests { + #[test] + fn test_v2_jsonb_to_serde_json_escaped_keys() + -> std::result::Result<(), Box> { + let expected = serde_json::json!({"": 9, "a\"b": {"a\\b": "\u{1f600}\n"}}); + let input = expected.to_string(); + let value = jsonb::parse_value(input.as_bytes()).map_err(|e| e.to_string())?; + assert_eq!(jsonb_to_serde_json(&value.to_vec())?, expected); + Ok(()) + } + use super::*; #[test] diff --git a/src/frontend/src/instance/otlp.rs b/src/frontend/src/instance/otlp.rs index 4c80f1d17f2..d72b79a4275 100644 --- a/src/frontend/src/instance/otlp.rs +++ b/src/frontend/src/instance/otlp.rs @@ -214,10 +214,10 @@ impl OpenTelemetryProtocolHandler for Instance { #[tracing::instrument(skip_all)] async fn traces( &self, - pipeline_handler: PipelineHandlerRef, + _pipeline_handler: PipelineHandlerRef, request: ExportTraceServiceRequest, pipeline: PipelineWay, - pipeline_params: GreptimePipelineParams, + _pipeline_params: GreptimePipelineParams, table_name: String, ctx: QueryContextRef, ) -> ServerResult { @@ -239,16 +239,8 @@ impl OpenTelemetryProtocolHandler for Instance { let targets = trace_permission_targets(&table_name, &spans, &ctx); self.check_table_permission(&ctx, PermissionReq::Action(OTLP_WRITE), targets) .context(AuthSnafu)?; - self.ingest_trace_spans( - pipeline_handler, - &pipeline, - &pipeline_params, - table_name, - spans, - &conventions, - ctx, - ) - .await + self.ingest_trace_spans(&pipeline, table_name, spans, &conventions, ctx) + .await } #[tracing::instrument(skip_all)] diff --git a/src/frontend/src/instance/otlp/README.md b/src/frontend/src/instance/otlp/README.md index 4489ab0c3f3..0ce3e94a26a 100644 --- a/src/frontend/src/instance/otlp/README.md +++ b/src/frontend/src/instance/otlp/README.md @@ -24,10 +24,12 @@ flowchart TD F -->|greptime_trace_v0| G["v0 conversion
fixed columns and JSON attributes"] F -->|greptime_trace_v1| H["v1 conversion
flattened dynamic attributes"] + F -->|greptime_trace_v2| T["v2 conversion
fixed columns and JSON2 attributes"] H --> I["Collect request-wide schema observations"] I --> J["Resolve types, prevalidate rewrites,
apply DDL, and recheck table schema"] G --> K["Write main-table chunks"] + T --> K J --> K K --> L{"Chunk result"} L -->|success| M["Accept spans and collect
service/operation data"] @@ -54,8 +56,8 @@ The route is registered as `POST /v1/otlp/v1/traces` in 3. resolves the requested pipeline; and 4. calls `OpenTelemetryProtocolHandler::traces`. -The trace converter currently accepts the internal `greptime_trace_v0` and -`greptime_trace_v1` pipelines. Their `PipelineWay` variants are defined in +The trace converter accepts the internal `greptime_trace_v0`, +`greptime_trace_v1`, and `greptime_trace_v2` pipelines. Their `PipelineWay` variants are defined in [`pipeline/src/manager.rs`](../../../../pipeline/src/manager.rs). ### 2. Parse and chunk @@ -77,6 +79,17 @@ it to 0 disables splitting. The option is defined in | --- | --- | --- | | v0 | [`trace/v0.rs`](../../../../servers/src/otlp/trace/v0.rs) writes a fixed span schema and keeps span, scope, and resource attributes in JSON columns. | Converts and writes one chunk at a time through the legacy log insertion path. | | v1 | [`trace/v1.rs`](../../../../servers/src/otlp/trace/v1.rs) flattens OTLP attributes into dynamic columns and records the value types seen in each chunk. | Scans chunks one at a time to build a request-wide schema plan, then materializes and writes one chunk at a time. | +| v2 | [`trace/v2.rs`](../../../../servers/src/otlp/trace/v2.rs) writes one row per span with fixed columns and preserves resource, scope, and span attributes in JSON2 columns. | Converts and writes fixed-schema chunks without dynamic-column reconciliation. | + +The v2 pipeline currently discards events and links. Their storage is deferred +until `ARRAY(JSON2)` is supported (TODO(LFC)). + +V2 always creates the nullable `service_name` tag and reads it only from the +`service.name` resource attribute. It rejects out-of-range start/end timestamps +before writing. A conversion failure isolates invalid spans and writes the +remaining chunk; a storage failure propagates without retry because it may have +partially committed. V1 and v2 requests cannot write to a table stamped with the +other model, including through automatic schema alteration. Consequently, v1 chunks from one request can have different schemas: attribute keys are data-dependent. The request-wide plan reconciles compatible column diff --git a/src/frontend/src/instance/otlp/trace_ingest.rs b/src/frontend/src/instance/otlp/trace_ingest.rs index 5951728e6f4..e5b16c7d191 100644 --- a/src/frontend/src/instance/otlp/trace_ingest.rs +++ b/src/frontend/src/instance/otlp/trace_ingest.rs @@ -28,20 +28,20 @@ use common_meta::rpc::ddl::TriggerReason; use common_telemetry::{debug, warn}; use datatypes::prelude::ConcreteDataType; use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; -use pipeline::{GreptimePipelineParams, PipelineWay}; +use pipeline::PipelineWay; use servers::error::{self, Result as ServerResult}; use servers::otlp; use servers::otlp::coerce::{coerce_value_data, is_supported_trace_coercion, trace_value_datatype}; use servers::otlp::trace::span::{TraceSpan, TraceSpanGroup}; use servers::otlp::trace::v1::{TraceBatchSchema, TraceBinaryType, TraceRetryColumns}; use servers::otlp::trace::{SERVICE_NAME_COLUMN, TraceAuxData}; -use servers::query_handler::{PipelineHandlerRef, TraceIngestOutcome}; +use servers::query_handler::TraceIngestOutcome; use session::context::QueryContextRef; use snafu::{IntoError, ResultExt}; use table::requests::{ SEMANTIC_ENTITY_SERVICE_ID, SEMANTIC_PIPELINE, SEMANTIC_SIGNAL_TYPE, SEMANTIC_SOURCE, SEMANTIC_TRACE_CONVENTIONS, SEMANTIC_VALUE_MIXED, SEMANTIC_VALUE_UNKNOWN, SIGNAL_TYPE_TRACE, - SOURCE_OPENTELEMETRY, TABLE_DATA_MODEL_TRACE_V1, + SOURCE_OPENTELEMETRY, TABLE_DATA_MODEL_TRACE_V1, TABLE_DATA_MODEL_TRACE_V2, }; use crate::instance::Instance; @@ -54,6 +54,22 @@ use crate::instance::otlp::trace_types::{ }; use crate::metrics::{OTLP_TRACES_FAILURE_COUNT, OTLP_TRACES_ROWS}; +/// Merge converted V2 rows only when their fixed schemas agree. +fn merge_trace_v2_rows(rows: &mut Rows, batch: Rows) -> ServerResult<()> { + if rows.schema.is_empty() { + rows.schema = batch.schema; + } else { + snafu::ensure!( + rows.schema == batch.schema, + error::InternalSnafu { + err_msg: "Trace V2 schema mismatch while merging spans", + } + ); + } + rows.rows.extend(batch.rows); + Ok(()) +} + const TRACE_FAILURE_MESSAGE_LIMIT: usize = 4; /// Maximum characters of a failure cause echoed to the client and the log. @@ -79,11 +95,8 @@ impl ChunkFailureReaction { /// Shared dependencies and request metadata used while ingesting trace chunks. struct TraceChunkIngestContext<'a> { - pipeline_handler: PipelineHandlerRef, pipeline: &'a PipelineWay, - pipeline_params: &'a GreptimePipelineParams, table_name: &'a str, - is_trace_v1_model: bool, } /// Accumulates trace outcomes, auxiliary rows, and bounded failure details. @@ -855,18 +868,20 @@ impl TraceColumnRequestSchema { impl Instance { /// Ingest OTLP trace spans with chunk-level writes and span-level fallback on /// deterministic chunk failures. - #[allow(clippy::too_many_arguments)] pub(super) async fn ingest_trace_spans( &self, - pipeline_handler: PipelineHandlerRef, pipeline: &PipelineWay, - pipeline_params: &GreptimePipelineParams, table_name: String, groups: Vec, conventions: &str, ctx: QueryContextRef, ) -> ServerResult { let is_trace_v1_model = matches!(pipeline, PipelineWay::OtlpTraceDirectV1); + let data_model = match pipeline { + PipelineWay::OtlpTraceDirectV1 => Some(TABLE_DATA_MODEL_TRACE_V1), + PipelineWay::OtlpTraceDirectV2 => Some(TABLE_DATA_MODEL_TRACE_V2), + _ => None, + }; // Only the main span table gets the identity; the derived `_services` / // `_operations` lookup tables keep the unstamped `ctx`. @@ -877,19 +892,30 @@ impl Instance { // `service_name` is a tag column in both trace models, so the main span // table declares the logical `service` entity (Layer 1 auto-stamp). c.set_extension(SEMANTIC_ENTITY_SERVICE_ID, SERVICE_NAME_COLUMN); - if is_trace_v1_model { - c.set_extension(SEMANTIC_PIPELINE, TABLE_DATA_MODEL_TRACE_V1); + if let Some(data_model) = data_model { + c.set_extension(SEMANTIC_PIPELINE, data_model); c.set_extension(SEMANTIC_TRACE_CONVENTIONS, conventions); } Arc::new(c) }; + if data_model.is_some() + && let Some(table) = self + .catalog_manager + .table( + ctx.current_catalog(), + &ctx.current_schema(), + &table_name, + None, + ) + .await? + { + operator::insert::validate_trace_table_model(&table.table_info(), &main_ctx)?; + } + let ingest_ctx = TraceChunkIngestContext { - pipeline_handler, pipeline, - pipeline_params, table_name: &table_name, - is_trace_v1_model, }; let mut ingest_state = TraceIngestState { aux_data: TraceAuxData::default(), @@ -975,7 +1001,7 @@ impl Instance { match aux_requests { Ok((aux_requests, _)) if !aux_requests.inserts.is_empty() => { match self - .insert_trace_requests(aux_requests, ingest_ctx.is_trace_v1_model, ctx) + .insert_trace_requests(aux_requests, ingest_ctx.pipeline, ctx) .await { Ok(output) => { @@ -1055,23 +1081,71 @@ impl Instance { async fn ingest_trace_chunk( &self, ingest_ctx: &TraceChunkIngestContext<'_>, - chunk: Vec, + mut chunk: Vec, ctx: QueryContextRef, ingest_state: &mut TraceIngestState, ) -> ServerResult<()> { // Try the fast path first so healthy batches keep their original // throughput and write amplification stays low. - let (requests, chunk_rows) = otlp::trace::to_grpc_insert_requests_from_spans( - &chunk, - ingest_ctx.pipeline, - ingest_ctx.pipeline_params, - ingest_ctx.table_name, - &ctx, - ingest_ctx.pipeline_handler.clone(), - )?; + let convert = |spans: &[TraceSpan]| { + otlp::trace::to_grpc_insert_requests_from_spans( + spans, + ingest_ctx.pipeline, + ingest_ctx.table_name, + ) + }; + let (requests, chunk_rows) = match convert(&chunk) { + Ok(x) => x, + Err(_) if matches!(ingest_ctx.pipeline, PipelineWay::OtlpTraceDirectV2) => { + let mut rows = Rows { + rows: Vec::with_capacity(chunk.len()), + ..Default::default() + }; + + for span in std::mem::take(&mut chunk) { + let result = convert(std::slice::from_ref(&span)); + match result { + Ok((requests, _)) => { + // V2 must always emit the same table and fixed column order. + for batch in requests.inserts.into_iter().filter_map(|r| r.rows) { + merge_trace_v2_rows(&mut rows, batch)?; + } + chunk.push(span); + } + Err(err) => { + ingest_state.outcome.rejected_spans += 1; + + let (cause, shown_cause) = Self::trace_failure_cause(&err); + Self::push_trace_failure_message( + &mut ingest_state.failure_messages, + "span_rejected", + &cause, + format!( + "Rejected span {}:{}: {}", + span.trace_id, span.span_id, shown_cause + ), + ); + } + } + } + + if rows.rows.is_empty() { + return Ok(()); + } + let chunk_rows = rows.rows.len(); + let requests = RowInsertRequests { + inserts: vec![RowInsertRequest { + table_name: ingest_ctx.table_name.to_string(), + rows: Some(rows), + }], + }; + (requests, chunk_rows) + } + Err(e) => return Err(e), + }; let output = match self - .insert_trace_requests(requests, ingest_ctx.is_trace_v1_model, ctx) + .insert_trace_requests(requests, ingest_ctx.pipeline, ctx) .await { Ok(output) => output, @@ -1364,12 +1438,17 @@ impl Instance { async fn insert_trace_requests( &self, mut requests: RowInsertRequests, - is_trace_v1_model: bool, + pipeline: &PipelineWay, ctx: QueryContextRef, ) -> ServerResult { - if is_trace_v1_model { + if matches!(pipeline, PipelineWay::OtlpTraceDirectV1) { self.reconcile_trace_column_types(&mut requests, &ctx) .await?; + } + if matches!( + pipeline, + PipelineWay::OtlpTraceDirectV1 | PipelineWay::OtlpTraceDirectV2 + ) { self.handle_trace_inserts(requests, ctx) .await .map_err(BoxedError::new) @@ -1394,6 +1473,9 @@ impl Instance { .catalog_manager .table(catalog, &schema, table_name, None) .await?; + if let Some(table) = &table { + operator::insert::validate_trace_table_model(&table.table_info(), ctx)?; + } let table_schema = table.as_ref().map(|table| table.schema()); let exclusions = request_schema.incompatible_schema_observations(table_schema.as_deref()); if !exclusions.is_empty() { diff --git a/src/frontend/src/instance/otlp/trace_ingest/tests.rs b/src/frontend/src/instance/otlp/trace_ingest/tests.rs index 2445da95bb8..c0ba1a91239 100644 --- a/src/frontend/src/instance/otlp/trace_ingest/tests.rs +++ b/src/frontend/src/instance/otlp/trace_ingest/tests.rs @@ -1335,3 +1335,34 @@ fn test_trace_conventions() { }; assert_eq!(trace_conventions(&conflicting), "mixed"); } + +#[test] +fn test_merge_trace_v2_rows() -> servers::error::Result<()> { + let batch = Rows { + schema: vec![ColumnSchema { + column_name: "trace_id".to_string(), + datatype: ColumnDataType::String as i32, + ..Default::default() + }], + rows: vec![Row { + values: vec![Value { + value_data: Some(ValueData::StringValue("trace".to_string())), + }], + }], + }; + let mut rows = Rows::default(); + super::merge_trace_v2_rows(&mut rows, batch.clone())?; + super::merge_trace_v2_rows(&mut rows, batch.clone())?; + assert_eq!(rows.rows.len(), 2); + + let before = rows.clone(); + let mut incompatible = batch; + incompatible.schema[0].column_name = "span_id".to_string(); + let result = super::merge_trace_v2_rows(&mut rows, incompatible); + assert!(matches!( + result, + Err(servers::error::Error::Internal { .. }) + )); + assert_eq!(rows, before); + Ok(()) +} diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs index 7fb538dc775..35727ae9b35 100644 --- a/src/operator/src/insert.rs +++ b/src/operator/src/insert.rs @@ -68,8 +68,9 @@ use table::TableRef; use table::metadata::{TableInfo, TableInfoRef}; use table::requests::{ AUTO_CREATE_TABLE_KEY, InsertRequest as TableInsertRequest, SEMANTIC_PER_TABLE_INDEX_KEY, - TABLE_DATA_MODEL, TABLE_DATA_MODEL_TRACE_V1, TRACE_TABLE_PARTITIONS_HINT_KEY, - VALID_TABLE_OPTION_KEYS, is_semantic_option_key, validate_semantic_option, + SEMANTIC_PIPELINE, TABLE_DATA_MODEL, TABLE_DATA_MODEL_TRACE_V1, TABLE_DATA_MODEL_TRACE_V2, + TRACE_TABLE_PARTITIONS_HINT_KEY, VALID_TABLE_OPTION_KEYS, is_semantic_option_key, + validate_semantic_option, }; use table::table_reference::TableReference; @@ -774,6 +775,9 @@ impl Inserter { } }; let table_info = table.table_info(); + if matches!(auto_create_table_type, AutoCreateTableType::Trace { .. }) { + validate_trace_table_model(&table_info, ctx)?; + } if table_info.is_ttl_instant_table() { instant_table_ids.insert(table_info.table_id()); } @@ -799,6 +803,9 @@ impl Inserter { match self.get_table(catalog, &schema, &req.table_name).await? { Some(table) => { let table_info = table.table_info(); + if matches!(auto_create_table_type, AutoCreateTableType::Trace { .. }) { + validate_trace_table_model(&table_info, ctx)?; + } if table_info.is_ttl_instant_table() { instant_table_ids.insert(table_info.table_id()); } @@ -990,7 +997,9 @@ impl Inserter { // use table_options to mark table model version create_table.table_options.insert( TABLE_DATA_MODEL.to_string(), - TABLE_DATA_MODEL_TRACE_V1.to_string(), + ctx.extension(SEMANTIC_PIPELINE) + .unwrap_or(TABLE_DATA_MODEL_TRACE_V1) + .to_string(), ); let table = self @@ -1307,7 +1316,10 @@ impl Inserter { ); match res { - Ok(table) => Ok(table), + Ok(table) => { + validate_trace_table_model(&table.table_info(), ctx)?; + Ok(table) + } Err(err) => { error!(err; "Failed to create table {}", table_ref); Err(err) @@ -1406,6 +1418,28 @@ fn validate_column_count_match(requests: &RowInsertRequests) -> Result<()> { Ok(()) } +/// Rejects writes from a different built-in trace model before schema mutation. +/// Unstamped, explicitly created tables remain subject to normal schema validation. +pub fn validate_trace_table_model(table_info: &TableInfo, ctx: &QueryContextRef) -> Result<()> { + let Some(expected @ (TABLE_DATA_MODEL_TRACE_V1 | TABLE_DATA_MODEL_TRACE_V2)) = + ctx.extension(SEMANTIC_PIPELINE) + else { + return Ok(()); + }; + if let Some(actual) = table_info.meta.options.extra_options.get(TABLE_DATA_MODEL) { + ensure!( + actual == expected, + InvalidInsertRequestSnafu { + reason: format!( + "Trace table `{}` uses {actual}, but the request uses {expected}", + table_info.name, + ), + } + ); + } + Ok(()) +} + /// Fill table options for a new table by create type. pub fn fill_table_options_for_create( table_options: &mut std::collections::HashMap, diff --git a/src/pipeline/src/lib.rs b/src/pipeline/src/lib.rs index e323859d02e..5d1acb7462e 100644 --- a/src/pipeline/src/lib.rs +++ b/src/pipeline/src/lib.rs @@ -33,8 +33,9 @@ pub use etl::{ }; pub use manager::{ GREPTIME_INTERNAL_IDENTITY_PIPELINE_NAME, GREPTIME_INTERNAL_TRACE_PIPELINE_V1_NAME, - IdentityTimeIndex, PipelineContext, PipelineDefinition, PipelineInfo, PipelineRef, - PipelineTableRef, PipelineVersion, PipelineWay, SelectInfo, pipeline_operator, table, util, + GREPTIME_INTERNAL_TRACE_PIPELINE_V2_NAME, IdentityTimeIndex, PipelineContext, + PipelineDefinition, PipelineInfo, PipelineRef, PipelineTableRef, PipelineVersion, PipelineWay, + SelectInfo, pipeline_operator, table, util, }; pub use options::PipelineOptions; diff --git a/src/pipeline/src/manager.rs b/src/pipeline/src/manager.rs index f2779d2d864..98505c818bf 100644 --- a/src/pipeline/src/manager.rs +++ b/src/pipeline/src/manager.rs @@ -80,6 +80,8 @@ impl SelectInfo { pub const GREPTIME_INTERNAL_IDENTITY_PIPELINE_NAME: &str = "greptime_identity"; pub const GREPTIME_INTERNAL_TRACE_PIPELINE_V0_NAME: &str = "greptime_trace_v0"; pub const GREPTIME_INTERNAL_TRACE_PIPELINE_V1_NAME: &str = "greptime_trace_v1"; +/// Built-in OTLP trace pipeline that stores semi-structured fields as JSON2. +pub const GREPTIME_INTERNAL_TRACE_PIPELINE_V2_NAME: &str = "greptime_trace_v2"; /// Enum for holding information of a pipeline, which is either pipeline itself, /// or information that be used to retrieve a pipeline from `PipelineHandler` @@ -146,6 +148,7 @@ pub enum PipelineWay { Pipeline(PipelineDefinition), OtlpTraceDirectV0, OtlpTraceDirectV1, + OtlpTraceDirectV2, } impl PipelineWay { @@ -155,7 +158,9 @@ impl PipelineWay { default_pipeline: Option, ) -> Result { if let Some(pipeline_name) = name { - if pipeline_name == GREPTIME_INTERNAL_TRACE_PIPELINE_V1_NAME { + if pipeline_name == GREPTIME_INTERNAL_TRACE_PIPELINE_V2_NAME { + Ok(PipelineWay::OtlpTraceDirectV2) + } else if pipeline_name == GREPTIME_INTERNAL_TRACE_PIPELINE_V1_NAME { Ok(PipelineWay::OtlpTraceDirectV1) } else if pipeline_name == GREPTIME_INTERNAL_TRACE_PIPELINE_V0_NAME { Ok(PipelineWay::OtlpTraceDirectV0) diff --git a/src/query/Cargo.toml b/src/query/Cargo.toml index 2a32292614f..b6dd2d1fdf4 100644 --- a/src/query/Cargo.toml +++ b/src/query/Cargo.toml @@ -54,6 +54,7 @@ futures-util.workspace = true greptime-proto.workspace = true humantime.workspace = true itertools.workspace = true +jsonb.workspace = true lazy_static.workspace = true log-query.workspace = true meter-core.workspace = true diff --git a/src/query/src/datafusion/json_expr_planner.rs b/src/query/src/datafusion/json_expr_planner.rs index 04acb48d199..c934565affa 100644 --- a/src/query/src/datafusion/json_expr_planner.rs +++ b/src/query/src/datafusion/json_expr_planner.rs @@ -33,17 +33,17 @@ use sqlparser::ast::BinaryOperator; /// Rewrites JSON-aware SQL expressions into DataFusion expressions. /// -/// This planner handles three cases: +/// This planner handles the following cases: /// - Rewrites compound identifiers on JSON extension columns into `json_get` function. -/// For example, `select a.b.c` => `select json_get(a, "b.c")`. +/// For example, `select a.b.c` => `select json_get(a, '$.b.c')`. /// - Extends a JSON path with list indexes and fields following an index. -/// For example, `select a.b[0].c` => `select json_get(a, "b[0][\"c\"]")`. +/// For example, `select a.b[0].c` => `select json_get(a, '$.b[0].c')`. /// - Pushes an "expected type" argument into the `json_get` function when it participates in a /// binary operator. So that `json_get` knows the wanted data type when dealing with variant /// JSON values. /// For example, `select json_get(a, "b.c") + 1` => `select json_get(a, "b.c", NULL::Int64) + 1`. /// - Infers the expected type from scalar, aggregate, and window function signatures. -/// For example, `select abs(a.b.c)` => `select abs(json_get(a, "b.c", NULL::Float64))`. +/// For example, `select abs(a.b.c)` => `select abs(json_get(a, '$.b.c', NULL::Float64))`. #[derive(Debug)] pub(crate) struct JsonExprPlanner; @@ -85,9 +85,9 @@ impl ExprPlanner for JsonExprPlanner { /// Extends the path of an untyped `json_get` with one field access. /// /// For `j.o.l[1].inner.l[2]`, `plan_compound_identifier` first produces - /// `json_get(j, "o.l")`. DataFusion then calls this method successively + /// `json_get(j, '$.o.l')`. DataFusion then calls this method successively /// with a list index, two named fields, and another list index, producing - /// the final path `o.l[1]["inner"]["l"][2]`. + /// the final path `$.o.l[1].inner.l[2]`. fn plan_field_access( &self, mut expr: RawFieldAccessExpr, @@ -107,12 +107,7 @@ impl ExprPlanner for JsonExprPlanner { let Some(name) = name.try_as_str().flatten() else { return Ok(PlannerResult::Original(expr)); }; - // Encode the field name as a JSON string before embedding it in the - // bracket accessor. This preserves dots as literal field-name characters - // and escapes quotes, backslashes, and control characters correctly. - let name = serde_json::to_string(name) - .map_err(|e| DataFusionError::External(Box::new(e)))?; - format!("[{name}]") + format!(".{}", json_path_field(name)?) } GetFieldAccess::ListRange { .. } => return Ok(PlannerResult::Original(expr)), }; @@ -141,7 +136,12 @@ impl ExprPlanner for JsonExprPlanner { LazyLock::new(|| Arc::new(create_udf(Arc::new(JsonGetWithType::default())))); let json_get = JSON_GET_UDF.clone(); - let path = nested_names.join("."); + let mut path = "$".to_string(); + for name in nested_names { + path.push('.'); + path.push_str(&json_path_field(name)?); + } + Ok(PlannerResult::Planned(Expr::ScalarFunction( ScalarFunction::new_udf( json_get, @@ -185,6 +185,16 @@ impl ExprPlanner for JsonExprPlanner { } } +/// Quotes field names containing JSONPath punctuation, preserving literal keys. +fn json_path_field(name: &str) -> Result { + if !name.is_empty() && name.chars().all(|c| c.is_alphanumeric() || c == '_') { + Ok(name.to_string()) + } else { + // Reuse serde_json's string escaping. + serde_json::to_string(name).map_err(|e| DataFusionError::External(Box::new(e))) + } +} + enum JsonGetTypeResolution { Fallback, Typed(Vec<(usize, DataType)>), @@ -538,7 +548,7 @@ mod tests { assert_eq!( func.args[1], Expr::Literal( - ScalarValue::Utf8(Some("list[0][\"a.b\"]".to_string())), + ScalarValue::Utf8(Some(r#"list[0]."a.b""#.to_string())), None ) ); @@ -568,12 +578,49 @@ mod tests { ); assert_eq!( func.args[1], - Expr::Literal(ScalarValue::Utf8(Some("payload.cpu".to_string())), None) + Expr::Literal(ScalarValue::Utf8(Some("$.payload.cpu".to_string())), None) ); } other => panic!("expected json_get scalar function, got {other:?}"), } + for (key, path) in [ + ("http.status_code", r#"$."http.status_code""#), + ("a\"b", r#"$."a\"b""#), + ("a\\b", r#"$."a\\b""#), + ] { + let planned = planner.plan_compound_identifier( + &Field::new("labels", DataType::Struct(Fields::empty()), true) + .with_extension_type(Json2ExtensionType::default()), + Some(&qualifier), + &[key.to_string()], + )?; + let PlannerResult::Planned(Expr::ScalarFunction(func)) = planned else { + unreachable!() + }; + assert_eq!( + func.args[1], + Expr::Literal(ScalarValue::Utf8(Some(path.to_string())), None) + ); + } + + let PlannerResult::Planned(Expr::ScalarFunction(func)) = planner.plan_compound_identifier( + &Field::new("labels", DataType::Struct(Fields::empty()), true) + .with_extension_type(Json2ExtensionType::default()), + Some(&qualifier), + &["resource".to_string(), "http.status_code".to_string()], + )? + else { + unreachable!() + }; + assert_eq!( + func.args[1], + Expr::Literal( + ScalarValue::Utf8(Some(r#"$.resource."http.status_code""#.to_string())), + None + ) + ); + let original = planner.plan_compound_identifier( &Field::new("plain", DataType::Utf8, true), Some(&qualifier), diff --git a/src/query/src/optimizer/json_type_concretize.rs b/src/query/src/optimizer/json_type_concretize.rs index db94daa9e03..9051ec0a05c 100644 --- a/src/query/src/optimizer/json_type_concretize.rs +++ b/src/query/src/optimizer/json_type_concretize.rs @@ -16,7 +16,7 @@ use std::any::Any; use std::collections::HashMap; use arrow_schema::DataType; -use common_function::scalars::json::json_get::JsonGetWithType; +use common_function::scalars::json::json_get::{JsonGetWithType, parse_json_get_path}; use datafusion::datasource::{DefaultTableSource, TableProvider}; use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion}; use datafusion_common::{Result, plan_datafusion_err, plan_err}; @@ -24,6 +24,7 @@ use datafusion_expr::{Expr, LogicalPlan}; use datafusion_optimizer::{OptimizerConfig, OptimizerRule}; use datatypes::extension::json::is_json2_extension_type; use datatypes::types::json_type::{JsonNativeType, JsonObjectType}; +use jsonb::jsonpath::Path; use table::table::adapter::DfTableProviderAdapter; use crate::dummy_catalog::DummyTableProvider; @@ -181,10 +182,15 @@ fn deduce_json_type(expr: &Expr) -> Result> { ); }; - // Object-only type deduction cannot represent bracket JSONPath access, so preserve the - // full Variant and let json_get apply the expression. - if path.contains('[') { - return Ok(Some((column.name.clone(), JsonNativeType::Variant))); + let path = parse_json_get_path(path) + .map_err(|e| plan_datafusion_err!("Invalid JSONPath {path:?}: {e}"))?; + + if path + .paths + .iter() + .all(|segment| matches!(segment, Path::Root)) + { + return Ok(Some((column.name.clone(), JsonNativeType::String))); } let with_type = f @@ -196,18 +202,17 @@ fn deduce_json_type(expr: &Expr) -> Result> { let with_type = JsonNativeType::try_from(&with_type).map_err(|e| plan_datafusion_err!("{e:?}"))?; - let mut split = path.rsplit("."); - let Some(leaf) = split.next().filter(|&x| !x.is_empty() && x != "$") else { - return Ok(Some((column.name.clone(), JsonNativeType::String))); - }; - - let mut object = JsonObjectType::new(); - object.insert(leaf.to_string(), with_type); - let mut root = JsonNativeType::Object(object); - - for s in split { + let mut root = with_type; + for segment in path.paths.into_iter().rev() { + let name = match segment { + Path::Root => continue, + Path::DotField(name) | Path::ColonField(name) | Path::ObjectField(name) => name, + // A full JSONPath expression can select arrays or use filters/wildcards. + // Keep the entire value when an object projection cannot represent it. + _ => return Ok(Some((column.name.clone(), JsonNativeType::Variant))), + }; let mut object = JsonObjectType::new(); - object.insert(s.to_string(), root); + object.insert(name.into_owned(), root); root = JsonNativeType::Object(object); } @@ -328,13 +333,52 @@ mod tests { } #[test] - fn test_deduce_json_type_with_list_index() -> Result<()> { - let expr = json_get_expr(col("j"), path_expr("l[0]"), Some(DataType::Int64))?; + fn test_deduce_json_type_object_paths() -> Result<()> { + for paths in [ + ["a.b", r#"$."a"."b""#, r#"["a"]["b"]"#], + [r#"$."a.b"."c.d""#, r#"["a.b"]["c.d"]"#, r#"$."a.b"["c.d"]"#], + ] { + let deduce = |path| { + deduce_json_type(&json_get_expr( + col("j"), + path_expr(path), + Some(DataType::Int64), + )?) + }; + let expected = deduce(paths[0])?; + assert!(!matches!(expected, Some((_, JsonNativeType::Variant)))); + for path in &paths[1..] { + assert_eq!(deduce(path)?, expected); + } + } + Ok(()) + } - assert_eq!( - Some(("j".to_string(), JsonNativeType::Variant)), - deduce_json_type(&expr)? - ); + #[test] + fn test_deduce_json_type_invalid_path() -> Result<()> { + let expr = json_get_expr(col("j"), path_expr("$.a["), Some(DataType::Int64))?; + let err = deduce_json_type(&expr).unwrap_err(); + assert!(err.to_string().contains("Invalid JSONPath"), "{err}"); + Ok(()) + } + + #[test] + fn test_deduce_json_type_with_list_index() -> Result<()> { + for path in [ + "l[0]", + "$.l[0]", + "$.l[*]", + "$.o.*", + "$.l[0 to 2]", + "$.l ? (@.a == 1)", + ] { + let expr = json_get_expr(col("j"), path_expr(path), Some(DataType::Int64))?; + assert_eq!( + Some(("j".to_string(), JsonNativeType::Variant)), + deduce_json_type(&expr)?, + "{path}" + ); + } Ok(()) } diff --git a/src/query/src/planner.rs b/src/query/src/planner.rs index 9a577c26f9c..cc1c9996483 100644 --- a/src/query/src/planner.rs +++ b/src/query/src/planner.rs @@ -639,7 +639,7 @@ impl DfLogicalPlanner { /// /// DataFusion first resolves the leading `j.o.l` through /// `JsonExprPlanner::plan_compound_identifier`, which produces an untyped -/// `json_get` with path `o.l`. Before invoking `JsonExprPlanner::plan_field_access`, +/// `json_get` with path `$.o.l`. Before invoking `JsonExprPlanner::plan_field_access`, /// however, DataFusion eagerly converts every remaining access into a /// `GetFieldAccess`. It accepts string values but not [`SqlExpr::Identifier`]s /// in [`AccessExpr::Dot`] after a subscript. Without this normalization, that @@ -652,8 +652,8 @@ impl DfLogicalPlanner { /// changes neither the SQL text nor the dot accesses into subscript nodes: the /// resulting AST is conceptually `j.o.l[1].'inner'.'l'[2]`. DataFusion converts /// the string-valued dot accesses into named field accesses, which -/// `plan_field_access` safely encodes as bracket members. It can then extend the -/// JSON path to `o.l[1]["inner"]["l"][2]`. +/// `plan_field_access` safely encodes as dot members, quoting names when needed. It can then extend the +/// JSON path to `$.o.l[1].inner.l[2]`. /// /// This behavior is unchanged in the latest upstream releases checked here: /// DataFusion 55.0.0 and sqlparser 0.62.0. diff --git a/src/servers/src/error.rs b/src/servers/src/error.rs index fe1d226a10c..30997a8536a 100644 --- a/src/servers/src/error.rs +++ b/src/servers/src/error.rs @@ -686,6 +686,13 @@ pub enum Error { location: Location, }, + #[snafu(transparent)] + Operator { + source: operator::error::Error, + #[snafu(implicit)] + location: Location, + }, + #[snafu(transparent)] GreptimeProto { source: api::error::Error, @@ -853,6 +860,7 @@ impl ErrorExt for Error { GreptimeProto { source, .. } => source.status_code(), Partition { source, .. } => source.status_code(), MetricEngine { source, .. } => source.status_code(), + Operator { source, .. } => source.status_code(), SubmitBatch { source, .. } => source.status_code(), } } @@ -887,6 +895,7 @@ impl ErrorExt for Error { GreptimeProto { source, .. } => source.retry_hint(), Partition { source, .. } => source.retry_hint(), MetricEngine { source, .. } => source.retry_hint(), + Operator { source, .. } => source.retry_hint(), SubmitBatch { source, .. } => source.retry_hint(), MemoryLimitExceeded { source, .. } => source.retry_hint(), diff --git a/src/servers/src/http/jaeger.rs b/src/servers/src/http/jaeger.rs index c729f6d85d9..db843bf74e2 100644 --- a/src/servers/src/http/jaeger.rs +++ b/src/servers/src/http/jaeger.rs @@ -1052,11 +1052,21 @@ fn to_keyvalue(key: String, value: JsonValue) -> Option { value_type: ValueType::String, value: Value::String(value.clone()), }), - JsonValue::Number(value) => Some(KeyValue { - key, - value_type: ValueType::Int64, - value: Value::Int64(value.as_i64().unwrap_or(0)), - }), + JsonValue::Number(value) => { + if value.is_i64() { + Some(KeyValue { + key, + value_type: ValueType::Int64, + value: Value::Int64(value.as_i64()?), + }) + } else { + Some(KeyValue { + key, + value_type: ValueType::Float64, + value: Value::Float64(value.as_f64()?), + }) + } + } JsonValue::Bool(value) => Some(KeyValue { key, value_type: ValueType::Boolean, @@ -1201,6 +1211,23 @@ mod tests { use super::*; use crate::http::{ColumnSchema, HttpRecordsOutput, OutputSchema}; + #[test] + fn test_numeric_tag_types() { + for (input, value_type, value) in [ + (json!(42), ValueType::Int64, Value::Int64(42)), + (json!(1.5), ValueType::Float64, Value::Float64(1.5)), + ] { + assert_eq!( + to_keyvalue("tag".to_string(), input), + Some(KeyValue { + key: "tag".to_string(), + value_type, + value, + }) + ); + } + } + #[test] fn test_services_from_records() { // The tests is the tuple of `(test_records, expected)`. diff --git a/src/servers/src/otlp/trace.rs b/src/servers/src/otlp/trace.rs index aca52c23e7b..ca1fe1defd4 100644 --- a/src/servers/src/otlp/trace.rs +++ b/src/servers/src/otlp/trace.rs @@ -16,11 +16,12 @@ pub mod attributes; pub mod span; pub mod v0; pub mod v1; +mod v2; use std::collections::HashSet; use api::v1::RowInsertRequests; -// Column names. The `greptime_trace_v1` fixed columns are canonically defined +// Column names. The trace fixed columns are canonically defined // in `common_catalog::consts` (shared with the read-time graph derivation). pub use common_catalog::consts::{ DURATION_NANO_COLUMN, SERVICE_NAME_COLUMN, SPAN_KIND_COLUMN, @@ -29,18 +30,19 @@ pub use common_catalog::consts::{ pub use common_catalog::consts::{ PARENT_SPAN_ID_COLUMN, SPAN_ID_COLUMN, SPAN_NAME_COLUMN, TRACE_ID_COLUMN, }; -use pipeline::{GreptimePipelineParams, PipelineWay}; -use session::context::QueryContextRef; +use pipeline::PipelineWay; use crate::error::{NotSupportedSnafu, Result}; use crate::otlp::trace::span::TraceSpan; -use crate::query_handler::PipelineHandlerRef; +pub const TIMESTAMP_END_COLUMN: &str = "timestamp_end"; pub const SPAN_STATUS_MESSAGE_COLUMN: &str = "span_status_message"; pub const SPAN_ATTRIBUTES_COLUMN: &str = "span_attributes"; pub const SPAN_EVENTS_COLUMN: &str = "span_events"; pub const SCOPE_NAME_COLUMN: &str = "scope_name"; pub const SCOPE_VERSION_COLUMN: &str = "scope_version"; +/// Column containing instrumentation scope attributes as a JSON2 object. +pub const SCOPE_ATTRIBUTES_COLUMN: &str = "scope_attributes"; pub const RESOURCE_ATTRIBUTES_COLUMN: &str = "resource_attributes"; pub const TRACE_STATE_COLUMN: &str = "trace_state"; @@ -112,28 +114,12 @@ impl TraceAuxData { pub fn to_grpc_insert_requests_from_spans( spans: &[TraceSpan], pipeline: &PipelineWay, - pipeline_params: &GreptimePipelineParams, table_name: &str, - query_ctx: &QueryContextRef, - pipeline_handler: PipelineHandlerRef, ) -> Result<(RowInsertRequests, usize)> { match pipeline { - PipelineWay::OtlpTraceDirectV0 => v0::v0_to_grpc_main_insert_requests( - spans, - pipeline, - pipeline_params, - table_name, - query_ctx, - pipeline_handler, - ), - PipelineWay::OtlpTraceDirectV1 => v1::v1_to_grpc_main_insert_requests( - spans, - pipeline, - pipeline_params, - table_name, - query_ctx, - pipeline_handler, - ), + PipelineWay::OtlpTraceDirectV0 => v0::v0_to_grpc_main_insert_requests(spans, table_name), + PipelineWay::OtlpTraceDirectV1 => v1::v1_to_grpc_main_insert_requests(spans, table_name), + PipelineWay::OtlpTraceDirectV2 => v2::v2_to_grpc_main_insert_requests(spans, table_name), _ => NotSupportedSnafu { feat: "Unsupported pipeline for trace", } @@ -153,7 +139,9 @@ pub fn to_grpc_insert_requests_for_aux_tables( ) -> Result<(RowInsertRequests, usize)> { match pipeline { PipelineWay::OtlpTraceDirectV0 => v0::build_aux_table_requests(aux_data, table_name), - PipelineWay::OtlpTraceDirectV1 => v1::build_aux_table_requests(aux_data, table_name), + PipelineWay::OtlpTraceDirectV1 | PipelineWay::OtlpTraceDirectV2 => { + v1::build_aux_table_requests(aux_data, table_name) + } _ => NotSupportedSnafu { feat: "Unsupported pipeline for trace", } diff --git a/src/servers/src/otlp/trace/span.rs b/src/servers/src/otlp/trace/span.rs index fe8b37fed0a..37a6e28935f 100644 --- a/src/servers/src/otlp/trace/span.rs +++ b/src/servers/src/otlp/trace/span.rs @@ -265,9 +265,14 @@ pub fn parse(request: ExportTraceServiceRequest) -> TraceSpanGroups { .resource .map(|r| r.attributes) .unwrap_or_default(); + + // "service.name" is required; SDKs MUST provide a fallback when it is unset. + // This is in the specification: + // https://opentelemetry.io/docs/specs/semconv/resource/service/ + // Tolerate missing names from senders without using unrelated attributes as names. let service_name = resource_attrs .iter() - .find_or_first(|kv| kv.key == KEY_SERVICE_NAME) + .find(|kv| kv.key == KEY_SERVICE_NAME) .and_then(|kv| kv.value.clone()) .and_then(|v| match v.value { Some(any_value::Value::StringValue(s)) => Some(s), diff --git a/src/servers/src/otlp/trace/v0.rs b/src/servers/src/otlp/trace/v0.rs index 1bc6752e900..f3f7bdc29fa 100644 --- a/src/servers/src/otlp/trace/v0.rs +++ b/src/servers/src/otlp/trace/v0.rs @@ -18,19 +18,16 @@ use api::v1::value::ValueData; use api::v1::{ColumnDataType, RowInsertRequests}; use common_catalog::consts::{trace_operations_table_name, trace_services_table_name}; use common_grpc::precision::Precision; -use pipeline::{GreptimePipelineParams, PipelineWay}; -use session::context::QueryContextRef; use crate::error::Result; use crate::otlp::trace::span::TraceSpan; use crate::otlp::trace::{ DURATION_NANO_COLUMN, PARENT_SPAN_ID_COLUMN, SERVICE_NAME_COLUMN, SPAN_ATTRIBUTES_COLUMN, SPAN_EVENTS_COLUMN, SPAN_ID_COLUMN, SPAN_KIND_COLUMN, SPAN_NAME_COLUMN, SPAN_STATUS_CODE, - SPAN_STATUS_MESSAGE_COLUMN, TIMESTAMP_COLUMN, TRACE_ID_COLUMN, TRACE_STATE_COLUMN, - TraceAuxData, + SPAN_STATUS_MESSAGE_COLUMN, TIMESTAMP_COLUMN, TIMESTAMP_END_COLUMN, TRACE_ID_COLUMN, + TRACE_STATE_COLUMN, TraceAuxData, }; use crate::otlp::utils::{make_column_data, make_string_column_data}; -use crate::query_handler::PipelineHandlerRef; use crate::row_writer::{self, MultiTableData, TableData}; const APPROXIMATE_COLUMN_COUNT: usize = 24; @@ -44,11 +41,7 @@ const MAX_TIMESTAMP: i64 = 4102444800000000000; /// caller can update them only after the main span write succeeds. pub fn v0_to_grpc_main_insert_requests( spans: &[TraceSpan], - _pipeline: &PipelineWay, - _pipeline_params: &GreptimePipelineParams, table_name: &str, - _query_ctx: &QueryContextRef, - _pipeline_handler: PipelineHandlerRef, ) -> Result<(RowInsertRequests, usize)> { let mut multi_table_writer = MultiTableData::default(); let trace_writer = build_trace_table_data(spans)?; @@ -102,7 +95,7 @@ pub fn write_span_to_row(writer: &mut TableData, span: TraceSpan) -> Result<()> // write fields let fields = vec![ make_column_data( - "timestamp_end", + TIMESTAMP_END_COLUMN, ColumnDataType::TimestampNanosecond, Some(ValueData::TimestampNanosecondValue( span.end_in_nanosecond as i64, diff --git a/src/servers/src/otlp/trace/v1.rs b/src/servers/src/otlp/trace/v1.rs index f55ab48251e..bd556c4f056 100644 --- a/src/servers/src/otlp/trace/v1.rs +++ b/src/servers/src/otlp/trace/v1.rs @@ -24,8 +24,6 @@ use api::v1::{ use common_catalog::consts::{trace_operations_table_name, trace_services_table_name}; use common_grpc::precision::Precision; use opentelemetry_proto::tonic::common::v1::any_value::Value as OtlpValue; -use pipeline::{GreptimePipelineParams, PipelineWay}; -use session::context::QueryContextRef; use crate::error::Result; use crate::otlp::trace::attributes::Attributes; @@ -34,10 +32,9 @@ use crate::otlp::trace::{ DURATION_NANO_COLUMN, KEY_SERVICE_NAME, PARENT_SPAN_ID_COLUMN, SCOPE_NAME_COLUMN, SCOPE_VERSION_COLUMN, SERVICE_NAME_COLUMN, SPAN_EVENTS_COLUMN, SPAN_ID_COLUMN, SPAN_KIND_COLUMN, SPAN_NAME_COLUMN, SPAN_STATUS_CODE, SPAN_STATUS_MESSAGE_COLUMN, - TIMESTAMP_COLUMN, TRACE_ID_COLUMN, TRACE_STATE_COLUMN, TraceAuxData, + TIMESTAMP_COLUMN, TIMESTAMP_END_COLUMN, TRACE_ID_COLUMN, TRACE_STATE_COLUMN, TraceAuxData, }; use crate::otlp::utils::any_value_to_jsonb; -use crate::query_handler::PipelineHandlerRef; use crate::row_writer::{self, MultiTableData, TableData}; const APPROXIMATE_COLUMN_COUNT: usize = 30; @@ -78,7 +75,7 @@ impl FixedTraceColumnIndexes { Ok(Self { timestamp, - timestamp_end: field("timestamp_end", ColumnDataType::TimestampNanosecond)?, + timestamp_end: field(TIMESTAMP_END_COLUMN, ColumnDataType::TimestampNanosecond)?, duration_nano: field(DURATION_NANO_COLUMN, ColumnDataType::Int64)?, parent_span_id: field(PARENT_SPAN_ID_COLUMN, ColumnDataType::String)?, trace_id: field(TRACE_ID_COLUMN, ColumnDataType::String)?, @@ -249,11 +246,7 @@ impl TraceBatchSchema { /// caller can update them only after the main span write succeeds. pub fn v1_to_grpc_main_insert_requests( spans: &[TraceSpan], - _pipeline: &PipelineWay, - _pipeline_params: &GreptimePipelineParams, table_name: &str, - _query_ctx: &QueryContextRef, - _pipeline_handler: PipelineHandlerRef, ) -> Result<(RowInsertRequests, usize)> { let requests = v1_to_grpc_main_insert_requests_from_iter(spans.iter().cloned(), table_name)?; Ok((requests, spans.len())) @@ -352,7 +345,7 @@ pub fn write_span_to_row(writer: &mut TableData, span: TraceSpan) -> Result<()> /// that does not fit `i64` saturates at `i64::MAX`. Clamping at the source /// keeps new Int64 tables and existing UInt64 tables behaving identically: /// the written value is always a non-negative, in-range `i64`. -fn span_duration_nano(span: &TraceSpan) -> i64 { +pub(super) fn span_duration_nano(span: &TraceSpan) -> i64 { span.end_in_nanosecond .saturating_sub(span.start_in_nanosecond) .min(i64::MAX as u64) as i64 @@ -756,7 +749,7 @@ mod tests { Some(ValueData::TimestampNanosecondValue(1)), ), ( - "timestamp_end", + TIMESTAMP_END_COLUMN, Some(ValueData::TimestampNanosecondValue(2)), ), (DURATION_NANO_COLUMN, Some(ValueData::I64Value(1))), diff --git a/src/servers/src/otlp/trace/v2.rs b/src/servers/src/otlp/trace/v2.rs new file mode 100644 index 00000000000..7d2dfb40adb --- /dev/null +++ b/src/servers/src/otlp/trace/v2.rs @@ -0,0 +1,146 @@ +// Copyright 2023 Greptime Team +// +// 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 api::v1::value::ValueData; +use api::v1::{ColumnDataType, ColumnSchema, RowInsertRequests, SemanticType}; +use common_grpc::precision::Precision; +use snafu::ensure; + +use crate::error::{Result, TimestampOverflowSnafu}; +use crate::otlp::trace::span::TraceSpan; +use crate::otlp::trace::v1::span_duration_nano; +use crate::otlp::trace::{ + DURATION_NANO_COLUMN, PARENT_SPAN_ID_COLUMN, RESOURCE_ATTRIBUTES_COLUMN, + SCOPE_ATTRIBUTES_COLUMN, SCOPE_NAME_COLUMN, SCOPE_VERSION_COLUMN, SERVICE_NAME_COLUMN, + SPAN_ATTRIBUTES_COLUMN, SPAN_ID_COLUMN, SPAN_KIND_COLUMN, SPAN_NAME_COLUMN, SPAN_STATUS_CODE, + SPAN_STATUS_MESSAGE_COLUMN, TIMESTAMP_COLUMN, TIMESTAMP_END_COLUMN, TRACE_ID_COLUMN, + TRACE_STATE_COLUMN, +}; +use crate::otlp::utils::{make_column_data, make_string_column_data}; +use crate::row_writer::{self, MultiTableData, TableData}; + +// Preallocate for the fixed v2 schema: 3 timing columns, 3 span/trace IDs, +// 2 kind/name columns, 2 status columns, 1 trace state, 2 scope name/version +// columns, 1 service name, and 3 JSON2 attribute columns (17 total). +// Events and links are excluded until ARRAY(JSON2) is supported. +const APPROXIMATE_COLUMN_COUNT: usize = 17; + +/// Converts trace spans into row insert requests for the main v2 trace table. +pub(super) fn v2_to_grpc_main_insert_requests( + spans: &[TraceSpan], + table_name: &str, +) -> Result<(RowInsertRequests, usize)> { + let mut tables = MultiTableData::default(); + tables.add_table_data(table_name, build_trace_table_data(spans)?); + Ok(tables.into_row_insert_requests()) +} + +/// Builds the fixed row-oriented payload for the main v2 trace table. +fn build_trace_table_data(spans: &[TraceSpan]) -> Result { + let mut writer = TableData::new(APPROXIMATE_COLUMN_COUNT, spans.len()); + for span in spans.iter().cloned() { + write_span_to_row(&mut writer, span)?; + } + Ok(writer) +} + +fn write_span_to_row(writer: &mut TableData, span: TraceSpan) -> Result<()> { + ensure!( + span.start_in_nanosecond <= i64::MAX as u64, + TimestampOverflowSnafu { + error: "`span.start_in_nanosecond`", + } + ); + ensure!( + span.end_in_nanosecond <= i64::MAX as u64, + TimestampOverflowSnafu { + error: "`span.end_in_nanosecond`", + } + ); + let mut row = writer.alloc_one_row(); + + row_writer::write_ts_to_nanos( + writer, + TIMESTAMP_COLUMN, + Some(span.start_in_nanosecond as i64), + Precision::Nanosecond, + &mut row, + )?; + row_writer::write_fields( + writer, + vec![ + make_column_data( + TIMESTAMP_END_COLUMN, + ColumnDataType::TimestampNanosecond, + Some(ValueData::TimestampNanosecondValue( + span.end_in_nanosecond as i64, + )), + ), + make_column_data( + DURATION_NANO_COLUMN, + ColumnDataType::Int64, + Some(ValueData::I64Value(span_duration_nano(&span))), + ), + make_string_column_data(PARENT_SPAN_ID_COLUMN, span.parent_span_id), + make_string_column_data(TRACE_ID_COLUMN, Some(span.trace_id)), + make_string_column_data(SPAN_ID_COLUMN, Some(span.span_id)), + make_string_column_data(SPAN_KIND_COLUMN, Some(span.span_kind)), + make_string_column_data(SPAN_NAME_COLUMN, Some(span.span_name)), + make_string_column_data(SPAN_STATUS_CODE, Some(span.span_status_code)), + make_string_column_data(SPAN_STATUS_MESSAGE_COLUMN, Some(span.span_status_message)), + make_string_column_data(TRACE_STATE_COLUMN, Some(span.trace_state)), + make_string_column_data(SCOPE_NAME_COLUMN, Some(span.scope_name)), + make_string_column_data(SCOPE_VERSION_COLUMN, Some(span.scope_version)), + ] + .into_iter(), + &mut row, + )?; + row_writer::write_by_schema( + writer, + std::iter::once(( + ColumnSchema { + column_name: SERVICE_NAME_COLUMN.to_string(), + datatype: ColumnDataType::String as i32, + semantic_type: SemanticType::Tag as i32, + ..Default::default() + }, + span.service_name.map(ValueData::StringValue), + )), + &mut row, + )?; + + row_writer::write_json2( + writer, + SPAN_ATTRIBUTES_COLUMN, + span.span_attributes, + &mut row, + )?; + row_writer::write_json2( + writer, + SCOPE_ATTRIBUTES_COLUMN, + span.scope_attributes, + &mut row, + )?; + row_writer::write_json2( + writer, + RESOURCE_ATTRIBUTES_COLUMN, + span.resource_attributes, + &mut row, + )?; + + // TODO(LFC): Store span_events and span_links once ARRAY(JSON2) is supported. + + writer.add_row(row); + Ok(()) +} diff --git a/src/servers/src/row_writer.rs b/src/servers/src/row_writer.rs index 61fd9d77e5b..7fce0d60711 100644 --- a/src/servers/src/row_writer.rs +++ b/src/servers/src/row_writer.rs @@ -13,23 +13,33 @@ // limitations under the License. use std::collections::HashMap; +use std::sync::Arc; use ahash::{HashMap as AHashMap, HashMapExt}; +use api::helper::encode_json_value; use api::v1::column_data_type_extension::TypeExt; use api::v1::helper::time_index_column_schema; use api::v1::value::ValueData; use api::v1::{ - ColumnDataType, ColumnDataTypeExtension, ColumnSchema, JsonTypeExtension, Row, + ColumnDataType, ColumnDataTypeExtension, ColumnOptions, ColumnSchema, JsonTypeExtension, Row, RowInsertRequest, RowInsertRequests, Rows, SemanticType, Value, }; +use arrow_schema::extension::{ + EXTENSION_TYPE_METADATA_KEY, EXTENSION_TYPE_NAME_KEY, ExtensionType, +}; use common_grpc::precision::Precision; use common_time::Timestamp; use common_time::timestamp::TimeUnit; use common_time::timestamp::TimeUnit::Nanosecond; +use datatypes::extension::json::{Json2ExtensionType, JsonMetadata}; +use datatypes::json::JsonSettings; +use datatypes::value::Value as DataValue; +use serde::Serialize; use snafu::{OptionExt, ResultExt, ensure}; use crate::error::{ - IncompatibleSchemaSnafu, Result, RowWriterSnafu, TimePrecisionSnafu, TimestampOverflowSnafu, + ConvertScalarValueSnafu, IncompatibleSchemaSnafu, InternalSnafu, Result, RowWriterSnafu, + TimePrecisionSnafu, TimestampOverflowSnafu, ToJsonSnafu, }; /// The intermediate data structure for building the write request. @@ -317,6 +327,56 @@ pub fn write_json( ) } +fn build_json2_column_schema(name: impl ToString) -> ColumnSchema { + let extension = Json2ExtensionType::new(Arc::new(JsonMetadata::new(JsonSettings::new_v2()))); + let mut options = ColumnOptions::default(); + options.options.insert( + EXTENSION_TYPE_NAME_KEY.to_string(), + Json2ExtensionType::NAME.to_string(), + ); + if let Some(metadata) = extension.serialize_metadata() { + options + .options + .insert(EXTENSION_TYPE_METADATA_KEY.to_string(), metadata); + } + + ColumnSchema { + column_name: name.to_string(), + datatype: ColumnDataType::Json as i32, + semantic_type: SemanticType::Field as i32, + options: Some(options), + ..Default::default() + } +} + +/// Writes a serializable value into a JSON2 field using the default v2 layout. +pub(crate) fn write_json2( + table_data: &mut TableData, + name: impl ToString, + value: impl Serialize, + one_row: &mut Vec, +) -> Result<()> { + let json = serde_json::to_value(value).context(ToJsonSnafu)?; + let value = JsonSettings::new_v2() + .encode(json) + .context(ConvertScalarValueSnafu)?; + let DataValue::Json(value) = value else { + return InternalSnafu { + err_msg: "JSON2 encoding returned a non-JSON value", + } + .fail(); + }; + + write_by_schema( + table_data, + std::iter::once(( + build_json2_column_schema(name), + Some(ValueData::JsonValue(encode_json_value(*value))), + )), + one_row, + ) +} + pub(crate) fn write_by_schema( table_data: &mut TableData, kv_iter: impl Iterator)>, diff --git a/src/table/src/requests.rs b/src/table/src/requests.rs index a21a8f6fed8..c57dc0b009d 100644 --- a/src/table/src/requests.rs +++ b/src/table/src/requests.rs @@ -60,6 +60,8 @@ pub const FILE_TABLE_FORMAT_KEY: &str = "format"; pub const TABLE_DATA_MODEL: &str = "table_data_model"; pub const TABLE_DATA_MODEL_TRACE_V1: &str = "greptime_trace_v1"; +/// Table data model used by the JSON2-based OTLP trace pipeline. +pub const TABLE_DATA_MODEL_TRACE_V2: &str = "greptime_trace_v2"; /// Returns true if the table stores spans in the `greptime_trace_v1` data model /// (fixed span columns), the shape the Jaeger query path and the entity-graph diff --git a/tests-integration/src/otlp.rs b/tests-integration/src/otlp.rs index 5d40a8bd7cb..fe4dfb8a9e5 100644 --- a/tests-integration/src/otlp.rs +++ b/tests-integration/src/otlp.rs @@ -18,7 +18,9 @@ mod test { use client::{DEFAULT_CATALOG_NAME, OutputData}; use common_recordbatch::RecordBatches; + use datatypes::arrow::array::AsArray; use frontend::instance::Instance; + use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest; use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest; use otel_arrow_rust::proto::opentelemetry::common::v1::any_value::Value as Val; use otel_arrow_rust::proto::opentelemetry::common::v1::{ @@ -30,6 +32,8 @@ mod test { NumberDataPoint, ResourceMetrics, ScopeMetrics, Sum, metric, }; use otel_arrow_rust::proto::opentelemetry::resource::v1::Resource; + use pipeline::{GreptimePipelineParams, PipelineWay}; + use serde_json::json; use servers::query_handler::OpenTelemetryProtocolHandler; use servers::query_handler::sql::SqlQueryHandler; use session::context::QueryContext; @@ -37,6 +41,417 @@ mod test { use crate::standalone::GreptimeDbStandaloneBuilder; use crate::tests; + #[tokio::test(flavor = "multi_thread")] + async fn test_trace_v2_on_standalone() -> Result<(), Box> { + let standalone = GreptimeDbStandaloneBuilder::new("trace_v2_standalone") + .build() + .await; + test_trace_v2(standalone.fe_instance()).await?; + Ok(()) + } + + #[tokio::test(flavor = "multi_thread")] + async fn test_trace_v2_on_distributed() -> Result<(), Box> { + let distributed = tests::create_distributed_instance("trace_v2_distributed").await; + test_trace_v2(&distributed.frontend()).await?; + Ok(()) + } + + async fn test_trace_v2(instance: &Arc) -> Result<(), Box> { + let mut context = QueryContext::with(DEFAULT_CATALOG_NAME, "public"); + context.set_extension( + common_catalog::consts::TRACE_TABLE_NAME_SESSION_KEY, + "trace_v2", + ); + let ctx = Arc::new(context); + let request: ExportTraceServiceRequest = serde_json::from_value(json!({ + "resourceSpans": [{ + "resource": {"attributes": [ + {"key": "deployment.environment", "value": {"stringValue": "prod"}}, + {"key": "cloud.region", "value": {"stringValue": "us-west"}}, + {"key": "capacity", "value": {"intValue": "8"}} + ]}, + "scopeSpans": [{ + "scope": {"attributes": [ + {"key": "sample.rate", "value": {"doubleValue": 0.5}}, + {"key": "enabled", "value": {"boolValue": true}}, + {"key": "config", "value": {"kvlistValue": {"values": [ + {"key": "batch_size", "value": {"intValue": "4"}} + ]}}} + ]}, + "spans": [{ + "traceId": "c05d7a4ec8e1f231f02ed6e8da8655b4", + "spanId": "9630f2916e2f7909", "name": "op", "kind": 2, + "startTimeUnixNano": "1736480942444376000", + "endTimeUnixNano": "1736480942444499000", + "attributes": [ + {"key": "http.status_code", "value": {"intValue": "200"}}, + {"key": "latency", "value": {"doubleValue": 12.5}}, + {"key": "empty"}, + {"key": "bytes", "value": {"bytesValue": "AQID"}}, + {"key": "nested", "value": {"kvlistValue": {"values": [ + {"key": "a.b", "value": {"arrayValue": {"values": [{"boolValue": true}, {"stringValue": "ok"}]}}} + ]}}} + ] + }] + }] + }] + }))?; + let result = instance + .traces( + instance.clone(), + request.clone(), + PipelineWay::OtlpTraceDirectV2, + GreptimePipelineParams::default(), + "trace_v2".to_string(), + ctx.clone(), + ) + .await?; + assert_eq!((result.accepted_spans, result.rejected_spans), (1, 0)); + + // Check the complete schema created by the first export, before later writes. + let output = instance + .do_query("SHOW CREATE TABLE trace_v2", ctx.clone()) + .await; + let batches = match output + .into_iter() + .next() + .ok_or("Expected non-empty SQL output")?? + .data + { + OutputData::Stream(stream) => RecordBatches::try_collect(stream).await?, + OutputData::RecordBatches(batches) => batches, + OutputData::AffectedRows(_) => return Err("Expected SHOW CREATE TABLE rows".into()), + }; + let expected = r#"CREATE TABLE IF NOT EXISTS "trace_v2" ( + "timestamp" TIMESTAMP(9) NOT NULL, + "timestamp_end" TIMESTAMP(9) NULL, + "duration_nano" BIGINT NULL, + "parent_span_id" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'), + "trace_id" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'), + "span_id" STRING NULL, + "span_kind" STRING NULL, + "span_name" STRING NULL, + "span_status_code" STRING NULL, + "span_status_message" STRING NULL, + "trace_state" STRING NULL, + "scope_name" STRING NULL, + "scope_version" STRING NULL, + "service_name" STRING NULL SKIPPING INDEX WITH(false_positive_rate = '0.01', granularity = '10240', type = 'BLOOM'), + "span_attributes" JSON2( + max_auto_expanded_paths = 100 + ) NULL, + "scope_attributes" JSON2( + max_auto_expanded_paths = 100 + ) NULL, + "resource_attributes" JSON2( + max_auto_expanded_paths = 100 + ) NULL, + TIME INDEX ("timestamp"), + PRIMARY KEY ("service_name") +) +PARTITION ON COLUMNS ("trace_id") ( + trace_id < '1', + trace_id >= '1' AND trace_id < '2', + trace_id >= '2' AND trace_id < '3', + trace_id >= '3' AND trace_id < '4', + trace_id >= '4' AND trace_id < '5', + trace_id >= '5' AND trace_id < '6', + trace_id >= '6' AND trace_id < '7', + trace_id >= '7' AND trace_id < '8', + trace_id >= '8' AND trace_id < '9', + trace_id >= '9' AND trace_id < 'a', + trace_id >= 'a' AND trace_id < 'b', + trace_id >= 'b' AND trace_id < 'c', + trace_id >= 'c' AND trace_id < 'd', + trace_id >= 'd' AND trace_id < 'e', + trace_id >= 'e' AND trace_id < 'f', + trace_id >= 'f' +) +ENGINE=mito +WITH( + 'comment' = 'Created on insertion', + append_mode = 'true', + 'greptime.semantic.entity.service.id' = 'service_name', + 'greptime.semantic.pipeline' = 'greptime_trace_v2', + 'greptime.semantic.signal_type' = 'trace', + 'greptime.semantic.source' = 'opentelemetry', + 'greptime.semantic.trace.conventions' = 'unknown', + table_data_model = 'greptime_trace_v2' +)"#; + let batches = batches.take(); + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].num_rows(), 1); + assert_eq!(batches[0].num_columns(), 2); + assert_eq!( + batches[0].iter_column_as_string(1).collect::>(), + vec![Some(expected.to_string())], + ); + + // First write has no service.name: it must still create the tag, but no auxiliary rows. + assert_trace_v2_query( + instance, + "SELECT COUNT(*) = 1 + FROM trace_v2 + WHERE service_name IS NULL", + ctx.clone(), + ) + .await?; + assert_trace_v2_query( + instance, + "SELECT COUNT(*) = 17 + FROM information_schema.columns + WHERE table_name = 'trace_v2'", + ctx.clone(), + ) + .await?; + assert_trace_v2_query( + instance, + "SELECT COUNT(*) = 0 + FROM information_schema.tables + WHERE table_name IN ('trace_v2_services', 'trace_v2_operations')", + ctx.clone(), + ) + .await?; + assert_trace_v2_query( + instance, + r#"SELECT COUNT(*) = 1 + FROM trace_v2 + WHERE span_attributes."http.status_code"::BIGINT = 200 + AND span_attributes.empty::STRING IS NULL + AND span_attributes.bytes[1]::BIGINT = 2 + AND span_attributes.nested."a.b"[0]::BOOLEAN = true"#, + ctx.clone(), + ) + .await?; + + // Exercise each JSON2 column and expressions combining columns, before and after flush. + let json_queries = [ + r#"SELECT COUNT(*) > 0 AND COUNT(*) = COUNT(CASE + WHEN span_attributes.latency::DOUBLE * 2 = 25 + AND upper(span_attributes.nested."a.b"[1]::STRING) = 'OK' + AND span_attributes.missing::STRING IS NULL + THEN 1 END) + FROM trace_v2"#, + r#"SELECT COUNT(*) > 0 AND COUNT(*) = COUNT(CASE + WHEN scope_attributes."sample.rate"::DOUBLE + 0.25 = 0.75 + AND scope_attributes.enabled::BOOLEAN + AND scope_attributes.config.batch_size::BIGINT * 2 = 8 + THEN 1 END) + FROM trace_v2"#, + r#"SELECT COUNT(*) > 0 AND COUNT(*) = COUNT(CASE + WHEN upper(resource_attributes."cloud.region"::STRING) = 'US-WEST' + AND resource_attributes.capacity::BIGINT - 2 = 6 + THEN 1 END) + FROM trace_v2"#, + r#"SELECT AVG(span_attributes.latency::DOUBLE + * scope_attributes."sample.rate"::DOUBLE) = 6.25 + AND SUM(resource_attributes.capacity::BIGINT) + = SUM(scope_attributes.config.batch_size::BIGINT) * 2 + FROM trace_v2"#, + ]; + for sql in json_queries { + assert_trace_v2_query(instance, sql, ctx.clone()).await?; + } + + let mut mixed = request.clone(); + mixed.resource_spans[0] + .resource + .as_mut() + .ok_or("Expected non-empty resource in trace fixture")? + .attributes[0] + .key = "service.name".to_string(); + let spans = &mut mixed.resource_spans[0].scope_spans[0].spans; + // Reject one span in the first chunk, merge its healthy rows, then write the next chunk. + spans.resize(514, spans[0].clone()); + for (index, span) in spans.iter_mut().enumerate() { + span.span_id = (index as u64).to_be_bytes().to_vec(); + } + spans[0].attributes[0] + .value + .as_mut() + .ok_or("Expected non-empty attribute value in trace fixture")? + .value = Some( + opentelemetry_proto::tonic::common::v1::any_value::Value::StringValue("ok".to_string()), + ); + spans[511].end_time_unix_nano = u64::MAX; + let result = instance + .traces( + instance.clone(), + mixed, + PipelineWay::OtlpTraceDirectV2, + GreptimePipelineParams::default(), + "trace_v2".to_string(), + ctx.clone(), + ) + .await?; + assert_eq!((result.accepted_spans, result.rejected_spans), (513, 1)); + assert!( + result + .error_message + .as_deref() + .is_some_and(|s| s.contains("Timestamp overflow")) + ); + assert_trace_v2_query( + instance, + "SELECT COUNT(*) = 514 + FROM trace_v2", + ctx.clone(), + ) + .await?; + assert_trace_v2_query( + instance, + r#"SELECT COUNT(*) = 1 + FROM trace_v2 + WHERE span_attributes."http.status_code"::STRING = 'ok'"#, + ctx.clone(), + ) + .await?; + assert_trace_v2_query( + instance, + "SELECT COUNT(*) = 1 + FROM trace_v2_services + WHERE service_name = 'prod'", + ctx.clone(), + ) + .await?; + assert_trace_v2_query( + instance, + "SELECT COUNT(*) = 1 + FROM trace_v2_operations + WHERE service_name = 'prod'", + ctx.clone(), + ) + .await?; + + let mut invalid = request.clone(); + invalid.resource_spans[0].scope_spans[0].spans[0].start_time_unix_nano = u64::MAX; + let result = instance + .traces( + instance.clone(), + invalid, + PipelineWay::OtlpTraceDirectV2, + GreptimePipelineParams::default(), + "trace_v2".to_string(), + ctx.clone(), + ) + .await?; + assert_eq!((result.accepted_spans, result.rejected_spans), (0, 1)); + + // A v1 request must fail before auto-ALTER can flatten attributes into this table. + let error = instance + .traces( + instance.clone(), + request.clone(), + PipelineWay::OtlpTraceDirectV1, + GreptimePipelineParams::default(), + "trace_v2".to_string(), + ctx.clone(), + ) + .await + .err() + .ok_or("Expected mixed trace models to be rejected")?; + let error = format!("{error:?}"); + assert!( + error.contains("Trace table `trace_v2` uses greptime_trace_v2, but the request uses greptime_trace_v1"), + "{error}", + ); + assert_trace_v2_query( + instance, + "SELECT COUNT(*) = 17 + FROM information_schema.columns + WHERE table_name = 'trace_v2'", + ctx.clone(), + ) + .await?; + + let mut context = (*ctx).clone(); + context.set_extension( + common_catalog::consts::TRACE_TABLE_NAME_SESSION_KEY, + "trace_v1", + ); + let v1_ctx = Arc::new(context); + // Create the v1 table with an explicit service identity for the isolation check. + let mut v1_request = request.clone(); + v1_request.resource_spans[0] + .resource + .as_mut() + .ok_or("Expected non-empty resource in trace fixture")? + .attributes[0] + .key = "service.name".to_string(); + let result = instance + .traces( + instance.clone(), + v1_request, + PipelineWay::OtlpTraceDirectV1, + GreptimePipelineParams::default(), + "trace_v1".to_string(), + v1_ctx.clone(), + ) + .await?; + assert_eq!(result.accepted_spans, 1); + let error = instance + .traces( + instance.clone(), + request, + PipelineWay::OtlpTraceDirectV2, + GreptimePipelineParams::default(), + "trace_v1".to_string(), + v1_ctx, + ) + .await + .err() + .ok_or("Expected mixed trace models to be rejected")?; + let error = format!("{error:?}"); + assert!( + error.contains("Trace table `trace_v1` uses greptime_trace_v1, but the request uses greptime_trace_v2"), + "{error}", + ); + + let output = instance + .do_query("ADMIN FLUSH_TABLE('trace_v2')", ctx.clone()) + .await; + assert!(output.into_iter().all(|r| r.is_ok())); + assert_trace_v2_query( + instance, + r#"SELECT COUNT(*) = 513 + FROM trace_v2 + WHERE span_attributes."http.status_code"::BIGINT = 200"#, + ctx.clone(), + ) + .await?; + for sql in json_queries { + assert_trace_v2_query(instance, sql, ctx.clone()).await?; + } + Ok(()) + } + + async fn assert_trace_v2_query( + instance: &Arc, + sql: &str, + ctx: Arc, + ) -> Result<(), Box> { + let output = instance.do_query(sql, ctx).await; + let OutputData::Stream(stream) = output + .into_iter() + .next() + .ok_or("Expected non-empty SQL output")?? + .data + else { + return Err(format!("Expected a query stream: {sql}").into()); + }; + let batches = RecordBatches::try_collect(stream).await?; + let batches = batches.take(); + assert_eq!(batches.len(), 1, "{sql}"); + assert_eq!(batches[0].num_rows(), 1, "{sql}"); + assert_eq!( + batches[0].column(0).as_boolean().iter().collect::>(), + vec![Some(true)], + "{sql}" + ); + Ok(()) + } + #[tokio::test(flavor = "multi_thread")] pub async fn test_otlp_on_standalone() { let standalone = GreptimeDbStandaloneBuilder::new("test_standalone_otlp") diff --git a/tests-integration/tests/http.rs b/tests-integration/tests/http.rs index f8a4de57654..c6aa6d2f20d 100644 --- a/tests-integration/tests/http.rs +++ b/tests-integration/tests/http.rs @@ -56,7 +56,9 @@ use opentelemetry_proto::tonic::collector::metrics::v1::{ use opentelemetry_proto::tonic::collector::trace::v1::{ ExportTraceServiceRequest, ExportTraceServiceResponse, }; -use pipeline::GREPTIME_INTERNAL_TRACE_PIPELINE_V1_NAME; +use pipeline::{ + GREPTIME_INTERNAL_TRACE_PIPELINE_V1_NAME, GREPTIME_INTERNAL_TRACE_PIPELINE_V2_NAME, +}; use prost::Message; use rstest_reuse::apply; use serde_json::{Value, json}; @@ -104,7 +106,12 @@ macro_rules! http_test { async fn [< $test >]() { let store_type = tests_integration::test_util::StorageType::$service; if store_type.test_on() { - let _ = $crate::http::$test(store_type).await; + // Support both unit tests and fallible tests without discarding errors. + let result = $crate::http::$test(store_type).await; + assert_eq!( + std::process::Termination::report(result), + std::process::ExitCode::SUCCESS, + ); } } )* @@ -171,6 +178,7 @@ macro_rules! http_tests { test_otlp_metrics_resource_info_conflicts, test_otlp_traces_v0, test_otlp_traces_v1, + test_otlp_traces_v2, test_otlp_traces_v1_entity_graph, test_otlp_logs, test_loki_pb_logs, @@ -7727,6 +7735,123 @@ pub async fn test_otlp_traces_v0(store_type: StorageType) { guard.remove_all().await; } +/// Exercises v2 protobuf ingestion and JSON2 paths. +pub(crate) async fn test_otlp_traces_v2( + store_type: StorageType, +) -> Result<(), Box> { + common_telemetry::init_default_ut_logging(); + + let (app, mut guard) = + setup_test_http_app_with_frontend(store_type, "test_otlp_traces_v2").await; + let client = TestClient::new(app).await; + let table_name = "trace_v2_spans"; + + let request: ExportTraceServiceRequest = serde_json::from_value(json!({ + "resourceSpans": [{ + "resource": { + "attributes": [ + make_string_attr("service.name", "frontend"), + make_string_attr("deployment.environment", "production") + ] + }, + "scopeSpans": [{ + "scope": { + "name": "trace-v2-test", + "version": "1.0.0", + "attributes": [make_bool_attr("enabled", true)] + }, + "spans": [{ + "traceId": "c05d7a4ec8e1f231f02ed6e8da8655b4", + "spanId": "9630f2916e2f7909", + "name": "GET /api", + "kind": 2, + "startTimeUnixNano": "1736480942444376000", + "endTimeUnixNano": "1736480942444499000", + "attributes": [ + make_int_attr("http.status_code", 200), + make_string_attr("http.route", "/api"), + {"key": "ratio", "value": {"doubleValue": 1.5}}, + make_string_attr("a\"b", "quoted"), + make_string_attr("a\\b", "backslash") + ], + "events": [{ + "timeUnixNano": "1736480942444400000", + "name": "cache.hit", + "attributes": [make_int_attr("event.code", 7)] + }], + "links": [{ + "traceId": "cc9e0991a2e63d274984bd44ee669203", + "spanId": "8f847259b0f6e1ab", + "attributes": [make_string_attr("link.type", "follows_from")] + }], + "status": { "message": "", "code": 1 } + }] + }], + "schemaUrl": "https://opentelemetry.io/schemas/1.4.0" + }] + }))?; + + let response = send_req( + &client, + vec![ + ( + HeaderName::from_static("content-type"), + HeaderValue::from_static("application/x-protobuf"), + ), + ( + HeaderName::from_static("x-greptime-pipeline-name"), + HeaderValue::from_static(GREPTIME_INTERNAL_TRACE_PIPELINE_V2_NAME), + ), + ( + HeaderName::from_static("x-greptime-trace-table-name"), + HeaderValue::from_static(table_name), + ), + ], + "/v1/otlp/v1/traces", + request.encode_to_vec(), + false, + ) + .await; + assert_eq!(StatusCode::OK, response.status()); + + validate_data( + "otlp_traces_v2_json2", + &client, + "select service_name, span_attributes.\"http.status_code\"::BIGINT, \ + resource_attributes.\"deployment.environment\"::STRING, \ + scope_attributes.enabled::BOOLEAN from trace_v2_spans;", + r#"[["frontend",200,"production",true]]"#, + ) + .await; + validate_data( + "otlp_traces_v2_fixed_schema", + &client, + "select count(*) from information_schema.columns where table_name = 'trace_v2_spans' \ + and (column_name like 'span_attributes.%' \ + or column_name in ('span_events', 'span_links'));", + "[[0]]", + ) + .await; + validate_data( + "otlp_traces_v2_escaped_keys", + &client, + r#"select span_attributes."a""b"::STRING, span_attributes."a\b"::STRING from trace_v2_spans;"#, + r#"[["quoted","backslash"]]"#, + ).await; + validate_data( + "otlp_traces_v2_semantics", + &client, + "select count(*) from information_schema.tables where table_name = 'trace_v2_spans' \ + and create_options like '%table_data_model=greptime_trace_v2%' \ + and create_options like '%greptime.semantic.pipeline=greptime_trace_v2%';", + "[[1]]", + ) + .await; + + guard.remove_all().await; + Ok(()) +} + /// One real OTLP export must come out of `semantic_relationships` as the /// zero-configuration chain; resources missing an identity column derive /// nothing extra. diff --git a/tests/cases/standalone/common/function/json/json_get.result b/tests/cases/standalone/common/function/json/json_get.result index 3c58ca91b93..0eb5aa05c97 100644 --- a/tests/cases/standalone/common/function/json/json_get.result +++ b/tests/cases/standalone/common/function/json/json_get.result @@ -1,4 +1,17 @@ -- json_get functions -- +-- Invalid paths must report errors; a valid path with no match returns NULL. +SELECT json_get_int(parse_json('{"a": 1}'), '$.a['); + +Error: 3001(EngineExecuteQuery), Execution error: Invalid JSONPath "$.a[": InvalidJsonPath + +SELECT json_get_int(parse_json('{"a": 1}'), '$.missing'); + ++--------------------------------------------------------------+ +| json_get_int(parse_json(Utf8("{"a": 1}")),Utf8("$.missing")) | ++--------------------------------------------------------------+ +| | ++--------------------------------------------------------------+ + SELECT json_get_int(parse_json('{"a": {"b": {"c": 1}}}'), 'a.b.c'); +------------------------------------------------------------------------+ diff --git a/tests/cases/standalone/common/function/json/json_get.sql b/tests/cases/standalone/common/function/json/json_get.sql index f892a684622..18734e60ef8 100644 --- a/tests/cases/standalone/common/function/json/json_get.sql +++ b/tests/cases/standalone/common/function/json/json_get.sql @@ -1,4 +1,9 @@ -- json_get functions -- +-- Invalid paths must report errors; a valid path with no match returns NULL. +SELECT json_get_int(parse_json('{"a": 1}'), '$.a['); + +SELECT json_get_int(parse_json('{"a": 1}'), '$.missing'); + SELECT json_get_int(parse_json('{"a": {"b": {"c": 1}}}'), 'a.b.c'); SELECT json_get_float(parse_json('{"a": {"b": {"c": 1.234}}}'), 'a:b.c'); diff --git a/tests/cases/standalone/common/types/json/json2.result b/tests/cases/standalone/common/types/json/json2.result index a6178496004..493b1d85c16 100644 --- a/tests/cases/standalone/common/types/json/json2.result +++ b/tests/cases/standalone/common/types/json/json2.result @@ -73,83 +73,83 @@ Affected Rows: 2 -- SQLNESS REPLACE (peers.*) REDACTED explain select j.a.b from json2_table; -+---------------+---------------------------------------------------+ -| plan_type | plan | -+---------------+---------------------------------------------------+ -| logical_plan | MergeScan [is_placeholder=false, remote_input=[ | -| | Projection: json_get(json2_table.j, Utf8("a.b")) | -| | TableScan: json2_table | -| | ]] | -| physical_plan | CooperativeExec | ++---------------+----------------------------------------------------+ +| plan_type | plan | ++---------------+----------------------------------------------------+ +| logical_plan | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: json_get(json2_table.j, Utf8("$.a.b")) | +| | TableScan: json2_table | +| | ]] | +| physical_plan | CooperativeExec | | | MergeScanExec: REDACTED -| | | -+---------------+---------------------------------------------------+ +| | | ++---------------+----------------------------------------------------+ -- SQLNESS REPLACE (peers.*) REDACTED explain select j.a.x::bool from json2_table; -+---------------+------------------------------------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+------------------------------------------------------------------------------------------------------------------------------------+ -| logical_plan | MergeScan [is_placeholder=false, remote_input=[ | -| | Projection: json_get(json2_table.j, Utf8("a.x"), Boolean(NULL)) AS arrow_cast(json_get(json2_table.j,Utf8("a.x")),Utf8("Boolean")) | -| | TableScan: json2_table | -| | ]] | -| physical_plan | CooperativeExec | ++---------------+----------------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+----------------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: json_get(json2_table.j, Utf8("$.a.x"), Boolean(NULL)) AS arrow_cast(json_get(json2_table.j,Utf8("$.a.x")),Utf8("Boolean")) | +| | TableScan: json2_table | +| | ]] | +| physical_plan | CooperativeExec | | | MergeScanExec: REDACTED -| | | -+---------------+------------------------------------------------------------------------------------------------------------------------------------+ +| | | ++---------------+----------------------------------------------------------------------------------------------------------------------------------------+ select j.a.b from json2_table order by ts; -+-------------------------------------+ -| json_get(json2_table.j,Utf8("a.b")) | -+-------------------------------------+ -| 1 | -| -2 | -| 3 | -| -4 | -| | -| | -| s7 | -| 8 | -| | -| 10 | -+-------------------------------------+ ++---------------------------------------+ +| json_get(json2_table.j,Utf8("$.a.b")) | ++---------------------------------------+ +| 1 | +| -2 | +| 3 | +| -4 | +| | +| | +| s7 | +| 8 | +| | +| 10 | ++---------------------------------------+ select j.a, j.a.x from json2_table order by ts; -+-----------------------------------+-------------------------------------+ -| json_get(json2_table.j,Utf8("a")) | json_get(json2_table.j,Utf8("a.x")) | -+-----------------------------------+-------------------------------------+ -| {"b":1} | | -| {"b":-2} | | -| {"b":3} | | -| {"b":-4} | | -| {} | | -| | | -| {"b":"s7"} | | -| {"b":8} | | -| {"x":true} | true | -| {"b":10} | | -+-----------------------------------+-------------------------------------+ ++-------------------------------------+---------------------------------------+ +| json_get(json2_table.j,Utf8("$.a")) | json_get(json2_table.j,Utf8("$.a.x")) | ++-------------------------------------+---------------------------------------+ +| {"b":1} | | +| {"b":-2} | | +| {"b":3} | | +| {"b":-4} | | +| {} | | +| | | +| {"b":"s7"} | | +| {"b":8} | | +| {"x":true} | true | +| {"b":10} | | ++-------------------------------------+---------------------------------------+ select j, j.a from json2_table order by ts; -+--------------------------------------------------+-----------------------------------+ -| j | json_get(json2_table.j,Utf8("a")) | -+--------------------------------------------------+-----------------------------------+ -| {"a":{"b":1},"c":"s1","d":[{"e":{"f":0.1}}]} | {"b":1} | -| {"a":{"b":-2},"c":"s2","d":[{"e":{"f":0.2}}]} | {"b":-2} | -| {"a":{"b":3},"c":"s3"} | {"b":3} | -| {"a":{"b":-4},"d":[{"e":{"g":-0.4}}]} | {"b":-4} | -| {"a":{},"c":"s5"} | {} | -| {"c":"s6"} | | -| {"a":{"b":"s7"},"c":[1],"d":[{"e":{"g":-0.7}}]} | {"b":"s7"} | -| {"a":{"b":8},"c":"s8"} | {"b":8} | -| {"a":{"x":true},"c":"s9","d":[{"e":{"g":-0.9}}]} | {"x":true} | -| {"a":{"b":10},"y":false} | {"b":10} | -+--------------------------------------------------+-----------------------------------+ ++--------------------------------------------------+-------------------------------------+ +| j | json_get(json2_table.j,Utf8("$.a")) | ++--------------------------------------------------+-------------------------------------+ +| {"a":{"b":1},"c":"s1","d":[{"e":{"f":0.1}}]} | {"b":1} | +| {"a":{"b":-2},"c":"s2","d":[{"e":{"f":0.2}}]} | {"b":-2} | +| {"a":{"b":3},"c":"s3"} | {"b":3} | +| {"a":{"b":-4},"d":[{"e":{"g":-0.4}}]} | {"b":-4} | +| {"a":{},"c":"s5"} | {} | +| {"c":"s6"} | | +| {"a":{"b":"s7"},"c":[1],"d":[{"e":{"g":-0.7}}]} | {"b":"s7"} | +| {"a":{"b":8},"c":"s8"} | {"b":8} | +| {"a":{"x":true},"c":"s9","d":[{"e":{"g":-0.9}}]} | {"x":true} | +| {"a":{"b":10},"y":false} | {"b":10} | ++--------------------------------------------------+-------------------------------------+ select j from json2_table where j.a.b = 1; @@ -161,20 +161,20 @@ select j from json2_table where j.a.b = 1; select j.c, j.y from json2_table order by ts; -+-----------------------------------+-----------------------------------+ -| json_get(json2_table.j,Utf8("c")) | json_get(json2_table.j,Utf8("y")) | -+-----------------------------------+-----------------------------------+ -| s1 | | -| s2 | | -| s3 | | -| | | -| s5 | | -| s6 | | -| [1] | | -| s8 | | -| s9 | | -| | false | -+-----------------------------------+-----------------------------------+ ++-------------------------------------+-------------------------------------+ +| json_get(json2_table.j,Utf8("$.c")) | json_get(json2_table.j,Utf8("$.y")) | ++-------------------------------------+-------------------------------------+ +| s1 | | +| s2 | | +| s3 | | +| | | +| s5 | | +| s6 | | +| [1] | | +| s8 | | +| s9 | | +| | false | ++-------------------------------------+-------------------------------------+ select j from json2_table order by ts; @@ -301,71 +301,45 @@ select json_get(j, '$') from json2_table order by ts; select json_get(j, '.') from json2_table order by ts; -+--------------------------------------------------+ -| json_get(json2_table.j,Utf8(".")) | -+--------------------------------------------------+ -| {"a":{"b":1},"c":"s1","d":[{"e":{"f":0.1}}]} | -| {"a":{"b":-2},"c":"s2","d":[{"e":{"f":0.2}}]} | -| {"a":{"b":3},"c":"s3"} | -| {"a":{"b":-4},"d":[{"e":{"g":-0.4}}]} | -| {"a":{},"c":"s5"} | -| {"c":"s6"} | -| {"a":{"b":"s7"},"c":[1],"d":[{"e":{"g":-0.7}}]} | -| {"a":{"b":8},"c":"s8"} | -| {"a":{"x":true},"c":"s9","d":[{"e":{"g":-0.9}}]} | -| {"a":{"b":10},"y":false} | -+--------------------------------------------------+ +Error: 3000(PlanQuery), Error during planning: Invalid JSONPath ".": InvalidJsonPath select json_get(j, '$.') from json2_table order by ts; -+--------------------------------------------------+ -| json_get(json2_table.j,Utf8("$.")) | -+--------------------------------------------------+ -| {"a":{"b":1},"c":"s1","d":[{"e":{"f":0.1}}]} | -| {"a":{"b":-2},"c":"s2","d":[{"e":{"f":0.2}}]} | -| {"a":{"b":3},"c":"s3"} | -| {"a":{"b":-4},"d":[{"e":{"g":-0.4}}]} | -| {"a":{},"c":"s5"} | -| {"c":"s6"} | -| {"a":{"b":"s7"},"c":[1],"d":[{"e":{"g":-0.7}}]} | -| {"a":{"b":8},"c":"s8"} | -| {"a":{"x":true},"c":"s9","d":[{"e":{"g":-0.9}}]} | -| {"a":{"b":10},"y":false} | -+--------------------------------------------------+ +Error: 3000(PlanQuery), Error during planning: Invalid JSONPath "$.": InvalidJsonPath select j.a.b + 1 from json2_table order by ts; -+------------------------------------------------------------+ -| json_get(json2_table.j,Utf8("a.b"),Int64(NULL)) + Int64(1) | -+------------------------------------------------------------+ -| 2 | -| -1 | -| 4 | -| -3 | -| | -| | -| | -| 9 | -| | -| 11 | -+------------------------------------------------------------+ ++--------------------------------------------------------------+ +| json_get(json2_table.j,Utf8("$.a.b"),Int64(NULL)) + Int64(1) | ++--------------------------------------------------------------+ +| 2 | +| -1 | +| 4 | +| -3 | +| | +| | +| | +| 9 | +| | +| 11 | ++--------------------------------------------------------------+ select j.d from json2_table order by ts; -+-----------------------------------+ -| json_get(json2_table.j,Utf8("d")) | -+-----------------------------------+ -| [{"e":{"f":0.1}}] | -| [{"e":{"f":0.2}}] | -| | -| [{"e":{"g":-0.4}}] | -| | -| | -| [{"e":{"g":-0.7}}] | -| | -| [{"e":{"g":-0.9}}] | -| | -+-----------------------------------+ ++-------------------------------------+ +| json_get(json2_table.j,Utf8("$.d")) | ++-------------------------------------+ +| [{"e":{"f":0.1}}] | +| [{"e":{"f":0.2}}] | +| | +| [{"e":{"g":-0.4}}] | +| | +| | +| [{"e":{"g":-0.7}}] | +| | +| [{"e":{"g":-0.9}}] | +| | ++-------------------------------------+ drop table json2_table; @@ -394,13 +368,13 @@ select ts, j.payload, j.payload is null as payload_is_null from json2_variant_null order by ts; -+-------------------------+------------------------------------------------+-----------------+ -| ts | json_get(json2_variant_null.j,Utf8("payload")) | payload_is_null | -+-------------------------+------------------------------------------------+-----------------+ -| 1970-01-01T00:00:00.001 | {"value":1} | false | -| 1970-01-01T00:00:00.002 | text | false | -| 1970-01-01T00:00:00.003 | | true | -+-------------------------+------------------------------------------------+-----------------+ ++-------------------------+--------------------------------------------------+-----------------+ +| ts | json_get(json2_variant_null.j,Utf8("$.payload")) | payload_is_null | ++-------------------------+--------------------------------------------------+-----------------+ +| 1970-01-01T00:00:00.001 | {"value":1} | false | +| 1970-01-01T00:00:00.002 | text | false | +| 1970-01-01T00:00:00.003 | | true | ++-------------------------+--------------------------------------------------+-----------------+ admin flush_table('json2_variant_null'); @@ -414,13 +388,13 @@ select ts, j.payload, j.payload is null as payload_is_null from json2_variant_null order by ts; -+-------------------------+------------------------------------------------+-----------------+ -| ts | json_get(json2_variant_null.j,Utf8("payload")) | payload_is_null | -+-------------------------+------------------------------------------------+-----------------+ -| 1970-01-01T00:00:00.001 | {"value":1} | false | -| 1970-01-01T00:00:00.002 | text | false | -| 1970-01-01T00:00:00.003 | | true | -+-------------------------+------------------------------------------------+-----------------+ ++-------------------------+--------------------------------------------------+-----------------+ +| ts | json_get(json2_variant_null.j,Utf8("$.payload")) | payload_is_null | ++-------------------------+--------------------------------------------------+-----------------+ +| 1970-01-01T00:00:00.001 | {"value":1} | false | +| 1970-01-01T00:00:00.002 | text | false | +| 1970-01-01T00:00:00.003 | | true | ++-------------------------+--------------------------------------------------+-----------------+ drop table json2_variant_null; @@ -499,14 +473,14 @@ select from json2_finite_paths order by ts; -+-------------------------+-------------------------------------------+---------------------------------------------+-------+------+-------------------------------------------------+ -| ts | j | json_get(json2_finite_paths.j,Utf8("hint")) | alpha | beta | json_get(json2_finite_paths.j,Utf8("conflict")) | -+-------------------------+-------------------------------------------+---------------------------------------------+-------+------+-------------------------------------------------+ -| 1970-01-01T00:00:00.001 | {"alpha":1,"conflict":1,"hint":"h1"} | h1 | 1 | | 1 | -| 1970-01-01T00:00:00.002 | {"alpha":2,"conflict":"text","hint":"h2"} | h2 | 2 | | text | -| 1970-01-01T00:00:00.003 | {"beta":3,"conflict":true,"hint":"h3"} | h3 | | 3 | true | -| 1970-01-01T00:00:00.004 | {"beta":4,"conflict":"other","hint":"h4"} | h4 | | 4 | other | -+-------------------------+-------------------------------------------+---------------------------------------------+-------+------+-------------------------------------------------+ ++-------------------------+-------------------------------------------+-----------------------------------------------+-------+------+---------------------------------------------------+ +| ts | j | json_get(json2_finite_paths.j,Utf8("$.hint")) | alpha | beta | json_get(json2_finite_paths.j,Utf8("$.conflict")) | ++-------------------------+-------------------------------------------+-----------------------------------------------+-------+------+---------------------------------------------------+ +| 1970-01-01T00:00:00.001 | {"alpha":1,"conflict":1,"hint":"h1"} | h1 | 1 | | 1 | +| 1970-01-01T00:00:00.002 | {"alpha":2,"conflict":"text","hint":"h2"} | h2 | 2 | | text | +| 1970-01-01T00:00:00.003 | {"beta":3,"conflict":true,"hint":"h3"} | h3 | | 3 | true | +| 1970-01-01T00:00:00.004 | {"beta":4,"conflict":"other","hint":"h4"} | h4 | | 4 | other | ++-------------------------+-------------------------------------------+-----------------------------------------------+-------+------+---------------------------------------------------+ admin compact_table('json2_finite_paths'); @@ -526,14 +500,14 @@ select from json2_finite_paths order by ts; -+-------------------------+-------------------------------------------+---------------------------------------------+-------+------+-------------------------------------------------+ -| ts | j | json_get(json2_finite_paths.j,Utf8("hint")) | alpha | beta | json_get(json2_finite_paths.j,Utf8("conflict")) | -+-------------------------+-------------------------------------------+---------------------------------------------+-------+------+-------------------------------------------------+ -| 1970-01-01T00:00:00.001 | {"alpha":1,"conflict":1,"hint":"h1"} | h1 | 1 | | 1 | -| 1970-01-01T00:00:00.002 | {"alpha":2,"conflict":"text","hint":"h2"} | h2 | 2 | | text | -| 1970-01-01T00:00:00.003 | {"beta":3,"conflict":true,"hint":"h3"} | h3 | | 3 | true | -| 1970-01-01T00:00:00.004 | {"beta":4,"conflict":"other","hint":"h4"} | h4 | | 4 | other | -+-------------------------+-------------------------------------------+---------------------------------------------+-------+------+-------------------------------------------------+ ++-------------------------+-------------------------------------------+-----------------------------------------------+-------+------+---------------------------------------------------+ +| ts | j | json_get(json2_finite_paths.j,Utf8("$.hint")) | alpha | beta | json_get(json2_finite_paths.j,Utf8("$.conflict")) | ++-------------------------+-------------------------------------------+-----------------------------------------------+-------+------+---------------------------------------------------+ +| 1970-01-01T00:00:00.001 | {"alpha":1,"conflict":1,"hint":"h1"} | h1 | 1 | | 1 | +| 1970-01-01T00:00:00.002 | {"alpha":2,"conflict":"text","hint":"h2"} | h2 | 2 | | text | +| 1970-01-01T00:00:00.003 | {"beta":3,"conflict":true,"hint":"h3"} | h3 | | 3 | true | +| 1970-01-01T00:00:00.004 | {"beta":4,"conflict":"other","hint":"h4"} | h4 | | 4 | other | ++-------------------------+-------------------------------------------+-----------------------------------------------+-------+------+---------------------------------------------------+ drop table json2_finite_paths; diff --git a/tests/cases/standalone/common/types/json/json2_deep_nesting.result b/tests/cases/standalone/common/types/json/json2_deep_nesting.result index 1f91ac497d4..39fef8bd543 100644 --- a/tests/cases/standalone/common/types/json/json2_deep_nesting.result +++ b/tests/cases/standalone/common/types/json/json2_deep_nesting.result @@ -19,7 +19,7 @@ Affected Rows: 1 select j.level_01 from json2_depth_49; +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| json_get(json2_depth_49.j,Utf8("level_01")) | +| json_get(json2_depth_49.j,Utf8("$.level_01")) | +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | {"level_02":{"level_03":{"level_04":{"level_05":{"level_06":{"level_07":{"level_08":{"level_09":{"level_10":{"level_11":{"level_12":{"level_13":{"level_14":{"level_15":{"level_16":{"level_17":{"level_18":{"level_19":{"level_20":{"level_21":{"level_22":{"level_23":{"level_24":{"level_25":{"level_26":{"level_27":{"level_28":{"level_29":{"level_30":{"level_31":{"level_32":{"level_33":{"level_34":{"level_35":{"level_36":{"level_37":{"level_38":{"level_39":{"level_40":{"level_41":{"level_42":{"level_43":{"level_44":{"level_45":{"level_46":{"level_47":{"level_48":{"level_49":1}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}} | +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ @@ -35,7 +35,7 @@ admin flush_table('json2_depth_49'); select j.level_01 from json2_depth_49; +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| json_get(json2_depth_49.j,Utf8("level_01")) | +| json_get(json2_depth_49.j,Utf8("$.level_01")) | +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | {"level_02":{"level_03":{"level_04":{"level_05":{"level_06":{"level_07":{"level_08":{"level_09":{"level_10":{"level_11":{"level_12":{"level_13":{"level_14":{"level_15":{"level_16":{"level_17":{"level_18":{"level_19":{"level_20":{"level_21":{"level_22":{"level_23":{"level_24":{"level_25":{"level_26":{"level_27":{"level_28":{"level_29":{"level_30":{"level_31":{"level_32":{"level_33":{"level_34":{"level_35":{"level_36":{"level_37":{"level_38":{"level_39":{"level_40":{"level_41":{"level_42":{"level_43":{"level_44":{"level_45":{"level_46":{"level_47":{"level_48":{"level_49":1}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}} | +---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ @@ -65,7 +65,7 @@ Affected Rows: 1 select j.level_01 from json2_depth_50; +----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| json_get(json2_depth_50.j,Utf8("level_01")) | +| json_get(json2_depth_50.j,Utf8("$.level_01")) | +----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | {"level_02":{"level_03":{"level_04":{"level_05":{"level_06":{"level_07":{"level_08":{"level_09":{"level_10":{"level_11":{"level_12":{"level_13":{"level_14":{"level_15":{"level_16":{"level_17":{"level_18":{"level_19":{"level_20":{"level_21":{"level_22":{"level_23":{"level_24":{"level_25":{"level_26":{"level_27":{"level_28":{"level_29":{"level_30":{"level_31":{"level_32":{"level_33":{"level_34":{"level_35":{"level_36":{"level_37":{"level_38":{"level_39":{"level_40":{"level_41":{"level_42":{"level_43":{"level_44":{"level_45":{"level_46":{"level_47":{"level_48":{"level_49":{"level_50":1}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}} | +----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ @@ -81,7 +81,7 @@ admin flush_table('json2_depth_50'); select j.level_01 from json2_depth_50; +----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| json_get(json2_depth_50.j,Utf8("level_01")) | +| json_get(json2_depth_50.j,Utf8("$.level_01")) | +----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | {"level_02":{"level_03":{"level_04":{"level_05":{"level_06":{"level_07":{"level_08":{"level_09":{"level_10":{"level_11":{"level_12":{"level_13":{"level_14":{"level_15":{"level_16":{"level_17":{"level_18":{"level_19":{"level_20":{"level_21":{"level_22":{"level_23":{"level_24":{"level_25":{"level_26":{"level_27":{"level_28":{"level_29":{"level_30":{"level_31":{"level_32":{"level_33":{"level_34":{"level_35":{"level_36":{"level_37":{"level_38":{"level_39":{"level_40":{"level_41":{"level_42":{"level_43":{"level_44":{"level_45":{"level_46":{"level_47":{"level_48":{"level_49":{"level_50":1}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}} | +----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ @@ -111,7 +111,7 @@ Affected Rows: 1 select j.level_01 from json2_depth_51; +-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| json_get(json2_depth_51.j,Utf8("level_01")) | +| json_get(json2_depth_51.j,Utf8("$.level_01")) | +-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | {"level_02":{"level_03":{"level_04":{"level_05":{"level_06":{"level_07":{"level_08":{"level_09":{"level_10":{"level_11":{"level_12":{"level_13":{"level_14":{"level_15":{"level_16":{"level_17":{"level_18":{"level_19":{"level_20":{"level_21":{"level_22":{"level_23":{"level_24":{"level_25":{"level_26":{"level_27":{"level_28":{"level_29":{"level_30":{"level_31":{"level_32":{"level_33":{"level_34":{"level_35":{"level_36":{"level_37":{"level_38":{"level_39":{"level_40":{"level_41":{"level_42":{"level_43":{"level_44":{"level_45":{"level_46":{"level_47":{"level_48":{"level_49":{"level_50":{"level_51":1}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}} | +-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ @@ -144,7 +144,7 @@ admin flush_table('json2_depth_51'); select j.level_01 from json2_depth_51; +-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| json_get(json2_depth_51.j,Utf8("level_01")) | +| json_get(json2_depth_51.j,Utf8("$.level_01")) | +-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | {"level_02":{"level_03":{"level_04":{"level_05":{"level_06":{"level_07":{"level_08":{"level_09":{"level_10":{"level_11":{"level_12":{"level_13":{"level_14":{"level_15":{"level_16":{"level_17":{"level_18":{"level_19":{"level_20":{"level_21":{"level_22":{"level_23":{"level_24":{"level_25":{"level_26":{"level_27":{"level_28":{"level_29":{"level_30":{"level_31":{"level_32":{"level_33":{"level_34":{"level_35":{"level_36":{"level_37":{"level_38":{"level_39":{"level_40":{"level_41":{"level_42":{"level_43":{"level_44":{"level_45":{"level_46":{"level_47":{"level_48":{"level_49":{"level_50":{"level_51":1}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}} | +-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ @@ -190,7 +190,7 @@ Affected Rows: 1 select j.level_01 from json2_depth_70; +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| json_get(json2_depth_70.j,Utf8("level_01")) | +| json_get(json2_depth_70.j,Utf8("$.level_01")) | +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | {"level_02":{"level_03":{"level_04":{"level_05":{"level_06":{"level_07":{"level_08":{"level_09":{"level_10":{"level_11":{"level_12":{"level_13":{"level_14":{"level_15":{"level_16":{"level_17":{"level_18":{"level_19":{"level_20":{"level_21":{"level_22":{"level_23":{"level_24":{"level_25":{"level_26":{"level_27":{"level_28":{"level_29":{"level_30":{"level_31":{"level_32":{"level_33":{"level_34":{"level_35":{"level_36":{"level_37":{"level_38":{"level_39":{"level_40":{"level_41":{"level_42":{"level_43":{"level_44":{"level_45":{"level_46":{"level_47":{"level_48":{"level_49":{"level_50":{"level_51":{"level_52":{"level_53":{"level_54":{"level_55":{"level_56":{"level_57":{"level_58":{"level_59":{"level_60":{"level_61":{"level_62":{"level_63":{"level_64":{"level_65":{"level_66":{"level_67":{"level_68":{"level_69":{"level_70":1}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}} | +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ @@ -215,7 +215,7 @@ admin flush_table('json2_depth_70'); select j.level_01 from json2_depth_70; +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ -| json_get(json2_depth_70.j,Utf8("level_01")) | +| json_get(json2_depth_70.j,Utf8("$.level_01")) | +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ | {"level_02":{"level_03":{"level_04":{"level_05":{"level_06":{"level_07":{"level_08":{"level_09":{"level_10":{"level_11":{"level_12":{"level_13":{"level_14":{"level_15":{"level_16":{"level_17":{"level_18":{"level_19":{"level_20":{"level_21":{"level_22":{"level_23":{"level_24":{"level_25":{"level_26":{"level_27":{"level_28":{"level_29":{"level_30":{"level_31":{"level_32":{"level_33":{"level_34":{"level_35":{"level_36":{"level_37":{"level_38":{"level_39":{"level_40":{"level_41":{"level_42":{"level_43":{"level_44":{"level_45":{"level_46":{"level_47":{"level_48":{"level_49":{"level_50":{"level_51":{"level_52":{"level_53":{"level_54":{"level_55":{"level_56":{"level_57":{"level_58":{"level_59":{"level_60":{"level_61":{"level_62":{"level_63":{"level_64":{"level_65":{"level_66":{"level_67":{"level_68":{"level_69":{"level_70":1}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}} | +------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+ diff --git a/tests/cases/standalone/common/types/json/json2_empty.result b/tests/cases/standalone/common/types/json/json2_empty.result index c9fec513964..f4f97c33e1b 100644 --- a/tests/cases/standalone/common/types/json/json2_empty.result +++ b/tests/cases/standalone/common/types/json/json2_empty.result @@ -27,11 +27,11 @@ select ts, j from json_empty order by ts; select ts, j.a from json_empty order by ts; -+-------------------------+----------------------------------+ -| ts | json_get(json_empty.j,Utf8("a")) | -+-------------------------+----------------------------------+ -| 1970-01-01T00:00:00.001 | | -+-------------------------+----------------------------------+ ++-------------------------+------------------------------------+ +| ts | json_get(json_empty.j,Utf8("$.a")) | ++-------------------------+------------------------------------+ +| 1970-01-01T00:00:00.001 | | ++-------------------------+------------------------------------+ admin flush_table('json_empty'); @@ -55,13 +55,13 @@ Affected Rows: 2 select ts, j, j.a from json_empty order by ts; -+-------------------------+---------+----------------------------------+ -| ts | j | json_get(json_empty.j,Utf8("a")) | -+-------------------------+---------+----------------------------------+ -| 1970-01-01T00:00:00.001 | {} | | -| 1970-01-01T00:00:00.002 | {} | | -| 1970-01-01T00:00:00.003 | {"a":1} | 1 | -+-------------------------+---------+----------------------------------+ ++-------------------------+---------+------------------------------------+ +| ts | j | json_get(json_empty.j,Utf8("$.a")) | ++-------------------------+---------+------------------------------------+ +| 1970-01-01T00:00:00.001 | {} | | +| 1970-01-01T00:00:00.002 | {} | | +| 1970-01-01T00:00:00.003 | {"a":1} | 1 | ++-------------------------+---------+------------------------------------+ admin flush_table('json_empty'); @@ -81,13 +81,13 @@ admin compact_table('json_empty'); select ts, j, j.a from json_empty order by ts; -+-------------------------+---------+----------------------------------+ -| ts | j | json_get(json_empty.j,Utf8("a")) | -+-------------------------+---------+----------------------------------+ -| 1970-01-01T00:00:00.001 | {} | | -| 1970-01-01T00:00:00.002 | {} | | -| 1970-01-01T00:00:00.003 | {"a":1} | 1 | -+-------------------------+---------+----------------------------------+ ++-------------------------+---------+------------------------------------+ +| ts | j | json_get(json_empty.j,Utf8("$.a")) | ++-------------------------+---------+------------------------------------+ +| 1970-01-01T00:00:00.001 | {} | | +| 1970-01-01T00:00:00.002 | {} | | +| 1970-01-01T00:00:00.003 | {"a":1} | 1 | ++-------------------------+---------+------------------------------------+ select count(*) from json_empty; @@ -327,16 +327,16 @@ Affected Rows: 2 select ts, j, j.a, j is null from json_empty_null_mixed order by ts; -+-------------------------+------------+---------------------------------------------+---------------------------------+ -| ts | j | json_get(json_empty_null_mixed.j,Utf8("a")) | json_empty_null_mixed.j IS NULL | -+-------------------------+------------+---------------------------------------------+---------------------------------+ -| 1970-01-01T00:00:00.001 | {} | | false | -| 1970-01-01T00:00:00.002 | | | true | -| 1970-01-01T00:00:00.003 | {"a":{}} | {} | false | -| 1970-01-01T00:00:00.004 | {"a":null} | | false | -| 1970-01-01T00:00:00.005 | {"a":1} | 1 | false | -| 1970-01-01T00:00:00.006 | | | true | -+-------------------------+------------+---------------------------------------------+---------------------------------+ ++-------------------------+------------+-----------------------------------------------+---------------------------------+ +| ts | j | json_get(json_empty_null_mixed.j,Utf8("$.a")) | json_empty_null_mixed.j IS NULL | ++-------------------------+------------+-----------------------------------------------+---------------------------------+ +| 1970-01-01T00:00:00.001 | {} | | false | +| 1970-01-01T00:00:00.002 | | | true | +| 1970-01-01T00:00:00.003 | {"a":{}} | {} | false | +| 1970-01-01T00:00:00.004 | {"a":null} | | false | +| 1970-01-01T00:00:00.005 | {"a":1} | 1 | false | +| 1970-01-01T00:00:00.006 | | | true | ++-------------------------+------------+-----------------------------------------------+---------------------------------+ admin flush_table('json_empty_null_mixed'); @@ -356,16 +356,16 @@ admin compact_table('json_empty_null_mixed'); select ts, j, j.a, j is null from json_empty_null_mixed order by ts; -+-------------------------+------------+---------------------------------------------+---------------------------------+ -| ts | j | json_get(json_empty_null_mixed.j,Utf8("a")) | json_empty_null_mixed.j IS NULL | -+-------------------------+------------+---------------------------------------------+---------------------------------+ -| 1970-01-01T00:00:00.001 | {} | | false | -| 1970-01-01T00:00:00.002 | | | true | -| 1970-01-01T00:00:00.003 | {"a":{}} | {} | false | -| 1970-01-01T00:00:00.004 | {"a":null} | | false | -| 1970-01-01T00:00:00.005 | {"a":1} | 1 | false | -| 1970-01-01T00:00:00.006 | | | true | -+-------------------------+------------+---------------------------------------------+---------------------------------+ ++-------------------------+------------+-----------------------------------------------+---------------------------------+ +| ts | j | json_get(json_empty_null_mixed.j,Utf8("$.a")) | json_empty_null_mixed.j IS NULL | ++-------------------------+------------+-----------------------------------------------+---------------------------------+ +| 1970-01-01T00:00:00.001 | {} | | false | +| 1970-01-01T00:00:00.002 | | | true | +| 1970-01-01T00:00:00.003 | {"a":{}} | {} | false | +| 1970-01-01T00:00:00.004 | {"a":null} | | false | +| 1970-01-01T00:00:00.005 | {"a":1} | 1 | false | +| 1970-01-01T00:00:00.006 | | | true | ++-------------------------+------------+-----------------------------------------------+---------------------------------+ select count(*) from json_empty_null_mixed; diff --git a/tests/cases/standalone/common/types/json/json2_list_index.result b/tests/cases/standalone/common/types/json/json2_list_index.result index e7415ce8e29..6ec5b951eba 100644 --- a/tests/cases/standalone/common/types/json/json2_list_index.result +++ b/tests/cases/standalone/common/types/json/json2_list_index.result @@ -10,7 +10,7 @@ CREATE TABLE json2_list_index ( Affected Rows: 0 INSERT INTO json2_list_index VALUES - (1, 'host1', '{"l":[[10,11],["a","b"]],"o":{"l":[{"inner":{"l":[1,2,3]}},{"inner":{"l":["x","y","z"]},"casekey":"normalized","UPPER":"quoted","a.b":"dotted"}]}}'), + (1, 'host1', '{"a.b":"root-dotted","l":[[10,11],["a","b"]],"o":{"l":[{"inner":{"l":[1,2,3]}},{"inner":{"l":["x","y","z"]},"casekey":"normalized","UPPER":"quoted","a.b":"dotted"}]}}'), (2, 'host2', '{"l":[[20],[21,22,23]],"o":{"l":[{"inner":{"l":[4,5,6]}},{"inner":{"l":[7,8]}}]}}'), (3, 'host3', '{"l":[null,[30]],"o":{"l":[null,{"inner":{"l":[true,false,null]}}]}}'); @@ -110,6 +110,21 @@ ORDER BY ts; | 1970-01-01T00:00:00.006 | | | | +-------------------------+------------+--------------+------------+ +SELECT ts, j."a.b" AS root_dotted_key +FROM json2_list_index +ORDER BY ts; + ++-------------------------+-----------------+ +| ts | root_dotted_key | ++-------------------------+-----------------+ +| 1970-01-01T00:00:00.001 | root-dotted | +| 1970-01-01T00:00:00.002 | | +| 1970-01-01T00:00:00.003 | | +| 1970-01-01T00:00:00.004 | | +| 1970-01-01T00:00:00.005 | | +| 1970-01-01T00:00:00.006 | | ++-------------------------+-----------------+ + SELECT ts, j.l[0][0]::DOUBLE * 2 AS calculated FROM json2_list_index ORDER BY ts; @@ -125,6 +140,42 @@ ORDER BY ts; | 1970-01-01T00:00:00.006 | | +-------------------------+------------+ +SELECT ts, + json_get(j, '$."a.b"') AS quoted_path, + json_get(j, '$["a.b"]') AS bracket_path, + json_get(j, '$.o.l[1]."a.b"') AS nested_path +FROM json2_list_index +ORDER BY ts; + ++-------------------------+-------------+--------------+-------------+ +| ts | quoted_path | bracket_path | nested_path | ++-------------------------+-------------+--------------+-------------+ +| 1970-01-01T00:00:00.001 | root-dotted | root-dotted | dotted | +| 1970-01-01T00:00:00.002 | | | | +| 1970-01-01T00:00:00.003 | | | | +| 1970-01-01T00:00:00.004 | | | | +| 1970-01-01T00:00:00.005 | | | | +| 1970-01-01T00:00:00.006 | | | | ++-------------------------+-------------+--------------+-------------+ + +-- Invalid paths must fail planning instead of falling back to Variant. +SELECT json_get(j, '$.o[')::BIGINT FROM json2_list_index; + +Error: 3000(PlanQuery), Error during planning: Invalid JSONPath "$.o[": InvalidJsonPath + +SELECT json_get(j, '$.missing')::BIGINT FROM json2_list_index ORDER BY ts; + ++------------------------------------------------+ +| json_get(json2_list_index.j,Utf8("$.missing")) | ++------------------------------------------------+ +| | +| | +| | +| | +| | +| | ++------------------------------------------------+ + DROP TABLE json2_list_index; Affected Rows: 0 diff --git a/tests/cases/standalone/common/types/json/json2_list_index.sql b/tests/cases/standalone/common/types/json/json2_list_index.sql index d14c0a59966..43c8c2040ed 100644 --- a/tests/cases/standalone/common/types/json/json2_list_index.sql +++ b/tests/cases/standalone/common/types/json/json2_list_index.sql @@ -8,7 +8,7 @@ CREATE TABLE json2_list_index ( ); INSERT INTO json2_list_index VALUES - (1, 'host1', '{"l":[[10,11],["a","b"]],"o":{"l":[{"inner":{"l":[1,2,3]}},{"inner":{"l":["x","y","z"]},"casekey":"normalized","UPPER":"quoted","a.b":"dotted"}]}}'), + (1, 'host1', '{"a.b":"root-dotted","l":[[10,11],["a","b"]],"o":{"l":[{"inner":{"l":[1,2,3]}},{"inner":{"l":["x","y","z"]},"casekey":"normalized","UPPER":"quoted","a.b":"dotted"}]}}'), (2, 'host2', '{"l":[[20],[21,22,23]],"o":{"l":[{"inner":{"l":[4,5,6]}},{"inner":{"l":[7,8]}}]}}'), (3, 'host3', '{"l":[null,[30]],"o":{"l":[null,{"inner":{"l":[true,false,null]}}]}}'); @@ -42,8 +42,24 @@ SELECT ts, FROM json2_list_index ORDER BY ts; +SELECT ts, j."a.b" AS root_dotted_key +FROM json2_list_index +ORDER BY ts; + SELECT ts, j.l[0][0]::DOUBLE * 2 AS calculated FROM json2_list_index ORDER BY ts; +SELECT ts, + json_get(j, '$."a.b"') AS quoted_path, + json_get(j, '$["a.b"]') AS bracket_path, + json_get(j, '$.o.l[1]."a.b"') AS nested_path +FROM json2_list_index +ORDER BY ts; + +-- Invalid paths must fail planning instead of falling back to Variant. +SELECT json_get(j, '$.o[')::BIGINT FROM json2_list_index; + +SELECT json_get(j, '$.missing')::BIGINT FROM json2_list_index ORDER BY ts; + DROP TABLE json2_list_index; diff --git a/tests/cases/standalone/common/types/json/json2_nested.result b/tests/cases/standalone/common/types/json/json2_nested.result index 044d9e04462..1fd06accbd9 100644 --- a/tests/cases/standalone/common/types/json/json2_nested.result +++ b/tests/cases/standalone/common/types/json/json2_nested.result @@ -33,25 +33,25 @@ Affected Rows: 2 SELECT j.a.b FROM json2_nested_fallback ORDER BY ts; -+-----------------------------------------------+ -| json_get(json2_nested_fallback.j,Utf8("a.b")) | -+-----------------------------------------------+ -| | -| 2 | -| 3 | -| | -+-----------------------------------------------+ ++-------------------------------------------------+ +| json_get(json2_nested_fallback.j,Utf8("$.a.b")) | ++-------------------------------------------------+ +| | +| 2 | +| 3 | +| | ++-------------------------------------------------+ SELECT j.a, j.a.b FROM json2_nested_fallback ORDER BY ts; -+---------------------------------------------+-----------------------------------------------+ -| json_get(json2_nested_fallback.j,Utf8("a")) | json_get(json2_nested_fallback.j,Utf8("a.b")) | -+---------------------------------------------+-----------------------------------------------+ -| 1 | | -| {"b":2} | 2 | -| {"b":3} | 3 | -| 2 | | -+---------------------------------------------+-----------------------------------------------+ ++-----------------------------------------------+-------------------------------------------------+ +| json_get(json2_nested_fallback.j,Utf8("$.a")) | json_get(json2_nested_fallback.j,Utf8("$.a.b")) | ++-----------------------------------------------+-------------------------------------------------+ +| 1 | | +| {"b":2} | 2 | +| {"b":3} | 3 | +| 2 | | ++-----------------------------------------------+-------------------------------------------------+ DROP TABLE json2_nested_fallback; @@ -100,13 +100,13 @@ ADMIN FLUSH_TABLE('json2_nested_fallback_multi'); SELECT j.a.b, j.a.c, j.d.e, j.b.x, k.x.y FROM json2_nested_fallback_multi ORDER BY ts; -+-----------------------------------------------------+-----------------------------------------------------+-----------------------------------------------------+-----------------------------------------------------+-----------------------------------------------------+ -| json_get(json2_nested_fallback_multi.j,Utf8("a.b")) | json_get(json2_nested_fallback_multi.j,Utf8("a.c")) | json_get(json2_nested_fallback_multi.j,Utf8("d.e")) | json_get(json2_nested_fallback_multi.j,Utf8("b.x")) | json_get(json2_nested_fallback_multi.k,Utf8("x.y")) | -+-----------------------------------------------------+-----------------------------------------------------+-----------------------------------------------------+-----------------------------------------------------+-----------------------------------------------------+ -| | | 10 | exact1 | | -| 2 | two | | exact2 | ky2 | -| 3 | three | 30 | exact3 | ky3 | -+-----------------------------------------------------+-----------------------------------------------------+-----------------------------------------------------+-----------------------------------------------------+-----------------------------------------------------+ ++-------------------------------------------------------+-------------------------------------------------------+-------------------------------------------------------+-------------------------------------------------------+-------------------------------------------------------+ +| json_get(json2_nested_fallback_multi.j,Utf8("$.a.b")) | json_get(json2_nested_fallback_multi.j,Utf8("$.a.c")) | json_get(json2_nested_fallback_multi.j,Utf8("$.d.e")) | json_get(json2_nested_fallback_multi.j,Utf8("$.b.x")) | json_get(json2_nested_fallback_multi.k,Utf8("$.x.y")) | ++-------------------------------------------------------+-------------------------------------------------------+-------------------------------------------------------+-------------------------------------------------------+-------------------------------------------------------+ +| | | 10 | exact1 | | +| 2 | two | | exact2 | ky2 | +| 3 | three | 30 | exact3 | ky3 | ++-------------------------------------------------------+-------------------------------------------------------+-------------------------------------------------------+-------------------------------------------------------+-------------------------------------------------------+ DROP TABLE json2_nested_fallback_multi; @@ -154,13 +154,13 @@ ADMIN FLUSH_TABLE('json2_nested_fallback_to_parent'); SELECT j.a.b.c FROM json2_nested_fallback_to_parent ORDER BY ts; -+-----------------------------------------------------------+ -| json_get(json2_nested_fallback_to_parent.j,Utf8("a.b.c")) | -+-----------------------------------------------------------+ -| | -| from_parent | -| exact_parent | -+-----------------------------------------------------------+ ++-------------------------------------------------------------+ +| json_get(json2_nested_fallback_to_parent.j,Utf8("$.a.b.c")) | ++-------------------------------------------------------------+ +| | +| from_parent | +| exact_parent | ++-------------------------------------------------------------+ DROP TABLE json2_nested_fallback_to_parent; @@ -208,13 +208,13 @@ ADMIN FLUSH_TABLE('json2_nested_deep_multi'); SELECT j.a.b.c, j.a.b.d FROM json2_nested_deep_multi ORDER BY ts; -+---------------------------------------------------+---------------------------------------------------+ -| json_get(json2_nested_deep_multi.j,Utf8("a.b.c")) | json_get(json2_nested_deep_multi.j,Utf8("a.b.d")) | -+---------------------------------------------------+---------------------------------------------------+ -| | | -| c2 | d2 | -| c3 | d3 | -+---------------------------------------------------+---------------------------------------------------+ ++-----------------------------------------------------+-----------------------------------------------------+ +| json_get(json2_nested_deep_multi.j,Utf8("$.a.b.c")) | json_get(json2_nested_deep_multi.j,Utf8("$.a.b.d")) | ++-----------------------------------------------------+-----------------------------------------------------+ +| | | +| c2 | d2 | +| c3 | d3 | ++-----------------------------------------------------+-----------------------------------------------------+ DROP TABLE json2_nested_deep_multi; @@ -262,13 +262,13 @@ ADMIN FLUSH_TABLE('json2_nested_null_parent'); SELECT j.a.b FROM json2_nested_null_parent ORDER BY ts; -+--------------------------------------------------+ -| json_get(json2_nested_null_parent.j,Utf8("a.b")) | -+--------------------------------------------------+ -| | -| from_null_parent | -| exact_null_parent | -+--------------------------------------------------+ ++----------------------------------------------------+ +| json_get(json2_nested_null_parent.j,Utf8("$.a.b")) | ++----------------------------------------------------+ +| | +| from_null_parent | +| exact_null_parent | ++----------------------------------------------------+ DROP TABLE json2_nested_null_parent; @@ -316,13 +316,13 @@ ADMIN FLUSH_TABLE('json2_nested_array_parent'); SELECT j.a.b FROM json2_nested_array_parent ORDER BY ts; -+---------------------------------------------------+ -| json_get(json2_nested_array_parent.j,Utf8("a.b")) | -+---------------------------------------------------+ -| | -| from_array_parent | -| exact_array_parent | -+---------------------------------------------------+ ++-----------------------------------------------------+ +| json_get(json2_nested_array_parent.j,Utf8("$.a.b")) | ++-----------------------------------------------------+ +| | +| from_array_parent | +| exact_array_parent | ++-----------------------------------------------------+ DROP TABLE json2_nested_array_parent; @@ -371,14 +371,14 @@ ADMIN FLUSH_TABLE('json2_nested_typed_cast'); SELECT j.a.flag::BOOLEAN, j.a.score::DOUBLE, j.a.n::INT64 FROM json2_nested_typed_cast ORDER BY ts; -+----------------------------------------------------+-----------------------------------------------------+---------------------------------------------------------------------------+ -| json_get(json2_nested_typed_cast.j,Utf8("a.flag")) | json_get(json2_nested_typed_cast.j,Utf8("a.score")) | arrow_cast(json_get(json2_nested_typed_cast.j,Utf8("a.n")),Utf8("Int64")) | -+----------------------------------------------------+-----------------------------------------------------+---------------------------------------------------------------------------+ -| | | | -| true | 1.5 | 10 | -| | | | -| false | 2.5 | 20 | -+----------------------------------------------------+-----------------------------------------------------+---------------------------------------------------------------------------+ ++------------------------------------------------------+-------------------------------------------------------+-----------------------------------------------------------------------------+ +| json_get(json2_nested_typed_cast.j,Utf8("$.a.flag")) | json_get(json2_nested_typed_cast.j,Utf8("$.a.score")) | arrow_cast(json_get(json2_nested_typed_cast.j,Utf8("$.a.n")),Utf8("Int64")) | ++------------------------------------------------------+-------------------------------------------------------+-----------------------------------------------------------------------------+ +| | | | +| true | 1.5 | 10 | +| | | | +| false | 2.5 | 20 | ++------------------------------------------------------+-------------------------------------------------------+-----------------------------------------------------------------------------+ DROP TABLE json2_nested_typed_cast; diff --git a/tests/cases/standalone/common/types/json/json2_type_hints.result b/tests/cases/standalone/common/types/json/json2_type_hints.result index d9a384e0844..4e997cb1451 100644 --- a/tests/cases/standalone/common/types/json/json2_type_hints.result +++ b/tests/cases/standalone/common/types/json/json2_type_hints.result @@ -51,13 +51,13 @@ SELECT FROM json2_type_hints ORDER BY ts; -+-----------------------------------------------+------------------------------------------------+--------------------------------------------------+--------------------------------------------+ -| json_get(json2_type_hints.j,Utf8("user.age")) | json_get(json2_type_hints.j,Utf8("user.name")) | json_get(json2_type_hints.j,Utf8("user.active")) | json_get(json2_type_hints.j,Utf8("score")) | -+-----------------------------------------------+------------------------------------------------+--------------------------------------------------+--------------------------------------------+ -| 42 | Alice | true | 3.25 | -| 18 | Bob | | 1.5 | -| 18 | unknown | | 1.5 | -+-----------------------------------------------+------------------------------------------------+--------------------------------------------------+--------------------------------------------+ ++-------------------------------------------------+--------------------------------------------------+----------------------------------------------------+----------------------------------------------+ +| json_get(json2_type_hints.j,Utf8("$.user.age")) | json_get(json2_type_hints.j,Utf8("$.user.name")) | json_get(json2_type_hints.j,Utf8("$.user.active")) | json_get(json2_type_hints.j,Utf8("$.score")) | ++-------------------------------------------------+--------------------------------------------------+----------------------------------------------------+----------------------------------------------+ +| 42 | Alice | true | 3.25 | +| 18 | Bob | | 1.5 | +| 18 | unknown | | 1.5 | ++-------------------------------------------------+--------------------------------------------------+----------------------------------------------------+----------------------------------------------+ INSERT INTO json2_type_hints VALUES (4, '{"user":{"age":"bad"}}'); diff --git a/tests/cases/standalone/common/types/json/jsonbench.result b/tests/cases/standalone/common/types/json/jsonbench.result index bc039e0b080..2c99b219605 100644 --- a/tests/cases/standalone/common/types/json/jsonbench.result +++ b/tests/cases/standalone/common/types/json/jsonbench.result @@ -162,17 +162,17 @@ EXPLAIN SELECT date_part('hour', to_timestamp_micros(arrow_cast(data.time_us, 'Int64'))) as hour_of_day FROM bluesky; -+---------------+-------------------------------------------------------------------------------------------------------------------------------+ -| plan_type | plan | -+---------------+-------------------------------------------------------------------------------------------------------------------------------+ -| logical_plan | MergeScan [is_placeholder=false, remote_input=[ | -| | Projection: date_part(Utf8("hour"), to_timestamp_micros(json_get(bluesky.data, Utf8("time_us"), Int64(NULL)))) AS hour_of_day | -| | TableScan: bluesky | -| | ]] | -| physical_plan | CooperativeExec | ++---------------+---------------------------------------------------------------------------------------------------------------------------------+ +| plan_type | plan | ++---------------+---------------------------------------------------------------------------------------------------------------------------------+ +| logical_plan | MergeScan [is_placeholder=false, remote_input=[ | +| | Projection: date_part(Utf8("hour"), to_timestamp_micros(json_get(bluesky.data, Utf8("$.time_us"), Int64(NULL)))) AS hour_of_day | +| | TableScan: bluesky | +| | ]] | +| physical_plan | CooperativeExec | | | MergeScanExec: REDACTED -| | | -+---------------+-------------------------------------------------------------------------------------------------------------------------------+ +| | | ++---------------+---------------------------------------------------------------------------------------------------------------------------------+ DROP TABLE bluesky; diff --git a/tests/compatibility/cases/trace_v2_json2/case.toml b/tests/compatibility/cases/trace_v2_json2/case.toml new file mode 100644 index 00000000000..f4136b4e2d8 --- /dev/null +++ b/tests/compatibility/cases/trace_v2_json2/case.toml @@ -0,0 +1,8 @@ +name = "trace_v2_json2" +reason = "Preserve the trace v2 schema, JSON2 attributes and model metadata across restart and upgrade." +introduced_by = "greptime_trace_v2" +topologies = ["standalone", "distributed"] +from_range = [">=v1.3.0-alpha.1"] +to_range = [">=v1.3.0-alpha.1"] +features = ["trace", "json2", "table", "metadata", "read"] +owner = "frontend" diff --git a/tests/compatibility/cases/trace_v2_json2/setup.sql b/tests/compatibility/cases/trace_v2_json2/setup.sql new file mode 100644 index 00000000000..02538b0ea70 --- /dev/null +++ b/tests/compatibility/cases/trace_v2_json2/setup.sql @@ -0,0 +1,33 @@ +CREATE TABLE t_trace_v2 ( + "timestamp" TIMESTAMP(9) TIME INDEX, + timestamp_end TIMESTAMP(9), + duration_nano BIGINT, + parent_span_id STRING, + trace_id STRING, + span_id STRING, + span_kind STRING, + span_name STRING, + span_status_code STRING, + span_status_message STRING, + trace_state STRING, + scope_name STRING, + scope_version STRING, + service_name STRING PRIMARY KEY, + span_attributes JSON2, + scope_attributes JSON2, + resource_attributes JSON2 +) WITH ( + 'append_mode' = 'true', + 'table_data_model' = 'greptime_trace_v2', + 'greptime.semantic.pipeline' = 'greptime_trace_v2', + 'greptime.semantic.signal_type' = 'trace', + 'greptime.semantic.source' = 'opentelemetry', + 'greptime.semantic.entity.service.id' = 'service_name', + 'greptime.semantic.trace.conventions' = 'unknown' +); + +INSERT INTO t_trace_v2 ("timestamp", timestamp_end, duration_nano, service_name, trace_id, span_id, span_attributes, scope_attributes, resource_attributes) VALUES + (1, 11, 10, NULL, 'trace-1', 'span-1', '{"http.status_code":200,"nested":{"a.b":[true,"ok"]}}', '{}', '{}'), + (1, 21, 20, NULL, 'trace-2', 'span-2', '{"http.status_code":"ok","bytes":[1,2,3]}', '{"enabled":true}', '{"deployment.environment":"prod"}'); + +ADMIN FLUSH_TABLE('t_trace_v2'); diff --git a/tests/compatibility/cases/trace_v2_json2/verify.result b/tests/compatibility/cases/trace_v2_json2/verify.result new file mode 100644 index 00000000000..7d3d9f2bf7f --- /dev/null +++ b/tests/compatibility/cases/trace_v2_json2/verify.result @@ -0,0 +1,49 @@ +SELECT COUNT(*) AS n FROM t_trace_v2; + ++---+ +| n | ++---+ +| 2 | ++---+ + +SELECT COUNT(*) AS n FROM t_trace_v2 +WHERE span_attributes."http.status_code"::BIGINT = 200 + AND span_attributes.nested."a.b"[0]::BOOLEAN = true; + ++---+ +| n | ++---+ +| 1 | ++---+ + +SELECT COUNT(*) AS n FROM t_trace_v2 +WHERE span_attributes."http.status_code"::STRING = 'ok' + AND span_attributes.bytes[1]::BIGINT = 2 + AND scope_attributes.enabled::BOOLEAN = true + AND resource_attributes."deployment.environment"::STRING = 'prod'; + ++---+ +| n | ++---+ +| 1 | ++---+ + +SELECT COUNT(*) AS n FROM information_schema.columns +WHERE table_name = 't_trace_v2'; + ++----+ +| n | ++----+ +| 17 | ++----+ + +SELECT COUNT(*) AS n FROM information_schema.tables +WHERE table_name = 't_trace_v2' + AND create_options LIKE '%table_data_model=greptime_trace_v2%' + AND create_options LIKE '%greptime.semantic.pipeline=greptime_trace_v2%'; + ++---+ +| n | ++---+ +| 1 | ++---+ diff --git a/tests/compatibility/cases/trace_v2_json2/verify.sql b/tests/compatibility/cases/trace_v2_json2/verify.sql new file mode 100644 index 00000000000..7a1b0cd50bf --- /dev/null +++ b/tests/compatibility/cases/trace_v2_json2/verify.sql @@ -0,0 +1,19 @@ +SELECT COUNT(*) AS n FROM t_trace_v2; + +SELECT COUNT(*) AS n FROM t_trace_v2 +WHERE span_attributes."http.status_code"::BIGINT = 200 + AND span_attributes.nested."a.b"[0]::BOOLEAN = true; + +SELECT COUNT(*) AS n FROM t_trace_v2 +WHERE span_attributes."http.status_code"::STRING = 'ok' + AND span_attributes.bytes[1]::BIGINT = 2 + AND scope_attributes.enabled::BOOLEAN = true + AND resource_attributes."deployment.environment"::STRING = 'prod'; + +SELECT COUNT(*) AS n FROM information_schema.columns +WHERE table_name = 't_trace_v2'; + +SELECT COUNT(*) AS n FROM information_schema.tables +WHERE table_name = 't_trace_v2' + AND create_options LIKE '%table_data_model=greptime_trace_v2%' + AND create_options LIKE '%greptime.semantic.pipeline=greptime_trace_v2%';