diff --git a/src/datatypes/src/series/frame/tests.rs b/src/datatypes/src/series/frame/tests.rs index beb5f9ae0ec..6247c4924ee 100644 --- a/src/datatypes/src/series/frame/tests.rs +++ b/src/datatypes/src/series/frame/tests.rs @@ -247,6 +247,98 @@ fn assert_float_bits(actual: &Samples, expected: &[Option]) { } } +#[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::>().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::() + .unwrap(); + let second_values = second_table + .as_arrow() + .column(2) + .as_any() + .downcast_ref::() + .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::() + .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::>(); 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::() + .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::() - .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, diff --git a/src/datatypes/src/series/samples.rs b/src/datatypes/src/series/samples.rs index eb02ccfd4db..6b2056df691 100644 --- a/src/datatypes/src/series/samples.rs +++ b/src/datatypes/src/series/samples.rs @@ -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 { 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, diff --git a/src/datatypes/src/series/samples/tests.rs b/src/datatypes/src/series/samples/tests.rs index 3f672ed418a..b623937945d 100644 --- a/src/datatypes/src/series/samples/tests.rs +++ b/src/datatypes/src/series/samples/tests.rs @@ -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::() + .unwrap(); + let present_starts = present + .as_arrow() + .column(6) + .as_any() + .downcast_ref::() + .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]);