1use linkrs_core::types::{EdgeId, Timestamp};
7use linkrs_core::{StorageError, StorageResult, Value};
8
9use std::collections::HashMap;
10use std::sync::Arc;
11
12use crate::column_stats::{compute_stats_streaming, deserialize_stat_value, serialize_stat_value};
13use crate::encoding::EncodingType;
14
15#[derive(Debug, Clone)]
17pub struct TombstoneStats {
18 pub count: usize,
20 pub memory_bytes: usize,
22 pub oldest_delete_ts: Option<Timestamp>,
24 pub newest_delete_ts: Option<Timestamp>,
26}
27
28impl TombstoneStats {
29 pub fn estimate_memory(count: usize) -> usize {
36 const HASH_NODE_OVERHEAD: usize = 48;
37 const AUTHORITY_RECORD: usize =
38 std::mem::size_of::<EdgeId>() + 2 * std::mem::size_of::<Timestamp>();
39 count * (AUTHORITY_RECORD + HASH_NODE_OVERHEAD)
40 }
41}
42
43#[derive(Debug, Clone, Default)]
50pub struct DeletionStats {
51 pub total_deleted_edges: u64,
53 pub total_live_edges: u64,
55}
56
57impl DeletionStats {
58 pub fn deletion_ratio(&self) -> f64 {
60 let total = self
61 .total_live_edges
62 .saturating_add(self.total_deleted_edges);
63 if total == 0 {
64 0.0
65 } else {
66 self.total_deleted_edges as f64 / total as f64
67 }
68 }
69
70 pub fn deletion_percentage(&self) -> f64 {
72 self.deletion_ratio() * 100.0
73 }
74
75 pub fn is_significant(&self) -> bool {
77 self.deletion_ratio() > 0.1
78 }
79}
80
81#[derive(Debug, Clone)]
92pub struct GroupSegmentStats {
93 pub group: u32,
94 pub row_count: u64,
95 pub live_count: u64,
96 pub null_count: u64,
97 pub sort_min: Option<Value>,
98 pub sort_max: Option<Value>,
99 pub columns: HashMap<Arc<str>, crate::column_stats::ColumnStats>,
100}
101
102impl GroupSegmentStats {
103 pub fn collect(
107 group: u32,
108 row_count: u64,
109 live_count: u64,
110 endpoints: &[u32],
111 column_values: &HashMap<Arc<str>, Vec<Option<Value>>>,
112 column_encodings: &HashMap<Arc<str>, EncodingType>,
113 ) -> Self {
114 let endpoint_values = endpoints
115 .iter()
116 .map(|endpoint| Some(Value::BigInt(*endpoint as i64)));
117 let endpoint_stats = compute_stats_streaming(endpoint_values, EncodingType::None, 0, 0);
118 let (sort_min, sort_max) = (
119 endpoint_stats.min_value.clone(),
120 endpoint_stats.max_value.clone(),
121 );
122 let mut null_count = 0u64;
123 let mut columns = HashMap::with_capacity(column_values.len());
124 for (name, values) in column_values {
125 let encoding = column_encodings
126 .get(name)
127 .copied()
128 .unwrap_or(EncodingType::None);
129 let stats = compute_stats_streaming(values.iter().cloned(), encoding, 0, 0);
130 null_count = null_count.saturating_add(stats.null_count);
131 columns.insert(name.clone(), stats);
132 }
133 Self {
134 group,
135 row_count,
136 live_count,
137 null_count,
138 sort_min,
139 sort_max,
140 columns,
141 }
142 }
143
144 pub fn widen_with(&mut self, fresh: &GroupSegmentStats) {
151 self.row_count = fresh.row_count;
152 self.live_count = fresh.live_count;
153 self.null_count = fresh.null_count;
154 match (&self.sort_min, &fresh.sort_min) {
155 (_, None) => {}
156 (None, Some(_)) => self.sort_min = fresh.sort_min.clone(),
157 (Some(current), Some(next)) => {
158 if crate::cursor::compare_stat_values(next, current) == std::cmp::Ordering::Less {
159 self.sort_min = fresh.sort_min.clone();
160 }
161 }
162 }
163 match (&self.sort_max, &fresh.sort_max) {
164 (_, None) => {}
165 (None, Some(_)) => self.sort_max = fresh.sort_max.clone(),
166 (Some(current), Some(next)) => {
167 if crate::cursor::compare_stat_values(next, current) == std::cmp::Ordering::Greater
168 {
169 self.sort_max = fresh.sort_max.clone();
170 }
171 }
172 }
173 for (name, fresh_stats) in &fresh.columns {
174 match self.columns.get_mut(name) {
175 None => {
176 self.columns.insert(name.clone(), fresh_stats.clone());
177 }
178 Some(current) => {
179 match (¤t.min_value, &fresh_stats.min_value) {
180 (_, None) => {}
181 (None, Some(_)) => current.min_value = fresh_stats.min_value.clone(),
182 (Some(a), Some(b)) => {
183 if crate::cursor::compare_stat_values(b, a) == std::cmp::Ordering::Less
184 {
185 current.min_value = fresh_stats.min_value.clone();
186 }
187 }
188 }
189 match (¤t.max_value, &fresh_stats.max_value) {
190 (_, None) => {}
191 (None, Some(_)) => current.max_value = fresh_stats.max_value.clone(),
192 (Some(a), Some(b)) => {
193 if crate::cursor::compare_stat_values(b, a)
194 == std::cmp::Ordering::Greater
195 {
196 current.max_value = fresh_stats.max_value.clone();
197 }
198 }
199 }
200 current.null_count = fresh_stats.null_count;
201 match (¤t.hll, &fresh_stats.hll) {
202 (Some(_), Some(fresh_hll)) => {
203 if let Some(merged) = current.hll.clone() {
204 let mut combined = merged;
205 combined.merge(fresh_hll);
206 current.distinct_count = Some(combined.estimate());
207 current.hll = Some(combined);
208 } else {
209 current.hll = fresh_stats.hll.clone();
210 current.distinct_count = fresh_stats.distinct_count;
211 }
212 }
213 (None, Some(_)) => {
214 current.hll = fresh_stats.hll.clone();
215 current.distinct_count = fresh_stats.distinct_count;
216 }
217 _ => {
218 current.distinct_count = fresh_stats.distinct_count;
219 }
220 }
221 current.encoding_type = fresh_stats.encoding_type;
222 }
223 }
224 }
225 }
226
227 pub fn may_contain(&self, predicates: &[crate::cursor::ScanPredicate]) -> bool {
235 if predicates.is_empty() {
236 return true;
237 }
238 for range in crate::cursor::ScanPredicate::merged_ranges(predicates) {
239 let Some(stats) = self.columns.get(&*range.column) else {
240 continue;
241 };
242 let (Some(min), Some(max)) = (stats.min_value.as_ref(), stats.max_value.as_ref())
243 else {
244 continue;
245 };
246 if !range.overlaps(min, max) {
247 return false;
248 }
249 }
250 true
251 }
252
253 pub fn encode(&self) -> Vec<u8> {
254 let mut out = Vec::new();
255 out.extend_from_slice(&self.group.to_le_bytes());
256 out.extend_from_slice(&self.row_count.to_le_bytes());
257 out.extend_from_slice(&self.live_count.to_le_bytes());
258 out.extend_from_slice(&self.null_count.to_le_bytes());
259 out.push(self.sort_min.is_some() as u8);
260 if let Some(ref value) = self.sort_min {
261 let mut buf = Vec::new();
262 if serialize_stat_value(&mut buf, value).is_ok() {
263 out.extend_from_slice(&(buf.len() as u32).to_le_bytes());
264 out.extend_from_slice(&buf);
265 } else {
266 out.extend_from_slice(&0u32.to_le_bytes());
267 }
268 }
269 out.push(self.sort_max.is_some() as u8);
270 if let Some(ref value) = self.sort_max {
271 let mut buf = Vec::new();
272 if serialize_stat_value(&mut buf, value).is_ok() {
273 out.extend_from_slice(&(buf.len() as u32).to_le_bytes());
274 out.extend_from_slice(&buf);
275 } else {
276 out.extend_from_slice(&0u32.to_le_bytes());
277 }
278 }
279 let mut names: Vec<&Arc<str>> = self.columns.keys().collect();
280 names.sort();
281 out.extend_from_slice(&(names.len() as u32).to_le_bytes());
282 for name in names {
283 let bytes = name.as_bytes();
284 out.extend_from_slice(&(bytes.len() as u32).to_le_bytes());
285 out.extend_from_slice(bytes);
286 let mut stats_buf = Vec::new();
287 if self.columns[name].serialize_meta(&mut stats_buf).is_ok() {
288 out.extend_from_slice(&(stats_buf.len() as u32).to_le_bytes());
289 out.extend_from_slice(&stats_buf);
290 } else {
291 out.extend_from_slice(&0u32.to_le_bytes());
292 }
293 }
294 out
295 }
296
297 pub fn decode(data: &[u8], cursor: &mut usize) -> StorageResult<Self> {
298 let take = |data: &[u8], cursor: &mut usize, len: usize| -> StorageResult<Vec<u8>> {
299 if data.len().saturating_sub(*cursor) < len {
300 return Err(StorageError::deserialize_error(
301 "segment stats record too short",
302 ));
303 }
304 let slice = data[*cursor..*cursor + len].to_vec();
305 *cursor += len;
306 Ok(slice)
307 };
308 let group = u32::from_le_bytes(
309 take(data, cursor, 4)?
310 .try_into()
311 .map_err(|_| StorageError::deserialize_error("segment group too short"))?,
312 );
313 let row_count = u64::from_le_bytes(
314 take(data, cursor, 8)?
315 .try_into()
316 .map_err(|_| StorageError::deserialize_error("segment row count too short"))?,
317 );
318 let live_count = u64::from_le_bytes(
319 take(data, cursor, 8)?
320 .try_into()
321 .map_err(|_| StorageError::deserialize_error("segment live count too short"))?,
322 );
323 let null_count = u64::from_le_bytes(
324 take(data, cursor, 8)?
325 .try_into()
326 .map_err(|_| StorageError::deserialize_error("segment null count too short"))?,
327 );
328 let has_min = take(data, cursor, 1)?[0] != 0;
329 let sort_min = if has_min {
330 let len = u32::from_le_bytes(take(data, cursor, 4)?.try_into().map_err(|_| {
331 StorageError::deserialize_error("segment sort min length too short")
332 })?) as usize;
333 let bytes = take(data, cursor, len)?;
334 let mut slice = &bytes[..];
335 Some(deserialize_stat_value(&mut slice)?)
336 } else {
337 None
338 };
339 let has_max = take(data, cursor, 1)?[0] != 0;
340 let sort_max = if has_max {
341 let len = u32::from_le_bytes(take(data, cursor, 4)?.try_into().map_err(|_| {
342 StorageError::deserialize_error("segment sort max length too short")
343 })?) as usize;
344 let bytes = take(data, cursor, len)?;
345 let mut slice = &bytes[..];
346 Some(deserialize_stat_value(&mut slice)?)
347 } else {
348 None
349 };
350 let column_count = u32::from_le_bytes(
351 take(data, cursor, 4)?
352 .try_into()
353 .map_err(|_| StorageError::deserialize_error("segment column count too short"))?,
354 ) as usize;
355 let mut columns = HashMap::with_capacity(column_count);
356 for _ in 0..column_count {
357 let name_len = u32::from_le_bytes(take(data, cursor, 4)?.try_into().map_err(|_| {
358 StorageError::deserialize_error("segment column name length too short")
359 })?) as usize;
360 let name_bytes = take(data, cursor, name_len)?;
361 let name = String::from_utf8(name_bytes)
362 .map_err(|e| StorageError::deserialize_error(e.to_string()))?;
363 let stats_len = u32::from_le_bytes(take(data, cursor, 4)?.try_into().map_err(|_| {
364 StorageError::deserialize_error("segment column stats length too short")
365 })?) as usize;
366 let stats_bytes = take(data, cursor, stats_len)?;
367 let mut slice = &stats_bytes[..];
368 let stats = crate::column_stats::ColumnStats::deserialize_meta(&mut slice)?;
369 if !slice.is_empty() {
370 return Err(StorageError::deserialize_error(
371 "unexpected trailing data in segment column stats",
372 ));
373 }
374 columns.insert(Arc::from(name.as_str()), stats);
375 }
376 Ok(Self {
377 group,
378 row_count,
379 live_count,
380 null_count,
381 sort_min,
382 sort_max,
383 columns,
384 })
385 }
386}
387
388pub fn encode_segment_snapshot(stats: &HashMap<u32, GroupSegmentStats>) -> Vec<u8> {
390 let mut groups: Vec<u32> = stats.keys().copied().collect();
391 groups.sort_unstable();
392 let mut out = Vec::new();
393 out.extend_from_slice(&(groups.len() as u32).to_le_bytes());
394 for group in groups {
395 out.extend_from_slice(&stats[&group].encode());
396 }
397 out
398}
399
400pub fn decode_segment_snapshot(data: &[u8]) -> StorageResult<HashMap<u32, GroupSegmentStats>> {
403 let mut cursor = 0usize;
404 let take = |data: &[u8], cursor: &mut usize, len: usize| -> StorageResult<Vec<u8>> {
405 if data.len().saturating_sub(*cursor) < len {
406 return Err(StorageError::deserialize_error(
407 "segment snapshot too short",
408 ));
409 }
410 let slice = data[*cursor..*cursor + len].to_vec();
411 *cursor += len;
412 Ok(slice)
413 };
414 let count = u32::from_le_bytes(
415 take(data, &mut cursor, 4)?
416 .try_into()
417 .map_err(|_| StorageError::deserialize_error("segment snapshot count too short"))?,
418 ) as usize;
419 let mut out = HashMap::with_capacity(count);
420 for _ in 0..count {
421 let record = GroupSegmentStats::decode(data, &mut cursor)?;
422 out.insert(record.group, record);
423 }
424 if cursor != data.len() {
425 return Err(StorageError::deserialize_error(
426 "unexpected trailing data in segment snapshot",
427 ));
428 }
429 Ok(out)
430}
431
432#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
434pub struct ScanPruneReport {
435 pub segments_total: usize,
436 pub segments_pruned: usize,
437 pub rows_scanned: usize,
438 pub rows_filtered: usize,
439}
440
441impl ScanPruneReport {
442 pub fn prune_rate(&self) -> f64 {
443 if self.segments_total == 0 {
444 0.0
445 } else {
446 self.segments_pruned as f64 / self.segments_total as f64
447 }
448 }
449
450 pub fn filter_rate(&self) -> f64 {
451 if self.rows_scanned == 0 {
452 0.0
453 } else {
454 self.rows_filtered as f64 / self.rows_scanned as f64
455 }
456 }
457}
458
459#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
468pub struct FormProfileSnapshot {
469 pub width_bytes: u64,
470 pub width_samples: u64,
471 pub reads: u64,
472 pub writes: u64,
473}
474
475impl FormProfileSnapshot {
476 pub fn is_unknown(&self) -> bool {
477 self.width_samples == 0
478 }
479
480 pub fn avg_width_bytes(&self) -> Option<f64> {
481 if self.width_samples == 0 {
482 None
483 } else {
484 Some(self.width_bytes as f64 / self.width_samples as f64)
485 }
486 }
487
488 pub fn write_share(&self) -> Option<f64> {
489 let total = self.reads.saturating_add(self.writes);
490 if total == 0 {
491 None
492 } else {
493 Some(self.writes as f64 / total as f64)
494 }
495 }
496
497 pub fn read_to_write_ratio(&self) -> Option<f64> {
498 if self.writes == 0 {
499 None
500 } else {
501 Some(self.reads as f64 / self.writes as f64)
502 }
503 }
504
505 pub fn encode(&self) -> Vec<u8> {
506 let mut out = Vec::with_capacity(32);
507 out.extend_from_slice(&self.width_bytes.to_le_bytes());
508 out.extend_from_slice(&self.width_samples.to_le_bytes());
509 out.extend_from_slice(&self.reads.to_le_bytes());
510 out.extend_from_slice(&self.writes.to_le_bytes());
511 out
512 }
513
514 pub fn decode(data: &[u8]) -> StorageResult<Self> {
515 if data.len() != 32 {
516 return Err(StorageError::deserialize_error(
517 "form profile snapshot must hold 32 bytes",
518 ));
519 }
520 let mut cursor = 0usize;
521 let take_u64 = |data: &[u8], cursor: &mut usize| {
522 let mut bytes = [0u8; 8];
523 bytes.copy_from_slice(&data[*cursor..*cursor + 8]);
524 *cursor += 8;
525 u64::from_le_bytes(bytes)
526 };
527 Ok(Self {
528 width_bytes: take_u64(data, &mut cursor),
529 width_samples: take_u64(data, &mut cursor),
530 reads: take_u64(data, &mut cursor),
531 writes: take_u64(data, &mut cursor),
532 })
533 }
534}
535
536pub fn estimate_value_bytes(value: &Value) -> u64 {
542 match value {
543 Value::Empty => 0,
544 Value::Null(_) => 0,
545 Value::Bool(_) => 1,
546 Value::SmallInt(_) => 2,
547 Value::Int(_) => 4,
548 Value::BigInt(_) => 8,
549 Value::Float(_) => 4,
550 Value::Double(_) => 8,
551 Value::Decimal128(_) => 16,
552 Value::String(s) => s.len() as u64,
553 Value::FixedString(s) => s.len() as u64,
554 Value::Blob(b) => b.len() as u64,
555 Value::Date(_) => 8,
556 Value::Time(_) => 8,
557 Value::DateTime(_) => 8,
558 Value::Uuid(_) => 16,
559 Value::Interval(_) => 16,
560 Value::Json(_) | Value::JsonB(_) => 32,
561 Value::List(_) | Value::Map(_) | Value::Set(_) => 32,
562 Value::Geography(_) => 32,
563 Value::Vector(v) => (v.dimension() as u64).saturating_mul(4),
564 Value::DataSet(_) => 32,
565 Value::Struct(_) | Value::Array(_) => 32,
566 Value::Vertex(_) | Value::Edge(_) | Value::EdgeHeader(_) | Value::Path(_) => 32,
567 Value::VertexId(_) | Value::EdgeId(_) => 8,
568 }
569}
570
571pub fn estimate_props_bytes(props: &[(std::sync::Arc<str>, Value)]) -> u64 {
573 props.iter().map(|(_, v)| estimate_value_bytes(v)).sum()
574}
575
576#[cfg(test)]
577mod tests {
578 use super::super::super::{EdgeSchema, EdgeStrategy, RecordForm};
579 use super::*;
580 use crate::edge::edge_table::config::EdgeTableConfig;
581 use crate::edge::edge_table::core::EdgeStore;
582 use crate::types::StoragePropertyDef;
583 use linkrs_core::types::DataType;
584 use linkrs_core::Value;
585
586 fn create_edge_table() -> EdgeStore {
587 let schema = EdgeSchema {
588 label_id: 0,
589 label_name: "knows".into(),
590 src_label: 0,
591 dst_label: 0,
592 properties: vec![],
593 oe_strategy: EdgeStrategy::Multiple,
594 ie_strategy: EdgeStrategy::Multiple,
595 schema_version: 1,
596 record_form: RecordForm::default(),
597 };
598 EdgeStore::with_config(schema, EdgeTableConfig::default()).unwrap()
599 }
600
601 fn create_edge_table_with_props() -> EdgeStore {
602 let schema = EdgeSchema {
603 label_id: 0,
604 label_name: "knows".into(),
605 src_label: 0,
606 dst_label: 0,
607 properties: vec![StoragePropertyDef {
608 name: "weight".into(),
609 data_type: DataType::Double,
610 nullable: false,
611 default_value: Some(Value::Double(0.0)),
612 }],
613 oe_strategy: EdgeStrategy::Multiple,
614 ie_strategy: EdgeStrategy::Multiple,
615 schema_version: 1,
616 record_form: RecordForm::default(),
617 };
618 EdgeStore::with_config(schema, EdgeTableConfig::default()).unwrap()
619 }
620
621 #[test]
622 fn test_deletion_ratio() {
623 let mut stats = DeletionStats::default();
624
625 assert_eq!(stats.deletion_ratio(), 0.0);
626 assert_eq!(stats.deletion_percentage(), 0.0);
627 assert!(!stats.is_significant());
628
629 stats.total_live_edges = 100;
630 stats.total_deleted_edges = 50;
631 assert!((stats.deletion_ratio() - 50.0 / 150.0).abs() < 1e-9);
632 assert!((stats.deletion_percentage() - 100.0 * 50.0 / 150.0).abs() < 1e-9);
633 assert!(stats.is_significant());
634
635 stats.total_deleted_edges = 5;
636 assert!((stats.deletion_ratio() - 5.0 / 105.0).abs() < 1e-9);
637 assert!(!stats.is_significant());
638
639 stats.total_live_edges = 0;
641 stats.total_deleted_edges = 3;
642 assert_eq!(stats.deletion_ratio(), 1.0);
643 }
644
645 #[test]
646 fn test_tombstone_stats_accuracy() {
647 let mut table = create_edge_table_with_props();
648
649 table
650 .insert_edge(0, 1, 0, &[("weight".into(), Value::Double(1.0))], 50)
651 .unwrap();
652 table
653 .insert_edge(0, 2, 0, &[("weight".into(), Value::Double(2.0))], 100)
654 .unwrap();
655 table
656 .insert_edge(0, 3, 0, &[("weight".into(), Value::Double(3.0))], 150)
657 .unwrap();
658
659 table.delete_edge(0, 1, 0, 200).unwrap();
660 table.delete_edge(0, 2, 0, 250).unwrap();
661 table.delete_edge(0, 3, 0, 300).unwrap();
662
663 let stats = table.mvcc.tombstone_stats();
664 assert_eq!(stats.count, 3);
666 assert!(stats.memory_bytes > 0);
667 assert_eq!(stats.oldest_delete_ts, Some(200));
668 assert_eq!(stats.newest_delete_ts, Some(300));
669 }
670
671 #[test]
672 fn test_deletion_stats_tracking() {
673 let mut table = create_edge_table_with_props();
674
675 for i in 0..5u64 {
676 table
677 .insert_edge(
678 0,
679 1,
680 i as i64,
681 &[("weight".into(), Value::Double(i as f64))],
682 100 + i,
683 )
684 .unwrap();
685 }
686
687 let stats = table.deletion_stats();
688 assert_eq!(stats.total_deleted_edges, 0);
689 assert_eq!(stats.deletion_percentage(), 0.0);
690
691 table.delete_edge(0, 1, 0, 110).unwrap();
692 table.delete_edge(0, 1, 1, 111).unwrap();
693
694 let stats = table.deletion_stats();
695 assert_eq!(stats.total_deleted_edges, 2);
696 assert!(stats.deletion_percentage() >= 0.0);
697 }
698
699 #[test]
700 fn test_deletion_stats_complete_table_deletion() {
701 let mut table = create_edge_table();
702
703 for i in 0..3 {
704 table.insert_edge(0, 1, i as i64, &[], 100).unwrap();
705 }
706
707 for i in 0..3 {
708 table.delete_edge(0, 1, i as i64, 110).unwrap();
709 }
710
711 let stats = table.deletion_stats();
712 assert_eq!(stats.total_deleted_edges, 3);
713 assert_eq!(stats.total_live_edges, 0);
714 }
715
716 #[test]
717 fn test_segment_stats_collect_prune_and_roundtrip() {
718 use crate::cursor::ScanPredicate;
719 use std::collections::HashMap;
720
721 let mut column_values: HashMap<Arc<str>, Vec<Option<Value>>> = HashMap::new();
722 column_values.insert(
723 "weight".into(),
724 vec![
725 Some(Value::Double(1.0)),
726 Some(Value::Double(2.0)),
727 None,
728 Some(Value::Double(9.0)),
729 ],
730 );
731 let stats =
732 GroupSegmentStats::collect(0, 4096, 3, &[10, 20, 30], &column_values, &HashMap::new());
733 assert_eq!(stats.group, 0);
734 assert_eq!(stats.row_count, 4096);
735 assert_eq!(stats.live_count, 3);
736 assert_eq!(stats.null_count, 1);
737 assert_eq!(stats.sort_min, Some(Value::BigInt(10)));
738 assert_eq!(stats.sort_max, Some(Value::BigInt(30)));
739
740 let inside = vec![ScanPredicate::ColumnEqual {
741 column: "weight".into(),
742 value: Value::Double(2.0),
743 }];
744 assert!(stats.may_contain(&inside));
745 let outside = vec![ScanPredicate::ColumnRange {
746 column: "weight".into(),
747 lower: Some(Value::Double(100.0)),
748 upper: None,
749 include_lower: true,
750 include_upper: true,
751 }];
752 assert!(!stats.may_contain(&outside));
753 let unknown = vec![ScanPredicate::ColumnEqual {
754 column: "missing".into(),
755 value: Value::Double(1.0),
756 }];
757 assert!(stats.may_contain(&unknown));
758
759 let mut cursor = 0usize;
760 let encoded = stats.encode();
761 let decoded = GroupSegmentStats::decode(&encoded, &mut cursor).unwrap();
762 assert_eq!(cursor, encoded.len());
763 assert_eq!(decoded.group, stats.group);
764 assert_eq!(decoded.row_count, stats.row_count);
765 assert_eq!(decoded.live_count, stats.live_count);
766 assert_eq!(decoded.null_count, stats.null_count);
767 assert_eq!(decoded.sort_min, stats.sort_min);
768 assert_eq!(decoded.sort_max, stats.sort_max);
769
770 let mut snapshot = HashMap::new();
771 snapshot.insert(0u32, stats);
772 let bytes = encode_segment_snapshot(&snapshot);
773 let restored = decode_segment_snapshot(&bytes).unwrap();
774 assert_eq!(restored.len(), 1);
775 assert!(restored.contains_key(&0));
776
777 let mut bad = bytes.clone();
778 bad.push(0);
779 assert!(decode_segment_snapshot(&bad).is_err());
780 }
781
782 #[test]
783 fn test_segment_stats_widen_only_bounds() {
784 use std::collections::HashMap;
785
786 let mut first_values: HashMap<Arc<str>, Vec<Option<Value>>> = HashMap::new();
787 first_values.insert(
788 "weight".into(),
789 vec![Some(Value::Double(5.0)), Some(Value::Double(6.0))],
790 );
791 let mut stats =
792 GroupSegmentStats::collect(0, 4096, 2, &[5], &first_values, &HashMap::new());
793 let mut second_values: HashMap<Arc<str>, Vec<Option<Value>>> = HashMap::new();
794 second_values.insert("weight".into(), vec![Some(Value::Double(1.0))]);
795 let fresh = GroupSegmentStats::collect(1, 4096, 1, &[50], &second_values, &HashMap::new());
796 stats.widen_with(&fresh);
797 assert_eq!(stats.row_count, 4096);
798 assert_eq!(stats.live_count, 1);
799 assert_eq!(stats.sort_min, Some(Value::BigInt(5)));
800 assert_eq!(stats.sort_max, Some(Value::BigInt(50)));
801 let column = stats.columns.get("weight").unwrap();
802 assert_eq!(column.min_value, Some(Value::Double(1.0)));
803 assert_eq!(column.max_value, Some(Value::Double(6.0)));
804 }
805
806 #[test]
807 fn test_prune_report_rates_are_observable() {
808 let report = ScanPruneReport {
809 segments_total: 4,
810 segments_pruned: 1,
811 rows_scanned: 100,
812 rows_filtered: 25,
813 };
814 assert!((report.prune_rate() - 0.25).abs() < 1e-9);
815 assert!((report.filter_rate() - 0.25).abs() < 1e-9);
816 assert_eq!(ScanPruneReport::default().prune_rate(), 0.0);
817 assert_eq!(ScanPruneReport::default().filter_rate(), 0.0);
818 }
819}