diff --git a/src/datatypes/src/schema/column_schema.rs b/src/datatypes/src/schema/column_schema.rs
index 2479f4fc41..d3ecd90116 100644
--- a/src/datatypes/src/schema/column_schema.rs
+++ b/src/datatypes/src/schema/column_schema.rs
@@ -1230,6 +1230,7 @@ mod tests {
     use arrow::datatypes::{DataType as ArrowDataType, TimeUnit};
 
     use super::*;
+    use crate::types::{StructField, StructType};
     use crate::value::Value;
     use crate::vectors::Int32Vector;
 
@@ -1343,6 +1344,24 @@ mod tests {
         assert!(v.only_null());
     }
 
+    #[test]
+    fn test_column_schema_create_default_null_struct_with_list() {
+        let list_type =
+            ConcreteDataType::list_datatype(Arc::new(ConcreteDataType::int32_datatype()));
+        let struct_type =
+            ConcreteDataType::struct_datatype(StructType::new(Arc::new(vec![StructField::new(
+                "values".to_string(),
+                list_type,
+                true,
+            )])));
+        let column_schema = ColumnSchema::new("test", struct_type, true);
+
+        let v = column_schema.create_default_vector(5).unwrap().unwrap();
+
+        assert_eq!(5, v.len());
+        assert!(v.only_null());
+    }
+
     #[test]
     fn test_column_schema_no_default() {
         let column_schema = ColumnSchema::new("test", ConcreteDataType::int32_datatype(), false);
diff --git a/src/datatypes/src/vectors/operations.rs b/src/datatypes/src/vectors/operations.rs
index 4eb74b0307..1dc9e21a0d 100644
--- a/src/datatypes/src/vectors/operations.rs
+++ b/src/datatypes/src/vectors/operations.rs
@@ -89,7 +89,25 @@ macro_rules! impl_scalar_vector_op {
     )+};
 }
 
-impl_scalar_vector_op!(BinaryVector, BooleanVector, ListVector, StringVector);
+impl_scalar_vector_op!(BinaryVector, BooleanVector, StringVector);
+
+impl VectorOp for ListVector {
+    fn replicate(&self, offsets: &[usize]) -> VectorRef {
+        replicate::replicate_list(self, offsets)
+    }
+
+    fn filter(&self, filter: &BooleanVector) -> Result<VectorRef> {
+        filter::filter_non_constant!(self, ListVector, filter)
+    }
+
+    fn cast(&self, to_type: &ConcreteDataType) -> Result<VectorRef> {
+        cast::cast_non_constant!(self, to_type)
+    }
+
+    fn take(&self, indices: &UInt32Vector) -> Result<VectorRef> {
+        take::take_indices!(self, ListVector, indices)
+    }
+}
 
 impl VectorOp for Decimal128Vector {
     fn replicate(&self, offsets: &[usize]) -> VectorRef {
diff --git a/src/datatypes/src/vectors/operations/replicate.rs b/src/datatypes/src/vectors/operations/replicate.rs
index 9e0591efa3..dee0b3855b 100644
--- a/src/datatypes/src/vectors/operations/replicate.rs
+++ b/src/datatypes/src/vectors/operations/replicate.rs
@@ -16,6 +16,7 @@ use crate::prelude::*;
 pub(crate) use crate::vectors::decimal::replicate_decimal128;
 pub(crate) use crate::vectors::null::replicate_null;
 pub(crate) use crate::vectors::primitive::replicate_primitive;
+use crate::vectors::{ListVector, ListVectorBuilder};
 
 pub(crate) fn replicate_scalar<C: ScalarVector>(c: &C, offsets: &[usize]) -> VectorRef {
     assert_eq!(offsets.len(), c.len());
@@ -36,6 +37,26 @@ pub(crate) fn replicate_scalar<C: ScalarVector>(c: &C, offsets: &[usize]) -> Vec
     builder.to_vector()
 }
 
+pub(crate) fn replicate_list(c: &ListVector, offsets: &[usize]) -> VectorRef {
+    assert_eq!(offsets.len(), c.len());
+
+    if offsets.is_empty() {
+        return c.slice(0, 0);
+    }
+    let mut builder =
+        ListVectorBuilder::with_type_capacity(c.item_type(), *offsets.last().unwrap());
+
+    let mut previous_offset = 0;
+    for (i, offset) in offsets.iter().enumerate() {
+        let data = c.get_data(i);
+        for _ in previous_offset..*offset {
+            builder.push(data.clone());
+        }
+        previous_offset = *offset;
+    }
+    builder.to_vector()
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
@@ -45,8 +66,11 @@ mod tests {
     use paste::paste;
 
     use super::*;
+    use crate::value::{ListValue, ListValueRef};
     use crate::vectors::constant::ConstantVector;
-    use crate::vectors::{Decimal128Vector, Int32Vector, NullVector, StringVector, VectorOp};
+    use crate::vectors::{
+        Decimal128Vector, Int32Vector, ListVectorBuilder, NullVector, StringVector, VectorOp,
+    };
 
     #[test]
     fn test_replicate_primitive() {
@@ -93,6 +117,26 @@ mod tests {
         assert_eq!(expect, v);
     }
 
+    #[test]
+    fn test_replicate_list() {
+        let item_type = Arc::new(ConcreteDataType::int32_datatype());
+        let first = ListValue::new(vec![Value::Int32(1), Value::Int32(2)], item_type.clone());
+        let second = ListValue::new(vec![Value::Int32(3)], item_type.clone());
+        let mut builder = ListVectorBuilder::with_type_capacity(item_type, 2);
+        builder.push(Some(ListValueRef::Ref { val: &first }));
+        builder.push(Some(ListValueRef::Ref { val: &second }));
+        let v = builder.finish();
+
+        let v = v.replicate(&[1, 3]);
+        let mut expect_builder =
+            ListVectorBuilder::with_type_capacity(Arc::new(ConcreteDataType::int32_datatype()), 3);
+        expect_builder.push(Some(ListValueRef::Ref { val: &first }));
+        expect_builder.push(Some(ListValueRef::Ref { val: &second }));
+        expect_builder.push(Some(ListValueRef::Ref { val: &second }));
+        let expect = expect_builder.to_vector();
+        assert_eq!(expect, v);
+    }
+
     #[test]
     fn test_replicate_constant() {
         let v = Arc::new(StringVector::from_slice(&["hello"]));
diff --git a/src/datatypes/src/vectors/struct_vector.rs b/src/datatypes/src/vectors/struct_vector.rs
index 79abef255b..1a5495522e 100644
--- a/src/datatypes/src/vectors/struct_vector.rs
+++ b/src/datatypes/src/vectors/struct_vector.rs
@@ -144,22 +144,25 @@ impl Vector for StructVector {
 
 impl VectorOp for StructVector {
     fn replicate(&self, offsets: &[usize]) -> VectorRef {
-        let column_arrays = self
-            .array
-            .columns()
-            .iter()
-            .map(|col| {
-                let vector = Helper::try_into_vector(col)
-                    .expect("Failed to replicate struct vector columns");
-                vector.replicate(offsets).to_arrow_array()
-            })
-            .collect::<Vec<_>>();
-        let replicated_array = StructArray::new(
-            self.array.fields().clone(),
-            column_arrays,
-            self.array.nulls().cloned(),
+        assert_eq!(offsets.len(), self.len());
+
+        if offsets.is_empty() {
+            return self.slice(0, 0);
+        }
+        let mut builder = StructVectorBuilder::with_type_and_capacity(
+            self.fields.clone(),
+            *offsets.last().unwrap(),
         );
-        Arc::new(StructVector::try_new(self.fields.clone(), replicated_array).unwrap())
+
+        let mut previous_offset = 0;
+        for (i, offset) in offsets.iter().enumerate() {
+            let data = self.get_data(i);
+            for _ in previous_offset..*offset {
+                builder.push(data.clone());
+            }
+            previous_offset = *offset;
+        }
+        builder.to_vector()
     }
 
     fn cast(&self, _to_type: &ConcreteDataType) -> Result<VectorRef> {
diff --git a/src/servers/src/http.rs b/src/servers/src/http.rs
index 10ffba39db..2350d457bf 100644
--- a/src/servers/src/http.rs
+++ b/src/servers/src/http.rs
@@ -59,8 +59,8 @@ use crate::http::influxdb::{influxdb_health, influxdb_ping, influxdb_write_v1, i
 use crate::http::otlp::OtlpState;
 use crate::http::prom_store::PromStoreState;
 use crate::http::prometheus::{
-    build_info_query, format_query, instant_query, label_values_query, labels_query, parse_query,
-    range_query, series_query,
+    build_info_query, format_query, instant_query, label_values_query, labels_query,
+    metadata_query, parse_query, range_query, series_query,
 };
 use crate::http::result::arrow_result::ArrowResponse;
 use crate::http::result::csv_result::CsvResponse;
@@ -1164,6 +1164,7 @@ impl HttpServer {
             .route("/query", routing::post(instant_query).get(instant_query))
             .route("/query_range", routing::post(range_query).get(range_query))
             .route("/labels", routing::post(labels_query).get(labels_query))
+            .route("/metadata", routing::get(metadata_query))
             .route("/series", routing::post(series_query).get(series_query))
             .route("/parse_query", routing::post(parse_query).get(parse_query))
             .route(
diff --git a/src/servers/src/http/prometheus.rs b/src/servers/src/http/prometheus.rs
index a81131f11a..0db5872302 100644
--- a/src/servers/src/http/prometheus.rs
+++ b/src/servers/src/http/prometheus.rs
@@ -34,6 +34,7 @@ use common_catalog::parse_catalog_and_schema_from_db_string;
 use common_decimal::Decimal128;
 use common_error::ext::ErrorExt;
 use common_error::status_code::StatusCode;
+use common_query::native_histogram::is_native_histogram_value_type;
 use common_query::{Output, OutputData};
 use common_recordbatch::{RecordBatch, RecordBatches};
 use common_telemetry::{debug, tracing};
@@ -63,6 +64,8 @@ use snafu::{Location, OptionExt, ResultExt};
 use store_api::metric_engine_consts::{
     DATA_SCHEMA_TABLE_ID_COLUMN_NAME, DATA_SCHEMA_TSID_COLUMN_NAME, LOGICAL_TABLE_METADATA_KEY,
 };
+use table::metadata::TableInfo;
+use table::requests::{SEMANTIC_METRIC_TYPE, SEMANTIC_METRIC_UNIT};
 use table::TableRef;
 
 pub use super::result::prometheus_resp::PrometheusJsonResponse;
@@ -128,6 +131,14 @@ pub struct PromData {
     pub result: PromQueryResult,
 }
 
+#[derive(Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct PromMetadata {
+    #[serde(rename = "type")]
+    pub metric_type: String,
+    pub unit: String,
+    pub help: String,
+}
+
 /// A "holder" for the reference([Arc]) to a column name,
 /// to help avoiding cloning [String]s when used as a [HashMap] key.
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
@@ -145,6 +156,7 @@ pub enum PrometheusResponse {
     PromData(PromData),
     Labels(Vec<String>),
     Series(Vec<HashMap<Column, String>>),
+    Metadata(BTreeMap<String, Vec<PromMetadata>>),
     LabelValues(Vec<String>),
     FormatQuery(String),
     BuildInfo(OwnedBuildInfo),
@@ -201,6 +213,14 @@ pub struct FormatQuery {
     query: Option<String>,
 }
 
+#[derive(Debug, Default, Serialize, Deserialize)]
+pub struct MetadataQuery {
+    db: Option<String>,
+    limit: Option<usize>,
+    limit_per_metric: Option<usize>,
+    metric: Option<String>,
+}
+
 #[axum_macros::debug_handler]
 #[tracing::instrument(
     skip_all,
@@ -238,6 +258,31 @@ pub async fn build_info_query() -> PrometheusJsonResponse {
     PrometheusJsonResponse::success(PrometheusResponse::BuildInfo(build_info.into()))
 }
 
+#[axum_macros::debug_handler]
+#[tracing::instrument(
+    skip_all,
+    fields(protocol = "prometheus", request_type = "metadata_query")
+)]
+pub async fn metadata_query(
+    State(handler): State<PrometheusHandlerRef>,
+    Query(params): Query<MetadataQuery>,
+    Extension(mut query_ctx): Extension<QueryContext>,
+) -> PrometheusJsonResponse {
+    let (catalog, schema) = get_catalog_schema(&params.db, &query_ctx);
+    try_update_catalog_schema(&mut query_ctx, &catalog, &schema);
+
+    let metadata = match retrieve_metric_metadata(&query_ctx, handler.catalog_manager(), &params)
+        .await
+    {
+        Ok(metadata) => metadata,
+        Err(err) => {
+            return PrometheusJsonResponse::error(StatusCode::InvalidArguments, err.to_string());
+        }
+    };
+
+    PrometheusJsonResponse::success(PrometheusResponse::Metadata(metadata))
+}
+
 #[derive(Debug, Default, Serialize, Deserialize)]
 pub struct InstantQuery {
     query: Option<String>,
@@ -1157,15 +1202,15 @@ fn record_batches_to_labels_name(
     labels: &mut HashSet<String>,
 ) -> Result<()> {
     let mut column_indices = Vec::new();
-    let mut field_column_indices = Vec::new();
+    let mut value_column_indices = Vec::new();
     for (i, column) in batches.schema().column_schemas().iter().enumerate() {
-        if let ConcreteDataType::Float64(_) = column.data_type {
-            field_column_indices.push(i);
+        if is_prometheus_value_column(&column.data_type) {
+            value_column_indices.push(i);
         }
         column_indices.push(i);
     }
 
-    if field_column_indices.is_empty() {
+    if value_column_indices.is_empty() {
         return Err(Error::Internal {
             err_msg: "no value column found".to_string(),
         });
@@ -1177,21 +1222,18 @@ fn record_batches_to_labels_name(
             .map(|c| batches.schema().column_name_by_index(*c).to_string())
             .collect::<Vec<_>>();
 
-        let field_columns = field_column_indices
+        let value_columns = value_column_indices
             .iter()
-            .map(|i| {
-                let column = batch.column(*i);
-                column.as_primitive::<Float64Type>()
-            })
+            .map(|i| batch.column(*i))
             .collect::<Vec<_>>();
 
         for row_index in 0..batch.num_rows() {
-            // if all field columns are null, skip this row
-            if field_columns.iter().all(|c| c.is_null(row_index)) {
+            // if all value columns are null, skip this row
+            if value_columns.iter().all(|c| c.is_null(row_index)) {
                 continue;
             }
 
-            // if a field is not null, record the tag name and return
+            // if a value is not null, record the tag name and return
             names.iter().for_each(|name| {
                 let _ = labels.insert(name.clone());
             });
@@ -1201,6 +1243,10 @@ fn record_batches_to_labels_name(
     Ok(())
 }
 
+fn is_prometheus_value_column(data_type: &ConcreteDataType) -> bool {
+    matches!(data_type, ConcreteDataType::Float64(_)) || is_native_histogram_value_type(data_type)
+}
+
 pub(crate) fn retrieve_metric_name_and_result_type(
     promql_expr: &PromqlExpr,
 ) -> (Option<String>, ValueType) {
@@ -1861,6 +1907,83 @@ async fn retrieve_table_names(
     Ok(table_names)
 }
 
+async fn retrieve_metric_metadata(
+    query_ctx: &QueryContext,
+    manager: CatalogManagerRef,
+    params: &MetadataQuery,
+) -> Result<BTreeMap<String, Vec<PromMetadata>>> {
+    let mut metadata = BTreeMap::new();
+    if params.limit == Some(0) {
+        return Ok(metadata);
+    }
+
+    let catalog = query_ctx.current_catalog();
+    let schema = query_ctx.current_schema();
+    let mut tables_stream = manager.tables(catalog, &schema, Some(query_ctx));
+
+    while let Some(table) = tables_stream.next().await {
+        let table = table?;
+        let table_info = table.table_info();
+        if !is_prometheus_metric_table(table_info.as_ref()) {
+            continue;
+        }
+
+        if let Some(metric) = &params.metric
+            && table_info.name.as_str() != metric
+        {
+            continue;
+        }
+
+        metadata.insert(
+            table_info.name.clone(),
+            vec![prometheus_metadata_from_table(table_info.as_ref())],
+        );
+        if let Some(limit) = params.limit
+            && metadata.len() >= limit
+        {
+            break;
+        }
+    }
+
+    Ok(metadata)
+}
+
+fn is_prometheus_metric_table(table_info: &TableInfo) -> bool {
+    table_info
+        .meta
+        .options
+        .extra_options
+        .contains_key(LOGICAL_TABLE_METADATA_KEY)
+}
+
+fn prometheus_metadata_from_table(table_info: &TableInfo) -> PromMetadata {
+    let options = &table_info.meta.options.extra_options;
+    let metric_type = options
+        .get(SEMANTIC_METRIC_TYPE)
+        .cloned()
+        .or_else(|| table_has_native_histogram_value(table_info).then_some("histogram".to_string()))
+        .unwrap_or_default();
+    let unit = options
+        .get(SEMANTIC_METRIC_UNIT)
+        .cloned()
+        .unwrap_or_default();
+
+    PromMetadata {
+        metric_type,
+        unit,
+        help: String::new(),
+    }
+}
+
+fn table_has_native_histogram_value(table_info: &TableInfo) -> bool {
+    table_info
+        .meta
+        .schema
+        .column_schemas()
+        .iter()
+        .any(|column| is_native_histogram_value_type(&column.data_type))
+}
+
 async fn retrieve_field_names(
     query_ctx: &QueryContext,
     manager: CatalogManagerRef,
@@ -2162,11 +2285,12 @@ mod tests {
     use catalog::memory::MemoryCatalogManager;
     use catalog::{RegisterSchemaRequest, RegisterTableRequest};
     use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
+    use common_query::native_histogram::native_histogram_value_type;
     use datatypes::prelude::ConcreteDataType;
     use datatypes::schema::{ColumnSchema, Schema};
     use promql_parser::parser::value::ValueType;
     use table::metadata::{TableInfoBuilder, TableMetaBuilder, TableType, TableVersion};
-    use table::requests::TableOptions;
+    use table::requests::{SEMANTIC_METRIC_TYPE, SEMANTIC_METRIC_UNIT, TableOptions};
     use table::test_util::EmptyTable;
     use table::test_util::table_info::test_table_info;
 
@@ -3138,4 +3262,82 @@ mod tests {
             (PermissionTableTargets::resolved(Vec::new()), 2)
         );
     }
+
+    #[test]
+    fn test_record_batches_to_labels_name_accepts_native_histogram_value() {
+        let schema = Arc::new(Schema::new(vec![
+            ColumnSchema::new("host", ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new(
+                "greptime_native_histogram",
+                native_histogram_value_type().clone(),
+                true,
+            ),
+        ]));
+        let batch = RecordBatch::new_empty(schema.clone());
+        let batches = RecordBatches::try_new(schema, vec![batch]).unwrap();
+        let mut labels = HashSet::new();
+
+        record_batches_to_labels_name(batches, &mut labels).unwrap();
+
+        assert!(labels.is_empty());
+    }
+
+    #[tokio::test]
+    async fn test_retrieve_metric_metadata_uses_semantic_options() {
+        let schema = Arc::new(Schema::new(vec![
+            ColumnSchema::new(
+                "greptime_timestamp",
+                ConcreteDataType::timestamp_millisecond_datatype(),
+                false,
+            )
+            .with_time_index(true),
+            ColumnSchema::new("host", ConcreteDataType::string_datatype(), false),
+            ColumnSchema::new("value", ConcreteDataType::float64_datatype(), true),
+        ]));
+        let mut options = TableOptions::default();
+        options.extra_options.insert(
+            LOGICAL_TABLE_METADATA_KEY.to_string(),
+            "greptime_physical_table".to_string(),
+        );
+        options
+            .extra_options
+            .insert(SEMANTIC_METRIC_TYPE.to_string(), "counter".to_string());
+        options
+            .extra_options
+            .insert(SEMANTIC_METRIC_UNIT.to_string(), "By".to_string());
+        let meta = TableMetaBuilder::empty()
+            .schema(schema)
+            .primary_key_indices(vec![1])
+            .engine("metric".to_string())
+            .next_column_id(3)
+            .options(options)
+            .build()
+            .unwrap();
+        let table_info = TableInfoBuilder::default()
+            .table_id(1025)
+            .table_version(0 as TableVersion)
+            .name("http_requests_total")
+            .catalog_name(DEFAULT_CATALOG_NAME)
+            .schema_name(DEFAULT_SCHEMA_NAME)
+            .table_type(TableType::Base)
+            .meta(meta)
+            .build()
+            .unwrap();
+        let manager: CatalogManagerRef =
+            MemoryCatalogManager::new_with_table(EmptyTable::from_table_info(&table_info));
+        let query_ctx = QueryContext::with(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME);
+
+        let metadata = retrieve_metric_metadata(&query_ctx, manager, &MetadataQuery::default())
+            .await
+            .unwrap();
+
+        assert_eq!(
+            metadata.get("http_requests_total"),
+            Some(&vec![PromMetadata {
+                metric_type: "counter".to_string(),
+                unit: "By".to_string(),
+                help: String::new(),
+            }])
+        );
+    }
 }