From b462d5d19e2edf11246b93fefca2ccdb22e7cb14 Mon Sep 17 00:00:00 2001 From: shuiyisong <113876041+shuiyisong@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:35:28 +0800 Subject: [PATCH] fix: honor default prefix for all metric columns (#8640) * fix: honor default prefix for metric columns Signed-off-by: shuiyisong * fix: cr issue Signed-off-by: shuiyisong --------- Signed-off-by: shuiyisong --- config/config.md | 6 +- config/datanode.example.toml | 3 +- config/frontend.example.toml | 3 +- config/standalone.example.toml | 3 +- src/common/query/src/native_histogram.rs | 4 +- src/common/query/src/prelude.rs | 20 ++++ src/metric-engine/src/data_region.rs | 7 +- src/metric-engine/src/engine/alter.rs | 7 +- .../src/engine/alter/extract_new_columns.rs | 5 +- src/metric-engine/src/engine/create.rs | 13 +-- .../src/engine/create/extract_new_columns.rs | 7 +- src/metric-engine/src/engine/put.rs | 7 +- src/mito2/src/sst.rs | 98 +++++-------------- src/operator/src/insert.rs | 45 +++++++-- src/servers/src/otlp/metrics.rs | 42 ++++++++ src/servers/src/prom_remote_write/README.md | 6 +- src/servers/src/prom_remote_write/v2.rs | 34 +++++-- src/servers/tests/http/prom_store_test.rs | 7 +- .../tests/prom_remote_write_v2_test.rs | 8 +- 19 files changed, 189 insertions(+), 136 deletions(-) diff --git a/config/config.md b/config/config.md index 8a31fc0eb2..af4e5f61f8 100644 --- a/config/config.md +++ b/config/config.md @@ -13,7 +13,7 @@ | Key | Type | Default | Descriptions | | --- | -----| ------- | ----------- | | `default_timezone` | String | Unset | The default timezone of the server. | -| `default_column_prefix` | String | Unset | The default column prefix for auto-created time index and value columns. | +| `default_column_prefix` | String | Unset | The default column prefix for auto-created time index, value, and native histogram columns.
Legacy OTLP summary columns keep their historical `greptime_` prefix. | | `auto_create_table` | Bool | `true` | Server-side global switch for auto table creation on write.
When `false`, a missing table is never auto-created even if the request sets the `auto_create_table` hint to `true`. Default: `true`. | | `user_provider` | String | Unset | The user provider for authentication.
Examples: "static_user_provider:file:/path/to/users", "static_user_provider:cmd:greptime_user=greptime_pwd"
Password verifier formats: "plain:", "pbkdf2_sha256:::",
"mysql_native_password:",
"pg_scram_sha256::::"
"pbkdf2_sha256" and "pg_scram_sha256" protect passwords at rest, but cannot authenticate over MySQL's
native password handshake; a MySQL client must send the password in cleartext for such users.
"mysql_native_password" is MySQL-specific and cannot authenticate over PostgreSQL at all.
PostgreSQL SCRAM only covers "plain" and "pg_scram_sha256" users; if any user is "pbkdf2_sha256" or
"mysql_native_password", PostgreSQL falls back to cleartext password auth for every user.
For "pg_scram_sha256" users, keep the default iteration count (4096) and salt length (16): both are
observable in the SCRAM server-first message, and non-default values weaken resistance to username
enumeration. | | `max_in_flight_write_bytes` | String | Unset | Maximum total memory for all concurrent write request bodies and messages (HTTP, gRPC, Flight).
Set to 0 to disable the limit. Default: "0" (unlimited) | @@ -238,7 +238,7 @@ | Key | Type | Default | Descriptions | | --- | -----| ------- | ----------- | | `default_timezone` | String | Unset | The default timezone of the server. | -| `default_column_prefix` | String | Unset | The default column prefix for auto-created time index and value columns. | +| `default_column_prefix` | String | Unset | The default column prefix for auto-created time index, value, and native histogram columns.
Legacy OTLP summary columns keep their historical `greptime_` prefix. | | `auto_create_table` | Bool | `true` | Server-side global switch for auto table creation on write.
When `false`, a missing table is never auto-created even if the request sets the `auto_create_table` hint to `true`. Default: `true`. | | `user_provider` | String | Unset | The user provider for authentication.
Examples: "static_user_provider:file:/path/to/users", "static_user_provider:cmd:greptime_user=greptime_pwd"
Password verifier formats: "plain:", "pbkdf2_sha256:::",
"mysql_native_password:",
"pg_scram_sha256::::"
"pbkdf2_sha256" and "pg_scram_sha256" protect passwords at rest, but cannot authenticate over MySQL's
native password handshake; a MySQL client must send the password in cleartext for such users.
"mysql_native_password" is MySQL-specific and cannot authenticate over PostgreSQL at all.
PostgreSQL SCRAM only covers "plain" and "pg_scram_sha256" users; if any user is "pbkdf2_sha256" or
"mysql_native_password", PostgreSQL falls back to cleartext password auth for every user.
For "pg_scram_sha256" users, keep the default iteration count (4096) and salt length (16): both are
observable in the SCRAM server-first message, and non-default values weaken resistance to username
enumeration. | | `max_in_flight_write_bytes` | String | Unset | Maximum total memory for all concurrent write request bodies and messages (HTTP, gRPC, Flight).
Set to 0 to disable the limit. Default: "0" (unlimited) | @@ -465,7 +465,7 @@ | Key | Type | Default | Descriptions | | --- | -----| ------- | ----------- | | `node_id` | Integer | Unset | The datanode identifier and should be unique in the cluster. | -| `default_column_prefix` | String | Unset | The default column prefix for auto-created time index and value columns. | +| `default_column_prefix` | String | Unset | The default column prefix for auto-created time index, value, and native histogram columns.
Legacy OTLP summary columns keep their historical `greptime_` prefix. | | `require_lease_before_startup` | Bool | `false` | Start services after regions have obtained leases.
It will block the datanode start if it can't receive leases in the heartbeat from metasrv. | | `init_regions_in_background` | Bool | `false` | Initialize all regions in the background during the startup.
By default, it provides services after all regions have been initialized. | | `init_regions_parallelism` | Integer | `16` | Parallelism of initializing regions. | diff --git a/config/datanode.example.toml b/config/datanode.example.toml index 9646936f00..9b5fc2e9a4 100644 --- a/config/datanode.example.toml +++ b/config/datanode.example.toml @@ -2,7 +2,8 @@ ## @toml2docs:none-default node_id = 42 -## The default column prefix for auto-created time index and value columns. +## The default column prefix for auto-created time index, value, and native histogram columns. +## Legacy OTLP summary columns keep their historical `greptime_` prefix. ## @toml2docs:none-default default_column_prefix = "greptime" diff --git a/config/frontend.example.toml b/config/frontend.example.toml index a0dffcfa97..5d3cc97bdf 100644 --- a/config/frontend.example.toml +++ b/config/frontend.example.toml @@ -2,7 +2,8 @@ ## @toml2docs:none-default default_timezone = "UTC" -## The default column prefix for auto-created time index and value columns. +## The default column prefix for auto-created time index, value, and native histogram columns. +## Legacy OTLP summary columns keep their historical `greptime_` prefix. ## @toml2docs:none-default default_column_prefix = "greptime" diff --git a/config/standalone.example.toml b/config/standalone.example.toml index 3f141c8ad6..1758fd6086 100644 --- a/config/standalone.example.toml +++ b/config/standalone.example.toml @@ -2,7 +2,8 @@ ## @toml2docs:none-default default_timezone = "UTC" -## The default column prefix for auto-created time index and value columns. +## The default column prefix for auto-created time index, value, and native histogram columns. +## Legacy OTLP summary columns keep their historical `greptime_` prefix. ## @toml2docs:none-default default_column_prefix = "greptime" diff --git a/src/common/query/src/native_histogram.rs b/src/common/query/src/native_histogram.rs index ca77326ed9..9d07db19c3 100644 --- a/src/common/query/src/native_histogram.rs +++ b/src/common/query/src/native_histogram.rs @@ -106,8 +106,8 @@ pub fn native_histogram_value_type() -> &'static ConcreteDataType { &NATIVE_HISTOGRAM_VALUE_TYPE } -pub fn is_native_histogram_value_schema(name: &str, data_type: &ConcreteDataType) -> bool { - name == NATIVE_HISTOGRAM_FIELD && data_type == native_histogram_value_type() +pub fn is_native_histogram_value_type(data_type: &ConcreteDataType) -> bool { + data_type == native_histogram_value_type() } // --------------------------------------------------------------------------- diff --git a/src/common/query/src/prelude.rs b/src/common/query/src/prelude.rs index 50668bbbb1..41e5d895f0 100644 --- a/src/common/query/src/prelude.rs +++ b/src/common/query/src/prelude.rs @@ -19,6 +19,7 @@ use snafu::ensure; pub use crate::columnar_value::ColumnarValue; use crate::error::{InvalidColumnPrefixSnafu, Result}; +use crate::native_histogram::NATIVE_HISTOGRAM_FIELD; /// Default time index column name. static GREPTIME_TIMESTAMP_CELL: OnceCell = OnceCell::new(); @@ -26,6 +27,9 @@ static GREPTIME_TIMESTAMP_CELL: OnceCell = OnceCell::new(); /// Default value column name. static GREPTIME_VALUE_CELL: OnceCell = OnceCell::new(); +/// Default native histogram column name. +static GREPTIME_NATIVE_HISTOGRAM_CELL: OnceCell = OnceCell::new(); + pub fn set_default_prefix(prefix: Option<&str>) -> Result<()> { // Strip surrounding double quotes as a defensive measure against upstream // sources (scripts, CI, template engines, incorrect shell escaping) that may @@ -41,11 +45,13 @@ pub fn set_default_prefix(prefix: Option<&str>) -> Result<()> { // use default greptime prefix GREPTIME_TIMESTAMP_CELL.get_or_init(|| GREPTIME_TIMESTAMP.to_string()); GREPTIME_VALUE_CELL.get_or_init(|| GREPTIME_VALUE.to_string()); + GREPTIME_NATIVE_HISTOGRAM_CELL.get_or_init(|| NATIVE_HISTOGRAM_FIELD.to_string()); } Some(s) if s.trim().is_empty() => { // use "" to disable prefix GREPTIME_TIMESTAMP_CELL.get_or_init(|| "timestamp".to_string()); GREPTIME_VALUE_CELL.get_or_init(|| "value".to_string()); + GREPTIME_NATIVE_HISTOGRAM_CELL.get_or_init(|| "native_histogram".to_string()); } Some(x) => { ensure!( @@ -54,6 +60,7 @@ pub fn set_default_prefix(prefix: Option<&str>) -> Result<()> { ); GREPTIME_TIMESTAMP_CELL.get_or_init(|| format!("{}_timestamp", x)); GREPTIME_VALUE_CELL.get_or_init(|| format!("{}_value", x)); + GREPTIME_NATIVE_HISTOGRAM_CELL.get_or_init(|| format!("{}_native_histogram", x)); } } Ok(()) @@ -71,6 +78,15 @@ pub fn greptime_value() -> &'static str { GREPTIME_VALUE_CELL.get_or_init(|| GREPTIME_VALUE.to_string()) } +/// Get the default native histogram column name. +/// Returns the configured value, or `greptime_native_histogram` if not set. +#[inline] +pub fn greptime_native_histogram() -> &'static str { + GREPTIME_NATIVE_HISTOGRAM_CELL + .get() + .map_or(NATIVE_HISTOGRAM_FIELD, String::as_str) +} + /// Default timestamp column name constant for backward compatibility. const GREPTIME_TIMESTAMP: &str = "greptime_timestamp"; /// Default value column name constant for backward compatibility. @@ -92,6 +108,7 @@ mod tests { set_default_prefix(None).unwrap(); assert_eq!(greptime_timestamp(), "greptime_timestamp"); assert_eq!(greptime_value(), "greptime_value"); + assert_eq!(greptime_native_histogram(), "greptime_native_histogram"); } #[test] @@ -99,6 +116,7 @@ mod tests { set_default_prefix(Some("")).unwrap(); assert_eq!(greptime_timestamp(), "timestamp"); assert_eq!(greptime_value(), "value"); + assert_eq!(greptime_native_histogram(), "native_histogram"); } #[test] @@ -107,6 +125,7 @@ mod tests { set_default_prefix(Some("\"\"")).unwrap(); assert_eq!(greptime_timestamp(), "timestamp"); assert_eq!(greptime_value(), "value"); + assert_eq!(greptime_native_histogram(), "native_histogram"); } #[test] @@ -114,6 +133,7 @@ mod tests { set_default_prefix(Some("mydb")).unwrap(); assert_eq!(greptime_timestamp(), "mydb_timestamp"); assert_eq!(greptime_value(), "mydb_value"); + assert_eq!(greptime_native_histogram(), "mydb_native_histogram"); } #[test] diff --git a/src/metric-engine/src/data_region.rs b/src/metric-engine/src/data_region.rs index ec9ff15c8d..55b9eaaf68 100644 --- a/src/metric-engine/src/data_region.rs +++ b/src/metric-engine/src/data_region.rs @@ -13,7 +13,7 @@ // limitations under the License. use api::v1::SemanticType; -use common_query::native_histogram::is_native_histogram_value_schema; +use common_query::native_histogram::is_native_histogram_value_type; use common_telemetry::{debug, info}; use datatypes::schema::{SkippingIndexOptions, SkippingIndexType}; use mito2::engine::MitoEngine; @@ -138,10 +138,7 @@ impl DataRegion { // table for native histograms; ordinary metric fields are // created with the logical table. SemanticType::Field - if is_native_histogram_value_schema( - &c.column_schema.name, - &c.column_schema.data_type, - ) => {} + if is_native_histogram_value_type(&c.column_schema.data_type) => {} _ => { return AddingFieldColumnSnafu { name: &c.column_schema.name, diff --git a/src/metric-engine/src/engine/alter.rs b/src/metric-engine/src/engine/alter.rs index e9d8e878df..517d66ae86 100644 --- a/src/metric-engine/src/engine/alter.rs +++ b/src/metric-engine/src/engine/alter.rs @@ -18,7 +18,7 @@ mod validate; use std::collections::{BTreeSet, HashMap, HashSet}; use api::v1::SemanticType; -use common_query::native_histogram::is_native_histogram_value_schema; +use common_query::native_histogram::is_native_histogram_value_type; use extract_new_columns::extract_new_columns; use snafu::{OptionExt, ResultExt, ensure}; use store_api::metadata::ColumnMetadata; @@ -246,10 +246,7 @@ impl MetricEngineInner { ensure!( fields.len() == 1 - && is_native_histogram_value_schema( - &fields[0].column_schema.name, - &fields[0].column_schema.data_type - ), + && is_native_histogram_value_type(&fields[0].column_schema.data_type), AddingFieldColumnSnafu { name: first_added_field.column_metadata.column_schema.name.clone(), } diff --git a/src/metric-engine/src/engine/alter/extract_new_columns.rs b/src/metric-engine/src/engine/alter/extract_new_columns.rs index 3573762d74..36c7dbf416 100644 --- a/src/metric-engine/src/engine/alter/extract_new_columns.rs +++ b/src/metric-engine/src/engine/alter/extract_new_columns.rs @@ -15,7 +15,7 @@ use std::collections::{HashMap, HashSet}; use api::v1::SemanticType; -use common_query::native_histogram::is_native_histogram_value_schema; +use common_query::native_histogram::is_native_histogram_value_type; use snafu::ensure; use store_api::metadata::ColumnMetadata; use store_api::region_request::{AlterKind, RegionAlterRequest}; @@ -45,8 +45,7 @@ pub fn extract_new_columns<'a>( { ensure!( col.column_metadata.semantic_type != SemanticType::Field - || is_native_histogram_value_schema( - &col.column_metadata.column_schema.name, + || is_native_histogram_value_type( &col.column_metadata.column_schema.data_type ), AddingFieldColumnSnafu { diff --git a/src/metric-engine/src/engine/create.rs b/src/metric-engine/src/engine/create.rs index a9b5e16bab..952643cd7a 100644 --- a/src/metric-engine/src/engine/create.rs +++ b/src/metric-engine/src/engine/create.rs @@ -17,7 +17,7 @@ mod extract_new_columns; use std::collections::{HashMap, HashSet}; use api::v1::SemanticType; -use common_query::native_histogram::is_native_histogram_value_schema; +use common_query::native_histogram::is_native_histogram_value_type; use common_telemetry::info; use common_time::{FOREVER, Timestamp}; use datatypes::data_type::ConcreteDataType; @@ -423,10 +423,7 @@ impl MetricEngineInner { .fail(); }; - if is_native_histogram_value_schema( - &field_col.column_schema.name, - &field_col.column_schema.data_type, - ) { + if is_native_histogram_value_type(&field_col.column_schema.data_type) { return Ok(()); } @@ -660,8 +657,8 @@ pub(crate) fn region_options_for_metadata_region( mod test { use common_meta::ddl::test_util::assert_column_name_and_id; use common_meta::ddl::utils::{parse_column_metadatas, parse_manifest_infos_from_extensions}; - use common_query::native_histogram::{NATIVE_HISTOGRAM_FIELD, native_histogram_value_type}; - use common_query::prelude::{greptime_timestamp, greptime_value}; + use common_query::native_histogram::native_histogram_value_type; + use common_query::prelude::{greptime_native_histogram, greptime_timestamp, greptime_value}; use store_api::metric_engine_consts::{METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY}; use store_api::region_request::{BatchRegionDdlRequest, RegionRequirements}; @@ -874,7 +871,7 @@ mod test { column_id: 2, semantic_type: SemanticType::Field, column_schema: ColumnSchema::new( - NATIVE_HISTOGRAM_FIELD, + greptime_native_histogram(), native_histogram_value_type().clone(), true, ), diff --git a/src/metric-engine/src/engine/create/extract_new_columns.rs b/src/metric-engine/src/engine/create/extract_new_columns.rs index f0d3af8748..add29916ca 100644 --- a/src/metric-engine/src/engine/create/extract_new_columns.rs +++ b/src/metric-engine/src/engine/create/extract_new_columns.rs @@ -15,7 +15,7 @@ use std::collections::{HashMap, HashSet}; use api::v1::SemanticType; -use common_query::native_histogram::is_native_histogram_value_schema; +use common_query::native_histogram::is_native_histogram_value_type; use snafu::ensure; use store_api::metadata::ColumnMetadata; use store_api::region_request::RegionCreateRequest; @@ -37,10 +37,7 @@ pub fn extract_new_columns<'a>( { ensure!( col.semantic_type != SemanticType::Field - || is_native_histogram_value_schema( - &col.column_schema.name, - &col.column_schema.data_type - ), + || is_native_histogram_value_type(&col.column_schema.data_type), AddingFieldColumnSnafu { name: col.column_schema.name.clone(), } diff --git a/src/metric-engine/src/engine/put.rs b/src/metric-engine/src/engine/put.rs index 21865ecdbb..96ed2c3e60 100644 --- a/src/metric-engine/src/engine/put.rs +++ b/src/metric-engine/src/engine/put.rs @@ -767,8 +767,7 @@ mod tests { use common_error::ext::ErrorExt; use common_error::status_code::StatusCode; use common_function::utils::partition_expr_version; - use common_query::native_histogram::NATIVE_HISTOGRAM_FIELD; - use common_query::prelude::{greptime_timestamp, greptime_value}; + use common_query::prelude::{greptime_native_histogram, greptime_timestamp, greptime_value}; use common_recordbatch::RecordBatches; use datatypes::prelude::ConcreteDataType; use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema}; @@ -1014,7 +1013,7 @@ mod tests { options: None, }; let histogram = PbColumnSchema { - column_name: NATIVE_HISTOGRAM_FIELD.to_string(), + column_name: greptime_native_histogram().to_string(), datatype: ColumnDataType::Struct as i32, semantic_type: SemanticType::Field as _, datatype_extension: None, @@ -1066,7 +1065,7 @@ mod tests { }; let value_idx = column_index(&merged_request, greptime_value()); - let histogram_idx = column_index(&merged_request, NATIVE_HISTOGRAM_FIELD); + let histogram_idx = column_index(&merged_request, greptime_native_histogram()); assert!(matches!( merged_request.rows[0].values[value_idx].value_data, Some(ValueData::F64Value(_)) diff --git a/src/mito2/src/sst.rs b/src/mito2/src/sst.rs index 170f25a2b4..a1f6e3e7f8 100644 --- a/src/mito2/src/sst.rs +++ b/src/mito2/src/sst.rs @@ -22,8 +22,7 @@ use arrow_schema::DataType; use arrow_schema::extension::{EXTENSION_TYPE_NAME_KEY, ExtensionType}; use common_base::readable_size::ReadableSize; use common_query::native_histogram::{ - is_native_histogram_value_schema, native_histogram_list_element_id, - native_histogram_subfield_id, + is_native_histogram_value_type, native_histogram_list_element_id, native_histogram_subfield_id, }; use datatypes::arrow::datatypes::{ DataType as ArrowDataType, Field, FieldRef, Fields, Schema, SchemaRef, @@ -87,10 +86,10 @@ pub fn with_field_id(mut field: Field, column_id: u32) -> Field { /// fields), so external readers can identify it by extension and resolve /// nested fields by id. /// -/// Detection is by the native-histogram column name and struct type -/// (`is_native_histogram_value_schema`); other struct columns are left -/// untouched. mito2 reads SST columns by schema position, never by field -/// metadata, so this only affects external readers. +/// Detection is by the exact native-histogram struct type +/// (`is_native_histogram_value_type`); other struct columns are left untouched. +/// mito2 reads SST columns by schema position, never by field metadata, so this +/// only affects external readers. /// /// Returns an error if the parent column's `PARQUET:field_id` is missing, /// malformed, or exceeds `i32::MAX`, or if a sub-field id cannot be derived @@ -98,10 +97,7 @@ pub fn with_field_id(mut field: Field, column_id: u32) -> Field { /// the derived id overflows a positive `i32` (an absurdly large parent /// `column_id`); see [`native_histogram_subfield_id`]. fn stamp_native_histogram_subfield_ids(field: &mut Field) -> crate::error::Result<()> { - if !is_native_histogram_value_schema( - field.name(), - &ConcreteDataType::from_arrow_type(field.data_type()), - ) { + if !is_native_histogram_value_type(&ConcreteDataType::from_arrow_type(field.data_type())) { return Ok(()); } // Namespace sub-field ids by the parent column's field id (its @@ -509,6 +505,7 @@ impl SeriesEstimator { mod tests { use std::sync::Arc; + use common_query::prelude::greptime_native_histogram; use datatypes::arrow::array::{ BinaryArray, DictionaryArray, TimestampMillisecondArray, UInt8Array, UInt32Array, UInt64Array, @@ -750,20 +747,18 @@ mod tests { #[test] fn test_maybe_wrap_schema_native_histogram() { - use common_query::native_histogram::NATIVE_HISTOGRAM_FIELD; - let schema = Arc::new(Schema::new(vec![ Field::new( "greptime_timestamp", ArrowDataType::Timestamp(TimeUnit::Millisecond, None), false, ), - histogram_field(NATIVE_HISTOGRAM_FIELD, 1), + histogram_field(greptime_native_histogram(), 1), ])); let wrapped = maybe_wrap_schema(&schema).unwrap(); let hist = wrapped - .field_with_name(NATIVE_HISTOGRAM_FIELD) + .field_with_name(greptime_native_histogram()) .expect("histogram field present"); // The struct has 18 sub-fields. let ArrowDataType::Struct(children) = hist.data_type() else { @@ -778,13 +773,11 @@ mod tests { // Two histogram columns with distinct parent column ids get disjoint // sub-field ids (defensive: the metric engine yields at most one // histogram column, but the scheme must stay correct if more appear). - use common_query::native_histogram::{ - NATIVE_HISTOGRAM_FIELD, native_histogram_subfield_id, - }; + use common_query::native_histogram::native_histogram_subfield_id; let schema = Arc::new(Schema::new(vec![ - histogram_field(NATIVE_HISTOGRAM_FIELD, 1), - histogram_field(NATIVE_HISTOGRAM_FIELD, 7), + histogram_field(greptime_native_histogram(), 1), + histogram_field(greptime_native_histogram(), 7), ])); let wrapped = maybe_wrap_schema(&schema).unwrap(); let h1 = &wrapped.fields()[0]; @@ -799,26 +792,12 @@ mod tests { } #[test] - fn test_maybe_wrap_schema_requires_canonical_name() { - // Detection requires the canonical column name: a histogram-typed field - // named differently is left untouched. - use arrow_schema::extension::EXTENSION_TYPE_NAME_KEY; - use common_query::native_histogram::native_histogram_value_type; - use datatypes::data_type::DataType; - - let hist_arrow = native_histogram_value_type().as_arrow_type(); - let schema = Arc::new(Schema::new(vec![Field::new( - "custom_histogram", - hist_arrow, - true, - )])); + fn test_maybe_wrap_schema_recognizes_histogram_by_type() { + let schema = Arc::new(Schema::new(vec![histogram_field("custom_histogram", 5)])); let wrapped = maybe_wrap_schema(&schema).unwrap(); let hist = wrapped.field_with_name("custom_histogram").unwrap(); - assert!( - hist.metadata().get(EXTENSION_TYPE_NAME_KEY).is_none(), - "a histogram-typed field without the canonical name must not be stamped" - ); + assert_histogram_stamped(hist, 5); } #[test] @@ -911,65 +890,42 @@ mod tests { // On-disk contract: after writing through the parquet writer path, the // footer still carries the greptime.histogram extension and every // nested (sub-field + list-element) PARQUET:field_id. - use common_query::native_histogram::NATIVE_HISTOGRAM_FIELD; - let schema = Arc::new(Schema::new(vec![ Field::new( "greptime_timestamp", ArrowDataType::Timestamp(TimeUnit::Millisecond, None), false, ), - histogram_field(NATIVE_HISTOGRAM_FIELD, 3), + histogram_field(greptime_native_histogram(), 3), ])); let on_disk = parquet_footer_arrow_schema(&schema); let hist = on_disk - .field_with_name(NATIVE_HISTOGRAM_FIELD) + .field_with_name(greptime_native_histogram()) .expect("histogram field present"); assert_histogram_stamped(hist, 3); } #[test] - fn test_parquet_roundtrip_noncanonical_struct_untouched() { - // A histogram-shaped struct without the canonical column name is left - // untouched on disk: no extension, no nested field ids. - use arrow_schema::extension::EXTENSION_TYPE_NAME_KEY; - use common_query::native_histogram::native_histogram_value_type; - use datatypes::data_type::DataType; - - let hist_arrow = native_histogram_value_type().as_arrow_type(); + fn test_parquet_roundtrip_recognizes_histogram_by_type() { + // The persisted type, rather than a process-local configured name, + // identifies native histograms across upgrades and prefix changes. let schema = Arc::new(Schema::new(vec![ Field::new( "ts", ArrowDataType::Timestamp(TimeUnit::Millisecond, None), false, ), - Field::new("custom_histogram", hist_arrow, true), + histogram_field("custom_histogram", 9), ])); let on_disk = parquet_footer_arrow_schema(&schema); let hist = on_disk.field_with_name("custom_histogram").unwrap(); - assert!( - hist.metadata().get(EXTENSION_TYPE_NAME_KEY).is_none(), - "a histogram-typed field without the canonical name must not be stamped on disk" - ); - if let ArrowDataType::Struct(children) = hist.data_type() { - for child in children { - assert!( - child.metadata().get(PARQUET_FIELD_ID_KEY).is_none(), - "non-histogram sub-field {} must not get a field id on disk", - child.name() - ); - } - } else { - panic!("expected a struct, got {:?}", hist.data_type()); - } + assert_histogram_stamped(hist, 9); } #[test] fn test_maybe_wrap_schema_overflows_return_error() { - use common_query::native_histogram::NATIVE_HISTOGRAM_FIELD; - // A column id of 12_582_912 makes the derived sub-field id overflow // i32 (BASE + column_id*64 == i32::MAX + 1). The write path must // surface this as an error rather than silently dropping the field @@ -980,7 +936,7 @@ mod tests { ArrowDataType::Timestamp(TimeUnit::Millisecond, None), false, ), - histogram_field(NATIVE_HISTOGRAM_FIELD, 12_582_912), + histogram_field(greptime_native_histogram(), 12_582_912), ])); let err = maybe_wrap_schema(&schema).unwrap_err(); assert!( @@ -1000,11 +956,11 @@ mod tests { // without the write path's stamping), the writer must fail loudly // rather than silently namespace under column 0, which would collide // with that column's nested ids. - use common_query::native_histogram::{NATIVE_HISTOGRAM_FIELD, native_histogram_value_type}; + use common_query::native_histogram::native_histogram_value_type; use datatypes::data_type::DataType; let field = Field::new( - NATIVE_HISTOGRAM_FIELD, + greptime_native_histogram(), native_histogram_value_type().as_arrow_type(), true, ); @@ -1030,15 +986,13 @@ mod tests { // above i32::MAX (e.g. u32::MAX) must not be silently parsed as a // failed i32 and collapsed onto column 0's nested ids. It must // surface a checked-conversion error instead. - use common_query::native_histogram::NATIVE_HISTOGRAM_FIELD; - let schema = Arc::new(Schema::new(vec![ Field::new( "greptime_timestamp", ArrowDataType::Timestamp(TimeUnit::Millisecond, None), false, ), - histogram_field(NATIVE_HISTOGRAM_FIELD, u32::MAX), + histogram_field(greptime_native_histogram(), u32::MAX), ])); let err = maybe_wrap_schema(&schema).unwrap_err(); assert!( diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs index c3e924eaef..74ecb6b3cd 100644 --- a/src/operator/src/insert.rs +++ b/src/operator/src/insert.rs @@ -37,9 +37,7 @@ use common_meta::cache::TableFlownodeSetCacheRef; use common_meta::node_manager::{AffectedRows, NodeManagerRef}; use common_meta::peer::Peer; use common_query::Output; -use common_query::native_histogram::{ - NATIVE_HISTOGRAM_FIELD, is_native_histogram_value_schema, native_histogram_value_type, -}; +use common_query::native_histogram::{is_native_histogram_value_type, native_histogram_value_type}; use common_query::prelude::{greptime_timestamp, greptime_value}; use common_telemetry::tracing_context::TracingContext; use common_telemetry::{error, info, warn}; @@ -1143,7 +1141,6 @@ fn request_is_native_histogram(request_schema: &[ColumnSchema]) -> bool { }; fields.next().is_none() - && col.column_name == NATIVE_HISTOGRAM_FIELD && api::helper::is_column_type_value_eq( col.datatype, col.datatype_extension.clone(), @@ -1157,7 +1154,7 @@ fn table_is_native_histogram(table: &TableRef) -> bool { return false; }; - fields.next().is_none() && is_native_histogram_value_schema(&col.name, &col.data_type) + fields.next().is_none() && is_native_histogram_value_type(&col.data_type) } fn validate_column_count_match(requests: &RowInsertRequests) -> Result<()> { @@ -1431,12 +1428,15 @@ impl FlowMirrorTask { mod tests { use std::sync::Arc; + use api::helper::ColumnDataTypeWrapper; use api::v1::helper::{field_column_schema, time_index_column_schema}; use api::v1::{RowInsertRequest, Rows, Value}; 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::test_util::MockDatanodeManager; + use common_query::native_histogram::NATIVE_HISTOGRAM_FIELD; + use common_query::prelude::{greptime_native_histogram, set_default_prefix}; use datatypes::data_type::ConcreteDataType; use datatypes::schema::ColumnSchema; use moka::future::Cache; @@ -1448,7 +1448,11 @@ mod tests { use super::*; use crate::tests::{create_partition_rule_manager, prepare_mocked_backend}; - fn make_table_ref_with_schema(ts_name: &str, field_name: &str) -> TableRef { + fn make_table_ref_with_schema( + ts_name: &str, + field_name: &str, + field_type: ConcreteDataType, + ) -> TableRef { let schema = datatypes::schema::SchemaBuilder::try_from_columns(vec![ ColumnSchema::new( ts_name, @@ -1456,7 +1460,7 @@ mod tests { false, ) .with_time_index(true), - ColumnSchema::new(field_name, ConcreteDataType::float64_datatype(), true), + ColumnSchema::new(field_name, field_type, true), ]) .unwrap() .build() @@ -1495,7 +1499,8 @@ mod tests { async fn test_accommodate_existing_schema_logic() { let ts_name = "my_ts"; let field_name = "my_field"; - let table = make_table_ref_with_schema(ts_name, field_name); + let table = + make_table_ref_with_schema(ts_name, field_name, ConcreteDataType::float64_datatype()); // The request uses different names for timestamp and field columns let mut req = RowInsertRequest { @@ -1545,6 +1550,30 @@ mod tests { assert_eq!(req_schema[1].column_name, field_name); } + #[test] + fn test_native_histogram_detection_survives_prefix_change() { + set_default_prefix(Some("custom")).unwrap(); + let table = make_table_ref_with_schema( + "custom_timestamp", + NATIVE_HISTOGRAM_FIELD, + native_histogram_value_type().clone(), + ); + let (datatype, datatype_extension) = + ColumnDataTypeWrapper::try_from(native_histogram_value_type().clone()) + .unwrap() + .into_parts(); + let request_schema = [api::v1::ColumnSchema { + column_name: greptime_native_histogram().to_string(), + datatype: datatype as i32, + semantic_type: SemanticType::Field as i32, + datatype_extension, + options: None, + }]; + + assert!(request_is_native_histogram(&request_schema)); + assert!(table_is_native_histogram(&table)); + } + #[test] fn test_last_non_null_create_options_preserve_default_without_append_mode() { let ctx = Arc::new(QueryContext::with( diff --git a/src/servers/src/otlp/metrics.rs b/src/servers/src/otlp/metrics.rs index 73d05684ec..30a10aebed 100644 --- a/src/servers/src/otlp/metrics.rs +++ b/src/servers/src/otlp/metrics.rs @@ -853,6 +853,7 @@ fn encode_summary( #[cfg(test)] mod tests { + use common_query::prelude::set_default_prefix; use otel_arrow_rust::proto::opentelemetry::common::v1::AnyValue; use otel_arrow_rust::proto::opentelemetry::common::v1::any_value::Value as Val; use otel_arrow_rust::proto::opentelemetry::metrics::v1::number_data_point::Value; @@ -1054,6 +1055,47 @@ mod tests { ); } + #[test] + fn test_encode_legacy_summary_keeps_legacy_column_names() { + set_default_prefix(Some("custom")).unwrap(); + let mut tables = MultiTableData::default(); + let summary = Summary { + data_points: vec![SummaryDataPoint { + attributes: vec![keyvalue("host", "testserver")], + time_unix_nano: 100, + count: 25, + quantile_values: vec![ValueAtQuantile { + quantile: 0.90, + value: 1000.0, + }], + ..Default::default() + }], + }; + + encode_summary( + &mut tables, + "datamon", + &summary, + None, + None, + &OtlpMetricCtx { + is_legacy: true, + ..Default::default() + }, + ) + .unwrap(); + + let table = tables.get_or_default_table_data("datamon", 0, 0); + assert_eq!( + table + .columns() + .iter() + .map(|column| column.column_name.as_str()) + .collect::>(), + vec!["host", "custom_timestamp", "greptime_p90", GREPTIME_COUNT,] + ); + } + #[test] fn test_encode_histogram() { let mut tables = MultiTableData::default(); diff --git a/src/servers/src/prom_remote_write/README.md b/src/servers/src/prom_remote_write/README.md index acac7fd685..1f15f8288a 100644 --- a/src/servers/src/prom_remote_write/README.md +++ b/src/servers/src/prom_remote_write/README.md @@ -25,7 +25,7 @@ flowchart TD J --> K["same metric-engine flag as samples, no batcher"] K --> L["table: "] - L --> M["field: greptime_native_histogram Struct"] + L --> M["field: configured native-histogram Struct"] M --> N["struct children: counts, spans, buckets, sum, schema"] H --> P["written headers and counters"] K --> P @@ -41,7 +41,9 @@ Native histogram rows follow the same metric-engine switch as samples. They do not use the pending rows batcher yet because the batcher assumes the classic timestamp + Float64 value + string tags shape. -Each histogram row stores `greptime_native_histogram` as one Struct field: +Each histogram row stores one Struct field named +`_native_histogram`. The default name is +`greptime_native_histogram`; an empty prefix produces `native_histogram`. - common scalar children: `schema`, `zero_threshold`, `sum`, `reset_hint`, `start_timestamp`; diff --git a/src/servers/src/prom_remote_write/v2.rs b/src/servers/src/prom_remote_write/v2.rs index b4128994ce..ceb3d67d15 100644 --- a/src/servers/src/prom_remote_write/v2.rs +++ b/src/servers/src/prom_remote_write/v2.rs @@ -25,7 +25,7 @@ use api::v1::{ColumnSchema, ListValue, RowInsertRequest, Rows, SemanticType, Val use bytes::Bytes; use common_grpc::precision::Precision; use common_query::native_histogram::*; -use common_query::prelude::{greptime_timestamp, greptime_value}; +use common_query::prelude::{greptime_native_histogram, greptime_timestamp, greptime_value}; use pipeline::{ContextOpt, ContextReq}; use prost::Message; use snafu::{OptionExt, ResultExt, ensure}; @@ -270,7 +270,7 @@ fn native_histogram_column_schema() -> ColumnSchema { .into_parts(); ColumnSchema { - column_name: NATIVE_HISTOGRAM_FIELD.to_string(), + column_name: greptime_native_histogram().to_string(), datatype: datatype as i32, semantic_type: SemanticType::Field as i32, datatype_extension, @@ -400,7 +400,7 @@ fn ensure_no_internal_histogram_labels(tags: &PromTags) -> Result<()> { // The histogram field column is generated from the protobuf payload. for (name, _) in tags { ensure!( - name != NATIVE_HISTOGRAM_FIELD, + name != greptime_native_histogram() && name != NATIVE_HISTOGRAM_FIELD, error::InvalidPromRemoteRequestSnafu { msg: format!( "remote write v2 label `{name}` conflicts with an internal native histogram label" @@ -625,7 +625,7 @@ mod tests { use std::sync::Arc; use api::v1::value::ValueData; - use common_query::prelude::{greptime_timestamp, greptime_value}; + use common_query::prelude::{greptime_timestamp, greptime_value, set_default_prefix}; use session::context::QueryContext; use super::*; @@ -854,7 +854,7 @@ mod tests { "internal histogram label on samples", request_with_sample(vec![ (METRIC_NAME_LABEL, "metric"), - (NATIVE_HISTOGRAM_FIELD, "user_value"), + (greptime_native_histogram(), "user_value"), ]), "conflicts with an internal native histogram label", )); @@ -962,7 +962,7 @@ mod tests { .iter() .map(|col| col.column_name.as_str()) .collect::>(), - vec![greptime_timestamp(), NATIVE_HISTOGRAM_FIELD] + vec![greptime_timestamp(), greptime_native_histogram()] ); assert_eq!( rows.rows[0].values[0].value_data, @@ -1005,7 +1005,7 @@ mod tests { let mut request = test_util::request_with_labels_and_samples( vec![ (METRIC_NAME_LABEL, "metric"), - (NATIVE_HISTOGRAM_FIELD, "user_value"), + (greptime_native_histogram(), "user_value"), ], vec![], ); @@ -1021,6 +1021,22 @@ mod tests { ); } + #[test] + fn test_rejects_legacy_histogram_label_after_prefix_change() { + set_default_prefix(Some("custom")).unwrap(); + assert_eq!(greptime_native_histogram(), "custom_native_histogram"); + + let err = ensure_no_internal_histogram_labels(&vec![( + NATIVE_HISTOGRAM_FIELD.to_string(), + "user_value".to_string(), + )]) + .unwrap_err(); + assert!( + err.to_string() + .contains("conflicts with an internal native histogram label") + ); + } + #[test] fn test_into_context_req_converts_int_and_float_histograms_to_one_schema() { let float_histogram = Histogram { @@ -1070,7 +1086,7 @@ mod tests { .iter() .map(|col| col.column_name.as_str()) .collect::>(), - vec![greptime_timestamp(), NATIVE_HISTOGRAM_FIELD] + vec![greptime_timestamp(), greptime_native_histogram()] ); assert_eq!( @@ -1138,7 +1154,7 @@ mod tests { } fn histogram_field_value(rows: &Rows, row_idx: usize, field_name: &str) -> Option { - let histogram_idx = column_index(&rows.schema, NATIVE_HISTOGRAM_FIELD); + let histogram_idx = column_index(&rows.schema, greptime_native_histogram()); let Some(ValueData::StructValue(histogram)) = &rows.rows[row_idx].values[histogram_idx].value_data else { diff --git a/src/servers/tests/http/prom_store_test.rs b/src/servers/tests/http/prom_store_test.rs index 3c4cfbe4e9..2d573a3bfb 100644 --- a/src/servers/tests/http/prom_store_test.rs +++ b/src/servers/tests/http/prom_store_test.rs @@ -28,8 +28,9 @@ use async_trait::async_trait; use axum::Router; use axum::http::HeaderMap; use common_query::Output; -use common_query::native_histogram::NATIVE_HISTOGRAM_FIELD; -use common_query::prelude::{GREPTIME_PHYSICAL_TABLE, greptime_timestamp, greptime_value}; +use common_query::prelude::{ + GREPTIME_PHYSICAL_TABLE, greptime_native_histogram, greptime_timestamp, greptime_value, +}; use common_test_util::ports; use datafusion_expr::LogicalPlan; use prost::Message; @@ -675,7 +676,7 @@ async fn test_prometheus_remote_write_v2_writes_histogram_only_series() { assert!( rows.schema .iter() - .any(|column| column.column_name == NATIVE_HISTOGRAM_FIELD + .any(|column| column.column_name == greptime_native_histogram() && column.datatype == ColumnDataType::Struct as i32) ); assert!(write_rx.try_recv().is_err()); diff --git a/src/servers/tests/prom_remote_write_v2_test.rs b/src/servers/tests/prom_remote_write_v2_test.rs index c06291b437..a44583dccd 100644 --- a/src/servers/tests/prom_remote_write_v2_test.rs +++ b/src/servers/tests/prom_remote_write_v2_test.rs @@ -18,10 +18,10 @@ use api::v1::value::ValueData; use api::v1::{ColumnSchema, Rows}; use bytes::Bytes; use common_query::native_histogram::{ - COUNT_U64_FIELD, NATIVE_HISTOGRAM_FIELD, NATIVE_HISTOGRAM_FIELD_NAMES, - POSITIVE_BUCKETS_F64_FIELD, POSITIVE_BUCKETS_I64_FIELD, POSITIVE_SPAN_OFFSETS_FIELD, - SCHEMA_FIELD, + COUNT_U64_FIELD, NATIVE_HISTOGRAM_FIELD_NAMES, POSITIVE_BUCKETS_F64_FIELD, + POSITIVE_BUCKETS_I64_FIELD, POSITIVE_SPAN_OFFSETS_FIELD, SCHEMA_FIELD, }; +use common_query::prelude::greptime_native_histogram; use servers::prom_remote_write::v2::test_util as remote_write_v2; #[test] @@ -126,7 +126,7 @@ fn column_index(schema: &[ColumnSchema], column_name: &str) -> usize { } fn histogram_field_value(rows: &Rows, row_idx: usize, field_name: &str) -> Option { - let histogram_idx = column_index(&rows.schema, NATIVE_HISTOGRAM_FIELD); + let histogram_idx = column_index(&rows.schema, greptime_native_histogram()); let Some(ValueData::StructValue(histogram)) = &rows.rows[row_idx].values[histogram_idx].value_data else {