From bdc12034e559cbabc227ce3e53405928f568a22a Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Thu, 15 Oct 2026 17:56:57 +0800 Subject: [PATCH 1/3] fix(datatypes): address series foundation review feedback Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/common/query/src/native_histogram.rs | 58 +++++++----------------- src/datatypes/src/series/time.rs | 10 ++-- 2 files changed, 20 insertions(+), 48 deletions(-) diff --git a/src/common/query/src/native_histogram.rs b/src/common/query/src/native_histogram.rs index d8a559a2dc5..05e391419ae 100644 --- a/src/common/query/src/native_histogram.rs +++ b/src/common/query/src/native_histogram.rs @@ -35,6 +35,7 @@ use datafusion::arrow::datatypes::{ }; use datafusion_common::{DataFusionError, Result as DfResult}; use datatypes::data_type::{ConcreteDataType, DataType}; +pub use datatypes::series::histogram::CounterResetHint; use datatypes::types::{StructField, StructType}; pub use encoding::{NativeHistogramError, encode_native_histogram, native_histogram_column_schema}; use once_cell::sync::Lazy; @@ -141,48 +142,6 @@ pub const CUSTOM_BUCKETS_SCHEMA: i32 = -53; const MIN_EXPONENTIAL_SCHEMA: i32 = -4; const MAX_EXPONENTIAL_SCHEMA: i32 = 8; -/// Information carried by a native histogram about a possible counter reset. -/// -/// The persisted and protobuf representation remains an `i32`. Unrecognized -/// values are retained so data written by a newer producer can still round-trip. -#[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 for CounterResetHint { - fn from(value: i32) -> Self { - match value { - 0 => Self::Unknown, - 1 => Self::CounterReset, - 2 => Self::NotCounterReset, - 3 => Self::Gauge, - value => Self::Unrecognized(value), - } - } -} - -impl From for i32 { - fn from(value: CounterResetHint) -> Self { - match value { - CounterResetHint::Unknown => 0, - CounterResetHint::CounterReset => 1, - CounterResetHint::NotCounterReset => 2, - CounterResetHint::Gauge => 3, - CounterResetHint::Unrecognized(value) => value, - } - } -} - /// Reset hint indicating that the payload must be inspected. pub const UNKNOWN_COUNTER_RESET_HINT: CounterResetHint = CounterResetHint::Unknown; /// Reset hint indicating the first histogram after a counter reset. @@ -1753,6 +1712,21 @@ mod tests { ); } + #[test] + fn datatypes_reset_hints_round_trip_through_native_histogram() { + use datatypes::series::histogram::CounterResetHint as DatatypesCounterResetHint; + + let values = [-1, 0, 1, 2, 3, 77, i32::MIN, i32::MAX]; + for value in values { + let datatypes_hint = DatatypesCounterResetHint::from(value); + let query_hint: CounterResetHint = datatypes_hint; + let mut histogram = histogram(Vec::new(), Vec::new()); + histogram.reset_hint = query_hint; + let datatypes_hint: DatatypesCounterResetHint = histogram.reset_hint; + assert_eq!(i32::from(datatypes_hint), value); + } + } + #[test] fn arrow_round_trip_preserves_unrecognized_reset_hint() { let mut expected = histogram( diff --git a/src/datatypes/src/series/time.rs b/src/datatypes/src/series/time.rs index 66da94684d7..49b222010da 100644 --- a/src/datatypes/src/series/time.rs +++ b/src/datatypes/src/series/time.rs @@ -165,7 +165,10 @@ impl PartialEq for GridRange { impl Eq for GridRange {} /// A timestamp axis represented as a grid or explicit shared Arrow buffers. -#[derive(Clone, Debug, PartialEq, Eq)] +/// +/// Checked constructors validate timestamp ordering. Public payloads are validated again by the +/// downstream series boundary. +#[derive(Clone, Debug)] pub enum TimeAxis { /// A slice of a shared regular evaluation grid. Grid(GridRange), @@ -186,13 +189,9 @@ impl TimeAxis { /// Validates and stores timestamp chunks without concatenating their buffers. pub fn try_from_chunks(chunks: Vec>) -> Result { - 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() @@ -207,7 +206,6 @@ impl TimeAxis { nonempty.push(chunk); } } - let _ = total; Ok(Self::ExplicitChunks(nonempty.into())) } From 70a6c4551fc021c49f85d5e3b03fa008d017df4d Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:14:47 +0800 Subject: [PATCH 2/3] refactor(query): share native histogram bucket assembly Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/common/query/src/native_histogram.rs | 109 ++++++++++++++++------- 1 file changed, 79 insertions(+), 30 deletions(-) diff --git a/src/common/query/src/native_histogram.rs b/src/common/query/src/native_histogram.rs index 781ead65a1d..b70661c5dd6 100644 --- a/src/common/query/src/native_histogram.rs +++ b/src/common/query/src/native_histogram.rs @@ -589,18 +589,12 @@ impl NativeHistogram { } fn all_buckets(&self) -> Option> { - let mut buckets = self.side_buckets(false)?; - buckets.reverse(); - if self.zero_count != 0.0 { - buckets.push(Bucket { - lower: -self.zero_threshold, - upper: self.zero_threshold, - count: self.zero_count, - boundary_rule: BoundaryRule::ClosedBoth, - }); - } - buckets.extend(self.side_buckets(true)?); - Some(buckets) + Some(assemble_bucket_sides( + self.side_buckets(false)?, + self.zero_threshold, + self.zero_count, + self.side_buckets(true)?, + )) } fn side_buckets(&self, positive: bool) -> Option> { @@ -1053,28 +1047,35 @@ pub fn quantile_histogram_view( view.negative.buckets.is_empty(), view.positive.buckets.is_empty(), || { - let mut buckets = - build_view_side(view.negative, view.schema, view.custom_values, false)?; - buckets.reverse(); - if view.zero_count != 0.0 { - buckets.push(Bucket { - lower: -view.zero_threshold, - upper: view.zero_threshold, - count: view.zero_count, - boundary_rule: BoundaryRule::ClosedBoth, - }); - } - buckets.extend(build_view_side( - view.positive, - view.schema, - view.custom_values, - true, - )?); - Some(buckets) + Some(assemble_bucket_sides( + build_view_side(view.negative, view.schema, view.custom_values, false)?, + view.zero_threshold, + view.zero_count, + build_view_side(view.positive, view.schema, view.custom_values, true)?, + )) }, ) } +fn assemble_bucket_sides( + mut negative: Vec, + zero_threshold: f64, + zero_count: f64, + positive: Vec, +) -> Vec { + negative.reverse(); + if zero_count != 0.0 { + negative.push(Bucket { + lower: -zero_threshold, + upper: zero_threshold, + count: zero_count, + boundary_rule: BoundaryRule::ClosedBoth, + }); + } + negative.extend(positive); + negative +} + fn build_view_side( side: BucketSideView<'_>, schema: i32, @@ -1974,6 +1975,54 @@ mod tests { use super::*; use datatypes::series::histogram::HistogramBatch; + #[test] + fn assembles_bucket_sides_in_order_and_preserves_zero_count_semantics() { + let bucket = |lower, upper, count, boundary_rule| Bucket { + lower, + upper, + count, + boundary_rule, + }; + let negative = vec![ + bucket(-3.0, -2.0, 1.0, BoundaryRule::OpenRight), + bucket(-4.0, -3.0, 2.0, BoundaryRule::OpenRight), + ]; + let positive = vec![bucket(2.0, 3.0, 4.0, BoundaryRule::OpenLeft)]; + + for (zero_count, expected) in [ + ( + 5.0, + vec![ + bucket(-4.0, -3.0, 2.0, BoundaryRule::OpenRight), + bucket(-3.0, -2.0, 1.0, BoundaryRule::OpenRight), + bucket(-0.5, 0.5, 5.0, BoundaryRule::ClosedBoth), + bucket(2.0, 3.0, 4.0, BoundaryRule::OpenLeft), + ], + ), + ( + 0.0, + vec![ + bucket(-4.0, -3.0, 2.0, BoundaryRule::OpenRight), + bucket(-3.0, -2.0, 1.0, BoundaryRule::OpenRight), + bucket(2.0, 3.0, 4.0, BoundaryRule::OpenLeft), + ], + ), + ( + -0.0, + vec![ + bucket(-4.0, -3.0, 2.0, BoundaryRule::OpenRight), + bucket(-3.0, -2.0, 1.0, BoundaryRule::OpenRight), + bucket(2.0, 3.0, 4.0, BoundaryRule::OpenLeft), + ], + ), + ] { + assert_eq!( + assemble_bucket_sides(negative.clone(), 0.5, zero_count, positive.clone()), + expected + ); + } + } + fn assert_histogram_view_eq(view: &HistogramView<'_>, expected: &NativeHistogram) { assert_eq!(view.schema, expected.schema); assert_eq!( From ad1d1ee0fb748dca664726a01508d3fdf499ffe8 Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:15:01 +0800 Subject: [PATCH 3/3] refactor(datatypes): share series time validation Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- src/datatypes/src/series/samples.rs | 37 +------------- src/datatypes/src/series/samples/tests.rs | 17 ++++++- src/datatypes/src/series/time.rs | 60 ++++++++++++++++------- 3 files changed, 57 insertions(+), 57 deletions(-) diff --git a/src/datatypes/src/series/samples.rs b/src/datatypes/src/series/samples.rs index aeccfdf47e3..dead85c6d68 100644 --- a/src/datatypes/src/series/samples.rs +++ b/src/datatypes/src/series/samples.rs @@ -579,7 +579,7 @@ impl Series { if fields.len() != schema.fields.len() { return invalid("series field count does not match schema"); } - let time_len = checked_time_len(&time)?; + let time_len = time.try_len()?; if fields.iter().any(|f| f.len() != time_len) { return invalid("series field length does not match time axis"); } @@ -603,38 +603,3 @@ impl Series { }) } } - -fn checked_time_len(time: &TimeAxis) -> Result { - match time { - TimeAxis::Grid(range) => Ok(range.len()), - TimeAxis::Explicit(timestamps) => { - if timestamps.windows(2).any(|pair| pair[0] >= pair[1]) { - return invalid("timestamps must be strictly increasing"); - } - Ok(timestamps.len()) - } - TimeAxis::ExplicitChunks(chunks) => { - let mut total = 0usize; - let mut previous = None; - for chunk in chunks.iter() { - total = total.checked_add(chunk.len()).ok_or_else(|| { - InvalidSeriesSnafu { - reason: "timestamp length overflow".to_string(), - } - .build() - })?; - 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); - } - } - Ok(total) - } - } -} diff --git a/src/datatypes/src/series/samples/tests.rs b/src/datatypes/src/series/samples/tests.rs index 3f672ed418a..7cfb56bcd03 100644 --- a/src/datatypes/src/series/samples/tests.rs +++ b/src/datatypes/src/series/samples/tests.rs @@ -743,9 +743,22 @@ fn float_fields_clones_and_series_keep_real_arrow_buffer_owner_alive() { fields, ) .unwrap(); - drop(cloned); - assert_eq!(drops.load(Ordering::SeqCst), 0); + let DenseSamplesRef::Float(cloned_values) = cloned.get(0).unwrap().dense().unwrap() else { + panic!() + }; + let DenseSamplesRef::Float(series_values) = series.fields.get(0).unwrap().dense().unwrap() + else { + panic!() + }; + assert_eq!( + cloned_values.values().as_ptr(), + series_values.values().as_ptr() + ); + assert_eq!(cloned_values.values(), &[7.0, 8.0]); + drop(series); + assert_eq!(drops.load(Ordering::SeqCst), 0); + drop(cloned); assert_eq!(drops.load(Ordering::SeqCst), 1); } diff --git a/src/datatypes/src/series/time.rs b/src/datatypes/src/series/time.rs index 49b222010da..8003b461393 100644 --- a/src/datatypes/src/series/time.rs +++ b/src/datatypes/src/series/time.rs @@ -181,32 +181,54 @@ pub enum TimeAxis { impl TimeAxis { /// Validates an explicit, strictly increasing timestamp sequence. pub fn try_new(timestamps: ScalarBuffer) -> Result { - if timestamps.windows(2).any(|pair| pair[0] >= pair[1]) { - return invalid("timestamps must be strictly increasing"); - } - Ok(Self::Explicit(timestamps)) + let axis = Self::Explicit(timestamps); + axis.try_len()?; + Ok(axis) } /// Validates and stores timestamp chunks without concatenating their buffers. pub fn try_from_chunks(chunks: Vec>) -> Result { - let mut previous = None; - let mut nonempty = Vec::new(); - for chunk in chunks { - 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"); + let axis = Self::ExplicitChunks( + chunks + .into_iter() + .filter(|chunk| !chunk.is_empty()) + .collect(), + ); + axis.try_len()?; + Ok(axis) + } + + /// Returns the validated timestamp count. + pub(crate) fn try_len(&self) -> Result { + match self { + Self::Grid(range) => Ok(range.len()), + Self::Explicit(values) => { + if values.windows(2).any(|pair| pair[0] >= pair[1]) { + return invalid("timestamps must be strictly increasing"); + } + Ok(values.len()) } - if let Some(last) = chunk.last() { - previous = Some(*last); - } - if !chunk.is_empty() { - nonempty.push(chunk); + Self::ExplicitChunks(chunks) => { + let mut total = 0usize; + let mut previous = None; + for chunk in chunks.iter() { + total = total + .checked_add(chunk.len()) + .ok_or_else(|| invalid_series("timestamp length overflow"))?; + 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); + } + } + Ok(total) } } - Ok(Self::ExplicitChunks(nonempty.into())) } /// Returns the timestamp count.