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

fix(datatypes): retain histogram table views across frame densification

Browse Source 
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-10-15 16:31:56 +08:00
parent cdca81ed9e
commit 3717cc2e2a
3 changed files with 439 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/datatypes/src/series/frame/tests.rs
+125 -78
View File
@@ -247,6 +247,98 @@ fn assert_float_bits(actual: &Samples, expected: &[Option<f64>]) {
}
}
#[test]
fn sliced_histogram_frames_densify_without_duplicating_shared_table() {
let schema = test_schema(&["histogram"]);
let source_table = histogram_batch();
let source_samples =
Samples::try_from_histogram(UInt32Array::from(vec![Some(0); 64]), source_table).unwrap();
let samples = Samples::try_from_chunks(
(0..64)
.map(|position| source_samples.slice(position, 1).unwrap())
.collect(),
)
.unwrap();
let chunk = RawSeriesChunk::try_new(
&schema,
Buffer::from(vec![0x42_u8]),
None,
RawTimeAxis::try_new((0_i64..64).collect::<Vec<_>>().into()).unwrap(),
SeriesFields::one(samples),
true,
)
.unwrap();
let frames = pack_raw_series_chunk(&schema, RawLabelMode::Key, &chunk).unwrap();
assert_eq!(frames.len(), 64);
let mut decoded_chunks = Vec::with_capacity(frames.len());
for frame in &frames {
let decoded = unpack_raw_series_frame(&schema, RawLabelMode::Key, frame).unwrap();
assert_eq!(decoded.len(), 1);
assert_eq!(decoded[0].time.len(), 1);
decoded_chunks.push(decoded.into_iter().next().unwrap());
}
let first_table = match decoded_chunks[0].fields.get(0).unwrap().dense().unwrap() {
DenseSamplesRef::Histogram(histograms) => histograms.batch().clone(),
_ => panic!("expected histogram samples"),
};
let second_table = match decoded_chunks[1].fields.get(0).unwrap().dense().unwrap() {
DenseSamplesRef::Histogram(histograms) => histograms.batch(),
_ => panic!("expected histogram samples"),
};
assert!(!Arc::ptr_eq(&first_table, second_table));
let first_values = first_table
.as_arrow()
.column(2)
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
let second_values = second_table
.as_arrow()
.column(2)
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
assert_eq!(
first_values.values().as_ptr(),
second_values.values().as_ptr()
);
let samples = Samples::try_from_chunks(
decoded_chunks
.iter()
.map(|decoded| decoded.fields.get(0).unwrap().clone())
.collect(),
)
.unwrap();
drop(chunk);
drop(frames);
drop(decoded_chunks);
drop(source_samples);
let dense = samples.into_dense().unwrap();
let DenseSamplesRef::Histogram(histograms) = dense.dense().unwrap() else {
panic!("expected dense histogram samples");
};
assert_eq!(histograms.len(), 64);
assert_eq!(histograms.batch().len(), first_table.len());
for position in 0..64 {
assert_histogram(histograms.view_at(position).unwrap(), -9.5);
}
let dense_values = histograms
.batch()
.as_arrow()
.column(2)
.as_any()
.downcast_ref::<Float64Array>()
.unwrap();
assert_eq!(
dense_values.values().as_ptr(),
first_values.values().as_ptr()
);
}
#[test]
fn schema_and_key_explicit_round_trips_preserve_identity_and_bits() {
let schema = test_schema(&["value"]);
@@ -315,26 +407,6 @@ fn schema_and_key_explicit_round_trips_preserve_identity_and_bits() {
expected_key_metadata,
);
assert_eq!(key_schema.as_ref(), &expected_key);
assert_eq!(key_schema.metadata().len(), 4);
assert_eq!(
key_schema.metadata().get("greptime.series.layout").unwrap(),
"v1"
);
assert_eq!(
key_schema
.metadata()
.get("greptime.series.content")
.unwrap(),
"raw"
);
assert_eq!(
key_schema.metadata().get("greptime.series.labels").unwrap(),
"key"
);
assert_eq!(
key_schema.metadata().get("greptime.series.time").unwrap(),
"timestamps"
);
let mut explicit_fields = expected_key.fields().iter().cloned().collect::<Vec<_>>();
explicit_fields.push(Field::new("name", DataType::Utf8, true).into());
explicit_fields.push(
@@ -360,18 +432,6 @@ fn schema_and_key_explicit_round_trips_preserve_identity_and_bits() {
let expected_explicit = Schema::new_with_metadata(explicit_fields, explicit_metadata);
assert_eq!(explicit_schema.as_ref(), &expected_explicit);
assert_eq!(
expected_key
.metadata()
.get("greptime.series.labels")
.unwrap(),
"key"
);
assert_eq!(
expected_key.metadata().get("greptime.series.time").unwrap(),
"timestamps"
);
let nan = f64::from_bits(0x7ff8_0000_0000_0042);
let stale_nan = f64::from_bits(0x7ff0_0000_0000_0002);
let values = [Some(-0.0), None, Some(nan), Some(stale_nan)];
@@ -628,26 +688,6 @@ fn segmented_mixed_fields_emit_every_aligned_position_and_marker() {
);
drop(aligned_chunk);
drop(aligned_frames);
assert_eq!(
aligned_a,
[
Some(1.0_f64.to_bits()),
Some(2.0_f64.to_bits()),
Some(3.0_f64.to_bits()),
None,
Some(5.0_f64.to_bits())
]
);
assert_eq!(
aligned_b,
[
Some(10.0_f64.to_bits()),
None,
Some(30.0_f64.to_bits()),
Some(40.0_f64.to_bits()),
Some(50.0_f64.to_bits())
]
);
let mut distinct_rows = frames.clone();
let alternate_schema = test_schema(&["first", "second"]);
@@ -817,24 +857,19 @@ fn histogram_payload_and_present_all_null_representation_survive() {
else {
panic!("histogram representation was normalized");
};
assert_eq!(histograms.batch().len(), 2);
assert_eq!(histograms.indices().len(), 2);
assert!(histograms.indices().is_valid(0));
assert!(histograms.indices().is_valid(1));
assert_eq!(histograms.indices().value(0), 0);
assert_eq!(histograms.indices().value(1), 0);
assert_histogram(histograms.batch().view(0).unwrap(), -9.5);
assert_unreferenced_histogram(histograms.batch().view(1).unwrap());
assert_histogram(histograms.view_at(0).unwrap(), -9.5);
assert_histogram(histograms.view_at(1).unwrap(), -9.5);
let DenseSamplesRef::Histogram(all_null_histograms) =
decoded[0].fields.get(1).unwrap().dense().unwrap()
else {
panic!("all-null histogram representation was normalized");
};
assert_eq!(all_null_histograms.batch().len(), 2);
assert_eq!(all_null_histograms.indices().len(), 2);
assert!((0..2).all(|i| !all_null_histograms.indices().is_valid(i)));
assert_histogram(all_null_histograms.batch().view(0).unwrap(), -9.5);
assert_unreferenced_histogram(all_null_histograms.batch().view(1).unwrap());
let DenseSamplesRef::Float(all_null_float) = decoded[0].fields.get(2).unwrap().dense().unwrap()
else {
panic!("all-null float representation was normalized");
@@ -877,16 +912,14 @@ fn histogram_payload_and_present_all_null_representation_survive() {
panic!("all-null mixed representation was normalized");
};
assert_eq!(mixed.floats().len(), 2);
assert_eq!(mixed.histograms().batch().len(), 2);
assert_eq!(mixed.histograms().indices().len(), 2);
assert!(
(0..2).all(|i| !mixed.floats().is_valid(i) && !mixed.histograms().indices().is_valid(i))
);
assert_histogram(mixed.histograms().batch().view(0).unwrap(), -9.5);
assert_unreferenced_histogram(mixed.histograms().batch().view(1).unwrap());
}
#[test]
fn rejects_grid_and_invalid_schema_even_for_empty_batch_and_accepts_valid_empty() {
fn rejects_invalid_schema_and_metadata_and_accepts_valid_empty() {
let schema = test_schema(&["value"]);
let grid_schema = SeriesSchema {
fields: schema.fields.clone(),
@@ -1000,7 +1033,15 @@ fn rejects_grid_and_invalid_schema_even_for_empty_batch_and_accepts_valid_empty(
.unwrap();
assert!(unpack_raw_series_frame(&schema, RawLabelMode::Key, &malformed).is_err());
}
}
#[test]
fn rejects_invalid_timestamps_and_empty_raw_payloads() {
let schema = test_schema(&["value"]);
let chunk = float_chunk(&schema, &[1], None, vec![1], vec![Some(2.0)], false);
let valid = pack_raw_series_chunk(&schema, RawLabelMode::Key, &chunk)
.unwrap()
.remove(0);
for bad_timestamps in [vec![], vec![1, 1], vec![2, 1]] {
let malformed = with_timestamp_values(&valid, bad_timestamps);
assert!(unpack_raw_series_frame(&schema, RawLabelMode::Key, &malformed).is_err());
@@ -1107,6 +1148,23 @@ fn rejects_grid_and_invalid_schema_even_for_empty_batch_and_accepts_valid_empty(
// the decoder can inspect it; keep this a constructor-invariant assertion.
assert!(arrow::record_batch::RecordBatch::try_new(valid.schema(), columns).is_err());
let mut bad = valid.columns().to_vec();
bad[2] = Arc::new(
with_timestamp_values(&valid, vec![1, 1])
.column(2)
.as_any()
.downcast_ref::<StructArray>()
.unwrap()
.clone(),
);
let invalid_later_row = arrow::record_batch::RecordBatch::try_new(valid.schema(), bad).unwrap();
let later_invalid =
concat_batches(&valid.schema(), &[valid.clone(), invalid_later_row]).unwrap();
assert!(unpack_raw_series_frame(&schema, RawLabelMode::Key, &later_invalid).is_err());
}
#[test]
fn rejects_invalid_explicit_labels() {
let explicit_schema = test_schema(&["value"]);
let explicit_chunk = float_chunk(
&explicit_schema,
@@ -1213,21 +1271,10 @@ fn rejects_grid_and_invalid_schema_even_for_empty_batch_and_accepts_valid_empty(
let reserved =
arrow::record_batch::RecordBatch::try_new(explicit.schema(), reserved_labels).unwrap();
assert!(unpack_raw_series_frame(&explicit_schema, RawLabelMode::Explicit, &reserved).is_err());
}
let mut bad = valid.columns().to_vec();
bad[2] = Arc::new(
with_timestamp_values(&valid, vec![1, 1])
.column(2)
.as_any()
.downcast_ref::<StructArray>()
.unwrap()
.clone(),
);
let invalid_later_row = arrow::record_batch::RecordBatch::try_new(valid.schema(), bad).unwrap();
let later_invalid =
concat_batches(&valid.schema(), &[valid.clone(), invalid_later_row]).unwrap();
assert!(unpack_raw_series_frame(&schema, RawLabelMode::Key, &later_invalid).is_err());
#[test]
fn rejects_malformed_histogram_and_payload_encodings() {
let hist_schema = test_schema(&["hist"]);
let valid_hist = pack_raw_series_chunk(
&hist_schema,
src/datatypes/src/series/samples.rs
+10 -5
View File
@@ -355,8 +355,7 @@ impl Samples {
}
}
/// Materialize chunks into one dense array. When all histogram-bearing chunks reference the
/// same table, reuse it; otherwise concatenate tables per chunk and remap indices, including
/// unused rows.
/// same Arrow table view, reuse it; otherwise concatenate tables per chunk and remap indices.
pub fn into_dense(self) -> Result<Self> {
let SampleStorage::Chunks(chunks, total_len) = self.storage else {
return Ok(self);
@@ -390,9 +389,15 @@ impl Samples {
SampleStorage::Mixed(mixed) => Some(&mixed.histograms.batch),
SampleStorage::Float(_) | SampleStorage::Chunks(_, _) => None,
});
let shared_batch = histogram_batches
.next()
.filter(|first| histogram_batches.all(|batch| Arc::ptr_eq(first, batch)));
let shared_batch = histogram_batches.next().filter(|first| {
histogram_batches.all(|batch| {
Arc::ptr_eq(first, batch)
|| first
.as_arrow()
.to_data()
.ptr_eq(&batch.as_arrow().to_data())
})
});
let needs_float = chunks.iter().any(|chunk| {
matches!(
chunk.storage,
src/datatypes/src/series/samples/tests.rs
+304 -2
View File
@@ -15,8 +15,8 @@
use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};
use arrow::array::{Float64Array, UInt32Array};
use arrow::buffer::{Buffer, ScalarBuffer};
use arrow::array::{ArrayRef, Float64Array, Int64Array, StructArray, UInt32Array};
use arrow::buffer::{BooleanBuffer, Buffer, NullBuffer, ScalarBuffer};
use crate::series::histogram::{BucketSideView, CounterResetHint, HistogramBatchBuilder};
@@ -235,6 +235,308 @@ fn chunk_iterator_is_sequential_exact_and_handles_many_singletons() {
assert!(iter.next().is_none());
}
#[test]
fn densification_reuses_distinct_wrappers_of_the_same_arrow_table_view() {
let base_row = HistogramView {
schema: 2,
zero_threshold: 0.25,
sum: 7.0,
count: 8.0,
zero_count: 1.0,
reset_hint: CounterResetHint::Gauge,
start_timestamp: None,
custom_values: &[],
positive: BucketSideView {
span_offsets: &[0],
span_lengths: &[1],
buckets: &[9.0],
},
negative: BucketSideView {
span_offsets: &[],
span_lengths: &[],
buckets: &[],
},
};
let rows = [
HistogramView {
schema: 1,
zero_threshold: 0.0,
sum: 1.0,
count: 2.0,
zero_count: 0.0,
reset_hint: CounterResetHint::Unknown,
start_timestamp: Some(11),
custom_values: &[1.0],
positive: BucketSideView {
span_offsets: &[0],
span_lengths: &[1],
buckets: &[3.0],
},
negative: BucketSideView {
span_offsets: &[],
span_lengths: &[],
buckets: &[],
},
},
HistogramView {
sum: -0.0,
count: f64::NAN,
start_timestamp: Some(42),
custom_values: &[f64::from_bits(0x7ff8_0000_0000_0042)],
..base_row
},
];
let backing = histogram_batch(&[rows[0], rows[1], rows[1]]);
let original =
Arc::new(HistogramBatch::try_from_arrow(backing.as_arrow().slice(1, 1)).unwrap());
let wrapped = Arc::new(HistogramBatch::try_from_arrow(original.as_arrow().clone()).unwrap());
assert!(!Arc::ptr_eq(&original, &wrapped));
assert!(
original
.as_arrow()
.to_data()
.ptr_eq(&wrapped.as_arrow().to_data())
);
let mixed = Samples::try_from_mixed(
Float64Array::from(vec![
Some(-0.0),
Some(f64::from_bits(0x7ff8_0000_0000_0042)),
None,
]),
HistogramSamples::try_new(UInt32Array::from(vec![None, None, Some(0)]), wrapped).unwrap(),
)
.unwrap();
let dense = Samples::try_from_chunks(vec![
Samples::try_from_histogram(UInt32Array::from(vec![Some(0), None]), original.clone())
.unwrap(),
mixed,
])
.unwrap()
.into_dense()
.unwrap();
let DenseSamplesRef::Mixed(values) = dense.dense().unwrap() else {
panic!()
};
assert_eq!(values.histograms.batch().len(), 1);
assert!(Arc::ptr_eq(values.histograms.batch(), &original));
assert_eq!(values.histograms.indices().value(0), 0);
assert!(values.histograms.indices().is_null(1));
assert!(values.histograms.indices().is_null(2));
assert!(values.histograms.indices().is_null(3));
assert_eq!(values.histograms.indices().value(4), 0);
let expected = [
SampleRef::Histogram(original.view(0).unwrap()),
SampleRef::Missing,
SampleRef::Float(-0.0),
SampleRef::Float(f64::from_bits(0x7ff8_0000_0000_0042)),
SampleRef::Histogram(original.view(0).unwrap()),
];
for (index, expected) in expected.into_iter().enumerate() {
assert!(
same_sample(dense.sample(index), expected),
"position {index}"
);
}
}
#[test]
fn densification_does_not_merge_equal_content_distinct_buffers() {
let first = batch(2.0, 3.0, &[4.0]);
let equal = batch(2.0, 3.0, &[4.0]);
assert!(
!first
.as_arrow()
.to_data()
.ptr_eq(&equal.as_arrow().to_data())
);
let dense = Samples::try_from_chunks(vec![
Samples::try_from_histogram(UInt32Array::from(vec![Some(0)]), first.clone()).unwrap(),
Samples::try_from_histogram(UInt32Array::from(vec![Some(0)]), equal.clone()).unwrap(),
])
.unwrap()
.into_dense()
.unwrap();
let DenseSamplesRef::Histogram(values) = dense.dense().unwrap() else {
panic!()
};
assert_eq!(values.batch().len(), 2);
assert_eq!(values.indices().value(0), 0);
assert_eq!(values.indices().value(1), 1);
assert!(same_sample(
dense.sample(0),
SampleRef::Histogram(first.view(0).unwrap())
));
assert!(same_sample(
dense.sample(1),
SampleRef::Histogram(equal.view(0).unwrap())
));
}
#[test]
fn densification_does_not_merge_distinct_table_slices() {
let base_row = HistogramView {
schema: 1,
zero_threshold: 0.0,
sum: 2.0,
count: 3.0,
zero_count: 0.0,
reset_hint: CounterResetHint::Unknown,
start_timestamp: None,
custom_values: &[],
positive: BucketSideView {
span_offsets: &[0],
span_lengths: &[1],
buckets: &[4.0],
},
negative: BucketSideView {
span_offsets: &[],
span_lengths: &[],
buckets: &[],
},
};
let rows = [
base_row,
HistogramView {
start_timestamp: Some(42),
sum: 12.0,
..base_row
},
HistogramView {
start_timestamp: Some(99),
sum: 19.0,
..base_row
},
];
let backing = histogram_batch(&rows);
let first = Arc::new(HistogramBatch::try_from_arrow(backing.as_arrow().slice(0, 1)).unwrap());
let equal_len_other_offset =
Arc::new(HistogramBatch::try_from_arrow(backing.as_arrow().slice(1, 1)).unwrap());
let longer_other_offset =
Arc::new(HistogramBatch::try_from_arrow(backing.as_arrow().slice(1, 2)).unwrap());
for (second, expected_table_len, second_expected) in [
(
equal_len_other_offset,
2,
SampleRef::Histogram(backing.view(1).unwrap()),
),
(
longer_other_offset,
3,
SampleRef::Histogram(backing.view(1).unwrap()),
),
] {
let dense = Samples::try_from_chunks(vec![
Samples::try_from_histogram(UInt32Array::from(vec![Some(0)]), first.clone()).unwrap(),
Samples::try_from_histogram(UInt32Array::from(vec![Some(0)]), second).unwrap(),
])
.unwrap()
.into_dense()
.unwrap();
let DenseSamplesRef::Histogram(values) = dense.dense().unwrap() else {
panic!()
};
assert_eq!(values.batch().len(), expected_table_len);
assert_eq!(values.indices().value(0), 0);
assert_eq!(values.indices().value(1), 1);
assert!(same_sample(
dense.sample(0),
SampleRef::Histogram(first.view(0).unwrap())
));
assert!(same_sample(dense.sample(1), second_expected));
}
}
#[test]
fn densification_does_not_merge_shared_value_buffer_with_different_null_bitmap_offsets() {
let source = batch(2.0, 3.0, &[4.0]);
let start_values = ScalarBuffer::from(vec![42_i64]);
let bitmap = Buffer::from(vec![0b10_u8]);
let missing_starts = Int64Array::new(
start_values.clone(),
Some(NullBuffer::new(BooleanBuffer::new(bitmap.clone(), 0, 1))),
);
let present_starts = Int64Array::new(
start_values.clone(),
Some(NullBuffer::new(BooleanBuffer::new(bitmap.clone(), 1, 1))),
);
assert_eq!(
missing_starts.values().inner().as_ptr(),
present_starts.values().inner().as_ptr()
);
let missing_nulls = missing_starts.nulls().unwrap();
let present_nulls = present_starts.nulls().unwrap();
assert_eq!(
missing_nulls.buffer().as_ptr(),
present_nulls.buffer().as_ptr()
);
assert_ne!(missing_nulls.offset(), present_nulls.offset());
assert!(missing_starts.is_null(0));
assert_eq!(present_starts.value(0), 42);
let with_start = |starts: Int64Array| {
let mut columns = source.as_arrow().columns().to_vec();
columns[6] = Arc::new(starts) as ArrayRef;
Arc::new(
HistogramBatch::try_from_arrow(
StructArray::try_new(source.as_arrow().fields().clone(), columns, None).unwrap(),
)
.unwrap(),
)
};
let missing = with_start(missing_starts);
let present = with_start(present_starts);
let missing_starts = missing
.as_arrow()
.column(6)
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
let present_starts = present
.as_arrow()
.column(6)
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
assert_eq!(
missing_starts.values().inner().as_ptr(),
present_starts.values().inner().as_ptr()
);
assert_eq!(
missing_starts.nulls().unwrap().buffer().as_ptr(),
present_starts.nulls().unwrap().buffer().as_ptr()
);
assert_ne!(
missing_starts.nulls().unwrap().offset(),
present_starts.nulls().unwrap().offset()
);
assert!(missing_starts.is_null(0));
assert_eq!(present_starts.value(0), 42);
assert!(
!missing
.as_arrow()
.to_data()
.ptr_eq(&present.as_arrow().to_data())
);
let dense = Samples::try_from_chunks(vec![
Samples::try_from_histogram(UInt32Array::from(vec![Some(0)]), missing.clone()).unwrap(),
Samples::try_from_histogram(UInt32Array::from(vec![Some(0)]), present.clone()).unwrap(),
])
.unwrap()
.into_dense()
.unwrap();
assert_eq!(dense.len(), 2);
assert!(same_sample(
dense.sample(0),
SampleRef::Histogram(missing.view(0).unwrap())
));
assert!(same_sample(
dense.sample(1),
SampleRef::Histogram(present.view(0).unwrap())
));
}
#[test]
fn densifying_slices_keeps_shared_histogram_table() {
let table = batch(2.0, 3.0, &[4.0, 5.0]);