diff --git a/config/config.md b/config/config.md
index 0cf78c0fb95..3a6e187375f 100644
--- a/config/config.md
+++ b/config/config.md
@@ -32,6 +32,7 @@
| `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default
This allows browser to access http APIs without CORS restrictions |
| `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. |
| `http.prom_validation_mode` | String | `strict` | Whether to enable validation for Prometheus remote write requests.
Available options:
- strict: deny invalid UTF-8 strings (default).
- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).
- unchecked: do not valid strings. |
+| `http.experimental_enable_prometheus_native_histogram` | Bool | `false` | Experimental: enable Prometheus remote write v2 native histogram ingestion. |
| `http.experimental_enable_explain_analyze_stream` | Bool | `false` | Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics. |
| `grpc` | -- | -- | The gRPC server options. |
| `grpc.bind_addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
@@ -248,6 +249,7 @@
| `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default
This allows browser to access http APIs without CORS restrictions |
| `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. |
| `http.prom_validation_mode` | String | `strict` | Whether to enable validation for Prometheus remote write requests.
Available options:
- strict: deny invalid UTF-8 strings (default).
- lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).
- unchecked: do not valid strings. |
+| `http.experimental_enable_prometheus_native_histogram` | Bool | `false` | Experimental: enable Prometheus remote write v2 native histogram ingestion. |
| `http.experimental_enable_explain_analyze_stream` | Bool | `false` | Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics. |
| `grpc` | -- | -- | The gRPC server options. |
| `grpc.bind_addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
diff --git a/config/frontend.example.toml b/config/frontend.example.toml
index 06b4efd36e2..badba500623 100644
--- a/config/frontend.example.toml
+++ b/config/frontend.example.toml
@@ -57,6 +57,8 @@ cors_allowed_origins = ["https://example.com"]
## - lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).
## - unchecked: do not valid strings.
prom_validation_mode = "strict"
+## Experimental: enable Prometheus remote write v2 native histogram ingestion.
+experimental_enable_prometheus_native_histogram = false
## Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics.
experimental_enable_explain_analyze_stream = false
diff --git a/config/standalone.example.toml b/config/standalone.example.toml
index 407280278d8..de3fa7a49d8 100644
--- a/config/standalone.example.toml
+++ b/config/standalone.example.toml
@@ -71,6 +71,8 @@ cors_allowed_origins = ["https://example.com"]
## - lossy: allow invalid UTF-8 strings, replace invalid characters with REPLACEMENT_CHARACTER(U+FFFD).
## - unchecked: do not valid strings.
prom_validation_mode = "strict"
+## Experimental: enable Prometheus remote write v2 native histogram ingestion.
+experimental_enable_prometheus_native_histogram = false
## Experimental: enable POST /v1/sql/analyze/stream for streaming EXPLAIN ANALYZE VERBOSE metrics.
experimental_enable_explain_analyze_stream = false
diff --git a/src/common/query/src/lib.rs b/src/common/query/src/lib.rs
index 91a417d3560..bdba49eaf70 100644
--- a/src/common/query/src/lib.rs
+++ b/src/common/query/src/lib.rs
@@ -15,6 +15,7 @@
pub mod columnar_value;
pub mod error;
pub mod logical_plan;
+pub mod native_histogram;
pub mod prelude;
pub mod request;
pub mod stream;
diff --git a/src/common/query/src/native_histogram.rs b/src/common/query/src/native_histogram.rs
new file mode 100644
index 00000000000..b10c78104a5
--- /dev/null
+++ b/src/common/query/src/native_histogram.rs
@@ -0,0 +1,111 @@
+// 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.
+
+//! Shared native histogram field contract.
+//!
+//! Prom remote-write v2 stores these names as children of one Struct field, while
+//! metric-engine uses the same contract to recognize native histogram tables.
+
+use std::sync::Arc;
+
+use datatypes::data_type::ConcreteDataType;
+use datatypes::types::{StructField, StructType};
+use once_cell::sync::Lazy;
+
+pub const NATIVE_HISTOGRAM_FIELD: &str = "greptime_native_histogram";
+pub const SCHEMA_FIELD: &str = "schema";
+pub const ZERO_THRESHOLD_FIELD: &str = "zero_threshold";
+pub const SUM_FIELD: &str = "sum";
+pub const RESET_HINT_FIELD: &str = "reset_hint";
+pub const START_TIMESTAMP_FIELD: &str = "start_timestamp";
+pub const CUSTOM_VALUES_FIELD: &str = "custom_values";
+pub const POSITIVE_SPAN_OFFSETS_FIELD: &str = "positive_span_offsets";
+pub const POSITIVE_SPAN_LENGTHS_FIELD: &str = "positive_span_lengths";
+pub const NEGATIVE_SPAN_OFFSETS_FIELD: &str = "negative_span_offsets";
+pub const NEGATIVE_SPAN_LENGTHS_FIELD: &str = "negative_span_lengths";
+pub const COUNT_U64_FIELD: &str = "count_u64";
+pub const ZERO_COUNT_U64_FIELD: &str = "zero_count_u64";
+pub const POSITIVE_BUCKETS_I64_FIELD: &str = "positive_buckets_i64";
+pub const NEGATIVE_BUCKETS_I64_FIELD: &str = "negative_buckets_i64";
+pub const COUNT_F64_FIELD: &str = "count_f64";
+pub const ZERO_COUNT_F64_FIELD: &str = "zero_count_f64";
+pub const POSITIVE_BUCKETS_F64_FIELD: &str = "positive_buckets_f64";
+pub const NEGATIVE_BUCKETS_F64_FIELD: &str = "negative_buckets_f64";
+
+// Keep int and float payloads in separate columns. The populated family is the
+// type signal, so we don't need to persist an extra histogram-type tag.
+pub const NATIVE_HISTOGRAM_FIELD_NAMES: &[&str] = &[
+ SCHEMA_FIELD,
+ ZERO_THRESHOLD_FIELD,
+ SUM_FIELD,
+ RESET_HINT_FIELD,
+ START_TIMESTAMP_FIELD,
+ CUSTOM_VALUES_FIELD,
+ POSITIVE_SPAN_OFFSETS_FIELD,
+ POSITIVE_SPAN_LENGTHS_FIELD,
+ NEGATIVE_SPAN_OFFSETS_FIELD,
+ NEGATIVE_SPAN_LENGTHS_FIELD,
+ COUNT_U64_FIELD,
+ ZERO_COUNT_U64_FIELD,
+ POSITIVE_BUCKETS_I64_FIELD,
+ NEGATIVE_BUCKETS_I64_FIELD,
+ COUNT_F64_FIELD,
+ ZERO_COUNT_F64_FIELD,
+ POSITIVE_BUCKETS_F64_FIELD,
+ NEGATIVE_BUCKETS_F64_FIELD,
+];
+
+static NATIVE_HISTOGRAM_VALUE_TYPE: Lazy = Lazy::new(|| {
+ let fields = NATIVE_HISTOGRAM_FIELD_NAMES
+ .iter()
+ .filter_map(|name| {
+ let data_type = native_histogram_field_type(name)?;
+ Some(StructField::new((*name).to_string(), data_type, true))
+ })
+ .collect();
+ ConcreteDataType::struct_datatype(StructType::new(Arc::new(fields)))
+});
+
+/// Returns the exact Greptime type for a persisted native histogram field.
+pub fn native_histogram_field_type(name: &str) -> Option {
+ match name {
+ SCHEMA_FIELD | RESET_HINT_FIELD => Some(ConcreteDataType::int32_datatype()),
+ ZERO_THRESHOLD_FIELD | SUM_FIELD | COUNT_F64_FIELD | ZERO_COUNT_F64_FIELD => {
+ Some(ConcreteDataType::float64_datatype())
+ }
+ START_TIMESTAMP_FIELD => Some(ConcreteDataType::timestamp_millisecond_datatype()),
+ CUSTOM_VALUES_FIELD | POSITIVE_BUCKETS_F64_FIELD | NEGATIVE_BUCKETS_F64_FIELD => Some(
+ ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::float64_datatype())),
+ ),
+ POSITIVE_SPAN_OFFSETS_FIELD | NEGATIVE_SPAN_OFFSETS_FIELD => Some(
+ ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::int32_datatype())),
+ ),
+ POSITIVE_SPAN_LENGTHS_FIELD | NEGATIVE_SPAN_LENGTHS_FIELD => Some(
+ ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::uint32_datatype())),
+ ),
+ COUNT_U64_FIELD | ZERO_COUNT_U64_FIELD => Some(ConcreteDataType::uint64_datatype()),
+ POSITIVE_BUCKETS_I64_FIELD | NEGATIVE_BUCKETS_I64_FIELD => Some(
+ ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::int64_datatype())),
+ ),
+ _ => None,
+ }
+}
+
+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()
+}
diff --git a/src/metric-engine/AGENTS.md b/src/metric-engine/AGENTS.md
index 86b676db48c..4471de3faee 100644
--- a/src/metric-engine/AGENTS.md
+++ b/src/metric-engine/AGENTS.md
@@ -59,7 +59,9 @@ Mito2 data region. Queries against a physical region pass straight through.
`store-api`'s metric engine consts; keep them in sync with `engine.rs`.
- **Metadata K-V encoding** (`metadata_region.rs`): changes the on-disk metadata layout.
- **RegionId group mapping** (`utils.rs`): data vs metadata region derivation.
-- **Physical column rules** (`engine/alter/`): a physical region allows only one field column.
+- **Physical column rules** (`engine/alter/`): ordinary metric fields stay on
+ logical tables; shared physical field adds are only for the native histogram
+ Struct field.
## Testing
diff --git a/src/metric-engine/src/data_region.rs b/src/metric-engine/src/data_region.rs
index b64d61a287e..ec9ff15c8d9 100644
--- a/src/metric-engine/src/data_region.rs
+++ b/src/metric-engine/src/data_region.rs
@@ -13,6 +13,7 @@
// limitations under the License.
use api::v1::SemanticType;
+use common_query::native_histogram::is_native_histogram_value_schema;
use common_telemetry::{debug, info};
use datatypes::schema::{SkippingIndexOptions, SkippingIndexType};
use mito2::engine::MitoEngine;
@@ -123,42 +124,55 @@ impl DataRegion {
.into_iter()
.enumerate()
.map(|(delta, mut c)| {
- if c.semantic_type == SemanticType::Tag {
- if !c.column_schema.data_type.is_string() {
- return ColumnTypeMismatchSnafu {
- expect: ConcreteDataType::string_datatype(),
- actual: c.column_schema.data_type.clone(),
+ match c.semantic_type {
+ SemanticType::Tag => {
+ if !c.column_schema.data_type.is_string() {
+ return ColumnTypeMismatchSnafu {
+ expect: ConcreteDataType::string_datatype(),
+ actual: c.column_schema.data_type.clone(),
+ }
+ .fail();
+ }
+ }
+ // Field columns can only be added to the shared physical
+ // 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,
+ ) => {}
+ _ => {
+ return AddingFieldColumnSnafu {
+ name: &c.column_schema.name,
}
.fail();
}
- } else {
- return AddingFieldColumnSnafu {
- name: &c.column_schema.name,
- }
- .fail();
- };
+ }
c.column_id = new_column_id_start + delta as u32;
c.column_schema.set_nullable();
- match index_options {
- IndexOptions::None => {}
- IndexOptions::Inverted => {
- c.column_schema.set_inverted_index(true);
- }
- IndexOptions::Skipping {
- granularity,
- false_positive_rate,
- } => {
- c.column_schema
- .set_skipping_options(
- &SkippingIndexOptions::new(
- granularity,
- false_positive_rate,
- SkippingIndexType::BloomFilter,
+ if c.semantic_type == SemanticType::Tag {
+ match index_options {
+ IndexOptions::None => {}
+ IndexOptions::Inverted => {
+ c.column_schema.set_inverted_index(true);
+ }
+ IndexOptions::Skipping {
+ granularity,
+ false_positive_rate,
+ } => {
+ c.column_schema
+ .set_skipping_options(
+ &SkippingIndexOptions::new(
+ granularity,
+ false_positive_rate,
+ SkippingIndexType::BloomFilter,
+ )
+ .context(SetSkippingIndexOptionSnafu)?,
)
- .context(SetSkippingIndexOptionSnafu)?,
- )
- .context(SetSkippingIndexOptionSnafu)?;
+ .context(SetSkippingIndexOptionSnafu)?;
+ }
}
}
diff --git a/src/metric-engine/src/engine/alter.rs b/src/metric-engine/src/engine/alter.rs
index 892e0c91c2d..e9d8e878df7 100644
--- a/src/metric-engine/src/engine/alter.rs
+++ b/src/metric-engine/src/engine/alter.rs
@@ -17,6 +17,8 @@ mod validate;
use std::collections::{BTreeSet, HashMap, HashSet};
+use api::v1::SemanticType;
+use common_query::native_histogram::is_native_histogram_value_schema;
use extract_new_columns::extract_new_columns;
use snafu::{OptionExt, ResultExt, ensure};
use store_api::metadata::ColumnMetadata;
@@ -27,8 +29,8 @@ use validate::validate_alter_region_requests;
use crate::engine::MetricEngineInner;
use crate::error::{
- LogicalRegionNotFoundSnafu, PhysicalRegionNotFoundSnafu, Result, SerializeColumnMetadataSnafu,
- UnexpectedRequestSnafu,
+ AddingFieldColumnSnafu, LogicalRegionNotFoundSnafu, PhysicalRegionNotFoundSnafu, Result,
+ SerializeColumnMetadataSnafu, UnexpectedRequestSnafu,
};
use crate::utils::{append_manifest_info, encode_manifest_info_to_extensions, to_data_region_id};
@@ -117,6 +119,8 @@ impl MetricEngineInner {
) -> Result {
// Checks all alter requests are add columns.
validate_alter_region_requests(&requests)?;
+ self.validate_logical_field_alters(physical_region_id, &requests)
+ .await?;
// Finds new columns to add
let mut new_column_names = HashSet::new();
@@ -209,6 +213,52 @@ impl MetricEngineInner {
Ok(0)
}
+ async fn validate_logical_field_alters(
+ &self,
+ physical_region_id: RegionId,
+ requests: &[(RegionId, RegionAlterRequest)],
+ ) -> Result<()> {
+ // Logical metric tables have one field column. Native histograms are a
+ // special struct field, so field alters must leave exactly that field.
+ for (region_id, request) in requests {
+ let AlterKind::AddColumns { columns } = &request.kind else {
+ unreachable!()
+ };
+ let added_fields = columns
+ .iter()
+ .filter(|col| col.column_metadata.semantic_type == SemanticType::Field)
+ .collect::>();
+ let Some(&first_added_field) = added_fields.first() else {
+ continue;
+ };
+
+ let mut fields = self
+ .load_logical_columns(physical_region_id, *region_id)
+ .await?
+ .into_iter()
+ .filter(|col| col.semantic_type == SemanticType::Field)
+ .collect::>();
+ fields.extend(
+ added_fields
+ .into_iter()
+ .map(|col| col.column_metadata.clone()),
+ );
+
+ ensure!(
+ fields.len() == 1
+ && is_native_histogram_value_schema(
+ &fields[0].column_schema.name,
+ &fields[0].column_schema.data_type
+ ),
+ AddingFieldColumnSnafu {
+ name: first_added_field.column_metadata.column_schema.name.clone(),
+ }
+ );
+ }
+
+ Ok(())
+ }
+
async fn alter_physical_region(
&self,
region_id: RegionId,
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 c3bef1ba4fd..3573762d742 100644
--- a/src/metric-engine/src/engine/alter/extract_new_columns.rs
+++ b/src/metric-engine/src/engine/alter/extract_new_columns.rs
@@ -15,6 +15,7 @@
use std::collections::{HashMap, HashSet};
use api::v1::SemanticType;
+use common_query::native_histogram::is_native_histogram_value_schema;
use snafu::ensure;
use store_api::metadata::ColumnMetadata;
use store_api::region_request::{AlterKind, RegionAlterRequest};
@@ -43,7 +44,11 @@ pub fn extract_new_columns<'a>(
&& !new_column_names.contains(column_name)
{
ensure!(
- col.column_metadata.semantic_type != SemanticType::Field,
+ col.column_metadata.semantic_type != SemanticType::Field
+ || is_native_histogram_value_schema(
+ &col.column_metadata.column_schema.name,
+ &col.column_metadata.column_schema.data_type
+ ),
AddingFieldColumnSnafu {
name: column_name.to_string(),
}
diff --git a/src/metric-engine/src/engine/create.rs b/src/metric-engine/src/engine/create.rs
index c0b5da4145f..a5cbd566723 100644
--- a/src/metric-engine/src/engine/create.rs
+++ b/src/metric-engine/src/engine/create.rs
@@ -17,6 +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_telemetry::info;
use common_time::{FOREVER, Timestamp};
use datatypes::data_type::ConcreteDataType;
@@ -390,8 +391,8 @@ impl MetricEngineInner {
ConflictRegionOptionSnafu {}
);
- // check if only one field column is declared, and all tag columns are string
- let mut field_col: Option<&ColumnMetadata> = None;
+ // check if field columns are either a normal metric value or a native histogram.
+ let mut field_cols = Vec::new();
for col in &request.column_metadatas {
// Verified in above steps.
if is_metric_engine_internal_column(&col.column_schema.name) {
@@ -406,21 +407,30 @@ impl MetricEngineInner {
}
),
SemanticType::Field => {
- if let Some(field_col) = field_col {
- MultipleFieldColumnSnafu {
- previous: field_col.column_schema.name.clone(),
- current: col.column_schema.name.clone(),
- }
- .fail()?;
- }
- field_col = Some(col)
+ field_cols.push(col);
}
SemanticType::Timestamp => {}
}
}
- let field_col = field_col.context(NoFieldColumnSnafu)?;
+ let [field_col] = field_cols.as_slice() else {
+ if field_cols.is_empty() {
+ NoFieldColumnSnafu.fail()?;
+ }
+ return MultipleFieldColumnSnafu {
+ previous: field_cols[0].column_schema.name.clone(),
+ current: field_cols[1].column_schema.name.clone(),
+ }
+ .fail();
+ };
- // make sure the field column is float64 type
+ if is_native_histogram_value_schema(
+ &field_col.column_schema.name,
+ &field_col.column_schema.data_type,
+ ) {
+ return Ok(());
+ }
+
+ // make sure the normal field column is float64 type
ensure!(
field_col.column_schema.data_type == ConcreteDataType::float64_datatype(),
ColumnTypeMismatchSnafu {
@@ -653,6 +663,7 @@ 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 store_api::metric_engine_consts::{METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY};
use store_api::region_request::{BatchRegionDdlRequest, RegionRequirements};
@@ -845,6 +856,96 @@ mod test {
MetricEngineInner::verify_region_create_request(&request).unwrap();
}
+ #[test]
+ fn test_verify_region_create_request_native_histogram_fields() {
+ let native_histogram_columns = vec![
+ ColumnMetadata {
+ column_id: 0,
+ semantic_type: SemanticType::Timestamp,
+ column_schema: ColumnSchema::new(
+ greptime_timestamp(),
+ ConcreteDataType::timestamp_millisecond_datatype(),
+ false,
+ ),
+ },
+ ColumnMetadata {
+ column_id: 1,
+ semantic_type: SemanticType::Tag,
+ column_schema: ColumnSchema::new("job", ConcreteDataType::string_datatype(), true),
+ },
+ ColumnMetadata {
+ column_id: 2,
+ semantic_type: SemanticType::Field,
+ column_schema: ColumnSchema::new(
+ NATIVE_HISTOGRAM_FIELD,
+ native_histogram_value_type().clone(),
+ true,
+ ),
+ },
+ ];
+ let request = RegionCreateRequest {
+ column_metadatas: native_histogram_columns,
+ table_dir: "test_dir".to_string(),
+ path_type: PathType::Bare,
+ engine: METRIC_ENGINE_NAME.to_string(),
+ primary_key: vec![],
+ options: [(
+ LOGICAL_TABLE_METADATA_KEY.to_string(),
+ "physical".to_string(),
+ )]
+ .into_iter()
+ .collect(),
+ partition_expr_json: Some("".to_string()),
+ requirements: Default::default(),
+ };
+ MetricEngineInner::verify_region_create_request(&request).unwrap();
+
+ let request = RegionCreateRequest {
+ column_metadatas: vec![
+ ColumnMetadata {
+ column_id: 0,
+ semantic_type: SemanticType::Timestamp,
+ column_schema: ColumnSchema::new(
+ greptime_timestamp(),
+ ConcreteDataType::timestamp_millisecond_datatype(),
+ false,
+ ),
+ },
+ ColumnMetadata {
+ column_id: 1,
+ semantic_type: SemanticType::Field,
+ column_schema: ColumnSchema::new(
+ "value_a",
+ ConcreteDataType::float64_datatype(),
+ true,
+ ),
+ },
+ ColumnMetadata {
+ column_id: 2,
+ semantic_type: SemanticType::Field,
+ column_schema: ColumnSchema::new(
+ "value_b",
+ ConcreteDataType::float64_datatype(),
+ true,
+ ),
+ },
+ ],
+ table_dir: "test_dir".to_string(),
+ path_type: PathType::Bare,
+ engine: METRIC_ENGINE_NAME.to_string(),
+ primary_key: vec![],
+ options: [(
+ LOGICAL_TABLE_METADATA_KEY.to_string(),
+ "physical".to_string(),
+ )]
+ .into_iter()
+ .collect(),
+ partition_expr_json: Some("".to_string()),
+ requirements: Default::default(),
+ };
+ assert!(MetricEngineInner::verify_region_create_request(&request).is_err());
+ }
+
#[tokio::test]
async fn test_create_request_for_physical_regions() {
// original request
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 9859caa06b8..f0d3af87487 100644
--- a/src/metric-engine/src/engine/create/extract_new_columns.rs
+++ b/src/metric-engine/src/engine/create/extract_new_columns.rs
@@ -15,6 +15,7 @@
use std::collections::{HashMap, HashSet};
use api::v1::SemanticType;
+use common_query::native_histogram::is_native_histogram_value_schema;
use snafu::ensure;
use store_api::metadata::ColumnMetadata;
use store_api::region_request::RegionCreateRequest;
@@ -35,7 +36,11 @@ pub fn extract_new_columns<'a>(
&& !new_column_names.contains(&col.column_schema.name.as_str())
{
ensure!(
- col.semantic_type != SemanticType::Field,
+ col.semantic_type != SemanticType::Field
+ || is_native_histogram_value_schema(
+ &col.column_schema.name,
+ &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 103b15e596e..21865ecdbb1 100644
--- a/src/metric-engine/src/engine/put.rs
+++ b/src/metric-engine/src/engine/put.rs
@@ -195,12 +195,10 @@ impl MetricEngineInner {
requests: Vec<(RegionId, RegionPutRequest)>,
) -> Result<(RegionPutRequest, AffectedRows)> {
let total_rows: usize = requests.iter().map(|(_, req)| req.rows.rows.len()).sum();
- let mut merged_rows = Vec::with_capacity(total_rows);
+ let mut modified_requests = Vec::with_capacity(requests.len());
let mut total_affected_rows: AffectedRows = 0;
- let mut output_schema: Option> = None;
let mut merged_version: Option = None;
- // Modify and collect rows from each request
for (logical_region_id, mut request) in requests {
if let Some(request_version) = request.partition_expr_version {
if let Some(merged_version) = merged_version {
@@ -224,17 +222,15 @@ impl MetricEngineInner {
let row_count = request.rows.rows.len();
total_affected_rows += row_count as AffectedRows;
-
- // Capture the output schema from the first modified request
- if output_schema.is_none() {
- output_schema = Some(request.rows.schema.clone());
- }
-
- merged_rows.extend(request.rows.rows);
+ modified_requests.push(request.rows);
}
- // Safe to unwrap: requests is guaranteed non-empty by caller
- let schema = output_schema.unwrap();
+ let schema =
+ Self::build_union_schema(modified_requests.iter().map(|rows| rows.schema.as_slice()));
+ let mut merged_rows = Vec::with_capacity(total_rows);
+ for rows in modified_requests {
+ merged_rows.extend(Self::align_rows_to_schema(rows, &schema));
+ }
let merged_request = RegionPutRequest {
rows: Rows {
@@ -261,7 +257,8 @@ impl MetricEngineInner {
requests: Vec<(RegionId, RegionPutRequest)>,
) -> Result<(RegionPutRequest, AffectedRows)> {
// Build union schema from all requests
- let merged_schema = Self::build_union_schema(&requests);
+ let merged_schema =
+ Self::build_union_schema(requests.iter().map(|(_, req)| req.rows.schema.as_slice()));
// Align all rows to the merged schema and collect table_ids
let (merged_rows, table_ids, merged_version) =
@@ -302,17 +299,21 @@ impl MetricEngineInner {
Ok((merged_request, table_ids.len() as AffectedRows))
}
- /// Builds a union schema containing all columns from all requests.
- fn build_union_schema(requests: &[(RegionId, RegionPutRequest)]) -> Vec {
- let mut schema_map: HashMap<&str, ColumnSchema> = HashMap::new();
- for (_, request) in requests {
- for col in &request.rows.schema {
- schema_map
- .entry(col.column_name.as_str())
- .or_insert_with(|| col.clone());
+ fn build_union_schema<'a>(
+ schemas: impl IntoIterator- ,
+ ) -> Vec {
+ let mut schema = Vec::new();
+ for columns in schemas {
+ for col in columns {
+ if !schema
+ .iter()
+ .any(|existing: &ColumnSchema| existing.column_name == col.column_name)
+ {
+ schema.push(col.clone());
+ }
}
}
- schema_map.into_values().collect()
+ schema
}
fn align_requests_to_schema(
@@ -325,8 +326,6 @@ impl MetricEngineInner {
let mut table_ids = Vec::with_capacity(total_rows);
let mut merged_version: Option = None;
- let null_value = Value { value_data: None };
-
for (logical_region_id, request) in requests {
if let Some(request_version) = request.partition_expr_version {
if let Some(merged_version) = merged_version {
@@ -342,47 +341,54 @@ impl MetricEngineInner {
}
}
let table_id = logical_region_id.table_id();
-
- // Build column name to index mapping once per request
- let col_name_to_idx: FxHashMap<&str, usize> = request
- .rows
- .schema
- .iter()
- .enumerate()
- .map(|(idx, col)| (col.column_name.as_str(), idx))
- .collect();
-
- // Build column mapping array once per request
- // col_mapping[i] = Some(idx) means merged_schema[i] is at request.schema[idx]
- // col_mapping[i] = None means merged_schema[i] doesn't exist in request.schema
- let col_mapping: Vec> = merged_schema
- .iter()
- .map(|merged_col| {
- col_name_to_idx
- .get(merged_col.column_name.as_str())
- .copied()
- })
- .collect();
-
- // Apply the mapping to all rows
- for mut row in request.rows.rows {
- let mut aligned_values = Vec::with_capacity(merged_schema.len());
- for &opt_idx in &col_mapping {
- aligned_values.push(match opt_idx {
- Some(idx) => std::mem::take(&mut row.values[idx]),
- None => null_value.clone(),
- });
- }
- merged_rows.push(Row {
- values: aligned_values,
- });
- table_ids.push(table_id);
- }
+ let row_count = request.rows.rows.len();
+ merged_rows.extend(Self::align_rows_to_schema(request.rows, merged_schema));
+ table_ids.extend(std::iter::repeat_n(table_id, row_count));
}
Ok((merged_rows, table_ids, merged_version))
}
+ fn align_rows_to_schema(rows: Rows, merged_schema: &[ColumnSchema]) -> Vec
{
+ let Rows { schema, rows } = rows;
+ if schema.len() == merged_schema.len()
+ && schema
+ .iter()
+ .zip(merged_schema)
+ .all(|(left, right)| left.column_name == right.column_name)
+ {
+ return rows;
+ }
+
+ let col_name_to_idx: FxHashMap<&str, usize> = schema
+ .iter()
+ .enumerate()
+ .map(|(idx, col)| (col.column_name.as_str(), idx))
+ .collect();
+ let col_mapping: Vec> = merged_schema
+ .iter()
+ .map(|merged_col| {
+ col_name_to_idx
+ .get(merged_col.column_name.as_str())
+ .copied()
+ })
+ .collect();
+ let null_value = Value { value_data: None };
+
+ rows.into_iter()
+ .map(|mut row| {
+ let values = col_mapping
+ .iter()
+ .map(|opt_idx| match opt_idx {
+ Some(idx) => std::mem::take(&mut row.values[*idx]),
+ None => null_value.clone(),
+ })
+ .collect();
+ Row { values }
+ })
+ .collect()
+ }
+
/// Find the physical region id for a logical region.
fn find_physical_region_id(&self, logical_region_id: RegionId) -> Result {
let state = self.state.read().unwrap();
@@ -519,7 +525,7 @@ impl MetricEngineInner {
/// - Check if every column in the request exists in the physical region
/// - Check each column's datatype and semantic type match the physical region's schema
/// - Check the time index column is present
- /// - When `check_fields` is true, check every physical field column is present.
+ /// - When `check_fields` is true, check every logical field column is present.
/// Set this to `false` for delete requests, which legitimately carry only
/// the primary key + timestamp.
async fn verify_rows(
@@ -531,23 +537,29 @@ impl MetricEngineInner {
) -> Result<()> {
// Check if the region exists
let data_region_id = to_data_region_id(physical_region_id);
- let state = self.state.read().unwrap();
- if !state.is_logical_region_exist(logical_region_id) {
- error!("Trying to write to an nonexistent region {logical_region_id}");
- return LogicalRegionNotFoundSnafu {
- region_id: logical_region_id,
+ let (physical_columns, ts_name) = {
+ let state = self.state.read().unwrap();
+ if !state.is_logical_region_exist(logical_region_id) {
+ error!("Trying to write to an nonexistent region {logical_region_id}");
+ return LogicalRegionNotFoundSnafu {
+ region_id: logical_region_id,
+ }
+ .fail();
}
- .fail();
- }
+
+ let physical_state = state
+ .physical_region_states()
+ .get(&data_region_id)
+ .context(PhysicalRegionNotFoundSnafu {
+ region_id: data_region_id,
+ })?;
+ (
+ physical_state.physical_columns().clone(),
+ physical_state.time_index_column_name().to_string(),
+ )
+ };
// Type + semantic check on every column in the request schema.
- let physical_state = state
- .physical_region_states()
- .get(&data_region_id)
- .context(PhysicalRegionNotFoundSnafu {
- region_id: data_region_id,
- })?;
- let physical_columns = physical_state.physical_columns();
for col in &rows.schema {
let info = physical_columns
.get(&col.column_name)
@@ -593,7 +605,6 @@ impl MetricEngineInner {
);
}
- let ts_name = physical_state.time_index_column_name();
ensure!(
rows.schema.iter().any(|col| col.column_name == ts_name),
InvalidRequestSnafu {
@@ -602,17 +613,56 @@ impl MetricEngineInner {
}
);
+ let logical_columns = self
+ .load_logical_columns(physical_region_id, logical_region_id)
+ .await?;
+ let logical_fields = logical_columns
+ .iter()
+ .filter(|col| col.semantic_type == SemanticType::Field)
+ .map(|col| (col.column_schema.name.as_str(), col))
+ .collect::>();
+
+ for col in &rows.schema {
+ if api::helper::is_semantic_type_eq(col.semantic_type, SemanticType::Field) {
+ ensure!(
+ logical_fields.contains_key(col.column_name.as_str()),
+ InvalidRequestSnafu {
+ region_id: logical_region_id,
+ reason: format!(
+ "field column {} does not belong to logical region {logical_region_id}",
+ col.column_name,
+ ),
+ }
+ );
+ }
+ }
+
if check_fields {
- let field_name = physical_state.field_column_name();
- if !rows.schema.iter().any(|col| col.column_name == field_name) {
- let field_meta =
- physical_columns
- .get(field_name)
- .with_context(|| ColumnNotFoundSnafu {
- name: field_name,
- region_id: logical_region_id,
- })?;
- Self::fill_missing_field_column(logical_region_id, field_name, field_meta, rows)?;
+ // Sparse logical writes may omit nullable field columns. Fill them
+ // before the rows are rewritten for the shared physical table.
+ for (field_name, field_meta) in logical_fields {
+ if !rows.schema.iter().any(|col| col.column_name == field_name) {
+ Self::fill_missing_field_column(
+ logical_region_id,
+ field_name,
+ field_meta,
+ rows,
+ )?;
+ }
+ }
+
+ for (field_name, field_meta) in physical_columns
+ .iter()
+ .filter(|(_, col)| col.semantic_type == SemanticType::Field)
+ {
+ if !rows.schema.iter().any(|col| col.column_name == *field_name) {
+ Self::fill_missing_field_column(
+ logical_region_id,
+ field_name,
+ field_meta,
+ rows,
+ )?;
+ }
}
}
@@ -625,6 +675,8 @@ impl MetricEngineInner {
field_meta: &ColumnMetadata,
rows: &mut Rows,
) -> Result<()> {
+ // This is only for schema columns with a concrete default, usually NULL
+ // for field columns from other logical tables sharing this physical table.
ensure!(
!field_meta.column_schema.is_default_impure(),
UnexpectedRequestSnafu {
@@ -715,6 +767,8 @@ 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_recordbatch::RecordBatches;
use datatypes::prelude::ConcreteDataType;
use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema};
@@ -814,6 +868,13 @@ mod tests {
.unwrap();
}
+ fn column_index(rows: &Rows, name: &str) -> usize {
+ rows.schema
+ .iter()
+ .position(|col| col.column_name == name)
+ .unwrap()
+ }
+
async fn run_batch_write_with_schema_variants(
env: &TestEnv,
physical_region_id: RegionId,
@@ -929,6 +990,103 @@ mod tests {
assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 5);
}
+ #[test]
+ fn test_sparse_batch_aligns_mixed_field_order() {
+ let primary_key = PbColumnSchema {
+ column_name: PRIMARY_KEY_COLUMN_NAME.to_string(),
+ datatype: ColumnDataType::Binary as i32,
+ semantic_type: SemanticType::Tag as _,
+ datatype_extension: None,
+ options: None,
+ };
+ let timestamp = PbColumnSchema {
+ column_name: greptime_timestamp().to_string(),
+ datatype: ColumnDataType::TimestampMillisecond as i32,
+ semantic_type: SemanticType::Timestamp as _,
+ datatype_extension: None,
+ options: None,
+ };
+ let value = PbColumnSchema {
+ column_name: greptime_value().to_string(),
+ datatype: ColumnDataType::Float64 as i32,
+ semantic_type: SemanticType::Field as _,
+ datatype_extension: None,
+ options: None,
+ };
+ let histogram = PbColumnSchema {
+ column_name: NATIVE_HISTOGRAM_FIELD.to_string(),
+ datatype: ColumnDataType::Struct as i32,
+ semantic_type: SemanticType::Field as _,
+ datatype_extension: None,
+ options: None,
+ };
+
+ let sample_rows = Rows {
+ schema: vec![
+ primary_key.clone(),
+ timestamp.clone(),
+ value.clone(),
+ histogram.clone(),
+ ],
+ rows: vec![Row {
+ values: vec![
+ ValueData::BinaryValue(vec![1]).into(),
+ ValueData::TimestampMillisecondValue(0).into(),
+ ValueData::F64Value(1.0).into(),
+ Value { value_data: None },
+ ],
+ }],
+ };
+ let histogram_rows = Rows {
+ schema: vec![primary_key, timestamp, histogram, value],
+ rows: vec![Row {
+ values: vec![
+ ValueData::BinaryValue(vec![2]).into(),
+ ValueData::TimestampMillisecondValue(0).into(),
+ ValueData::StructValue(api::v1::StructValue { items: vec![] }).into(),
+ Value { value_data: None },
+ ],
+ }],
+ };
+
+ let schema = MetricEngineInner::build_union_schema([
+ sample_rows.schema.as_slice(),
+ histogram_rows.schema.as_slice(),
+ ]);
+ let merged_rows = MetricEngineInner::align_rows_to_schema(sample_rows, &schema)
+ .into_iter()
+ .chain(MetricEngineInner::align_rows_to_schema(
+ histogram_rows,
+ &schema,
+ ))
+ .collect();
+ let merged_request = Rows {
+ schema,
+ rows: merged_rows,
+ };
+
+ let value_idx = column_index(&merged_request, greptime_value());
+ let histogram_idx = column_index(&merged_request, NATIVE_HISTOGRAM_FIELD);
+ assert!(matches!(
+ merged_request.rows[0].values[value_idx].value_data,
+ Some(ValueData::F64Value(_))
+ ));
+ assert!(
+ merged_request.rows[0].values[histogram_idx]
+ .value_data
+ .is_none()
+ );
+ assert!(
+ merged_request.rows[1].values[value_idx]
+ .value_data
+ .is_none()
+ );
+ assert!(matches!(
+ merged_request.rows[1].values[histogram_idx].value_data,
+ Some(ValueData::StructValue(_))
+ ));
+ }
+
#[tokio::test]
async fn test_write_logical_region() {
let env = TestEnv::new().await;
diff --git a/src/metric-engine/src/engine/state.rs b/src/metric-engine/src/engine/state.rs
index 25416380f12..06d5f19c4bd 100644
--- a/src/metric-engine/src/engine/state.rs
+++ b/src/metric-engine/src/engine/state.rs
@@ -17,7 +17,6 @@
use std::collections::{HashMap, HashSet};
use api::v1::SemanticType;
-use common_telemetry::warn;
use common_time::timestamp::TimeUnit;
use snafu::OptionExt;
use store_api::codec::PrimaryKeyEncoding;
@@ -37,10 +36,6 @@ pub struct PhysicalRegionState {
/// row batch. The time index is fixed at region creation and never
/// changes, so this stays in sync with `physical_columns`.
time_index_column_name: String,
- /// Name of the field column. Metric regions have exactly one field column
- /// verified at creation time, so the write path can validate completeness
- /// without consulting per-logical-region metadata.
- field_column_name: String,
primary_key_encoding: PrimaryKeyEncoding,
options: PhysicalRegionOptions,
time_index_unit: TimeUnit,
@@ -61,16 +56,10 @@ impl PhysicalRegionState {
.find(|(_, meta)| meta.semantic_type == SemanticType::Timestamp)
.map(|(name, _)| name.clone())
.unwrap_or_default();
- let field_column_name = physical_columns
- .iter()
- .find(|(_, meta)| meta.semantic_type == SemanticType::Field)
- .map(|(name, _)| name.clone())
- .unwrap_or_default();
Self {
logical_regions: HashSet::new(),
physical_columns,
time_index_column_name,
- field_column_name,
primary_key_encoding,
options,
time_index_unit,
@@ -92,11 +81,6 @@ impl PhysicalRegionState {
&self.time_index_column_name
}
- /// Returns the cached name of the field column.
- pub fn field_column_name(&self) -> &str {
- &self.field_column_name
- }
-
/// Returns a reference to the physical region options.
pub fn options(&self) -> &PhysicalRegionOptions {
&self.options
@@ -160,12 +144,6 @@ impl MetricEngineState {
SemanticType::Timestamp,
"unexpected time index column {col} added to an existing physical region"
);
- if meta.semantic_type == SemanticType::Field {
- warn!(
- "Unexpected field column {col} added to physical region {physical_region_id}; cached field column remains {}",
- state.field_column_name
- );
- }
state.physical_columns.insert(col, meta);
}
}
diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs
index 080c270f928..a433602a36a 100644
--- a/src/operator/src/insert.rs
+++ b/src/operator/src/insert.rs
@@ -37,6 +37,9 @@ 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::prelude::{greptime_timestamp, greptime_value};
use common_telemetry::tracing_context::TracingContext;
use common_telemetry::{error, info, warn};
@@ -907,6 +910,10 @@ impl Inserter {
let table_name = table.table_info().name.clone();
let request_schema = req.rows.as_ref().unwrap().schema.as_slice();
+ let request_field_count = request_schema
+ .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 {
@@ -915,12 +922,22 @@ impl Inserter {
// If accommodate_existing_schema is true, update request schema for Timestamp/Field columns
if accommodate_existing_schema {
+ let request_is_native_histogram = request_is_native_histogram(request_schema);
+ let table_is_native_histogram = table_is_native_histogram(table);
+ ensure!(
+ request_is_native_histogram == table_is_native_histogram,
+ InvalidInsertRequestSnafu {
+ reason: format!(
+ "Table `{table_name}` cannot mix native histogram and float sample fields"
+ ),
+ }
+ );
let table_schema = table.schema();
// Find timestamp column name
let ts_col_name = table_schema.timestamp_column().map(|c| c.name.clone());
// Find field column name if there is only one and `is_single_value` is true.
let mut field_col_name = None;
- if is_single_value {
+ if is_single_value && request_field_count <= 1 {
let mut multiple_field_cols = false;
table.field_columns().for_each(|col| {
if field_col_name.is_none() {
@@ -1063,6 +1080,32 @@ impl Inserter {
}
}
+fn request_is_native_histogram(request_schema: &[ColumnSchema]) -> bool {
+ let mut fields = request_schema
+ .iter()
+ .filter(|col| col.semantic_type == SemanticType::Field as i32);
+ let Some(col) = fields.next() else {
+ return false;
+ };
+
+ fields.next().is_none()
+ && col.column_name == NATIVE_HISTOGRAM_FIELD
+ && api::helper::is_column_type_value_eq(
+ col.datatype,
+ col.datatype_extension.clone(),
+ native_histogram_value_type(),
+ )
+}
+
+fn table_is_native_histogram(table: &TableRef) -> bool {
+ let mut fields = table.field_columns();
+ let Some(col) = fields.next() else {
+ return false;
+ };
+
+ fields.next().is_none() && is_native_histogram_value_schema(&col.name, &col.data_type)
+}
+
fn validate_column_count_match(requests: &RowInsertRequests) -> Result<()> {
for request in &requests.inserts {
let rows = request.rows.as_ref().unwrap();
diff --git a/src/servers/src/http.rs b/src/servers/src/http.rs
index aa0633ec45f..3713e99454d 100644
--- a/src/servers/src/http.rs
+++ b/src/servers/src/http.rs
@@ -163,6 +163,9 @@ pub struct HttpOptions {
/// Validation mode while decoding Prometheus remote write requests.
pub prom_validation_mode: PromValidationMode,
+ /// Enables experimental Prometheus remote write v2 native histogram ingestion.
+ pub experimental_enable_prometheus_native_histogram: bool,
+
pub cors_allowed_origins: Vec,
pub enable_cors: bool,
@@ -180,6 +183,7 @@ impl Default for HttpOptions {
cors_allowed_origins: Vec::new(),
enable_cors: true,
prom_validation_mode: PromValidationMode::Strict,
+ experimental_enable_prometheus_native_histogram: false,
experimental_enable_explain_analyze_stream: false,
}
}
@@ -601,6 +605,9 @@ impl HttpServerBuilder {
pipeline_handler,
prom_store_with_metric_engine,
prom_validation_mode,
+ experimental_enable_prometheus_native_histogram: self
+ .options
+ .experimental_enable_prometheus_native_histogram,
pending_rows_batcher,
};
diff --git a/src/servers/src/http/prom_store.rs b/src/servers/src/http/prom_store.rs
index 07cd1e19a9a..49e9ab64e64 100644
--- a/src/servers/src/http/prom_store.rs
+++ b/src/servers/src/http/prom_store.rs
@@ -45,7 +45,7 @@ use crate::http::header::{
use crate::pending_rows_batcher::PendingRowsBatcher;
use crate::prom_remote_write::decode::PromSeriesProcessor;
use crate::prom_remote_write::decode_remote_write_request;
-use crate::prom_remote_write::v2::{RemoteWriteV2RequestExt, decode_remote_write_v2_request};
+use crate::prom_remote_write::v2::{decode_remote_write_v2_request, into_write_requests};
use crate::prom_remote_write::validation::PromValidationMode;
use crate::prom_store::{extract_schema_from_read_request, snappy_decompress};
use crate::query_handler::{PipelineHandlerRef, PromStoreProtocolHandlerRef, PromStoreResponse};
@@ -71,6 +71,7 @@ pub struct PromStoreState {
pub pipeline_handler: Option,
pub prom_store_with_metric_engine: bool,
pub prom_validation_mode: PromValidationMode,
+ pub experimental_enable_prometheus_native_histogram: bool,
pub pending_rows_batcher: Option>,
}
@@ -142,6 +143,7 @@ async fn remote_write_v1(
pipeline_handler,
prom_store_with_metric_engine,
prom_validation_mode,
+ experimental_enable_prometheus_native_histogram: _,
pending_rows_batcher,
} = state;
@@ -177,15 +179,22 @@ async fn remote_write_v1(
req.as_insert_requests()
};
- let outcome = write_prometheus_rows(
+ let outcome = match write_prometheus_rows_with_progress(
prom_store_handler,
pending_rows_batcher,
prom_store_with_metric_engine,
- &db,
query_ctx,
req,
)
- .await?;
+ .await
+ {
+ Ok(outcome) => outcome,
+ Err(error) => {
+ record_remote_write_samples(&db, error.rows_written);
+ return Err(error.error);
+ }
+ };
+ record_remote_write_samples(&db, outcome.rows_written);
Ok((
StatusCode::NO_CONTENT,
@@ -207,6 +216,7 @@ async fn remote_write_v2(
pipeline_handler: _,
prom_store_with_metric_engine,
prom_validation_mode: _,
+ experimental_enable_prometheus_native_histogram,
pending_rows_batcher,
} = state;
@@ -225,38 +235,96 @@ async fn remote_write_v2(
Ok(request) => request,
Err(error) => return Ok(remote_write_v2_error_response(error, 0, 0, 0)),
};
- let req = match request.into_context_req() {
+ if !experimental_enable_prometheus_native_histogram && request_has_native_histograms(&request) {
+ return Ok(remote_write_v2_error_response(
+ error::InvalidPromRemoteRequestSnafu {
+ msg: "prometheus remote write v2 native histogram ingestion is experimental; set http.experimental_enable_prometheus_native_histogram = true to enable it"
+ .to_string(),
+ }
+ .build(),
+ 0,
+ 0,
+ 0,
+ ));
+ }
+ let req = match into_write_requests(request) {
Ok(req) => req,
Err(error) => return Ok(remote_write_v2_error_response(error, 0, 0, 0)),
};
- let outcome = match write_prometheus_rows_with_progress(
- prom_store_handler,
- pending_rows_batcher,
- prom_store_with_metric_engine,
- &db,
- query_ctx,
- req,
- )
- .await
- {
- Ok(outcome) => outcome,
- Err(error) => {
- return Ok(remote_write_v2_error_response(
- error.error,
- error.rows_written,
- 0,
- 0,
- ));
+ let outcome = if req.sample_count > 0 {
+ match write_prometheus_rows_with_progress(
+ prom_store_handler.clone(),
+ pending_rows_batcher.clone(),
+ prom_store_with_metric_engine,
+ query_ctx.clone(),
+ req.samples,
+ )
+ .await
+ {
+ Ok(outcome) => outcome,
+ Err(error) => {
+ record_remote_write_samples(&db, error.rows_written);
+ return Ok(remote_write_v2_error_response(
+ error.error,
+ error.rows_written,
+ 0,
+ 0,
+ ));
+ }
+ }
+ } else {
+ PromWriteOutcome {
+ write_cost: 0,
+ rows_written: 0,
}
};
+ let samples_written = outcome.rows_written;
+ record_remote_write_samples(&db, samples_written);
+ let mut histograms_written = 0;
+ let mut write_cost = outcome.write_cost;
- let mut headers = write_cost_header_map(outcome.write_cost);
- append_remote_write_v2_written_headers(&mut headers, outcome.rows_written, 0, 0);
+ if req.histogram_count > 0 {
+ let outcome = match write_prometheus_rows_with_progress(
+ prom_store_handler,
+ None,
+ prom_store_with_metric_engine,
+ query_ctx,
+ req.histograms,
+ )
+ .await
+ {
+ Ok(outcome) => outcome,
+ Err(error) => {
+ record_remote_write_histograms(&db, error.rows_written);
+ return Ok(remote_write_v2_error_response(
+ error.error,
+ samples_written,
+ error.rows_written,
+ 0,
+ ));
+ }
+ };
+ histograms_written = outcome.rows_written;
+ record_remote_write_histograms(&db, histograms_written);
+ write_cost += outcome.write_cost;
+ }
+
+ let mut headers = write_cost_header_map(write_cost);
+ append_remote_write_v2_written_headers(&mut headers, samples_written, histograms_written, 0);
Ok((StatusCode::NO_CONTENT, headers).into_response())
}
+fn request_has_native_histograms(
+ request: &api::greptime_proto::io::prometheus::write::v2::Request,
+) -> bool {
+ request
+ .timeseries
+ .iter()
+ .any(|series| !series.histograms.is_empty())
+}
+
fn vm_proto_version_response(params: &RemoteWriteQuery) -> Option {
params
.get_vm_proto_version
@@ -302,31 +370,14 @@ struct PromWriteError {
rows_written: u64,
}
-async fn write_prometheus_rows(
- prom_store_handler: PromStoreProtocolHandlerRef,
- pending_rows_batcher: Option>,
- prom_store_with_metric_engine: bool,
- db: &str,
- query_ctx: Arc,
- req: ContextReq,
-) -> Result {
- write_prometheus_rows_with_progress(
- prom_store_handler,
- pending_rows_batcher,
- prom_store_with_metric_engine,
- db,
- query_ctx,
- req,
- )
- .await
- .map_err(|error| error.error)
-}
-
+/// Writes one decoded PRW batch and keeps the number of persisted rows on error.
+///
+/// The v2 handler uses that partial progress to return Prometheus' written
+/// sample/histogram headers even when a later table write fails.
async fn write_prometheus_rows_with_progress(
prom_store_handler: PromStoreProtocolHandlerRef,
pending_rows_batcher: Option>,
prom_store_with_metric_engine: bool,
- db: &str,
query_ctx: Arc,
req: ContextReq,
) -> std::result::Result {
@@ -347,9 +398,6 @@ async fn write_prometheus_rows_with_progress(
error,
rows_written,
})?;
- crate::metrics::PROM_STORE_REMOTE_WRITE_SAMPLES
- .with_label_values(&[db])
- .inc_by(rows);
rows_written += rows;
}
return Ok(PromWriteOutcome {
@@ -373,9 +421,6 @@ async fn write_prometheus_rows_with_progress(
error,
rows_written,
})?;
- crate::metrics::PROM_STORE_REMOTE_WRITE_SAMPLES
- .with_label_values(&[db])
- .inc_by(cnt);
write_cost += output.meta.cost;
rows_written += cnt;
}
@@ -386,6 +431,24 @@ async fn write_prometheus_rows_with_progress(
})
}
+fn record_remote_write_samples(db: &str, rows: u64) {
+ if rows == 0 {
+ return;
+ }
+ crate::metrics::PROM_STORE_REMOTE_WRITE_SAMPLES
+ .with_label_values(&[db])
+ .inc_by(rows);
+}
+
+fn record_remote_write_histograms(db: &str, rows: u64) {
+ if rows == 0 {
+ return;
+ }
+ crate::metrics::PROM_STORE_REMOTE_WRITE_HISTOGRAMS
+ .with_label_values(&[db])
+ .inc_by(rows);
+}
+
fn remote_write_v2_error_response(
error: error::Error,
samples: u64,
@@ -647,6 +710,7 @@ mod tests {
pipeline_handler: None,
prom_store_with_metric_engine: false,
prom_validation_mode: PromValidationMode::Strict,
+ experimental_enable_prometheus_native_histogram: false,
pending_rows_batcher: None,
}
}
diff --git a/src/servers/src/metrics.rs b/src/servers/src/metrics.rs
index 6f151db5393..838cc445594 100644
--- a/src/servers/src/metrics.rs
+++ b/src/servers/src/metrics.rs
@@ -128,6 +128,13 @@ lazy_static! {
&[METRIC_DB_LABEL]
)
.unwrap();
+ /// The native histograms count of Prometheus remote write.
+ pub static ref PROM_STORE_REMOTE_WRITE_HISTOGRAMS: IntCounterVec = register_int_counter_vec!(
+ "greptime_servers_prometheus_remote_write_histograms",
+ "frontend prometheus remote write native histograms",
+ &[METRIC_DB_LABEL]
+ )
+ .unwrap();
pub static ref PENDING_BATCHES: IntGauge = register_int_gauge!(
"greptime_prom_store_pending_batches",
"Number of pending batches waiting to be flushed"
diff --git a/src/servers/src/otlp/trace/v1.rs b/src/servers/src/otlp/trace/v1.rs
index cce6891b0dd..e04d344e54e 100644
--- a/src/servers/src/otlp/trace/v1.rs
+++ b/src/servers/src/otlp/trace/v1.rs
@@ -303,7 +303,7 @@ pub(crate) fn write_attributes(
mod tests {
use api::v1::value::ValueData;
use opentelemetry_proto::tonic::common::v1::any_value::Value as OtlpValue;
- use opentelemetry_proto::tonic::common::v1::{AnyValue, KeyValue};
+ use opentelemetry_proto::tonic::common::v1::{AnyValue, ArrayValue, KeyValue};
use super::*;
use crate::otlp::trace::TraceAuxData;
@@ -533,6 +533,42 @@ mod tests {
);
}
+ #[test]
+ fn test_keep_mixed_binary_and_json_values_until_frontend_reconciliation() {
+ let mut writer = TableData::new(4, 2);
+
+ let attrs1 = Attributes::from(vec![make_kv(
+ "val",
+ OtlpValue::BytesValue(vec![1_u8, 2, 3]),
+ )]);
+ let mut row1 = writer.alloc_one_row();
+ write_attributes(&mut writer, "attr", attrs1, &mut row1).unwrap();
+ writer.add_row(row1);
+
+ let attrs2 = Attributes::from(vec![make_kv(
+ "val",
+ OtlpValue::ArrayValue(ArrayValue {
+ values: vec![AnyValue {
+ value: Some(OtlpValue::IntValue(1)),
+ }],
+ }),
+ )]);
+ let mut row2 = writer.alloc_one_row();
+ write_attributes(&mut writer, "attr", attrs2, &mut row2).unwrap();
+ writer.add_row(row2);
+
+ let (schema, rows) = writer.into_schema_and_rows();
+ let col_idx = schema
+ .iter()
+ .position(|c| c.column_name == "attr.val")
+ .unwrap();
+ assert_eq!(schema[col_idx].datatype, ColumnDataType::Binary as i32);
+ assert!(matches!(
+ rows[1].values[col_idx].value_data.as_ref(),
+ Some(ValueData::BinaryValue(_))
+ ));
+ }
+
#[test]
fn test_build_aux_table_requests_deduplicates_services_and_operations() {
let spans = vec![
diff --git a/src/servers/src/prom_remote_write/README.md b/src/servers/src/prom_remote_write/README.md
new file mode 100644
index 00000000000..acac7fd6854
--- /dev/null
+++ b/src/servers/src/prom_remote_write/README.md
@@ -0,0 +1,54 @@
+# Prometheus Remote Write
+
+This module decodes Prometheus remote write requests and converts them into
+Greptime row insert requests.
+
+## Remote Write V2
+
+Remote write v2 enters through `remote_write_v2` in
+`src/servers/src/http/prom_store.rs`.
+
+```mermaid
+flowchart TD
+ A["HTTP /v1/prometheus/write"] --> B["remote_write_v2"]
+ B --> C["decode_remote_write_v2_request"]
+ C --> D["into_write_requests"]
+
+ D --> E["samples ContextReq"]
+ D --> F["histograms ContextReq"]
+
+ E --> G["write_prometheus_rows_with_progress"]
+ G --> H["metric engine / pending batcher when enabled"]
+
+ F --> I["histogram ContextOpt"]
+ I --> J["write_prometheus_rows_with_progress"]
+ J --> K["same metric-engine flag as samples, no batcher"]
+ K --> L["table: "]
+
+ L --> M["field: greptime_native_histogram Struct"]
+ M --> N["struct children: counts, spans, buckets, sum, schema"]
+ H --> P["written headers and counters"]
+ K --> P
+```
+
+The conversion step splits one v2 request into two `ContextReq`s:
+
+- samples keep the existing sample table name and can use the metric-engine
+ physical table path and pending rows batcher;
+- native histograms keep the existing metric table name.
+
+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:
+
+- common scalar children: `schema`, `zero_threshold`, `sum`, `reset_hint`,
+ `start_timestamp`;
+- count children: `count_u64` / `zero_count_u64` or `count_f64` / `zero_count_f64`;
+- list children for custom values, spans, and positive/negative buckets;
+- original Prometheus labels as Greptime tags.
+
+The v2 response always reports written sample, histogram, and exemplar counts in
+Prometheus remote-write headers. Exemplars are currently ignored and reported as
+zero.
diff --git a/src/servers/src/prom_remote_write/v2.rs b/src/servers/src/prom_remote_write/v2.rs
index 8ee84553388..b4128994cef 100644
--- a/src/servers/src/prom_remote_write/v2.rs
+++ b/src/servers/src/prom_remote_write/v2.rs
@@ -13,14 +13,18 @@
// limitations under the License.
use std::collections::hash_map::Entry;
-use std::collections::{HashMap, HashSet};
+use ahash::{HashMap, HashMapExt, HashSet, HashSetExt};
+use api::greptime_proto::io::prometheus::write::v2::histogram::{Count, ZeroCount};
#[cfg(test)]
-use api::greptime_proto::io::prometheus::write::v2::{Exemplar, Histogram, Metadata, metadata};
-use api::greptime_proto::io::prometheus::write::v2::{Request, Sample, TimeSeries};
-use api::v1::{RowInsertRequest, Rows, Value};
+use api::greptime_proto::io::prometheus::write::v2::{Exemplar, Metadata, metadata};
+use api::greptime_proto::io::prometheus::write::v2::{Histogram, Request, Sample, TimeSeries};
+use api::helper::ColumnDataTypeWrapper;
+use api::v1::value::ValueData;
+use api::v1::{ColumnSchema, ListValue, RowInsertRequest, Rows, SemanticType, Value};
use bytes::Bytes;
use common_grpc::precision::Precision;
+use common_query::native_histogram::*;
use common_query::prelude::{greptime_timestamp, greptime_value};
use pipeline::{ContextOpt, ContextReq};
use prost::Message;
@@ -29,6 +33,7 @@ use snafu::{OptionExt, ResultExt, ensure};
use crate::error::{self, Result};
use crate::prom_remote_write::row_builder::PromCtx;
use crate::prom_remote_write::try_decompress;
+use crate::prom_remote_write::validation::validate_label_name;
#[allow(deprecated)]
use crate::prom_store::{
DATABASE_LABEL, DATABASE_LABEL_ALT, METRIC_NAME_LABEL, PHYSICAL_TABLE_LABEL,
@@ -53,49 +58,132 @@ pub(crate) fn decode_remote_write_v2_request(is_zstd: bool, body: Bytes) -> Resu
Request::decode(&buf[..]).context(error::DecodePromRemoteRequestSnafu)
}
-pub(crate) trait RemoteWriteV2RequestExt {
- fn into_context_req(self) -> Result;
+pub(crate) struct RemoteWriteV2WriteRequests {
+ pub samples: ContextReq,
+ pub histograms: ContextReq,
+ pub sample_count: u64,
+ pub histogram_count: u64,
}
-impl RemoteWriteV2RequestExt for Request {
- fn into_context_req(self) -> Result {
- let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_CONVERT_ELAPSED.start_timer();
- let Request {
- symbols,
- timeseries,
- } = self;
+/// Converts a PRW v2 request into normal sample writes and native histogram writes.
+///
+/// A metric name may appear in only one payload kind per request because the
+/// metric-engine logical table is either a float metric or a native histogram.
+pub(crate) fn into_write_requests(request: Request) -> Result {
+ let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_CONVERT_ELAPSED.start_timer();
+ let Request {
+ symbols,
+ timeseries,
+ } = request;
- ensure!(
- symbols.first().map(|s| s.as_str()) == Some(""),
- error::InvalidPromRemoteRequestSnafu {
- msg: "remote write v2 symbols must start with an empty string".to_string(),
- }
- );
+ ensure!(
+ symbols.first().map(|s| s.as_str()) == Some(""),
+ error::InvalidPromRemoteRequestSnafu {
+ msg: "remote write v2 symbols must start with an empty string".to_string(),
+ }
+ );
- let mut tables = HashMap::>::new();
+ let mut sample_tables = HashMap::>::new();
+ let mut histogram_tables = HashMap::>::new();
+ let mut sample_metrics = HashSet::<(PromCtx, String)>::new();
+ let mut histogram_metrics = HashSet::<(PromCtx, String)>::new();
+ let mut sample_count_total = 0;
+ let mut histogram_count_total = 0;
- for series in timeseries {
- // Native histograms and exemplars are intentionally ignored for now.
- // They will be converted into Greptime rows once their ingestion path is implemented.
- let sample_count = series.samples.len();
- if sample_count == 0 {
- continue;
- }
-
- let (prom_ctx, table_name, tags) = resolve_series_labels(&symbols, &series)?;
- let table_data = match tables.entry(prom_ctx).or_default().entry(table_name) {
- Entry::Occupied(entry) => {
- let table_data = entry.into_mut();
- table_data.reserve_rows(sample_count);
- table_data
- }
- Entry::Vacant(entry) => entry.insert(TableData::new(tags.len() + 2, sample_count)),
- };
-
- write_samples(table_data, series.samples, tags)?;
+ for series in timeseries {
+ // Exemplars are intentionally ignored for now.
+ let sample_count = series.samples.len();
+ let histogram_count = series.histograms.len();
+ if sample_count == 0 && histogram_count == 0 {
+ continue;
}
- Ok(into_context_req(tables))
+ let (prom_ctx, table_name, tags) = resolve_series_labels(&symbols, &series)?;
+ ensure_no_internal_histogram_labels(&tags)?;
+ let metric_key = (prom_ctx.clone(), table_name.clone());
+ if sample_count > 0 {
+ ensure!(
+ !histogram_metrics.contains(&metric_key),
+ error::InvalidPromRemoteRequestSnafu {
+ msg: format!(
+ "remote write v2 metric `{table_name}` contains both samples and native histograms"
+ ),
+ }
+ );
+ sample_metrics.insert(metric_key.clone());
+ }
+ if histogram_count > 0 {
+ ensure!(
+ !sample_metrics.contains(&metric_key),
+ error::InvalidPromRemoteRequestSnafu {
+ msg: format!(
+ "remote write v2 metric `{table_name}` contains both samples and native histograms"
+ ),
+ }
+ );
+ histogram_metrics.insert(metric_key);
+ }
+
+ if sample_count > 0 && histogram_count == 0 {
+ // Fast path for regular sample-only series. Move the resolved labels into
+ // the sample writer instead of cloning them for a histogram path we won't use.
+ let table_data = get_or_create_table_data(
+ &mut sample_tables,
+ prom_ctx,
+ table_name,
+ tags.len() + 2,
+ sample_count,
+ );
+
+ write_samples(table_data, series.samples, tags)?;
+ sample_count_total += sample_count as u64;
+ // The owned labels were moved above, so skip the mixed-series path below.
+ continue;
+ }
+
+ if histogram_count > 0 {
+ let table_data = get_or_create_table_data(
+ &mut histogram_tables,
+ prom_ctx,
+ table_name,
+ tags.len() + 2,
+ histogram_count,
+ );
+
+ let mut histograms = series.histograms;
+ let Some(last_histogram) = histograms.pop() else {
+ continue;
+ };
+ for histogram in &histograms {
+ write_native_histogram(table_data, histogram, tags.iter().cloned())?;
+ }
+ write_native_histogram(table_data, &last_histogram, tags.into_iter())?;
+ histogram_count_total += histogram_count as u64;
+ }
+ }
+
+ Ok(RemoteWriteV2WriteRequests {
+ samples: into_context_req(sample_tables),
+ histograms: into_context_req(histogram_tables),
+ sample_count: sample_count_total,
+ histogram_count: histogram_count_total,
+ })
+}
+
+fn get_or_create_table_data(
+ tables: &mut HashMap>,
+ prom_ctx: PromCtx,
+ table_name: String,
+ column_count: usize,
+ row_count: usize,
+) -> &mut TableData {
+ match tables.entry(prom_ctx).or_default().entry(table_name) {
+ Entry::Occupied(entry) => {
+ let table_data = entry.into_mut();
+ table_data.reserve_rows(row_count);
+ table_data
+ }
+ Entry::Vacant(entry) => entry.insert(TableData::new(column_count, row_count)),
}
}
@@ -135,6 +223,195 @@ fn write_sample(
Ok(())
}
+fn write_native_histogram(
+ table_data: &mut TableData,
+ histogram: &Histogram,
+ tags: impl Iterator- ,
+) -> Result<()> {
+ // Persist both int and float families into the logical table schema. Only one
+ // family is populated per row; the other is written as NULL so PromQL can
+ // infer the original histogram flavor without a separate type column.
+ let mut row = table_data.alloc_one_row();
+ row_writer::write_ts_to_millis(
+ table_data,
+ greptime_timestamp(),
+ Some(histogram.timestamp),
+ Precision::Millisecond,
+ &mut row,
+ )?;
+
+ write_native_histogram_value(table_data, histogram, &mut row)?;
+
+ row_writer::write_tags(table_data, tags, &mut row)?;
+ table_data.add_row(row);
+
+ Ok(())
+}
+
+fn write_native_histogram_value(
+ table_data: &mut TableData,
+ histogram: &Histogram,
+ row: &mut Vec,
+) -> Result<()> {
+ let column_schema = native_histogram_column_schema();
+ let value = native_histogram_struct_value(histogram)?;
+
+ row_writer::write_by_schema(
+ table_data,
+ std::iter::once((column_schema, Some(value))),
+ row,
+ )
+}
+
+fn native_histogram_column_schema() -> ColumnSchema {
+ let (datatype, datatype_extension) =
+ ColumnDataTypeWrapper::try_from(native_histogram_value_type().clone())
+ .expect("native histogram type is convertible to protobuf")
+ .into_parts();
+
+ ColumnSchema {
+ column_name: NATIVE_HISTOGRAM_FIELD.to_string(),
+ datatype: datatype as i32,
+ semantic_type: SemanticType::Field as i32,
+ datatype_extension,
+ options: None,
+ }
+}
+
+fn native_histogram_struct_value(histogram: &Histogram) -> Result {
+ let mut items = Vec::with_capacity(NATIVE_HISTOGRAM_FIELD_NAMES.len());
+ items.extend([
+ pb_value(ValueData::I32Value(histogram.schema)),
+ pb_value(ValueData::F64Value(histogram.zero_threshold)),
+ pb_value(ValueData::F64Value(histogram.sum)),
+ pb_value(ValueData::I32Value(histogram.reset_hint)),
+ optional_pb_value((histogram.start_timestamp != 0).then_some(
+ ValueData::TimestampMillisecondValue(histogram.start_timestamp),
+ )),
+ f64_list_value(histogram.custom_values.iter().copied()),
+ i32_list_value(histogram.positive_spans.iter().map(|span| span.offset)),
+ u32_list_value(histogram.positive_spans.iter().map(|span| span.length)),
+ i32_list_value(histogram.negative_spans.iter().map(|span| span.offset)),
+ u32_list_value(histogram.negative_spans.iter().map(|span| span.length)),
+ ]);
+
+ let int_counts = match (&histogram.count, &histogram.zero_count) {
+ (Some(Count::CountInt(count)), Some(ZeroCount::ZeroCountInt(zero_count))) => {
+ (*count, *zero_count)
+ }
+ (Some(Count::CountInt(count)), _) => (*count, 0),
+ (_, Some(ZeroCount::ZeroCountInt(zero_count))) => (0, *zero_count),
+ _ => (0, 0),
+ };
+ let float_counts = match (&histogram.count, &histogram.zero_count) {
+ (Some(Count::CountFloat(count)), Some(ZeroCount::ZeroCountFloat(zero_count))) => {
+ Some((*count, *zero_count))
+ }
+ (Some(Count::CountFloat(count)), _) => Some((*count, 0.0)),
+ _ => None,
+ };
+
+ if let Some(counts) = float_counts {
+ items.extend([
+ null_pb_value(),
+ null_pb_value(),
+ i64_list_value(std::iter::empty()),
+ i64_list_value(std::iter::empty()),
+ pb_value(ValueData::F64Value(counts.0)),
+ pb_value(ValueData::F64Value(counts.1)),
+ f64_list_value(histogram.positive_counts.iter().copied()),
+ f64_list_value(histogram.negative_counts.iter().copied()),
+ ]);
+ } else {
+ let positive_buckets = bucket_counts_from_deltas(&histogram.positive_deltas)?;
+ let negative_buckets = bucket_counts_from_deltas(&histogram.negative_deltas)?;
+ items.extend([
+ pb_value(ValueData::U64Value(int_counts.0)),
+ pb_value(ValueData::U64Value(int_counts.1)),
+ i64_list_value(positive_buckets.iter().copied()),
+ i64_list_value(negative_buckets.iter().copied()),
+ null_pb_value(),
+ null_pb_value(),
+ f64_list_value(std::iter::empty()),
+ f64_list_value(std::iter::empty()),
+ ]);
+ }
+
+ Ok(ValueData::StructValue(api::v1::StructValue { items }))
+}
+
+fn pb_value(value_data: ValueData) -> Value {
+ optional_pb_value(Some(value_data))
+}
+
+fn null_pb_value() -> Value {
+ optional_pb_value(None)
+}
+
+fn optional_pb_value(value_data: Option) -> Value {
+ Value { value_data }
+}
+
+fn list_value(values: impl IntoIterator
- ) -> Value {
+ pb_value(ValueData::ListValue(ListValue {
+ items: values.into_iter().map(pb_value).collect(),
+ }))
+}
+
+fn i32_list_value(values: impl IntoIterator
- ) -> Value {
+ list_value(values.into_iter().map(ValueData::I32Value))
+}
+
+fn u32_list_value(values: impl IntoIterator
- ) -> Value {
+ list_value(values.into_iter().map(ValueData::U32Value))
+}
+
+fn i64_list_value(values: impl IntoIterator
- ) -> Value {
+ list_value(values.into_iter().map(ValueData::I64Value))
+}
+
+fn f64_list_value(values: impl IntoIterator
- ) -> Value {
+ list_value(values.into_iter().map(ValueData::F64Value))
+}
+
+fn bucket_counts_from_deltas(deltas: &[i64]) -> Result> {
+ let mut count = 0_i64;
+ let mut buckets = Vec::with_capacity(deltas.len());
+
+ for delta in deltas {
+ count = count
+ .checked_add(*delta)
+ .context(error::InvalidPromRemoteRequestSnafu {
+ msg: "remote write v2 native histogram bucket count overflows i64".to_string(),
+ })?;
+ ensure!(
+ count >= 0,
+ error::InvalidPromRemoteRequestSnafu {
+ msg: "remote write v2 native histogram bucket count is negative".to_string(),
+ }
+ );
+ buckets.push(count);
+ }
+
+ Ok(buckets)
+}
+
+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,
+ error::InvalidPromRemoteRequestSnafu {
+ msg: format!(
+ "remote write v2 label `{name}` conflicts with an internal native histogram label"
+ ),
+ }
+ );
+ }
+
+ Ok(())
+}
+
fn resolve_series_labels(symbols: &[String], series: &TimeSeries) -> Result {
ensure!(
series.labels_refs.len().is_multiple_of(2),
@@ -151,7 +428,7 @@ fn resolve_series_labels(symbols: &[String], series: &TimeSeries) -> Result Result Result<()> {
+fn validate_label(name: &str) -> Result<()> {
ensure!(
- !name.is_empty(),
+ validate_label_name(name.as_bytes()),
error::InvalidPromRemoteRequestSnafu {
- msg: "remote write v2 label names must not be empty".to_string(),
- }
- );
- ensure!(
- !value.is_empty(),
- error::InvalidPromRemoteRequestSnafu {
- msg: format!("remote write v2 label `{name}` value must not be empty"),
+ msg: format!("remote write v2 invalid label name `{name}`"),
}
);
@@ -266,6 +543,10 @@ fn table_data_to_row_insert_request(table_name: String, table_data: TableData) -
#[cfg(any(test, feature = "testing"))]
pub mod test_util {
use api::greptime_proto::io::prometheus::write::v2::{Histogram, Request, Sample, TimeSeries};
+ use api::v1::RowInsertRequest;
+ use bytes::Bytes;
+
+ use crate::error::Result;
pub fn request_with_labels_and_samples(
labels: Vec<(&str, &str)>,
@@ -281,6 +562,22 @@ pub mod test_util {
request_with_labels(labels, Vec::new(), histograms)
}
+ pub fn decode_request(is_zstd: bool, body: Bytes) -> Result {
+ super::decode_remote_write_v2_request(is_zstd, body)
+ }
+
+ pub fn write_requests(
+ request: Request,
+ ) -> Result<(Vec, Vec, u64, u64)> {
+ let requests = super::into_write_requests(request)?;
+ Ok((
+ requests.samples.all_req().collect(),
+ requests.histograms.all_req().collect(),
+ requests.sample_count,
+ requests.histogram_count,
+ ))
+ }
+
pub fn histogram(timestamp: i64) -> Histogram {
Histogram {
timestamp,
@@ -375,7 +672,7 @@ mod tests {
#[test]
fn test_into_context_req_samples() {
- let ctx_req = test_util::request_with_labels_and_samples(
+ let ctx_req = into_write_requests(test_util::request_with_labels_and_samples(
vec![
(METRIC_NAME_LABEL, "http_requests_total"),
("job", "api"),
@@ -393,11 +690,13 @@ mod tests {
start_timestamp: 0,
},
],
- )
- .into_context_req()
+ ))
.unwrap();
- let mut inserts = ctx_req.all_req().collect::>();
+ assert_eq!(ctx_req.sample_count, 2);
+ assert_eq!(ctx_req.histogram_count, 0);
+ assert_eq!(ctx_req.histograms.all_req().count(), 0);
+ let mut inserts = ctx_req.samples.all_req().collect::>();
assert_eq!(inserts.len(), 1);
let request = inserts.pop().unwrap();
@@ -437,51 +736,9 @@ mod tests {
);
}
- #[test]
- fn test_into_context_req_accepts_utf8_label_names() {
- let ctx_req = test_util::request_with_labels_and_samples(
- vec![
- (METRIC_NAME_LABEL, "http_requests_total"),
- ("service.name", "api"),
- ("区域", "华东"),
- ],
- vec![Sample {
- value: 42.0,
- timestamp: 1000,
- start_timestamp: 0,
- }],
- )
- .into_context_req()
- .unwrap();
-
- let mut inserts = ctx_req.all_req().collect::>();
- assert_eq!(inserts.len(), 1);
- let rows = inserts.pop().unwrap().rows.unwrap();
- assert_eq!(
- rows.schema
- .iter()
- .map(|col| col.column_name.as_str())
- .collect::>(),
- vec![
- greptime_timestamp(),
- greptime_value(),
- "service.name",
- "区域"
- ]
- );
- assert_eq!(
- rows.rows[0].values[2].value_data,
- Some(ValueData::StringValue("api".to_string()))
- );
- assert_eq!(
- rows.rows[0].values[3].value_data,
- Some(ValueData::StringValue("华东".to_string()))
- );
- }
-
#[test]
fn test_into_context_req_special_labels() {
- let ctx_req = test_util::request_with_labels_and_samples(
+ let ctx_req = into_write_requests(test_util::request_with_labels_and_samples(
vec![
(METRIC_NAME_LABEL, "cpu_usage"),
(DATABASE_LABEL, "tenant_a"),
@@ -493,11 +750,12 @@ mod tests {
timestamp: 1000,
start_timestamp: 0,
}],
- )
- .into_context_req()
+ ))
.unwrap();
- let mut iter = ctx_req.as_req_iter(Arc::new(QueryContext::with("greptime", "public")));
+ let mut iter = ctx_req
+ .samples
+ .as_req_iter(Arc::new(QueryContext::with("greptime", "public")));
let (ctx, reqs) = iter.next().unwrap();
assert!(iter.next().is_none());
@@ -565,13 +823,40 @@ mod tests {
cases.push((
"empty label name",
request_with_sample(vec![(METRIC_NAME_LABEL, "metric"), ("", "api")]),
- "label names must not be empty",
+ "invalid label name",
));
cases.push((
- "empty label value",
- request_with_sample(vec![(METRIC_NAME_LABEL, "metric"), ("job", "")]),
- "label `job` value must not be empty",
+ "invalid label name",
+ request_with_sample(vec![(METRIC_NAME_LABEL, "metric"), ("has-dash", "api")]),
+ "invalid label name",
+ ));
+
+ cases.push((
+ "dotted label name",
+ request_with_sample(vec![(METRIC_NAME_LABEL, "metric"), ("service.name", "api")]),
+ "invalid label name",
+ ));
+
+ cases.push((
+ "non-ascii label name",
+ request_with_sample(vec![(METRIC_NAME_LABEL, "metric"), ("区域", "api")]),
+ "invalid label name",
+ ));
+
+ cases.push((
+ "empty metric name",
+ request_with_sample(vec![(METRIC_NAME_LABEL, "")]),
+ "label `__name__` value must not be empty",
+ ));
+
+ cases.push((
+ "internal histogram label on samples",
+ request_with_sample(vec![
+ (METRIC_NAME_LABEL, "metric"),
+ (NATIVE_HISTOGRAM_FIELD, "user_value"),
+ ]),
+ "conflicts with an internal native histogram label",
));
for (name, request, expected) in cases {
@@ -580,7 +865,27 @@ mod tests {
}
#[test]
- fn test_into_context_req_ignores_histograms_and_exemplars() {
+ fn test_into_context_req_allows_empty_label_values() {
+ let ctx_req = into_write_requests(test_util::request_with_labels_and_samples(
+ vec![(METRIC_NAME_LABEL, "metric"), ("job", "")],
+ vec![Sample {
+ value: 1.0,
+ timestamp: 1000,
+ start_timestamp: 0,
+ }],
+ ))
+ .unwrap();
+
+ let rows = ctx_req.samples.all_req().next().unwrap().rows.unwrap();
+ let job_idx = column_index(&rows.schema, "job");
+ assert_eq!(
+ rows.rows[0].values[job_idx].value_data,
+ Some(ValueData::StringValue(String::new()))
+ );
+ }
+
+ #[test]
+ fn test_into_context_req_rejects_same_metric_samples_and_histograms() {
let mut request = test_util::request_with_labels_and_samples(
vec![(METRIC_NAME_LABEL, "metric")],
vec![Sample {
@@ -590,22 +895,213 @@ mod tests {
}],
);
request.timeseries[0].histograms.push(Histogram::default());
- request.timeseries[0].exemplars.push(Exemplar::default());
- let ctx_req = request.into_context_req().unwrap();
-
- assert_eq!(ctx_req.all_req().count(), 1);
+ assert_invalid(
+ "same metric samples and histograms",
+ request,
+ "contains both samples and native histograms",
+ );
}
#[test]
- fn test_into_context_req_skips_histogram_only_series() {
+ fn test_into_context_req_converts_histograms_and_ignores_exemplars() {
+ let request = Request {
+ symbols: vec![
+ "".to_string(),
+ METRIC_NAME_LABEL.to_string(),
+ "sample_metric".to_string(),
+ "histogram_metric".to_string(),
+ ],
+ timeseries: vec![
+ TimeSeries {
+ labels_refs: vec![1, 2],
+ samples: vec![Sample {
+ value: 1.0,
+ timestamp: 1000,
+ start_timestamp: 0,
+ }],
+ ..Default::default()
+ },
+ TimeSeries {
+ labels_refs: vec![1, 3],
+ histograms: vec![Histogram::default()],
+ exemplars: vec![Exemplar::default()],
+ ..Default::default()
+ },
+ ],
+ };
+
+ let ctx_req = into_write_requests(request).unwrap();
+
+ assert_eq!(ctx_req.sample_count, 1);
+ assert_eq!(ctx_req.histogram_count, 1);
+ assert_eq!(ctx_req.samples.all_req().count(), 1);
+ assert_eq!(ctx_req.histograms.all_req().count(), 1);
+ }
+
+ #[test]
+ fn test_into_context_req_converts_histogram_only_series() {
let mut request =
test_util::request_with_labels_and_samples(vec![(METRIC_NAME_LABEL, "metric")], vec![]);
request.timeseries[0].histograms.push(Histogram::default());
- let ctx_req = request.into_context_req().unwrap();
+ let ctx_req = into_write_requests(request).unwrap();
- assert_eq!(ctx_req.all_req().count(), 0);
+ assert_eq!(ctx_req.sample_count, 0);
+ assert_eq!(ctx_req.histogram_count, 1);
+ assert_eq!(ctx_req.samples.all_req().count(), 0);
+ let mut inserts = ctx_req.histograms.all_req().collect::>();
+ assert_eq!(inserts.len(), 1);
+
+ let request = inserts.pop().unwrap();
+ assert_eq!(request.table_name, "metric");
+ let rows = request.rows.unwrap();
+ assert_eq!(rows.rows.len(), 1);
+ assert_eq!(
+ rows.schema
+ .iter()
+ .map(|col| col.column_name.as_str())
+ .collect::>(),
+ vec![greptime_timestamp(), NATIVE_HISTOGRAM_FIELD]
+ );
+ assert_eq!(
+ rows.rows[0].values[0].value_data,
+ Some(ValueData::TimestampMillisecondValue(0))
+ );
+ assert_eq!(
+ histogram_field_value(&rows, 0, SCHEMA_FIELD),
+ Some(ValueData::I32Value(0))
+ );
+ assert_eq!(
+ histogram_field_value(&rows, 0, COUNT_U64_FIELD),
+ Some(ValueData::U64Value(0))
+ );
+ assert_eq!(histogram_field_value(&rows, 0, COUNT_F64_FIELD), None);
+ }
+
+ #[test]
+ fn test_into_context_req_preserves_histogram_start_timestamp() {
+ let ctx_req = into_write_requests(test_util::request_with_labels_and_histograms(
+ vec![(METRIC_NAME_LABEL, "metric")],
+ vec![Histogram {
+ timestamp: 2000,
+ start_timestamp: 1000,
+ ..Default::default()
+ }],
+ ))
+ .unwrap();
+
+ let mut inserts = ctx_req.histograms.all_req().collect::>();
+ let rows = inserts.pop().unwrap().rows.unwrap();
+
+ assert_eq!(
+ histogram_field_value(&rows, 0, START_TIMESTAMP_FIELD),
+ Some(ValueData::TimestampMillisecondValue(1000))
+ );
+ }
+
+ #[test]
+ fn test_into_context_req_rejects_internal_histogram_labels() {
+ let mut request = test_util::request_with_labels_and_samples(
+ vec![
+ (METRIC_NAME_LABEL, "metric"),
+ (NATIVE_HISTOGRAM_FIELD, "user_value"),
+ ],
+ vec![],
+ );
+ request.timeseries[0].histograms.push(Histogram::default());
+
+ let err = match into_write_requests(request) {
+ Ok(_) => panic!("expected invalid request error"),
+ Err(err) => err,
+ };
+ assert_eq!(
+ err.to_string(),
+ "Invalid prometheus remote request, msg: remote write v2 label `greptime_native_histogram` 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 {
+ count: Some(api::greptime_proto::io::prometheus::write::v2::histogram::Count::CountFloat(3.5)),
+ zero_count: Some(
+ api::greptime_proto::io::prometheus::write::v2::histogram::ZeroCount::ZeroCountFloat(
+ 0.5,
+ ),
+ ),
+ positive_counts: vec![2.0, 3.5],
+ positive_spans: vec![api::greptime_proto::io::prometheus::write::v2::BucketSpan {
+ offset: 3,
+ length: 2,
+ }],
+ timestamp: 2000,
+ ..Default::default()
+ };
+ let request = Request {
+ symbols: vec![
+ "".to_string(),
+ METRIC_NAME_LABEL.to_string(),
+ "metric".to_string(),
+ ],
+ timeseries: vec![
+ TimeSeries {
+ labels_refs: vec![1, 2],
+ histograms: vec![test_util::histogram(1000)],
+ ..Default::default()
+ },
+ TimeSeries {
+ labels_refs: vec![1, 2],
+ histograms: vec![float_histogram],
+ ..Default::default()
+ },
+ ],
+ };
+
+ let ctx_req = into_write_requests(request).unwrap();
+
+ assert_eq!(ctx_req.histogram_count, 2);
+ let mut inserts = ctx_req.histograms.all_req().collect::>();
+ assert_eq!(inserts.len(), 1);
+ let rows = inserts.pop().unwrap().rows.unwrap();
+ assert_eq!(rows.rows.len(), 2);
+ assert_eq!(
+ rows.schema
+ .iter()
+ .map(|col| col.column_name.as_str())
+ .collect::>(),
+ vec![greptime_timestamp(), NATIVE_HISTOGRAM_FIELD]
+ );
+
+ assert_eq!(
+ histogram_field_value(&rows, 0, COUNT_U64_FIELD),
+ Some(ValueData::U64Value(0))
+ );
+ assert_eq!(histogram_field_value(&rows, 0, COUNT_F64_FIELD), None);
+ assert!(matches!(
+ histogram_field_value(&rows, 0, POSITIVE_BUCKETS_I64_FIELD),
+ Some(ValueData::ListValue(_))
+ ));
+ assert!(is_empty_list(histogram_field_value(
+ &rows,
+ 0,
+ POSITIVE_BUCKETS_F64_FIELD
+ )));
+
+ assert_eq!(histogram_field_value(&rows, 1, COUNT_U64_FIELD), None);
+ assert_eq!(
+ histogram_field_value(&rows, 1, COUNT_F64_FIELD),
+ Some(ValueData::F64Value(3.5))
+ );
+ assert!(is_empty_list(histogram_field_value(
+ &rows,
+ 1,
+ POSITIVE_BUCKETS_I64_FIELD
+ )));
+ assert!(matches!(
+ histogram_field_value(&rows, 1, POSITIVE_BUCKETS_F64_FIELD),
+ Some(ValueData::ListValue(_))
+ ));
}
fn request_with_sample(labels: Vec<(&str, &str)>) -> Request {
@@ -620,7 +1116,10 @@ mod tests {
}
fn assert_invalid(name: &str, request: Request, expected: &str) {
- let err = request.into_context_req().unwrap_err();
+ let err = match into_write_requests(request) {
+ Ok(_) => panic!("{name}: expected invalid request error"),
+ Err(err) => err,
+ };
assert!(
matches!(err, error::Error::InvalidPromRemoteRequest { .. }),
"{name}: expected invalid request error, got {err}"
@@ -630,4 +1129,29 @@ mod tests {
"{name}: expected error containing {expected:?}, got {err}"
);
}
+
+ fn column_index(schema: &[ColumnSchema], column_name: &str) -> usize {
+ schema
+ .iter()
+ .position(|column| column.column_name == column_name)
+ .unwrap()
+ }
+
+ fn histogram_field_value(rows: &Rows, row_idx: usize, field_name: &str) -> Option {
+ let histogram_idx = column_index(&rows.schema, NATIVE_HISTOGRAM_FIELD);
+ let Some(ValueData::StructValue(histogram)) =
+ &rows.rows[row_idx].values[histogram_idx].value_data
+ else {
+ panic!("expected native histogram struct value");
+ };
+ let field_idx = NATIVE_HISTOGRAM_FIELD_NAMES
+ .iter()
+ .position(|name| *name == field_name)
+ .unwrap();
+ histogram.items[field_idx].value_data.clone()
+ }
+
+ fn is_empty_list(value: Option) -> bool {
+ matches!(value, Some(ValueData::ListValue(list)) if list.items.is_empty())
+ }
}
diff --git a/src/servers/src/row_writer.rs b/src/servers/src/row_writer.rs
index d92ebc8367f..fb3351323c9 100644
--- a/src/servers/src/row_writer.rs
+++ b/src/servers/src/row_writer.rs
@@ -263,7 +263,7 @@ pub fn write_json(
)
}
-fn write_by_schema(
+pub(crate) fn write_by_schema(
table_data: &mut TableData,
kv_iter: impl Iterator
- )>,
one_row: &mut Vec,
diff --git a/src/servers/tests/http/prom_store_test.rs b/src/servers/tests/http/prom_store_test.rs
index 0de22f34e63..47969d3ef94 100644
--- a/src/servers/tests/http/prom_store_test.rs
+++ b/src/servers/tests/http/prom_store_test.rs
@@ -28,7 +28,8 @@ use async_trait::async_trait;
use axum::Router;
use axum::http::HeaderMap;
use common_query::Output;
-use common_query::prelude::{greptime_timestamp, greptime_value};
+use common_query::native_histogram::NATIVE_HISTOGRAM_FIELD;
+use common_query::prelude::{GREPTIME_PHYSICAL_TABLE, greptime_timestamp, greptime_value};
use common_test_util::ports;
use datafusion_expr::LogicalPlan;
use prost::Message;
@@ -62,6 +63,7 @@ struct DummyInstance {
struct RemoteWriteCapture {
schema: String,
physical_table: Option,
+ with_metric_engine: bool,
request: RowInsertRequests,
}
@@ -71,7 +73,7 @@ impl PromStoreProtocolHandler for DummyInstance {
&self,
request: RowInsertRequests,
ctx: QueryContextRef,
- _with_metric_engine: bool,
+ with_metric_engine: bool,
) -> Result