rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-11-13 22:36:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: synchronize raw frame stream stack base

Browse Source 
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-10-15 18:23:37 +08:00
parent 696259a1ce 489c9453cd
commit cfb66f5f2e
4 changed files with 156 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/query/src/native_histogram.rs
+95 -32
View File
@@ -36,6 +36,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::series::histogram::{BucketSideView, HistogramBatchBuilder, HistogramView};
use datatypes::series::samples::Samples;
use datatypes::types::{StructField, StructType};
@@ -144,8 +145,6 @@ pub const CUSTOM_BUCKETS_SCHEMA: i32 = -53;
const MIN_EXPONENTIAL_SCHEMA: i32 = -4;
const MAX_EXPONENTIAL_SCHEMA: i32 = 8;
pub use datatypes::series::histogram::CounterResetHint;
/// 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.
@@ -589,18 +588,12 @@ impl NativeHistogram {
}
fn all_buckets(&self) -> Option<Vec<Bucket>> {
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<Vec<Bucket>> {
@@ -1053,28 +1046,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<Bucket>,
zero_threshold: f64,
zero_count: f64,
positive: Vec<Bucket>,
) -> Vec<Bucket> {
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 +1974,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!(
@@ -2516,6 +2564,21 @@ mod tests {
assert!(histogram_samples_from_arrow(&malformed_spans).is_err());
}
#[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(
src/datatypes/src/series/samples.rs
+1 -36
View File
@@ -624,7 +624,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");
}
@@ -648,38 +648,3 @@ impl Series {
})
}
}
fn checked_time_len(time: &TimeAxis) -> Result<usize> {
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)
}
}
}
src/datatypes/src/series/samples/tests.rs
+15 -2
View File
@@ -1045,9 +1045,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);
}
src/datatypes/src/series/time.rs
+45 -25
View File
@@ -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),
@@ -178,37 +181,54 @@ pub enum TimeAxis {
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))
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<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");
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<usize> {
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)
}
}
let _ = total;
Ok(Self::ExplicitChunks(nonempty.into()))
}
/// Returns the timestamp count.