1use crate::shared_entity::{lookup, EntityFields, Fields, SharedEntity};
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10use std::borrow::Cow;
11use std::cmp::Ordering;
12use std::collections::{BTreeMap, BTreeSet, BinaryHeap, HashMap};
13use std::sync::Arc;
14
15use crate::row_scope::ScopeField;
16
17#[derive(Debug, Clone, PartialEq, Eq)]
21pub struct SortKey {
22 sort_value: SortValue,
24 entity_key: String,
26 order: SortOrder,
28 decimal: Option<DecimalKey>,
32}
33
34impl SortKey {
35 fn new(sort_value: SortValue, entity_key: String, order: SortOrder) -> Self {
36 Self {
37 decimal: DecimalKey::of(&sort_value),
38 sort_value,
39 entity_key,
40 order,
41 }
42 }
43
44 fn set_sort_value(&mut self, sort_value: SortValue) {
46 self.decimal = DecimalKey::of(&sort_value);
47 self.sort_value = sort_value;
48 }
49
50 fn ranked(&self) -> Ranked<'_> {
51 Ranked {
52 sort_value: &self.sort_value,
53 decimal: self.decimal,
54 entity_key: &self.entity_key,
55 order: self.order,
56 }
57 }
58}
59
60impl PartialOrd for SortKey {
61 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
62 Some(self.cmp(other))
63 }
64}
65
66impl Ord for SortKey {
67 fn cmp(&self, other: &Self) -> Ordering {
68 self.ranked().cmp(&other.ranked())
69 }
70}
71
72struct Ranked<'a> {
75 sort_value: &'a SortValue,
76 decimal: Option<DecimalKey>,
77 entity_key: &'a str,
78 order: SortOrder,
79}
80
81impl Ranked<'_> {
82 fn cmp_values(&self, other: &Self) -> Ordering {
84 match (self.sort_value, other.sort_value) {
85 (SortValue::String(left), SortValue::String(right)) => {
86 match (&self.decimal, &other.decimal) {
87 (Some(a), Some(b)) => a.cmp(left, b, right),
88 (Some(_), None) => Ordering::Less,
89 (None, Some(_)) => Ordering::Greater,
90 (None, None) => left.cmp(right),
91 }
92 }
93 (left, right) => left.cmp(right),
94 }
95 }
96
97 fn cmp(&self, other: &Self) -> Ordering {
98 if self.order != other.order {
99 return match (self.order, other.order) {
100 (SortOrder::Asc, SortOrder::Desc) => Ordering::Less,
101 (SortOrder::Desc, SortOrder::Asc) => Ordering::Greater,
102 _ => Ordering::Equal,
103 };
104 }
105
106 let sort_order = match (self.sort_value, other.sort_value, self.order) {
111 (SortValue::Null, SortValue::Null, _) => Ordering::Equal,
112 (SortValue::Null, _, _) => Ordering::Greater,
113 (_, SortValue::Null, _) => Ordering::Less,
114 (_, _, SortOrder::Asc) => self.cmp_values(other),
115 (_, _, SortOrder::Desc) => self.cmp_values(other).reverse(),
116 };
117
118 match sort_order {
119 Ordering::Equal => self.entity_key.cmp(other.entity_key),
120 other => other,
121 }
122 }
123}
124
125#[derive(Debug, Clone, PartialEq, Eq)]
127pub enum SortValue {
128 Null,
129 Bool(bool),
130 Integer(i64),
131 Float(OrderedFloat),
132 String(String),
133}
134
135impl Ord for SortValue {
136 fn cmp(&self, other: &Self) -> Ordering {
137 match (self, other) {
138 (SortValue::Null, SortValue::Null) => Ordering::Equal,
139 (SortValue::Null, _) => Ordering::Less,
140 (_, SortValue::Null) => Ordering::Greater,
141 (SortValue::Bool(a), SortValue::Bool(b)) => a.cmp(b),
142 (SortValue::Integer(a), SortValue::Integer(b)) => a.cmp(b),
143 (SortValue::Float(a), SortValue::Float(b)) => a.cmp(b),
144 (SortValue::String(a), SortValue::String(b)) => compare_strings(a, b),
145 (SortValue::Integer(_), SortValue::String(_)) => Ordering::Less,
147 (SortValue::String(_), SortValue::Integer(_)) => Ordering::Greater,
148 (SortValue::Float(_), SortValue::String(_)) => Ordering::Less,
149 (SortValue::String(_), SortValue::Float(_)) => Ordering::Greater,
150 (SortValue::Integer(a), SortValue::Float(b)) => OrderedFloat(*a as f64).cmp(b),
152 (SortValue::Float(a), SortValue::Integer(b)) => a.cmp(&OrderedFloat(*b as f64)),
153 (SortValue::Bool(_), _) => Ordering::Less,
155 (_, SortValue::Bool(_)) => Ordering::Greater,
156 }
157 }
158}
159
160impl PartialOrd for SortValue {
161 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
162 Some(self.cmp(other))
163 }
164}
165
166#[derive(Debug, Clone, Copy, PartialEq)]
168pub struct OrderedFloat(pub f64);
169
170impl Eq for OrderedFloat {}
171
172impl Ord for OrderedFloat {
173 fn cmp(&self, other: &Self) -> Ordering {
174 self.0.partial_cmp(&other.0).unwrap_or_else(|| {
175 if self.0.is_nan() && other.0.is_nan() {
176 Ordering::Equal
177 } else if self.0.is_nan() {
178 Ordering::Less
179 } else {
180 Ordering::Greater
181 }
182 })
183 }
184}
185
186impl PartialOrd for OrderedFloat {
187 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
188 Some(self.cmp(other))
189 }
190}
191
192#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
194#[serde(rename_all = "lowercase")]
195pub enum SortOrder {
196 Asc,
197 Desc,
198}
199
200impl From<crate::materialized_view::SortOrder> for SortOrder {
201 fn from(order: crate::materialized_view::SortOrder) -> Self {
202 match order {
203 crate::materialized_view::SortOrder::Asc => SortOrder::Asc,
204 crate::materialized_view::SortOrder::Desc => SortOrder::Desc,
205 }
206 }
207}
208
209#[derive(Debug, Clone, PartialEq)]
211pub enum ViewDelta {
212 None,
214 Add { key: String, entity: Value },
216 Remove { key: String },
218 Update { key: String, entity: Value },
220}
221
222#[derive(Debug)]
247pub struct SortedViewCache {
248 view_id: String,
250 sort_field: Vec<String>,
252 order: SortOrder,
254 sorted: BTreeMap<SortKey, ()>,
256 entities: HashMap<String, (SortKey, SharedEntity)>,
258 keys_cache: Vec<String>,
260 cache_dirty: bool,
262 placeholder: SharedEntity,
264 partitions: Option<Box<ScopePartitions>>,
268}
269
270impl SortedViewCache {
271 pub fn new(view_id: String, sort_field: Vec<String>, order: SortOrder) -> Self {
272 Self {
273 view_id,
274 sort_field,
275 order,
276 sorted: BTreeMap::new(),
277 entities: HashMap::new(),
278 keys_cache: Vec::new(),
279 cache_dirty: true,
280 placeholder: SharedEntity::new(Value::Null),
281 partitions: None,
282 }
283 }
284
285 pub fn view_id(&self) -> &str {
286 &self.view_id
287 }
288
289 pub fn len(&self) -> usize {
290 match &self.partitions {
291 Some(partitions) => partitions.len(),
292 None => self.entities.len(),
293 }
294 }
295
296 pub fn is_empty(&self) -> bool {
297 self.len() == 0
298 }
299
300 pub fn partition_by_scope(&mut self, field: ScopeField, max_partitions: usize) {
323 let rows = match self.partitions.take() {
324 Some(partitions) => partitions.into_rows(),
325 None => {
326 let rows = self.ordered_entities();
327 self.sorted.clear();
328 self.entities.clear();
329 self.keys_cache.clear();
330 self.cache_dirty = true;
331 rows
332 }
333 };
334 self.partitions = Some(Box::new(ScopePartitions {
335 field,
336 max_partitions: max_partitions.max(1),
337 view_id: self.view_id.clone(),
338 sort_field: self.sort_field.clone(),
339 order: self.order,
340 parts: HashMap::new(),
341 key_partition: HashMap::new(),
342 clock: 0,
343 evictions: 0,
344 }));
345 for (key, entity) in rows {
346 self.put(key, entity);
347 }
348 }
349
350 pub fn is_scope_partitioned(&self) -> bool {
352 self.partitions.is_some()
353 }
354
355 pub fn scope_partition_count(&self) -> usize {
357 self.partitions
358 .as_ref()
359 .map_or(0, |partitions| partitions.parts.len())
360 }
361
362 pub fn take_scope_partition_evictions(&mut self) -> u64 {
364 self.partitions
365 .as_mut()
366 .map_or(0, |partitions| std::mem::take(&mut partitions.evictions))
367 }
368
369 pub fn iter_ordered_in_scopes<'a>(
373 &'a self,
374 scopes: &'a BTreeSet<String>,
375 ) -> impl Iterator<Item = (&'a str, &'a SharedEntity)> + 'a {
376 let iter: Box<dyn Iterator<Item = (&'a str, &'a SharedEntity)> + 'a> =
377 match &self.partitions {
378 Some(partitions) => Box::new(MergedOrder::new(
379 std::iter::once("")
380 .chain(
381 scopes
382 .iter()
383 .map(String::as_str)
384 .filter(|scope| !scope.is_empty()),
385 )
386 .filter_map(|scope| partitions.parts.get(scope))
387 .map(|partition| partition.cache.iter_ranked()),
388 )),
389 None => Box::new(self.iter_ordered()),
390 };
391 iter
392 }
393
394 pub fn upsert(&mut self, entity_key: String, entity: impl Into<SharedEntity>) -> UpsertResult {
403 let entity = entity.into();
404 if let Some(partitions) = self.partitions.as_deref_mut() {
405 return partitions.upsert(entity_key, entity);
406 }
407 let sort_value = self.extract_sort_value(&entity);
408 let moved = self.put_with_sort_value(Cow::Borrowed(&entity_key), entity, sort_value);
409 let position = self.find_position(&entity_key);
410 match moved {
411 Placed::Unmoved => UpsertResult::Updated { position },
412 Placed::Moved => UpsertResult::Inserted { position },
413 }
414 }
415
416 pub fn put(&mut self, entity_key: String, entity: impl Into<SharedEntity>) {
418 let entity = entity.into();
419 if let Some(partitions) = self.partitions.as_deref_mut() {
420 partitions.put(entity_key, entity);
421 return;
422 }
423 let sort_value = self.extract_sort_value(&entity);
424 self.put_with_sort_value(Cow::Owned(entity_key), entity, sort_value);
425 }
426
427 fn put_with_sort_value(
429 &mut self,
430 entity_key: Cow<'_, str>,
431 entity: SharedEntity,
432 sort_value: SortValue,
433 ) -> Placed {
434 if let Some((sort_key, held)) = self.entities.get_mut(entity_key.as_ref()) {
435 let old_entity = std::mem::replace(held, self.placeholder.clone());
437 *held = Self::merge_entity(old_entity, entity);
438
439 let keeps_old_value = matches!(sort_value, SortValue::Null)
441 && !matches!(sort_key.sort_value, SortValue::Null);
442 if keeps_old_value || sort_key.sort_value == sort_value {
443 return Placed::Unmoved;
445 }
446
447 let mut ordered = match self.sorted.remove_entry(sort_key) {
450 Some((ordered, ())) => ordered,
451 None => {
454 let stale = sort_key.entity_key.as_str();
455 self.sorted.retain(|held, ()| held.entity_key != stale);
456 sort_key.clone()
457 }
458 };
459 ordered.set_sort_value(sort_value);
460 sort_key.sort_value = ordered.sort_value.clone();
461 sort_key.decimal = ordered.decimal;
462 self.sorted.insert(ordered, ());
463 self.cache_dirty = true;
464 return Placed::Moved;
465 }
466
467 let entity_key = entity_key.into_owned();
468 let new_sort_key = SortKey::new(sort_value, entity_key.clone(), self.order);
469 self.sorted.insert(new_sort_key.clone(), ());
470 self.entities.insert(entity_key, (new_sort_key, entity));
471 self.cache_dirty = true;
472 Placed::Moved
473 }
474
475 pub fn would_keep<E: EntityFields + ?Sized>(
482 &self,
483 entity_key: &str,
484 entity: &E,
485 max_entries: usize,
486 ) -> bool {
487 if let Some(partitions) = &self.partitions {
488 return partitions.would_keep(entity_key, entity, max_entries);
489 }
490 self.keeps(entity_key, || self.extract_sort_value(entity), max_entries)
491 .0
492 }
493
494 fn keeps(
497 &self,
498 entity_key: &str,
499 sort_value: impl FnOnce() -> SortValue,
500 max_entries: usize,
501 ) -> (bool, Option<SortValue>) {
502 if self.entities.contains_key(entity_key) || self.sorted.len() < max_entries {
503 return (true, None);
504 }
505 let Some((last, ())) = self.sorted.last_key_value() else {
506 return (true, None);
507 };
508 let sort_value = sort_value();
509 let candidate = Ranked {
510 sort_value: &sort_value,
511 decimal: DecimalKey::of(&sort_value),
512 entity_key,
513 order: self.order,
514 };
515 let kept = candidate.cmp(&last.ranked()).is_lt();
516 (kept, Some(sort_value))
517 }
518
519 pub fn upsert_bounded(
525 &mut self,
526 entity_key: String,
527 entity: impl Into<SharedEntity>,
528 max_entries: usize,
529 ) -> UpsertResult {
530 let result = self.upsert(entity_key, entity);
531 self.trim_to_max_entries(max_entries);
532 result
533 }
534
535 pub fn upsert_if_kept(
541 &mut self,
542 entity_key: &str,
543 entity: &SharedEntity,
544 max_entries: usize,
545 ) -> bool {
546 if let Some(partitions) = self.partitions.as_deref_mut() {
547 return partitions.upsert_if_kept(entity_key, entity, max_entries);
548 }
549 let (kept, sort_value) =
550 self.keeps(entity_key, || self.extract_sort_value(entity), max_entries);
551 if !kept {
552 return false;
553 }
554 let sort_value = sort_value.unwrap_or_else(|| self.extract_sort_value(entity));
555 self.put_with_sort_value(Cow::Borrowed(entity_key), entity.clone(), sort_value);
556 self.trim_to_max_entries(max_entries);
557 true
558 }
559
560 pub fn trim_to_max_entries(&mut self, max_entries: usize) -> usize {
568 if let Some(partitions) = self.partitions.as_deref_mut() {
569 return partitions.trim(max_entries);
570 }
571 self.trim_evicting(max_entries, |_| {})
572 }
573
574 fn trim_evicting(&mut self, max_entries: usize, mut on_evicted: impl FnMut(String)) -> usize {
577 let mut evicted = 0;
578 while self.sorted.len() > max_entries {
579 let Some((sort_key, ())) = self.sorted.pop_last() else {
580 break;
581 };
582 self.entities.remove(&sort_key.entity_key);
583 on_evicted(sort_key.entity_key);
584 evicted += 1;
585 }
586 if evicted > 0 && !self.cache_dirty {
587 self.keys_cache.truncate(self.sorted.len());
590 }
591 evicted
592 }
593
594 fn merge_entity(base: SharedEntity, patch: SharedEntity) -> SharedEntity {
598 let (base_fields, base_version) = base.into_parts();
599 let (patch_fields, patch_version) = patch.into_parts();
600 let (Fields::Object(base_members), Fields::Object(patch_members)) =
601 (&*base_fields, &*patch_fields)
602 else {
603 return SharedEntity::from_parts(patch_fields, patch_version);
604 };
605 let keeps = !Arc::ptr_eq(&base_fields, &patch_fields)
608 && base_members.iter().any(|(key, base)| {
609 patch_members
610 .get(key)
611 .is_none_or(|patch| !Arc::ptr_eq(base, patch) && keeps_fields(base, patch))
612 });
613 let fields = if keeps {
614 let base = match Arc::try_unwrap(base_fields) {
615 Ok(fields) => fields.into_value(),
616 Err(shared) => shared.to_value(),
617 };
618 Arc::new(Fields::from_value(Self::deep_merge(
619 base,
620 patch_fields.to_value(),
621 )))
622 } else {
623 patch_fields
624 };
625 let version = match (base_version, patch_version) {
626 (Some(base), Some(patch)) => Some(Self::deep_merge(base, patch)),
627 (base, None) => base,
628 (None, patch) => patch,
629 };
630 SharedEntity::from_parts(fields, version)
631 }
632
633 fn deep_merge(base: Value, patch: Value) -> Value {
634 match (base, patch) {
635 (Value::Object(mut base_map), Value::Object(patch_map)) => {
636 for (key, patch_value) in patch_map {
637 if let Some(base_value) = base_map.remove(&key) {
638 base_map.insert(key, Self::deep_merge(base_value, patch_value));
639 } else {
640 base_map.insert(key, patch_value);
641 }
642 }
643 Value::Object(base_map)
644 }
645 (_, patch) => patch,
646 }
647 }
648
649 pub fn clear(&mut self) {
651 if let Some(partitions) = self.partitions.as_deref_mut() {
652 partitions.parts.clear();
653 partitions.key_partition.clear();
654 return;
655 }
656 self.sorted.clear();
657 self.entities.clear();
658 self.cache_dirty = true;
659 }
660
661 pub fn remove(&mut self, entity_key: &str) -> Option<usize> {
665 if let Some(partitions) = self.partitions.as_deref_mut() {
666 return partitions.remove(entity_key);
667 }
668 let (sort_key, _) = self.entities.get(entity_key)?;
669 let position = self.find_position_by_sort_key(sort_key);
670 self.remove_key(entity_key);
671 Some(position)
672 }
673
674 pub fn remove_key(&mut self, entity_key: &str) -> bool {
676 if let Some(partitions) = self.partitions.as_deref_mut() {
677 return partitions.remove_key(entity_key);
678 }
679 let Some((sort_key, _)) = self.entities.remove(entity_key) else {
680 return false;
681 };
682 self.sorted.remove(&sort_key);
683 self.cache_dirty = true;
684 true
685 }
686
687 pub fn get(&self, entity_key: &str) -> Option<&SharedEntity> {
689 if let Some(partitions) = &self.partitions {
690 return partitions.get(entity_key);
691 }
692 self.entities.get(entity_key).map(|(_, v)| v)
693 }
694
695 pub fn ordered_keys(&mut self) -> &[String] {
700 if self.partitions.is_some() {
701 self.keys_cache = self
702 .iter_ordered()
703 .map(|(key, _)| key.to_string())
704 .collect();
705 return &self.keys_cache;
706 }
707 if self.cache_dirty {
708 self.rebuild_keys_cache();
709 }
710 &self.keys_cache
711 }
712
713 pub fn get_window(&mut self, skip: usize, take: usize) -> Vec<(String, Value)> {
715 self.iter_ordered()
716 .skip(skip)
717 .take(take)
718 .map(|(key, entity)| (key.to_string(), entity.to_value()))
719 .collect()
720 }
721
722 pub fn iter_ordered(&self) -> impl Iterator<Item = (&str, &SharedEntity)> + '_ {
725 let iter: Box<dyn Iterator<Item = (&str, &SharedEntity)> + '_> = match &self.partitions {
726 Some(partitions) => Box::new(MergedOrder::new(
727 partitions
728 .parts
729 .values()
730 .map(|partition| partition.cache.iter_ranked()),
731 )),
732 None => Box::new(
733 self.iter_ranked()
734 .map(|(sort_key, entity)| (sort_key.entity_key.as_str(), entity)),
735 ),
736 };
737 iter
738 }
739
740 fn iter_ranked(&self) -> RankedRows<'_> {
742 Box::new(self.sorted.keys().filter_map(|sort_key| {
743 self.entities
744 .get(&sort_key.entity_key)
745 .map(|(_, entity)| (sort_key, entity))
746 }))
747 }
748
749 pub fn get_all_ordered(&mut self) -> Vec<(String, Value)> {
751 self.ordered_entities()
752 .into_iter()
753 .map(|(key, entity)| (key, entity.into_value()))
754 .collect()
755 }
756
757 pub fn ordered_entities(&mut self) -> Vec<(String, SharedEntity)> {
760 self.iter_ordered()
761 .map(|(key, entity)| (key.to_string(), entity.clone()))
762 .collect()
763 }
764
765 pub fn compute_window_deltas(
767 &mut self,
768 old_window_keys: &[String],
769 skip: usize,
770 take: usize,
771 ) -> Vec<ViewDelta> {
772 let new_window_keys: Vec<&str> = self
773 .iter_ordered()
774 .map(|(key, _)| key)
775 .skip(skip)
776 .take(take)
777 .collect();
778
779 let old_set: std::collections::HashSet<&str> =
780 old_window_keys.iter().map(String::as_str).collect();
781 let new_set: std::collections::HashSet<&str> = new_window_keys.iter().copied().collect();
782
783 let mut deltas = Vec::new();
784
785 for key in old_set.difference(&new_set) {
787 deltas.push(ViewDelta::Remove {
788 key: (*key).to_string(),
789 });
790 }
791
792 for key in new_set.difference(&old_set) {
794 if let Some(entity) = self.get(key) {
795 deltas.push(ViewDelta::Add {
796 key: (*key).to_string(),
797 entity: entity.to_value(),
798 });
799 }
800 }
801
802 deltas
803 }
804
805 fn extract_sort_value<E: EntityFields + ?Sized>(&self, entity: &E) -> SortValue {
806 match lookup(entity, &self.sort_field) {
807 Some(value) => value_to_sort_value(&value),
808 None => SortValue::Null,
809 }
810 }
811
812 fn find_position(&self, entity_key: &str) -> usize {
813 if let Some((sort_key, _)) = self.entities.get(entity_key) {
814 self.find_position_by_sort_key(sort_key)
815 } else {
816 0
817 }
818 }
819
820 fn find_position_by_sort_key(&self, sort_key: &SortKey) -> usize {
821 self.sorted.range(..sort_key).count()
822 }
823
824 fn rebuild_keys_cache(&mut self) {
825 self.keys_cache = self.sorted.keys().map(|sk| sk.entity_key.clone()).collect();
826 self.cache_dirty = false;
827 }
828}
829
830type RankedRows<'a> = Box<dyn Iterator<Item = (&'a SortKey, &'a SharedEntity)> + 'a>;
832
833#[derive(Debug)]
836struct ScopePartitions {
837 field: ScopeField,
838 max_partitions: usize,
839 view_id: String,
840 sort_field: Vec<String>,
841 order: SortOrder,
842 parts: HashMap<String, Partition>,
844 key_partition: HashMap<String, String>,
846 clock: u64,
848 evictions: u64,
850}
851
852#[derive(Debug)]
853struct Partition {
854 cache: SortedViewCache,
855 last_write: u64,
856}
857
858impl ScopePartitions {
859 fn len(&self) -> usize {
860 self.parts
861 .values()
862 .map(|partition| partition.cache.len())
863 .sum()
864 }
865
866 fn into_rows(self) -> Vec<(String, SharedEntity)> {
867 let mut rows = Vec::new();
868 for (_, mut partition) in self.parts {
869 rows.extend(partition.cache.ordered_entities());
870 }
871 rows
872 }
873
874 fn get(&self, entity_key: &str) -> Option<&SharedEntity> {
875 let scope = self.key_partition.get(entity_key)?;
876 self.parts.get(scope)?.cache.get(entity_key)
877 }
878
879 fn route<E: EntityFields + ?Sized>(&mut self, entity_key: &str, entity: &E) -> Option<String> {
883 let scope = self.field.scope_of(entity);
884 let Some(partition) = scope.partition() else {
885 self.remove(entity_key);
886 return None;
887 };
888 if self
889 .key_partition
890 .get(entity_key)
891 .is_some_and(|held| held != partition)
892 {
893 self.remove(entity_key);
894 }
895 self.clock += 1;
896 if !self.parts.contains_key(partition) {
897 if !partition.is_empty() {
898 self.make_room();
899 }
900 self.parts.insert(
901 partition.to_string(),
902 Partition {
903 cache: SortedViewCache::new(
904 self.view_id.clone(),
905 self.sort_field.clone(),
906 self.order,
907 ),
908 last_write: self.clock,
909 },
910 );
911 }
912 if let Some(held) = self.parts.get_mut(partition) {
913 held.last_write = self.clock;
914 }
915 Some(partition.to_string())
916 }
917
918 fn make_room(&mut self) {
921 while self.parts.keys().filter(|scope| !scope.is_empty()).count() >= self.max_partitions {
922 let Some(oldest) = self
923 .parts
924 .iter()
925 .filter(|(scope, _)| !scope.is_empty())
926 .min_by_key(|(_, partition)| partition.last_write)
927 .map(|(scope, _)| scope.clone())
928 else {
929 return;
930 };
931 if let Some(partition) = self.parts.remove(&oldest) {
932 for (key, _) in partition.cache.iter_ordered() {
933 self.key_partition.remove(key);
934 }
935 }
936 self.evictions += 1;
937 tracing::debug!(
938 view_id = %self.view_id,
939 max_partitions = self.max_partitions,
940 "Evicted a scope partition at the cap"
941 );
942 }
943 }
944
945 fn settle(&mut self, entity_key: &str, partition: &str) {
948 match self.parts.get(partition) {
949 Some(held) if held.cache.get(entity_key).is_some() => {
950 self.key_partition
951 .insert(entity_key.to_string(), partition.to_string());
952 }
953 Some(held) if held.cache.is_empty() => {
954 self.parts.remove(partition);
955 }
956 _ => {}
957 }
958 }
959
960 fn upsert(&mut self, entity_key: String, entity: SharedEntity) -> UpsertResult {
961 let Some(partition) = self.route(&entity_key, &entity) else {
962 return UpsertResult::Inserted {
963 position: usize::MAX,
964 };
965 };
966 let result = match self.parts.get_mut(&partition) {
967 Some(held) => held.cache.upsert(entity_key.clone(), entity),
968 None => {
969 return UpsertResult::Inserted {
970 position: usize::MAX,
971 }
972 }
973 };
974 self.settle(&entity_key, &partition);
975 result
976 }
977
978 fn put(&mut self, entity_key: String, entity: SharedEntity) {
979 let Some(partition) = self.route(&entity_key, &entity) else {
980 return;
981 };
982 if let Some(held) = self.parts.get_mut(&partition) {
983 held.cache.put(entity_key.clone(), entity);
984 }
985 self.settle(&entity_key, &partition);
986 }
987
988 fn would_keep<E: EntityFields + ?Sized>(
989 &self,
990 entity_key: &str,
991 entity: &E,
992 max_entries: usize,
993 ) -> bool {
994 match self.field.scope_of(entity).partition() {
995 None => false,
996 Some(partition) => self
997 .parts
998 .get(partition)
999 .is_none_or(|held| held.cache.would_keep(entity_key, entity, max_entries)),
1000 }
1001 }
1002
1003 fn upsert_if_kept(
1004 &mut self,
1005 entity_key: &str,
1006 entity: &SharedEntity,
1007 max_entries: usize,
1008 ) -> bool {
1009 if !self.would_keep(entity_key, entity, max_entries) {
1010 if self
1012 .key_partition
1013 .get(entity_key)
1014 .is_some_and(|held| Some(held.as_str()) != self.field.scope_of(entity).partition())
1015 {
1016 self.remove(entity_key);
1017 }
1018 return false;
1019 }
1020 let Some(partition) = self.route(entity_key, entity) else {
1021 return false;
1022 };
1023 let mut evicted = Vec::new();
1024 if let Some(held) = self.parts.get_mut(&partition) {
1025 held.cache.put(entity_key.to_string(), entity.clone());
1026 held.cache
1027 .trim_evicting(max_entries, |key| evicted.push(key));
1028 }
1029 for key in evicted {
1030 self.key_partition.remove(&key);
1031 }
1032 self.settle(entity_key, &partition);
1033 self.key_partition.contains_key(entity_key)
1034 }
1035
1036 fn trim(&mut self, max_entries: usize) -> usize {
1037 let mut evicted = Vec::new();
1038 for partition in self.parts.values_mut() {
1039 partition
1040 .cache
1041 .trim_evicting(max_entries, |key| evicted.push(key));
1042 }
1043 for key in &evicted {
1044 self.key_partition.remove(key);
1045 }
1046 evicted.len()
1047 }
1048
1049 fn remove_key(&mut self, entity_key: &str) -> bool {
1050 let Some(partition) = self.key_partition.remove(entity_key) else {
1051 return false;
1052 };
1053 let Some(held) = self.parts.get_mut(&partition) else {
1054 return false;
1055 };
1056 let removed = held.cache.remove_key(entity_key);
1057 if held.cache.is_empty() {
1058 self.parts.remove(&partition);
1059 }
1060 removed
1061 }
1062
1063 fn remove(&mut self, entity_key: &str) -> Option<usize> {
1064 let partition = self.key_partition.remove(entity_key)?;
1065 let held = self.parts.get_mut(&partition)?;
1066 let position = held.cache.remove(entity_key);
1067 if held.cache.is_empty() {
1068 self.parts.remove(&partition);
1069 }
1070 position
1071 }
1072}
1073
1074struct MergedOrder<'a> {
1078 sources: Vec<RankedRows<'a>>,
1079 heads: BinaryHeap<Head<'a>>,
1080}
1081
1082struct Head<'a> {
1083 sort_key: &'a SortKey,
1084 entity: &'a SharedEntity,
1085 source: usize,
1086}
1087
1088impl PartialEq for Head<'_> {
1089 fn eq(&self, other: &Self) -> bool {
1090 self.sort_key == other.sort_key
1091 }
1092}
1093
1094impl Eq for Head<'_> {}
1095
1096impl PartialOrd for Head<'_> {
1097 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
1098 Some(self.cmp(other))
1099 }
1100}
1101
1102impl Ord for Head<'_> {
1103 fn cmp(&self, other: &Self) -> Ordering {
1105 other.sort_key.cmp(self.sort_key)
1106 }
1107}
1108
1109impl<'a> MergedOrder<'a> {
1110 fn new(sources: impl IntoIterator<Item = RankedRows<'a>>) -> Self {
1111 let mut sources: Vec<RankedRows<'a>> = sources.into_iter().collect();
1112 let mut heads = BinaryHeap::with_capacity(sources.len());
1113 for (source, rows) in sources.iter_mut().enumerate() {
1114 if let Some((sort_key, entity)) = rows.next() {
1115 heads.push(Head {
1116 sort_key,
1117 entity,
1118 source,
1119 });
1120 }
1121 }
1122 Self { sources, heads }
1123 }
1124}
1125
1126impl<'a> Iterator for MergedOrder<'a> {
1127 type Item = (&'a str, &'a SharedEntity);
1128
1129 fn next(&mut self) -> Option<Self::Item> {
1130 let head = self.heads.pop()?;
1131 if let Some((sort_key, entity)) = self.sources[head.source].next() {
1132 self.heads.push(Head {
1133 sort_key,
1134 entity,
1135 source: head.source,
1136 });
1137 }
1138 Some((head.sort_key.entity_key.as_str(), head.entity))
1139 }
1140}
1141
1142enum Placed {
1144 Unmoved,
1145 Moved,
1146}
1147
1148#[derive(Debug, Clone, PartialEq)]
1150pub enum UpsertResult {
1151 Inserted { position: usize },
1153 Updated { position: usize },
1155}
1156
1157fn keeps_fields(base: &Value, patch: &Value) -> bool {
1160 match (base, patch) {
1161 (Value::Object(base), Value::Object(patch)) => base
1162 .iter()
1163 .any(|(key, base)| patch.get(key).is_none_or(|patch| keeps_fields(base, patch))),
1164 _ => false,
1165 }
1166}
1167
1168fn value_to_sort_value(v: &Value) -> SortValue {
1169 match v {
1170 Value::Null => SortValue::Null,
1171 Value::Bool(b) => SortValue::Bool(*b),
1172 Value::Number(n) => {
1173 if let Some(i) = n.as_i64() {
1174 SortValue::Integer(i)
1175 } else if let Some(f) = n.as_f64() {
1176 SortValue::Float(OrderedFloat(f))
1177 } else {
1178 SortValue::Null
1179 }
1180 }
1181 Value::String(s) => SortValue::String(s.clone()),
1182 _ => SortValue::Null,
1183 }
1184}
1185
1186#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1190struct DecimalKey {
1191 negative: bool,
1192 start: usize,
1194 value: Option<u128>,
1196}
1197
1198impl DecimalKey {
1199 fn of(sort_value: &SortValue) -> Option<Self> {
1200 match sort_value {
1201 SortValue::String(text) => Self::parse(text),
1202 _ => None,
1203 }
1204 }
1205
1206 fn parse(text: &str) -> Option<Self> {
1207 let (negative, digits) = match text.strip_prefix('-') {
1208 Some(digits) => (true, digits),
1209 None => (false, text),
1210 };
1211 if digits.is_empty() || !digits.bytes().all(|byte| byte.is_ascii_digit()) {
1212 return None;
1213 }
1214 let significant = digits.trim_start_matches('0');
1215 let significant = if significant.is_empty() {
1216 &digits[digits.len() - 1..]
1217 } else {
1218 significant
1219 };
1220 Some(Self {
1221 negative: negative && significant != "0",
1222 start: text.len() - significant.len(),
1223 value: significant.parse().ok(),
1224 })
1225 }
1226
1227 fn cmp(&self, left: &str, other: &Self, right: &str) -> Ordering {
1230 match (self.negative, other.negative) {
1231 (true, false) => Ordering::Less,
1232 (false, true) => Ordering::Greater,
1233 _ => {
1234 let magnitude = match (self.value, other.value) {
1235 (Some(a), Some(b)) => a.cmp(&b),
1236 _ => {
1237 let (a, b) = (&left[self.start..], &right[other.start..]);
1238 a.len().cmp(&b.len()).then_with(|| a.cmp(b))
1239 }
1240 };
1241 if self.negative {
1242 magnitude.reverse()
1243 } else {
1244 magnitude
1245 }
1246 }
1247 }
1248 }
1249}
1250
1251fn compare_strings(left: &str, right: &str) -> Ordering {
1258 compare_decimal_strings(left, right).unwrap_or_else(|| {
1259 let left_decimal = DecimalKey::parse(left).is_some();
1260 let right_decimal = DecimalKey::parse(right).is_some();
1261 right_decimal
1262 .cmp(&left_decimal)
1263 .then_with(|| left.cmp(right))
1264 })
1265}
1266
1267fn compare_decimal_strings(left: &str, right: &str) -> Option<Ordering> {
1268 fn parts(value: &str) -> Option<(bool, &str)> {
1269 let (negative, digits) = match value.strip_prefix('-') {
1270 Some(digits) => (true, digits),
1271 None => (false, value),
1272 };
1273 if digits.is_empty() || !digits.bytes().all(|byte| byte.is_ascii_digit()) {
1274 return None;
1275 }
1276
1277 let digits = digits.trim_start_matches('0');
1278 let digits = if digits.is_empty() { "0" } else { digits };
1279 Some((negative && digits != "0", digits))
1280 }
1281
1282 let (left_negative, left_digits) = parts(left)?;
1283 let (right_negative, right_digits) = parts(right)?;
1284
1285 match (left_negative, right_negative) {
1286 (true, false) => Some(Ordering::Less),
1287 (false, true) => Some(Ordering::Greater),
1288 _ => {
1289 let magnitude = left_digits
1290 .len()
1291 .cmp(&right_digits.len())
1292 .then_with(|| left_digits.cmp(right_digits));
1293 Some(if left_negative {
1294 magnitude.reverse()
1295 } else {
1296 magnitude
1297 })
1298 }
1299 }
1300}
1301
1302#[cfg(test)]
1303mod tests {
1304 use super::*;
1305 use serde_json::json;
1306
1307 #[test]
1308 fn test_sorted_cache_basic() {
1309 let mut cache = SortedViewCache::new(
1310 "test/latest".to_string(),
1311 vec!["id".to_string()],
1312 SortOrder::Desc,
1313 );
1314
1315 cache.upsert("a".to_string(), json!({"id": 1, "name": "first"}));
1316 cache.upsert("b".to_string(), json!({"id": 3, "name": "third"}));
1317 cache.upsert("c".to_string(), json!({"id": 2, "name": "second"}));
1318
1319 let keys = cache.ordered_keys();
1320 assert_eq!(keys, vec!["b", "c", "a"]);
1322 }
1323
1324 #[test]
1325 fn test_sorted_cache_window() {
1326 let mut cache = SortedViewCache::new(
1327 "test/latest".to_string(),
1328 vec!["id".to_string()],
1329 SortOrder::Desc,
1330 );
1331
1332 for i in 1..=10 {
1333 cache.upsert(format!("e{}", i), json!({"id": i}));
1334 }
1335
1336 let window = cache.get_window(0, 3);
1338 assert_eq!(window.len(), 3);
1339 assert_eq!(window[0].0, "e10");
1340 assert_eq!(window[1].0, "e9");
1341 assert_eq!(window[2].0, "e8");
1342
1343 let window = cache.get_window(3, 3);
1344 assert_eq!(window[0].0, "e7");
1345 }
1346
1347 #[test]
1348 fn all_ordered_preserves_stable_sort_and_tie_breaking() {
1349 let mut cache = SortedViewCache::new(
1350 "test/latest".to_string(),
1351 vec!["score".to_string()],
1352 SortOrder::Desc,
1353 );
1354 cache.upsert("b".to_string(), json!({"score": 10}));
1355 cache.upsert("a".to_string(), json!({"score": 10}));
1356 cache.upsert("c".to_string(), json!({"score": 9}));
1357
1358 let keys: Vec<_> = cache
1359 .get_all_ordered()
1360 .into_iter()
1361 .map(|(key, _)| key)
1362 .collect();
1363 assert_eq!(keys, ["a", "b", "c"]);
1364 }
1365
1366 #[test]
1367 fn test_sorted_cache_update_moves_position() {
1368 let mut cache = SortedViewCache::new(
1369 "test/latest".to_string(),
1370 vec!["score".to_string()],
1371 SortOrder::Desc,
1372 );
1373
1374 cache.upsert("a".to_string(), json!({"score": 10}));
1375 cache.upsert("b".to_string(), json!({"score": 20}));
1376 cache.upsert("c".to_string(), json!({"score": 15}));
1377
1378 assert_eq!(cache.ordered_keys(), vec!["b", "c", "a"]);
1380
1381 cache.upsert("a".to_string(), json!({"score": 25}));
1383
1384 assert_eq!(cache.ordered_keys(), vec!["a", "b", "c"]);
1386 }
1387
1388 #[test]
1389 fn test_sorted_cache_remove() {
1390 let mut cache = SortedViewCache::new(
1391 "test/latest".to_string(),
1392 vec!["id".to_string()],
1393 SortOrder::Asc,
1394 );
1395
1396 cache.upsert("a".to_string(), json!({"id": 1}));
1397 cache.upsert("b".to_string(), json!({"id": 2}));
1398 cache.upsert("c".to_string(), json!({"id": 3}));
1399
1400 assert_eq!(cache.len(), 3);
1401
1402 let pos = cache.remove("b");
1403 assert_eq!(pos, Some(1));
1404 assert_eq!(cache.len(), 2);
1405 assert_eq!(cache.ordered_keys(), vec!["a", "c"]);
1406 }
1407
1408 #[test]
1409 fn test_compute_window_deltas() {
1410 let mut cache = SortedViewCache::new(
1411 "test/latest".to_string(),
1412 vec!["id".to_string()],
1413 SortOrder::Desc,
1414 );
1415
1416 for i in 1..=5 {
1418 cache.upsert(format!("e{}", i), json!({"id": i}));
1419 }
1420
1421 let old_window: Vec<String> = vec!["e5".to_string(), "e4".to_string(), "e3".to_string()];
1422
1423 cache.upsert("e6".to_string(), json!({"id": 6}));
1425
1426 let deltas = cache.compute_window_deltas(&old_window, 0, 3);
1429
1430 assert_eq!(deltas.len(), 2);
1431 assert!(deltas
1433 .iter()
1434 .any(|d| matches!(d, ViewDelta::Remove { key } if key == "e3")));
1435 assert!(deltas
1437 .iter()
1438 .any(|d| matches!(d, ViewDelta::Add { key, .. } if key == "e6")));
1439 }
1440
1441 fn keys(cache: &mut SortedViewCache) -> Vec<String> {
1442 cache.ordered_keys().to_vec()
1443 }
1444
1445 #[test]
1446 fn bounded_upsert_evicts_bottom_of_desc_order() {
1447 let mut cache = SortedViewCache::new(
1448 "test/top".to_string(),
1449 vec!["score".to_string()],
1450 SortOrder::Desc,
1451 );
1452
1453 for i in 1..=10 {
1454 cache.upsert_bounded(format!("e{i}"), json!({"score": i}), 4);
1455 assert!(cache.len() <= 4);
1456 }
1457
1458 assert_eq!(cache.len(), 4);
1459 assert_eq!(keys(&mut cache), ["e10", "e9", "e8", "e7"]);
1460 assert!(cache.get("e1").is_none());
1461 assert!(cache.get("e6").is_none());
1462 }
1463
1464 #[test]
1465 fn would_keep_matches_what_a_bounded_upsert_keeps() {
1466 let mut cache = SortedViewCache::new(
1467 "test/top".to_string(),
1468 vec!["score".to_string()],
1469 SortOrder::Desc,
1470 );
1471 for i in 1..=10 {
1472 let entity = json!({"score": i});
1473 let expected = cache.would_keep(&format!("e{i}"), &entity, 4);
1474 cache.upsert_bounded(format!("e{i}"), entity, 4);
1475 assert!(expected, "an entity above a full cache's tail is kept");
1476 }
1477 for (key, score, kept) in [("e0", 0, false), ("e6", 6, false), ("e11", 11, true)] {
1479 let entity = json!({"score": score});
1480 assert_eq!(cache.would_keep(key, &entity, 4), kept, "{key}");
1481 let mut copy = SortedViewCache::new(
1482 "test/top".to_string(),
1483 vec!["score".to_string()],
1484 SortOrder::Desc,
1485 );
1486 for existing in keys(&mut cache) {
1487 let value = cache.get(&existing).unwrap().clone();
1488 copy.upsert_bounded(existing, value, 4);
1489 }
1490 copy.upsert_bounded(key.to_string(), entity, 4);
1491 assert_eq!(copy.get(key).is_some(), kept, "{key}");
1492 }
1493 assert!(cache.would_keep("e8", &json!({"score": -1}), 4));
1495 }
1496
1497 #[test]
1498 fn bounded_upsert_evicts_bottom_of_asc_order() {
1499 let mut cache = SortedViewCache::new(
1500 "test/bottom".to_string(),
1501 vec!["score".to_string()],
1502 SortOrder::Asc,
1503 );
1504
1505 for i in (1..=10).rev() {
1506 cache.upsert_bounded(format!("e{i}"), json!({"score": i}), 4);
1507 assert!(cache.len() <= 4);
1508 }
1509
1510 assert_eq!(cache.len(), 4);
1511 assert_eq!(keys(&mut cache), ["e1", "e2", "e3", "e4"]);
1512 assert!(cache.get("e10").is_none());
1513 }
1514
1515 #[test]
1516 fn bounded_upsert_does_not_evict_stale_top_entities() {
1517 let mut cache = SortedViewCache::new(
1518 "test/top".to_string(),
1519 vec!["score".to_string()],
1520 SortOrder::Desc,
1521 );
1522
1523 cache.upsert_bounded("leader".to_string(), json!({"score": 1_000}), 3);
1526 for i in 1..=20 {
1527 cache.upsert_bounded(format!("e{i}"), json!({"score": i}), 3);
1528 }
1529
1530 assert_eq!(keys(&mut cache), ["leader", "e20", "e19"]);
1531 }
1532
1533 #[test]
1534 fn windows_within_cap_match_unbounded_cache() {
1535 let mut bounded = SortedViewCache::new(
1536 "test/top".to_string(),
1537 vec!["score".to_string()],
1538 SortOrder::Desc,
1539 );
1540 let mut unbounded = SortedViewCache::new(
1541 "test/top".to_string(),
1542 vec!["score".to_string()],
1543 SortOrder::Desc,
1544 );
1545
1546 for i in 0..200u64 {
1551 let key = format!("e{}", i % 60);
1552 let score = (i / 60) * 1_000 + (i * 37) % 101;
1553 let entity = json!({"score": score, "n": i});
1554 bounded.upsert_bounded(key.clone(), entity.clone(), 25);
1555 unbounded.upsert(key, entity);
1556 }
1557
1558 assert_eq!(bounded.len(), 25);
1559 for (skip, take) in [(0, 25), (0, 10), (5, 20), (24, 1)] {
1560 assert_eq!(
1561 bounded.get_window(skip, take),
1562 unbounded.get_window(skip, take),
1563 "window skip={skip} take={take}"
1564 );
1565 }
1566 }
1567
1568 #[test]
1569 fn evicted_entity_is_missing_after_top_moves_down_until_it_updates() {
1570 let mut cache = SortedViewCache::new(
1571 "test/top".to_string(),
1572 vec!["score".to_string()],
1573 SortOrder::Desc,
1574 );
1575
1576 for i in 1..=4 {
1577 cache.upsert_bounded(format!("e{i}"), json!({"score": i}), 3);
1578 }
1579 assert_eq!(keys(&mut cache), ["e4", "e3", "e2"]);
1580
1581 cache.upsert_bounded("e4".to_string(), json!({"score": 0}), 3);
1584 assert_eq!(keys(&mut cache), ["e3", "e2", "e4"]);
1585
1586 cache.upsert_bounded("e1".to_string(), json!({"score": 1}), 3);
1588 assert_eq!(keys(&mut cache), ["e3", "e2", "e1"]);
1589 }
1590
1591 #[test]
1592 fn evicted_entity_reenters_when_upserted_again() {
1593 let mut cache = SortedViewCache::new(
1594 "test/top".to_string(),
1595 vec!["score".to_string()],
1596 SortOrder::Desc,
1597 );
1598
1599 for i in 1..=5 {
1600 cache.upsert_bounded(format!("e{i}"), json!({"score": i}), 3);
1601 }
1602 assert!(cache.get("e1").is_none());
1603
1604 let result = cache.upsert_bounded("e1".to_string(), json!({"score": 100}), 3);
1605 assert_eq!(result, UpsertResult::Inserted { position: 0 });
1606 assert_eq!(keys(&mut cache), ["e1", "e5", "e4"]);
1607 assert_eq!(cache.len(), 3);
1608 }
1609
1610 #[test]
1611 fn upsert_below_full_cap_is_evicted_immediately() {
1612 let mut cache = SortedViewCache::new(
1613 "test/top".to_string(),
1614 vec!["score".to_string()],
1615 SortOrder::Desc,
1616 );
1617
1618 for i in 10..=12 {
1619 cache.upsert_bounded(format!("e{i}"), json!({"score": i}), 3);
1620 }
1621 let result = cache.upsert_bounded("low".to_string(), json!({"score": 1}), 3);
1622
1623 assert_eq!(result, UpsertResult::Inserted { position: 3 });
1624 assert!(cache.get("low").is_none());
1625 assert_eq!(keys(&mut cache), ["e12", "e11", "e10"]);
1626 }
1627
1628 #[test]
1629 fn trim_keeps_keys_cache_and_entities_consistent() {
1630 let mut cache = SortedViewCache::new(
1631 "test/top".to_string(),
1632 vec!["score".to_string()],
1633 SortOrder::Desc,
1634 );
1635
1636 for i in 1..=10 {
1637 cache.upsert(format!("e{i}"), json!({"score": i}));
1638 }
1639 assert_eq!(cache.ordered_keys().len(), 10);
1641
1642 assert_eq!(cache.trim_to_max_entries(4), 6);
1643 assert_eq!(cache.trim_to_max_entries(4), 0);
1644 assert_eq!(cache.len(), 4);
1645 assert_eq!(keys(&mut cache), ["e10", "e9", "e8", "e7"]);
1646 assert_eq!(cache.get_all_ordered().len(), 4);
1647 assert_eq!(cache.remove("e7"), Some(3));
1648 assert_eq!(keys(&mut cache), ["e10", "e9", "e8"]);
1649 }
1650
1651 #[test]
1652 fn test_nested_sort_field() {
1653 let mut cache = SortedViewCache::new(
1654 "test/latest".to_string(),
1655 vec!["id".to_string(), "round_id".to_string()],
1656 SortOrder::Desc,
1657 );
1658
1659 cache.upsert("a".to_string(), json!({"id": {"round_id": 1}}));
1660 cache.upsert("b".to_string(), json!({"id": {"round_id": 3}}));
1661 cache.upsert("c".to_string(), json!({"id": {"round_id": 2}}));
1662
1663 let keys = cache.ordered_keys();
1664 assert_eq!(keys, vec!["b", "c", "a"]);
1665 }
1666
1667 #[test]
1668 fn test_nested_decimal_string_sort_field() {
1669 let mut cache = SortedViewCache::new(
1670 "test/latest".to_string(),
1671 vec!["id".to_string(), "round_id".to_string()],
1672 SortOrder::Desc,
1673 );
1674
1675 cache.upsert("9".to_string(), json!({"id": {"round_id": "9"}}));
1676 cache.upsert("100".to_string(), json!({"id": {"round_id": "100"}}));
1677 cache.upsert("10".to_string(), json!({"id": {"round_id": "10"}}));
1678
1679 assert_eq!(cache.ordered_keys(), vec!["100", "10", "9"]);
1680 }
1681
1682 #[test]
1685 fn test_mixed_decimal_and_text_sort_values() {
1686 let mut cache = SortedViewCache::new(
1687 "test/latest".to_string(),
1688 vec!["name".to_string()],
1689 SortOrder::Asc,
1690 );
1691
1692 cache.upsert("two".to_string(), json!({"name": "2"}));
1693 cache.upsert("ten".to_string(), json!({"name": "10"}));
1694 cache.upsert("text".to_string(), json!({"name": "1a"}));
1695 assert_eq!(cache.ordered_keys(), vec!["two", "ten", "text"]);
1696
1697 cache.put("ten".to_string(), json!({"name": "11"}));
1698 assert_eq!(cache.ordered_keys(), vec!["two", "ten", "text"]);
1699 cache.put("ten".to_string(), json!({"name": "1"}));
1700 assert_eq!(cache.ordered_keys(), vec!["ten", "two", "text"]);
1701 cache.put("text".to_string(), json!({"name": "0"}));
1702 assert_eq!(cache.ordered_keys(), vec!["text", "ten", "two"]);
1703 assert_eq!(cache.len(), 3);
1704 assert_eq!(cache.sorted.len(), 3);
1705 }
1706
1707 #[test]
1709 fn string_order_is_transitive() {
1710 let texts = [
1711 "", "-", "-1", "-10", "0", "007", "1", "10", "1a", "2", "a", "+5", "1.5",
1712 ];
1713 for a in texts {
1714 for b in texts {
1715 for c in texts {
1716 if compare_strings(a, b) != Ordering::Greater
1717 && compare_strings(b, c) != Ordering::Greater
1718 {
1719 assert_ne!(
1720 compare_strings(a, c),
1721 Ordering::Greater,
1722 "{a:?} <= {b:?} <= {c:?}"
1723 );
1724 }
1725 }
1726 }
1727 }
1728 }
1729
1730 #[test]
1731 fn test_descending_string_sort_field() {
1732 let mut cache = SortedViewCache::new(
1733 "test/latest".to_string(),
1734 vec!["name".to_string()],
1735 SortOrder::Desc,
1736 );
1737
1738 cache.upsert("a".to_string(), json!({"name": "alpha"}));
1739 cache.upsert("c".to_string(), json!({"name": "charlie"}));
1740 cache.upsert("b".to_string(), json!({"name": "bravo"}));
1741
1742 assert_eq!(cache.ordered_keys(), vec!["c", "b", "a"]);
1743 }
1744
1745 #[test]
1746 fn test_update_with_missing_sort_field_preserves_position() {
1747 let mut cache = SortedViewCache::new(
1748 "test/latest".to_string(),
1749 vec!["id".to_string(), "round_id".to_string()],
1750 SortOrder::Desc,
1751 );
1752
1753 cache.upsert(
1754 "100".to_string(),
1755 json!({"id": {"round_id": 100}, "data": "initial"}),
1756 );
1757 cache.upsert(
1758 "200".to_string(),
1759 json!({"id": {"round_id": 200}, "data": "initial"}),
1760 );
1761 cache.upsert(
1762 "300".to_string(),
1763 json!({"id": {"round_id": 300}, "data": "initial"}),
1764 );
1765
1766 assert_eq!(cache.ordered_keys(), vec!["300", "200", "100"]);
1767
1768 cache.upsert("200".to_string(), json!({"data": "updated_without_id"}));
1769
1770 assert_eq!(
1771 cache.ordered_keys(),
1772 vec!["300", "200", "100"],
1773 "Entity 200 should retain its position even when updated without sort field"
1774 );
1775
1776 let entity = cache.get("200").unwrap();
1777 assert_eq!(entity["data"], "updated_without_id");
1778 }
1779
1780 #[test]
1781 fn test_new_entity_with_missing_sort_field_sorts_last() {
1782 for order in [SortOrder::Desc, SortOrder::Asc] {
1783 let mut cache = SortedViewCache::new(
1784 "test/latest".to_string(),
1785 vec!["id".to_string(), "round_id".to_string()],
1786 order,
1787 );
1788
1789 cache.upsert("100".to_string(), json!({"id": {"round_id": 100}}));
1790 cache.upsert("200".to_string(), json!({"id": {"round_id": 200}}));
1791 cache.upsert("new".to_string(), json!({"data": "no_sort_field"}));
1792 cache.upsert("nil".to_string(), json!({"id": {"round_id": null}}));
1793
1794 let ranked = match order {
1795 SortOrder::Desc => ["200", "100"],
1796 SortOrder::Asc => ["100", "200"],
1797 };
1798 assert_eq!(
1799 cache.ordered_keys(),
1800 [ranked[0], ranked[1], "new", "nil"],
1801 "{order:?}: an entity without a sort value never outranks one with it; \
1802 among themselves they order by key"
1803 );
1804 }
1805 }
1806
1807 #[test]
1810 fn a_full_window_does_not_admit_an_entity_without_a_sort_value() {
1811 let mut cache = SortedViewCache::new(
1812 "test/latest".to_string(),
1813 vec!["id".to_string(), "round_id".to_string()],
1814 SortOrder::Desc,
1815 );
1816 for round in [100, 200, 300] {
1817 cache.upsert_bounded(round.to_string(), json!({"id": {"round_id": round}}), 3);
1818 }
1819 let partial = json!({"metrics": {"checkpoint_count": 7}});
1820 assert!(!cache.would_keep("1", &partial, 3));
1821 cache.upsert_bounded("1".to_string(), partial, 3);
1822 assert_eq!(cache.ordered_keys(), ["300", "200", "100"]);
1823 }
1824
1825 #[test]
1828 fn a_whole_update_shares_the_given_entity() {
1829 let mut cache = SortedViewCache::new(
1830 "test/top".to_string(),
1831 vec!["score".to_string()],
1832 SortOrder::Desc,
1833 );
1834 cache.upsert(
1835 "a".to_string(),
1836 json!({"score": 1, "name": "a", "_version": "e:1"}),
1837 );
1838 let update =
1839 SharedEntity::new(json!({"score": 2, "name": "a", "extra": true, "_version": "e:2"}));
1840 cache.upsert("a".to_string(), update.clone());
1841 let held = cache.get("a").unwrap();
1842 assert!(held.shares_fields_with(&update));
1843 assert_eq!(*held, update);
1844 }
1845
1846 #[test]
1849 fn an_update_lacking_held_fields_merges_into_its_own_copy() {
1850 let mut cache = SortedViewCache::new(
1851 "test/top".to_string(),
1852 vec!["score".to_string()],
1853 SortOrder::Desc,
1854 );
1855 cache.upsert(
1856 "a".to_string(),
1857 json!({"score": 1, "nested": {"x": 1, "y": 1}, "_version": "e:1"}),
1858 );
1859 let update = SharedEntity::new(json!({"score": 2, "nested": {"x": 2}}));
1860 cache.upsert("a".to_string(), update.clone());
1861 let held = cache.get("a").unwrap();
1862 assert!(!held.shares_fields_with(&update));
1863 assert_eq!(
1864 held.to_value(),
1865 json!({"score": 2, "nested": {"x": 2, "y": 1}, "_version": "e:1"})
1866 );
1867 assert_eq!(update.to_value(), json!({"score": 2, "nested": {"x": 2}}));
1868 }
1869
1870 #[test]
1873 fn a_version_sort_reads_each_copys_own_version() {
1874 let mut cache = SortedViewCache::new(
1875 "test/versions".to_string(),
1876 vec!["_version".to_string()],
1877 SortOrder::Desc,
1878 );
1879 let first = SharedEntity::new(json!({"id": 1, "_version": "e:1"}));
1880 let (fields, _) = first.clone().into_parts();
1881 cache.upsert("a".to_string(), first);
1882 cache.upsert(
1883 "b".to_string(),
1884 SharedEntity::from_parts(fields, Some(json!("e:2"))),
1885 );
1886 assert_eq!(cache.ordered_keys(), ["b", "a"]);
1887 }
1888
1889 #[test]
1892 fn pre_parsed_decimals_compare_as_the_strings_do() {
1893 let huge = "9".repeat(45);
1894 let huger = format!("1{}", "0".repeat(45));
1895 let u128_max = u128::MAX.to_string();
1896 let past_u128 = (u128::MAX - 1).to_string() + "0";
1897 let texts = [
1898 "0",
1899 "-0",
1900 "000",
1901 "-000",
1902 "7",
1903 "007",
1904 "-7",
1905 "-007",
1906 "10",
1907 "9",
1908 "-10",
1909 "-9",
1910 "18446744073709551616",
1911 "-18446744073709551616",
1912 &u128_max,
1913 &past_u128,
1914 &huge,
1915 &huger,
1916 "-",
1917 "",
1918 "+5",
1919 "5a",
1920 "a",
1921 "abc",
1922 " 1",
1923 "1.5",
1924 "-1.5",
1925 "0x10",
1926 ];
1927 for left in texts {
1928 for right in texts {
1929 let expected = compare_strings(left, right);
1930 for order in [SortOrder::Asc, SortOrder::Desc] {
1931 let a = SortKey::new(SortValue::String(left.to_string()), "k".into(), order);
1932 let b = SortKey::new(SortValue::String(right.to_string()), "k".into(), order);
1933 let expected = match order {
1934 SortOrder::Asc => expected,
1935 SortOrder::Desc => expected.reverse(),
1936 };
1937 assert_eq!(a.cmp(&b), expected, "{left:?} vs {right:?} ({order:?})");
1938 }
1939 }
1940 }
1941 }
1942
1943 #[test]
1947 fn unpositioned_writes_match_positioned_ones() {
1948 struct Rng(u64);
1949 impl Rng {
1950 fn below(&mut self, n: u64) -> u64 {
1951 self.0 = self
1952 .0
1953 .wrapping_mul(6364136223846793005)
1954 .wrapping_add(1442695040888963407);
1955 (self.0 >> 33) % n
1956 }
1957 }
1958 for (field, order) in [
1959 ("amount", SortOrder::Desc),
1960 ("amount", SortOrder::Asc),
1961 ("label", SortOrder::Asc),
1962 ] {
1963 let cache = || SortedViewCache::new("test/top".into(), vec![field.into()], order);
1964 let (mut positioned, mut unpositioned) = (cache(), cache());
1965 let mut rng = Rng(7);
1966 for step in 0..4_000 {
1967 let key = format!("k{}", rng.below(60));
1968 let max_entries = 1 + rng.below(40) as usize;
1969 let value = match rng.below(6) {
1970 0 => Value::Null,
1971 1 => json!(rng.below(1_000)),
1972 2 => json!(format!("{}", rng.below(1_000))),
1973 3 => json!(format!("-{:03}", rng.below(1_000))),
1974 4 => json!(format!("{}{}", "9".repeat(40), rng.below(10))),
1975 _ => json!(format!("x{}", rng.below(100))),
1976 };
1977 let entity = SharedEntity::new(json!({
1978 field: value,
1979 "step": step,
1980 "_version": format!("e:{step}"),
1981 }));
1982 match rng.below(4) {
1983 0 => {
1984 positioned.remove(&key);
1985 unpositioned.remove_key(&key);
1986 }
1987 1 => {
1988 positioned.upsert(key.clone(), entity.clone());
1989 unpositioned.put(key, entity);
1990 }
1991 _ => {
1992 let keeps = positioned.would_keep(&key, &entity, max_entries);
1993 if keeps {
1994 positioned.upsert_bounded(key.clone(), entity.clone(), max_entries);
1995 }
1996 assert_eq!(
1997 unpositioned.upsert_if_kept(&key, &entity, max_entries),
1998 keeps
1999 );
2000 }
2001 }
2002 let expected: Vec<(String, Value)> = positioned
2003 .ordered_keys()
2004 .to_vec()
2005 .into_iter()
2006 .map(|key| {
2007 let entity = positioned.get(&key).unwrap().to_value();
2008 (key, entity)
2009 })
2010 .collect();
2011 assert_eq!(unpositioned.get_all_ordered(), expected, "step {step}");
2012 let skip = rng.below(8) as usize;
2013 assert_eq!(
2014 unpositioned.get_window(skip, 5),
2015 expected
2016 .iter()
2017 .skip(skip)
2018 .take(5)
2019 .cloned()
2020 .collect::<Vec<_>>()
2021 );
2022 assert_eq!(unpositioned.ordered_keys(), positioned.ordered_keys());
2023
2024 let mut reference: Vec<(SortValue, String)> = expected
2027 .iter()
2028 .map(|(key, _)| {
2029 let (sort_key, _) = &unpositioned.entities[key];
2030 (sort_key.sort_value.clone(), key.clone())
2031 })
2032 .collect();
2033 reference.sort_by(|(a, a_key), (b, b_key)| {
2034 let by_value = match (a, b) {
2035 (SortValue::Null, SortValue::Null) => Ordering::Equal,
2036 (SortValue::Null, _) => Ordering::Greater,
2037 (_, SortValue::Null) => Ordering::Less,
2038 _ if order == SortOrder::Desc => a.cmp(b).reverse(),
2039 _ => a.cmp(b),
2040 };
2041 by_value.then_with(|| a_key.cmp(b_key))
2042 });
2043 let reference: Vec<&String> = reference.iter().map(|(_, key)| key).collect();
2044 let held: Vec<&String> = expected.iter().map(|(key, _)| key).collect();
2045 assert_eq!(held, reference, "step {step}");
2046 }
2047 }
2048 }
2049
2050 fn scoped_cache(max_partitions: usize) -> SortedViewCache {
2051 let mut cache =
2052 SortedViewCache::new("Thing/top".into(), vec!["amount".into()], SortOrder::Desc);
2053 cache.partition_by_scope(ScopeField::new("scope"), max_partitions);
2054 cache
2055 }
2056
2057 fn row(scope: &str, amount: i64) -> SharedEntity {
2058 SharedEntity::new(json!({"scope": scope, "amount": amount}))
2059 }
2060
2061 fn scopes(values: &[&str]) -> BTreeSet<String> {
2062 values.iter().map(|value| value.to_string()).collect()
2063 }
2064
2065 fn keys_in<'a>(cache: &'a SortedViewCache, readable: &'a BTreeSet<String>) -> Vec<&'a str> {
2066 cache
2067 .iter_ordered_in_scopes(readable)
2068 .map(|(key, _)| key)
2069 .collect()
2070 }
2071
2072 #[test]
2073 fn each_scope_keeps_its_own_top_rows() {
2074 let mut cache = scoped_cache(16);
2075 for index in 0..10 {
2077 cache.upsert_if_kept(&format!("a{index}"), &row("A", index), 5);
2078 cache.upsert_if_kept(&format!("b{index}"), &row("B", 100 + index), 5);
2079 }
2080 cache.upsert_if_kept("p", &row("", 6), 5);
2081 assert_eq!(cache.len(), 11);
2082 assert_eq!(cache.scope_partition_count(), 3);
2083 assert_eq!(
2084 keys_in(&cache, &scopes(&["A"])),
2085 ["a9", "a8", "a7", "a6", "p", "a5"]
2086 );
2087 assert_eq!(keys_in(&cache, &scopes(&[])), ["p"]);
2088 let all: Vec<&str> = cache.iter_ordered().map(|(key, _)| key).collect();
2090 assert_eq!(all[..2], ["b9", "b8"]);
2091 assert_eq!(all.len(), 11);
2092 }
2093
2094 #[test]
2095 fn a_row_follows_its_scope_between_partitions() {
2096 let mut cache = scoped_cache(16);
2097 cache.upsert_if_kept("k", &row("", 1), 10);
2098 assert_eq!(keys_in(&cache, &scopes(&[])), ["k"]);
2099 cache.upsert_if_kept("k", &row("A", 2), 10);
2100 assert!(keys_in(&cache, &scopes(&[])).is_empty());
2101 assert_eq!(keys_in(&cache, &scopes(&["A"])), ["k"]);
2102 assert_eq!(cache.get("k").unwrap().field("scope"), Some(&json!("A")));
2103 cache.upsert_if_kept(
2105 "k",
2106 &SharedEntity::new(json!({"scope": 7, "amount": 3})),
2107 10,
2108 );
2109 assert!(cache.get("k").is_none());
2110 assert!(cache.is_empty());
2111 assert_eq!(
2112 cache.scope_partition_count(),
2113 0,
2114 "empty partitions are dropped"
2115 );
2116 cache.put("k".into(), row("B", 4));
2117 assert!(cache.remove_key("k"));
2118 assert!(cache.is_empty());
2119 }
2120
2121 #[test]
2122 fn the_scope_written_least_recently_is_evicted_at_the_cap() {
2123 let mut cache = scoped_cache(2);
2124 cache.upsert_if_kept("p", &row("", 1), 10);
2125 cache.upsert_if_kept("a", &row("A", 1), 10);
2126 cache.upsert_if_kept("b", &row("B", 1), 10);
2127 cache.upsert_if_kept("a2", &row("A", 2), 10);
2128 assert_eq!(cache.take_scope_partition_evictions(), 0);
2129 cache.upsert_if_kept("c", &row("C", 1), 10);
2131 assert_eq!(cache.take_scope_partition_evictions(), 1);
2132 assert_eq!(cache.take_scope_partition_evictions(), 0);
2133 assert!(cache.get("b").is_none());
2134 assert_eq!(
2135 keys_in(&cache, &scopes(&["A", "B", "C"])),
2136 ["a2", "a", "c", "p"]
2137 );
2138 assert_eq!(cache.scope_partition_count(), 3);
2139 }
2140
2141 #[test]
2142 fn partitioning_moves_rows_already_held() {
2143 let mut cache =
2144 SortedViewCache::new("Thing/top".into(), vec!["amount".into()], SortOrder::Desc);
2145 cache.put("a".into(), row("A", 1));
2146 cache.put("b".into(), row("B", 2));
2147 cache.partition_by_scope(ScopeField::new("scope"), 8);
2148 assert_eq!(keys_in(&cache, &scopes(&["A"])), ["a"]);
2149 assert_eq!(cache.ordered_keys(), ["b", "a"]);
2150 assert_eq!(cache.trim_to_max_entries(0), 2);
2151 assert!(cache.is_empty());
2152 }
2153}