rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-07-08 23:10:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: accept JSONB OTLP logs against existing schema (#8250)

Browse Source 
* fix: accept JSONB OTLP logs against existing schema

Handle existing OTLP log JSONB columns that round-trip as Json while direct log rows carry pre-encoded binary JSONB payloads.

Add integration coverage for repeated OTLP log batches after table auto-creation.

Files: src/servers/src/otlp/logs.rs, tests-integration/tests/http.rs
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>

* perf: cache OTLP log JSONB schema check

Cache whether an existing OTLP log column is legacy JSONB once per column before coercing rows.

Files: src/servers/src/otlp/logs.rs
Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>

* fix: narrow down schema coercion

Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>

---------

Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
This commit is contained in:
Lei, HUANG
2026-06-08 09:57:23 +08:00
committed by GitHub
parent 7baa4f6854
commit e7ce3ac0c7
2 changed files with 53 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
src/servers/src/otlp/logs.rs
View File

@@ -327,6 +327,28 @@ struct ExistingLogColumn {
datatype: ColumnDataType,
}
impl ExistingLogColumn {
fn schema_for_request_type(&self, request_type: ColumnDataType) -> ColumnSchema {
let mut schema = self.schema.clone();
if request_type == ColumnDataType::Binary && self.is_json_binary() {
schema.datatype = ColumnDataType::Binary as i32;
}
schema
}
fn is_json_binary(&self) -> bool {
self.datatype == ColumnDataType::Json
&& matches!(
self.schema
.datatype_extension
.as_ref()
.and_then(|datatype_extension| datatype_extension.type_ext.as_ref()),
Some(TypeExt::JsonType(json_type))
if *json_type == JsonTypeExtension::JsonBinary as i32
)
}
}
#[derive(Default)]
struct ExistingLogSchema {
columns: HashMap<String, ExistingLogColumn>,
@@ -581,6 +603,7 @@ fn decide_existing_column_schema_and_convert_value(
existing_column.datatype,
existing_column.schema.semantic_type(),
request_type,
existing_column.is_json_binary(),
column_name,
table_name,
)?;
@@ -646,6 +669,7 @@ fn align_rows_with_existing_schema(
let target_type = existing_column.datatype;
let semantic_type = existing_column.schema.semantic_type();
let target_is_json_binary = existing_column.is_json_binary();
for row in rows.iter_mut() {
let Some(value) = row.values.get_mut(column_idx) else {
continue;
@@ -655,11 +679,12 @@ fn align_rows_with_existing_schema(
target_type,
semantic_type,
request_type,
target_is_json_binary,
&schema.column_name,
table_name,
)?;
}
*schema = existing_column.schema.clone();
*schema = existing_column.schema_for_request_type(request_type);
}
Ok(())
@@ -670,6 +695,7 @@ fn coerce_log_value_data(
target_type: ColumnDataType,
_semantic_type: SemanticType,
request_type: ColumnDataType,
target_is_json_binary: bool,
column_name: &str,
table_name: &str,
) -> Result<Option<ValueData>> {
@@ -681,6 +707,10 @@ fn coerce_log_value_data(
return Ok(Some(value_data));
}
if request_type == ColumnDataType::Binary && target_is_json_binary {
return Ok(Some(value_data));
}
if is_timestamp_type(request_type)
&& let Some(target_unit) = timestamp_unit(target_type)
{

22
tests-integration/tests/http.rs
View File

@@ -6387,6 +6387,28 @@ pub async fn test_otlp_logs(store_type: StorageType) {
"[[1]]",
)
.await;
// Write another batch after the table exists. This verifies existing-schema
// alignment accepts the built-in JSONB columns in direct OTLP logs.
let res = send_req(
&client,
vec![(
HeaderName::from_static("content-type"),
HeaderValue::from_static("application/x-protobuf"),
)],
"/v1/otlp/v1/logs?db=public",
body.clone(),
false,
)
.await;
assert_eq!(StatusCode::OK, res.status());
validate_data(
"otlp_logs_multiple_batches",
&client,
"select count(*) from opentelemetry_logs;",
"[[4]]",
)
.await;
}
{