rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-29 15:06:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(otlp)!: support raw delta exponential histograms (#9754)

Browse Source 
* feat(otlp): support raw delta exponential histograms

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix(otlp): guard point accounting and benchmark metrics ingestion

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix: remove otlp_metric_compat logic

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* perf(otlp): cache physical schema lookups during metric preflight

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* chore: minor rename

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* perf(otlp): batch metric rows and reuse schema validation

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* chore: comments

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* fix: replace super in import path

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

---------

Signed-off-by: shuiyisong <xixing.sys@gmail.com>
This commit is contained in:
shuiyisong
2026-10-22 13:36:30 +00:00
committed by GitHub
parent bb3f07b482
commit 2eff52c757
45 changed files with 5051 additions and 1014 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/frontend/src/metrics.rs
+4 -5
View File
@@ -14,12 +14,12 @@
//! Cache lookup counters shared by frontend, pipeline, and protocol handlers.
//!
//! The `type` label identifies `otlp_trace_aux`, `otlp_metrics_legacy`,
//! The `type` label identifies `otlp_trace_aux`,
//! `pipeline_table`, `pipeline`, `pipeline_original`, `pipeline_failover`, or
//! `mysql_prepared_stmt`. Counters aggregate across catalogs and connections.
//! Trace lookups count distinct service/operation keys once per request, excluding
//! rechecks; legacy metric lookups count each requested metric name. Pipeline
//! loaders count successful coalesced waiters as hits and failed loads as misses.
//! rechecks. Pipeline loaders count successful coalesced waiters as hits and
//! failed loads as misses.
//! Failover counts one outcome after resolving all candidate schemas; an ambiguous
//! result is a miss. Table and prepared-statement caches count each lookup.
//! Both outcomes for all known caches are exported from the first metric update.
@@ -27,9 +27,8 @@
use lazy_static::lazy_static;
use prometheus::{IntCounterVec, register_int_counter_vec};
const CACHE_TYPES: [&str; 7] = [
const CACHE_TYPES: [&str; 6] = [
"otlp_trace_aux",
"otlp_metrics_legacy",
"pipeline_table",
"pipeline",
"pipeline_original",
src/common/query/src/prelude.rs
-2
View File
@@ -91,8 +91,6 @@ pub fn greptime_native_histogram() -> &'static str {
const GREPTIME_TIMESTAMP: &str = "greptime_timestamp";
/// Default value column name constant for backward compatibility.
const GREPTIME_VALUE: &str = "greptime_value";
/// Default counter column name for OTLP metrics (legacy mode).
pub const GREPTIME_COUNT: &str = "greptime_count";
/// Stored series label that opts ordinary float samples into raw-delta math.
pub const OTLP_AGGREGATION_TEMPORALITY_LABEL: &str = "otlp_aggregation_temporality";
/// Authoritative raw-delta value for [`OTLP_AGGREGATION_TEMPORALITY_LABEL`].
src/frontend/AGENTS.md
+4 -1
View File
@@ -74,8 +74,11 @@ including caches owned by pipeline and protocol handlers.
These tables are ingestion-managed; manual mutation requires restarting
serving frontends.
- **OTLP metrics** (`instance/otlp.rs`): authorize all emitted destinations before
schema preflight can reject points or expose destination schemas.
- **Logical-table batching** (`instance/logical_batcher.rs`): `Services` initializes
one shared batcher for opted-in HTTP Prom and nonlegacy OTLP metric-engine
one shared batcher for opted-in HTTP Prom and OTLP metric-engine
writes. Downstream routers can initialize it when enabling replacement endpoints.
`Instance::handle_otlp_metric_row_inserts` shares OTLP eligibility and dispatch
between converters and falls back for incompatible tables.
src/frontend/Cargo.toml
-1
View File
@@ -50,7 +50,6 @@ common-stat.workspace = true
common-telemetry.workspace = true
common-time.workspace = true
common-version.workspace = true
dashmap.workspace = true
datafusion.workspace = true
datafusion-expr.workspace = true
datatypes.workspace = true
src/frontend/src/instance.rs
+2 -193
View File
@@ -53,11 +53,9 @@ use common_base::cancellation::CancellableFuture;
use common_batcher::request_limiter::RequestLimiter;
use common_error::ext::{BoxedError, ErrorExt};
use common_event_recorder::EventRecorderRef;
use common_frontend::metrics::{CACHE_HIT, CACHE_MISS};
use common_meta::cache::TableFlownodeSetCacheRef;
use common_meta::cache_invalidator::CacheInvalidatorRef;
use common_meta::key::TableMetadataManagerRef;
use common_meta::key::table_name::TableNameKey;
use common_meta::node_manager::NodeManagerRef;
use common_meta::procedure_executor::ProcedureExecutorRef;
use common_query::Output;
@@ -65,12 +63,10 @@ use common_recordbatch::RecordBatchStreamWrapper;
use common_recordbatch::error::StreamTimeoutSnafu;
use common_telemetry::logging::SlowQueryOptions;
use common_telemetry::{debug, error, tracing};
use dashmap::DashMap;
use datafusion::dataframe::DataFrame;
use datafusion::physical_plan::ExecutionPlan;
use datafusion_expr::LogicalPlan;
use futures::{Stream, StreamExt, future};
use lazy_static::lazy_static;
use operator::delete::DeleterRef;
use operator::insert::InserterRef;
use operator::statement::{StatementExecutor, StatementExecutorRef};
@@ -84,14 +80,10 @@ use query::parser::{PromQuery, QueryStatement};
use query::query_engine::DescribeResult;
use query::query_engine::options::{QueryOptions, validate_catalog_and_schema};
use servers::batcher::logical_table::LogicalTablePendingRowsBatcher;
use servers::error::{
self as server_error, AuthSnafu, CommonMetaSnafu, ExecuteQuerySnafu,
OtlpMetricModeIncompatibleSnafu, UnexpectedResultSnafu,
};
use servers::error::{self as server_error, AuthSnafu, ExecuteQuerySnafu, UnexpectedResultSnafu};
use servers::interceptor::{
PromQueryInterceptor, PromQueryInterceptorRef, SqlQueryInterceptor, SqlQueryInterceptorRef,
};
use servers::otlp::metrics::legacy_normalize_otlp_name;
use servers::prometheus_handler::{
ParsedPromQuery, PrometheusHandler, resolve_schema_from_matchers,
};
@@ -108,7 +100,6 @@ use sql::statements::statement::Statement;
use sql::statements::tql::Tql;
use sql::util::{extract_tables_from_prom_expr_checked, extract_tables_from_statement_checked};
use sqlparser::ast::{AnalyzeFormat, ObjectName};
use table::requests::{OTLP_METRIC_COMPAT_KEY, OTLP_METRIC_COMPAT_PROM};
use tracing::Span;
use crate::error::{
@@ -119,10 +110,6 @@ use crate::error::{
use crate::service_config::InfluxdbMergeMode;
use crate::stream_wrapper::CancellableStreamWrapper;
lazy_static! {
static ref OTLP_LEGACY_DEFAULT_VALUE: String = "legacy".to_string();
}
/// The frontend instance contains necessary components, and implements many
/// traits, like [`servers::query_handler::grpc::GrpcQueryHandler`],
/// [`servers::query_handler::sql::SqlQueryHandler`], etc.
@@ -155,12 +142,6 @@ pub struct Instance {
common_grpc::flow_write::Reply,
>,
>,
// cache for otlp metrics
// first layer key: db-string
// key: direct input metric name
// value: if runs in legacy mode
otlp_metrics_table_legacy_cache: Arc<DashMap<String, DashMap<String, bool>>>,
}
impl Instance {
@@ -499,137 +480,6 @@ impl Instance {
.await
.context(TableOperationSnafu)
}
async fn check_otlp_legacy(
&self,
names: &[String],
ctx: &QueryContextRef,
) -> server_error::Result<bool> {
let db_string = ctx.get_db_string();
// fast cache check
let cache = self
.otlp_metrics_table_legacy_cache
.entry(db_string.clone())
.or_default();
if let Some(flag) = fast_legacy_check(&cache, names)? {
return Ok(flag);
}
// release cache reference to avoid lock contention
drop(cache);
let catalog = ctx.current_catalog();
let schema = ctx.current_schema();
// query legacy table names
let normalized_names = names
.iter()
.map(|n| legacy_normalize_otlp_name(n))
.collect::<Vec<_>>();
let table_names = normalized_names
.iter()
.map(|n| TableNameKey::new(catalog, &schema, n))
.collect::<Vec<_>>();
let table_values = self
.table_metadata_manager()
.table_name_manager()
.batch_get(table_names)
.await
.context(CommonMetaSnafu)?;
let table_ids = table_values
.into_iter()
.filter_map(|v| v.map(|vi| vi.table_id()))
.collect::<Vec<_>>();
// means no existing table is found, use new mode
if table_ids.is_empty() {
return Ok(false);
}
// has existing table, check table options
let table_infos = self
.table_metadata_manager()
.table_info_manager()
.batch_get(&table_ids)
.await
.context(CommonMetaSnafu)?;
let options = table_infos
.values()
.map(|info| {
info.table_info
.meta
.options
.extra_options
.get(OTLP_METRIC_COMPAT_KEY)
.unwrap_or(&OTLP_LEGACY_DEFAULT_VALUE)
})
.collect::<Vec<_>>();
if !options.is_empty() {
// check value consistency
let has_prom = options.iter().any(|opt| *opt == OTLP_METRIC_COMPAT_PROM);
let has_legacy = options
.iter()
.any(|opt| *opt == OTLP_LEGACY_DEFAULT_VALUE.as_str());
ensure!(!(has_prom && has_legacy), OtlpMetricModeIncompatibleSnafu);
Ok(has_legacy)
} else {
// no table info, use new mode
Ok(false)
}
}
fn cache_otlp_legacy(
&self,
names: &[String],
ctx: &QueryContextRef,
is_legacy: bool,
) -> server_error::Result<()> {
let cache = self
.otlp_metrics_table_legacy_cache
.entry(ctx.get_db_string())
.or_default();
cache_legacy_mode(&cache, names, is_legacy)
}
}
fn fast_legacy_check(
cache: &DashMap<String, bool>,
names: &[String],
) -> server_error::Result<Option<bool>> {
let hit_cache = names
.iter()
.filter_map(|name| cache.get(name))
.collect::<Vec<_>>();
CACHE_HIT
.with_label_values(&["otlp_metrics_legacy"])
.inc_by(hit_cache.len() as u64);
CACHE_MISS
.with_label_values(&["otlp_metrics_legacy"])
.inc_by((names.len() - hit_cache.len()) as u64);
if !hit_cache.is_empty() {
let hit_legacy = hit_cache.iter().any(|en| *en.value());
let hit_prom = hit_cache.iter().any(|en| !*en.value());
// hit but have true and false, means both legacy and new mode are used
// we cannot handle this case, so return error
// add doc links in err msg later
ensure!(!(hit_legacy && hit_prom), OtlpMetricModeIncompatibleSnafu);
Ok(Some(hit_legacy))
} else {
Ok(None)
}
}
fn cache_legacy_mode(
cache: &DashMap<String, bool>,
names: &[String],
is_legacy: bool,
) -> server_error::Result<()> {
for name in names {
let cached = cache.entry(name.clone()).or_insert(is_legacy);
ensure!(*cached == is_legacy, OtlpMetricModeIncompatibleSnafu);
}
Ok(())
}
/// If the relevant variables are set, the timeout is enforced for all PostgreSQL statements.
@@ -1996,6 +1846,7 @@ mod tests {
use common_error::ext::{BoxedError, ErrorExt, PlainError};
use common_error::status_code::StatusCode;
use common_event_recorder::{Event, EventRecorder, EventTypeFilter, EventTypeFilterRef};
use common_frontend::metrics::{CACHE_HIT, CACHE_MISS};
use common_frontend::slow_query_event::SlowQueryEvent;
use common_meta::cache::LayeredCacheRegistryBuilder;
use common_meta::kv_backend::memory::MemoryKvBackend;
@@ -3456,21 +3307,6 @@ mod tests {
Ok(())
}
#[tokio::test]
async fn test_instance_shared_state() -> TestResult<()> {
let instance =
test_instance_with_tables(test_table(1024, "source")?, test_table(1025, "target")?)
.await?;
let _event_recorder = instance.event_recorder();
assert!(Arc::ptr_eq(
&instance.otlp_metrics_table_legacy_cache,
&instance.clone().otlp_metrics_table_legacy_cache,
));
Ok(())
}
#[tokio::test]
async fn test_restricted_endpoint_handlers_check_permissions() -> TestResult<()> {
let checker = Arc::new(RejectEndpointPermissionChecker::default());
@@ -3927,33 +3763,6 @@ mod tests {
Ok(())
}
#[test]
fn test_fast_legacy_check_is_read_only() {
let hits = CACHE_HIT.with_label_values(&["otlp_metrics_legacy"]);
let misses = CACHE_MISS.with_label_values(&["otlp_metrics_legacy"]);
let before = (hits.get(), misses.get());
let cache = DashMap::new();
cache.insert("metric1".to_string(), true);
let names = vec!["metric1".to_string(), "metric2".to_string()];
assert_eq!(Some(true), fast_legacy_check(&cache, &names).unwrap());
assert_eq!((hits.get(), misses.get()), (before.0 + 1, before.1 + 1));
assert!(!cache.contains_key("metric2"));
cache_legacy_mode(&cache, &names, true).unwrap();
assert!(*cache.get("metric2").unwrap().value());
assert!(cache_legacy_mode(&cache, &names, false).is_err());
assert!(*cache.get("metric2").unwrap().value());
let cache_incompatible = DashMap::new();
cache_incompatible.insert("metric1".to_string(), true);
cache_incompatible.insert("metric2".to_string(), false);
assert!(fast_legacy_check(&cache_incompatible, &names).is_err());
assert_eq!((hits.get(), misses.get()), (before.0 + 3, before.1 + 1));
assert_eq!(None, fast_legacy_check(&cache, &[]).unwrap());
assert_eq!((hits.get(), misses.get()), (before.0 + 3, before.1 + 1));
}
#[test]
fn test_should_track_statement_process() {
assert!(should_track_statement_process(&parse_one_sql(
src/frontend/src/instance/builder.rs
-2
View File
@@ -32,7 +32,6 @@ use common_meta::key::flow::FlowMetadataManager;
use common_meta::kv_backend::KvBackendRef;
use common_meta::node_manager::NodeManagerRef;
use common_meta::procedure_executor::ProcedureExecutorRef;
use dashmap::DashMap;
use operator::batcher::PendingRowsBatcher;
use operator::delete::Deleter;
use operator::flow::FlowServiceOperator;
@@ -432,7 +431,6 @@ impl FrontendBuilder {
event_recorder,
slow_query_recorder,
process_manager,
otlp_metrics_table_legacy_cache: Arc::new(DashMap::new()),
slow_query_options: self.options.slow_query.clone(),
influxdb_default_merge_mode: self.options.influxdb.default_merge_mode,
trace_ingest_chunk_size: self.options.otlp.trace_ingest_chunk_size,
src/frontend/src/instance/otlp.rs
+41 -23
View File
@@ -16,6 +16,7 @@ mod trace_ingest;
pub mod trace_semconv;
pub mod trace_types;
use std::collections::HashSet;
use std::sync::Arc;
use api::v1::RowInsertRequests;
@@ -108,31 +109,53 @@ impl OpenTelemetryProtocolHandler for Instance {
interceptor_ref.pre_execute(ctx.clone())?;
let ctx = Arc::new(ctx.fork());
let input_names = request
.resource_metrics
.iter()
.flat_map(|r| r.scope_metrics.iter())
.flat_map(|s| s.metrics.iter().map(|m| m.name.clone()))
.collect::<Vec<_>>();
// See [`OtlpMetricCtx`] for details
let is_legacy = self.check_otlp_legacy(&input_names, &ctx).await?;
let mut metric_ctx = ctx
.protocol_ctx()
.get_otlp_metric_ctx()
.cloned()
.unwrap_or_default();
metric_ctx.is_legacy = is_legacy;
metric_ctx.resource_info = self.otlp_resource_info;
let mut prepared = otlp::metrics::prepare_metrics(request, &mut metric_ctx)?;
// Check all destinations before preflight can expose schemas or discard points.
let targets = {
let catalog = ctx.current_catalog();
let schema = ctx.current_schema();
let mut table_names = HashSet::new();
prepared
.requests
.iter()
.filter(|request| table_names.insert(request.table_name.as_str()))
.map(|request| PermissionTableTarget::new(catalog, &schema, &request.table_name))
.collect()
};
self.check_table_permission(
&ctx,
PermissionReq::Action(OTLP_WRITE),
PermissionTableTargets::resolved(targets),
)
.context(AuthSnafu)?;
let preflight = self
.inserter
.validate_and_normalize_otlp_metrics(
&mut prepared.requests,
&ctx,
metric_ctx.with_metric_engine,
ctx.extension(PHYSICAL_TABLE_PARAM)
.unwrap_or(GREPTIME_PHYSICAL_TABLE),
)
.await
.map_err(BoxedError::new)
.context(error::ExecuteGrpcQuerySnafu)?;
let rejections = prepared.point_rejections(preflight.rejections)?;
let otlp::metrics::MetricsConversion {
requests,
rows,
semantic_index,
resource_info,
mut outcome,
} = otlp::metrics::to_grpc_insert_requests(request, &mut metric_ctx)?;
} = prepared.finish(rejections, &preflight.physical_tables)?;
if outcome.rejected_data_points > 0 {
warn!(
"Rejected {} OTLP metrics data points: {}",
@@ -144,13 +167,10 @@ impl OpenTelemetryProtocolHandler for Instance {
return Ok(outcome);
}
self.check_row_insert_permission(&requests, &ctx, PermissionReq::Action(OTLP_WRITE))
.context(AuthSnafu)?;
let ctx = admit_write(rows as u64, &ctx)
.await
.map_err(BoxedError::new)
.context(error::ExecuteGrpcQuerySnafu)?;
self.cache_otlp_legacy(&input_names, &ctx, is_legacy)?;
OTLP_METRICS_ROWS.inc_by(rows as u64);
let ctx = {
@@ -163,9 +183,7 @@ impl OpenTelemetryProtocolHandler for Instance {
if let Some(index) = semantic_index.encode(&c.current_schema()) {
c.set_extension(SEMANTIC_PER_TABLE_INDEX_KEY, index);
}
if !is_legacy {
c.set_extension(OTLP_METRIC_COMPAT_KEY, OTLP_METRIC_COMPAT_PROM.to_string());
}
c.set_extension(OTLP_METRIC_COMPAT_KEY, OTLP_METRIC_COMPAT_PROM.to_string());
Arc::new(c)
};
@@ -333,9 +351,9 @@ impl Instance {
// The bulk path converts each request's time index unit to the
// destination table's unit during batch alignment, so no pre-gate
// alignment is needed here.
let batcher = self.logical_batcher().filter(|_| {
ctx.logical_batching_enabled() && !metric_ctx.is_legacy && metric_ctx.with_metric_engine
});
let batcher = self
.logical_batcher()
.filter(|_| ctx.logical_batching_enabled() && metric_ctx.with_metric_engine);
let batcher = if batcher.is_some()
&& self
.inserter
@@ -349,7 +367,7 @@ impl Instance {
None
};
// OTLP tables have one sample field in both the legacy and physical paths.
// OTLP tables have one sample field in both the ordinary and physical paths.
if let Some(batcher) = batcher {
let (rows, cost) = batcher
.submit_with(requests, ctx.clone(), |mut requests| {
@@ -366,7 +384,7 @@ impl Instance {
OutputData::AffectedRows(rows as usize),
OutputMeta::new_with_cost(cost as _),
))
} else if metric_ctx.is_legacy || !metric_ctx.with_metric_engine {
} else if !metric_ctx.with_metric_engine {
self.handle_row_inserts(requests, ctx.clone(), false, true)
.await
.map_err(BoxedError::new)
src/metric-engine/src/engine.rs
+20
View File
@@ -579,6 +579,26 @@ impl MetricEngine {
}
}
/// Shared definitions must agree before CREATE/ALTER mutates either region.
fn validate_shared_column(
region_id: store_api::storage::RegionId,
existing: &store_api::metadata::ColumnMetadata,
proposed: &store_api::metadata::ColumnMetadata,
) -> crate::error::Result<()> {
snafu::ensure!(
existing.column_schema.data_type == proposed.column_schema.data_type
&& existing.semantic_type == proposed.semantic_type,
crate::error::InvalidRequestSnafu {
region_id,
reason: format!(
"incompatible shared physical column `{}`",
proposed.column_schema.name
),
}
);
Ok(())
}
#[cfg(test)]
impl MetricEngine {
pub async fn scan_to_stream(
src/metric-engine/src/engine/alter.rs
+4 -4
View File
@@ -15,7 +15,7 @@
mod extract_new_columns;
mod validate;
use std::collections::{BTreeSet, HashMap, HashSet};
use std::collections::{BTreeSet, HashMap};
use api::v1::SemanticType;
use common_query::native_histogram::is_native_histogram_value_type;
@@ -123,7 +123,7 @@ impl MetricEngineInner {
.await?;
// Finds new columns to add
let mut new_column_names = HashSet::new();
let mut new_column_indices = HashMap::new();
let mut new_columns_to_add = vec![];
let index_options = {
@@ -139,7 +139,7 @@ impl MetricEngineInner {
extract_new_columns(
&requests,
physical_columns,
&mut new_column_names,
&mut new_column_indices,
&mut new_columns_to_add,
)?;
@@ -190,7 +190,7 @@ impl MetricEngineInner {
)
});
let new_add_columns = new_column_names.iter().map(|name| {
let new_add_columns = new_column_indices.keys().map(|name| {
// Safety: previous steps ensure the physical region exist
let column_metadata = *physical_schema_map.get(name).unwrap();
(name.to_string(), column_metadata.clone())
src/metric-engine/src/engine/alter/extract_new_columns.rs
+90 -10
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::collections::{HashMap, HashSet};
use std::collections::HashMap;
use api::v1::SemanticType;
use common_query::native_histogram::is_native_histogram_value_type;
@@ -31,18 +31,22 @@ use crate::error::{AddingFieldColumnSnafu, Result};
pub fn extract_new_columns<'a>(
requests: &'a [(RegionId, RegionAlterRequest)],
physical_columns: &HashMap<String, ColumnMetadata>,
new_column_names: &mut HashSet<&'a str>,
new_column_indices: &mut HashMap<&'a str, usize>,
new_columns: &mut Vec<ColumnMetadata>,
) -> Result<()> {
for (_, request) in requests {
for (region_id, request) in requests {
let AlterKind::AddColumns { columns } = &request.kind else {
unreachable!()
};
for col in columns {
let column_name = col.column_metadata.column_schema.name.as_str();
if !physical_columns.contains_key(column_name)
&& !new_column_names.contains(column_name)
{
if let Some(existing) = physical_columns.get(column_name).or_else(|| {
new_column_indices
.get(column_name)
.map(|&index| &new_columns[index])
}) {
crate::engine::validate_shared_column(*region_id, existing, &col.column_metadata)?;
} else {
ensure!(
col.column_metadata.semantic_type != SemanticType::Field
|| is_native_histogram_value_type(
@@ -52,7 +56,7 @@ pub fn extract_new_columns<'a>(
name: column_name.to_string(),
}
);
new_column_names.insert(column_name);
new_column_indices.insert(column_name, new_columns.len());
// TODO(weny): avoid clone
new_columns.push(col.column_metadata.clone());
}
@@ -64,7 +68,7 @@ pub fn extract_new_columns<'a>(
#[cfg(test)]
mod tests {
use std::collections::{HashMap, HashSet};
use std::collections::HashMap;
use api::v1::SemanticType;
use datatypes::prelude::ConcreteDataType;
@@ -76,6 +80,82 @@ mod tests {
use super::*;
use crate::error::Error;
#[test]
fn test_alter_rejects_conflicting_shared_definitions() {
let column = ColumnMetadata {
column_schema: ColumnSchema::new("shared", ConcreteDataType::string_datatype(), true),
semantic_type: SemanticType::Tag,
column_id: 0,
};
let mut incompatible = column.clone();
incompatible.column_schema =
ColumnSchema::new("shared", ConcreteDataType::int64_datatype(), true);
for existing in [false, true] {
let physical = if existing {
HashMap::from([("shared".into(), column.clone())])
} else {
HashMap::new()
};
let requests = [(
RegionId::new(1, 1),
RegionAlterRequest {
kind: AlterKind::AddColumns {
columns: [column.clone(), incompatible.clone()]
.into_iter()
.map(|column_metadata| AddColumn {
column_metadata,
location: None,
})
.collect(),
},
},
)];
let error =
extract_new_columns(&requests, &physical, &mut HashMap::new(), &mut Vec::new())
.unwrap_err();
assert!(matches!(error, Error::InvalidRequest { .. }));
}
}
#[test]
fn test_extract_new_columns_preserves_first_seen_order() {
let requests = [["b", "a"], ["a", "b"]].map(|names| {
(
RegionId::new(1, 1),
RegionAlterRequest {
kind: AlterKind::AddColumns {
columns: names
.into_iter()
.map(|name| AddColumn {
column_metadata: ColumnMetadata {
column_schema: ColumnSchema::new(
name,
ConcreteDataType::string_datatype(),
true,
),
semantic_type: SemanticType::Tag,
column_id: 0,
},
location: None,
})
.collect(),
},
},
)
});
let mut indices = HashMap::new();
let mut columns = Vec::new();
extract_new_columns(&requests, &HashMap::new(), &mut indices, &mut columns).unwrap();
assert_eq!(indices, HashMap::from([("b", 0), ("a", 1)]));
assert_eq!(
columns
.iter()
.map(|c| c.column_schema.name.as_str())
.collect::<Vec<_>>(),
["b", "a"]
);
}
#[test]
fn test_extract_new_columns_with_field_type() {
let requests = vec![(
@@ -99,13 +179,13 @@ mod tests {
)];
let physical_columns = HashMap::new();
let mut new_column_names = HashSet::new();
let mut new_column_indices = HashMap::new();
let mut new_columns = Vec::new();
let err = extract_new_columns(
&requests,
&physical_columns,
&mut new_column_names,
&mut new_column_indices,
&mut new_columns,
)
.unwrap_err();
src/metric-engine/src/engine/create.rs
+4 -4
View File
@@ -14,7 +14,7 @@
mod extract_new_columns;
use std::collections::{HashMap, HashSet};
use std::collections::HashMap;
use api::v1::SemanticType;
use common_query::native_histogram::is_native_histogram_value_type;
@@ -266,7 +266,7 @@ impl MetricEngineInner {
};
// Finds new columns to add to physical region
let mut new_column_names = HashSet::new();
let mut new_column_indices = HashMap::new();
let mut new_columns = Vec::new();
let index_option = {
@@ -282,7 +282,7 @@ impl MetricEngineInner {
extract_new_columns(
&requests,
physical_columns,
&mut new_column_names,
&mut new_column_indices,
&mut new_columns,
)?;
region_state.options().index
@@ -319,7 +319,7 @@ impl MetricEngineInner {
)
});
let new_add_columns = new_column_names.iter().map(|name| {
let new_add_columns = new_column_indices.keys().map(|name| {
// Safety: previous steps ensure the physical region exist
let column_metadata = *physical_schema_map.get(name).unwrap();
(name.to_string(), column_metadata.clone())
src/metric-engine/src/engine/create/extract_new_columns.rs
+68 -15
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::collections::{HashMap, HashSet};
use std::collections::HashMap;
use api::v1::SemanticType;
use common_query::native_histogram::is_native_histogram_value_type;
@@ -27,14 +27,18 @@ use crate::error::{AddingFieldColumnSnafu, Result};
pub fn extract_new_columns<'a>(
requests: &'a [(RegionId, RegionCreateRequest)],
physical_columns: &HashMap<String, ColumnMetadata>,
new_column_names: &mut HashSet<&'a str>,
new_column_indices: &mut HashMap<&'a str, usize>,
new_columns: &mut Vec<ColumnMetadata>,
) -> Result<()> {
for (_, request) in requests {
for (region_id, request) in requests {
for col in &request.column_metadatas {
if !physical_columns.contains_key(&col.column_schema.name)
&& !new_column_names.contains(&col.column_schema.name.as_str())
{
if let Some(existing) = physical_columns.get(&col.column_schema.name).or_else(|| {
new_column_indices
.get(col.column_schema.name.as_str())
.map(|&index| &new_columns[index])
}) {
crate::engine::validate_shared_column(*region_id, existing, col)?;
} else {
ensure!(
col.semantic_type != SemanticType::Field
|| is_native_histogram_value_type(&col.column_schema.data_type),
@@ -42,7 +46,7 @@ pub fn extract_new_columns<'a>(
name: col.column_schema.name.clone(),
}
);
new_column_names.insert(&col.column_schema.name);
new_column_indices.insert(&col.column_schema.name, new_columns.len());
// TODO(weny): avoid clone
new_columns.push(col.clone());
}
@@ -55,7 +59,7 @@ pub fn extract_new_columns<'a>(
#[cfg(test)]
mod tests {
use std::assert_matches;
use std::collections::{HashMap, HashSet};
use std::collections::HashMap;
use api::v1::SemanticType;
use datatypes::prelude::ConcreteDataType;
@@ -67,9 +71,51 @@ mod tests {
use super::*;
use crate::error::Error;
#[test]
fn test_conflicting_shared_definitions_rejected() {
let column = ColumnMetadata {
column_schema: ColumnSchema::new("shared", ConcreteDataType::string_datatype(), true),
semantic_type: SemanticType::Tag,
column_id: 0,
};
let mut incompatible = column.clone();
incompatible.semantic_type = SemanticType::Field;
let region = RegionId::new(1, 1);
assert!(crate::engine::validate_shared_column(region, &column, &incompatible).is_err());
incompatible.semantic_type = SemanticType::Tag;
incompatible.column_schema =
ColumnSchema::new("shared", ConcreteDataType::int64_datatype(), true);
assert!(crate::engine::validate_shared_column(region, &column, &incompatible).is_err());
assert!(crate::engine::validate_shared_column(region, &column, &column).is_ok());
for existing in [false, true] {
let physical = if existing {
HashMap::from([("shared".into(), column.clone())])
} else {
HashMap::new()
};
let requests = [(
region,
RegionCreateRequest {
column_metadatas: vec![column.clone(), incompatible.clone()],
engine: "metric".into(),
primary_key: vec![0],
options: HashMap::new(),
table_dir: "test".into(),
path_type: PathType::Bare,
partition_expr_json: None,
requirements: Default::default(),
},
)];
let error =
extract_new_columns(&requests, &physical, &mut HashMap::new(), &mut Vec::new())
.unwrap_err();
assert!(matches!(error, Error::InvalidRequest { .. }));
}
}
#[test]
fn test_extract_new_columns() {
let requests = vec![
let mut requests = vec![
(
RegionId::new(1, 1),
RegionCreateRequest {
@@ -126,6 +172,11 @@ mod tests {
),
];
let mut another = requests[0].1.column_metadatas[1].clone();
another.column_schema.name = "another_column".into();
requests[0].1.column_metadatas.push(another.clone());
requests[1].1.column_metadatas.push(another);
let mut physical_columns = HashMap::new();
physical_columns.insert(
"existing_column".to_string(),
@@ -139,20 +190,22 @@ mod tests {
column_id: 0,
},
);
let mut new_column_names = HashSet::new();
let mut new_column_indices = HashMap::new();
let mut new_columns = Vec::new();
let result = extract_new_columns(
&requests,
&physical_columns,
&mut new_column_names,
&mut new_column_indices,
&mut new_columns,
);
assert!(result.is_ok());
assert!(new_column_names.contains("new_column"));
assert_eq!(new_columns.len(), 1);
assert_eq!(new_column_indices["new_column"], 0);
assert_eq!(new_column_indices["another_column"], 1);
assert_eq!(new_columns.len(), 2);
assert_eq!(new_columns[0].column_schema.name, "new_column");
assert_eq!(new_columns[1].column_schema.name, "another_column");
}
#[test]
@@ -180,13 +233,13 @@ mod tests {
)];
let physical_columns = HashMap::new();
let mut new_column_names = HashSet::new();
let mut new_column_indices = HashMap::new();
let mut new_columns = Vec::new();
let err = extract_new_columns(
&requests,
&physical_columns,
&mut new_column_names,
&mut new_column_indices,
&mut new_columns,
)
.unwrap_err();
src/operator/src/insert.rs
+45 -31
View File
@@ -14,6 +14,9 @@
#[cfg(feature = "enterprise")]
mod flow_write;
mod otlp;
pub use otlp::MetricInsertPreflight;
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicU64, Ordering};
@@ -583,6 +586,9 @@ impl Inserter {
}
/// Handles row inserts request with metric engine.
///
/// Resolves or creates `physical_table` only when a logical target is missing.
/// Existing targets retain their backing tables and time-index units.
pub async fn handle_metric_row_inserts(
&self,
mut requests: RowInsertRequests,
@@ -601,22 +607,30 @@ impl Inserter {
});
validate_column_count_match(&requests)?;
// check and create physical table
let physical_table_ref = self
.create_physical_table_on_demand(&ctx, physical_table.clone(), statement_executor)
.await?;
// check and create logical tables; `create_or_alter_tables_on_demand`
// aligns each request's time index unit with the unit of the table it
// targets, inside its existing table lookups: existing tables keep
// their own unit (which matches the physical table they are bound
// to), and new tables use the selected physical table's unit (from
// `physical_table_ref`). Ingestion endpoints encode timestamps in a
// fixed unit (prometheus remote write always uses millisecond; OTLP
// keeps nanosecond precision on the metric engine path), and the
// metric engine requires each logical table's requests to match its
// time index unit. Narrowing conversions truncate the sub-unit part
// (floor), following `Timestamp::convert_to`.
// The selected backing table is needed only by new logical targets.
let mut has_new_tables = false;
for request in &requests.inserts {
if self
.get_table(
ctx.current_catalog(),
&ctx.current_schema(),
&request.table_name,
)
.await?
.is_none()
{
has_new_tables = true;
break;
}
}
let new_table_unit = if has_new_tables {
let physical = self
.create_physical_table_on_demand(&ctx, physical_table.clone(), statement_executor)
.await?;
table_time_index_unit(&physical)
} else {
None
};
let CreateAlterTableResult {
instant_table_ids,
table_infos,
@@ -628,7 +642,7 @@ impl Inserter {
statement_executor,
true,
true,
table_time_index_unit(&physical_table_ref),
new_table_unit,
)
.await?;
let name_to_info = table_infos
@@ -1065,11 +1079,10 @@ impl Inserter {
/// remote write. This will modify the `RowInsertRequests` in place.
/// `is_single_value` indicates whether the default schema only contains single value column so we can accommodate it.
///
/// `align_time_index_unit` is the selected physical metric table's time
/// index unit; passing `Some` (metric engine path only) rewrites each
/// request's time index column, inside this function's existing table
/// lookups (no extra catalog access): existing destination tables are
/// converted to their own unit, and new tables to the given unit.
/// `align_time_index_unit` supplies the selected physical table's time-index
/// unit for new logical tables. Existing logical targets always use their
/// own unit, even when this argument is `None` because no new target needs
/// the selected backing table.
#[allow(clippy::too_many_arguments)]
async fn create_or_alter_tables_on_demand(
&self,
@@ -1123,7 +1136,7 @@ impl Inserter {
// Metric path: an existing destination table keeps its own
// time index unit (it may be bound to another physical
// table than the one selected by this request).
if align_time_index_unit.is_some()
if matches!(auto_create_table_type, AutoCreateTableType::Logical(_))
&& let Some(rows) = req.rows.as_mut()
&& let Some(target_unit) = table_time_index_unit(&table)
{
@@ -1167,7 +1180,7 @@ impl Inserter {
// Metric path: an existing destination table keeps its
// own time index unit (it may be bound to another
// physical table than the one selected by this request).
if align_time_index_unit.is_some()
if matches!(auto_create_table_type, AutoCreateTableType::Logical(_))
&& let Some(rows) = req.rows.as_mut()
&& let Some(target_unit) = table_time_index_unit(&table)
{
@@ -1582,7 +1595,8 @@ impl Inserter {
/// When `accommodate_existing_schema` is true, it may modify the input `req` to
/// accommodate it with existing schema. See [`create_or_alter_tables_on_demand`](Self::create_or_alter_tables_on_demand)
/// for more details.
/// When `is_single_value` is true, it also rejects native-histogram/float kind changes.
/// When `is_single_value` is true, it also rejects native-histogram/float kind
/// changes even when no columns need adding.
/// When both options are true, it considers fields when modifying the input `req`.
fn get_alter_table_expr_on_demand(
&self,
@@ -1612,12 +1626,6 @@ impl Inserter {
.iter()
.filter(|col| col.semantic_type == SemanticType::Field as i32)
.count();
let column_exprs = ColumnExpr::from_column_schemas(request_schema);
let add_columns = expr_helper::extract_add_columns_expr(&table.schema(), column_exprs)?;
let Some(mut add_columns) = add_columns else {
return Ok(None);
};
if is_single_value {
let request_is_native_histogram = request_is_native_histogram(request_schema);
let table_is_native_histogram = table_is_native_histogram(table);
@@ -1631,6 +1639,12 @@ impl Inserter {
);
}
let column_exprs = ColumnExpr::from_column_schemas(request_schema);
let add_columns = expr_helper::extract_add_columns_expr(&table.schema(), column_exprs)?;
let Some(mut add_columns) = add_columns else {
return Ok(None);
};
// If accommodate_existing_schema is true, update request schema for Timestamp/Field columns
if accommodate_existing_schema {
let table_schema = table.schema();
src/operator/src/insert/otlp.rs
+828
View File
@@ -0,0 +1,828 @@
// 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.
//! Side-effect-free destination validation for prepared OTLP batches. Catalog
//! failures propagate; row errors let the converter reject complete source-point fan-outs.
use std::collections::HashSet;
use ahash::{HashMap as AHashMap, HashMapExt};
use api::helper::{ColumnDataTypeWrapper, to_grpc_value};
use api::v1::alter_table_expr::Kind;
use api::v1::column_def::try_as_column_schema;
use api::v1::{
ColumnDataType, ColumnDef, ColumnSchema, Row, RowInsertRequest, Rows, SemanticType, Value,
};
use common_query::native_histogram::native_histogram_value_type;
use common_query::prelude::{greptime_timestamp, greptime_value};
use common_time::timestamp::TimeUnit;
use datatypes::schema::ColumnSchema as DataColumn;
use session::context::QueryContextRef;
use snafu::prelude::*;
use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, PHYSICAL_TABLE_METADATA_KEY};
use table::TableRef;
use crate::error::{self, InvalidInsertRequestSnafu, Result};
use crate::insert::{
AutoCreateTableType, Inserter, convert_rows_time_unit, convert_timestamp_value_data,
request_is_native_histogram, table_is_native_histogram, table_time_index_unit,
};
use crate::req_convert::insert::{ImpureDefaultFiller, value_matches_type};
pub struct MetricInsertPreflight {
/// Rejections in batch order, then row order.
pub rejections: Vec<Vec<Option<String>>>,
pub physical_tables: std::collections::HashMap<String, String>,
}
struct PhysicalColumns {
by_name: AHashMap<String, (DataColumn, SemanticType)>,
fields: Vec<String>,
}
impl PhysicalColumns {
fn new(columns: Vec<(DataColumn, SemanticType)>) -> Self {
let fields = columns
.iter()
.filter(|(_, semantic)| *semantic == SemanticType::Field)
.map(|(column, _)| column.name.clone())
.collect();
let by_name = columns
.into_iter()
.map(|(column, semantic)| (column.name.clone(), (column, semantic)))
.collect();
Self { by_name, fields }
}
}
impl Inserter {
/// Validate each row independently, sharing schema planning within each batch.
/// Fresh cross-point conflicts are resolved by the converter after this pass.
pub async fn validate_and_normalize_otlp_metrics(
&self,
batches: &mut [RowInsertRequest],
ctx: &QueryContextRef,
metric_engine: bool,
physical_table: &str,
) -> Result<MetricInsertPreflight> {
let disabled = self.auto_create_disabled_reason(ctx)?;
let mut tables = AHashMap::new();
let mut physical_tables = std::collections::HashMap::new();
for request in batches.iter() {
if !tables.contains_key(&request.table_name) {
let table = self
.get_table(
ctx.current_catalog(),
&ctx.current_schema(),
&request.table_name,
)
.await?;
tables.insert(request.table_name.clone(), table);
}
let backing = match tables[&request.table_name].as_ref() {
Some(table) => table
.table_info()
.meta
.options
.extra_options
.get(LOGICAL_TABLE_METADATA_KEY)
.cloned(),
None if metric_engine => Some(physical_table.to_string()),
None => None,
};
if let Some(backing) = backing {
physical_tables.insert(request.table_name.clone(), backing);
}
}
// Shared schemas accumulate tags across logical tables; index them once per request.
let mut physical_columns = AHashMap::new();
for backing in physical_tables.values() {
if !tables.contains_key(backing) {
let table = self
.get_table(ctx.current_catalog(), &ctx.current_schema(), backing)
.await?;
tables.insert(backing.clone(), table);
}
physical_columns.entry(backing.as_str()).or_insert_with(|| {
PhysicalColumns::new(
tables[backing]
.as_ref()
.map(table_columns)
.unwrap_or_else(default_physical_columns),
)
});
}
// Match ordinary insertion: impure defaults are evaluated once per table
// and request, while supplied nulls remain supplied nulls.
let mut fillers = AHashMap::new();
let mut build_plan = |source: &mut RowInsertRequest| {
let rows = source.rows.as_mut().context(InvalidInsertRequestSnafu {
reason: "missing metric rows",
})?;
let input_columns = rows.schema.len();
let source_timestamp = timestamp_column(&rows.schema);
let request = &mut RowInsertRequest {
table_name: source.table_name.clone(),
rows: Some(Rows {
schema: std::mem::take(&mut rows.schema),
rows: Vec::new(),
}),
};
let auto_create = disabled.is_none()
|| Self::is_auto_create_exempt_private_table(
&ctx.current_schema(),
&request.table_name,
);
let target = tables[&request.table_name].as_ref();
let backing = physical_tables.get(&request.table_name);
let physical = backing.and_then(|name| tables[name].as_ref());
if let Some(physical) = physical {
ensure!(
physical
.table_info()
.meta
.options
.extra_options
.contains_key(PHYSICAL_TABLE_METADATA_KEY),
InvalidInsertRequestSnafu {
reason: "selected backing table is not a physical metric table"
}
);
}
if metric_engine {
ensure!(
backing.is_none() || physical.is_some() || auto_create,
InvalidInsertRequestSnafu {
reason: format!(
"missing physical table and {}",
disabled.unwrap_or_default()
),
}
);
let unit = target
.and_then(table_time_index_unit)
.or_else(|| physical.and_then(table_time_index_unit))
.unwrap_or(TimeUnit::Millisecond);
if let Some(rows) = &mut request.rows {
convert_rows_time_unit(rows, unit)?;
}
}
let mut columns = if let Some(table) = target {
ensure!(
!table
.table_info()
.meta
.options
.extra_options
.contains_key(PHYSICAL_TABLE_METADATA_KEY),
InvalidInsertRequestSnafu {
reason: "direct writes to a physical metric table are forbidden",
}
);
ensure!(
request_is_native_histogram(&request.rows.as_ref().unwrap().schema)
== table_is_native_histogram(table),
InvalidInsertRequestSnafu {
reason: format!(
"Table `{}` cannot mix native histogram and float sample fields",
request.table_name
),
}
);
let mut columns = table_columns(table);
if auto_create {
// Ordinary OTLP insertion does not rename columns.
if let Some(alter) = self.get_alter_table_expr_on_demand(
request,
table,
ctx,
metric_engine,
true,
true,
)? && let Some(Kind::AddColumns(add)) = alter.kind
{
for column in add.add_columns {
let def =
column
.column_def
.with_context(|| InvalidInsertRequestSnafu {
reason: "missing column definition",
})?;
columns.push(column_definition(&def)?);
}
}
}
columns
} else {
ensure!(
auto_create,
InvalidInsertRequestSnafu {
reason: format!(
"missing table `{}` and {}",
request.table_name,
disabled.unwrap_or_default()
),
}
);
let create_type = if metric_engine {
AutoCreateTableType::Logical(physical_table.to_string())
} else {
AutoCreateTableType::Physical
};
let create =
self.get_create_table_expr_on_demand(request, &create_type, ctx, None)?;
if metric_engine {
ensure!(
create
.column_defs
.iter()
.filter(|c| c.semantic_type == SemanticType::Field as i32)
.count()
== 1,
InvalidInsertRequestSnafu {
reason: "logical metrics require exactly one sample field"
}
);
}
create
.column_defs
.iter()
.map(column_definition)
.collect::<Result<Vec<_>>>()?
};
let rows = request.rows.as_mut().context(InvalidInsertRequestSnafu {
reason: "missing metric rows",
})?;
if let Some(backing) = backing {
// Logical tag nullability follows the shared physical schema.
for (column, semantic) in &mut columns {
if *semantic == SemanticType::Tag {
column.set_nullable();
}
}
validate_physical_columns(rows, &physical_columns[backing.as_str()], auto_create)?;
}
// Only defaults need values in the planning template.
rows.rows.push(Row::default());
if let Some(table) = target {
if !fillers.contains_key(&request.table_name) {
fillers.insert(
request.table_name.clone(),
ImpureDefaultFiller::new(table.table_info())?,
);
}
fillers[&request.table_name].fill_rows(rows);
}
RowValidationPlan::new(
input_columns,
source_timestamp,
std::mem::take(rows),
&columns,
)
};
let mut rejections = Vec::with_capacity(batches.len());
for request in batches {
let num_rows = request
.rows
.as_ref()
.context(InvalidInsertRequestSnafu {
reason: "missing metric rows",
})?
.rows
.len();
let plan = match build_plan(request) {
Ok(plan) => plan,
Err(err) if attributable(&err) => {
rejections.push(vec![Some(err.to_string()); num_rows]);
continue;
}
Err(err) => return Err(err),
};
let rows = request.rows.as_mut().context(InvalidInsertRequestSnafu {
reason: "missing metric rows",
})?;
let mut errors = Vec::with_capacity(num_rows);
for row in &mut rows.rows {
match plan.apply_row(row) {
Ok(()) => errors.push(None),
Err(err) if attributable(&err) => errors.push(Some(err.to_string())),
Err(err) => return Err(err),
}
}
rows.schema = plan.schema;
rejections.push(errors);
}
Ok(MetricInsertPreflight {
rejections,
physical_tables,
})
}
}
fn attributable(err: &error::Error) -> bool {
matches!(
err,
error::Error::InvalidInsertRequest { .. }
| error::Error::BuildCreateExprOnInsertion { .. }
| error::Error::FindNewColumnsOnInsertion { .. }
| error::Error::InvalidColumnDef { .. }
)
}
fn invalid(reason: impl Into<String>) -> error::Error {
InvalidInsertRequestSnafu {
reason: reason.into(),
}
.build()
}
fn column_definition(def: &ColumnDef) -> Result<(DataColumn, SemanticType)> {
let column = try_as_column_schema(def).map_err(|err| invalid(err.to_string()))?;
let semantic =
SemanticType::try_from(def.semantic_type).map_err(|err| invalid(err.to_string()))?;
Ok((column, semantic))
}
fn table_columns(table: &TableRef) -> Vec<(DataColumn, SemanticType)> {
let info = table.table_info();
let primary_keys: HashSet<_> = info.meta.primary_key_indices.iter().copied().collect();
info.meta
.schema
.column_schemas()
.iter()
.enumerate()
.map(|(index, column)| {
let semantic = if column.is_time_index() {
SemanticType::Timestamp
} else if primary_keys.contains(&index) {
SemanticType::Tag
} else {
SemanticType::Field
};
(column.clone(), semantic)
})
.collect()
}
fn default_physical_columns() -> Vec<(DataColumn, SemanticType)> {
use datatypes::prelude::ConcreteDataType;
vec![
(
DataColumn::new(
greptime_timestamp(),
ConcreteDataType::timestamp_millisecond_datatype(),
false,
)
.with_time_index(true),
SemanticType::Timestamp,
),
(
DataColumn::new(greptime_value(), ConcreteDataType::float64_datatype(), true),
SemanticType::Field,
),
]
}
fn check_definition(
source: &ColumnSchema,
target: &DataColumn,
semantic: SemanticType,
) -> Result<()> {
ensure!(
api::helper::is_column_type_value_eq(
source.datatype,
source.datatype_extension.clone(),
&target.data_type
) && source.semantic_type == semantic as i32,
InvalidInsertRequestSnafu {
reason: format!(
"incompatible datatype or semantic type for column `{}`",
source.column_name
),
}
);
Ok(())
}
fn validate_physical_columns(
rows: &Rows,
columns: &PhysicalColumns,
auto_create: bool,
) -> Result<()> {
for source in &rows.schema {
if let Some((target, semantic)) = columns.by_name.get(&source.column_name) {
check_definition(source, target, *semantic)?;
} else {
ensure!(
auto_create,
InvalidInsertRequestSnafu {
reason: format!(
"missing physical column `{}` while auto-DDL is disabled",
source.column_name
)
}
);
ensure!(
source.semantic_type == SemanticType::Tag as i32
|| (source.semantic_type == SemanticType::Field as i32
&& api::helper::is_column_type_value_eq(
source.datatype,
source.datatype_extension.clone(),
native_histogram_value_type()
)),
InvalidInsertRequestSnafu {
reason: format!("cannot add physical column `{}`", source.column_name)
}
);
}
}
for name in &columns.fields {
if !rows.schema.iter().any(|c| c.column_name == *name) {
let (column, _) = &columns.by_name[name];
ensure!(
!column.is_default_impure(),
InvalidInsertRequestSnafu {
reason: format!("impure default for physical field `{}`", column.name)
}
);
ensure!(
column
.create_default()
.map_err(|e| invalid(e.to_string()))?
.is_some(),
InvalidInsertRequestSnafu {
reason: format!("missing required physical field `{}`", column.name)
}
);
}
}
Ok(())
}
/// Schema decisions are shared; row values are always checked independently.
struct RowValidationPlan {
input_columns: usize,
schema: Vec<ColumnSchema>,
nullable: Vec<bool>,
defaults: Vec<Value>,
timestamp: Option<(usize, TimeUnit, TimeUnit)>,
}
impl RowValidationPlan {
fn new(
input_columns: usize,
source_timestamp: Option<(usize, TimeUnit)>,
mut template: Rows,
columns: &[(DataColumn, SemanticType)],
) -> Result<Self> {
let by_name: AHashMap<_, _> = columns
.iter()
.map(|(column, semantic)| (column.name.as_str(), (column, *semantic)))
.collect();
let mut names = HashSet::new();
let mut nullable = Vec::with_capacity(columns.len());
for source in &template.schema {
ensure!(
names.insert(source.column_name.as_str()),
InvalidInsertRequestSnafu {
reason: format!("duplicate column `{}`", source.column_name),
}
);
let (target, semantic) = by_name
.get(source.column_name.as_str())
.ok_or_else(|| invalid(format!("unknown column `{}`", source.column_name)))?;
check_definition(source, target, *semantic)?;
nullable.push(target.is_nullable());
}
let missing = columns
.iter()
.filter(|(column, _)| !names.contains(column.name.as_str()))
.collect::<Vec<_>>();
let mut row = template.rows.pop().context(InvalidInsertRequestSnafu {
reason: "missing normalization template",
})?;
// Resolve omissions before schemas are merged; an explicit null never uses a default.
for (column, semantic) in missing {
let value = column
.create_default()
.map_err(|err| invalid(err.to_string()))?
.ok_or_else(|| invalid(format!("missing required column `{}`", column.name)))?;
let (datatype, datatype_extension) =
ColumnDataTypeWrapper::try_from(column.data_type.clone())
.map_err(|err| invalid(err.to_string()))?
.into_parts();
template.schema.push(ColumnSchema {
column_name: column.name.clone(),
datatype: datatype as i32,
semantic_type: *semantic as i32,
datatype_extension,
options: None,
});
row.values.push(to_grpc_value(value));
nullable.push(column.is_nullable());
}
ensure!(
row.values.len() + input_columns == template.schema.len(),
InvalidInsertRequestSnafu {
reason: "invalid normalization template"
}
);
let defaults = row.values;
for ((source, nullable), value) in template.schema[input_columns..]
.iter()
.zip(&nullable[input_columns..])
.zip(&defaults)
{
validate_value(value, source, *nullable)?;
}
let timestamp = source_timestamp.and_then(|(index, source)| {
let target = ColumnDataType::try_from(template.schema[index].datatype)
.ok()
.and_then(api::helper::timestamp_unit)?;
(source != target).then_some((index, source, target))
});
Ok(Self {
input_columns,
schema: template.schema,
nullable,
defaults,
timestamp,
})
}
fn apply_row(&self, row: &mut Row) -> Result<()> {
ensure!(
row.values.len() == self.input_columns,
InvalidInsertRequestSnafu {
reason: "column count mismatch",
}
);
if let Some((index, source, target)) = self.timestamp
&& let Some(value) = row.values[index].value_data.take()
{
row.values[index].value_data =
convert_timestamp_value_data(value, source, target, index)?;
}
for ((source, nullable), value) in self.schema.iter().zip(&self.nullable).zip(&row.values) {
validate_value(value, source, *nullable)?;
}
row.values.extend_from_slice(&self.defaults);
Ok(())
}
}
fn timestamp_column(schema: &[ColumnSchema]) -> Option<(usize, TimeUnit)> {
let index = schema
.iter()
.position(|column| column.semantic_type == SemanticType::Timestamp as i32)?;
let unit = ColumnDataType::try_from(schema[index].datatype)
.ok()
.and_then(api::helper::timestamp_unit)?;
Some((index, unit))
}
fn validate_value(value: &Value, source: &ColumnSchema, nullable: bool) -> Result<()> {
ensure!(
value_matches_type(value, source.datatype, source.datatype_extension.as_ref()),
InvalidInsertRequestSnafu {
reason: format!("value datatype differs for `{}`", source.column_name)
}
);
ensure!(
value.value_data.is_some() || nullable,
InvalidInsertRequestSnafu {
reason: format!(
"null supplied for non-nullable column `{}`",
source.column_name
),
}
);
Ok(())
}
#[cfg(test)]
fn validate_and_fill_rows(rows: &mut Rows, columns: &[(DataColumn, SemanticType)]) -> Result<()> {
let template = Rows {
schema: rows.schema.clone(),
rows: vec![Row::default()],
};
let plan = RowValidationPlan::new(
rows.schema.len(),
timestamp_column(&rows.schema),
template,
columns,
)?;
for row in &mut rows.rows {
plan.apply_row(row)?;
}
rows.schema = plan.schema;
Ok(())
}
#[cfg(test)]
mod tests {
use api::v1::value::ValueData;
use api::v1::{ColumnDataType, Row};
use datatypes::prelude::ConcreteDataType;
use datatypes::schema::ColumnDefaultConstraint;
use datatypes::value::Value as DataValue;
use super::*;
#[test]
fn test_reused_plan_checks_each_value_and_timestamp() {
let schema = vec![api::v1::helper::time_index_column_schema(
greptime_timestamp(),
ColumnDataType::TimestampSecond,
)];
let mut template = Rows {
schema: schema.clone(),
rows: Vec::new(),
};
convert_rows_time_unit(&mut template, TimeUnit::Nanosecond).unwrap();
let columns = vec![(
DataColumn::new(
greptime_timestamp(),
ConcreteDataType::timestamp_nanosecond_datatype(),
false,
)
.with_time_index(true),
SemanticType::Timestamp,
)];
template.rows.push(Row::default());
let plan =
RowValidationPlan::new(schema.len(), timestamp_column(&schema), template, &columns)
.unwrap();
for (value, valid) in [
(Some(ValueData::TimestampSecondValue(1)), true),
(Some(ValueData::StringValue("invalid".into())), false),
(None, false),
(Some(ValueData::TimestampSecondValue(i64::MAX)), false),
(Some(ValueData::TimestampSecondValue(2)), true),
] {
let mut rows = Rows {
schema: schema.clone(),
rows: vec![Row {
values: vec![Value { value_data: value }],
}],
};
assert_eq!(plan.apply_row(&mut rows.rows[0]).is_ok(), valid);
if valid {
assert_eq!(
plan.schema[0].datatype,
ColumnDataType::TimestampNanosecond as i32
);
assert!(matches!(
rows.rows[0].values[0].value_data,
Some(ValueData::TimestampNanosecondValue(
1_000_000_000 | 2_000_000_000
))
));
}
}
}
#[test]
fn test_preflight_omissions_defaults_and_explicit_nulls() {
let mut columns = default_physical_columns();
columns.push((
DataColumn::new("host", ConcreteDataType::string_datatype(), false)
.with_default_constraint(Some(ColumnDefaultConstraint::Value(DataValue::String(
"unknown".into(),
))))
.unwrap(),
SemanticType::Tag,
));
let row = || Rows {
schema: vec![ColumnSchema {
column_name: greptime_timestamp().into(),
datatype: ColumnDataType::TimestampMillisecond as i32,
semantic_type: SemanticType::Timestamp as i32,
..Default::default()
}],
rows: vec![Row {
values: vec![Value {
value_data: Some(ValueData::TimestampMillisecondValue(1)),
}],
}],
};
let mut omitted = row();
validate_and_fill_rows(&mut omitted, &columns).unwrap();
assert_eq!(omitted.schema[2].column_name, "host");
assert_eq!(
omitted.rows[0].values[2].value_data,
Some(ValueData::StringValue("unknown".into()))
);
omitted.rows[0].values[2] = Value::default();
assert!(validate_and_fill_rows(&mut omitted, &columns).is_err());
columns[2].0 = DataColumn::new("host", ConcreteDataType::string_datatype(), false);
assert!(validate_and_fill_rows(&mut row(), &columns).is_err());
columns[2].0.set_nullable();
let mut nullable = row();
validate_and_fill_rows(&mut nullable, &columns).unwrap();
assert_eq!(nullable.rows[0].values[2].value_data, None);
}
#[test]
fn test_preflight_existing_columns_and_physical_definitions() {
let columns = default_physical_columns();
let mut rows = Rows {
schema: vec![ColumnSchema {
column_name: greptime_timestamp().into(),
datatype: ColumnDataType::TimestampMillisecond as i32,
semantic_type: SemanticType::Timestamp as i32,
..Default::default()
}],
rows: vec![Row {
values: vec![Value {
value_data: Some(ValueData::TimestampMillisecondValue(1)),
}],
}],
};
validate_and_fill_rows(&mut rows, &columns).unwrap();
for case in 0..3 {
let mut invalid = rows.clone();
match case {
0 => invalid.schema[0].semantic_type = SemanticType::Tag as i32,
1 => invalid.schema[0].datatype = ColumnDataType::TimestampNanosecond as i32,
_ => {
invalid.rows[0].values[0].value_data =
Some(ValueData::StringValue("bad".into()))
}
}
assert!(validate_and_fill_rows(&mut invalid, &columns).is_err());
}
let columns = PhysicalColumns::new(columns);
rows.schema[1].column_name = "new_field".into();
assert!(validate_physical_columns(&rows, &columns, true).is_err());
rows.schema[1].semantic_type = SemanticType::Tag as i32;
rows.schema[1].datatype = ColumnDataType::String as i32;
assert!(validate_physical_columns(&rows, &columns, false).is_err());
assert!(validate_physical_columns(&rows, &columns, true).is_ok());
}
#[test]
fn test_preflight_shared_physical_field_defaults() {
let rows = Rows {
schema: vec![api::v1::helper::time_index_column_schema(
greptime_timestamp(),
ColumnDataType::TimestampMillisecond,
)],
..Default::default()
};
let required = DataColumn::new("required", ConcreteDataType::float64_datatype(), false);
let defaulted = required
.clone()
.with_default_constraint(Some(ColumnDefaultConstraint::Value(DataValue::Float64(
1.0.into(),
))))
.unwrap();
let impure = DataColumn::new(
"impure",
ConcreteDataType::timestamp_millisecond_datatype(),
false,
)
.with_default_constraint(Some(ColumnDefaultConstraint::Function(
"current_timestamp()".into(),
)))
.unwrap();
for (field, error) in [
(required, Some("missing required physical field `required`")),
(defaulted, None),
(impure, Some("impure default for physical field `impure`")),
] {
let mut columns = default_physical_columns();
columns.extend((0..1024).map(|index| {
(
DataColumn::new(
format!("unrelated_tag_{index}"),
ConcreteDataType::string_datatype(),
true,
),
SemanticType::Tag,
)
}));
columns.push((field, SemanticType::Field));
let columns = PhysicalColumns::new(columns);
for auto_create in [false, true] {
let result = validate_physical_columns(&rows, &columns, auto_create);
if let Some(error) = error {
assert!(result.unwrap_err().to_string().contains(error));
} else {
result.unwrap();
}
}
}
}
}
src/operator/src/req_convert/insert.rs
+1
View File
@@ -25,6 +25,7 @@ pub use column_to_row::ColumnToRow;
pub(crate) use fill_impure_default::ImpureDefaultFiller;
pub use fill_impure_default::fill_reqs_with_impure_default;
pub use row_to_batch::rows_to_record_batch;
pub(crate) use row_to_batch::value_matches_type;
pub use row_to_region::RowToRegion;
use snafu::{OptionExt, ResultExt};
pub use stmt_to_region::StatementToRegion;
src/operator/src/req_convert/insert/row_to_batch.rs
+1 -1
View File
@@ -151,7 +151,7 @@ pub fn rows_to_record_batch(rows: &Rows, table_info: &TableInfo) -> Result<Recor
}
// Validate nested values before the infallible protobuf conversion reads type extensions.
fn value_matches_type(
pub(crate) fn value_matches_type(
value: &Value,
datatype: i32,
extension: Option<&ColumnDataTypeExtension>,
src/query/src/optimizer/windowed_sort.rs
+19
View File
@@ -85,6 +85,10 @@ impl WindowedSortPhysicalRule {
let Some(scanner_info) = fetch_partition_range(sort_input.clone())? else {
return Ok(Transformed::no(plan));
};
// Windowed sorting requires partition ranges, which file-engine scans lack.
if scanner_info.partition_ranges.is_empty() {
return Ok(Transformed::no(plan));
}
let input_schema = sort_input.schema();
let first_sort_expr = sort_exec.expr().first();
@@ -350,6 +354,21 @@ mod test {
use super::*;
#[test]
fn test_keep_ordinary_sort_when_scan_has_no_partition_ranges() {
let ordering = datafusion_physical_expr::LexOrdering::new([
datafusion_physical_expr::PhysicalSortExpr {
expr: Arc::new(PhysicalColumn::new("ts", 1)),
options: Default::default(),
},
])
.unwrap();
let sort = Arc::new(SortExec::new(ordering, new_region_scan()));
let optimized =
WindowedSortPhysicalRule::do_optimize(sort, &ConfigOptions::default()).unwrap();
assert!(optimized.is::<SortExec>());
}
#[test]
fn test_is_time_index_expr_tracks_aliases_through_projection() {
let scan = new_region_scan();
src/servers/Cargo.toml
+8
View File
@@ -207,3 +207,11 @@ harness = false
[[bench]]
name = "flush_batch_physical"
harness = false
[[bench]]
name = "otlp_metrics"
harness = false
[[bench]]
name = "otlp_metrics_preflight"
harness = false
src/servers/benches/otlp_metrics.rs
+84
View File
@@ -0,0 +1,84 @@
// 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.
#[path = "otlp_metrics/fixtures.rs"]
mod fixtures;
use std::hint::black_box;
use std::time::Duration;
use criterion::{BatchSize, BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
use servers::otlp::metrics::to_grpc_insert_requests;
use session::protocol_ctx::OtlpMetricCtx;
fn bench_conversion(c: &mut Criterion) {
let fixtures = fixtures::fixtures();
for (group_name, delta) in [
("otlp_conversion/common", false),
("otlp_conversion/delta", true),
] {
let mut group = c.benchmark_group(group_name);
for fixture in fixtures
.iter()
.filter(|fixture| fixture.name.starts_with("delta") == delta)
{
group.throughput(Throughput::Elements(fixture.points));
for resource_info in [false, true] {
let ctx = OtlpMetricCtx {
resource_info,
with_metric_engine: true,
..Default::default()
};
group.bench_function(
BenchmarkId::new(
fixture.name,
if resource_info {
"descriptors_on"
} else {
"descriptors_off"
},
),
|b| {
let output =
to_grpc_insert_requests(fixture.request.clone(), &mut ctx.clone())
.unwrap();
assert_eq!(
output.outcome.accepted_data_points,
fixture.points as i64 - fixture.rejected
);
assert_eq!(output.outcome.rejected_data_points, fixture.rejected);
assert_eq!(output.resource_info.is_some(), resource_info);
b.iter_batched(
|| fixture.request.clone(),
|request| {
black_box(
to_grpc_insert_requests(request, &mut ctx.clone()).unwrap(),
)
},
BatchSize::LargeInput,
);
},
);
}
}
group.finish();
}
}
criterion_group! {
name = benches;
config = Criterion::default().sample_size(20).warm_up_time(Duration::from_secs(1)).measurement_time(Duration::from_secs(3));
targets = bench_conversion
}
criterion_main!(benches);
src/servers/benches/otlp_metrics/README.md
+51
View File
@@ -0,0 +1,51 @@
# OTLP metrics benchmarks
Run from the workspace root:
```sh
cargo bench -p servers --bench otlp_metrics
cargo bench -p servers --bench otlp_metrics_preflight
```
The conversion benchmark measures `to_grpc_insert_requests`, including point
preparation, encoding, survivor merging, and optional resource descriptors. The
preflight benchmark measures destination validation against existing tables in
an in-memory catalog, with a fresh per-request schema cache. The same binary also
measures `otlp_ingestion`, which runs preparation, destination validation, and
survivor finalization together. Request cloning, fixture construction, and catalog
setup are outside the timed section. Neither
benchmark measures network requests, table creation, storage writes, or the full
ingestion endpoint. Both use the server's jemalloc allocator on non-Windows hosts.
Cases cover one point, 1,024 points with eight tags, 64 tags, classic histogram
fan-out, cumulative and delta exponential histograms, and mixed valid/invalid
points. `mixed_1024` rejects every tenth point during conversion;
`required_tag_1024` converts all points but rejects every tenth during preflight
because an ordinary table requires a missing tag. Other preflight cases use
logical tables with a shared physical schema. Each case checks accepted or
rejected counts before measuring. Conversion runs with descriptors on and off;
preflight does not validate the separate descriptor write.
To compare with `main`, copy `otlp_metrics.rs` and `otlp_metrics/fixtures.rs`
into an isolated checkout of that revision and register the `otlp_metrics` bench
in `src/servers/Cargo.toml` with `harness = false`. Use the same toolchain, build
profile, and machine for both revisions. Run serially, with other CPU-heavy work
stopped, and store both runs in the same absolute `CRITERION_HOME`:
```sh
# In the base checkout:
cargo bench -p servers --bench otlp_metrics -- otlp_conversion/common --save-baseline main
# In the feature checkout:
cargo bench -p servers --bench otlp_metrics -- otlp_conversion/common --baseline main
```
Delta conversion, preflight, and combined ingestion are measured only on the
feature branch because the base does not support them. A before/after conversion
ratio measures the whole branch's conversion changes, not the overhead of
preflight. These are local microbenchmarks; use an ingestion workload to measure
end-to-end throughput.
For staged refactors, save a baseline for the unmodified feature head and each
step using `--save-baseline <name>`. Run both binaries serially with the same
fixtures and settings. Compare each step against the preceding step and the
original head; only common conversion cases can also be compared with `main`.
src/servers/benches/otlp_metrics/fixtures.rs
+157
View File
@@ -0,0 +1,157 @@
// 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 otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, KeyValue, any_value};
use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
AggregationTemporality, ExponentialHistogram, ExponentialHistogramDataPoint, Gauge, Histogram,
HistogramDataPoint, Metric, NumberDataPoint, ResourceMetrics, ScopeMetrics,
exponential_histogram_data_point::Buckets, metric, number_data_point,
};
use otel_arrow_rust::proto::opentelemetry::resource::v1::Resource;
// Match the server allocator when measuring allocation-heavy conversion and validation.
#[cfg(not(windows))]
#[global_allocator]
static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
pub struct Fixture {
pub name: &'static str,
pub request: ExportMetricsServiceRequest,
pub points: u64,
pub rejected: i64,
}
pub fn fixtures() -> Vec<Fixture> {
[
("gauge_1", 1, 8),
("gauge_1024", 1024, 8),
("wide_1024", 1024, 64),
("histogram_1024", 1024, 8),
("cumulative_1024", 1024, 8),
("mixed_1024", 1024, 8),
("required_tag_1024", 1024, 8),
("delta_1024", 1024, 8),
]
.into_iter()
.map(|(name, points, labels)| {
let attributes = |index: u64| {
(0..labels)
.filter(|&label| {
!(name == "required_tag_1024" && label == 0 && index.is_multiple_of(10))
})
.map(|label| {
attribute(&format!("label_{label}"), &format!("value_{}", index % 100))
})
.collect()
};
let time = |index| 60_000_000_000 + index * 1_000_000;
let data = if name.starts_with("histogram") {
metric::Data::Histogram(Histogram {
aggregation_temporality: AggregationTemporality::Cumulative as i32,
data_points: (0..points)
.map(|index| HistogramDataPoint {
attributes: attributes(index),
time_unix_nano: time(index),
count: 11,
sum: Some(55.0),
bucket_counts: vec![1; 11],
explicit_bounds: (1..=10).map(f64::from).collect(),
..Default::default()
})
.collect(),
})
} else if name.starts_with("cumulative")
|| name.starts_with("mixed")
|| name.starts_with("delta")
{
metric::Data::ExponentialHistogram(ExponentialHistogram {
aggregation_temporality: if name.starts_with("delta") {
AggregationTemporality::Delta
} else {
AggregationTemporality::Cumulative
} as i32,
data_points: (0..points)
.map(|index| ExponentialHistogramDataPoint {
attributes: attributes(index),
time_unix_nano: time(index),
count: 16,
sum: Some(100.0),
scale: if name.starts_with("mixed") && index.is_multiple_of(10) {
-5
} else {
0
},
zero_threshold: 0.001,
positive: Some(Buckets {
offset: 0,
bucket_counts: vec![1; 16],
}),
..Default::default()
})
.collect(),
})
} else {
metric::Data::Gauge(Gauge {
data_points: (0..points)
.map(|index| NumberDataPoint {
attributes: attributes(index),
time_unix_nano: time(index),
value: Some(number_data_point::Value::AsDouble(index as f64)),
..Default::default()
})
.collect(),
})
};
Fixture {
name,
points,
rejected: if name.starts_with("mixed") {
points.div_ceil(10) as i64
} else {
0
},
request: ExportMetricsServiceRequest {
resource_metrics: vec![ResourceMetrics {
resource: Some(Resource {
attributes: vec![
attribute("service.name", "bench"),
attribute("host.id", "host"),
],
..Default::default()
}),
scope_metrics: vec![ScopeMetrics {
metrics: vec![Metric {
name: name.into(),
data: Some(data),
..Default::default()
}],
..Default::default()
}],
..Default::default()
}],
},
}
})
.collect()
}
fn attribute(key: &str, value: &str) -> KeyValue {
KeyValue {
key: key.into(),
value: Some(AnyValue {
value: Some(any_value::Value::StringValue(value.into())),
}),
}
}
src/servers/benches/otlp_metrics_preflight.rs
+260
View File
@@ -0,0 +1,260 @@
// 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.
#[path = "otlp_metrics/fixtures.rs"]
mod fixtures;
use std::collections::BTreeMap;
use std::hint::black_box;
use std::sync::Arc;
use std::time::Duration;
use api::helper::ColumnDataTypeWrapper;
use api::v1::{ColumnDataType, ColumnSchema, RowInsertRequests, SemanticType};
use catalog::RegisterTableRequest;
use catalog::memory::MemoryCatalogManager;
use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
use common_meta::cache::new_table_flownode_set_cache;
use common_meta::ddl::test_util::datanode_handler::NaiveDatanodeHandler;
use common_meta::kv_backend::memory::MemoryKvBackend;
use common_meta::test_util::MockDatanodeManager;
use common_query::prelude::greptime_value;
use criterion::{BatchSize, Criterion, Throughput, criterion_group, criterion_main};
use datatypes::schema::{ColumnSchema as DataColumn, Schema};
use moka::future::Cache;
use operator::insert::Inserter;
use operator::test_util::create_partition_rule_manager;
use servers::otlp::metrics::{prepare_metrics, to_grpc_insert_requests};
use session::context::QueryContext;
use session::protocol_ctx::OtlpMetricCtx;
use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, PHYSICAL_TABLE_METADATA_KEY};
use table::dist_table::DummyDataSource;
use table::metadata::FilterPushDownType;
use table::test_util::table_info::test_table_info;
fn register_table(catalog: &MemoryCatalogManager, id: u32, name: &str, columns: &[ColumnSchema]) {
let schema = Schema::new(
columns
.iter()
.map(|column| {
let timestamp = column.semantic_type == SemanticType::Timestamp as i32;
let required = name == "required_tag_1024" && column.column_name == "label_0";
DataColumn::new(
&column.column_name,
ColumnDataTypeWrapper::try_new(
column.datatype,
column.datatype_extension.clone(),
)
.unwrap()
.into(),
!timestamp && !required,
)
.with_time_index(timestamp)
})
.collect(),
);
let mut info = test_table_info(
id,
name,
DEFAULT_SCHEMA_NAME,
DEFAULT_CATALOG_NAME,
Arc::new(schema),
);
info.meta.primary_key_indices = columns
.iter()
.enumerate()
.filter_map(|(i, c)| (c.semantic_type == SemanticType::Tag as i32).then_some(i))
.collect();
info.meta.value_indices = columns
.iter()
.enumerate()
.filter_map(|(i, c)| (c.semantic_type == SemanticType::Field as i32).then_some(i))
.collect();
// Ordinary tables retain required tags; logical tables inherit nullable physical tags.
if name != "required_tag_1024" {
info.meta.engine = "metric".into();
let (key, value) = if name == "physical" {
(PHYSICAL_TABLE_METADATA_KEY, "true")
} else {
(LOGICAL_TABLE_METADATA_KEY, "physical")
};
info.meta
.options
.extra_options
.insert(key.into(), value.into());
}
catalog
.register_table_sync(RegisterTableRequest {
catalog: DEFAULT_CATALOG_NAME.into(),
schema: DEFAULT_SCHEMA_NAME.into(),
table_name: name.into(),
table_id: id,
table: Arc::new(table::Table::new(
Arc::new(info),
FilterPushDownType::Unsupported,
Arc::new(DummyDataSource),
)),
})
.unwrap();
}
async fn inserter(requests: &RowInsertRequests) -> Inserter {
let catalog = MemoryCatalogManager::with_default_setup();
let mut physical = BTreeMap::new();
physical.insert(
greptime_value().to_string(),
api::v1::helper::field_column_schema(greptime_value(), ColumnDataType::Float64),
);
for (index, request) in requests.inserts.iter().enumerate() {
let columns = &request.rows.as_ref().unwrap().schema;
register_table(&catalog, index as u32 + 1, &request.table_name, columns);
physical.extend(
columns
.iter()
.map(|column| (column.column_name.clone(), column.clone())),
);
}
register_table(
&catalog,
0,
"physical",
&physical.into_values().collect::<Vec<_>>(),
);
let backend = Arc::new(MemoryKvBackend::default());
Inserter::new(
catalog,
create_partition_rule_manager(backend.clone()).await,
Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)),
Arc::new(new_table_flownode_set_cache(
String::new(),
Cache::new(100),
backend,
)),
true,
)
}
fn bench_preflight(c: &mut Criterion) {
let runtime = tokio::runtime::Runtime::new().unwrap();
let ctx = QueryContext::arc();
let mut group = c.benchmark_group("otlp_preflight");
for fixture in fixtures::fixtures() {
let mut metric_ctx = OtlpMetricCtx {
with_metric_engine: true,
..Default::default()
};
let converted = to_grpc_insert_requests(fixture.request.clone(), &mut metric_ctx).unwrap();
assert_eq!(
converted.outcome.accepted_data_points,
fixture.points as i64 - fixture.rejected
);
let inserter = runtime.block_on(inserter(&converted.requests));
let prepared = prepare_metrics(fixture.request, &mut metric_ctx).unwrap();
group.throughput(Throughput::Elements(prepared.num_points() as u64));
group.bench_function(fixture.name, |b| {
let mut requests = prepared.requests.clone();
let result = runtime
.block_on(inserter.validate_and_normalize_otlp_metrics(
&mut requests,
&ctx,
true,
"physical",
))
.unwrap();
let rejections = prepared.point_rejections(result.rejections).unwrap();
assert_eq!(rejections.len(), prepared.num_points());
assert_eq!(
rejections.iter().filter(|r| r.is_some()).count(),
if fixture.name == "required_tag_1024" {
fixture.points.div_ceil(10) as usize
} else {
0
}
);
b.iter_batched_ref(
|| prepared.requests.clone(),
|requests| {
black_box(
runtime
.block_on(inserter.validate_and_normalize_otlp_metrics(
requests, &ctx, true, "physical",
))
.unwrap(),
)
},
BatchSize::LargeInput,
);
});
}
group.finish();
}
fn bench_ingestion(c: &mut Criterion) {
let runtime = tokio::runtime::Runtime::new().unwrap();
let ctx = QueryContext::arc();
let mut group = c.benchmark_group("otlp_ingestion");
for fixture in fixtures::fixtures() {
let metric_ctx = OtlpMetricCtx {
with_metric_engine: true,
..Default::default()
};
let converted =
to_grpc_insert_requests(fixture.request.clone(), &mut metric_ctx.clone()).unwrap();
let inserter = runtime.block_on(inserter(&converted.requests));
let convert = |request| {
let mut prepared = prepare_metrics(request, &mut metric_ctx.clone()).unwrap();
let validated = runtime
.block_on(inserter.validate_and_normalize_otlp_metrics(
&mut prepared.requests,
&ctx,
true,
"physical",
))
.unwrap();
let rejections = prepared.point_rejections(validated.rejections).unwrap();
prepared
.finish(rejections, &validated.physical_tables)
.unwrap()
};
let output = convert(fixture.request.clone());
let rejected = if fixture.name == "required_tag_1024" {
fixture.points.div_ceil(10) as i64
} else {
fixture.rejected
};
assert_eq!(output.outcome.rejected_data_points, rejected);
assert_eq!(
output.outcome.accepted_data_points,
fixture.points as i64 - rejected
);
group.throughput(Throughput::Elements(fixture.points));
group.bench_function(fixture.name, |b| {
b.iter_batched(
|| fixture.request.clone(),
|request| {
black_box(convert(request));
},
BatchSize::LargeInput,
);
});
}
group.finish();
}
criterion_group! {
name = benches;
config = Criterion::default().sample_size(20).warm_up_time(Duration::from_secs(1)).measurement_time(Duration::from_secs(3));
targets = bench_preflight, bench_ingestion
}
criterion_main!(benches);
src/servers/src/error.rs
+1 -10
View File
@@ -282,14 +282,6 @@ pub enum Error {
location: Location,
},
#[snafu(display(
"OTLP metric input have incompatible existing tables, please refer to docs for details"
))]
OtlpMetricModeIncompatible {
#[snafu(implicit)]
location: Location,
},
#[snafu(display("Common Meta error"))]
CommonMeta {
#[snafu(implicit)]
@@ -767,8 +759,7 @@ impl ErrorExt for Error {
AddressBind { .. }
| AlreadyStarted { .. }
| InvalidPromRemoteReadQueryResult { .. }
| OtlpMetricModeIncompatible { .. } => StatusCode::IllegalState,
| InvalidPromRemoteReadQueryResult { .. } => StatusCode::IllegalState,
UnsupportedDataType { .. } | UnsupportedOperation { .. } => StatusCode::Unsupported,
src/servers/src/http/otlp.rs
-1
View File
@@ -116,7 +116,6 @@ pub async fn metrics(
promote_scope_attrs: http_opts.promote_scope_attrs,
with_metric_engine,
// set by the frontend from its config
is_legacy: false,
resource_info: false,
metric_type: MetricType::Init,
metric_translation_strategy: http_opts.metric_translation_strategy,
src/servers/src/otlp/metrics.rs
+358 -483
View File
File diff suppressed because it is too large Load Diff
src/servers/src/otlp/metrics/prepare.rs
+623
View File
@@ -0,0 +1,623 @@
// 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 std::collections::{HashMap, HashSet};
use std::hash::{BuildHasher, Hash, Hasher};
use std::ops::Range;
use ahash::{HashMap as AHashMap, HashMapExt};
use api::v1::{ColumnSchema, RowInsertRequest, RowInsertRequests, Rows, SemanticType};
use common_base::regex_pattern::NAME_PATTERN_REG;
use common_query::native_histogram::native_histogram_value_type;
use otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest;
use otel_arrow_rust::proto::opentelemetry::common::v1::{KeyValue, any_value};
use otel_arrow_rust::proto::opentelemetry::metrics::v1::{Metric, NumberDataPoint, metric};
use session::protocol_ctx::OtlpMetricCtx;
use snafu::{OptionExt, ensure};
use table::requests::{
METADATA_QUALITY_DECLARED, SEMANTIC_METRIC_METADATA_QUALITY, SEMANTIC_METRIC_TYPE,
};
use crate::error::{self, Result};
use crate::metrics::OTLP_EXPONENTIAL_HISTOGRAM_REJECTED_DATA_POINTS;
use crate::otlp::metrics::{
MetricMetadata, MetricsConversion, OTEL_RESOURCE_INFO_TABLE_NAME, ResourceInfoData,
add_accepted_data_points, encode_metrics, from_metric_type, has_no_recorded_value,
process_resource_attrs, process_scope_attrs, reject_data_points, validate_sample_kind,
};
use crate::query_handler::MetricsIngestOutcome;
use crate::row_writer::{MultiTableData, TableData};
use crate::semantic::SemanticIndex;
/// Rows share exact input schemas while retaining their source-point boundaries.
pub struct PreparedMetrics {
/// Destination/schema batches. Normalization must preserve batch and row order/counts.
pub requests: Vec<RowInsertRequest>,
points: Vec<PreparedPoint>,
row_locations: Vec<(usize, usize)>,
metrics: Vec<MetricMetadata>,
resources: Vec<Vec<KeyValue>>,
outcome: MetricsIngestOutcome,
}
struct PreparedPoint {
metric: usize,
resource: usize,
timestamp: i64,
rows: Range<usize>,
}
#[derive(Default)]
struct MetricWriter {
tables: MultiTableData,
// TableData schemas only grow. A new column invalidates the cached batch.
batches: AHashMap<String, (usize, usize)>,
}
pub fn prepare_metrics(
request: ExportMetricsServiceRequest,
metric_ctx: &mut OtlpMetricCtx,
) -> Result<PreparedMetrics> {
let mut prepared = PreparedMetrics {
requests: Vec::new(),
points: Vec::new(),
row_locations: Vec::new(),
metrics: Vec::new(),
resources: Vec::new(),
outcome: MetricsIngestOutcome::default(),
};
let mut schema_batches = AHashMap::<u64, Vec<usize>>::new();
for resource in request.resource_metrics {
let raw_attrs = resource.resource.map(|r| r.attributes).unwrap_or_default();
let mut attrs = raw_attrs.clone();
process_resource_attrs(&mut attrs, metric_ctx);
let resource_index = prepared.resources.len();
prepared.resources.push(if metric_ctx.resource_info {
raw_attrs
} else {
Vec::new()
});
for scope in resource.scope_metrics {
let scope_attrs = process_scope_attrs(&scope, metric_ctx);
for mut metric in scope.metrics {
let Some(data) = metric.data.as_ref() else {
continue;
};
metric_ctx.set_metric_type(from_metric_type(data));
let metric_index = prepared.metrics.len();
prepared
.metrics
.push(MetricMetadata::new(&metric, metric_ctx)?);
let mut writer = MetricWriter::default();
let mut layout: (Vec<String>, bool) = (Vec::new(), false);
let mut writers = AHashMap::new();
// Move each payload once; never clone the accumulated request.
macro_rules! prepare_points {
($data:ident, $variant:ident, $omits_value:expr) => {
for point in std::mem::take(&mut $data.data_points) {
// Omissions must not become explicit nulls in a retained schema.
let omits_value = ($omits_value)(&point);
if omits_value != layout.1
|| !scalar_attribute_names(&point.attributes)
.eq(layout.0.iter().map(String::as_str))
{
let next_layout = (
scalar_attribute_names(&point.attributes)
.map(str::to_owned)
.collect(),
omits_value,
);
let next_writer = writers.remove(&next_layout).unwrap_or_default();
let previous = std::mem::replace(&mut writer, next_writer);
let previous_layout = std::mem::replace(&mut layout, next_layout);
if !previous.batches.is_empty() {
writers.insert(previous_layout, previous);
}
}
let timestamp = point.time_unix_nano;
let start = point.start_time_unix_nano;
$data.data_points.push(point);
metric.data = Some(metric::Data::$variant($data));
if !prepared.push_point(
&metric,
metric_index,
timestamp,
start,
&attrs,
scope_attrs.as_ref(),
resource_index,
metric_ctx,
&mut writer,
&mut schema_batches,
)? {
writer = MetricWriter::default();
}
let Some(metric::Data::$variant(mut remaining)) = metric.data.take()
else {
unreachable!()
};
remaining.data_points.clear();
$data = remaining;
}
};
}
let Some(data) = metric.data.take() else {
continue;
};
match data {
metric::Data::Gauge(mut data) => {
prepare_points!(data, Gauge, |point: &NumberDataPoint| point
.value
.is_none())
}
metric::Data::Sum(mut data) => {
prepare_points!(data, Sum, |point: &NumberDataPoint| point.value.is_none()
&& !has_no_recorded_value(point.flags))
}
metric::Data::Histogram(mut data) => {
prepare_points!(data, Histogram, |_| false)
}
metric::Data::Summary(mut data) => prepare_points!(data, Summary, |_| false),
metric::Data::ExponentialHistogram(mut data) => {
prepare_points!(data, ExponentialHistogram, |_| false)
}
}
}
}
}
Ok(prepared)
}
fn scalar_attribute_names(attrs: &[KeyValue]) -> impl Iterator<Item = &str> {
attrs.iter().filter_map(|attr| {
matches!(
attr.value.as_ref()?.value.as_ref()?,
any_value::Value::StringValue(_)
| any_value::Value::BoolValue(_)
| any_value::Value::IntValue(_)
| any_value::Value::DoubleValue(_)
)
.then_some(attr.key.as_str())
})
}
impl PreparedMetrics {
#[allow(clippy::too_many_arguments)]
fn push_point(
&mut self,
metric: &Metric,
metric_index: usize,
timestamp: u64,
start: u64,
resource_attrs: &Vec<KeyValue>,
scope_attrs: Option<&Vec<KeyValue>>,
resource: usize,
ctx: &OtlpMetricCtx,
writer: &mut MetricWriter,
schema_batches: &mut AHashMap<u64, Vec<usize>>,
) -> Result<bool> {
let metadata = &self.metrics[metric_index];
let exponential = metadata.native_schema.is_some();
// OTLP gauges ignore StartTimeUnixNano.
let invalid_interval =
!matches!(metric.data, Some(metric::Data::Gauge(_))) && start > timestamp;
let mut outcome = MetricsIngestOutcome::default();
let encoded = if !exponential && (timestamp > i64::MAX as u64 || invalid_interval) {
Err(error::InvalidOtlpMetricInputSnafu {
reason: "invalid OTLP data-point timestamp or interval",
}
.build())
} else {
encode_metrics(
&mut writer.tables,
metric,
metadata,
Some(resource_attrs),
scope_attrs,
ctx,
&mut outcome,
)
};
if let Err(err) = encoded {
match err {
error::Error::InvalidOtlpMetricInput { .. }
| error::Error::IncompatibleSchema { .. }
| error::Error::TimestampOverflow { .. }
| error::Error::TimePrecision { .. }
| error::Error::InvalidParameter { .. } => {
self.reject(exponential, err.to_string())?;
return Ok(false);
}
_ => return Err(err),
}
}
if outcome.rejected_data_points != 0 {
// Payload encoders already counted histogram-specific rejections.
reject_data_points(&mut self.outcome, 1, || {
outcome.error_message.unwrap_or_default()
})?;
return Ok(false);
}
if let Some((name, _)) = writer.tables.iter().find(|(name, table)| {
table.num_rows() != 0
&& !writer.batches.contains_key(*name)
&& !NAME_PATTERN_REG.is_match(name)
}) {
self.reject(exponential, format!("invalid output table name `{name}`"))?;
return Ok(false);
}
let first_row = self.row_locations.len();
for (name, schema, rows) in writer.tables.drain_rows() {
let batch = match writer.batches.get(name) {
Some(&(columns, batch)) if columns == schema.len() => batch,
_ => {
let mut hash = schema_batches.hasher().build_hasher();
name.hash(&mut hash);
for column in schema {
(&column.column_name, column.datatype, column.semantic_type)
.hash(&mut hash);
}
let candidates = schema_batches.entry(hash.finish()).or_default();
let batch = candidates.iter().copied().find(|&index| {
self.requests[index].table_name == name
&& self.requests[index]
.rows
.as_ref()
.is_some_and(|rows| rows.schema == schema)
});
let batch = if let Some(batch) = batch {
batch
} else {
validate_sample_kind(name, schema)?;
let batch = self.requests.len();
self.requests.push(RowInsertRequest {
table_name: name.to_owned(),
rows: Some(Rows {
schema: schema.to_vec(),
rows: Vec::new(),
}),
});
candidates.push(batch);
batch
};
writer
.batches
.insert(name.to_owned(), (schema.len(), batch));
batch
}
};
let output = self.requests[batch]
.rows
.as_mut()
.context(error::InternalSnafu {
err_msg: "missing prepared batch",
})?;
for row in rows {
self.row_locations.push((batch, output.rows.len()));
output.rows.push(row);
}
}
self.points.push(PreparedPoint {
metric: metric_index,
resource,
timestamp: timestamp as i64,
rows: first_row..self.row_locations.len(),
});
Ok(true)
}
fn reject(&mut self, exponential: bool, reason: String) -> Result<()> {
reject_data_points(&mut self.outcome, 1, || reason)?;
if exponential {
OTLP_EXPONENTIAL_HISTOGRAM_REJECTED_DATA_POINTS
.with_label_values(&["invalid_data_point"])
.inc();
}
Ok(())
}
/// Number of source points that survived payload conversion.
pub fn num_points(&self) -> usize {
self.points.len()
}
fn validate_batches(&self) -> Result<()> {
let mut counts = vec![0; self.requests.len()];
for &(batch, row) in &self.row_locations {
let count = counts.get_mut(batch).context(error::InternalSnafu {
err_msg: "missing prepared batch",
})?;
ensure!(
row == *count,
error::InternalSnafu {
err_msg: "invalid prepared row order",
}
);
*count += 1;
}
ensure!(
self.requests.iter().zip(counts).all(|(request, count)| {
count != 0
&& request
.rows
.as_ref()
.is_some_and(|rows| rows.rows.len() == count)
}),
error::InternalSnafu {
err_msg: "OTLP prepared batch row counts changed"
}
);
Ok(())
}
/// Rejects a source point if any row in its fan-out failed destination validation.
pub fn point_rejections(&self, rows: Vec<Vec<Option<String>>>) -> Result<Vec<Option<String>>> {
self.validate_batches()?;
ensure!(
rows.len() == self.requests.len()
&& rows.iter().zip(&self.requests).all(|(errors, request)| {
request
.rows
.as_ref()
.is_some_and(|rows| errors.len() == rows.rows.len())
}),
error::InternalSnafu {
err_msg: "OTLP batch rejections must match prepared rows"
}
);
Ok(self
.points
.iter()
.map(|point| {
self.row_locations[point.rows.clone()]
.iter()
.find_map(|&(batch, row)| rows[batch][row].clone())
})
.collect())
}
/// Complete authoritative validation before calling this method. Fresh
/// logical and physical conflicts are then collected from one survivor set.
pub fn finish(
mut self,
mut rejections: Vec<Option<String>>,
physical_tables: &HashMap<String, String>,
) -> Result<MetricsConversion> {
ensure!(
rejections.len() == self.points.len(),
error::InternalSnafu {
err_msg: "OTLP prepared points and rejections must have equal lengths",
}
);
self.validate_batches()?;
let conflicts = conflicting_points(
&self.requests,
&self.points,
&self.row_locations,
&rejections,
physical_tables,
);
for index in conflicts {
rejections[index] =
Some("incompatible metric sample kinds or column definitions".to_string());
}
let reuse_batches = rejections.iter().all(Option::is_none) && {
let mut tables = HashSet::new();
self.requests.len() <= 1 || self.requests.iter().all(|r| tables.insert(&r.table_name))
};
let mut outputs: Vec<(String, TableData)> = Vec::new();
let mut output_indices = AHashMap::new();
let mut mappings: Vec<Option<(usize, Vec<usize>, bool)>> = if reuse_batches {
Vec::new()
} else {
vec![None; self.requests.len()]
};
let mut semantic_index = SemanticIndex::default();
let mut resource_info = ResourceInfoData::default();
let mut resource_times = vec![Vec::new(); self.resources.len()];
for (point, rejection) in std::mem::take(&mut self.points).into_iter().zip(rejections) {
if let Some(reason) = rejection {
let metadata = &self.metrics[point.metric];
self.reject(
metadata.native_schema.is_some(),
format!("metric `{}`: {reason}", metadata.original_name),
)?;
continue;
}
for &(batch, row) in &self.row_locations[point.rows] {
let request = &mut self.requests[batch];
semantic_index.merge_table_from(
&mut self.metrics[point.metric].semantics,
&request.table_name,
);
if reuse_batches {
continue;
}
let rows = request.rows.as_mut().context(error::InternalSnafu {
err_msg: "missing prepared batch",
})?;
if mappings[batch].is_none() {
let output = *output_indices
.entry(request.table_name.clone())
.or_insert_with(|| {
let index = outputs.len();
outputs.push((
request.table_name.clone(),
TableData::new(rows.schema.len(), rows.rows.len()),
));
index
});
let columns = std::mem::take(&mut rows.schema)
.into_iter()
.map(|column| outputs[output].1.ensure_column(column))
.collect::<Result<Vec<_>>>()?;
let identity = columns
.iter()
.enumerate()
.all(|(index, &column)| index == column);
mappings[batch] = Some((output, columns, identity));
}
let (output, columns, identity) =
mappings[batch].as_ref().context(error::InternalSnafu {
err_msg: "missing output column mapping",
})?;
let table = &mut outputs[*output].1;
let values = std::mem::take(&mut rows.rows[row].values);
if *identity {
table.add_row(values);
} else {
let mut output = table.alloc_one_row();
for (value, &column) in values.into_iter().zip(columns) {
output[column] = value;
}
table.add_row(output);
}
}
add_accepted_data_points(&mut self.outcome, 1)?;
if !self.resources[point.resource].is_empty() {
resource_times[point.resource].push(point.timestamp);
}
}
for (attrs, times) in self.resources.iter().zip(resource_times) {
if !times.is_empty() {
resource_info.observe_times(attrs, times);
}
}
let requests = RowInsertRequests {
inserts: if reuse_batches {
std::mem::take(&mut self.requests)
} else {
outputs
.into_iter()
.map(|(table_name, table)| RowInsertRequest {
table_name,
rows: Some(table.into_rows()),
})
.collect()
},
};
let rows = requests
.inserts
.iter()
.filter_map(|r| r.rows.as_ref())
.map(|rows| rows.rows.len())
.sum();
let resource_info = if !resource_info.is_empty()
&& requests
.inserts
.iter()
.any(|r| r.table_name == OTEL_RESOURCE_INFO_TABLE_NAME)
{
common_telemetry::warn!(
"Skipping OTLP resource descriptor synthesis: a metric uses its table name"
);
None
} else {
resource_info.into_row_insert_requests()?
};
if resource_info.is_some() {
semantic_index.record_scalar(
OTEL_RESOURCE_INFO_TABLE_NAME,
SEMANTIC_METRIC_TYPE,
crate::semantic::METRIC_TYPE_INFO,
);
semantic_index.record_scalar(
OTEL_RESOURCE_INFO_TABLE_NAME,
SEMANTIC_METRIC_METADATA_QUALITY,
METADATA_QUALITY_DECLARED,
);
}
Ok(MetricsConversion {
requests,
rows,
semantic_index,
resource_info,
outcome: self.outcome,
})
}
}
fn conflicting_points(
requests: &[RowInsertRequest],
points: &[PreparedPoint],
row_locations: &[(usize, usize)],
rejections: &[Option<String>],
physical_tables: &HashMap<String, String>,
) -> HashSet<usize> {
let mut contributors = vec![Vec::new(); requests.len()];
for (index, point) in points
.iter()
.enumerate()
.filter(|(index, _)| rejections[*index].is_none())
{
for &(batch, _) in &row_locations[point.rows.clone()] {
if contributors[batch].last() != Some(&index) {
contributors[batch].push(index);
}
}
}
let mut columns = AHashMap::<_, Vec<(usize, &ColumnSchema)>>::new();
let mut kinds: AHashMap<&str, Vec<(usize, bool)>> = AHashMap::new();
for (batch, request) in requests
.iter()
.enumerate()
.filter(|(batch, _)| !contributors[*batch].is_empty())
{
let Some(rows) = &request.rows else { continue };
for column in &rows.schema {
columns
.entry((false, &request.table_name, &column.column_name))
.or_default()
.push((batch, column));
if let Some(physical) = physical_tables.get(&request.table_name) {
columns
.entry((true, physical, &column.column_name))
.or_default()
.push((batch, column));
}
if column.semantic_type == SemanticType::Field as i32 {
let native = api::helper::is_column_type_value_eq(
column.datatype,
column.datatype_extension.clone(),
native_histogram_value_type(),
);
kinds
.entry(&request.table_name)
.or_default()
.push((batch, native));
}
}
}
let mut rejected = HashSet::new();
for group in columns.values() {
let first = group[0].1;
if group.iter().any(|(_, column)| {
column.datatype != first.datatype
|| column.datatype_extension != first.datatype_extension
|| column.semantic_type != first.semantic_type
}) {
for (batch, _) in group {
rejected.extend(contributors[*batch].iter().copied());
}
}
}
for group in kinds.values() {
if group.iter().any(|(_, native)| *native) && group.iter().any(|(_, native)| !*native) {
for (batch, _) in group {
rejected.extend(contributors[*batch].iter().copied());
}
}
}
rejected
}
src/servers/src/otlp/metrics/resource_info.rs
+37 -96
View File
@@ -27,15 +27,10 @@ use common_catalog::consts::SEMANTIC_GRAPH_WINDOW_NANOS;
use common_grpc::precision::Precision;
use common_query::prelude::{greptime_timestamp, greptime_value};
use otel_arrow_rust::proto::opentelemetry::common::v1::KeyValue;
use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
AggregationTemporality, ResourceMetrics, metric,
};
use crate::error::Result;
use crate::otlp::metrics::{
INSTANCE_KEY, JOB_KEY, ServiceIdentity, exponential_histogram_gate,
exponential_histogram_value, histogram_data_point_rejection, scalar_value_string,
service_identity,
INSTANCE_KEY, JOB_KEY, ServiceIdentity, scalar_value_string, service_identity,
};
use crate::otlp::trace::{
KEY_CONTAINER_ID, KEY_CONTAINER_NAME, KEY_HOST_ID, KEY_HOST_NAME, KEY_K8S_CONTAINER_NAME,
@@ -88,8 +83,15 @@ pub struct ResourceInfoData {
}
impl ResourceInfoData {
/// Takes the raw attributes, before the promote filter runs on them.
pub fn observe(&mut self, raw_attrs: &[KeyValue], resource: &ResourceMetrics) {
pub(crate) fn is_empty(&self) -> bool {
self.rows.is_empty()
}
pub(super) fn observe_times(
&mut self,
raw_attrs: &[KeyValue],
times: impl IntoIterator<Item = i64>,
) {
let mut tags = Vec::with_capacity(MAX_PROJECTED_TAGS);
let ServiceIdentity { job, instance } = service_identity(raw_attrs);
if let Some(job) = job {
@@ -113,13 +115,13 @@ impl ResourceInfoData {
tags.sort_unstable();
let mut observed: BTreeMap<i64, i64> = BTreeMap::new();
for_each_encoded_time(resource, |ts| {
for ts in times {
let window = ts - ts.rem_euclid(SEMANTIC_GRAPH_WINDOW_NANOS);
observed
.entry(window)
.and_modify(|newest| *newest = (*newest).max(ts))
.or_insert(ts);
});
}
if observed.is_empty() {
return;
}
@@ -167,58 +169,6 @@ impl ResourceInfoData {
}
}
/// Visits the times of the data points the encoder writes rows for, so a
/// resource is described exactly where it is measured rather than wherever
/// its request happens to reach.
fn for_each_encoded_time(resource: &ResourceMetrics, mut visit: impl FnMut(i64)) {
fn visit_all(points: impl Iterator<Item = u64>, visit: &mut impl FnMut(i64)) {
for ts in points {
visit(ts as i64);
}
}
for scope in &resource.scope_metrics {
for m in &scope.metrics {
match &m.data {
Some(metric::Data::Gauge(g)) => {
visit_all(g.data_points.iter().map(|p| p.time_unix_nano), &mut visit)
}
Some(metric::Data::Sum(s)) => {
visit_all(s.data_points.iter().map(|p| p.time_unix_nano), &mut visit)
}
Some(metric::Data::Histogram(h)) => {
let is_delta = matches!(
AggregationTemporality::try_from(h.aggregation_temporality),
Ok(AggregationTemporality::Delta)
);
visit_all(
h.data_points
.iter()
.filter(|point| {
histogram_data_point_rejection(point, is_delta).is_none()
})
.map(|point| point.time_unix_nano),
&mut visit,
)
}
Some(metric::Data::Summary(s)) => {
visit_all(s.data_points.iter().map(|p| p.time_unix_nano), &mut visit)
}
Some(metric::Data::ExponentialHistogram(h))
if exponential_histogram_gate(h).is_ok() =>
{
for point in &h.data_points {
if let Ok((_, ts)) = exponential_histogram_value(point) {
visit(ts);
}
}
}
Some(metric::Data::ExponentialHistogram(_)) => {}
None => {}
}
}
}
}
#[cfg(test)]
mod tests {
use api::v1::SemanticType;
@@ -226,8 +176,8 @@ mod tests {
use common_query::prelude::set_default_prefix;
use otel_arrow_rust::proto::opentelemetry::common::v1::{AnyValue, any_value};
use otel_arrow_rust::proto::opentelemetry::metrics::v1::{
AggregationTemporality, ExponentialHistogram, ExponentialHistogramDataPoint, Gauge, Metric,
NumberDataPoint, ScopeMetrics,
AggregationTemporality, ExponentialHistogram, ExponentialHistogramDataPoint, Metric,
ResourceMetrics, ScopeMetrics, metric,
};
use super::*;
@@ -243,27 +193,6 @@ mod tests {
}
}
fn gauge_at(times: &[i64]) -> ResourceMetrics {
ResourceMetrics {
scope_metrics: vec![ScopeMetrics {
metrics: vec![Metric {
data: Some(metric::Data::Gauge(Gauge {
data_points: times
.iter()
.map(|ts| NumberDataPoint {
time_unix_nano: *ts as u64,
..Default::default()
})
.collect(),
})),
..Default::default()
}],
..Default::default()
}],
..Default::default()
}
}
#[test]
fn observe_projects_allowlist_and_dedups_per_request() {
let mut data = ResourceInfoData::default();
@@ -275,7 +204,7 @@ mod tests {
kv("k8s.node.name", "node-a"),
kv("os.type", "linux"),
];
data.observe(&attrs, &gauge_at(&[100, 50]));
data.observe_times(&attrs, [100, 50]);
assert_eq!(data.rows.len(), 1);
let (tags, windows) = data.rows.iter().next().unwrap();
assert_eq!(windows.values().copied().collect::<Vec<_>>(), vec![100]);
@@ -288,11 +217,11 @@ mod tests {
.all(|(k, _)| k != "os.type" && k != "service.instance.id")
);
data.observe(&[kv("host.id", "h-2")], &gauge_at(&[10]));
data.observe_times(&[kv("host.id", "h-2")], [10]);
assert_eq!(data.rows.len(), 2);
let mut empty = ResourceInfoData::default();
empty.observe(&[kv("os.type", "linux")], &gauge_at(&[100]));
empty.observe_times(&[kv("os.type", "linux")], [100]);
assert!(empty.into_row_insert_requests().unwrap().is_none());
}
@@ -301,9 +230,9 @@ mod tests {
fn observe_keeps_one_row_per_graph_window() {
let window = SEMANTIC_GRAPH_WINDOW_NANOS;
let mut data = ResourceInfoData::default();
data.observe(
data.observe_times(
&[kv("service.name", "api")],
&gauge_at(&[window + 1, window + 2, 3 * window + 7]),
[window + 1, window + 2, 3 * window + 7],
);
let windows = data.rows.values().next().unwrap();
@@ -320,6 +249,7 @@ mod tests {
let exponential = |temporality: AggregationTemporality| ResourceMetrics {
scope_metrics: vec![ScopeMetrics {
metrics: vec![Metric {
name: "latency".into(),
data: Some(metric::Data::ExponentialHistogram(ExponentialHistogram {
data_points: vec![ExponentialHistogramDataPoint {
time_unix_nano: 100,
@@ -337,10 +267,21 @@ mod tests {
AggregationTemporality::Delta,
AggregationTemporality::Unspecified,
] {
let resource = exponential(temporality);
let mut data = ResourceInfoData::default();
data.observe(&[kv("service.name", "api")], &resource);
assert!(data.into_row_insert_requests().unwrap().is_none());
let mut resource = exponential(temporality);
resource.resource = Some(
otel_arrow_rust::proto::opentelemetry::resource::v1::Resource {
attributes: vec![kv("service.name", "api")],
..Default::default()
},
);
let conversion = crate::otlp::metrics::to_grpc_insert_requests(
otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest { resource_metrics: vec![resource] },
&mut session::protocol_ctx::OtlpMetricCtx { resource_info: true, ..Default::default() },
).unwrap();
assert_eq!(
conversion.resource_info.is_none(),
temporality == AggregationTemporality::Unspecified
);
}
}
@@ -348,9 +289,9 @@ mod tests {
fn rows_carry_raw_key_tags_value_and_millis_timestamp() {
set_default_prefix(None).unwrap();
let mut data = ResourceInfoData::default();
data.observe(
data.observe_times(
&[kv("service.name", "api"), kv("host.id", "h-1")],
&gauge_at(&[1_700_000_000_123_456_789]),
[1_700_000_000_123_456_789],
);
let requests = data.into_row_insert_requests().unwrap().unwrap();
assert_eq!(requests.inserts.len(), 1);
src/servers/src/otlp/metrics/resource_info/tests/delta.rs
+35 -6
View File
@@ -22,8 +22,15 @@ use super::*;
fn observe_ignores_rejected_classic_histogram_windows() {
let window = SEMANTIC_GRAPH_WINDOW_NANOS;
let resource = ResourceMetrics {
resource: Some(
otel_arrow_rust::proto::opentelemetry::resource::v1::Resource {
attributes: vec![kv("service.name", "api")],
..Default::default()
},
),
scope_metrics: vec![ScopeMetrics {
metrics: vec![Metric {
name: "latency".into(),
data: Some(metric::Data::Histogram(Histogram {
data_points: vec![
HistogramDataPoint {
@@ -55,12 +62,34 @@ fn observe_ignores_rejected_classic_histogram_windows() {
}],
..Default::default()
};
let mut data = ResourceInfoData::default();
data.observe(&[kv("service.name", "api")], &resource);
let windows = data.rows.values().next().unwrap();
let conversion = crate::otlp::metrics::to_grpc_insert_requests(
otel_arrow_rust::proto::opentelemetry::collector::metrics::v1::ExportMetricsServiceRequest {
resource_metrics: vec![resource],
},
&mut session::protocol_ctx::OtlpMetricCtx { resource_info: true, ..Default::default() },
).unwrap();
assert_eq!(
vec![window + 1, 4 * window + 1],
windows.values().copied().collect::<Vec<_>>()
(
conversion.outcome.accepted_data_points,
conversion.outcome.rejected_data_points
),
(2, 1)
);
let descriptors = conversion.resource_info.unwrap();
let rows = descriptors.inserts[0].rows.as_ref().unwrap();
let timestamp = rows
.schema
.iter()
.position(|c| c.column_name == greptime_timestamp())
.unwrap();
assert_eq!(
vec![
Some(ValueData::TimestampMillisecondValue(window / 1_000_000)),
Some(ValueData::TimestampMillisecondValue(4 * window / 1_000_000))
],
rows.rows
.iter()
.map(|r| r.values[timestamp].value_data.clone())
.collect::<Vec<_>>()
);
}
src/servers/src/otlp/metrics/tests/delta.rs
+642 -3
View File
@@ -320,15 +320,16 @@ fn test_reserved_temporality_label_uses_final_persisted_key() {
})),
..Default::default()
};
let error = to_grpc_insert_requests(
let conversion = to_grpc_insert_requests(
metrics_request(vec![gauge(vec![keyvalue(
OTLP_AGGREGATION_TEMPORALITY_LABEL,
"user",
)])]),
&mut OtlpMetricCtx::default(),
)
.unwrap_err();
assert!(matches!(error, error::Error::InvalidOtlpMetricInput { .. }));
.unwrap();
assert_eq!(conversion.outcome.rejected_data_points, 1);
assert_eq!(conversion.outcome.accepted_data_points, 0);
let mut request = metrics_request(vec![gauge(vec![])]);
request.resource_metrics[0].scope_metrics[0].scope = Some(InstrumentationScope {
@@ -392,3 +393,641 @@ fn test_histogram_semantics_follow_only_emitted_rows() {
);
}
}
#[test]
fn test_shared_metric_metadata_follows_surviving_outputs() {
let mut metric = histogram_metric("latency");
let Some(metric::Data::Histogram(histogram)) = metric.data.as_mut() else {
unreachable!()
};
let with_sum = histogram.data_points[0].clone();
histogram.data_points[0].sum = None;
histogram.data_points.push(with_sum);
for (rejections, has_sum) in [
(vec![None, Some("rejected sum".into())], false),
(vec![Some("rejected first point".into()), None], true),
(vec![None, None], true),
] {
let prepared = prepare_metrics(
metrics_request(vec![metric.clone()]),
&mut OtlpMetricCtx::default(),
)
.unwrap();
let conversion = prepared.finish(rejections, &Default::default()).unwrap();
let semantics = decode(&conversion.semantic_index);
assert_eq!(semantics.contains_key("latency_sum"), has_sum);
for table in ["latency_count", "latency_bucket"] {
assert_eq!(
semantics[table][SEMANTIC_METRIC_TEMPORALITY],
METRIC_TEMPORALITY_CUMULATIVE
);
}
}
}
#[test]
fn test_delta_exponential_interval_encoding() {
use common_query::native_histogram::{COUNT_I64_FIELD, RESET_HINT_FIELD, SUM_FIELD};
use common_query::prelude::greptime_native_histogram;
for with_metric_engine in [false, true] {
let count = (1u64 << 53) + 1;
let point = ExponentialHistogramDataPoint {
start_time_unix_nano: 1_000_000_123,
time_unix_nano: 2_000_000_987,
count,
zero_count: count,
sum: Some(12.0),
zero_threshold: 0.1,
..Default::default()
};
let mut stale = point.clone();
stale.flags = DataPointFlags::NoRecordedValueMask as u32;
let conversion = to_grpc_insert_requests(
metrics_request(vec![exponential_metric(
"interval",
vec![point, stale],
AggregationTemporality::Delta,
)]),
&mut OtlpMetricCtx {
with_metric_engine,
..Default::default()
},
)
.unwrap();
assert_eq!(conversion.outcome.accepted_data_points, 2);
assert_eq!(conversion.outcome.rejected_data_points, 0);
let rows = table_rows(&conversion.requests, "interval");
let field = column_index(rows, greptime_native_histogram());
let tag = column_index(rows, OTLP_AGGREGATION_TEMPORALITY_LABEL);
let ts = column_index(rows, greptime_timestamp());
assert_eq!(
rows.schema[ts].datatype,
if with_metric_engine {
ColumnDataType::TimestampNanosecond
} else {
ColumnDataType::TimestampMillisecond
} as i32
);
for row in &rows.rows {
assert_eq!(
row.values[tag].value_data,
Some(ValueData::StringValue("delta".into()))
);
assert_eq!(
native_field(
row.values[field].value_data.as_ref().unwrap(),
RESET_HINT_FIELD
),
Some(ValueData::I32Value(ResetHint::Gauge as i32))
);
}
assert_eq!(
native_field(
rows.rows[0].values[field].value_data.as_ref().unwrap(),
COUNT_I64_FIELD
),
Some(ValueData::I64Value(count as i64))
);
let Some(ValueData::F64Value(stale)) = native_field(
rows.rows[1].values[field].value_data.as_ref().unwrap(),
SUM_FIELD,
) else {
panic!("missing sum")
};
assert_eq!(stale.to_bits(), PROMETHEUS_STALE_NAN_BITS);
}
}
fn gauge(name: &str) -> Metric {
Metric {
name: name.into(),
data: Some(metric::Data::Gauge(Gauge {
data_points: vec![NumberDataPoint {
value: Some(Value::AsDouble(7.0)),
..Default::default()
}],
})),
..Default::default()
}
}
#[test]
fn test_shared_schemas_preserve_source_order_and_omissions() {
let mut metric = gauge("ordered");
let Some(metric::Data::Gauge(data)) = &mut metric.data else {
unreachable!()
};
data.data_points = [
keyvalue("a", "first"),
keyvalue("b", "second"),
keyvalue("a", "third"),
KeyValue {
key: "a".into(),
value: None,
},
keyvalue("a", "fifth"),
keyvalue("a", "sixth"),
]
.into_iter()
.enumerate()
.map(|(index, attribute)| NumberDataPoint {
attributes: vec![attribute],
time_unix_nano: (index as u64 + 1) * 1_000_000,
value: (index != 5).then_some(Value::AsDouble(index as f64)),
..Default::default()
})
.collect();
for reject_first in [false, true] {
let prepared = prepare_metrics(
metrics_request(vec![metric.clone()]),
&mut OtlpMetricCtx::default(),
)
.unwrap();
assert_eq!(prepared.num_points(), 6);
// The fourth point omits a, even though its raw attribute key is unchanged.
assert_eq!(prepared.requests.len(), 4);
assert!(
prepared
.requests
.last()
.unwrap()
.rows
.as_ref()
.unwrap()
.schema
.iter()
.all(|column| column.column_name != greptime_value())
);
let mut rejections = vec![None; 6];
if reject_first {
rejections[0] = Some("rejected first point".into());
}
let conversion = prepared.finish(rejections, &Default::default()).unwrap();
let rows = table_rows(&conversion.requests, "ordered");
let a = column_index(rows, "a");
let b = column_index(rows, "b");
assert_eq!(b < a, reject_first);
let value = column_index(rows, greptime_value());
let timestamp = column_index(rows, greptime_timestamp());
for (index, row) in (usize::from(reject_first)..6).zip(&rows.rows) {
assert_eq!(
row.values[value].value_data,
(index != 5).then_some(ValueData::F64Value(index as f64))
);
assert_eq!(
row.values[timestamp].value_data,
Some(ValueData::TimestampMillisecondValue(index as i64 + 1))
);
assert_eq!(
row.values[a].value_data.is_some(),
matches!(index, 0 | 2 | 4 | 5)
);
assert_eq!(row.values[b].value_data.is_some(), index == 1);
}
assert_eq!(rows.rows.len(), 6 - usize::from(reject_first));
}
}
#[test]
fn test_reused_sum_layouts_preserve_omitted_and_stale_values() {
let points = [false, true, false, true, false]
.into_iter()
.enumerate()
.map(|(index, stale)| NumberDataPoint {
attributes: vec![keyvalue("host", "a")],
time_unix_nano: (index as u64 + 1) * 1_000_000,
flags: if stale {
DataPointFlags::NoRecordedValueMask as u32
} else {
0
},
..Default::default()
})
.collect();
let metric = Metric {
name: "requests".into(),
data: Some(metric::Data::Sum(Sum {
data_points: points,
aggregation_temporality: AggregationTemporality::Delta as i32,
..Default::default()
})),
..Default::default()
};
let prepared =
prepare_metrics(metrics_request(vec![metric]), &mut OtlpMetricCtx::default()).unwrap();
assert_eq!(prepared.requests.len(), 2);
for request in &prepared.requests {
let rows = request.rows.as_ref().unwrap();
let has_value = rows
.schema
.iter()
.any(|c| c.column_name == greptime_value());
assert_eq!(rows.rows.len(), if has_value { 2 } else { 3 });
assert!(
rows.rows
.iter()
.all(|row| row.values.len() == rows.schema.len())
);
}
let conversion = prepared.finish(vec![None; 5], &Default::default()).unwrap();
assert_eq!(conversion.outcome.accepted_data_points, 5);
let rows = table_rows(&conversion.requests, "requests");
let value = column_index(rows, greptime_value());
for (index, row) in rows.rows.iter().enumerate() {
if index % 2 == 0 {
assert!(row.values[value].value_data.is_none());
} else {
let Some(ValueData::F64Value(value)) = row.values[value].value_data else {
panic!("missing stale value")
};
assert_eq!(value.to_bits(), PROMETHEUS_STALE_NAN_BITS);
}
}
}
#[test]
fn test_unmerged_batches_preserve_normalized_rows_and_metadata() {
let mut first = gauge("same");
first.unit = "seconds".into();
let mut second = first.clone();
if let Some(metric::Data::Gauge(data)) = second.data.as_mut() {
data.data_points[0].value = Some(Value::AsDouble(8.0));
}
let mut prepared = prepare_metrics(
metrics_request(vec![first, gauge("other"), second]),
&mut OtlpMetricCtx::default(),
)
.unwrap();
assert_eq!(prepared.requests.len(), 2);
for request in &mut prepared.requests {
let rows = request.rows.as_mut().unwrap();
let timestamp = column_index(rows, greptime_timestamp());
rows.schema[timestamp].datatype = ColumnDataType::TimestampMicrosecond as i32;
for row in &mut rows.rows {
row.values[timestamp].value_data = Some(ValueData::TimestampMicrosecondValue(123));
}
}
let expected = prepared.requests.clone();
let conversion = prepared.finish(vec![None; 3], &Default::default()).unwrap();
assert_eq!(conversion.requests.inserts, expected);
assert_eq!(conversion.rows, 3);
assert_eq!(conversion.outcome.accepted_data_points, 3);
let semantics = decode(&conversion.semantic_index);
for request in &conversion.requests.inserts {
assert_eq!(
semantics[&request.table_name][SEMANTIC_METRIC_TYPE],
METRIC_TYPE_GAUGE
);
}
}
#[test]
fn test_shared_batch_row_error_rejects_complete_histogram_point() {
let mut metric = histogram_metric("latency");
let Some(metric::Data::Histogram(data)) = &mut metric.data else {
unreachable!()
};
data.data_points.push(data.data_points[0].clone());
let prepared =
prepare_metrics(metrics_request(vec![metric]), &mut OtlpMetricCtx::default()).unwrap();
let mut errors: Vec<_> = prepared
.requests
.iter()
.map(|request| vec![None; request.rows.as_ref().unwrap().rows.len()])
.collect();
let bucket = prepared
.requests
.iter()
.position(|r| r.table_name == "latency_bucket")
.unwrap();
let second_point = errors[bucket].len() / 2;
errors[bucket][second_point] = Some("invalid bucket".into());
assert!(prepared.point_rejections(Vec::new()).is_err());
let rejections = prepared.point_rejections(errors).unwrap();
assert_eq!(rejections, vec![None, Some("invalid bucket".into())]);
let conversion = prepared.finish(rejections, &Default::default()).unwrap();
assert_eq!(conversion.outcome.accepted_data_points, 1);
assert_eq!(conversion.outcome.rejected_data_points, 1);
for table in ["latency_count", "latency_sum", "latency_bucket"] {
assert_eq!(table_rows(&conversion.requests, table).rows.len(), 1);
}
}
#[test]
fn test_gauge_ignores_start_time_but_validates_timestamp() {
for with_metric_engine in [false, true] {
let mut ctx = OtlpMetricCtx {
with_metric_engine,
..Default::default()
};
let point = NumberDataPoint {
time_unix_nano: 60_000_000_000,
value: Some(Value::AsDouble(7.0)),
..Default::default()
};
let mut metric = gauge("gauge");
let Some(metric::Data::Gauge(data)) = &mut metric.data else {
unreachable!()
};
data.data_points = vec![point.clone()];
let baseline =
to_grpc_insert_requests(metrics_request(vec![metric.clone()]), &mut ctx).unwrap();
assert_eq!(baseline.outcome.accepted_data_points, 1);
for (start, timestamp, accepted) in [
(120_000_000_000, point.time_unix_nano, 1),
(u64::MAX, point.time_unix_nano, 1),
(0, i64::MAX as u64 + 1, 0),
] {
let Some(metric::Data::Gauge(data)) = &mut metric.data else {
unreachable!()
};
data.data_points[0].start_time_unix_nano = start;
data.data_points[0].time_unix_nano = timestamp;
let conversion =
to_grpc_insert_requests(metrics_request(vec![metric.clone()]), &mut ctx).unwrap();
assert_eq!(conversion.outcome.accepted_data_points, accepted);
assert_eq!(conversion.outcome.rejected_data_points, 1 - accepted);
if accepted == 1 {
assert_eq!(conversion.requests, baseline.requests);
} else {
assert!(conversion.requests.inserts.is_empty());
}
}
let sum = Metric {
name: "sum".into(),
data: Some(metric::Data::Sum(Sum {
data_points: vec![NumberDataPoint {
start_time_unix_nano: 120_000_000_000,
..point
}],
aggregation_temporality: AggregationTemporality::Delta as i32,
..Default::default()
})),
..Default::default()
};
let conversion = to_grpc_insert_requests(metrics_request(vec![sum]), &mut ctx).unwrap();
assert_eq!(conversion.outcome.accepted_data_points, 0);
assert_eq!(conversion.outcome.rejected_data_points, 1);
assert!(conversion.requests.inserts.is_empty());
}
}
#[test]
fn test_all_metric_families_isolate_invalid_attributes_and_fanout() {
let families = [
gauge("gauge"),
Metric {
name: "sum".into(),
data: Some(metric::Data::Sum(Sum {
data_points: vec![NumberDataPoint::default()],
..Default::default()
})),
..Default::default()
},
histogram_metric("histogram"),
Metric {
name: "summary".into(),
data: Some(metric::Data::Summary(Summary {
data_points: vec![SummaryDataPoint {
count: 1,
sum: 1.0,
..Default::default()
}],
})),
..Default::default()
},
exponential_metric(
"native",
vec![exponential_point()],
AggregationTemporality::Delta,
),
];
for family in families {
for invalid_first in [false, true] {
let mut metric = family.clone();
macro_rules! add_invalid {
($data:ident) => {{
let mut invalid = $data.data_points[0].clone();
invalid.attributes = vec![
keyvalue("rejected_only", "bad"),
keyvalue(OTLP_AGGREGATION_TEMPORALITY_LABEL, "bad"),
];
if invalid_first {
$data.data_points.insert(0, invalid);
} else {
$data.data_points.push(invalid);
}
}};
}
match metric.data.as_mut().unwrap() {
metric::Data::Gauge(data) => add_invalid!(data),
metric::Data::Sum(data) => add_invalid!(data),
metric::Data::Histogram(data) => add_invalid!(data),
metric::Data::Summary(data) => add_invalid!(data),
metric::Data::ExponentialHistogram(data) => add_invalid!(data),
}
let conversion = to_grpc_insert_requests(
metrics_request(vec![metric, gauge("independent")]),
&mut OtlpMetricCtx::default(),
)
.unwrap();
assert_eq!(
(
conversion.outcome.accepted_data_points,
conversion.outcome.rejected_data_points
),
(2, 1),
"{}",
family.name
);
assert!(conversion.requests.inserts.iter().all(|r| {
r.rows
.as_ref()
.unwrap()
.schema
.iter()
.all(|c| c.column_name != "rejected_only")
}));
}
}
}
#[test]
fn test_fresh_conflicts_are_simultaneous_and_authoritative_rejection_wins() {
let metrics = [
histogram_metric("latency"),
exponential_metric(
"latency_bucket",
vec![exponential_point()],
AggregationTemporality::Delta,
),
exponential_metric(
"latency_count",
vec![exponential_point()],
AggregationTemporality::Delta,
),
];
for order in [
[0, 1, 2],
[0, 2, 1],
[1, 0, 2],
[1, 2, 0],
[2, 0, 1],
[2, 1, 0],
] {
let mut input = order
.iter()
.map(|&i| metrics[i].clone())
.collect::<Vec<_>>();
input.push(gauge("independent"));
let conversion = to_grpc_insert_requests(
metrics_request(input.clone()),
&mut OtlpMetricCtx::default(),
)
.unwrap();
assert_eq!(
(
conversion.outcome.accepted_data_points,
conversion.outcome.rejected_data_points
),
(1, 3)
);
assert_eq!(conversion.requests.inserts.len(), 1);
assert_eq!(conversion.requests.inserts[0].table_name, "independent");
let prepared =
prepare_metrics(metrics_request(input), &mut OtlpMetricCtx::default()).unwrap();
let mut rejections = vec![None; 4];
rejections[order.iter().position(|i| *i == 0).unwrap()] =
Some("incompatible existing latency_sum".into());
let conversion = prepared
.finish(rejections, &std::collections::HashMap::new())
.unwrap();
assert_eq!(
(
conversion.outcome.accepted_data_points,
conversion.outcome.rejected_data_points
),
(3, 1)
);
assert!(!decode(&conversion.semantic_index).contains_key("latency_sum"));
}
}
#[test]
fn test_semantics_only_describe_actual_outputs() {
let mut histogram = histogram_metric("latency");
if let Some(metric::Data::Histogram(data)) = histogram.data.as_mut() {
data.data_points[0].sum = None;
}
let summary = Metric {
name: "summary".into(),
data: Some(metric::Data::Summary(Summary {
data_points: vec![SummaryDataPoint::default()],
})),
..Default::default()
};
for reverse in [false, true] {
let mut input = vec![
histogram.clone(),
gauge("latency_sum"),
summary.clone(),
gauge("summary"),
];
if reverse {
input.reverse();
}
let conversion =
to_grpc_insert_requests(metrics_request(input), &mut OtlpMetricCtx::default()).unwrap();
for name in ["latency_sum", "summary"] {
let semantics = decode(&conversion.semantic_index);
assert_eq!(semantics[name][SEMANTIC_METRIC_TYPE], METRIC_TYPE_GAUGE);
assert!(!semantics[name].contains_key(SEMANTIC_METRIC_TEMPORALITY));
}
}
}
#[test]
fn test_invalid_emitted_names_and_valid_empty_base_fanout() {
for reverse in [false, true] {
let mut metrics = vec![gauge("!!!"), histogram_metric("!!!"), gauge("valid")];
if reverse {
metrics.reverse();
}
let conversion =
to_grpc_insert_requests(metrics_request(metrics), &mut OtlpMetricCtx::default())
.unwrap();
assert_eq!(
(
conversion.outcome.accepted_data_points,
conversion.outcome.rejected_data_points
),
(2, 1)
);
assert!(
conversion
.requests
.inserts
.iter()
.all(|r| !r.table_name.is_empty())
);
assert!(
conversion
.requests
.inserts
.iter()
.any(|r| r.table_name == "_bucket")
);
}
}
#[test]
fn test_shared_physical_conflicts_and_independent_backings() {
use common_query::prelude::greptime_native_histogram;
for reverse in [false, true] {
let mut scalar = gauge("scalar");
if let Some(metric::Data::Gauge(data)) = scalar.data.as_mut() {
data.data_points[0]
.attributes
.push(keyvalue(greptime_native_histogram(), "tag"));
}
let mut metrics = vec![
scalar,
exponential_metric(
"native",
vec![exponential_point()],
AggregationTemporality::Delta,
),
gauge("independent"),
];
if reverse {
metrics.reverse();
}
for shared in [false, true] {
let prepared = prepare_metrics(
metrics_request(metrics.clone()),
&mut OtlpMetricCtx::default(),
)
.unwrap();
let backing = std::collections::HashMap::from([
("scalar".into(), "physical".into()),
(
"native".into(),
if shared { "physical" } else { "other" }.into(),
),
("independent".into(), "physical".into()),
]);
let conversion = prepared.finish(vec![None; 3], &backing).unwrap();
assert_eq!(
(
conversion.outcome.accepted_data_points,
conversion.outcome.rejected_data_points
),
if shared { (1, 2) } else { (3, 0) }
);
}
}
}
src/servers/src/otlp/metrics/translator.rs
-34
View File
@@ -286,16 +286,6 @@ pub fn normalize_label_name(name: &str) -> String {
n.to_string()
}
/// Normalize otlp instrumentation, metric and attribute names
///
/// <https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/api.md#instrument-name-syntax>
/// - since the name are case-insensitive, we transform them to lowercase for
/// better sql usability
/// - replace `.` and `-` with `_`
pub fn legacy_normalize_otlp_name(name: &str) -> String {
name.to_lowercase().replace(['.', '-'], "_")
}
#[cfg(test)]
mod tests {
use otel_arrow_rust::proto::opentelemetry::metrics::v1::Metric;
@@ -306,30 +296,6 @@ mod tests {
use super::*;
#[test]
fn test_legacy_normalize_otlp_name() {
assert_eq!(
legacy_normalize_otlp_name("jvm.memory.free"),
"jvm_memory_free"
);
assert_eq!(
legacy_normalize_otlp_name("jvm-memory-free"),
"jvm_memory_free"
);
assert_eq!(
legacy_normalize_otlp_name("jvm_memory_free"),
"jvm_memory_free"
);
assert_eq!(
legacy_normalize_otlp_name("JVM_MEMORY_FREE"),
"jvm_memory_free"
);
assert_eq!(
legacy_normalize_otlp_name("JVM_memory_FREE"),
"jvm_memory_free"
);
}
#[test]
fn test_translate_metric_name_strategies() {
let metric = Metric {
src/servers/src/row_writer.rs
+34 -9
View File
@@ -131,6 +131,16 @@ impl TableData {
(self.schema, self.rows)
}
pub(crate) fn into_rows(self) -> Rows {
let (schema, mut rows) = self.into_schema_and_rows();
for row in &mut rows {
if row.values.len() < schema.len() {
row.values.resize(schema.len(), Value::default());
}
}
Rows { schema, rows }
}
/// Writes a field value without enforcing that later writes use the same datatype
/// as the first-seen schema entry.
///
@@ -207,6 +217,29 @@ impl MultiTableData {
.insert(table_name.to_string(), table_data);
}
pub(crate) fn iter(&self) -> impl Iterator<Item = (&str, &TableData)> {
self.table_data_map
.iter()
.map(|(name, table)| (name.as_str(), table))
}
/// Retains schemas and row capacity for the next point with the same attributes.
pub(crate) fn drain_rows(
&mut self,
) -> impl Iterator<Item = (&str, &[ColumnSchema], std::vec::Drain<'_, Row>)> {
self.table_data_map.iter_mut().filter_map(|(name, table)| {
if table.rows.is_empty() {
return None;
}
for row in &mut table.rows {
if row.values.len() < table.schema.len() {
row.values.resize(table.schema.len(), Value::default());
}
}
Some((name.as_str(), table.schema.as_slice(), table.rows.drain(..)))
})
}
#[allow(dead_code)]
pub fn num_tables(&self) -> usize {
self.table_data_map.len()
@@ -220,17 +253,9 @@ impl MultiTableData {
.into_iter()
.map(|(table_name, table_data)| {
total_rows += table_data.num_rows();
let num_columns = table_data.num_columns();
let (schema, mut rows) = table_data.into_schema_and_rows();
for row in &mut rows {
if num_columns > row.values.len() {
row.values.resize(num_columns, Value { value_data: None });
}
}
RowInsertRequest {
table_name,
rows: Some(Rows { schema, rows }),
rows: Some(table_data.into_rows()),
}
})
.collect::<Vec<_>>();
src/servers/src/semantic.rs
+10 -1
View File
@@ -108,7 +108,7 @@ impl SemanticIndex {
serde_json::to_string(&BTreeMap::from([(schema, &self.tables)])).ok()
}
fn merge_from(&mut self, other: &SemanticIndex) {
pub(crate) fn merge_from(&mut self, other: &SemanticIndex) {
for (table, scalars) in &other.tables {
for (key, value) in scalars {
self.record_scalar(table, key, value);
@@ -116,6 +116,15 @@ impl SemanticIndex {
}
}
/// Moves a metric's metadata on its first accepted output for this table.
pub(crate) fn merge_table_from(&mut self, other: &mut SemanticIndex, table: &str) {
if let Some(scalars) = other.tables.remove(table) {
for (key, value) in scalars {
self.record_scalar(table, key, &value);
}
}
}
#[cfg(test)]
fn options_of(&self, table: &str) -> Option<&BTreeMap<&'static str, String>> {
self.tables.get(table)
src/session/src/protocol_ctx.rs
+2 -6
View File
@@ -42,18 +42,14 @@ impl ProtocolCtx {
/// If true, all scope attributes will be promoted to the final table schema.
/// Along with the scope name, scope version and scope schema URL.
/// - `with_metric_engine`
/// - `is_legacy`
/// If the user uses OTLP metrics ingestion before v0.16, it uses the old path.
/// So we call this path 'legacy'.
/// After v0.16, we store the OTLP metrics using prometheus compatible format, the new path.
/// The difference is how we convert the input data into the final table schema.
/// Use Metric Engine for auto-created tables and retain timestamp precision
/// until alignment with the destination table's time index.
#[derive(Debug, Clone, Default)]
pub struct OtlpMetricCtx {
pub promote_all_resource_attrs: bool,
pub resource_attrs: HashSet<String>,
pub promote_scope_attrs: bool,
pub with_metric_engine: bool,
pub is_legacy: bool,
/// Set from the server's `otlp.experimental_enable_resource_info`; off
/// means the resource descriptor is not synthesized at all.
pub resource_info: bool,
src/table/src/requests.rs
+1
View File
@@ -73,6 +73,7 @@ pub fn is_trace_table(table_info: &crate::metadata::TableInfo) -> bool {
)
}
/// Retained table metadata; OTLP metrics always use Prometheus-compatible conversion.
pub const OTLP_METRIC_COMPAT_KEY: &str = "otlp_metric_compat";
pub const OTLP_METRIC_COMPAT_PROM: &str = "prom";