diff --git a/src/datatypes/src/error.rs b/src/datatypes/src/error.rs
index 745268f6e35..9c29853ddcd 100644
--- a/src/datatypes/src/error.rs
+++ b/src/datatypes/src/error.rs
@@ -335,6 +335,13 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Invalid series: {reason}"))]
+    InvalidSeries {
+        reason: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("unexpected: {reason}"))]
     Unexpected {
         reason: String,
@@ -368,7 +375,8 @@ impl ErrorExt for Error {
             | InvalidVector { .. }
             | InvalidFulltextOption { .. }
             | InvalidSkippingIndexOption { .. }
-            | InvalidCompressionOption { .. } => StatusCode::InvalidArguments,
+            | InvalidCompressionOption { .. }
+            | InvalidSeries { .. } => StatusCode::InvalidArguments,
 
             ValueExceedsPrecision { .. }
             | CastType { .. }
diff --git a/src/datatypes/src/lib.rs b/src/datatypes/src/lib.rs
index 0dcce168571..fee2ab7f819 100644
--- a/src/datatypes/src/lib.rs
+++ b/src/datatypes/src/lib.rs
@@ -24,6 +24,7 @@ pub mod prelude;
 pub mod scalars;
 pub mod schema;
 pub mod serialize;
+pub mod series;
 pub mod time;
 pub mod timestamp;
 pub mod type_id;
diff --git a/src/datatypes/src/series.rs b/src/datatypes/src/series.rs
new file mode 100644
index 00000000000..24152366d47
--- /dev/null
+++ b/src/datatypes/src/series.rs
@@ -0,0 +1,162 @@
+// 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.
+
+pub mod histogram;
+pub mod time;
+
+#[cfg(test)]
+#[path = "series/tests.rs"]
+mod tests;
+
+use std::sync::Arc;
+
+use crate::error::{InvalidSeriesSnafu, Result};
+
+/// A label with shared UTF-8 name and value strings.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Label {
+    /// Label name.
+    pub name: Arc<str>,
+    /// Label value.
+    pub value: Arc<str>,
+}
+
+/// A metric name and its name-sorted ordinary labels.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct MetricName {
+    name: Option<Arc<str>>,
+    tags: Arc<[Label]>,
+}
+
+impl MetricName {
+    /// Creates a metric name, sorting labels and rejecting duplicates or `__name__` tags.
+    pub fn try_new(name: Option<Arc<str>>, mut tags: Vec<Label>) -> Result<Self> {
+        tags.sort_unstable_by(|a, b| a.name.cmp(&b.name));
+        if tags.iter().any(|tag| tag.name.as_ref() == "__name__") {
+            return invalid_series("__name__ must be stored as the metric name");
+        }
+        if tags.windows(2).any(|pair| pair[0].name == pair[1].name) {
+            return invalid_series("duplicate label name");
+        }
+        Ok(Self {
+            name,
+            tags: tags.into(),
+        })
+    }
+
+    /// Returns the metric name, if present.
+    pub fn name(&self) -> Option<&str> {
+        self.name.as_deref()
+    }
+
+    /// Returns ordinary labels sorted by name.
+    pub fn tags(&self) -> &[Label] {
+        &self.tags
+    }
+
+    /// Returns a label value, distinguishing absence from an empty value.
+    pub fn get(&self, label: &str) -> Option<&str> {
+        if label == "__name__" {
+            self.name()
+        } else {
+            self.tags
+                .binary_search_by(|tag| tag.name.as_ref().cmp(label))
+                .ok()
+                .map(|index| self.tags[index].value.as_ref())
+        }
+    }
+
+    /// Sets the metric name or an ordinary label, preserving other shared labels.
+    pub fn set(&mut self, label: &str, value: Arc<str>) -> Result<()> {
+        if label == "__name__" {
+            self.name = Some(value);
+            return Ok(());
+        }
+        match self
+            .tags
+            .binary_search_by(|tag| tag.name.as_ref().cmp(label))
+        {
+            Ok(index) => {
+                let mut tags = self.tags.to_vec();
+                tags[index].value = value;
+                self.tags = tags.into();
+            }
+            Err(index) => {
+                let mut tags = self.tags.to_vec();
+                tags.insert(
+                    index,
+                    Label {
+                        name: Arc::from(label),
+                        value,
+                    },
+                );
+                self.tags = tags.into();
+            }
+        }
+        Ok(())
+    }
+
+    /// Removes a label or clears the metric name.
+    pub fn remove(&mut self, label: &str) {
+        if label == "__name__" {
+            self.name = None;
+        } else if let Ok(index) = self
+            .tags
+            .binary_search_by(|tag| tag.name.as_ref().cmp(label))
+        {
+            let mut tags = self.tags.to_vec();
+            tags.remove(index);
+            self.tags = tags.into();
+        }
+    }
+
+    /// Keeps only the named labels and metric name.
+    pub fn keep_only(&mut self, labels: &[&str]) {
+        if !labels.contains(&"__name__") {
+            self.name = None;
+        }
+        let kept: Vec<_> = self
+            .tags
+            .iter()
+            .filter(|tag| labels.contains(&tag.name.as_ref()))
+            .cloned()
+            .collect();
+        if kept.len() != self.tags.len() {
+            self.tags = kept.into();
+        }
+    }
+
+    /// Removes the selected labels and metric name.
+    pub fn remove_all(&mut self, labels: &[&str]) {
+        if labels.contains(&"__name__") {
+            self.name = None;
+        }
+        let kept: Vec<_> = self
+            .tags
+            .iter()
+            .filter(|tag| !labels.contains(&tag.name.as_ref()))
+            .cloned()
+            .collect();
+        if kept.len() != self.tags.len() {
+            self.tags = kept.into();
+        }
+    }
+}
+
+fn invalid_series<T>(reason: &str) -> Result<T> {
+    InvalidSeriesSnafu {
+        reason: reason.to_string(),
+    }
+    .fail()
+}
diff --git a/src/datatypes/src/series/histogram.rs b/src/datatypes/src/series/histogram.rs
new file mode 100644
index 00000000000..9b9322c2774
--- /dev/null
+++ b/src/datatypes/src/series/histogram.rs
@@ -0,0 +1,493 @@
+// 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::sync::Arc;
+
+use arrow::array::{
+    Array, ArrayBuilder, ArrayRef, Float64Array, Float64Builder, Int32Array, Int32Builder,
+    Int64Array, Int64Builder, ListArray, ListBuilder, PrimitiveArray, StructArray,
+};
+use arrow::buffer::OffsetBuffer;
+use arrow::datatypes::{ArrowPrimitiveType, DataType, Field, Fields, Float64Type, Int32Type};
+use arrow::error::ArrowError;
+
+use crate::Result;
+use crate::error::InvalidSeriesSnafu;
+
+/// Reset hint retained by native histograms, including values unknown to this version.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum CounterResetHint {
+    /// The payload must be inspected to determine whether a reset occurred.
+    Unknown,
+    /// This is the first histogram after a counter reset.
+    CounterReset,
+    /// No counter reset occurred since the previous histogram.
+    NotCounterReset,
+    /// This is a gauge histogram, so counter resets do not apply.
+    Gauge,
+    /// A value not understood by this version.
+    Unrecognized(i32),
+}
+impl From<i32> for CounterResetHint {
+    fn from(value: i32) -> Self {
+        match value {
+            0 => Self::Unknown,
+            1 => Self::CounterReset,
+            2 => Self::NotCounterReset,
+            3 => Self::Gauge,
+            v => Self::Unrecognized(v),
+        }
+    }
+}
+impl From<CounterResetHint> for i32 {
+    fn from(value: CounterResetHint) -> Self {
+        match value {
+            CounterResetHint::Unknown => 0,
+            CounterResetHint::CounterReset => 1,
+            CounterResetHint::NotCounterReset => 2,
+            CounterResetHint::Gauge => 3,
+            CounterResetHint::Unrecognized(v) => v,
+        }
+    }
+}
+pub(crate) fn fields() -> Fields {
+    let list = |t| DataType::List(Arc::new(Field::new("item", t, false)));
+    vec![
+        Field::new("schema", DataType::Int32, false),
+        Field::new("zero_threshold", DataType::Float64, false),
+        Field::new("sum", DataType::Float64, false),
+        Field::new("count", DataType::Float64, false),
+        Field::new("zero_count", DataType::Float64, false),
+        Field::new("reset_hint", DataType::Int32, false),
+        Field::new("start_timestamp", DataType::Int64, true),
+        Field::new("custom_values", list(DataType::Float64), false),
+        Field::new("positive_span_offsets", list(DataType::Int32), false),
+        Field::new("positive_span_lengths", list(DataType::Int32), false),
+        Field::new("positive_buckets", list(DataType::Float64), false),
+        Field::new("negative_span_offsets", list(DataType::Int32), false),
+        Field::new("negative_span_lengths", list(DataType::Int32), false),
+        Field::new("negative_buckets", list(DataType::Float64), false),
+    ]
+    .into()
+}
+fn invalid<T>(reason: impl Into<String>) -> Result<T> {
+    Err(InvalidSeriesSnafu {
+        reason: reason.into(),
+    }
+    .build())
+}
+fn arrow_result<T>(r: std::result::Result<T, ArrowError>) -> Result<T> {
+    r.map_err(|e| {
+        InvalidSeriesSnafu {
+            reason: format!("invalid histogram Arrow array: {e}"),
+        }
+        .build()
+    })
+}
+
+/// A borrowed set of spans and absolute bucket counts.
+#[derive(Clone, Copy, Debug)]
+pub struct BucketSideView<'a> {
+    /// Span offsets.
+    pub span_offsets: &'a [i32],
+    /// Span lengths.
+    pub span_lengths: &'a [i32],
+    /// Absolute bucket counts.
+    pub buckets: &'a [f64],
+}
+/// A borrowed histogram row.
+#[derive(Clone, Copy, Debug)]
+pub struct HistogramView<'a> {
+    /// Histogram schema.
+    pub schema: i32,
+    /// Zero bucket threshold.
+    pub zero_threshold: f64,
+    /// Sum of observations.
+    pub sum: f64,
+    /// Total count.
+    pub count: f64,
+    /// Count in the zero bucket.
+    pub zero_count: f64,
+    /// Counter reset hint.
+    pub reset_hint: CounterResetHint,
+    /// Optional start timestamp in milliseconds.
+    pub start_timestamp: Option<i64>,
+    /// Custom bucket bounds.
+    pub custom_values: &'a [f64],
+    /// Positive bucket side.
+    pub positive: BucketSideView<'a>,
+    /// Negative bucket side.
+    pub negative: BucketSideView<'a>,
+}
+
+#[derive(Debug)]
+struct HistogramList<T: ArrowPrimitiveType> {
+    offsets: OffsetBuffer<i32>,
+    values: PrimitiveArray<T>,
+}
+impl<T: ArrowPrimitiveType> HistogramList<T> {
+    fn try_new(a: &ListArray) -> Result<Self> {
+        let v = a
+            .values()
+            .as_any()
+            .downcast_ref::<PrimitiveArray<T>>()
+            .ok_or_else(|| {
+                InvalidSeriesSnafu {
+                    reason: "histogram list item type mismatch".to_string(),
+                }
+                .build()
+            })?;
+        if a.null_count() != 0 || v.null_count() != 0 {
+            return invalid("histogram list rows and items must not be null");
+        }
+        Ok(Self {
+            offsets: a.offsets().clone(),
+            values: v.clone(),
+        })
+    }
+    fn view(&self, row: usize) -> &[T::Native] {
+        let s = self.offsets[row] as usize;
+        let e = self.offsets[row + 1] as usize;
+        &self.values.values()[s..e]
+    }
+}
+/// Columnar owner of native histogram rows.
+#[derive(Debug)]
+pub struct HistogramBatch {
+    array: StructArray,
+    schema: Int32Array,
+    threshold: Float64Array,
+    sum: Float64Array,
+    count: Float64Array,
+    zero: Float64Array,
+    hint: Int32Array,
+    start: Int64Array,
+    custom: HistogramList<Float64Type>,
+    po: HistogramList<Int32Type>,
+    pl: HistogramList<Int32Type>,
+    pb: HistogramList<Float64Type>,
+    no: HistogramList<Int32Type>,
+    nl: HistogramList<Int32Type>,
+    nb: HistogramList<Float64Type>,
+}
+impl HistogramBatch {
+    /// Validates and wraps an Arrow struct with the required ordered schema.
+    pub fn try_from_arrow(array: StructArray) -> Result<Self> {
+        if array.fields() != &fields() {
+            return invalid("histogram fields, order, types, or nullability mismatch");
+        }
+        if array.null_count() != 0 {
+            return invalid("histogram struct rows must not be null");
+        }
+        let c = array.columns();
+        if c.iter()
+            .enumerate()
+            .any(|(i, a)| i != 6 && a.null_count() != 0)
+        {
+            return invalid("non-nullable histogram column contains null");
+        }
+        macro_rules! col {
+            ($i:expr,$t:ty) => {
+                c[$i].as_any().downcast_ref::<$t>().cloned().ok_or_else(|| {
+                    InvalidSeriesSnafu {
+                        reason: format!("histogram column {} type mismatch", $i),
+                    }
+                    .build()
+                })
+            };
+        }
+        macro_rules! list {
+            ($i:expr,$t:ty) => {{
+                let a = c[$i].as_any().downcast_ref::<ListArray>().ok_or_else(|| {
+                    InvalidSeriesSnafu {
+                        reason: format!("histogram list column {} type mismatch", $i),
+                    }
+                    .build()
+                })?;
+                HistogramList::<$t>::try_new(a)?
+            }};
+        }
+        let batch = Self {
+            schema: col!(0, Int32Array)?,
+            threshold: col!(1, Float64Array)?,
+            sum: col!(2, Float64Array)?,
+            count: col!(3, Float64Array)?,
+            zero: col!(4, Float64Array)?,
+            hint: col!(5, Int32Array)?,
+            start: col!(6, Int64Array)?,
+            custom: list!(7, Float64Type),
+            po: list!(8, Int32Type),
+            pl: list!(9, Int32Type),
+            pb: list!(10, Float64Type),
+            no: list!(11, Int32Type),
+            nl: list!(12, Int32Type),
+            nb: list!(13, Float64Type),
+            array,
+        };
+        for i in 0..batch.len() {
+            validate_spans(batch.po.view(i), batch.pl.view(i), batch.pb.view(i).len())?;
+            validate_spans(batch.no.view(i), batch.nl.view(i), batch.nb.view(i).len())?
+        }
+        Ok(batch)
+    }
+    /// Returns the owning Arrow struct array.
+    pub fn as_arrow(&self) -> &StructArray {
+        &self.array
+    }
+    /// Number of rows.
+    pub fn len(&self) -> usize {
+        self.array.len()
+    }
+    /// Whether there are no rows.
+    pub fn is_empty(&self) -> bool {
+        self.array.is_empty()
+    }
+    /// Returns a borrowed row, or `None` when out of bounds.
+    pub fn view(&self, i: usize) -> Option<HistogramView<'_>> {
+        if i >= self.len() {
+            return None;
+        }
+        Some(HistogramView {
+            schema: self.schema.value(i),
+            zero_threshold: self.threshold.value(i),
+            sum: self.sum.value(i),
+            count: self.count.value(i),
+            zero_count: self.zero.value(i),
+            reset_hint: self.hint.value(i).into(),
+            start_timestamp: self.start.is_valid(i).then(|| self.start.value(i)),
+            custom_values: self.custom.view(i),
+            positive: BucketSideView {
+                span_offsets: self.po.view(i),
+                span_lengths: self.pl.view(i),
+                buckets: self.pb.view(i),
+            },
+            negative: BucketSideView {
+                span_offsets: self.no.view(i),
+                span_lengths: self.nl.view(i),
+                buckets: self.nb.view(i),
+            },
+        })
+    }
+}
+fn validate_spans(offsets: &[i32], lengths: &[i32], buckets: usize) -> Result<()> {
+    if offsets.len() != lengths.len() {
+        return invalid("span offsets and lengths differ");
+    }
+    let mut cursor = 0i32;
+    let mut total = 0usize;
+    for (i, (&offset, &length)) in offsets.iter().zip(lengths).enumerate() {
+        if length < 0 || (i > 0 && offset < 0) {
+            return invalid("negative span length or noninitial offset");
+        }
+        if i == 0 {
+            cursor = offset
+        } else {
+            cursor = cursor.checked_add(offset).ok_or_else(|| {
+                InvalidSeriesSnafu {
+                    reason: "span index overflow".to_string(),
+                }
+                .build()
+            })?
+        }
+        total = total.checked_add(length as usize).ok_or_else(|| {
+            InvalidSeriesSnafu {
+                reason: "bucket count overflow".to_string(),
+            }
+            .build()
+        })?;
+        cursor = cursor.checked_add(length).ok_or_else(|| {
+            InvalidSeriesSnafu {
+                reason: "span cursor overflow".to_string(),
+            }
+            .build()
+        })?;
+    }
+    if total != buckets {
+        return invalid("span lengths do not match bucket count");
+    }
+    Ok(())
+}
+
+/// Builds a histogram batch directly in Arrow builders.
+pub struct HistogramBatchBuilder {
+    schema: Int32Builder,
+    threshold: Float64Builder,
+    sum: Float64Builder,
+    count: Float64Builder,
+    zero: Float64Builder,
+    hint: Int32Builder,
+    start: Int64Builder,
+    custom: ListBuilder<Float64Builder>,
+    po: ListBuilder<Int32Builder>,
+    pl: ListBuilder<Int32Builder>,
+    pb: ListBuilder<Float64Builder>,
+    no: ListBuilder<Int32Builder>,
+    nl: ListBuilder<Int32Builder>,
+    nb: ListBuilder<Float64Builder>,
+    rows: usize,
+}
+impl Default for HistogramBatchBuilder {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+impl HistogramBatchBuilder {
+    /// Creates an empty builder.
+    pub fn new() -> Self {
+        Self {
+            schema: Int32Builder::new(),
+            threshold: Float64Builder::new(),
+            sum: Float64Builder::new(),
+            count: Float64Builder::new(),
+            zero: Float64Builder::new(),
+            hint: Int32Builder::new(),
+            start: Int64Builder::new(),
+            custom: lb64(),
+            po: lb32(),
+            pl: lb32(),
+            pb: lb64(),
+            no: lb32(),
+            nl: lb32(),
+            nb: lb64(),
+            rows: 0,
+        }
+    }
+    /// Validates and appends a row, returning its index without staging bucket vectors.
+    pub fn push(&mut self, r: HistogramView<'_>) -> Result<u32> {
+        preflight(
+            r,
+            self.rows,
+            [
+                self.custom.values_ref().len(),
+                self.po.values_ref().len(),
+                self.pl.values_ref().len(),
+                self.pb.values_ref().len(),
+                self.no.values_ref().len(),
+                self.nl.values_ref().len(),
+                self.nb.values_ref().len(),
+            ],
+        )?;
+        let index = u32::try_from(self.rows).map_err(|_| {
+            InvalidSeriesSnafu {
+                reason: "histogram row index overflow".to_string(),
+            }
+            .build()
+        })?;
+        self.schema.append_value(r.schema);
+        self.threshold.append_value(r.zero_threshold);
+        self.sum.append_value(r.sum);
+        self.count.append_value(r.count);
+        self.zero.append_value(r.zero_count);
+        self.hint.append_value(r.reset_hint.into());
+        if let Some(v) = r.start_timestamp {
+            self.start.append_value(v)
+        } else {
+            self.start.append_null()
+        }
+        append64(&mut self.custom, r.custom_values);
+        append32(&mut self.po, r.positive.span_offsets);
+        append32(&mut self.pl, r.positive.span_lengths);
+        append64(&mut self.pb, r.positive.buckets);
+        append32(&mut self.no, r.negative.span_offsets);
+        append32(&mut self.nl, r.negative.span_lengths);
+        append64(&mut self.nb, r.negative.buckets);
+        self.rows += 1;
+        Ok(index)
+    }
+    /// Finishes by transferring builder buffers to the batch.
+    pub fn finish(mut self) -> Result<Arc<HistogramBatch>> {
+        let arrays: Vec<ArrayRef> = vec![
+            Arc::new(self.schema.finish()),
+            Arc::new(self.threshold.finish()),
+            Arc::new(self.sum.finish()),
+            Arc::new(self.count.finish()),
+            Arc::new(self.zero.finish()),
+            Arc::new(self.hint.finish()),
+            Arc::new(self.start.finish()),
+            Arc::new(self.custom.finish()),
+            Arc::new(self.po.finish()),
+            Arc::new(self.pl.finish()),
+            Arc::new(self.pb.finish()),
+            Arc::new(self.no.finish()),
+            Arc::new(self.nl.finish()),
+            Arc::new(self.nb.finish()),
+        ];
+        Ok(Arc::new(HistogramBatch::try_from_arrow(arrow_result(
+            StructArray::try_new(fields(), arrays, None),
+        )?)?))
+    }
+}
+fn preflight(r: HistogramView<'_>, rows: usize, existing: [usize; 7]) -> Result<()> {
+    if rows >= u32::MAX as usize {
+        return invalid("histogram row index overflow");
+    }
+    let added = [
+        r.custom_values.len(),
+        r.positive.span_offsets.len(),
+        r.positive.span_lengths.len(),
+        r.positive.buckets.len(),
+        r.negative.span_offsets.len(),
+        r.negative.span_lengths.len(),
+        r.negative.buckets.len(),
+    ];
+    for (old, new) in existing.into_iter().zip(added) {
+        if old
+            .checked_add(new)
+            .filter(|n| *n <= i32::MAX as usize)
+            .is_none()
+        {
+            return invalid("histogram list values exceed Arrow offset limit");
+        }
+    }
+    validate_spans(
+        r.positive.span_offsets,
+        r.positive.span_lengths,
+        r.positive.buckets.len(),
+    )?;
+    validate_spans(
+        r.negative.span_offsets,
+        r.negative.span_lengths,
+        r.negative.buckets.len(),
+    )
+}
+fn lb64() -> ListBuilder<Float64Builder> {
+    ListBuilder::new(Float64Builder::new()).with_field(Arc::new(Field::new(
+        "item",
+        DataType::Float64,
+        false,
+    )))
+}
+fn lb32() -> ListBuilder<Int32Builder> {
+    ListBuilder::new(Int32Builder::new()).with_field(Arc::new(Field::new(
+        "item",
+        DataType::Int32,
+        false,
+    )))
+}
+fn append64(b: &mut ListBuilder<Float64Builder>, v: &[f64]) {
+    for x in v {
+        b.values().append_value(*x)
+    }
+    b.append(true)
+}
+fn append32(b: &mut ListBuilder<Int32Builder>, v: &[i32]) {
+    for x in v {
+        b.values().append_value(*x)
+    }
+    b.append(true)
+}
+
+#[cfg(test)]
+#[path = "histogram/tests.rs"]
+mod tests;
diff --git a/src/datatypes/src/series/histogram/tests.rs b/src/datatypes/src/series/histogram/tests.rs
new file mode 100644
index 00000000000..58d761a5ac7
--- /dev/null
+++ b/src/datatypes/src/series/histogram/tests.rs
@@ -0,0 +1,270 @@
+// 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 super::*;
+
+fn row(start: Option<i64>, hint: CounterResetHint) -> HistogramView<'static> {
+    HistogramView {
+        schema: 1,
+        zero_threshold: 0.5,
+        sum: f64::from_bits(0x7ff8_0000_0000_0042),
+        count: 3.0,
+        zero_count: -0.0,
+        reset_hint: hint,
+        start_timestamp: start,
+        custom_values: &[1.25, 2.5],
+        positive: BucketSideView {
+            span_offsets: &[0, 1],
+            span_lengths: &[2, 1],
+            buckets: &[0.0, -0.0, 7.0],
+        },
+        negative: BucketSideView {
+            span_offsets: &[-3],
+            span_lengths: &[1],
+            buckets: &[4.0],
+        },
+    }
+}
+fn batch() -> Arc<HistogramBatch> {
+    let mut b = HistogramBatchBuilder::new();
+    b.push(row(None, CounterResetHint::Unrecognized(42)))
+        .unwrap();
+    b.push(row(Some(123), CounterResetHint::Gauge)).unwrap();
+    b.finish().unwrap()
+}
+
+#[test]
+fn builder_views_and_arrow_buffer_borrowing() {
+    let empty = HistogramBatchBuilder::new();
+    assert!(empty.finish().unwrap().is_empty());
+    let original = row(None, CounterResetHint::Unrecognized(42));
+    let mut b = HistogramBatchBuilder::new();
+    assert_eq!(b.push(original).unwrap(), 0);
+    assert_eq!(b.push(row(Some(123), CounterResetHint::Gauge)).unwrap(), 1);
+    let b = b.finish().unwrap();
+    let v = b.view(0).unwrap();
+    assert_eq!(
+        (v.schema, v.zero_threshold, v.sum.to_bits(), v.count),
+        (1, 0.5, original.sum.to_bits(), 3.0)
+    );
+    assert_eq!(v.reset_hint, CounterResetHint::Unrecognized(42));
+    assert_eq!(v.start_timestamp, None);
+    assert_eq!(v.zero_count.to_bits(), (-0.0f64).to_bits());
+    assert_eq!(v.custom_values, &[1.25, 2.5]);
+    assert_eq!(v.positive.span_offsets, &[0, 1]);
+    assert_eq!(v.positive.span_lengths, &[2, 1]);
+    assert_eq!(v.positive.buckets[1].to_bits(), (-0.0f64).to_bits());
+    assert_eq!(v.negative.span_offsets, &[-3]);
+    assert_eq!(v.negative.span_lengths, &[1]);
+    assert_eq!(v.negative.buckets, &[4.0]);
+    let second = b.view(1).unwrap();
+    assert_eq!(second.start_timestamp, Some(123));
+    assert_eq!(second.reset_hint, CounterResetHint::Gauge);
+    assert!(b.view(usize::MAX).is_none());
+    let field = b
+        .as_arrow()
+        .column(10)
+        .as_any()
+        .downcast_ref::<ListArray>()
+        .unwrap();
+    let values = field
+        .values()
+        .as_any()
+        .downcast_ref::<Float64Array>()
+        .unwrap();
+    assert_eq!(v.positive.buckets.as_ptr(), values.values().as_ptr());
+    let retained = b.as_arrow().clone();
+    drop(b);
+    let after = HistogramBatch::try_from_arrow(retained).unwrap();
+    assert_eq!(after.view(0).unwrap().positive.buckets[2], 7.0);
+}
+
+#[test]
+fn failed_push_does_not_advance_index() {
+    let bad = HistogramView {
+        positive: BucketSideView {
+            span_offsets: &[0],
+            span_lengths: &[-1],
+            buckets: &[],
+        },
+        ..row(None, CounterResetHint::Unknown)
+    };
+    let mut b = HistogramBatchBuilder::new();
+    assert!(b.push(bad).is_err());
+    assert_eq!(b.push(row(None, CounterResetHint::Unknown)).unwrap(), 0);
+    assert_eq!(
+        b.finish().unwrap().view(0).unwrap().positive.buckets.len(),
+        3
+    );
+}
+
+#[test]
+fn rejects_invalid_spans() {
+    let cases = [
+        (&[0][..], &[][..], &[][..]),
+        (&[0][..], &[-1][..], &[][..]),
+        (&[0, -1][..], &[1, 0][..], &[1.0][..]),
+        (&[i32::MAX - 1, 1][..], &[1, 1][..], &[1.0, 2.0][..]),
+        (&[0][..], &[2][..], &[1.0][..]),
+    ];
+    for (o, l, buckets) in cases {
+        let r = HistogramView {
+            positive: BucketSideView {
+                span_offsets: o,
+                span_lengths: l,
+                buckets,
+            },
+            ..row(None, CounterResetHint::Unknown)
+        };
+        assert!(HistogramBatchBuilder::new().push(r).is_err());
+    }
+}
+
+#[test]
+fn reset_hint_codes_round_trip() {
+    for (code, hint) in [
+        (0, CounterResetHint::Unknown),
+        (1, CounterResetHint::CounterReset),
+        (2, CounterResetHint::NotCounterReset),
+        (3, CounterResetHint::Gauge),
+        (i32::MIN, CounterResetHint::Unrecognized(i32::MIN)),
+        (99, CounterResetHint::Unrecognized(99)),
+    ] {
+        assert_eq!(i32::from(CounterResetHint::from(code)), code);
+        assert_eq!(CounterResetHint::from(code), hint);
+    }
+}
+
+#[test]
+fn sliced_struct_list_and_primitive_values() {
+    let b = batch();
+    let source = b.as_arrow();
+    let wrapped = HistogramBatch::try_from_arrow(source.slice(1, 1)).unwrap();
+    assert_eq!(wrapped.view(0).unwrap().start_timestamp, Some(123));
+    let mut columns = source.columns().to_vec();
+    let original = columns[10].as_any().downcast_ref::<ListArray>().unwrap();
+    let list = original.slice(1, 1);
+    let values = list
+        .values()
+        .as_any()
+        .downcast_ref::<Float64Array>()
+        .unwrap();
+    let primitive = values.slice(1, values.len() - 1);
+    let rebuilt = ListArray::try_new(
+        Arc::new(Field::new_list_field(DataType::Float64, false)),
+        OffsetBuffer::new(vec![2i32, 5].into()),
+        Arc::new(primitive),
+        None,
+    )
+    .unwrap();
+    columns[10] = Arc::new(rebuilt);
+    for (i, c) in columns.iter_mut().enumerate() {
+        if i != 10 {
+            *c = c.slice(1, 1)
+        }
+    }
+    let struct_array = StructArray::try_new(source.fields().clone(), columns, None).unwrap();
+    let got = HistogramBatch::try_from_arrow(struct_array).unwrap();
+    assert_eq!(got.view(0).unwrap().positive.buckets, &[0.0, -0.0, 7.0]);
+}
+
+#[test]
+fn rejects_wrong_schema_type_order_and_nullability() {
+    let b = batch();
+    let source = b.as_arrow();
+    let mut c = source.columns().to_vec();
+    let mut f = source.fields().iter().cloned().collect::<Vec<_>>();
+    f.swap(0, 1);
+    c.swap(0, 1);
+    assert!(
+        HistogramBatch::try_from_arrow(StructArray::try_new(f.into(), c, None).unwrap()).is_err()
+    );
+    let mut f = source.fields().iter().cloned().collect::<Vec<_>>();
+    f[0] = Arc::new(Field::new("schema", DataType::Int32, true));
+    assert!(
+        HistogramBatch::try_from_arrow(
+            StructArray::try_new(f.into(), source.columns().to_vec(), None).unwrap()
+        )
+        .is_err()
+    );
+    let mut f = source.fields().iter().cloned().collect::<Vec<_>>();
+    f[0] = Arc::new(Field::new("schema", DataType::Float64, false));
+    let mut c = source.columns().to_vec();
+    c[0] = Arc::new(Float64Array::from(vec![1.0, 1.0]));
+    assert!(
+        HistogramBatch::try_from_arrow(StructArray::try_new(f.into(), c, None).unwrap()).is_err()
+    );
+}
+
+#[test]
+fn rejects_parent_and_list_item_nulls() {
+    let b = batch();
+    let source = b.as_arrow();
+    let parent = StructArray::try_new(
+        source.fields().clone(),
+        source.columns().to_vec(),
+        Some(arrow::buffer::NullBuffer::new_null(2)),
+    )
+    .unwrap();
+    assert!(HistogramBatch::try_from_arrow(parent).is_err());
+    let mut fields = source.fields().iter().cloned().collect::<Vec<_>>();
+    fields[10] = Arc::new(Field::new(
+        "positive_buckets",
+        DataType::List(Arc::new(Field::new("item", DataType::Float64, true))),
+        false,
+    ));
+    let list = source
+        .column(10)
+        .as_any()
+        .downcast_ref::<ListArray>()
+        .unwrap();
+    let values = list
+        .values()
+        .as_any()
+        .downcast_ref::<Float64Array>()
+        .unwrap();
+    let mut null_bits = vec![true; values.len()];
+    null_bits[1] = false;
+    let nullable = Float64Array::new(
+        values.values().clone(),
+        Some(arrow::buffer::NullBuffer::from(null_bits)),
+    );
+    assert!(
+        ListArray::try_new(
+            Arc::new(Field::new("item", DataType::Float64, false)),
+            list.offsets().clone(),
+            Arc::new(nullable.clone()),
+            None,
+        )
+        .is_err()
+    );
+    let rebuilt = ListArray::try_new(
+        Arc::new(Field::new("item", DataType::Float64, true)),
+        list.offsets().clone(),
+        Arc::new(nullable),
+        None,
+    )
+    .unwrap();
+    let mut cols = source.columns().to_vec();
+    cols[10] = Arc::new(rebuilt);
+    let child = StructArray::try_new(fields.into(), cols, None).unwrap();
+    assert!(HistogramBatch::try_from_arrow(child).is_err());
+}
+
+#[test]
+fn preflight_checks_list_limits_without_large_allocations() {
+    let r = row(None, CounterResetHint::Unknown);
+    assert!(preflight(r, 0, [i32::MAX as usize, 0, 0, 0, 0, 0, 0]).is_err());
+    assert!(preflight(r, 0, [0; 7]).is_ok());
+}
diff --git a/src/datatypes/src/series/tests.rs b/src/datatypes/src/series/tests.rs
new file mode 100644
index 00000000000..bf767529a06
--- /dev/null
+++ b/src/datatypes/src/series/tests.rs
@@ -0,0 +1,111 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use super::{Label, MetricName};
+
+fn label(name: &str, value: &str) -> Label {
+    Label {
+        name: Arc::from(name),
+        value: Arc::from(value),
+    }
+}
+
+#[test]
+fn metric_name_sorts_labels_and_rejects_ambiguous_identity() {
+    let metric = MetricName::try_new(
+        Some(Arc::from("cpu")),
+        vec![label("zone", "west"), label("host", "node-1")],
+    )
+    .unwrap();
+    assert_eq!(metric.name(), Some("cpu"));
+    assert_eq!(metric.tags()[0].name.as_ref(), "host");
+    assert_eq!(metric.tags()[1].name.as_ref(), "zone");
+
+    assert!(MetricName::try_new(None, vec![label("x", "1"), label("x", "2")]).is_err());
+    assert!(MetricName::try_new(None, vec![label("__name__", "cpu")]).is_err());
+}
+
+#[test]
+fn metric_identity_uses_contents_and_distinguishes_absent_from_empty() {
+    let one = MetricName::try_new(
+        Some(Arc::from(String::from("cpu"))),
+        vec![label("empty", ""), label("host", "a")],
+    )
+    .unwrap();
+    let same = MetricName::try_new(
+        Some(Arc::from(String::from("cpu"))),
+        vec![label("host", "a"), label("empty", "")],
+    )
+    .unwrap();
+    assert_eq!(one, same);
+    assert_eq!(one.get("empty"), Some(""));
+    assert_eq!(one.get("missing"), None);
+    assert_eq!(one.get("__name__"), Some("cpu"));
+    assert_eq!(one.get("__name_missing"), None);
+}
+
+#[test]
+fn metric_name_edits_retain_shared_labels_and_labels_are_copy_on_write() {
+    let mut metric = MetricName::try_new(
+        Some(Arc::from("cpu")),
+        vec![label("host", "a"), label("zone", "west")],
+    )
+    .unwrap();
+    let sibling = metric.clone();
+    assert!(Arc::ptr_eq(&metric.tags, &sibling.tags));
+
+    metric.set("__name__", Arc::from("memory")).unwrap();
+    assert!(Arc::ptr_eq(&metric.tags, &sibling.tags));
+    assert_eq!(sibling.name(), Some("cpu"));
+
+    metric.set("host", Arc::from("b")).unwrap();
+    metric.set("rack", Arc::from("r1")).unwrap();
+    assert_eq!(sibling.get("host"), Some("a"));
+    assert_eq!(sibling.get("rack"), None);
+    assert_eq!(
+        metric
+            .tags()
+            .iter()
+            .map(|tag| tag.name.as_ref())
+            .collect::<Vec<_>>(),
+        ["host", "rack", "zone"]
+    );
+    metric.remove("zone");
+    assert_eq!(metric.get("zone"), None);
+    assert_eq!(sibling.get("zone"), Some("west"));
+}
+
+#[test]
+fn keep_only_and_remove_all_handle_metric_name_and_ordinary_labels() {
+    let mut metric = MetricName::try_new(
+        Some(Arc::from("cpu")),
+        vec![label("host", "a"), label("zone", "west")],
+    )
+    .unwrap();
+    metric.keep_only(&["host"]);
+    assert_eq!(metric.name(), None);
+    assert_eq!(metric.tags().len(), 1);
+    assert_eq!(metric.get("host"), Some("a"));
+
+    metric.set("__name__", Arc::from("cpu")).unwrap();
+    metric.set("zone", Arc::from("west")).unwrap();
+    metric.remove_all(&["host", "__name__"]);
+    assert_eq!(metric.name(), None);
+    assert_eq!(metric.get("host"), None);
+    assert_eq!(metric.get("zone"), Some("west"));
+    metric.remove_all(&["zone"]);
+    assert!(metric.tags().is_empty());
+}
diff --git a/src/datatypes/src/series/time.rs b/src/datatypes/src/series/time.rs
new file mode 100644
index 00000000000..8ba6f321d50
--- /dev/null
+++ b/src/datatypes/src/series/time.rs
@@ -0,0 +1,301 @@
+// 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::sync::Arc;
+
+use arrow::buffer::ScalarBuffer;
+use snafu::OptionExt;
+
+use crate::error::{InvalidSeriesSnafu, Result};
+
+#[cfg(test)]
+#[path = "time/tests.rs"]
+mod tests;
+
+/// Value description of a non-empty, fixed-step evaluation grid.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct GridSpec {
+    /// First evaluation timestamp in milliseconds.
+    pub start_ms: i64,
+    /// Positive time step in milliseconds.
+    pub step_ms: i64,
+    /// Number of evaluation points.
+    pub len: u64,
+}
+
+impl GridSpec {
+    /// Creates a validated grid specification.
+    pub fn try_new(start_ms: i64, step_ms: i64, len: u64) -> Result<Self> {
+        if step_ms <= 0 || len == 0 {
+            return invalid("grid step must be positive and grid length nonzero");
+        }
+        let Some(last) = ((len - 1) as i128)
+            .checked_mul(step_ms as i128)
+            .and_then(|offset| (start_ms as i128).checked_add(offset))
+        else {
+            return invalid("grid timestamp overflows i128");
+        };
+        let Ok(platform_len) = usize::try_from(len) else {
+            return invalid("grid timestamp or length overflows supported range");
+        };
+        if i64::try_from(last).is_err()
+            || platform_len
+                .checked_mul(std::mem::size_of::<i64>())
+                .is_none()
+        {
+            return invalid("grid timestamp or length overflows supported range");
+        }
+        Ok(Self {
+            start_ms,
+            step_ms,
+            len,
+        })
+    }
+
+    /// Creates a grid for an inclusive query range, retaining the last point not after end.
+    pub fn try_from_range(start_ms: i64, end_ms: i64, step_ms: i64) -> Result<Self> {
+        if step_ms <= 0 || end_ms < start_ms {
+            return invalid("invalid inclusive grid range or step");
+        }
+        let len = ((end_ms as i128 - start_ms as i128) / step_ms as i128) + 1;
+        Self::try_new(
+            start_ms,
+            step_ms,
+            u64::try_from(len).map_err(|_| invalid_series("grid length overflow"))?,
+        )
+    }
+}
+
+/// A materialized, shareable timestamp grid.
+#[derive(Debug)]
+pub struct EvalGrid {
+    spec: GridSpec,
+    timestamps: ScalarBuffer<i64>,
+}
+
+impl EvalGrid {
+    /// Allocates the grid timestamps once for sharing across result series.
+    pub fn try_new(spec: GridSpec) -> Result<Self> {
+        GridSpec::try_new(spec.start_ms, spec.step_ms, spec.len)?;
+        let len =
+            usize::try_from(spec.len).map_err(|_| invalid_series("grid length overflows usize"))?;
+        let mut values = Vec::new();
+        values
+            .try_reserve_exact(len)
+            .map_err(|_| invalid_series("grid timestamp allocation failed"))?;
+        for index in 0..len {
+            // The validated endpoint bounds every point in this positive-step grid.
+            let timestamp = (spec.start_ms as i128) + (index as i128) * (spec.step_ms as i128);
+            values.push(timestamp as i64);
+        }
+        Ok(Self {
+            spec,
+            timestamps: values.into(),
+        })
+    }
+
+    /// Returns the value specification.
+    pub fn spec(&self) -> GridSpec {
+        self.spec
+    }
+
+    /// Returns the shared timestamp buffer.
+    pub fn timestamps(&self) -> &ScalarBuffer<i64> {
+        &self.timestamps
+    }
+}
+
+/// A contiguous range referencing a shared evaluation grid.
+#[derive(Clone, Debug)]
+pub struct GridRange {
+    grid: Arc<EvalGrid>,
+    start: usize,
+    len: usize,
+}
+
+impl GridRange {
+    /// Creates a checked subrange; an empty range may begin at the grid end.
+    pub fn try_new(grid: Arc<EvalGrid>, start: usize, len: usize) -> Result<Self> {
+        let end = start.checked_add(len).context(InvalidSeriesSnafu {
+            reason: "grid range overflow".to_string(),
+        })?;
+        if end > grid.timestamps.len() {
+            return invalid("grid range exceeds its grid");
+        }
+        Ok(Self { grid, start, len })
+    }
+
+    /// Returns the shared grid.
+    pub fn grid(&self) -> &Arc<EvalGrid> {
+        &self.grid
+    }
+    /// Returns the start index in the grid.
+    pub fn start(&self) -> usize {
+        self.start
+    }
+    /// Returns the range length.
+    pub fn len(&self) -> usize {
+        self.len
+    }
+    /// Returns whether this range is empty.
+    pub fn is_empty(&self) -> bool {
+        self.len == 0
+    }
+    /// Returns this range as a zero-copy Arrow buffer slice.
+    pub fn timestamps(&self) -> ScalarBuffer<i64> {
+        self.grid.timestamps.slice(self.start, self.len)
+    }
+}
+
+impl PartialEq for GridRange {
+    fn eq(&self, other: &Self) -> bool {
+        self.grid.spec == other.grid.spec && self.start == other.start && self.len == other.len
+    }
+}
+impl Eq for GridRange {}
+
+/// A timestamp axis represented as a grid or explicit shared Arrow buffers.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum TimeAxis {
+    /// A slice of a shared regular evaluation grid.
+    Grid(GridRange),
+    /// An explicit timestamp sequence.
+    Explicit(ScalarBuffer<i64>),
+    /// Explicit timestamps held in ordered chunks without flattening.
+    ExplicitChunks(Arc<[ScalarBuffer<i64>]>),
+}
+
+impl TimeAxis {
+    /// Validates an explicit, strictly increasing timestamp sequence.
+    pub fn try_new(timestamps: ScalarBuffer<i64>) -> Result<Self> {
+        if timestamps.windows(2).any(|pair| pair[0] >= pair[1]) {
+            return invalid("timestamps must be strictly increasing");
+        }
+        Ok(Self::Explicit(timestamps))
+    }
+
+    /// Validates and stores timestamp chunks without concatenating their buffers.
+    pub fn try_from_chunks(chunks: Vec<ScalarBuffer<i64>>) -> Result<Self> {
+        let mut total = 0usize;
+        let mut previous = None;
+        let mut nonempty = Vec::new();
+        for chunk in chunks {
+            total = total.checked_add(chunk.len()).context(InvalidSeriesSnafu {
+                reason: "timestamp length overflow".to_string(),
+            })?;
+            if chunk.windows(2).any(|pair| pair[0] >= pair[1])
+                || chunk
+                    .first()
+                    .is_some_and(|first| previous.is_some_and(|last| last >= *first))
+            {
+                return invalid("timestamps must be strictly increasing across chunks");
+            }
+            if let Some(last) = chunk.last() {
+                previous = Some(*last);
+            }
+            if !chunk.is_empty() {
+                nonempty.push(chunk);
+            }
+        }
+        let _ = total;
+        Ok(Self::ExplicitChunks(nonempty.into()))
+    }
+
+    /// Returns the timestamp count.
+    pub fn len(&self) -> usize {
+        match self {
+            Self::Grid(range) => range.len,
+            Self::Explicit(values) => values.len(),
+            Self::ExplicitChunks(chunks) => chunks.iter().map(|chunk| chunk.len()).sum(),
+        }
+    }
+    /// Returns whether there are no timestamps.
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    /// Returns a timestamp by checked index.
+    pub fn timestamp(&self, index: usize) -> Option<i64> {
+        match self {
+            Self::Grid(range) if index < range.len => {
+                range.grid.timestamps.get(range.start + index).copied()
+            }
+            Self::Explicit(values) => values.get(index).copied(),
+            Self::ExplicitChunks(chunks) => {
+                let mut offset = index;
+                for chunk in chunks.iter() {
+                    if offset < chunk.len() {
+                        return chunk.get(offset).copied();
+                    }
+                    offset -= chunk.len();
+                }
+                None
+            }
+            _ => None,
+        }
+    }
+
+    /// Returns a zero-copy slice, retaining Arrow owners; chunked axes remain chunked.
+    pub fn slice(&self, offset: usize, len: usize) -> Result<Self> {
+        let end = offset.checked_add(len).context(InvalidSeriesSnafu {
+            reason: "timestamp slice overflow".to_string(),
+        })?;
+        if end > self.len() {
+            return invalid("timestamp slice exceeds axis");
+        }
+        match self {
+            Self::Grid(range) => Ok(Self::Grid(GridRange::try_new(
+                range.grid.clone(),
+                range.start + offset,
+                len,
+            )?)),
+            Self::Explicit(values) => Ok(Self::Explicit(values.slice(offset, len))),
+            Self::ExplicitChunks(chunks) => {
+                let mut remain = len;
+                let mut skip = offset;
+                let mut sliced = Vec::new();
+                for chunk in chunks.iter() {
+                    if skip >= chunk.len() {
+                        skip -= chunk.len();
+                        continue;
+                    }
+                    let take = remain.min(chunk.len() - skip);
+                    if take > 0 {
+                        sliced.push(chunk.slice(skip, take));
+                        remain -= take;
+                    }
+                    skip = 0;
+                    if remain == 0 {
+                        break;
+                    }
+                }
+                Ok(Self::ExplicitChunks(sliced.into()))
+            }
+        }
+    }
+}
+
+fn invalid_series(reason: &str) -> crate::error::Error {
+    InvalidSeriesSnafu {
+        reason: reason.to_string(),
+    }
+    .build()
+}
+
+fn invalid<T>(reason: &str) -> Result<T> {
+    InvalidSeriesSnafu {
+        reason: reason.to_string(),
+    }
+    .fail()
+}
diff --git a/src/datatypes/src/series/time/tests.rs b/src/datatypes/src/series/time/tests.rs
new file mode 100644
index 00000000000..f3ee539a624
--- /dev/null
+++ b/src/datatypes/src/series/time/tests.rs
@@ -0,0 +1,249 @@
+// 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::sync::{
+    Arc,
+    atomic::{AtomicUsize, Ordering},
+};
+
+use arrow::buffer::{Buffer, ScalarBuffer};
+
+use super::{EvalGrid, GridRange, GridSpec, TimeAxis};
+
+#[test]
+fn grid_specs_validate_and_ranges_are_inclusive_without_rounding_up() {
+    assert_eq!(GridSpec::try_from_range(-5, 6, 6).unwrap().len, 2);
+    assert_eq!(GridSpec::try_from_range(4, 4, 2).unwrap().len, 1);
+    assert_eq!(GridSpec::try_from_range(4, 5, 2).unwrap().len, 1);
+    assert!(GridSpec::try_from_range(5, 4, 1).is_err());
+    assert!(GridSpec::try_from_range(i64::MIN, i64::MAX, 1).is_err());
+    assert!(GridSpec::try_from_range(0, 1, 0).is_err());
+    assert!(GridSpec::try_new(0, -1, 1).is_err());
+    assert!(GridSpec::try_new(0, 1, 0).is_err());
+    assert!(GridSpec::try_new(i64::MAX, 1, 2).is_err());
+    assert_eq!(
+        &EvalGrid::try_new(GridSpec::try_new(i64::MIN, i64::MAX, 3).unwrap())
+            .unwrap()
+            .timestamps()[..],
+        &[i64::MIN, -1, i64::MAX - 1]
+    );
+    assert!(GridSpec::try_new(i64::MIN, i64::MAX, 4).is_err());
+    assert!(GridSpec::try_new(0, i64::MAX, u64::MAX).is_err());
+    assert!(GridSpec::try_new(i64::MIN, 1, 2).is_ok());
+    assert!(GridSpec::try_new(i64::MAX - 1, 1, 2).is_ok());
+}
+
+#[test]
+fn grid_materialization_revalidates_public_specs_and_shares_ranges_by_value() {
+    assert!(
+        EvalGrid::try_new(GridSpec {
+            start_ms: 0,
+            step_ms: 0,
+            len: 1
+        })
+        .is_err()
+    );
+    assert!(
+        EvalGrid::try_new(GridSpec {
+            start_ms: 0,
+            step_ms: 1,
+            len: 0
+        })
+        .is_err()
+    );
+    assert!(
+        EvalGrid::try_new(GridSpec {
+            start_ms: i64::MAX,
+            step_ms: 1,
+            len: 2
+        })
+        .is_err()
+    );
+
+    let grid = Arc::new(EvalGrid::try_new(GridSpec::try_new(-5, 3, 5).unwrap()).unwrap());
+    assert_eq!(&grid.timestamps()[..], &[-5, -2, 1, 4, 7]);
+    let a = GridRange::try_new(grid.clone(), 1, 3).unwrap();
+    let b = GridRange::try_new(Arc::new(EvalGrid::try_new(grid.spec()).unwrap()), 1, 3).unwrap();
+    assert_eq!(a, b);
+    let values = a.timestamps();
+    assert_eq!(values.as_ptr(), grid.timestamps().as_ptr().wrapping_add(1));
+    assert_eq!(&values[..], &[-2, 1, 4]);
+    assert!(GridRange::try_new(grid.clone(), 5, 0).unwrap().is_empty());
+    assert!(GridRange::try_new(grid.clone(), 4, 2).is_err());
+    assert!(GridRange::try_new(grid, usize::MAX, 1).is_err());
+}
+
+#[test]
+fn grid_axis_slices_keep_absolute_indices_and_shared_grid_alive() {
+    let grid = Arc::new(EvalGrid::try_new(GridSpec::try_new(10, 10, 5).unwrap()).unwrap());
+    let axis = TimeAxis::Grid(GridRange::try_new(grid.clone(), 1, 4).unwrap());
+    let nested = axis.slice(1, 2).unwrap();
+    let TimeAxis::Grid(range) = &nested else {
+        panic!("grid slice changed representation")
+    };
+    assert_eq!(range.start(), 2);
+    assert_eq!(range.len(), 2);
+    drop(axis);
+    drop(grid);
+    assert_eq!(nested.timestamp(0), Some(30));
+    assert_eq!(nested.timestamp(1), Some(40));
+    assert_eq!(nested.timestamp(2), None);
+    assert!(nested.slice(1, usize::MAX).is_err());
+    assert!(nested.slice(2, 1).is_err());
+}
+
+#[test]
+fn explicit_timestamps_require_strict_order_and_slice_without_copying() {
+    let source = ScalarBuffer::from(vec![1_i64, 3, 5, 7]);
+    let axis = TimeAxis::try_new(source.clone()).unwrap();
+    let clone = axis.clone();
+    let TimeAxis::Explicit(values) = &axis else {
+        panic!("expected explicit axis")
+    };
+    let TimeAxis::Explicit(cloned_values) = &clone else {
+        panic!("expected explicit axis")
+    };
+    assert!(values.ptr_eq(cloned_values));
+    let slice = axis.slice(1, 2).unwrap();
+    let TimeAxis::Explicit(sliced) = &slice else {
+        panic!("expected explicit slice")
+    };
+    assert_eq!(sliced.as_ptr(), source.as_ptr().wrapping_add(1));
+    assert_eq!(&sliced[..], &[3, 5]);
+    let nested = slice.slice(1, 1).unwrap();
+    let TimeAxis::Explicit(nested_values) = &nested else {
+        panic!("expected nested explicit slice")
+    };
+    assert_eq!(nested_values.as_ptr(), source.as_ptr().wrapping_add(2));
+    assert!(slice.slice(1, usize::MAX).is_err());
+    assert!(slice.slice(2, 1).is_err());
+    drop(source);
+    drop(axis);
+    drop(clone);
+    drop(slice);
+    assert_eq!(nested.timestamp(0), Some(5));
+    assert_eq!(nested.timestamp(1), None);
+    assert!(TimeAxis::try_new(ScalarBuffer::from(vec![1_i64, 1])).is_err());
+    assert!(TimeAxis::try_new(ScalarBuffer::from(vec![2_i64, 1])).is_err());
+}
+
+#[test]
+fn chunked_timestamps_validate_boundaries_and_keep_chunked_slices() {
+    let axis = TimeAxis::try_from_chunks(vec![
+        ScalarBuffer::from(Vec::<i64>::new()),
+        ScalarBuffer::from(vec![1_i64, 3, 5]),
+        ScalarBuffer::from(Vec::<i64>::new()),
+        ScalarBuffer::from(vec![7_i64, 9]),
+    ])
+    .unwrap();
+    let TimeAxis::ExplicitChunks(chunks) = &axis else {
+        panic!("expected chunked axis")
+    };
+    assert_eq!(chunks.len(), 2);
+    let first_ptr = chunks[0].as_ptr();
+    let second_ptr = chunks[1].as_ptr();
+    let slice = axis.slice(1, 3).unwrap();
+    let TimeAxis::ExplicitChunks(sliced) = &slice else {
+        panic!("slice flattened chunks")
+    };
+    assert_eq!(sliced.len(), 2);
+    assert_eq!(sliced[0].as_ptr(), first_ptr.wrapping_add(1));
+    assert_eq!(sliced[1].as_ptr(), second_ptr);
+    assert_eq!(slice.timestamp(0), Some(3));
+    assert_eq!(slice.timestamp(1), Some(5));
+    assert_eq!(slice.timestamp(2), Some(7));
+    assert_eq!(slice.timestamp(3), None);
+    let tail = axis.slice(5, 0).unwrap();
+    assert!(tail.is_empty());
+    let nested = slice.slice(1, 2).unwrap();
+    let TimeAxis::ExplicitChunks(nested_chunks) = &nested else {
+        panic!("nested chunk slice flattened representation")
+    };
+    assert_eq!(nested_chunks.len(), 2);
+    assert_eq!(nested_chunks[0].as_ptr(), first_ptr.wrapping_add(2));
+    assert_eq!(nested_chunks[1].as_ptr(), second_ptr);
+    assert_eq!(&nested_chunks[0][..], &[5]);
+    assert_eq!(&nested_chunks[1][..], &[7]);
+    assert!(axis.slice(usize::MAX, 1).is_err());
+    drop(slice);
+    drop(axis);
+    assert_eq!(nested.timestamp(0), Some(5));
+    assert_eq!(nested.timestamp(1), Some(7));
+
+    assert!(TimeAxis::try_from_chunks(vec![ScalarBuffer::from(vec![1_i64, 1])]).is_err());
+    assert!(
+        TimeAxis::try_from_chunks(vec![
+            ScalarBuffer::from(vec![1_i64, 3]),
+            ScalarBuffer::from(vec![3_i64, 5]),
+        ])
+        .is_err()
+    );
+    assert!(
+        TimeAxis::try_from_chunks(vec![
+            ScalarBuffer::from(vec![1_i64, 3]),
+            ScalarBuffer::from(vec![2_i64, 5]),
+        ])
+        .is_err()
+    );
+}
+
+struct BufferOwner {
+    values: Vec<i64>,
+    drops: Arc<AtomicUsize>,
+}
+
+impl Drop for BufferOwner {
+    fn drop(&mut self) {
+        self.drops.fetch_add(1, Ordering::SeqCst);
+    }
+}
+
+fn owned_timestamps(values: Vec<i64>, drops: Arc<AtomicUsize>) -> ScalarBuffer<i64> {
+    let mut owner = BufferOwner { values, drops };
+    let pointer = std::ptr::NonNull::new(owner.values.as_mut_ptr().cast()).unwrap();
+    let len = owner.values.len() * std::mem::size_of::<i64>();
+    // SAFETY: the vector allocation remains stable in the Arc owner for the buffer lifetime.
+    let element_count = owner.values.len();
+    let buffer = unsafe { Buffer::from_custom_allocation(pointer, len, Arc::new(owner)) };
+    ScalarBuffer::new(buffer, 0, element_count)
+}
+
+#[test]
+fn sliced_explicit_and_chunk_buffers_retain_and_release_real_allocations() {
+    let explicit_drops = Arc::new(AtomicUsize::new(0));
+    let axis = TimeAxis::try_new(owned_timestamps(vec![2, 4, 6], explicit_drops.clone())).unwrap();
+    let slice = axis.slice(1, 2).unwrap();
+    drop(axis);
+    assert_eq!(explicit_drops.load(Ordering::SeqCst), 0);
+    assert_eq!(slice.timestamp(0), Some(4));
+    drop(slice);
+    assert_eq!(explicit_drops.load(Ordering::SeqCst), 1);
+
+    let first_drops = Arc::new(AtomicUsize::new(0));
+    let second_drops = Arc::new(AtomicUsize::new(0));
+    let axis = TimeAxis::try_from_chunks(vec![
+        owned_timestamps(vec![1, 3], first_drops.clone()),
+        owned_timestamps(vec![5, 7], second_drops.clone()),
+    ])
+    .unwrap();
+    let slice = axis.slice(1, 2).unwrap();
+    drop(axis);
+    assert_eq!(first_drops.load(Ordering::SeqCst), 0);
+    assert_eq!(second_drops.load(Ordering::SeqCst), 0);
+    assert_eq!(slice.timestamp(0), Some(3));
+    assert_eq!(slice.timestamp(1), Some(5));
+    drop(slice);
+    assert_eq!(first_drops.load(Ordering::SeqCst), 1);
+    assert_eq!(second_drops.load(Ordering::SeqCst), 1);
+}