mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-11-14 23:06:59 +00:00
feat(datatypes): add Narthex shared series identity and time foundations
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
@@ -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 { .. }
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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;
|
||||
@@ -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());
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
Reference in New Issue
Block a user