feat(otlp): add trace v2 ingestion with JSON2 attributes

Signed-off-by: luofucong <luofc@foxmail.com>
This commit is contained in:
luofucong
2026-09-17 09:23:32 +08:00
parent 259b42c039
commit aeadf27e2d
38 changed files with 1861 additions and 497 deletions
Generated
+1
View File
@@ -11926,6 +11926,7 @@ dependencies = [
"greptime-proto",
"humantime",
"itertools 0.14.0",
"jsonb",
"lazy_static",
"log-query",
"meter-core",
+291 -67
View File
@@ -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<Vec<u8>> {
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<JsonPath<'_>, 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<Option<Vec<String>>> {
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<Box<dyn JsonGetResultBuilder>> {
fn result_builder(len: usize, with_type: &DataType) -> Result<Box<dyn JsonGetResultBuilder>> {
let builder = match with_type {
DataType::Utf8 | DataType::LargeUtf8 | DataType::Utf8View => Box::new(StringResultBuilder {
inner: StringViewBuilder::with_capacity(len),
is_json2,
})
as Box<dyn JsonGetResultBuilder>,
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<ArrayRef> {
let segments = path
.trim_start_matches("$")
.split('.')
.filter(|segment| !segment.is_empty())
.collect::<Vec<_>>();
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<a: Struct<b: Int64>>`.
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<dyn std::error::Error>>
{
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<dyn std::error::Error>> {
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::<Int64Type>().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::<Int64Type>().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<dyn std::error::Error>> {
let json = parse_string_to_jsonb(r#"{"a": 1}"#)?;
let arrays: Vec<ArrayRef> = vec![
Arc::new(BinaryArray::from_iter_values([json])),
Arc::new(BinaryArray::from(vec![None::<&[u8]>])),
Arc::new(BinaryArray::from(Vec::<Option<&[u8]>>::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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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<dyn std::error::Error>>
{
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::<Int64Type>().value(0), 42);
assert!(!result.is_null(0));
}
Ok(())
}
}
+12 -7
View File
@@ -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<String> {
/// Converts a json type value to serde_json::Value
pub fn jsonb_to_serde_json(val: &[u8]) -> Result<serde_json::Value> {
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<Vec<u8>> {
#[cfg(test)]
mod tests {
#[test]
fn test_v2_jsonb_to_serde_json_escaped_keys()
-> std::result::Result<(), Box<dyn std::error::Error>> {
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]
+4 -12
View File
@@ -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<TraceIngestOutcome> {
@@ -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)]
+15 -2
View File
@@ -24,10 +24,12 @@ flowchart TD
F -->|greptime_trace_v0| G["v0 conversion<br/>fixed columns and JSON attributes"]
F -->|greptime_trace_v1| H["v1 conversion<br/>flattened dynamic attributes"]
F -->|greptime_trace_v2| T["v2 conversion<br/>fixed columns and JSON2 attributes"]
H --> I["Collect request-wide schema observations"]
I --> J["Resolve types, prevalidate rewrites,<br/>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<br/>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
+100 -27
View File
@@ -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;
@@ -79,11 +79,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 +852,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<TraceSpanGroup>,
conventions: &str,
ctx: QueryContextRef,
) -> ServerResult<TraceIngestOutcome> {
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 +876,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 +985,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 +1065,78 @@ impl Instance {
async fn ingest_trace_chunk(
&self,
ingest_ctx: &TraceChunkIngestContext<'_>,
chunk: Vec<TraceSpan>,
mut chunk: Vec<TraceSpan>,
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()
};
let convert_span = |span: &TraceSpan| {
let result = convert(std::slice::from_ref(span));
let is_ok = result.is_ok();
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) {
if rows.schema.is_empty() {
rows.schema = batch.schema;
} else {
debug_assert_eq!(rows.schema, batch.schema);
}
rows.rows.extend(batch.rows);
}
}
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
),
);
}
}
is_ok
};
chunk.retain(convert_span);
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 +1429,17 @@ impl Instance {
async fn insert_trace_requests(
&self,
mut requests: RowInsertRequests,
is_trace_v1_model: bool,
pipeline: &PipelineWay,
ctx: QueryContextRef,
) -> ServerResult<Output> {
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 +1464,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() {
+35 -3
View File
@@ -68,8 +68,9 @@ use table::TableRef;
use table::metadata::TableInfo;
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;
@@ -625,6 +626,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());
}
@@ -650,6 +654,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());
}
@@ -841,7 +848,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
@@ -1171,6 +1180,7 @@ impl Inserter {
match res {
Ok(table) => {
validate_trace_table_model(&table.table_info(), ctx)?;
info!(
"Successfully created table {} with options: {:?}",
table_ref, create_table_expr.table_options,
@@ -1278,6 +1288,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<String, String>,
+3 -2
View File
@@ -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;
+6 -1
View File
@@ -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<PipelineWay>,
) -> Result<PipelineWay> {
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)
+1
View File
@@ -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
+62 -15
View File
@@ -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<String> {
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),
+66 -22
View File
@@ -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<Option<(String, JsonNativeType)>> {
);
};
// 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<Option<(String, JsonNativeType)>> {
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(())
}
+3 -3
View File
@@ -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.
+9
View File
@@ -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(),
+32 -5
View File
@@ -1052,11 +1052,21 @@ fn to_keyvalue(key: String, value: JsonValue) -> Option<KeyValue> {
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)`.
+11 -24
View File
@@ -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,18 @@ 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 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(crate) 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 +113,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 +138,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",
}
+6 -1
View File
@@ -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),
-7
View File
@@ -18,8 +18,6 @@ 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;
@@ -30,7 +28,6 @@ use crate::otlp::trace::{
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)?;
+1 -8
View File
@@ -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;
@@ -37,7 +35,6 @@ use crate::otlp::trace::{
TIMESTAMP_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;
@@ -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
+145
View File
@@ -0,0 +1,145 @@
// 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, 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<TableData> {
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",
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(())
}
+59 -2
View File
@@ -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, Result, RowWriterSnafu, TimePrecisionSnafu,
TimestampOverflowSnafu, ToJsonSnafu,
};
/// The intermediate data structure for building the write request.
@@ -317,6 +327,53 @@ 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<Value>,
) -> 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 {
unreachable!()
};
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<Item = (ColumnSchema, Option<ValueData>)>,
+2
View File
@@ -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
+415
View File
@@ -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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
let distributed = tests::create_distributed_instance("trace_v2_distributed").await;
test_trace_v2(&distributed.frontend()).await?;
Ok(())
}
async fn test_trace_v2(instance: &Arc<Instance>) -> Result<(), Box<dyn std::error::Error>> {
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<_>>(),
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<Instance>,
sql: &str,
ctx: Arc<QueryContext>,
) -> Result<(), Box<dyn std::error::Error>> {
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<_>>(),
vec![Some(true)],
"{sql}"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
pub async fn test_otlp_on_standalone() {
let standalone = GreptimeDbStandaloneBuilder::new("test_standalone_otlp")
+127 -2
View File
@@ -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,
@@ -7718,6 +7726,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<dyn std::error::Error>> {
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.
@@ -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');
+------------------------------------------------------------------------+
@@ -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');
@@ -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;
@@ -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}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}}} |
+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
@@ -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;
@@ -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
@@ -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;
@@ -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;
@@ -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"}}');
@@ -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;
@@ -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"
@@ -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');
@@ -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 |
+---+
@@ -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%';