feat: implement native histogram read

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

Cargo.lock

@@ -11021,6 +11021,7 @@ dependencies = [
  "bytemuck",
  "common-error",
  "common-macro",
+ "common-query",
  "common-recordbatch",
  "common-telemetry",
  "criterion 0.7.0",

src/common/query/src/lib.rs

@@ -17,6 +17,7 @@ pub mod error;
 pub mod logical_plan;
 pub mod native_histogram;
 pub mod prelude;
+pub mod promql_annotations;
 pub mod request;
 pub mod stream;
 #[cfg(any(test, feature = "testing"))]
@@ -32,6 +33,8 @@ use datafusion::physical_plan::ExecutionPlan;
 use serde::{Deserialize, Serialize};
 use sqlparser_derive::{Visit, VisitMut};
 
+use crate::promql_annotations::PromqlAnnotationCollector;
+
 /// new Output struct with output data(previously Output) and output meta
 #[derive(Debug)]
 pub struct Output {
@@ -69,6 +72,9 @@ pub struct OutputMeta {
     /// May exist for query output. One can retrieve execution metrics from this plan.
     pub plan: Option<Arc<dyn ExecutionPlan>>,
     pub cost: OutputCost,
+    pub warnings: Vec<String>,
+    pub infos: Vec<String>,
+    pub promql_annotations: Option<PromqlAnnotationCollector>,
 }
 
 impl Output {
@@ -121,18 +127,47 @@ impl Debug for OutputData {
 
 impl OutputMeta {
     pub fn new(plan: Option<Arc<dyn ExecutionPlan>>, cost: usize) -> Self {
-        Self { plan, cost }
+        Self {
+            plan,
+            cost,
+            warnings: Vec::new(),
+            infos: Vec::new(),
+            promql_annotations: None,
+        }
     }
 
     pub fn new_with_plan(plan: Arc<dyn ExecutionPlan>) -> Self {
         Self {
             plan: Some(plan),
             cost: 0,
+            warnings: Vec::new(),
+            infos: Vec::new(),
+            promql_annotations: None,
         }
     }
 
     pub fn new_with_cost(cost: usize) -> Self {
-        Self { plan: None, cost }
+        Self {
+            plan: None,
+            cost,
+            warnings: Vec::new(),
+            infos: Vec::new(),
+            promql_annotations: None,
+        }
+    }
+
+    pub fn with_promql_annotations(
+        mut self,
+        promql_annotations: Option<PromqlAnnotationCollector>,
+    ) -> Self {
+        self.promql_annotations = promql_annotations;
+        self
+    }
+
+    pub fn collect_promql_annotations(&mut self) {
+        if let Some(collector) = &self.promql_annotations {
+            collector.append_to(&mut self.warnings, &mut self.infos);
+        }
     }
 }
 

src/common/query/src/promql_annotations.rs

@@ -0,0 +1,170 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::{BTreeSet, HashMap};
+use std::sync::{Arc, Mutex, Weak};
+
+use datafusion::config::{ConfigEntry, ConfigExtension, ExtensionOptions};
+use datafusion_common::{DataFusionError, Result as DfResult};
+use once_cell::sync::Lazy;
+
+type PromqlAnnotationStateRef = Mutex<PromqlAnnotationState>;
+
+static PROMQL_ANNOTATION_REGISTRY: Lazy<Mutex<HashMap<String, Weak<PromqlAnnotationStateRef>>>> =
+    Lazy::new(|| Mutex::new(HashMap::new()));
+
+#[derive(Debug, Default)]
+struct PromqlAnnotationState {
+    warnings: BTreeSet<String>,
+    infos: BTreeSet<String>,
+}
+
+#[derive(Clone, Debug)]
+pub struct PromqlAnnotationCollector {
+    inner: Arc<PromqlAnnotationStateRef>,
+}
+
+impl Default for PromqlAnnotationCollector {
+    fn default() -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(PromqlAnnotationState::default())),
+        }
+    }
+}
+
+impl PromqlAnnotationCollector {
+    pub fn record_warning<S: Into<String>>(&self, warning: S) {
+        self.inner
+            .lock()
+            .expect("promql annotation collector poisoned")
+            .warnings
+            .insert(warning.into());
+    }
+
+    pub fn record_info<S: Into<String>>(&self, info: S) {
+        self.inner
+            .lock()
+            .expect("promql annotation collector poisoned")
+            .infos
+            .insert(info.into());
+    }
+
+    pub fn append_to(&self, warnings: &mut Vec<String>, infos: &mut Vec<String>) {
+        let inner = self
+            .inner
+            .lock()
+            .expect("promql annotation collector poisoned");
+        append_unique(warnings, inner.warnings.iter().cloned());
+        append_unique(infos, inner.infos.iter().cloned());
+    }
+}
+
+impl ConfigExtension for PromqlAnnotationCollector {
+    const PREFIX: &'static str = "greptime_promql_annotations";
+}
+
+impl ExtensionOptions for PromqlAnnotationCollector {
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    fn as_any_mut(&mut self) -> &mut dyn std::any::Any {
+        self
+    }
+
+    fn cloned(&self) -> Box<dyn ExtensionOptions> {
+        Box::new(self.clone())
+    }
+
+    fn set(&mut self, key: &str, value: &str) -> DfResult<()> {
+        Err(DataFusionError::NotImplemented(format!(
+            "PromqlAnnotationCollector does not support set key: {key} with value: {value}"
+        )))
+    }
+
+    fn entries(&self) -> Vec<ConfigEntry> {
+        vec![]
+    }
+}
+
+pub fn promql_annotation_collector(query_id: &str) -> PromqlAnnotationCollector {
+    let mut registry = PROMQL_ANNOTATION_REGISTRY
+        .lock()
+        .expect("promql annotation registry poisoned");
+    registry.retain(|_, collector| collector.strong_count() > 0);
+
+    if let Some(inner) = registry.get(query_id).and_then(Weak::upgrade) {
+        return PromqlAnnotationCollector { inner };
+    }
+
+    let collector = PromqlAnnotationCollector::default();
+    registry.insert(query_id.to_string(), Arc::downgrade(&collector.inner));
+    collector
+}
+
+pub fn get_promql_annotation_collector(query_id: &str) -> Option<PromqlAnnotationCollector> {
+    let mut registry = PROMQL_ANNOTATION_REGISTRY
+        .lock()
+        .expect("promql annotation registry poisoned");
+    registry.retain(|_, collector| collector.strong_count() > 0);
+    registry
+        .get(query_id)
+        .and_then(Weak::upgrade)
+        .map(|inner| PromqlAnnotationCollector { inner })
+}
+
+fn append_unique(target: &mut Vec<String>, values: impl Iterator<Item = String>) {
+    let mut seen = target.iter().cloned().collect::<BTreeSet<_>>();
+    for value in values {
+        if seen.insert(value.clone()) {
+            target.push(value);
+        }
+    }
+    target.sort();
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn collector_deduplicates_and_orders_annotations() {
+        let collector = PromqlAnnotationCollector::default();
+        collector.record_info("z-info");
+        collector.record_info("a-info");
+        collector.record_info("z-info");
+        collector.record_warning("b-warning");
+        collector.record_warning("a-warning");
+
+        let mut warnings = vec!["c-warning".to_string()];
+        let mut infos = vec![];
+        collector.append_to(&mut warnings, &mut infos);
+
+        assert_eq!(warnings, vec!["a-warning", "b-warning", "c-warning"]);
+        assert_eq!(infos, vec!["a-info", "z-info"]);
+    }
+
+    #[test]
+    fn registry_prunes_dead_collectors() {
+        let key = "registry_prunes_dead_collectors";
+        {
+            let collector = promql_annotation_collector(key);
+            assert!(get_promql_annotation_collector(key).is_some());
+            drop(collector);
+        }
+
+        assert!(get_promql_annotation_collector(key).is_none());
+        assert!(!PROMQL_ANNOTATION_REGISTRY.lock().unwrap().contains_key(key));
+    }
+}

src/promql/Cargo.toml

@@ -13,6 +13,7 @@ async-trait.workspace = true
 bytemuck.workspace = true
 common-error.workspace = true
 common-macro.workspace = true
+common-query.workspace = true
 common-telemetry.workspace = true
 datafusion.workspace = true
 datafusion-common.workspace = true

src/promql/src/functions.rs

@@ -18,6 +18,7 @@ mod deriv;
 mod double_exponential_smoothing;
 mod extrapolate_rate;
 mod idelta;
+mod native_histogram;
 mod predict_linear;
 mod quantile;
 mod quantile_aggr;
@@ -42,6 +43,18 @@ pub use deriv::Deriv;
 pub use double_exponential_smoothing::DoubleExponentialSmoothing;
 pub use extrapolate_rate::{Delta, Increase, Rate};
 pub use idelta::IDelta;
+pub use native_histogram::{
+    NativeHistogramAbsentOverTime, NativeHistogramAdd, NativeHistogramAggAvg,
+    NativeHistogramAggSum, NativeHistogramAvg, NativeHistogramAvgOverTime, NativeHistogramChanges,
+    NativeHistogramCount, NativeHistogramCountOverTime, NativeHistogramDelta,
+    NativeHistogramDivScalar, NativeHistogramDrop, NativeHistogramEq, NativeHistogramFraction,
+    NativeHistogramIDelta, NativeHistogramIRate, NativeHistogramIncrease,
+    NativeHistogramLastOverTime, NativeHistogramMulScalar, NativeHistogramNeg,
+    NativeHistogramNotEq, NativeHistogramPresentOverTime, NativeHistogramQuantile,
+    NativeHistogramRate, NativeHistogramResets, NativeHistogramScalarMul, NativeHistogramStddev,
+    NativeHistogramStdvar, NativeHistogramSub, NativeHistogramSum, NativeHistogramSumOverTime,
+    NativeHistogramToString,
+};
 pub use predict_linear::PredictLinear;
 pub use quantile::QuantileOverTime;
 pub use quantile_aggr::{QUANTILE_NAME, quantile_udaf};

src/query/src/datafusion.rs

@@ -28,6 +28,7 @@ use common_catalog::consts::is_readonly_schema;
 use common_error::ext::BoxedError;
 use common_function::function::FunctionContext;
 use common_function::function_factory::ScalarFunctionFactory;
+use common_query::promql_annotations::get_promql_annotation_collector;
 use common_query::{Output, OutputData, OutputMeta};
 use common_recordbatch::adapter::{RecordBatchStreamAdapter, RegionQueryStatCounters};
 use common_recordbatch::{EmptyRecordBatchStream, SendableRecordBatchStream};
@@ -167,16 +168,19 @@ impl DatafusionQueryEngine {
         let physical_plan = self.create_physical_plan(&mut ctx, &plan).await?;
         let physical_plan = self.optimize_physical_plan(&mut ctx, physical_plan)?;
         let physical_plan = if let Some(wrapper) = self.plugins.get::<PhysicalPlanWrapperRef>() {
-            wrapper.wrap(physical_plan, query_ctx)
+            wrapper.wrap(physical_plan, query_ctx.clone())
         } else {
             physical_plan
         };
 
         let stream = self.execute_stream(&ctx, &physical_plan)?;
+        let promql_annotations = query_ctx
+            .remote_query_id()
+            .and_then(get_promql_annotation_collector);
 
         Ok(Output::new(
             OutputData::Stream(stream),
-            OutputMeta::new_with_plan(physical_plan),
+            OutputMeta::new_with_plan(physical_plan).with_promql_annotations(promql_annotations),
         ))
     }
 
@@ -597,6 +601,17 @@ impl QueryEngine for DatafusionQueryEngine {
                 state: self.engine_state().function_state(),
             });
 
+        if let Some(promql_annotations) = query_ctx
+            .remote_query_id()
+            .and_then(get_promql_annotation_collector)
+        {
+            state
+                .config_mut()
+                .options_mut()
+                .extensions
+                .insert(promql_annotations);
+        }
+
         // Carry scheduled Flow time through ConfigOptions.extensions so that
         // the distributed plan analyzer can read it during expression
         // simplification (preventing wall-clock constant-folding of `now()`).

src/query/src/planner.rs

@@ -22,6 +22,7 @@ use arrow_schema::DataType;
 use async_trait::async_trait;
 use catalog::table_source::DfTableSourceProvider;
 use common_error::ext::BoxedError;
+use common_query::promql_annotations::promql_annotation_collector;
 use common_telemetry::tracing;
 use datafusion::common::{DFSchema, plan_err};
 use datafusion::execution::SessionStateBuilder;
@@ -299,7 +300,15 @@ impl DfLogicalPlanner {
 
     #[tracing::instrument(skip_all)]
     async fn plan_pql(&self, stmt: &EvalStmt, query_ctx: QueryContextRef) -> Result<LogicalPlan> {
-        let scheduled_state = self.derive_session_state_with_scheduled_time(&query_ctx)?;
+        let mut scheduled_state = self.derive_session_state_with_scheduled_time(&query_ctx)?;
+        let promql_annotations = query_ctx.remote_query_id().map(promql_annotation_collector);
+        if let Some(collector) = &promql_annotations {
+            scheduled_state
+                .config_mut()
+                .options_mut()
+                .extensions
+                .insert(collector.clone());
+        }
         let plan_decoder = Arc::new(DefaultPlanDecoder::new(
             scheduled_state.clone(),
             &query_ctx,
@@ -314,10 +323,15 @@ impl DfLogicalPlanner {
                 .sql_parser
                 .enable_ident_normalization,
         );
-        let plan = PromPlanner::stmt_to_plan(table_provider, stmt, &self.engine_state)
-            .await
-            .map_err(BoxedError::new)
-            .context(QueryPlanSnafu)?;
+        let plan = PromPlanner::stmt_to_plan_with_annotations(
+            table_provider,
+            stmt,
+            &self.engine_state,
+            promql_annotations,
+        )
+        .await
+        .map_err(BoxedError::new)
+        .context(QueryPlanSnafu)?;
 
         let context = QueryEngineContext::new(scheduled_state, query_ctx);
         Ok(self

src/query/src/promql/planner.rs

@@ -22,7 +22,9 @@ use catalog::table_source::DfTableSourceProvider;
 use common_error::ext::ErrorExt;
 use common_error::status_code::StatusCode;
 use common_function::function::FunctionContext;
+use common_query::native_histogram::native_histogram_value_type;
 use common_query::prelude::greptime_value;
+use common_query::promql_annotations::PromqlAnnotationCollector;
 use datafusion::common::DFSchemaRef;
 use datafusion::datasource::DefaultTableSource;
 use datafusion::functions_aggregate::average::avg_udaf;
@@ -51,7 +53,7 @@ use datafusion_expr::{
     ExprSchemable, Literal, Projection, SortExpr, TableScan, TableSource, col, lit,
 };
 use datatypes::arrow::datatypes::{DataType as ArrowDataType, TimeUnit as ArrowTimeUnit};
-use datatypes::data_type::ConcreteDataType;
+use datatypes::data_type::{ConcreteDataType, DataType as GreptimeDataType};
 use itertools::Itertools;
 use once_cell::sync::Lazy;
 use promql::extension_plan::{
@@ -60,9 +62,18 @@ use promql::extension_plan::{
 };
 use promql::functions::{
     AbsentOverTime, AvgOverTime, Changes, CountOverTime, Delta, Deriv, DoubleExponentialSmoothing,
-    IDelta, Increase, LastOverTime, MaxOverTime, MinOverTime, PredictLinear, PresentOverTime,
-    QuantileOverTime, Rate, Resets, Round, StddevOverTime, StdvarOverTime, SumOverTime,
-    quantile_udaf,
+    IDelta, Increase, LastOverTime, MaxOverTime, MinOverTime, NativeHistogramAbsentOverTime,
+    NativeHistogramAdd, NativeHistogramAggAvg, NativeHistogramAggSum, NativeHistogramAvg,
+    NativeHistogramAvgOverTime, NativeHistogramChanges, NativeHistogramCount,
+    NativeHistogramCountOverTime, NativeHistogramDelta, NativeHistogramDivScalar,
+    NativeHistogramDrop, NativeHistogramEq, NativeHistogramFraction, NativeHistogramIDelta,
+    NativeHistogramIRate, NativeHistogramIncrease, NativeHistogramLastOverTime,
+    NativeHistogramMulScalar, NativeHistogramNeg, NativeHistogramNotEq,
+    NativeHistogramPresentOverTime, NativeHistogramQuantile, NativeHistogramRate,
+    NativeHistogramResets, NativeHistogramScalarMul, NativeHistogramStddev, NativeHistogramStdvar,
+    NativeHistogramSub, NativeHistogramSum, NativeHistogramSumOverTime, NativeHistogramToString,
+    PredictLinear, PresentOverTime, QuantileOverTime, Rate, Resets, Round, StddevOverTime,
+    StdvarOverTime, SumOverTime, quantile_udaf,
 };
 use promql_parser::label::{METRIC_NAME, MatchOp, Matcher, Matchers};
 use promql_parser::parser::token::TokenType;
@@ -374,6 +385,7 @@ impl PromPlannerContext {
 pub struct PromPlanner {
     table_provider: DfTableSourceProvider,
     ctx: PromPlannerContext,
+    promql_annotations: Option<PromqlAnnotationCollector>,
 }
 
 impl PromPlanner {
@@ -381,10 +393,20 @@ impl PromPlanner {
         table_provider: DfTableSourceProvider,
         stmt: &EvalStmt,
         query_engine_state: &QueryEngineState,
+    ) -> Result<LogicalPlan> {
+        Self::stmt_to_plan_with_annotations(table_provider, stmt, query_engine_state, None).await
+    }
+
+    pub async fn stmt_to_plan_with_annotations(
+        table_provider: DfTableSourceProvider,
+        stmt: &EvalStmt,
+        query_engine_state: &QueryEngineState,
+        promql_annotations: Option<PromqlAnnotationCollector>,
     ) -> Result<LogicalPlan> {
         let mut planner = Self {
             table_provider,
             ctx: PromPlannerContext::from_eval_stmt(stmt),
+            promql_annotations,
         };
 
         let plan = planner
@@ -699,6 +721,22 @@ impl PromPlanner {
 
         let group_exprs = self.agg_modifier_to_col(input.schema(), modifier, false)?;
 
+        if self.all_field_columns_are_native_histograms(input.schema()) {
+            let promql_annotations = self.promql_annotations.clone();
+            return self.projection_for_each_field_column(input, |col| {
+                Ok(DfExpr::ScalarFunction(ScalarFunction {
+                    func: Arc::new(NativeHistogramDrop::float_null_udf(
+                        format!(
+                            "{}: dropped native histogram samples because this aggregation is not supported for native histograms",
+                            op
+                        ),
+                        promql_annotations.clone(),
+                    )),
+                    args: vec![DfExpr::Column(Column::from_name(col))],
+                }))
+            });
+        }
+
         let val = Self::get_param_as_literal_expr(param, Some(*op), Some(ArrowDataType::Float64))?;
 
         // convert op and value columns to window exprs.
@@ -775,6 +813,14 @@ impl PromPlanner {
         let UnaryExpr { expr } = unary_expr;
         // Unary Expr in PromQL implys the `-` operator
         let input = self.prom_expr_to_plan(expr, query_engine_state).await?;
+        if self.all_field_columns_are_native_histograms(input.schema()) {
+            return self.projection_for_each_field_column(input, |col| {
+                Ok(DfExpr::ScalarFunction(ScalarFunction {
+                    func: Arc::new(NativeHistogramNeg::scalar_udf()),
+                    args: vec![DfExpr::Column(col.into())],
+                }))
+            });
+        }
         self.projection_for_each_field_column(input, |col| {
             Ok(DfExpr::Negative(Box::new(DfExpr::Column(col.into()))))
         })
@@ -818,6 +864,16 @@ impl PromPlanner {
             });
         }
 
+        if planned_leaves.iter().any(|leaf| {
+            leaf.ctx
+                .field_columns
+                .iter()
+                .any(|field| Self::field_column_is_native_histogram(leaf.plan.schema(), field))
+        }) {
+            self.ctx = original_ctx;
+            return Ok(None);
+        }
+
         if !Self::binary_island_join_contexts_supported(&planned_leaves) {
             self.ctx = original_ctx;
             return Ok(None);
@@ -1154,10 +1210,25 @@ impl PromPlanner {
                 if let Some(time_expr) = self.try_build_special_time_expr_with_context(lhs) {
                     expr = time_expr
                 }
+                let input_schema = input.schema().clone();
+                let promql_annotations = self.promql_annotations.clone();
                 let bin_expr_builder = |col: &String| {
                     let binary_expr_builder = Self::prom_token_to_binary_expr_builder(*op)?;
-                    let mut binary_expr =
-                        binary_expr_builder(expr.clone(), DfExpr::Column(col.into()))?;
+                    let rhs_is_histogram =
+                        Self::field_column_is_native_histogram(&input_schema, col);
+                    let rhs = DfExpr::Column(col.into());
+                    let mut binary_expr = match Self::native_histogram_binary_expr(
+                        *op,
+                        expr.clone(),
+                        false,
+                        rhs.clone(),
+                        rhs_is_histogram,
+                        is_comparison_op && !should_return_bool,
+                        promql_annotations.clone(),
+                    )? {
+                        Some(expr) => expr,
+                        None => binary_expr_builder(expr.clone(), rhs)?,
+                    };
 
                     if is_comparison_op && should_return_bool {
                         binary_expr = DfExpr::Cast(Cast {
@@ -1180,10 +1251,25 @@ impl PromPlanner {
                 if let Some(time_expr) = self.try_build_special_time_expr_with_context(rhs) {
                     expr = time_expr
                 }
+                let input_schema = input.schema().clone();
+                let promql_annotations = self.promql_annotations.clone();
                 let bin_expr_builder = |col: &String| {
                     let binary_expr_builder = Self::prom_token_to_binary_expr_builder(*op)?;
-                    let mut binary_expr =
-                        binary_expr_builder(DfExpr::Column(col.into()), expr.clone())?;
+                    let lhs_is_histogram =
+                        Self::field_column_is_native_histogram(&input_schema, col);
+                    let lhs = DfExpr::Column(col.into());
+                    let mut binary_expr = match Self::native_histogram_binary_expr(
+                        *op,
+                        lhs.clone(),
+                        lhs_is_histogram,
+                        expr.clone(),
+                        false,
+                        is_comparison_op && !should_return_bool,
+                        promql_annotations.clone(),
+                    )? {
+                        Some(expr) => expr,
+                        None => binary_expr_builder(lhs, expr.clone())?,
+                    };
 
                     if is_comparison_op && should_return_bool {
                         binary_expr = DfExpr::Cast(Cast {
@@ -1279,21 +1365,38 @@ impl PromPlanner {
                     &right_context,
                 )?;
                 let join_plan_schema = join_plan.schema().clone();
+                let promql_annotations = self.promql_annotations.clone();
 
                 let bin_expr_builder = |_: &String| {
                     let (left_col_name, right_col_name) = field_columns.next().unwrap();
-                    let left_col = join_plan_schema
+                    let left_field = join_plan_schema
                         .qualified_field_with_name(Some(&left_table_ref), left_col_name)
-                        .context(DataFusionPlanningSnafu)?
-                        .into();
-                    let right_col = join_plan_schema
+                        .context(DataFusionPlanningSnafu)?;
+                    let right_field = join_plan_schema
                         .qualified_field_with_name(Some(&right_table_ref), right_col_name)
-                        .context(DataFusionPlanningSnafu)?
-                        .into();
+                        .context(DataFusionPlanningSnafu)?;
+                    let left_is_histogram =
+                        left_field.1.data_type() == &Self::native_histogram_arrow_type();
+                    let right_is_histogram =
+                        right_field.1.data_type() == &Self::native_histogram_arrow_type();
+                    let left_col = left_field.into();
+                    let right_col = right_field.into();
 
                     let binary_expr_builder = Self::prom_token_to_binary_expr_builder(*op)?;
-                    let mut binary_expr =
-                        binary_expr_builder(DfExpr::Column(left_col), DfExpr::Column(right_col))?;
+                    let lhs = DfExpr::Column(left_col);
+                    let rhs = DfExpr::Column(right_col);
+                    let mut binary_expr = match Self::native_histogram_binary_expr(
+                        *op,
+                        lhs.clone(),
+                        left_is_histogram,
+                        rhs.clone(),
+                        right_is_histogram,
+                        is_comparison_op && !should_return_bool,
+                        promql_annotations.clone(),
+                    )? {
+                        Some(expr) => expr,
+                        None => binary_expr_builder(lhs, rhs)?,
+                    };
                     if is_comparison_op && should_return_bool {
                         binary_expr = DfExpr::Cast(Cast {
                             expr: Box::new(binary_expr),
@@ -1613,8 +1716,12 @@ impl PromPlanner {
                 ),
             })
         };
-        let (mut func_exprs, new_tags) =
-            self.create_function_expr(func, args.literals.clone(), query_engine_state)?;
+        let (mut func_exprs, new_tags) = self.create_function_expr(
+            func,
+            args.literals.clone(),
+            input.schema(),
+            query_engine_state,
+        )?;
         func_exprs.insert(0, self.create_time_index_column_expr()?);
         func_exprs.extend_from_slice(&self.create_tag_column_exprs()?);
         if let Some(tsid_col) =
@@ -2738,6 +2845,7 @@ impl PromPlanner {
         &mut self,
         func: &Function,
         other_input_exprs: Vec<DfExpr>,
+        input_schema: &DFSchemaRef,
         query_engine_state: &QueryEngineState,
     ) -> Result<(Vec<DfExpr>, Vec<String>)> {
         // TODO(ruihang): check function args list
@@ -2748,35 +2856,136 @@ impl PromPlanner {
         let mut exprs = Vec::with_capacity(self.ctx.field_columns.len());
         // New labels after executing the function, e.g. `label_replace` etc.
         let mut new_tags = vec![];
+        let promql_annotations = self.promql_annotations.clone();
+        let native_histogram_drop_udf = |name: &str| {
+            Arc::new(NativeHistogramDrop::float_null_udf(
+                format!(
+                    "{name}: dropped native histogram samples because this function is not supported for native histograms"
+                ),
+                promql_annotations.clone(),
+            ))
+        };
         let scalar_func = match func.name {
+            "increase" if self.all_field_columns_are_native_histogram_ranges(input_schema) => {
+                ScalarFunc::ExtrapolateUdf(
+                    Arc::new(NativeHistogramIncrease::scalar_udf_with_collector(
+                        self.promql_annotations.clone(),
+                    )),
+                    self.ctx.range.context(ExpectRangeSelectorSnafu)?,
+                )
+            }
             "increase" => ScalarFunc::ExtrapolateUdf(
                 Arc::new(Increase::scalar_udf()),
                 self.ctx.range.context(ExpectRangeSelectorSnafu)?,
             ),
+            "rate" if self.all_field_columns_are_native_histogram_ranges(input_schema) => {
+                ScalarFunc::ExtrapolateUdf(
+                    Arc::new(NativeHistogramRate::scalar_udf_with_collector(
+                        self.promql_annotations.clone(),
+                    )),
+                    self.ctx.range.context(ExpectRangeSelectorSnafu)?,
+                )
+            }
             "rate" => ScalarFunc::ExtrapolateUdf(
                 Arc::new(Rate::scalar_udf()),
                 self.ctx.range.context(ExpectRangeSelectorSnafu)?,
             ),
+            "delta" if self.all_field_columns_are_native_histogram_ranges(input_schema) => {
+                ScalarFunc::ExtrapolateUdf(
+                    Arc::new(NativeHistogramDelta::scalar_udf_with_collector(
+                        self.promql_annotations.clone(),
+                    )),
+                    self.ctx.range.context(ExpectRangeSelectorSnafu)?,
+                )
+            }
             "delta" => ScalarFunc::ExtrapolateUdf(
                 Arc::new(Delta::scalar_udf()),
                 self.ctx.range.context(ExpectRangeSelectorSnafu)?,
             ),
+            "idelta" if self.all_field_columns_are_native_histogram_ranges(input_schema) => {
+                ScalarFunc::Udf(Arc::new(NativeHistogramIDelta::scalar_udf_with_collector(
+                    self.promql_annotations.clone(),
+                )))
+            }
             "idelta" => ScalarFunc::Udf(Arc::new(IDelta::<false>::scalar_udf())),
+            "irate" if self.all_field_columns_are_native_histogram_ranges(input_schema) => {
+                ScalarFunc::Udf(Arc::new(NativeHistogramIRate::scalar_udf_with_collector(
+                    self.promql_annotations.clone(),
+                )))
+            }
             "irate" => ScalarFunc::Udf(Arc::new(IDelta::<true>::scalar_udf())),
+            "resets" if self.all_field_columns_are_native_histogram_ranges(input_schema) => {
+                ScalarFunc::Udf(Arc::new(NativeHistogramResets::scalar_udf()))
+            }
             "resets" => ScalarFunc::Udf(Arc::new(Resets::scalar_udf())),
+            "changes" if self.all_field_columns_are_native_histogram_ranges(input_schema) => {
+                ScalarFunc::Udf(Arc::new(NativeHistogramChanges::scalar_udf()))
+            }
             "changes" => ScalarFunc::Udf(Arc::new(Changes::scalar_udf())),
+            "deriv" if self.all_field_columns_are_native_histogram_ranges(input_schema) => {
+                ScalarFunc::Udf(native_histogram_drop_udf(func.name))
+            }
             "deriv" => ScalarFunc::Udf(Arc::new(Deriv::scalar_udf())),
+            "avg_over_time" if self.all_field_columns_are_native_histogram_ranges(input_schema) => {
+                ScalarFunc::Udf(Arc::new(
+                    NativeHistogramAvgOverTime::scalar_udf_with_collector(
+                        self.promql_annotations.clone(),
+                    ),
+                ))
+            }
             "avg_over_time" => ScalarFunc::Udf(Arc::new(AvgOverTime::scalar_udf())),
+            "min_over_time" | "max_over_time"
+                if self.all_field_columns_are_native_histogram_ranges(input_schema) =>
+            {
+                ScalarFunc::Udf(native_histogram_drop_udf(func.name))
+            }
             "min_over_time" => ScalarFunc::Udf(Arc::new(MinOverTime::scalar_udf())),
             "max_over_time" => ScalarFunc::Udf(Arc::new(MaxOverTime::scalar_udf())),
+            "sum_over_time" if self.all_field_columns_are_native_histogram_ranges(input_schema) => {
+                ScalarFunc::Udf(Arc::new(
+                    NativeHistogramSumOverTime::scalar_udf_with_collector(
+                        self.promql_annotations.clone(),
+                    ),
+                ))
+            }
             "sum_over_time" => ScalarFunc::Udf(Arc::new(SumOverTime::scalar_udf())),
+            "count_over_time"
+                if self.all_field_columns_are_native_histogram_ranges(input_schema) =>
+            {
+                ScalarFunc::Udf(Arc::new(NativeHistogramCountOverTime::scalar_udf()))
+            }
             "count_over_time" => ScalarFunc::Udf(Arc::new(CountOverTime::scalar_udf())),
+            "last_over_time"
+                if self.all_field_columns_are_native_histogram_ranges(input_schema) =>
+            {
+                ScalarFunc::Udf(Arc::new(NativeHistogramLastOverTime::scalar_udf()))
+            }
             "last_over_time" => ScalarFunc::Udf(Arc::new(LastOverTime::scalar_udf())),
+            "absent_over_time"
+                if self.all_field_columns_are_native_histogram_ranges(input_schema) =>
+            {
+                ScalarFunc::Udf(Arc::new(NativeHistogramAbsentOverTime::scalar_udf()))
+            }
             "absent_over_time" => ScalarFunc::Udf(Arc::new(AbsentOverTime::scalar_udf())),
+            "present_over_time"
+                if self.all_field_columns_are_native_histogram_ranges(input_schema) =>
+            {
+                ScalarFunc::Udf(Arc::new(NativeHistogramPresentOverTime::scalar_udf()))
+            }
             "present_over_time" => ScalarFunc::Udf(Arc::new(PresentOverTime::scalar_udf())),
+            "stddev_over_time" | "stdvar_over_time" | "quantile_over_time"
+                if self.all_field_columns_are_native_histogram_ranges(input_schema) =>
+            {
+                ScalarFunc::Udf(native_histogram_drop_udf(func.name))
+            }
             "stddev_over_time" => ScalarFunc::Udf(Arc::new(StddevOverTime::scalar_udf())),
             "stdvar_over_time" => ScalarFunc::Udf(Arc::new(StdvarOverTime::scalar_udf())),
             "quantile_over_time" => ScalarFunc::Udf(Arc::new(QuantileOverTime::scalar_udf())),
+            "predict_linear"
+                if self.all_field_columns_are_native_histogram_ranges(input_schema) =>
+            {
+                ScalarFunc::Udf(native_histogram_drop_udf(func.name))
+            }
             "predict_linear" => {
                 other_input_exprs[0] = DfExpr::Cast(Cast {
                     expr: Box::new(other_input_exprs[0].clone()),
@@ -2784,9 +2993,32 @@ impl PromPlanner {
                 });
                 ScalarFunc::Udf(Arc::new(PredictLinear::scalar_udf()))
             }
+            "double_exponential_smoothing" | "holt_winters"
+                if self.all_field_columns_are_native_histogram_ranges(input_schema) =>
+            {
+                ScalarFunc::Udf(native_histogram_drop_udf(func.name))
+            }
             "double_exponential_smoothing" | "holt_winters" => {
                 ScalarFunc::Udf(Arc::new(DoubleExponentialSmoothing::scalar_udf()))
             }
+            "histogram_count" => {
+                ScalarFunc::NativeHistogramUdf(Arc::new(NativeHistogramCount::scalar_udf()))
+            }
+            "histogram_sum" => {
+                ScalarFunc::NativeHistogramUdf(Arc::new(NativeHistogramSum::scalar_udf()))
+            }
+            "histogram_avg" => {
+                ScalarFunc::NativeHistogramUdf(Arc::new(NativeHistogramAvg::scalar_udf()))
+            }
+            "histogram_stddev" => {
+                ScalarFunc::NativeHistogramUdf(Arc::new(NativeHistogramStddev::scalar_udf()))
+            }
+            "histogram_stdvar" => {
+                ScalarFunc::NativeHistogramUdf(Arc::new(NativeHistogramStdvar::scalar_udf()))
+            }
+            "histogram_fraction" => {
+                ScalarFunc::NativeHistogramUdf(Arc::new(NativeHistogramFraction::scalar_udf()))
+            }
             "time" => {
                 exprs.push(build_special_time_expr(
                     self.ctx.time_index_column.as_ref().unwrap(),
@@ -2935,12 +3167,21 @@ impl PromPlanner {
 
                 ScalarFunc::GeneratedExpr
             }
+            "round" if self.all_field_columns_are_native_histograms(input_schema) => {
+                if other_input_exprs.is_empty() {
+                    other_input_exprs.push_front(0.0f64.lit());
+                }
+                ScalarFunc::DataFusionUdf(native_histogram_drop_udf(func.name))
+            }
             "round" => {
                 if other_input_exprs.is_empty() {
                     other_input_exprs.push_front(0.0f64.lit());
                 }
                 ScalarFunc::DataFusionUdf(Arc::new(Round::scalar_udf()))
             }
+            "rad" | "deg" | "sgn" if self.all_field_columns_are_native_histograms(input_schema) => {
+                ScalarFunc::DataFusionUdf(native_histogram_drop_udf(func.name))
+            }
             "rad" => ScalarFunc::DataFusionBuiltin(datafusion::functions::math::radians()),
             "deg" => ScalarFunc::DataFusionBuiltin(datafusion::functions::math::degrees()),
             "sgn" => ScalarFunc::DataFusionBuiltin(datafusion::functions::math::signum()),
@@ -2960,20 +3201,32 @@ impl PromPlanner {
                     .scalar_functions()
                     .get(func.name)
                 {
-                    ScalarFunc::DataFusionBuiltin(f.clone())
+                    if self.all_field_columns_are_native_histograms(input_schema) {
+                        ScalarFunc::DataFusionUdf(native_histogram_drop_udf(func.name))
+                    } else {
+                        ScalarFunc::DataFusionBuiltin(f.clone())
+                    }
                 } else if let Some(factory) = query_engine_state.scalar_function(func.name) {
-                    let func_state = query_engine_state.function_state();
-                    let query_ctx = self.table_provider.query_ctx();
+                    if self.all_field_columns_are_native_histograms(input_schema) {
+                        ScalarFunc::DataFusionUdf(native_histogram_drop_udf(func.name))
+                    } else {
+                        let func_state = query_engine_state.function_state();
+                        let query_ctx = self.table_provider.query_ctx();
 
-                    ScalarFunc::DataFusionUdf(Arc::new(factory.provide(FunctionContext {
-                        state: func_state,
-                        query_ctx: query_ctx.clone(),
-                    })))
+                        ScalarFunc::DataFusionUdf(Arc::new(factory.provide(FunctionContext {
+                            state: func_state,
+                            query_ctx: query_ctx.clone(),
+                        })))
+                    }
                 } else if let Some(f) = datafusion_functions::math::functions()
                     .iter()
                     .find(|f| f.name() == func.name)
                 {
-                    ScalarFunc::DataFusionUdf(f.clone())
+                    if self.all_field_columns_are_native_histograms(input_schema) {
+                        ScalarFunc::DataFusionUdf(native_histogram_drop_udf(func.name))
+                    } else {
+                        ScalarFunc::DataFusionUdf(f.clone())
+                    }
                 } else {
                     return UnsupportedExprSnafu {
                         name: func.name.to_string(),
@@ -3005,6 +3258,25 @@ impl PromPlanner {
                     .collect_vec();
                     exprs.push(DfExpr::ScalarFunction(ScalarFunction { func, args }))
                 }
+                ScalarFunc::NativeHistogramUdf(func) => {
+                    if Self::field_column_is_native_histogram(input_schema, value) {
+                        let args = itertools::chain!(
+                            other_input_exprs.iter().take(field_column_pos).cloned(),
+                            std::iter::once(col_expr),
+                            other_input_exprs.iter().skip(field_column_pos).cloned()
+                        )
+                        .collect_vec();
+                        exprs.push(DfExpr::ScalarFunction(ScalarFunction { func, args }));
+                    } else {
+                        exprs.push(
+                            DfExpr::Literal(ScalarValue::Float64(None), None).alias(format!(
+                                "{}_{}",
+                                func.name(),
+                                value
+                            )),
+                        );
+                    }
+                }
                 ScalarFunc::Udf(func) => {
                     let ts_range_expr = DfExpr::Column(Column::from_name(
                         RangeManipulate::build_timestamp_range_name(
@@ -3391,6 +3663,11 @@ impl PromPlanner {
                 }
             );
         }
+
+        if self.all_field_columns_are_native_histograms(input_plan.schema()) {
+            return self.create_native_histogram_aggregate_exprs(op, input_plan);
+        }
+
         let aggr = match op.id() {
             token::T_SUM => sum_udaf(),
             token::T_QUANTILE => {
@@ -3466,6 +3743,85 @@ impl PromPlanner {
         Ok((exprs, prev_field_exprs))
     }
 
+    fn create_native_histogram_aggregate_exprs(
+        &mut self,
+        op: TokenType,
+        input_plan: &LogicalPlan,
+    ) -> Result<(Vec<DfExpr>, Vec<DfExpr>)> {
+        let prev_field_exprs = if op.id() == token::T_COUNT_VALUES {
+            ensure!(
+                self.ctx.field_columns.len() == 1,
+                UnsupportedExprSnafu {
+                    name: "count_values on multi-value input"
+                }
+            );
+            self.ctx
+                .field_columns
+                .iter()
+                .map(|col| {
+                    DfExpr::ScalarFunction(ScalarFunction {
+                        func: Arc::new(NativeHistogramToString::scalar_udf()),
+                        args: vec![DfExpr::Column(Column::from_name(col))],
+                    })
+                })
+                .collect::<Vec<_>>()
+        } else {
+            vec![]
+        };
+
+        let exprs = self
+            .ctx
+            .field_columns
+            .iter()
+            .map(|col| {
+                let input = DfExpr::Column(Column::from_name(col));
+                let expr = match op.id() {
+                    token::T_SUM => Arc::new(NativeHistogramAggSum::aggregate_udf_with_collector(
+                        self.promql_annotations.clone(),
+                    ))
+                        .call(vec![input])
+                        .alias(col),
+                    token::T_AVG => Arc::new(NativeHistogramAggAvg::aggregate_udf_with_collector(
+                        self.promql_annotations.clone(),
+                    ))
+                        .call(vec![input])
+                        .alias(col),
+                    token::T_COUNT_VALUES | token::T_COUNT => count_udaf().call(vec![input]).alias(col),
+                    token::T_GROUP => max_udaf().call(vec![lit(1_f64)]).alias(col),
+                    token::T_MIN
+                    | token::T_MAX
+                    | token::T_STDDEV
+                    | token::T_STDVAR
+                    | token::T_QUANTILE
+                    | token::T_TOPK
+                    | token::T_BOTTOMK => sum_udaf()
+                        .call(vec![DfExpr::ScalarFunction(ScalarFunction {
+                            func: Arc::new(NativeHistogramDrop::float_null_udf(
+                                format!(
+                                    "{}: dropped native histogram samples because this aggregation is not supported for native histograms",
+                                    op
+                                ),
+                                self.promql_annotations.clone(),
+                            )),
+                            args: vec![input],
+                        })])
+                        .alias(col),
+                    _ => return UnexpectedTokenSnafu { token: op }.fail(),
+                };
+                Ok(expr)
+            })
+            .collect::<Result<Vec<_>>>()?;
+
+        let normalized_exprs =
+            normalize_cols(exprs.iter().cloned(), input_plan).context(DataFusionPlanningSnafu)?;
+        self.ctx.field_columns = normalized_exprs
+            .into_iter()
+            .map(|expr| expr.schema_name().to_string())
+            .collect();
+
+        Ok((exprs, prev_field_exprs))
+    }
+
     fn get_param_value_as_str(op: TokenType, param: &Option<Box<PromExpr>>) -> Result<&str> {
         let param = param
             .as_deref()
@@ -3634,6 +3990,10 @@ impl PromPlanner {
         let input_plan = self.strip_tsid_column(input_plan)?;
         self.ctx.use_tsid = false;
 
+        if self.all_field_columns_are_native_histograms(input_plan.schema()) {
+            return self.create_native_histogram_quantile_plan(phi, input_plan);
+        }
+
         if !self.ctx.has_le_tag() {
             // Return empty result instead of error when 'le' column is not found
             // This handles the case when histogram metrics don't exist
@@ -3677,6 +4037,43 @@ impl PromPlanner {
         }))
     }
 
+    fn create_native_histogram_quantile_plan(
+        &mut self,
+        phi: f64,
+        input_plan: LogicalPlan,
+    ) -> Result<LogicalPlan> {
+        ensure!(
+            self.ctx.field_columns.len() == 1,
+            MultiFieldsNotSupportedSnafu {
+                operator: SPECIAL_HISTOGRAM_QUANTILE
+            },
+        );
+
+        let field_column = self.ctx.field_columns[0].clone();
+        let quantile_expr = DfExpr::ScalarFunction(ScalarFunction {
+            func: Arc::new(NativeHistogramQuantile::scalar_udf()),
+            args: vec![
+                DfExpr::Column(Column::from_name(field_column)),
+                DfExpr::Literal(ScalarValue::Float64(Some(phi)), None),
+            ],
+        });
+        let display_name = quantile_expr.schema_name().to_string();
+        self.ctx.field_columns = vec![display_name.clone()];
+
+        let project_exprs = std::iter::once(self.create_time_index_column_expr()?)
+            .chain(std::iter::once(quantile_expr.alias(display_name)))
+            .chain(self.create_tag_column_exprs()?)
+            .collect::<Vec<_>>();
+
+        LogicalPlanBuilder::from(input_plan)
+            .project(project_exprs)
+            .context(DataFusionPlanningSnafu)?
+            .filter(self.create_empty_values_filter_expr()?)
+            .context(DataFusionPlanningSnafu)?
+            .build()
+            .context(DataFusionPlanningSnafu)
+    }
+
     /// Create a [SPECIAL_VECTOR_FUNCTION] plan
     async fn create_vector_plan(&mut self, args: &PromFunctionArgs) -> Result<LogicalPlan> {
         if args.args.len() != 1 {
@@ -3884,6 +4281,69 @@ impl PromPlanner {
         }
     }
 
+    fn native_histogram_binary_expr(
+        token: TokenType,
+        lhs: DfExpr,
+        lhs_is_histogram: bool,
+        rhs: DfExpr,
+        rhs_is_histogram: bool,
+        filter_context: bool,
+        promql_annotations: Option<PromqlAnnotationCollector>,
+    ) -> Result<Option<DfExpr>> {
+        if !lhs_is_histogram && !rhs_is_histogram {
+            return Ok(None);
+        }
+
+        let scalar_fn = |func: ScalarUdfDef, args| {
+            DfExpr::ScalarFunction(ScalarFunction {
+                func: Arc::new(func),
+                args,
+            })
+        };
+        let invalid_expr = || {
+            let message = format!(
+                "{}: dropped native histogram samples because this binary operation is not supported for native histograms",
+                token
+            );
+            let func = if filter_context {
+                NativeHistogramDrop::bool_false_udf(message, promql_annotations.clone())
+            } else {
+                NativeHistogramDrop::float_null_udf(message, promql_annotations.clone())
+            };
+            let args = vec![lhs.clone(), rhs.clone()];
+            scalar_fn(func, args)
+        };
+
+        let expr = match (token.id(), lhs_is_histogram, rhs_is_histogram) {
+            (token::T_ADD, true, true) => scalar_fn(
+                NativeHistogramAdd::scalar_udf_with_collector(promql_annotations.clone()),
+                vec![lhs, rhs],
+            ),
+            (token::T_SUB, true, true) => scalar_fn(
+                NativeHistogramSub::scalar_udf_with_collector(promql_annotations.clone()),
+                vec![lhs, rhs],
+            ),
+            (token::T_MUL, true, false) => {
+                scalar_fn(NativeHistogramMulScalar::scalar_udf(), vec![lhs, rhs])
+            }
+            (token::T_MUL, false, true) => {
+                scalar_fn(NativeHistogramScalarMul::scalar_udf(), vec![lhs, rhs])
+            }
+            (token::T_DIV, true, false) => {
+                scalar_fn(NativeHistogramDivScalar::scalar_udf(), vec![lhs, rhs])
+            }
+            (token::T_EQLC, true, true) => {
+                scalar_fn(NativeHistogramEq::scalar_udf(), vec![lhs, rhs])
+            }
+            (token::T_NEQ, true, true) => {
+                scalar_fn(NativeHistogramNotEq::scalar_udf(), vec![lhs, rhs])
+            }
+            _ => invalid_expr(),
+        };
+
+        Ok(Some(expr))
+    }
+
     /// Return a lambda to build binary expression from token.
     /// Because some binary operator are function in DataFusion like `atan2` or `^`.
     #[allow(clippy::type_complexity)]
@@ -3990,6 +4450,52 @@ impl PromPlanner {
             .any(|field| field.name() == DATA_SCHEMA_TSID_COLUMN_NAME)
     }
 
+    fn native_histogram_arrow_type() -> ArrowDataType {
+        native_histogram_value_type().as_arrow_type()
+    }
+
+    fn field_column_type<'a>(
+        schema: &'a DFSchemaRef,
+        field_column: &str,
+    ) -> Option<&'a ArrowDataType> {
+        schema
+            .index_of_column_by_name(None, field_column)
+            .map(|idx| schema.field(idx).data_type())
+    }
+
+    fn field_column_is_native_histogram(schema: &DFSchemaRef, field_column: &str) -> bool {
+        Self::field_column_type(schema, field_column)
+            .is_some_and(|data_type| data_type == &Self::native_histogram_arrow_type())
+    }
+
+    fn field_column_is_native_histogram_range(schema: &DFSchemaRef, field_column: &str) -> bool {
+        Self::field_column_type(schema, field_column).is_some_and(|data_type| {
+            matches!(
+                data_type,
+                ArrowDataType::Dictionary(_, value_type)
+                    if value_type.as_ref() == &Self::native_histogram_arrow_type()
+            )
+        })
+    }
+
+    fn all_field_columns_are_native_histograms(&self, schema: &DFSchemaRef) -> bool {
+        !self.ctx.field_columns.is_empty()
+            && self
+                .ctx
+                .field_columns
+                .iter()
+                .all(|field| Self::field_column_is_native_histogram(schema, field))
+    }
+
+    fn all_field_columns_are_native_histogram_ranges(&self, schema: &DFSchemaRef) -> bool {
+        !self.ctx.field_columns.is_empty()
+            && self
+                .ctx
+                .field_columns
+                .iter()
+                .all(|field| Self::field_column_is_native_histogram_range(schema, field))
+    }
+
     fn optional_tsid_projection(
         schema: &DFSchemaRef,
         table_ref: Option<&TableReference>,
@@ -4719,6 +5225,9 @@ enum ScalarFunc {
     /// Similar to DataFusionBuiltin but for custom functions not built into DataFusion.
     /// Processing: Direct pass-through with argument positioning.
     DataFusionUdf(Arc<ScalarUdfDef>),
+    /// Native histogram helper UDFs. Non-histogram inputs are projected as NULL
+    /// so the normal PromQL empty-value filter drops them.
+    NativeHistogramUdf(Arc<ScalarUdfDef>),
     /// PromQL-specific functions that operate on time series data with temporal context.
     /// These functions require both timestamp ranges and values to perform calculations.
     /// Processing: Automatically injects timestamp_range and value columns as first arguments.
@@ -4745,6 +5254,7 @@ mod test {
     use catalog::memory::{MemoryCatalogManager, new_memory_catalog_manager};
     use common_base::Plugins;
     use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
+    use common_query::native_histogram::NATIVE_HISTOGRAM_FIELD;
     use common_query::prelude::greptime_timestamp;
     use common_query::test_util::DummyDecoder;
     use datafusion::arrow::datatypes::Schema as ArrowSchema;
@@ -4925,6 +5435,62 @@ mod test {
         )
     }
 
+    async fn build_test_native_histogram_table_provider(table_name: &str) -> DfTableSourceProvider {
+        let catalog_list = MemoryCatalogManager::with_default_setup();
+        let columns = vec![
+            ColumnSchema::new(
+                "tag_0".to_string(),
+                ConcreteDataType::string_datatype(),
+                false,
+            ),
+            ColumnSchema::new(
+                "timestamp".to_string(),
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                false,
+            )
+            .with_time_index(true),
+            ColumnSchema::new(
+                NATIVE_HISTOGRAM_FIELD.to_string(),
+                native_histogram_value_type().clone(),
+                true,
+            ),
+        ];
+        let schema = Arc::new(Schema::new(columns));
+        let table_meta = TableMetaBuilder::empty()
+            .schema(schema)
+            .primary_key_indices(vec![0])
+            .value_indices(vec![2])
+            .next_column_id(1024)
+            .build()
+            .unwrap();
+        let table_info = TableInfoBuilder::default()
+            .name(table_name)
+            .meta(table_meta)
+            .build()
+            .unwrap();
+        let table = EmptyTable::from_table_info(&table_info);
+
+        assert!(
+            catalog_list
+                .register_table_sync(RegisterTableRequest {
+                    catalog: DEFAULT_CATALOG_NAME.to_string(),
+                    schema: DEFAULT_SCHEMA_NAME.to_string(),
+                    table_name: table_name.to_string(),
+                    table_id: 1024,
+                    table,
+                })
+                .is_ok()
+        );
+
+        DfTableSourceProvider::new(
+            catalog_list,
+            false,
+            QueryContext::arc(),
+            DummyDecoder::arc(),
+            false,
+        )
+    }
+
     async fn build_test_table_provider_with_tsid(
         table_name_tuples: &[(String, String)],
         num_tag: usize,
@@ -6278,6 +6844,7 @@ mod test {
         let mut planner = PromPlanner {
             table_provider,
             ctx: PromPlannerContext::from_eval_stmt(&eval_stmt),
+            promql_annotations: None,
         };
         let plan = planner
             .prom_expr_to_plan(&eval_stmt.expr, &build_query_engine_state())
@@ -7132,6 +7699,78 @@ mod test {
         indie_query_plan_compare(query, expected).await;
     }
 
+    async fn native_histogram_plan(query: &str) -> String {
+        let table_provider = build_test_native_histogram_table_provider("some_metric").await;
+        let plan = PromPlanner::stmt_to_plan(
+            table_provider,
+            &build_eval_stmt(query),
+            &build_query_engine_state(),
+        )
+        .await
+        .unwrap();
+        plan.display_indent_schema().to_string()
+    }
+
+    #[tokio::test]
+    async fn native_histogram_count_uses_native_udf() {
+        let plan = native_histogram_plan("histogram_count(some_metric)").await;
+
+        assert!(plan.contains("prom_native_histogram_count"), "{plan}");
+        assert!(!plan.contains("PromHistogramFold"), "{plan}");
+    }
+
+    #[tokio::test]
+    async fn native_histogram_rate_can_feed_count() {
+        let plan = native_histogram_plan("histogram_count(rate(some_metric[5m]))").await;
+
+        assert!(plan.contains("prom_native_histogram_rate"), "{plan}");
+        assert!(plan.contains("prom_native_histogram_count"), "{plan}");
+    }
+
+    #[tokio::test]
+    async fn native_histogram_quantile_skips_classic_fold() {
+        let plan = native_histogram_plan("histogram_quantile(0.9, some_metric)").await;
+
+        assert!(plan.contains("prom_native_histogram_quantile"), "{plan}");
+        assert!(!plan.contains("PromHistogramFold"), "{plan}");
+    }
+
+    #[tokio::test]
+    async fn native_histogram_topk_uses_drop_udf() {
+        let plan = native_histogram_plan("topk(1, some_metric)").await;
+
+        assert!(plan.contains("prom_native_histogram_drop_float"), "{plan}");
+    }
+
+    #[tokio::test]
+    async fn unsupported_native_histogram_functions_use_drop_udf() {
+        for query in [
+            "deriv(some_metric[5m])",
+            "min_over_time(some_metric[5m])",
+            "quantile_over_time(0.9, some_metric[5m])",
+            "predict_linear(some_metric[5m], 60)",
+            "round(some_metric)",
+            "abs(some_metric)",
+        ] {
+            let plan = native_histogram_plan(query).await;
+
+            assert!(
+                plan.contains("prom_native_histogram_drop_float"),
+                "{query}\n{plan}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn native_histogram_absent_over_time_uses_native_udf() {
+        let plan = native_histogram_plan("absent_over_time(some_metric[5m])").await;
+
+        assert!(
+            plan.contains("prom_native_histogram_absent_over_time"),
+            "{plan}"
+        );
+    }
+
     #[tokio::test]
     async fn less_filter_on_value() {
         let query = "some_metric < 1.2345";

src/servers/src/http/prometheus.rs

@@ -74,17 +74,31 @@ use crate::prometheus_handler::PrometheusHandlerRef;
 
 /// For [ValueType::Vector] result type
 #[derive(Debug, Default, Serialize, Deserialize, PartialEq)]
+#[serde(deny_unknown_fields)]
 pub struct PromSeriesVector {
     pub metric: BTreeMap<String, String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub value: Option<(f64, String)>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub histogram: Option<(f64, PromNativeHistogram)>,
 }
 
 /// For [ValueType::Matrix] result type
 #[derive(Debug, Default, Serialize, Deserialize, PartialEq)]
+#[serde(deny_unknown_fields)]
 pub struct PromSeriesMatrix {
     pub metric: BTreeMap<String, String>,
+    #[serde(skip_serializing_if = "Vec::is_empty", default)]
     pub values: Vec<(f64, String)>,
+    #[serde(skip_serializing_if = "Vec::is_empty", default)]
+    pub histograms: Vec<(f64, PromNativeHistogram)>,
+}
+
+#[derive(Debug, Default, Serialize, Deserialize, PartialEq)]
+pub struct PromNativeHistogram {
+    pub count: String,
+    pub sum: String,
+    pub buckets: Vec<(u8, String, String, String)>,
 }
 
 /// Variants corresponding to [ValueType]

src/servers/src/http/result/prometheus_resp.rs

@@ -16,7 +16,7 @@
 use std::cmp::Ordering;
 use std::collections::{BTreeMap, HashMap};
 
-use arrow::array::{Array, AsArray};
+use arrow::array::{Array, AsArray, StructArray};
 use arrow::datatypes::{Float64Type, TimestampMillisecondType};
 use arrow_schema::DataType;
 use axum::Json;
@@ -24,6 +24,9 @@ use axum::http::HeaderValue;
 use axum::response::{IntoResponse, Response};
 use common_error::ext::ErrorExt;
 use common_error::status_code::StatusCode;
+use common_query::native_histogram::{
+    NativeHistogram, is_native_histogram_value_type, read_histogram,
+};
 use common_query::{Output, OutputData};
 use common_recordbatch::RecordBatches;
 use datatypes::prelude::ConcreteDataType;
@@ -35,13 +38,37 @@ use serde_json::Value;
 use snafu::{OptionExt, ResultExt};
 
 use crate::error::{
-    ArrowSnafu, CollectRecordbatchSnafu, Result, UnexpectedResultSnafu, status_code_to_http_status,
+    ArrowSnafu, CollectRecordbatchSnafu, DataFusionSnafu, Result, UnexpectedResultSnafu,
+    status_code_to_http_status,
 };
 use crate::http::header::{GREPTIME_DB_HEADER_METRICS, collect_plan_metrics};
 use crate::http::prometheus::{
-    PromData, PromQueryResult, PromSeriesMatrix, PromSeriesVector, PrometheusResponse,
+    PromData, PromNativeHistogram, PromQueryResult, PromSeriesMatrix, PromSeriesVector,
+    PrometheusResponse,
 };
 
+#[derive(Default)]
+struct PromSeriesSamples {
+    values: Vec<(f64, String)>,
+    histograms: Vec<(f64, PromNativeHistogram)>,
+}
+
+fn prometheus_native_histogram(histogram: &NativeHistogram) -> Option<PromNativeHistogram> {
+    Some(PromNativeHistogram {
+        count: histogram.count.to_string(),
+        sum: histogram.sum.to_string(),
+        buckets: histogram.to_prometheus_buckets()?,
+    })
+}
+
+fn prometheus_native_histogram_or_error(
+    histogram: &NativeHistogram,
+) -> Result<PromNativeHistogram> {
+    prometheus_native_histogram(histogram).context(UnexpectedResultSnafu {
+        reason: "native histogram cannot be converted to Prometheus buckets".to_string(),
+    })
+}
+
 #[derive(Debug, Default, Serialize, Deserialize, PartialEq)]
 pub struct PrometheusJsonResponse {
     pub status: String,
@@ -55,6 +82,8 @@ pub struct PrometheusJsonResponse {
     pub error_type: Option<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub warnings: Option<Vec<String>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub infos: Option<Vec<String>>,
 
     #[serde(skip)]
     pub status_code: Option<StatusCode>,
@@ -99,6 +128,7 @@ impl PrometheusJsonResponse {
             error: Some(reason.into()),
             error_type: Some(error_type.to_string()),
             warnings: None,
+            infos: None,
             resp_metrics: Default::default(),
             status_code: Some(error_type),
         }
@@ -111,6 +141,7 @@ impl PrometheusJsonResponse {
             error: None,
             error_type: None,
             warnings: None,
+            infos: None,
             resp_metrics: Default::default(),
             status_code: None,
         }
@@ -124,6 +155,7 @@ impl PrometheusJsonResponse {
     ) -> Self {
         let response: Result<Self> = try {
             let result = result?;
+            let mut meta = result.meta;
             let mut resp =
                 match result.data {
                     OutputData::RecordBatches(batches) => Self::success(
@@ -144,8 +176,16 @@ impl PrometheusJsonResponse {
                         "expected data result, but got affected rows",
                     ),
                 };
+            meta.collect_promql_annotations();
 
-            if let Some(physical_plan) = result.meta.plan {
+            if !meta.warnings.is_empty() {
+                resp.warnings = Some(meta.warnings);
+            }
+            if !meta.infos.is_empty() {
+                resp.infos = Some(meta.infos);
+            }
+
+            if let Some(physical_plan) = meta.plan {
                 let mut result_map = HashMap::new();
                 let mut tmp = vec![&mut result_map];
                 collect_plan_metrics(&physical_plan, &mut tmp);
@@ -199,6 +239,7 @@ impl PrometheusJsonResponse {
         let mut timestamp_column_index = None;
         let mut tag_column_indices = Vec::new();
         let mut first_field_column_index = None;
+        let mut native_histogram_column_index = None;
 
         let mut num_label_columns = 0;
 
@@ -224,6 +265,11 @@ impl PrometheusJsonResponse {
                 {
                     first_field_column_index = Some(i);
                 }
+                _ if native_histogram_column_index.is_none()
+                    && is_native_histogram_value_type(&column.data_type) =>
+                {
+                    native_histogram_column_index = Some(i);
+                }
                 ConcreteDataType::String(_) => {
                     tag_column_indices.push(i);
                     num_label_columns += 1;
@@ -235,13 +281,16 @@ impl PrometheusJsonResponse {
         let timestamp_column_index = timestamp_column_index.context(UnexpectedResultSnafu {
             reason: "no timestamp column found".to_string(),
         })?;
-        let first_field_column_index = first_field_column_index.context(UnexpectedResultSnafu {
-            reason: "no value column found".to_string(),
-        })?;
+        if first_field_column_index.is_none() && native_histogram_column_index.is_none() {
+            return UnexpectedResultSnafu {
+                reason: "no value column found".to_string(),
+            }
+            .fail();
+        }
 
         // Preserves the order of output tags.
         // Tag order matters, e.g., after sorc and sort_desc, the output order must be kept.
-        let mut buffer = IndexMap::<Vec<(&str, &str)>, Vec<(f64, String)>>::new();
+        let mut buffer = IndexMap::<Vec<(&str, &str)>, PromSeriesSamples>::new();
 
         let schema = batches.schema();
         for batch in batches.iter() {
@@ -258,42 +307,73 @@ impl PrometheusJsonResponse {
                 .column(timestamp_column_index)
                 .as_primitive::<TimestampMillisecondType>();
 
-            let array =
-                arrow::compute::cast(batch.column(first_field_column_index), &DataType::Float64)
-                    .context(ArrowSnafu)?;
-            let field_column = array.as_primitive::<Float64Type>();
+            let field_array = first_field_column_index
+                .map(|index| arrow::compute::cast(batch.column(index), &DataType::Float64))
+                .transpose()
+                .context(ArrowSnafu)?;
+            let field_column = field_array
+                .as_ref()
+                .map(|array| array.as_primitive::<Float64Type>());
+            let native_histogram_column = native_histogram_column_index
+                .map(|index| {
+                    batch
+                        .column(index)
+                        .as_any()
+                        .downcast_ref::<StructArray>()
+                        .with_context(|| UnexpectedResultSnafu {
+                            reason: "native histogram column is not a struct array".to_string(),
+                        })
+                })
+                .transpose()?;
 
             // assemble rows
             for row_index in 0..batch.num_rows() {
-                // retrieve value
-                if field_column.is_valid(row_index) {
-                    let v = field_column.value(row_index);
-                    // ignore all NaN values to reduce the amount of data to be sent.
-                    if v.is_nan() {
-                        continue;
+                let value = field_column.and_then(|field_column| {
+                    if !field_column.is_valid(row_index) {
+                        return None;
                     }
+                    let value = field_column.value(row_index);
+                    (!value.is_nan()).then_some((timestamp_column.value(row_index), value))
+                });
+                let histogram = native_histogram_column
+                    .and_then(|column| {
+                        read_histogram(column, row_index)
+                            .context(DataFusionSnafu)
+                            .transpose()
+                    })
+                    .transpose()?
+                    .map(|histogram| {
+                        prometheus_native_histogram_or_error(&histogram)
+                            .map(|histogram| (timestamp_column.value(row_index), histogram))
+                    })
+                    .transpose()?;
 
-                    // retrieve tags
-                    let mut tags = Vec::with_capacity(num_label_columns + 1);
-                    if let Some(metric_name) = &metric_name {
-                        tags.push((METRIC_NAME, metric_name.as_str()));
-                    }
-                    for (tag_column, tag_name) in tag_columns.iter().zip(tag_names.iter()) {
-                        // TODO(ruihang): add test for NULL tag
-                        if tag_column.is_valid(row_index) {
-                            tags.push((tag_name, tag_column.value(row_index)));
-                        }
+                if value.is_none() && histogram.is_none() {
+                    continue;
+                }
+
+                // retrieve tags
+                let mut tags = Vec::with_capacity(num_label_columns + 1);
+                if let Some(metric_name) = &metric_name {
+                    tags.push((METRIC_NAME, metric_name.as_str()));
+                }
+                for (tag_column, tag_name) in tag_columns.iter().zip(tag_names.iter()) {
+                    // TODO(ruihang): add test for NULL tag
+                    if tag_column.is_valid(row_index) {
+                        tags.push((tag_name, tag_column.value(row_index)));
                     }
+                }
 
-                    // retrieve timestamp
-                    let timestamp_millis = timestamp_column.value(row_index);
-                    let timestamp = timestamp_millis as f64 / 1000.0;
-
-                    buffer
-                        .entry(tags)
-                        .or_default()
-                        .push((timestamp, Into::<f64>::into(v).to_string()));
-                };
+                let entry = buffer.entry(tags).or_default();
+                if let Some((timestamp_millis, histogram)) = histogram {
+                    entry
+                        .histograms
+                        .push((timestamp_millis as f64 / 1000.0, histogram));
+                } else if let Some((timestamp_millis, value)) = value {
+                    entry
+                        .values
+                        .push((timestamp_millis as f64 / 1000.0, value.to_string()));
+                }
             }
         }
 
@@ -306,28 +386,46 @@ impl PrometheusJsonResponse {
         };
 
         // accumulate data into result
-        buffer.into_iter().for_each(|(tags, mut values)| {
+        buffer.into_iter().for_each(|(tags, mut samples)| {
             let metric = tags
                 .into_iter()
                 .map(|(k, v)| (k.to_string(), v.to_string()))
                 .collect::<BTreeMap<_, _>>();
             match result {
                 PromQueryResult::Vector(ref mut v) => {
+                    let histogram = samples.histograms.pop();
+                    let value = if histogram.is_none() {
+                        samples.values.pop()
+                    } else {
+                        None
+                    };
                     v.push(PromSeriesVector {
                         metric,
-                        value: values.pop(),
+                        value,
+                        histogram,
                     });
                 }
                 PromQueryResult::Matrix(ref mut v) => {
                     // sort values by timestamp
-                    if !values.is_sorted_by(|a, b| a.0 <= b.0) {
-                        values.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(Ordering::Equal));
+                    if !samples.values.is_sorted_by(|a, b| a.0 <= b.0) {
+                        samples
+                            .values
+                            .sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(Ordering::Equal));
+                    }
+                    if !samples.histograms.is_sorted_by(|a, b| a.0 <= b.0) {
+                        samples
+                            .histograms
+                            .sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(Ordering::Equal));
                     }
 
-                    v.push(PromSeriesMatrix { metric, values });
+                    v.push(PromSeriesMatrix {
+                        metric,
+                        values: samples.values,
+                        histograms: samples.histograms,
+                    });
                 }
                 PromQueryResult::Scalar(ref mut v) => {
-                    *v = values.pop();
+                    *v = samples.values.pop();
                 }
                 PromQueryResult::String(ref mut _v) => {
                     // TODO(ruihang): Not supported yet
@@ -350,3 +448,267 @@ impl PrometheusJsonResponse {
         Ok(data)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use common_query::native_histogram::{
+        NATIVE_HISTOGRAM_FIELD, NativeHistogram, Span, build_histogram_array,
+        native_histogram_value_type,
+    };
+    use common_query::{Output, OutputData, OutputMeta};
+    use common_recordbatch::{RecordBatch, RecordBatches};
+    use datatypes::data_type::ConcreteDataType;
+    use datatypes::schema::{ColumnSchema, Schema};
+    use datatypes::vectors::{
+        Float64Vector, StringVector, StructVector, TimestampMillisecondVector, VectorRef,
+    };
+
+    use super::*;
+
+    fn sample_histogram() -> NativeHistogram {
+        NativeHistogram {
+            schema: 0,
+            zero_threshold: 0.001,
+            sum: 3.0,
+            reset_hint: 0,
+            start_timestamp: Some(0),
+            custom_values: vec![],
+            positive_spans: vec![Span {
+                offset: 0,
+                length: 1,
+            }],
+            negative_spans: vec![],
+            count: 2.0,
+            zero_count: 1.0,
+            positive_buckets: vec![1.0],
+            negative_buckets: vec![],
+        }
+    }
+
+    fn histogram_vector(values: &[Option<NativeHistogram>]) -> VectorRef {
+        let histogram_array = build_histogram_array(values);
+        let histogram_array = histogram_array
+            .as_any()
+            .downcast_ref::<StructArray>()
+            .unwrap()
+            .clone();
+        let ConcreteDataType::Struct(histogram_type) = native_histogram_value_type().clone() else {
+            unreachable!("native histogram type must be a struct")
+        };
+        Arc::new(StructVector::try_new(histogram_type, histogram_array).unwrap())
+    }
+
+    #[tokio::test]
+    async fn from_query_result_preserves_warnings_and_infos() {
+        let recordbatches = RecordBatches::empty();
+        let mut meta = OutputMeta::default();
+        meta.warnings.push("warn".to_string());
+        meta.infos.push("info".to_string());
+        let response = PrometheusJsonResponse::from_query_result(
+            Ok(Output::new(OutputData::RecordBatches(recordbatches), meta)),
+            None,
+            ValueType::Vector,
+        )
+        .await;
+
+        assert_eq!(response.warnings, Some(vec!["warn".to_string()]));
+        assert_eq!(response.infos, Some(vec!["info".to_string()]));
+    }
+
+    #[test]
+    fn record_batches_to_data_serializes_aliased_native_histogram_vector() {
+        let schema = Arc::new(Schema::new(vec![
+            ColumnSchema::new(
+                "greptime_timestamp",
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                false,
+            )
+            .with_time_index(true),
+            ColumnSchema::new("job", ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new(
+                "prom_native_histogram_rate(greptime_native_histogram)",
+                native_histogram_value_type().clone(),
+                true,
+            ),
+        ]));
+        let columns: Vec<VectorRef> = vec![
+            Arc::new(TimestampMillisecondVector::from_values([1000])),
+            Arc::new(StringVector::from(vec![Some("api")])),
+            histogram_vector(&[Some(sample_histogram())]),
+        ];
+        let recordbatch = RecordBatch::new(schema.clone(), columns).unwrap();
+        let recordbatches = RecordBatches::try_new(schema, vec![recordbatch]).unwrap();
+
+        let response = PrometheusJsonResponse::record_batches_to_data(
+            recordbatches,
+            Some("http_requests".to_string()),
+            ValueType::Vector,
+        )
+        .unwrap();
+
+        let PrometheusResponse::PromData(PromData {
+            result: PromQueryResult::Vector(series),
+            ..
+        }) = response
+        else {
+            panic!("expected vector response");
+        };
+        assert_eq!(series.len(), 1);
+        assert_eq!(series[0].metric["__name__"], "http_requests");
+        assert_eq!(series[0].metric["job"], "api");
+        assert!(series[0].value.is_none());
+
+        let (timestamp, histogram) = series[0].histogram.as_ref().unwrap();
+        assert_eq!(*timestamp, 1.0);
+        assert_eq!(histogram.count, "2");
+        assert_eq!(histogram.sum, "3");
+        assert_eq!(
+            histogram.buckets,
+            vec![
+                (
+                    3,
+                    "-0.001".to_string(),
+                    "0.001".to_string(),
+                    "1".to_string()
+                ),
+                (0, "0.5".to_string(), "1".to_string(), "1".to_string()),
+            ]
+        );
+    }
+
+    #[test]
+    fn record_batches_to_data_prefers_histogram_for_instant_vector() {
+        let schema = Arc::new(Schema::new(vec![
+            ColumnSchema::new(
+                "greptime_timestamp",
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                false,
+            )
+            .with_time_index(true),
+            ColumnSchema::new("job", ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new("value", ConcreteDataType::float64_datatype(), true),
+            ColumnSchema::new(
+                NATIVE_HISTOGRAM_FIELD,
+                native_histogram_value_type().clone(),
+                true,
+            ),
+        ]));
+        let columns: Vec<VectorRef> = vec![
+            Arc::new(TimestampMillisecondVector::from_values([1000])),
+            Arc::new(StringVector::from(vec![Some("api")])),
+            Arc::new(Float64Vector::from_slice([42.0])),
+            histogram_vector(&[Some(sample_histogram())]),
+        ];
+        let recordbatch = RecordBatch::new(schema.clone(), columns).unwrap();
+        let recordbatches = RecordBatches::try_new(schema, vec![recordbatch]).unwrap();
+
+        let response = PrometheusJsonResponse::record_batches_to_data(
+            recordbatches,
+            Some("http_requests".to_string()),
+            ValueType::Vector,
+        )
+        .unwrap();
+
+        let PrometheusResponse::PromData(PromData {
+            result: PromQueryResult::Vector(series),
+            ..
+        }) = response
+        else {
+            panic!("expected vector response");
+        };
+
+        assert_eq!(series.len(), 1);
+        assert!(series[0].value.is_none());
+        assert!(series[0].histogram.is_some());
+    }
+
+    #[test]
+    fn record_batches_to_data_prefers_histogram_for_matrix_row() {
+        let schema = Arc::new(Schema::new(vec![
+            ColumnSchema::new(
+                "greptime_timestamp",
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                false,
+            )
+            .with_time_index(true),
+            ColumnSchema::new("job", ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new("value", ConcreteDataType::float64_datatype(), true),
+            ColumnSchema::new(
+                NATIVE_HISTOGRAM_FIELD,
+                native_histogram_value_type().clone(),
+                true,
+            ),
+        ]));
+        let columns: Vec<VectorRef> = vec![
+            Arc::new(TimestampMillisecondVector::from_values([1000, 2000])),
+            Arc::new(StringVector::from(vec![Some("api"), Some("api")])),
+            Arc::new(Float64Vector::from_slice([42.0, 7.0])),
+            histogram_vector(&[Some(sample_histogram()), None]),
+        ];
+        let recordbatch = RecordBatch::new(schema.clone(), columns).unwrap();
+        let recordbatches = RecordBatches::try_new(schema, vec![recordbatch]).unwrap();
+
+        let response = PrometheusJsonResponse::record_batches_to_data(
+            recordbatches,
+            Some("http_requests".to_string()),
+            ValueType::Matrix,
+        )
+        .unwrap();
+
+        let PrometheusResponse::PromData(PromData {
+            result: PromQueryResult::Matrix(series),
+            ..
+        }) = response
+        else {
+            panic!("expected matrix response");
+        };
+
+        assert_eq!(series.len(), 1);
+        assert_eq!(series[0].values, vec![(2.0, "7".to_string())]);
+        assert_eq!(series[0].histograms.len(), 1);
+        assert_eq!(series[0].histograms[0].0, 1.0);
+    }
+
+    #[test]
+    fn record_batches_to_data_errors_on_unrenderable_native_histogram() {
+        let schema = Arc::new(Schema::new(vec![
+            ColumnSchema::new(
+                "greptime_timestamp",
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                false,
+            )
+            .with_time_index(true),
+            ColumnSchema::new("job", ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new(
+                NATIVE_HISTOGRAM_FIELD,
+                native_histogram_value_type().clone(),
+                true,
+            ),
+        ]));
+        let mut histogram = sample_histogram();
+        histogram.schema = 100;
+        let columns: Vec<VectorRef> = vec![
+            Arc::new(TimestampMillisecondVector::from_values([1000])),
+            Arc::new(StringVector::from(vec![Some("api")])),
+            histogram_vector(&[Some(histogram)]),
+        ];
+        let recordbatch = RecordBatch::new(schema.clone(), columns).unwrap();
+        let recordbatches = RecordBatches::try_new(schema, vec![recordbatch]).unwrap();
+
+        let error = PrometheusJsonResponse::record_batches_to_data(
+            recordbatches,
+            Some("http_requests".to_string()),
+            ValueType::Vector,
+        )
+        .unwrap_err();
+
+        assert!(
+            error
+                .to_string()
+                .contains("native histogram cannot be converted to Prometheus buckets"),
+            "{error}"
+        );
+    }
+}

src/servers/src/prom_remote_write/v2.rs

@@ -16,9 +16,11 @@ use std::collections::hash_map::Entry;
 
 use ahash::{HashMap, HashMapExt, HashSet, HashSetExt};
 use api::greptime_proto::io::prometheus::write::v2::histogram::{Count, ZeroCount};
+use api::greptime_proto::io::prometheus::write::v2::{
+    BucketSpan, Histogram, Request, Sample, TimeSeries,
+};
 #[cfg(test)]
 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};
@@ -43,6 +45,9 @@ use crate::row_writer::{self, TableData};
 
 type PromTags = Vec<(String, String)>;
 type ResolvedSeriesLabels = (PromCtx, String, PromTags);
+const MIN_REMOTE_WRITE_V2_SCHEMA: i32 = -4;
+const MAX_REMOTE_WRITE_V2_SCHEMA: i32 = 8;
+const MAX_REDUCIBLE_REMOTE_WRITE_V2_SCHEMA: i32 = 52;
 
 pub(crate) fn decode_remote_write_v2_request(is_zstd: bool, body: Bytes) -> Result<Request> {
     let _timer = crate::metrics::METRIC_HTTP_PROM_STORE_DECODE_ELAPSED.start_timer();
@@ -279,6 +284,9 @@ fn native_histogram_column_schema() -> ColumnSchema {
 }
 
 fn native_histogram_struct_value(histogram: &Histogram) -> Result<ValueData> {
+    let uses_float_counts = native_histogram_uses_float_counts(histogram)?;
+    validate_native_histogram(histogram, uses_float_counts)?;
+
     let mut items = Vec::with_capacity(NATIVE_HISTOGRAM_FIELD_NAMES.len());
     items.extend([
         pb_value(ValueData::I32Value(histogram.schema)),
@@ -295,39 +303,39 @@ fn native_histogram_struct_value(histogram: &Histogram) -> Result<ValueData> {
         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 {
+    if uses_float_counts {
+        let count = match histogram.count.as_ref() {
+            Some(Count::CountFloat(count)) => *count,
+            _ => 0.0,
+        };
+        let zero_count = match histogram.zero_count.as_ref() {
+            Some(ZeroCount::ZeroCountFloat(zero_count)) => *zero_count,
+            _ => 0.0,
+        };
         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)),
+            pb_value(ValueData::F64Value(count)),
+            pb_value(ValueData::F64Value(zero_count)),
             f64_list_value(histogram.positive_counts.iter().copied()),
             f64_list_value(histogram.negative_counts.iter().copied()),
         ]);
     } else {
+        let count = match histogram.count.as_ref() {
+            Some(Count::CountInt(count)) => *count,
+            _ => 0,
+        };
+        let zero_count = match histogram.zero_count.as_ref() {
+            Some(ZeroCount::ZeroCountInt(zero_count)) => *zero_count,
+            _ => 0,
+        };
         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)),
+            pb_value(ValueData::U64Value(count)),
+            pb_value(ValueData::U64Value(zero_count)),
             i64_list_value(positive_buckets.iter().copied()),
             i64_list_value(negative_buckets.iter().copied()),
             null_pb_value(),
@@ -340,6 +348,229 @@ fn native_histogram_struct_value(histogram: &Histogram) -> Result<ValueData> {
     Ok(ValueData::StructValue(api::v1::StructValue { items }))
 }
 
+fn validate_native_histogram(histogram: &Histogram, uses_float_counts: bool) -> Result<()> {
+    validate_native_histogram_schema(histogram.schema)?;
+    validate_native_histogram_custom_values(histogram)?;
+
+    let (positive_buckets, negative_buckets) = if uses_float_counts {
+        (
+            histogram.positive_counts.len(),
+            histogram.negative_counts.len(),
+        )
+    } else {
+        (
+            histogram.positive_deltas.len(),
+            histogram.negative_deltas.len(),
+        )
+    };
+    let custom_max_index =
+        (histogram.schema == CUSTOM_BUCKETS_SCHEMA).then_some(histogram.custom_values.len() as i32);
+    validate_native_histogram_spans(
+        "positive",
+        &histogram.positive_spans,
+        positive_buckets,
+        custom_max_index,
+    )?;
+    validate_native_histogram_spans(
+        "negative",
+        &histogram.negative_spans,
+        negative_buckets,
+        None,
+    )?;
+
+    if histogram.schema == CUSTOM_BUCKETS_SCHEMA {
+        ensure!(
+            histogram.zero_threshold == 0.0 && native_histogram_zero_count_is_zero(histogram),
+            error::InvalidPromRemoteRequestSnafu {
+                msg: "remote write v2 custom native histogram must not use a zero bucket"
+                    .to_string(),
+            }
+        );
+        ensure!(
+            histogram.negative_spans.is_empty()
+                && histogram.negative_deltas.is_empty()
+                && histogram.negative_counts.is_empty(),
+            error::InvalidPromRemoteRequestSnafu {
+                msg: "remote write v2 custom native histogram must not use negative buckets"
+                    .to_string(),
+            }
+        );
+    }
+
+    Ok(())
+}
+
+fn validate_native_histogram_schema(schema: i32) -> Result<()> {
+    if schema == CUSTOM_BUCKETS_SCHEMA
+        || (MIN_REMOTE_WRITE_V2_SCHEMA..=MAX_REMOTE_WRITE_V2_SCHEMA).contains(&schema)
+    {
+        return Ok(());
+    }
+
+    if (MAX_REMOTE_WRITE_V2_SCHEMA + 1..=MAX_REDUCIBLE_REMOTE_WRITE_V2_SCHEMA).contains(&schema) {
+        error::InvalidPromRemoteRequestSnafu {
+            msg: format!(
+                "remote write v2 native histogram schema {schema} must be reduced before ingestion"
+            ),
+        }
+        .fail()
+    } else {
+        error::InvalidPromRemoteRequestSnafu {
+            msg: format!("remote write v2 native histogram schema {schema} is unsupported"),
+        }
+        .fail()
+    }
+}
+
+fn validate_native_histogram_custom_values(histogram: &Histogram) -> Result<()> {
+    if histogram.schema != CUSTOM_BUCKETS_SCHEMA {
+        ensure!(
+            histogram.custom_values.is_empty(),
+            error::InvalidPromRemoteRequestSnafu {
+                msg: "remote write v2 standard native histogram must not use custom_values"
+                    .to_string(),
+            }
+        );
+        return Ok(());
+    }
+
+    for value in &histogram.custom_values {
+        ensure!(
+            !value.is_nan() && *value != f64::INFINITY,
+            error::InvalidPromRemoteRequestSnafu {
+                msg: "remote write v2 custom native histogram custom_values must not contain +Inf or NaN"
+                    .to_string(),
+            }
+        );
+    }
+    for values in histogram.custom_values.windows(2) {
+        ensure!(
+            values[0] < values[1],
+            error::InvalidPromRemoteRequestSnafu {
+                msg: "remote write v2 custom native histogram custom_values must be sorted"
+                    .to_string(),
+            }
+        );
+    }
+
+    Ok(())
+}
+
+fn validate_native_histogram_spans(
+    name: &str,
+    spans: &[BucketSpan],
+    bucket_count: usize,
+    custom_max_index: Option<i32>,
+) -> Result<()> {
+    let span_len = spans
+        .iter()
+        .try_fold(0usize, |sum, span| sum.checked_add(span.length as usize))
+        .context(error::InvalidPromRemoteRequestSnafu {
+            msg: format!("remote write v2 native histogram {name} spans overflow"),
+        })?;
+    ensure!(
+        span_len == bucket_count,
+        error::InvalidPromRemoteRequestSnafu {
+            msg: format!(
+                "remote write v2 native histogram {name} spans describe {span_len} buckets, found {bucket_count}"
+            ),
+        }
+    );
+
+    let mut current_index = 0i32;
+    let mut first = true;
+    for span in spans {
+        current_index = if first {
+            first = false;
+            span.offset
+        } else {
+            current_index.checked_add(span.offset).context(
+                error::InvalidPromRemoteRequestSnafu {
+                    msg: format!(
+                        "remote write v2 native histogram {name} span index overflows i32"
+                    ),
+                },
+            )?
+        };
+
+        for _ in 0..span.length {
+            if let Some(max_index) = custom_max_index {
+                ensure!(
+                    (0..=max_index).contains(&current_index),
+                    error::InvalidPromRemoteRequestSnafu {
+                        msg: format!(
+                            "remote write v2 custom native histogram {name} bucket index {current_index} is out of range"
+                        ),
+                    }
+                );
+            }
+            current_index =
+                current_index
+                    .checked_add(1)
+                    .context(error::InvalidPromRemoteRequestSnafu {
+                        msg: format!(
+                            "remote write v2 native histogram {name} span index overflows i32"
+                        ),
+                    })?;
+        }
+    }
+
+    Ok(())
+}
+
+fn native_histogram_zero_count_is_zero(histogram: &Histogram) -> bool {
+    match histogram.zero_count.as_ref() {
+        Some(ZeroCount::ZeroCountInt(zero_count)) => *zero_count == 0,
+        Some(ZeroCount::ZeroCountFloat(zero_count)) => *zero_count == 0.0,
+        None => true,
+    }
+}
+
+fn native_histogram_uses_float_counts(histogram: &Histogram) -> Result<bool> {
+    let uses_float_count = matches!(histogram.count, Some(Count::CountFloat(_)))
+        || matches!(histogram.zero_count, Some(ZeroCount::ZeroCountFloat(_)));
+    let uses_int_count = matches!(histogram.count, Some(Count::CountInt(_)))
+        || matches!(histogram.zero_count, Some(ZeroCount::ZeroCountInt(_)));
+    let uses_float_buckets =
+        !histogram.positive_counts.is_empty() || !histogram.negative_counts.is_empty();
+    let uses_int_buckets =
+        !histogram.positive_deltas.is_empty() || !histogram.negative_deltas.is_empty();
+
+    if matches!(
+        (&histogram.count, &histogram.zero_count),
+        (Some(Count::CountInt(_)), Some(ZeroCount::ZeroCountFloat(_)))
+            | (Some(Count::CountFloat(_)), Some(ZeroCount::ZeroCountInt(_)))
+    ) {
+        return error::InvalidPromRemoteRequestSnafu {
+            msg: "remote write v2 native histogram count and zero_count must use the same integer or float family".to_string(),
+        }
+        .fail();
+    }
+
+    ensure!(
+        !(uses_float_buckets && uses_int_buckets),
+        error::InvalidPromRemoteRequestSnafu {
+            msg: "remote write v2 native histogram bucket counts must use either integer deltas or float counts".to_string(),
+        }
+    );
+    ensure!(
+        !(uses_float_count && uses_int_buckets),
+        error::InvalidPromRemoteRequestSnafu {
+            msg: "remote write v2 float native histogram must not use integer bucket deltas"
+                .to_string(),
+        }
+    );
+    ensure!(
+        !(uses_int_count && uses_float_buckets),
+        error::InvalidPromRemoteRequestSnafu {
+            msg: "remote write v2 integer native histogram must not use float bucket counts"
+                .to_string(),
+        }
+    );
+
+    Ok(uses_float_count || uses_float_buckets)
+}
+
 fn pb_value(value_data: ValueData) -> Value {
     optional_pb_value(Some(value_data))
 }
@@ -859,6 +1090,126 @@ mod tests {
             "conflicts with an internal native histogram label",
         ));
 
+        cases.push((
+            "int count with float zero count",
+            request_with_histogram(Histogram {
+                count: Some(Count::CountInt(1)),
+                zero_count: Some(ZeroCount::ZeroCountFloat(0.5)),
+                ..Default::default()
+            }),
+            "count and zero_count must use the same integer or float family",
+        ));
+
+        cases.push((
+            "float count with int zero count",
+            request_with_histogram(Histogram {
+                count: Some(Count::CountFloat(1.0)),
+                zero_count: Some(ZeroCount::ZeroCountInt(1)),
+                ..Default::default()
+            }),
+            "count and zero_count must use the same integer or float family",
+        ));
+
+        cases.push((
+            "reducible schema",
+            request_with_histogram(Histogram {
+                schema: 9,
+                ..Default::default()
+            }),
+            "schema 9 must be reduced before ingestion",
+        ));
+
+        cases.push((
+            "unsupported schema",
+            request_with_histogram(Histogram {
+                schema: 53,
+                ..Default::default()
+            }),
+            "schema 53 is unsupported",
+        ));
+
+        cases.push((
+            "standard schema with custom values",
+            request_with_histogram(Histogram {
+                schema: 1,
+                custom_values: vec![1.0],
+                ..Default::default()
+            }),
+            "standard native histogram must not use custom_values",
+        ));
+
+        cases.push((
+            "custom values with inf",
+            request_with_histogram(Histogram {
+                schema: CUSTOM_BUCKETS_SCHEMA,
+                custom_values: vec![f64::INFINITY],
+                ..Default::default()
+            }),
+            "custom_values must not contain +Inf or NaN",
+        ));
+
+        cases.push((
+            "custom values not sorted",
+            request_with_histogram(Histogram {
+                schema: CUSTOM_BUCKETS_SCHEMA,
+                custom_values: vec![2.0, 1.0],
+                ..Default::default()
+            }),
+            "custom_values must be sorted",
+        ));
+
+        cases.push((
+            "custom schema with zero bucket",
+            request_with_histogram(Histogram {
+                schema: CUSTOM_BUCKETS_SCHEMA,
+                zero_count: Some(ZeroCount::ZeroCountInt(1)),
+                ..Default::default()
+            }),
+            "custom native histogram must not use a zero bucket",
+        ));
+
+        cases.push((
+            "custom schema with negative buckets",
+            request_with_histogram(Histogram {
+                schema: CUSTOM_BUCKETS_SCHEMA,
+                negative_spans: vec![BucketSpan {
+                    offset: -1,
+                    length: 1,
+                }],
+                negative_deltas: vec![1],
+                ..Default::default()
+            }),
+            "custom native histogram must not use negative buckets",
+        ));
+
+        cases.push((
+            "span count mismatch",
+            request_with_histogram(Histogram {
+                positive_spans: vec![BucketSpan {
+                    offset: 0,
+                    length: 2,
+                }],
+                positive_deltas: vec![1],
+                ..Default::default()
+            }),
+            "positive spans describe 2 buckets, found 1",
+        ));
+
+        cases.push((
+            "custom span index out of range",
+            request_with_histogram(Histogram {
+                schema: CUSTOM_BUCKETS_SCHEMA,
+                custom_values: vec![1.0],
+                positive_spans: vec![BucketSpan {
+                    offset: 2,
+                    length: 1,
+                }],
+                positive_deltas: vec![1],
+                ..Default::default()
+            }),
+            "positive bucket index 2 is out of range",
+        ));
+
         for (name, request, expected) in cases {
             assert_invalid(name, request, expected);
         }
@@ -1115,6 +1466,13 @@ mod tests {
         )
     }
 
+    fn request_with_histogram(histogram: Histogram) -> Request {
+        test_util::request_with_labels_and_histograms(
+            vec![(METRIC_NAME_LABEL, "metric")],
+            vec![histogram],
+        )
+    }
+
     fn assert_invalid(name: &str, request: Request, expected: &str) {
         let err = match into_write_requests(request) {
             Ok(_) => panic!("{name}: expected invalid request error"),

tests-integration/tests/grpc.rs

@@ -872,6 +872,7 @@ pub async fn test_prom_gateway_query(store_type: StorageType) {
                 .into_iter()
                 .collect(),
                 value: Some((5.0, "1".to_string())),
+                ..Default::default()
             },
             PromSeriesVector {
                 metric: [
@@ -881,6 +882,7 @@ pub async fn test_prom_gateway_query(store_type: StorageType) {
                 .into_iter()
                 .collect(),
                 value: Some((5.0, "2".to_string())),
+                ..Default::default()
             },
         ]
     );
@@ -932,6 +934,7 @@ pub async fn test_prom_gateway_query(store_type: StorageType) {
                 .into_iter()
                 .collect(),
                 values: vec![(5.0, "1".to_string()), (10.0, "1".to_string())],
+                ..Default::default()
             },
             PromSeriesMatrix {
                 metric: [
@@ -941,6 +944,7 @@ pub async fn test_prom_gateway_query(store_type: StorageType) {
                 .into_iter()
                 .collect(),
                 values: vec![(5.0, "2".to_string()), (10.0, "2".to_string())],
+                ..Default::default()
             },
         ]
     );
@@ -973,6 +977,7 @@ pub async fn test_prom_gateway_query(store_type: StorageType) {
         error: None,
         error_type: None,
         warnings: None,
+        infos: None,
         resp_metrics: Default::default(),
         status_code: None,
     };

tests-integration/tests/http.rs

@@ -2455,7 +2455,6 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag
                 reset_hint: 2,
                 timestamp: 3000,
                 start_timestamp: 1500,
-                custom_values: vec![0.5, 1.5],
                 ..Default::default()
             },
             Histogram {
@@ -2508,10 +2507,58 @@ pub async fn test_prometheus_remote_write_v2_native_histogram(store_type: Storag
         "prometheus_remote_write_v2_native_histogram_rows",
         &client,
         "select greptime_timestamp, greptime_native_histogram, job, instance from remote_write_v2_latency_seconds order by greptime_timestamp;",
-        "[[3000,{\"count_f64\":null,\"count_u64\":4,\"custom_values\":[0.5,1.5],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[1],\"negative_span_lengths\":[1],\"negative_span_offsets\":[-2],\"positive_buckets_f64\":[],\"positive_buckets_i64\":[1,3,2],\"positive_span_lengths\":[3],\"positive_span_offsets\":[0],\"reset_hint\":2,\"schema\":1,\"start_timestamp\":1500,\"sum\":10.0,\"zero_count_f64\":null,\"zero_count_u64\":1,\"zero_threshold\":0.001},\"api\",\"localhost:9090\"],[4000,{\"count_f64\":3.5,\"count_u64\":null,\"custom_values\":[],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[],\"negative_span_lengths\":[],\"negative_span_offsets\":[],\"positive_buckets_f64\":[2.0,3.5],\"positive_buckets_i64\":[],\"positive_span_lengths\":[2],\"positive_span_offsets\":[3],\"reset_hint\":3,\"schema\":2,\"start_timestamp\":2500,\"sum\":20.0,\"zero_count_f64\":0.5,\"zero_count_u64\":null,\"zero_threshold\":0.002},\"api\",\"localhost:9090\"]]",
+        "[[3000,{\"count_f64\":null,\"count_u64\":4,\"custom_values\":[],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[1],\"negative_span_lengths\":[1],\"negative_span_offsets\":[-2],\"positive_buckets_f64\":[],\"positive_buckets_i64\":[1,3,2],\"positive_span_lengths\":[3],\"positive_span_offsets\":[0],\"reset_hint\":2,\"schema\":1,\"start_timestamp\":1500,\"sum\":10.0,\"zero_count_f64\":null,\"zero_count_u64\":1,\"zero_threshold\":0.001},\"api\",\"localhost:9090\"],[4000,{\"count_f64\":3.5,\"count_u64\":null,\"custom_values\":[],\"negative_buckets_f64\":[],\"negative_buckets_i64\":[],\"negative_span_lengths\":[],\"negative_span_offsets\":[],\"positive_buckets_f64\":[2.0,3.5],\"positive_buckets_i64\":[],\"positive_span_lengths\":[2],\"positive_span_offsets\":[3],\"reset_hint\":3,\"schema\":2,\"start_timestamp\":2500,\"sum\":20.0,\"zero_count_f64\":0.5,\"zero_count_u64\":null,\"zero_threshold\":0.002},\"api\",\"localhost:9090\"]]",
     )
     .await;
 
+    let res = client
+        .get("/v1/prometheus/api/v1/query?query=histogram_count(remote_write_v2_latency_seconds)&time=4")
+        .send()
+        .await;
+    assert_eq!(res.status(), StatusCode::OK);
+    let body = serde_json::from_str::<Value>(&res.text().await).unwrap();
+    assert_eq!(body["status"], "success");
+    assert_eq!(body["data"]["resultType"], "vector");
+    let series = body["data"]["result"].as_array().unwrap();
+    assert_eq!(series.len(), 1);
+    let value = series[0]["value"].as_array().unwrap();
+    assert_eq!(value[0].as_f64(), Some(4.0));
+    assert_eq!(value[1], "3.5");
+
+    let res = client
+        .get("/v1/prometheus/api/v1/query?query=remote_write_v2_latency_seconds&time=4")
+        .send()
+        .await;
+    assert_eq!(res.status(), StatusCode::OK);
+    let body = serde_json::from_str::<Value>(&res.text().await).unwrap();
+    assert_eq!(body["status"], "success");
+    assert_eq!(body["data"]["resultType"], "vector");
+    let series = body["data"]["result"].as_array().unwrap();
+    assert_eq!(series.len(), 1);
+    assert!(series[0].get("value").is_none());
+    let histogram = series[0]["histogram"].as_array().unwrap();
+    assert_eq!(histogram[0].as_f64(), Some(4.0));
+    assert_eq!(histogram[1]["count"], "3.5");
+    assert_eq!(histogram[1]["sum"], "20");
+
+    let res = client
+        .get("/v1/prometheus/api/v1/query_range?query=remote_write_v2_latency_seconds&start=3&end=4&step=1")
+        .send()
+        .await;
+    assert_eq!(res.status(), StatusCode::OK);
+    let body = serde_json::from_str::<Value>(&res.text().await).unwrap();
+    assert_eq!(body["status"], "success");
+    assert_eq!(body["data"]["resultType"], "matrix");
+    let series = body["data"]["result"].as_array().unwrap();
+    assert_eq!(series.len(), 1);
+    assert!(series[0].get("values").is_none());
+    let histograms = series[0]["histograms"].as_array().unwrap();
+    assert_eq!(histograms.len(), 2);
+    assert_eq!(histograms[0][0].as_f64(), Some(3.0));
+    assert_eq!(histograms[0][1]["count"], "4");
+    assert_eq!(histograms[1][0].as_f64(), Some(4.0));
+    assert_eq!(histograms[1][1]["count"], "3.5");
+
     validate_data(
         "prometheus_remote_write_v2_native_histogram_table_created",
         &client,