1use crate::static_btree::entry::{Entry, Offset};
2use crate::static_btree::error::{Error, Result};
3use crate::static_btree::key::Key;
4use crate::static_btree::payload::PayloadEntry;
5#[cfg(feature = "http")]
6use http_range_client::{AsyncBufferedHttpRangeClient, AsyncHttpRangeClient};
7use log::{debug, info};
8use std::cmp::{max, min};
9use std::collections::{HashMap, VecDeque};
10use std::io::{Cursor, Read, Seek, SeekFrom, Write};
11use std::mem::size_of;
12use std::ops::Range;
13
14const PAYLOAD_TAG: Offset = 1u64 << 63;
16const PAYLOAD_MASK: Offset = !PAYLOAD_TAG;
18
19const DEFAULT_MIN_REQ_SIZE: usize = 1024 * 32;
20
21pub type NodeItem<K> = Entry<K>;
25
26impl<K: Key> NodeItem<K> {
28 pub fn new_with_key(key: K) -> NodeItem<K> {
29 NodeItem { key, offset: 0 }
30 }
31
32 pub fn create(offset: u64) -> NodeItem<K> {
33 NodeItem {
34 key: K::default(),
35 offset,
36 }
37 }
38
39 pub fn set_key(&mut self, key: K) {
40 self.key = key;
41 }
42
43 pub fn set_offset(&mut self, offset: u64) {
44 self.offset = offset;
45 }
46
47 pub fn equals(&self, other: &NodeItem<K>) -> bool {
48 self.key == other.key
49 }
50}
51
52fn in_bounds<K: Key>(
58 key: &K,
59 lower: &K,
60 lower_strict: bool,
61 upper: &K,
62 upper_strict: bool,
63) -> bool {
64 let lower_ok = if lower_strict {
65 key > lower
66 } else {
67 key >= lower
68 };
69 let upper_ok = if upper_strict {
70 key < upper
71 } else {
72 key <= upper
73 };
74 lower_ok && upper_ok
75}
76
77fn read_node_vec<K: Key>(node_items: &mut Vec<NodeItem<K>>, mut data: impl Read) -> Result<()> {
79 node_items.clear();
80 for _ in 0..node_items.capacity() {
81 node_items.push(NodeItem::from_reader(&mut data)?);
82 }
83 Ok(())
84}
85
86fn read_node_items<K: Key, R: Read + Seek + ?Sized>(
88 data: &mut R,
89 base: u64,
90 node_index: usize,
91 length: usize,
92) -> Result<Vec<NodeItem<K>>> {
93 let mut node_items = Vec::with_capacity(length);
94 data.seek(SeekFrom::Start(
95 base + (node_index * NodeItem::<K>::SERIALIZED_SIZE) as u64,
96 ))?;
97 read_node_vec(&mut node_items, data)?;
98 Ok(node_items)
99}
100
101#[cfg(feature = "http")]
103async fn read_http_node_items<K: Key, T: AsyncHttpRangeClient>(
104 client: &mut AsyncBufferedHttpRangeClient<T>,
105 base: usize,
106 node_ids: &Range<usize>,
107) -> Result<Vec<NodeItem<K>>> {
108 info!("sending request to fetch node items, base: {base}, node_ids: {node_ids:?}");
109
110 let begin = base + node_ids.start * NodeItem::<K>::SERIALIZED_SIZE;
111 let length = node_ids.len() * NodeItem::<K>::SERIALIZED_SIZE;
112 let bytes = client
113 .min_req_size(1024 * 1024)
115 .get_range(begin, length)
116 .await?;
117
118 let mut node_items = Vec::with_capacity(node_ids.len());
119 debug_assert_eq!(bytes.len(), length);
120 for node_item_bytes in bytes.chunks(NodeItem::<K>::SERIALIZED_SIZE) {
121 let node_item = NodeItem::from_reader(&mut Cursor::new(node_item_bytes))?;
122 node_items.push(node_item);
123 }
124 Ok(node_items)
125}
126
127#[cfg(feature = "http")]
128#[allow(dead_code)]
129async fn read_http_payload_data<T: AsyncHttpRangeClient>(
130 client: &mut AsyncBufferedHttpRangeClient<T>,
131 offset: usize,
132) -> Result<PayloadEntry> {
133 let temp_buffered_count_bytes_size = DEFAULT_MIN_REQ_SIZE; debug!("sending request to fetch payload, offset {offset:?}");
136
137 let payload_data = client
138 .get_range(offset, temp_buffered_count_bytes_size)
139 .await?;
140 let mut buf = Cursor::new(payload_data);
141
142 let (payload_entry, _) = PayloadEntry::deserialize(&mut buf)?;
143 Ok(payload_entry)
144}
145
146#[derive(Debug, Default)]
148pub struct PayloadCache {
149 data: Vec<u8>,
151 start_offset: usize,
153 end_offset: usize,
155}
156
157impl PayloadCache {
158 pub fn new() -> Self {
160 Self {
161 data: Vec::new(),
162 start_offset: 0,
163 end_offset: 0,
164 }
165 }
166
167 pub fn contains(&self, offset: usize) -> bool {
169 !self.data.is_empty() && offset >= self.start_offset && offset < self.end_offset
170 }
171
172 pub fn get_entry(&self, offset: usize) -> Result<PayloadEntry> {
174 if !self.contains(offset) {
175 return Err(Error::PayloadOffsetNotInCache);
176 }
177
178 let relative_offset = offset - self.start_offset;
179 let mut cursor = Cursor::new(&self.data[relative_offset..]);
180 let (entry, _) = PayloadEntry::deserialize(&mut cursor)?;
181 Ok(entry)
182 }
183
184 pub fn update(&mut self, start_offset: usize, data: Vec<u8>) {
186 self.data = data;
187 self.start_offset = start_offset;
188 self.end_offset = start_offset + self.data.len();
189 }
190}
191
192#[cfg(feature = "http")]
203pub async fn prefetch_payload<T: AsyncHttpRangeClient>(
204 client: &mut AsyncBufferedHttpRangeClient<T>,
205 payload_section_start: usize,
206 chunk_size: usize,
207) -> Result<PayloadCache> {
208 debug!(
209 "prefetching payload chunk: start={}, size={}",
210 payload_section_start, chunk_size
211 );
212
213 let mut cache = PayloadCache::new();
214
215 let payload_data = client.get_range(payload_section_start, chunk_size).await?;
217
218 cache.update(payload_section_start, payload_data.to_vec());
220
221 Ok(cache)
222}
223
224#[cfg(feature = "http")]
226#[allow(dead_code)]
227async fn read_payload_entry<T: AsyncHttpRangeClient>(
228 client: &mut AsyncBufferedHttpRangeClient<T>,
229 offset: usize,
230 cache: Option<&PayloadCache>,
231) -> Result<PayloadEntry> {
232 if let Some(cache) = cache {
234 if cache.contains(offset) {
235 return cache.get_entry(offset);
236 }
237 }
238
239 read_http_payload_data(client, offset).await
241}
242
243#[derive(Debug)]
245enum PayloadRef {
246 Direct(u64),
248 Indirect(usize),
250}
251
252#[cfg(feature = "http")]
254async fn batch_resolve_payloads<T: AsyncHttpRangeClient>(
255 client: &mut AsyncBufferedHttpRangeClient<T>,
256 payload_refs: Vec<PayloadRef>,
257 payload_section_start: usize,
258 feature_begin: usize,
259 cache: Option<&PayloadCache>,
260) -> Result<Vec<HttpSearchResultItem>> {
261 debug!("batch resolving {} payload references", payload_refs.len());
262
263 if payload_refs.is_empty() {
265 return Ok(Vec::new());
266 }
267
268 let mut results = Vec::new();
270 let mut payload_offsets_to_fetch = Vec::new();
271
272 for payload_ref in payload_refs {
274 match payload_ref {
275 PayloadRef::Direct(offset) => {
276 let start = feature_begin + offset as usize;
278 results.push(HttpSearchResultItem {
279 range: HttpRange::RangeFrom(start..),
280 });
281 }
282 PayloadRef::Indirect(rel_offset) => {
283 let abs_offset = payload_section_start + rel_offset;
284
285 if let Some(cache) = cache {
287 if cache.contains(abs_offset) {
288 match cache.get_entry(abs_offset) {
290 Ok(entry) => {
291 for offset in entry.offsets {
292 let start = feature_begin + offset as usize;
293 results.push(HttpSearchResultItem {
294 range: HttpRange::RangeFrom(start..),
295 });
296 }
297 continue;
298 }
299 Err(_) => {
300 payload_offsets_to_fetch.push(abs_offset);
302 }
303 }
304 } else {
305 payload_offsets_to_fetch.push(abs_offset);
307 }
308 } else {
309 payload_offsets_to_fetch.push(abs_offset);
311 }
312 }
313 }
314 }
315
316 if payload_offsets_to_fetch.is_empty() {
318 return Ok(results);
319 }
320
321 payload_offsets_to_fetch.sort();
323
324 payload_offsets_to_fetch.dedup();
326
327 debug!(
328 "fetching {} unique payload offsets",
329 payload_offsets_to_fetch.len()
330 );
331
332 let mut offset_ranges = Vec::new();
334 let mut current_range = (payload_offsets_to_fetch[0], payload_offsets_to_fetch[0]);
335
336 for &offset in payload_offsets_to_fetch.iter().skip(1) {
337 if offset <= current_range.1 + DEFAULT_MIN_REQ_SIZE {
339 current_range.1 = offset;
340 } else {
341 offset_ranges.push(current_range);
343 current_range = (offset, offset);
344 }
345 }
346 offset_ranges.push(current_range);
347
348 debug!(
349 "grouped into {} payload range requests",
350 offset_ranges.len()
351 );
352
353 let mut fetched_payloads = HashMap::new();
355
356 for (start, end) in offset_ranges {
357 let fetch_size = (end - start) + DEFAULT_MIN_REQ_SIZE;
360
361 let payload_data = client.get_range(start, fetch_size).await?;
362
363 for &offset in payload_offsets_to_fetch
365 .iter()
366 .filter(|&&o| o >= start && o <= end)
367 {
368 let relative_offset = offset - start;
369
370 if relative_offset < payload_data.len() {
372 let mut buf = Cursor::new(&payload_data[relative_offset..]);
373 match PayloadEntry::deserialize(&mut buf) {
374 Ok((entry, _)) => {
375 fetched_payloads.insert(offset, entry);
376 }
377 Err(e) => {
378 debug!("error deserializing payload at offset {}: {:?}", offset, e);
379 }
381 }
382 }
383 }
384 }
385
386 for &offset in &payload_offsets_to_fetch {
388 if let Some(entry) = fetched_payloads.get(&offset) {
389 for offset in &entry.offsets {
390 let start = feature_begin + *offset as usize;
391 results.push(HttpSearchResultItem {
392 range: HttpRange::RangeFrom(start..),
393 });
394 }
395 }
396 }
397
398 Ok(results)
399}
400
401#[derive(Debug)]
402pub struct SearchResultItem {
404 pub offset: usize,
406 pub index: usize,
408}
409
410#[derive(Debug, Clone)]
412pub struct Stree<K: Key> {
413 node_items: Vec<NodeItem<K>>,
414 num_leaf_nodes: usize, branching_factor: u16,
416 level_bounds: Vec<Range<usize>>,
417 payload_data: Vec<u8>,
419 payload_initialized: bool,
421}
422
423impl<K: Key> Stree<K> {
424 pub const DEFAULT_NODE_SIZE: u16 = 16;
425
426 pub const DEFAULT_PAYLOAD_PREFETCH_SIZE: usize = 1024 * 1024;
428
429 pub fn compute_payload_prefetch_size(
443 num_items: usize,
444 estimated_avg_payload_size: Option<usize>,
445 prefetch_factor: Option<f32>,
446 ) -> usize {
447 let avg_size = estimated_avg_payload_size.unwrap_or(64);
449
450 let factor = prefetch_factor.unwrap_or(1.0);
452
453 let estimated_payload_entries = (num_items as f32 * 0.1).ceil() as usize;
457
458 let estimated_payload_size = estimated_payload_entries * avg_size;
460
461 let prefetch_size = (estimated_payload_size as f32 * factor) as usize;
463
464 prefetch_size.clamp(16 * 1024, 4 * 1024 * 1024)
468 }
469
470 fn init(&mut self, branching_factor: u16) -> Result<()> {
472 if branching_factor < 2 {
477 return Err(Error::InvalidFormat(format!(
478 "branching factor must be at least 2, got {branching_factor}"
479 )));
480 }
481 if self.num_leaf_nodes == 0 {
482 return Err(Error::InvalidFormat(
483 "cannot build an attribute index with no entries".to_string(),
484 ));
485 }
486 self.branching_factor = branching_factor.clamp(2u16, 65535u16);
487 self.level_bounds =
488 Stree::<K>::generate_level_bounds(self.num_leaf_nodes, self.branching_factor);
489 let num_nodes = self
490 .level_bounds
491 .first()
492 .expect("Btree has at least one level when node_size >= 2 and num_items > 0")
493 .end;
494 self.node_items = vec![NodeItem::create(0); num_nodes]; Ok(())
496 }
497
498 fn generate_level_bounds(num_items: usize, branching_factor: u16) -> Vec<Range<usize>> {
500 assert!(branching_factor >= 2, "Node size must be at least 2");
501 assert!(num_items > 0, "Cannot create empty tree");
502 assert!(
503 num_items <= usize::MAX - ((num_items / branching_factor as usize) * 2),
504 "Number of items too large"
505 );
506
507 let mut level_num_nodes: Vec<usize> = Vec::new();
509 let mut n = num_items;
510 let mut num_nodes = n;
511 level_num_nodes.push(n);
512 loop {
513 n = n.div_ceil(branching_factor as usize);
514 num_nodes += n;
515 level_num_nodes.push(n);
516 if n < branching_factor as usize {
517 break;
518 }
519 }
520
521 let mut level_offsets: Vec<usize> = Vec::with_capacity(level_num_nodes.len());
523 n = num_nodes;
524 for size in &level_num_nodes {
525 level_offsets.push(n - size);
526 n -= size;
527 }
528 let mut level_bounds = Vec::with_capacity(level_num_nodes.len());
529 for i in 0..level_num_nodes.len() {
530 level_bounds.push(level_offsets[i]..level_offsets[i] + level_num_nodes[i]);
531 }
532 level_bounds
533 }
534
535 fn generate_nodes(&mut self) -> Result<()> {
536 let node_size = self.branching_factor as usize - 1;
537 let mut parent_min_key = HashMap::<usize, K>::new(); for level in 0..self.level_bounds.len() - 1 {
539 let children_level = &self.level_bounds[level];
540 let parent_level = &self.level_bounds[level + 1];
541
542 let mut parent_idx = parent_level.start;
543
544 let mut child_idx = children_level.start;
545
546 while child_idx < children_level.end {
552 if parent_idx >= parent_level.end {
553 break;
554 }
555 let child_idx_diff = child_idx - children_level.start;
556
557 let skip_size =
559 self.branching_factor as usize * (self.branching_factor as usize - 1);
560
561 let is_right_most_child = (node_size * node_size) <= (child_idx_diff % skip_size)
562 && (child_idx_diff % skip_size)
563 < (self.branching_factor as usize * self.branching_factor as usize);
564 let has_next_node = child_idx + node_size < children_level.end;
565
566 if is_right_most_child {
567 child_idx += node_size;
568 continue;
569 } else if !has_next_node {
570 let parent_key = K::max_value();
571 let parent_node = NodeItem::<K>::new(parent_key.clone(), child_idx as u64);
572 self.node_items[parent_idx] = parent_node;
573
574 let own_min = min(
575 self.node_items[child_idx].key.clone(),
576 parent_min_key
577 .get(&child_idx)
578 .unwrap_or(&K::max_value())
579 .clone(),
580 );
581 parent_min_key.insert(parent_idx, own_min);
582 parent_idx += 1;
583 child_idx += node_size;
584 continue;
585 } else {
586 let right_node_idx = child_idx + node_size;
587
588 let is_leaf_node = child_idx >= self.num_nodes() - self.num_leaf_nodes;
589 if is_leaf_node {
590 let parent_key = if right_node_idx < children_level.end {
591 self.node_items[right_node_idx].key.clone()
592 } else {
593 K::max_value()
594 };
595 let parent_node = NodeItem::<K>::new(parent_key.clone(), child_idx as u64);
596 self.node_items[parent_idx] = parent_node;
597 parent_min_key.insert(parent_idx, self.node_items[child_idx].key.clone());
598 parent_idx += 1;
599 child_idx += node_size;
600 continue;
601 }
602
603 let parent_key = if right_node_idx < children_level.end {
604 parent_min_key
605 .get(&(child_idx + node_size))
606 .expect("Parent node's key is the minimum key of the right children node's leaf node")
607 .clone()
608 } else {
609 K::max_value()
610 };
611 let parent_node = NodeItem::<K>::new(parent_key.clone(), child_idx as u64);
612 self.node_items[parent_idx] = parent_node;
613 parent_min_key.insert(
614 parent_idx,
615 parent_min_key
616 .get(&child_idx)
617 .expect("Parent node's key is the minimum key of the right children node's leaf node")
618 .clone(),
619 );
620 parent_idx += 1;
621 child_idx += node_size;
622
623 continue;
624 }
625 }
626 }
627 Ok(())
628 }
629
630 fn read_data(&mut self, data: impl Read) -> Result<()> {
631 read_node_vec(&mut self.node_items, data)?;
632 Ok(())
633 }
634
635 #[cfg(feature = "http")]
636 async fn read_http<T: AsyncHttpRangeClient>(
637 &mut self,
638 client: &mut AsyncBufferedHttpRangeClient<T>,
639 index_begin: usize,
640 ) -> Result<()> {
641 let min_req_size = Stree::<K>::index_size(
642 self.num_leaf_items(),
643 self.branching_factor(),
644 self.payload_size(),
645 ); let mut pos = index_begin;
647 for i in 0..self.num_nodes() {
648 let bytes = client
649 .min_req_size(min_req_size)
650 .get_range(pos, size_of::<NodeItem<K>>())
651 .await?;
652 let n = NodeItem::from_bytes(bytes)?;
653 self.node_items[i] = n;
654 pos += NodeItem::<K>::SERIALIZED_SIZE;
655 }
656 Ok(())
657 }
658
659 fn num_nodes(&self) -> usize {
660 self.node_items.len()
661 }
662
663 pub fn build(nodes: &[NodeItem<K>], branching_factor: u16) -> Result<Stree<K>> {
664 let branching_factor = branching_factor.clamp(2u16, 65535u16);
665 let mut nodes = nodes.to_vec();
667 nodes.sort_by_key(|item| item.key.clone());
668 const TAG_MASK: Offset = 1u64 << 63;
671 let mut payload_data = Vec::new();
672 let mut unique_leaves = Vec::new();
673 let mut i = 0;
674 while i < nodes.len() {
675 let key = nodes[i].key.clone();
676 let mut payload_entry = PayloadEntry::new();
677 payload_entry.add_offset(nodes[i].offset);
678 let mut j = i + 1;
679 while j < nodes.len() && nodes[j].key == key {
680 payload_entry.add_offset(nodes[j].offset);
681 j += 1;
682 }
683 if payload_entry.count == 1 {
684 let mut n = NodeItem::new_with_key(key);
686 n.set_offset(payload_entry.offsets[0]);
687 unique_leaves.push(n);
688 } else {
689 let rel = payload_data.len() as Offset;
691 let buf = payload_entry.serialize();
692 payload_data.extend_from_slice(&buf);
693 let mut n = NodeItem::new_with_key(key);
694 n.set_offset(TAG_MASK | rel);
695 unique_leaves.push(n);
696 }
697 i = j;
698 }
699 let mut tree = Stree::<K> {
701 node_items: Vec::new(),
702 num_leaf_nodes: unique_leaves.len(),
703 branching_factor,
704 level_bounds: Vec::new(),
705 payload_data,
706 payload_initialized: true,
707 };
708 tree.init(branching_factor)?;
709 let num_nodes = tree.num_nodes();
710 for (k, node) in unique_leaves.into_iter().enumerate() {
711 tree.node_items[num_nodes - tree.num_leaf_nodes + k] = node;
712 }
713 tree.generate_nodes()?;
714
715 Ok(tree)
716 }
717
718 pub fn from_buf(
719 mut data: impl Read,
720 num_items: usize,
721 branching_factor: u16,
722 ) -> Result<Stree<K>> {
723 let branching_factor = branching_factor.clamp(2u16, 65535u16);
725 let level_bounds = Stree::<K>::generate_level_bounds(num_items, branching_factor);
726 let num_nodes = level_bounds
727 .first()
728 .expect("Btree has at least one level when node_size >= 2 and num_items > 0")
729 .end;
730 let mut tree = Stree::<K> {
731 node_items: Vec::with_capacity(num_nodes),
732 num_leaf_nodes: num_items,
733 branching_factor,
734 level_bounds,
735 payload_data: Vec::new(),
736 payload_initialized: false,
737 };
738 tree.read_data(&mut data)?;
740 let mut payload = Vec::new();
742 data.read_to_end(&mut payload)?;
743 if !payload.is_empty() {
744 tree.payload_data = payload;
745 tree.payload_initialized = true;
746 }
747 Ok(tree)
748 }
749
750 #[cfg(feature = "http")]
751 pub async fn from_http<T: AsyncHttpRangeClient>(
752 client: &mut AsyncBufferedHttpRangeClient<T>,
753 index_begin: usize,
754 num_items: usize,
755 node_size: u16,
756 ) -> Result<Stree<K>> {
757 let mut tree = Stree::<K> {
758 node_items: Vec::new(),
759 num_leaf_nodes: num_items,
760 branching_factor: 0,
761 level_bounds: Vec::new(),
762 payload_data: Vec::new(),
763 payload_initialized: false,
764 };
765 tree.init(node_size)?;
766 tree.read_http(client, index_begin).await?;
767 Ok(tree)
768 }
769
770 pub fn find_exact(&self, key: K) -> Result<Vec<SearchResultItem>> {
771 let leaf_nodes_offset = self
772 .level_bounds
773 .first()
774 .expect("RTree has at least one level when node_size >= 2 and num_items > 0")
775 .start;
776 let search_entry = NodeItem::new_with_key(key);
777 let mut results = Vec::new();
778 let mut queue = VecDeque::new();
779 let node_size = self.branching_factor as usize - 1;
780
781 queue.push_back((0, self.level_bounds.len() - 1));
782 while let Some(next) = queue.pop_front() {
783 let node_index = next.0;
784 let level = next.1;
785
786 let is_leaf_node = level == 0;
788
789 let end = min(node_index + node_size, self.level_bounds[level].end);
791
792 let node_items = &self.node_items[node_index..end];
793
794 if node_items.is_empty() {
795 continue;
796 }
797
798 if !is_leaf_node {
801 let search_result =
802 node_items.binary_search_by(|item| item.key.cmp(&search_entry.key));
803 match search_result {
804 Ok(index) => {
805 let child = node_items[index].offset as usize + node_size;
814 let child_end = self.level_bounds[level - 1].end;
815 let child = if child >= child_end {
816 node_items[index].offset as usize
817 } else {
818 child
819 };
820 queue.push_back((child, level - 1));
821 }
822 Err(index) => {
823 if index == 0 {
824 queue.push_back((node_items[0].offset as usize, level - 1));
825 } else if index == node_items.len() {
826 queue.push_back((
827 node_items[node_items.len() - 1].offset as usize + node_size,
828 level - 1,
829 ));
830 } else {
831 queue.push_back((node_items[index].offset as usize, level - 1));
832 }
833 }
834 }
835 }
836
837 if is_leaf_node {
838 let result = node_items.binary_search_by(|item| item.key.cmp(&search_entry.key));
839 match result {
840 Ok(idx) => {
841 let off = node_items[idx].offset;
842 let base_index = node_index + idx - leaf_nodes_offset;
843 if self.payload_initialized && (off & PAYLOAD_TAG) != 0 {
845 let rel = (off & PAYLOAD_MASK) as usize;
846 let (entry, _) = PayloadEntry::deserialize(&mut Cursor::new(
847 &self.payload_data[rel..],
848 ))?;
849 for o in entry.offsets {
850 results.push(SearchResultItem {
851 offset: o as usize,
852 index: base_index,
853 });
854 }
855 } else {
856 results.push(SearchResultItem {
857 offset: off as usize,
858 index: base_index,
859 });
860 }
861 }
862 Err(_) => continue,
863 }
864 }
865 }
866 Ok(results)
867 }
868
869 pub fn stream_find_exact<R: Read + Seek + ?Sized>(
870 data: &mut R,
871 num_items: usize, branching_factor: u16,
873 key: K,
874 ) -> Result<Vec<SearchResultItem>> {
875 let search_entry = NodeItem::new_with_key(key);
876 let mut results = Vec::new();
877 let mut queue = VecDeque::new();
878 let node_size = branching_factor as usize - 1;
879 let level_bounds = Stree::<K>::generate_level_bounds(num_items, branching_factor);
880
881 let Range {
882 start: leaf_nodes_offset,
883 end: num_nodes,
884 } = level_bounds
885 .first()
886 .expect("RTree has at least one level when node_size >= 2 and num_items > 0");
887
888 let payload_data_start =
889 data.stream_position()? + (Entry::<K>::SERIALIZED_SIZE as u64) * (*num_nodes as u64);
890
891 let index_base: u64 = data.stream_position()?;
892
893 queue.push_back((0, level_bounds.len() - 1));
894 while let Some(next) = queue.pop_front() {
895 let node_index = next.0;
896 let level = next.1;
897
898 let is_leaf_node = level == 0;
900
901 let end = min(node_index + node_size, level_bounds[level].end);
903
904 let node_items = read_node_items(data, index_base, node_index, end - node_index)?;
905
906 if node_items.is_empty() {
907 continue;
908 }
909
910 if !is_leaf_node {
913 let search_result =
914 node_items.binary_search_by(|item: &Entry<K>| item.key.cmp(&search_entry.key));
915 match search_result {
916 Ok(index) => {
917 let child = node_items[index].offset as usize + node_size;
926 let child_end = level_bounds[level - 1].end;
927 let child = if child >= child_end {
928 node_items[index].offset as usize
929 } else {
930 child
931 };
932 queue.push_back((child, level - 1));
933 }
934 Err(index) => {
935 if index == 0 {
936 queue.push_back((node_items[0].offset as usize, level - 1));
937 } else if index == node_items.len() {
938 queue.push_back((
939 node_items[node_items.len() - 1].offset as usize + node_size,
940 level - 1,
941 ));
942 } else {
943 queue.push_back((node_items[index].offset as usize, level - 1));
944 }
945 }
946 }
947 }
948
949 if is_leaf_node {
950 let result = node_items.binary_search_by(|item| item.key.cmp(&search_entry.key));
951 match result {
952 Ok(idx) => {
953 let off = node_items[idx].offset;
954 let base_index = node_index + idx - leaf_nodes_offset;
955 if (off & PAYLOAD_TAG) != 0 {
957 let rel = (off & PAYLOAD_MASK) as usize;
958 data.seek(SeekFrom::Start(payload_data_start + rel as u64))?;
959 let (entry, _) = PayloadEntry::deserialize(data)?;
960 for o in entry.offsets {
961 results.push(SearchResultItem {
962 offset: o as usize,
963 index: base_index,
964 });
965 }
966 } else {
967 results.push(SearchResultItem {
968 offset: off as usize,
969 index: base_index,
970 });
971 }
972 }
973 Err(_) => continue,
974 }
975 }
976 }
977 Ok(results)
978 }
979
980 pub fn find_range(&self, lower: K, upper: K) -> Result<Vec<SearchResultItem>> {
991 self.find_range_strict(lower, false, upper, false)
992 }
993
994 pub fn find_range_strict(
1007 &self,
1008 lower: K,
1009 lower_strict: bool,
1010 upper: K,
1011 upper_strict: bool,
1012 ) -> Result<Vec<SearchResultItem>> {
1013 let leaf_nodes_offset = self
1014 .level_bounds
1015 .first()
1016 .expect("RTree has at least one level when node_size >= 2 and num_items > 0")
1017 .start;
1018 if lower > upper {
1020 return Ok(Vec::new());
1021 }
1022
1023 if lower == upper {
1024 if lower_strict || upper_strict {
1026 return Ok(Vec::new());
1027 }
1028 return self.find_exact(lower);
1031 }
1032
1033 let node_size = self.branching_factor as usize - 1;
1034 let mut results = Vec::new();
1035
1036 let lower_idx = self.find_partition(lower.clone())?;
1038 let upper_idx = self.find_partition(upper.clone())?;
1039
1040 let leaf_level = 0;
1042 let leaf_start = self.level_bounds[leaf_level].start;
1043 let leaf_end = self.level_bounds[leaf_level].end;
1044
1045 let start_idx = max(lower_idx, leaf_start);
1047 let end_idx = min(upper_idx + 2 * node_size, leaf_end);
1055
1056 let mut current_idx = start_idx;
1058 while current_idx < end_idx {
1059 let node_end = min(current_idx + node_size, end_idx);
1060 let node_items = &self.node_items[current_idx..node_end];
1061
1062 for (_i, item) in node_items.iter().enumerate() {
1064 if in_bounds(&item.key, &lower, lower_strict, &upper, upper_strict) {
1065 let off = item.offset;
1066 let idx = current_idx + _i - leaf_nodes_offset;
1067 if self.payload_initialized && (off & PAYLOAD_TAG) != 0 {
1068 let rel = (off & PAYLOAD_MASK) as usize;
1069 let (entry, _) =
1070 PayloadEntry::deserialize(&mut Cursor::new(&self.payload_data[rel..]))?;
1071 for o in entry.offsets {
1072 results.push(SearchResultItem {
1073 offset: o as usize,
1074 index: idx,
1075 });
1076 }
1077 } else {
1078 results.push(SearchResultItem {
1079 offset: off as usize,
1080 index: idx,
1081 });
1082 }
1083 }
1084 }
1085
1086 current_idx = node_end;
1087 }
1088
1089 Ok(results)
1090 }
1091
1092 pub fn stream_find_range<R: Read + Seek + ?Sized>(
1093 data: &mut R,
1094 num_items: usize, branching_factor: u16,
1096 lower: K,
1097 upper: K,
1098 ) -> Result<Vec<SearchResultItem>> {
1099 Self::stream_find_range_strict(
1100 data,
1101 num_items,
1102 branching_factor,
1103 lower,
1104 false,
1105 upper,
1106 false,
1107 )
1108 }
1109
1110 #[allow(clippy::too_many_arguments)]
1114 pub fn stream_find_range_strict<R: Read + Seek + ?Sized>(
1115 data: &mut R,
1116 num_items: usize, branching_factor: u16,
1118 lower: K,
1119 lower_strict: bool,
1120 upper: K,
1121 upper_strict: bool,
1122 ) -> Result<Vec<SearchResultItem>> {
1123 let node_size = branching_factor as usize - 1;
1124 let level_bounds = Stree::<K>::generate_level_bounds(num_items, branching_factor);
1125
1126 let Range {
1127 start: leaf_nodes_offset,
1128 end: num_nodes,
1129 } = level_bounds
1130 .first()
1131 .expect("RTree has at least one level when node_size >= 2 and num_items > 0");
1132
1133 let payload_data_start =
1134 data.stream_position()? + (Entry::<K>::SERIALIZED_SIZE as u64) * (*num_nodes as u64);
1135
1136 if lower > upper {
1138 return Ok(Vec::new());
1139 }
1140
1141 if lower == upper {
1142 if lower_strict || upper_strict {
1144 return Ok(Vec::new());
1145 }
1146 return Stree::stream_find_exact(data, num_items, branching_factor, lower);
1149 }
1150
1151 let mut results = Vec::new();
1152
1153 let upper_idx =
1155 Stree::stream_find_partition(data, num_items, branching_factor, upper.clone())?;
1156 let lower_idx =
1157 Stree::stream_find_partition(data, num_items, branching_factor, lower.clone())?;
1158
1159 let leaf_level = 0;
1161 let leaf_start = level_bounds[leaf_level].start;
1162 let leaf_end = level_bounds[leaf_level].end;
1163
1164 let start_idx = max(lower_idx, leaf_start);
1166 let end_idx = min(upper_idx + 2 * node_size, leaf_end);
1174
1175 let index_base: u64 = data.stream_position()?;
1176
1177 let mut current_idx = start_idx;
1179 while current_idx < end_idx {
1180 let node_end = min(current_idx + node_size, end_idx);
1181 let node_items: Vec<NodeItem<K>> =
1182 read_node_items(data, index_base, current_idx, node_end - current_idx)?;
1183
1184 for (_i, item) in node_items.iter().enumerate() {
1186 if in_bounds(&item.key, &lower, lower_strict, &upper, upper_strict) {
1187 let off = item.offset;
1188 let idx = current_idx + _i - leaf_nodes_offset;
1189 if (off & PAYLOAD_TAG) != 0 {
1190 let rel = (off & PAYLOAD_MASK) as usize;
1191 data.seek(SeekFrom::Start(payload_data_start + rel as u64))?;
1192 let (entry, _) = PayloadEntry::deserialize(data)?;
1193 for o in entry.offsets {
1194 results.push(SearchResultItem {
1195 offset: o as usize,
1196 index: idx,
1197 });
1198 }
1199 } else {
1200 results.push(SearchResultItem {
1201 offset: off as usize,
1202 index: idx,
1203 });
1204 }
1205 }
1206 }
1207
1208 current_idx = node_end;
1209 }
1210
1211 Ok(results)
1212 }
1213
1214 pub fn find_partition(&self, key: K) -> Result<usize> {
1222 let node_size = self.branching_factor as usize - 1;
1223 let mut node_index = 0;
1224
1225 for level in (1..self.level_bounds.len()).rev() {
1229 let end = min(node_index + node_size, self.level_bounds[level].end);
1230 let node_items = &self.node_items[node_index..end];
1231
1232 if node_items.is_empty() {
1233 continue;
1234 }
1235 match node_items.binary_search_by(|item| item.key.cmp(&key)) {
1237 Ok(index) => {
1238 node_index = node_items[index].offset as usize;
1241 }
1242 Err(index) => {
1243 if index == 0 {
1245 node_index = node_items[0].offset as usize;
1248 } else if index >= node_items.len() {
1249 node_index = node_items[node_items.len() - 1].offset as usize + node_size;
1252 } else {
1253 node_index = node_items[index].offset as usize;
1256 }
1257 }
1258 }
1259 }
1260
1261 Ok(node_index)
1264 }
1265
1266 pub fn stream_find_partition<R: Read + Seek + ?Sized>(
1267 data: &mut R,
1268 num_items: usize, branching_factor: u16,
1270 key: K,
1271 ) -> Result<usize> {
1272 let start_position = data.stream_position()?;
1273 let node_size = branching_factor as usize - 1;
1274 let level_bounds = Stree::<K>::generate_level_bounds(num_items, branching_factor);
1275
1276 let mut node_index = 0;
1277
1278 let index_base = data.stream_position()?;
1279
1280 for level in (1..level_bounds.len()).rev() {
1284 let end = min(node_index + node_size, level_bounds[level].end);
1285 let node_items = read_node_items(data, index_base, node_index, end - node_index)?;
1286
1287 if node_items.is_empty() {
1288 continue;
1289 }
1290 match node_items.binary_search_by(|item: &Entry<K>| item.key.cmp(&key)) {
1292 Ok(index) => {
1293 node_index = node_items[index].offset as usize;
1296 }
1297 Err(index) => {
1298 if index == 0 {
1300 node_index = node_items[0].offset as usize;
1303 } else if index >= node_items.len() {
1304 node_index = node_items[node_items.len() - 1].offset as usize + node_size;
1307 } else {
1308 node_index = node_items[index].offset as usize;
1311 }
1312 }
1313 }
1314 }
1315
1316 data.seek(SeekFrom::Start(start_position))?;
1317
1318 Ok(node_index)
1321 }
1322
1323 #[cfg(feature = "http")]
1324 #[allow(clippy::too_many_arguments)]
1325 pub async fn http_stream_find_exact<T: AsyncHttpRangeClient>(
1326 client: &mut AsyncBufferedHttpRangeClient<T>,
1327 index_begin: usize,
1328 feature_begin: usize,
1329 num_items: usize,
1330 branching_factor: u16,
1331 key: K,
1332 combine_request_threshold: usize,
1333 ) -> Result<Vec<HttpSearchResultItem>> {
1334 debug!("http_stream_find_exact starts: index_begin: {index_begin}, feature_begin: {feature_begin}, num_items: {num_items}, branching_factor: {branching_factor}, key: {key:?}");
1335
1336 if num_items == 0 {
1337 return Ok(vec![]);
1338 }
1339 let search_entry = NodeItem::new_with_key(key.clone());
1340 let node_size = branching_factor as usize - 1;
1341 let level_bounds = Stree::<K>::generate_level_bounds(num_items, branching_factor);
1342
1343 let Range {
1351 start: root_start,
1352 end: root_end,
1353 } = level_bounds
1354 .last()
1355 .expect("RTree has at least one level when node_size >= 2 and num_items > 0");
1356
1357 #[derive(Debug, PartialEq, Eq)]
1358 struct NodeRange {
1359 level: usize,
1360 nodes: Range<usize>,
1361 }
1362
1363 let mut queue = VecDeque::new();
1364 queue.push_back(NodeRange {
1365 nodes: *root_start..*root_end,
1366 level: level_bounds.len() - 1,
1367 });
1368
1369 let mut payload_refs = Vec::new();
1371
1372 let num_all_items = level_bounds
1373 .first()
1374 .expect("Btree has at least one level when node_size >= 2 and num_items > 0")
1375 .end;
1376
1377 let payload_data_start = index_begin + Stree::<K>::tree_size(num_all_items);
1378
1379 let prefetch_size = Self::compute_payload_prefetch_size(num_items, None, None);
1381 debug!("prefetching payload with size: {} bytes", prefetch_size);
1382
1383 let payload_cache = prefetch_payload(client, payload_data_start, prefetch_size).await?;
1385
1386 while let Some(node_range) = queue.pop_front() {
1387 debug!("next: {node_range:?}. {} items left in queue", queue.len());
1388 let is_leaf = node_range.level == 0;
1389 let node_items = read_http_node_items(client, index_begin, &node_range.nodes).await?;
1390 if node_items.is_empty() {
1391 continue;
1392 }
1393
1394 if is_leaf {
1395 let result = node_items
1396 .binary_search_by(|item: &NodeItem<K>| item.key.cmp(&search_entry.key));
1397 match result {
1398 Ok(idx) => {
1399 let off: u64 = node_items[idx].offset;
1400 if (off & PAYLOAD_TAG) != 0 {
1403 let rel = (off & PAYLOAD_MASK) as usize;
1404 payload_refs.push(PayloadRef::Indirect(rel));
1406 } else {
1407 payload_refs.push(PayloadRef::Direct(off));
1409 }
1410 }
1411 Err(_) => continue,
1412 }
1413 } else {
1414 let result = node_items
1415 .binary_search_by(|item: &NodeItem<K>| item.key.cmp(&search_entry.key));
1416 let mut _offset = 0;
1417 match result {
1418 Ok(idx) => {
1419 _offset = node_items[idx].offset as usize + node_size;
1420 }
1421 Err(idx) => {
1422 if idx == 0 {
1423 _offset = node_items[0].offset as usize;
1424 } else if idx == node_items.len() {
1425 _offset = node_items[node_items.len() - 1].offset as usize + node_size;
1426 } else {
1427 _offset = node_items[idx].offset as usize;
1428 }
1429 }
1430 }
1431 let children_level = node_range.level - 1;
1432 let mut children_nodes = _offset..(_offset + node_size);
1433 if children_level == 0 {
1434 children_nodes.end += 1;
1441 }
1442 children_nodes.end = min(children_nodes.end, level_bounds[children_level].end);
1443
1444 let children_range = NodeRange {
1445 nodes: children_nodes,
1446 level: children_level,
1447 };
1448
1449 let Some(tail) = queue.back_mut() else {
1450 debug!("Adding new request onto empty queue: {children_range:?}");
1451 queue.push_back(children_range);
1452 continue;
1453 };
1454
1455 if tail.level != children_level {
1456 debug!("Adding new request for new level: {children_range:?} (existing queue tail: {tail:?})");
1457 queue.push_back(children_range);
1458 continue;
1459 }
1460
1461 let wasted_bytes = {
1462 if children_range.nodes.start >= tail.nodes.end {
1463 (children_range.nodes.start - tail.nodes.end) * size_of::<NodeItem<K>>()
1464 } else {
1465 debug_assert_eq!(
1471 children_range.nodes.start + 1,
1472 tail.nodes.end,
1473 "we only ever fetch one extra node"
1474 );
1475 debug_assert_eq!(
1476 children_level, 0,
1477 "extra node fetching only happens with leaf nodes"
1478 );
1479 0
1480 }
1481 };
1482 if wasted_bytes > combine_request_threshold {
1483 debug!("Adding new request for: {children_range:?} rather than merging with distant NodeRange: {tail:?} (would waste {wasted_bytes} bytes)");
1484 queue.push_back(children_range);
1485 continue;
1486 }
1487
1488 debug!("Extending existing request {tail:?} with nearby children: {:?} (wastes {wasted_bytes} bytes)", &children_range.nodes);
1490 tail.nodes.end = children_range.nodes.end;
1491 }
1492 }
1493
1494 let results = batch_resolve_payloads(
1496 client,
1497 payload_refs,
1498 payload_data_start,
1499 feature_begin,
1500 Some(&payload_cache),
1501 )
1502 .await?;
1503
1504 Ok(results)
1505 }
1506
1507 #[cfg(feature = "http")]
1508 #[allow(clippy::too_many_arguments)]
1509 pub async fn http_stream_find_partition<T: AsyncHttpRangeClient>(
1510 client: &mut AsyncBufferedHttpRangeClient<T>,
1511 index_begin: usize,
1512 num_items: usize,
1513 branching_factor: u16,
1514 key: K,
1515 _combine_request_threshold: usize,
1516 ) -> Result<usize> {
1517 if num_items == 0 {
1518 return Ok(0);
1519 }
1520
1521 let node_size = branching_factor as usize - 1;
1522 let level_bounds = Self::generate_level_bounds(num_items, branching_factor);
1523
1524 debug!("http_stream_find_partition - index_begin: {index_begin}, num_items: {num_items}, branching_factor: {branching_factor}, level_bounds: {level_bounds:?}, key: {key:?}");
1525
1526 let mut node_index = 0;
1528
1529 for level in (1..level_bounds.len()).rev() {
1531 let end = min(node_index + node_size, level_bounds[level].end);
1532
1533 let node_range = Range {
1535 start: node_index,
1536 end,
1537 };
1538
1539 let node_items = read_http_node_items(client, index_begin, &node_range).await?;
1541
1542 if node_items.is_empty() {
1543 continue;
1544 }
1545
1546 match node_items.binary_search_by(|item: &NodeItem<K>| item.key.cmp(&key)) {
1548 Ok(index) => {
1549 node_index = node_items[index].offset as usize;
1551 }
1552
1553 Err(index) => {
1554 if index == 0 {
1556 node_index = node_items[0].offset as usize;
1559 } else if index >= node_items.len() {
1560 node_index = node_items[node_items.len() - 1].offset as usize + node_size;
1563 } else {
1564 node_index = node_items[index].offset as usize;
1567 }
1568 }
1569 }
1570 }
1571
1572 Ok(node_index)
1575 }
1576
1577 pub fn tree_size(num_items: usize) -> usize {
1578 num_items * Entry::<K>::SERIALIZED_SIZE
1579 }
1580
1581 pub fn estimate_payload_section_size(
1595 num_items: usize,
1596 duplicate_percentage: Option<f32>,
1597 avg_duplicates_per_key: Option<f32>,
1598 ) -> usize {
1599 let dup_pct = duplicate_percentage.unwrap_or(0.1); let avg_dups = avg_duplicates_per_key.unwrap_or(3.0); let num_dup_keys = (num_items as f32 * dup_pct).ceil() as usize;
1605
1606 let avg_entry_size = 4 + (avg_dups as usize * 8);
1610
1611 num_dup_keys * avg_entry_size
1613 }
1614
1615 pub fn index_size(num_items: usize, branching_factor: u16, payload_size: usize) -> usize {
1616 assert!(branching_factor >= 2, "Node size must be at least 2");
1617 assert!(num_items > 0, "Cannot create empty tree");
1618 let branching_factor_min = branching_factor.clamp(2, 65535) as usize;
1619 let mut n = num_items;
1625 let mut num_nodes = n;
1626
1627 loop {
1628 n = n.div_ceil(branching_factor_min);
1629 num_nodes += n;
1630 if n < branching_factor_min {
1631 break;
1632 }
1633 }
1634
1635 num_nodes * NodeItem::<K>::SERIALIZED_SIZE + payload_size
1636 }
1637
1638 pub fn payload_size(&self) -> usize {
1639 self.payload_data.len()
1640 }
1641
1642 pub fn num_leaf_items(&self) -> usize {
1643 self.num_leaf_nodes
1644 }
1645
1646 pub fn num_items(&self) -> usize {
1647 self.node_items.len()
1648 }
1649
1650 pub fn branching_factor(&self) -> u16 {
1651 self.branching_factor
1652 }
1653
1654 pub fn stream_write<W: Write>(&self, out: &mut W) -> Result<usize> {
1656 let mut written_bytes = 0;
1658 for item in &self.node_items {
1660 written_bytes += item.write_to(out)?;
1661 }
1662 if self.payload_initialized && !self.payload_data.is_empty() {
1664 out.write_all(&self.payload_data)?;
1665 written_bytes += self.payload_data.len();
1666 }
1667 Ok(written_bytes)
1668 }
1669
1670 #[cfg(feature = "http")]
1671 #[allow(clippy::too_many_arguments)]
1672 pub async fn http_stream_find_range<T: AsyncHttpRangeClient>(
1673 client: &mut AsyncBufferedHttpRangeClient<T>,
1674 index_begin: usize,
1675 feature_begin: usize,
1676 num_items: usize,
1677 branching_factor: u16,
1678 lower: K,
1679 upper: K,
1680 combine_request_threshold: usize,
1681 ) -> Result<Vec<HttpSearchResultItem>> {
1682 Self::http_stream_find_range_strict(
1683 client,
1684 index_begin,
1685 feature_begin,
1686 num_items,
1687 branching_factor,
1688 lower,
1689 false,
1690 upper,
1691 false,
1692 combine_request_threshold,
1693 )
1694 .await
1695 }
1696
1697 #[cfg(feature = "http")]
1701 #[allow(clippy::too_many_arguments)]
1702 pub async fn http_stream_find_range_strict<T: AsyncHttpRangeClient>(
1703 client: &mut AsyncBufferedHttpRangeClient<T>,
1704 index_begin: usize,
1705 feature_begin: usize,
1706 num_items: usize,
1707 branching_factor: u16,
1708 lower: K,
1709 lower_strict: bool,
1710 upper: K,
1711 upper_strict: bool,
1712 combine_request_threshold: usize,
1713 ) -> Result<Vec<HttpSearchResultItem>> {
1714 debug!("http_stream_find_range starts: index_begin: {index_begin}, feature_begin: {feature_begin}, num_items: {num_items}, branching_factor: {branching_factor}, lower: {lower:?}, upper: {upper:?}");
1715
1716 if lower > upper {
1718 return Ok(Vec::new());
1719 }
1720
1721 if lower == upper {
1722 if lower_strict || upper_strict {
1724 return Ok(Vec::new());
1725 }
1726 return Self::http_stream_find_exact(
1729 client,
1730 index_begin,
1731 feature_begin,
1732 num_items,
1733 branching_factor,
1734 lower,
1735 combine_request_threshold,
1736 )
1737 .await;
1738 }
1739
1740 let node_size = branching_factor as usize - 1;
1741 let level_bounds = Self::generate_level_bounds(num_items, branching_factor);
1742
1743 let num_all_items = level_bounds
1744 .first()
1745 .expect("Btree has at least one level when node_size >= 2 and num_items > 0")
1746 .end;
1747
1748 let payload_data_start = index_begin + Stree::<K>::tree_size(num_all_items);
1749
1750 let prefetch_size = Self::compute_payload_prefetch_size(num_items, None, Some(1.5));
1753 debug!("prefetching payload with size: {} bytes", prefetch_size);
1754
1755 let payload_cache = prefetch_payload(client, payload_data_start, prefetch_size).await?;
1757
1758 debug!("http_stream_find_range - index_begin: {index_begin}, feature_begin: {feature_begin}, num_items: {num_items}, branching_factor: {branching_factor}, level_bounds: {level_bounds:?}, lower: {lower:?}, upper: {upper:?}");
1759 let _ = level_bounds
1760 .first()
1761 .expect("RTree has at least one level when node_size >= 2 and num_items > 0");
1762
1763 let upper_idx = Self::http_stream_find_partition(
1765 client,
1766 index_begin,
1767 num_items,
1768 branching_factor,
1769 upper.clone(),
1770 combine_request_threshold,
1771 )
1772 .await?;
1773
1774 let lower_idx = Self::http_stream_find_partition(
1775 client,
1776 index_begin,
1777 num_items,
1778 branching_factor,
1779 lower.clone(),
1780 combine_request_threshold,
1781 )
1782 .await?;
1783
1784 let leaf_level = 0;
1786 let leaf_start = level_bounds[leaf_level].start;
1787 let leaf_end = level_bounds[leaf_level].end;
1788
1789 let start_idx = max(lower_idx, leaf_start);
1791 let end_idx = min(upper_idx + 2 * node_size, leaf_end);
1799
1800 let mut payload_refs = Vec::new();
1802
1803 let mut current_idx = start_idx;
1805 while current_idx < end_idx {
1806 let node_end = min(current_idx + node_size, end_idx);
1807
1808 let node_range = Range {
1810 start: current_idx,
1811 end: node_end,
1812 };
1813
1814 let node_items = read_http_node_items::<K, T>(client, index_begin, &node_range).await?;
1816
1817 for item in node_items.iter() {
1819 if in_bounds(&item.key, &lower, lower_strict, &upper, upper_strict) {
1820 let off = item.offset;
1821
1822 if (off & PAYLOAD_TAG) != 0 {
1823 let rel = (off & PAYLOAD_MASK) as usize;
1824 payload_refs.push(PayloadRef::Indirect(rel));
1826 } else {
1827 payload_refs.push(PayloadRef::Direct(off));
1829 }
1830 }
1831 }
1832
1833 current_idx = node_end;
1834 }
1835
1836 let results = batch_resolve_payloads(
1838 client,
1839 payload_refs,
1840 payload_data_start,
1841 feature_begin,
1842 Some(&payload_cache),
1843 )
1844 .await?;
1845
1846 Ok(results)
1847 }
1848}
1849
1850#[cfg(feature = "http")]
1851pub mod http {
1852 use std::ops::{Range, RangeFrom};
1853
1854 #[derive(Debug, Clone, Eq, PartialEq)]
1856 pub enum HttpRange {
1857 Range(Range<usize>),
1858 RangeFrom(RangeFrom<usize>),
1859 }
1860
1861 impl HttpRange {
1862 pub fn start(&self) -> usize {
1863 match self {
1864 Self::Range(range) => range.start,
1865 Self::RangeFrom(range) => range.start,
1866 }
1867 }
1868
1869 pub fn end(&self) -> Option<usize> {
1870 match self {
1871 Self::Range(range) => Some(range.end),
1872 Self::RangeFrom(_) => None,
1873 }
1874 }
1875
1876 pub fn with_end(self, end: Option<usize>) -> Self {
1877 match end {
1878 Some(end) => Self::Range(self.start()..end),
1879 None => Self::RangeFrom(self.start()..),
1880 }
1881 }
1882
1883 pub fn length(&self) -> Option<usize> {
1884 match self {
1885 Self::Range(range) => Some(range.end - range.start),
1886 Self::RangeFrom(_) => None,
1887 }
1888 }
1889 }
1890
1891 #[derive(Debug, Eq, PartialEq, Clone)]
1892 pub struct HttpSearchResultItem {
1894 pub range: HttpRange,
1896 }
1897}
1898#[cfg(feature = "http")]
1899pub(crate) use http::*;
1900
1901#[cfg(test)]
1902mod tests {
1903 use super::*;
1904 use crate::static_btree::error::Result;
1905 use crate::static_btree::key::FixedStringKey;
1906
1907 #[test]
1908 fn test_compute_payload_prefetch_size() -> Result<()> {
1909 let small_size = Stree::<i32>::compute_payload_prefetch_size(100, None, None);
1911 assert!(small_size >= 16 * 1024, "Minimum size should be enforced");
1912
1913 let medium_size = Stree::<i32>::compute_payload_prefetch_size(10000, None, None);
1915 assert!(
1916 medium_size > small_size,
1917 "Medium tree should have larger prefetch size"
1918 );
1919
1920 let large_size = Stree::<i32>::compute_payload_prefetch_size(100000, None, None);
1922 assert!(
1923 large_size > medium_size,
1924 "Large tree should have larger prefetch size"
1925 );
1926
1927 let custom_size = Stree::<i32>::compute_payload_prefetch_size(1000, Some(128), Some(2.0));
1929 assert!(
1930 custom_size > Stree::<i32>::compute_payload_prefetch_size(1000, None, None),
1931 "Custom settings should produce larger size"
1932 );
1933
1934 let huge_size =
1936 Stree::<i32>::compute_payload_prefetch_size(10000000, Some(1024), Some(10.0));
1937 assert!(
1938 huge_size <= 4 * 1024 * 1024,
1939 "Maximum size should be enforced"
1940 );
1941
1942 Ok(())
1943 }
1944
1945 #[test]
1946 fn test_estimate_payload_section_size() -> Result<()> {
1947 let small_size = Stree::<i32>::estimate_payload_section_size(100, None, None);
1949 assert_eq!(
1950 small_size,
1951 10 * (4 + 3 * 8),
1952 "Size calculation should match expected formula"
1953 );
1954
1955 let custom_size = Stree::<i32>::estimate_payload_section_size(1000, Some(0.2), Some(5.0));
1957 assert_eq!(
1958 custom_size,
1959 200 * (4 + 5 * 8),
1960 "Custom settings should be applied correctly"
1961 );
1962
1963 Ok(())
1964 }
1965
1966 #[tokio::test]
1967 async fn test_payload_cache() -> Result<()> {
1968 use crate::static_btree::payload::PayloadEntry;
1969
1970 let mut entry = PayloadEntry::new();
1972 entry.add_offset(42);
1973 entry.add_offset(43);
1974
1975 let serialized = entry.serialize();
1977
1978 let mut cache = PayloadCache::new();
1980 cache.update(1000, serialized.clone());
1981
1982 assert!(cache.contains(1000), "Offset should be in cache");
1984 assert!(!cache.contains(999), "Offset should not be in cache");
1985 assert!(
1986 !cache.contains(1000 + serialized.len()),
1987 "Offset should not be in cache"
1988 );
1989
1990 let retrieved_entry = cache.get_entry(1000)?;
1992 assert_eq!(retrieved_entry.count, 2, "Entry count should match");
1993 assert_eq!(
1994 retrieved_entry.offsets,
1995 vec![42, 43],
1996 "Entry offsets should match"
1997 );
1998
1999 let err = cache.get_entry(2000).unwrap_err();
2001 assert!(
2002 matches!(err, Error::PayloadOffsetNotInCache),
2003 "Should return correct error"
2004 );
2005
2006 Ok(())
2007 }
2008
2009 #[test]
2010 fn tree_2items() -> Result<()> {
2011 let mut nodes = Vec::new();
2012 nodes.push(NodeItem::new(0, 0));
2013 nodes.push(NodeItem::new(2, 0));
2014 assert!(nodes[0].equals(&NodeItem::new(0, 0)));
2015 assert!(nodes[1].equals(&NodeItem::new(2, 2)));
2016 let mut offset = 0;
2017 for node in &mut nodes {
2018 node.offset = offset;
2019 offset += NodeItem::<u64>::SERIALIZED_SIZE as u64;
2020 }
2021 let tree = Stree::build(&nodes, 2)?;
2022 let list = tree.find_exact(0)?;
2023 assert_eq!(list.len(), 1);
2024 assert_eq!(list[0].offset as u64, nodes[0].offset);
2025
2026 let list = tree.find_exact(2)?;
2027 assert_eq!(list.len(), 1);
2028 assert_eq!(list[0].offset as u64, nodes[1].offset);
2029
2030 let list = tree.find_exact(1)?;
2031 assert_eq!(list.len(), 0);
2032
2033 let list = tree.find_exact(3)?;
2034 assert_eq!(list.len(), 0);
2035
2036 Ok(())
2037 }
2038
2039 #[test]
2040 fn tree_19items_roundtrip_find_exact() -> Result<()> {
2041 let mut nodes = vec![
2042 NodeItem::new(0_i64, 0_u64),
2043 NodeItem::new(1_i64, 1_u64),
2044 NodeItem::new(2_i64, 2_u64),
2045 NodeItem::new(3_i64, 3_u64),
2046 NodeItem::new(4_i64, 4_u64),
2047 NodeItem::new(5_i64, 5_u64),
2048 NodeItem::new(6_i64, 6_u64),
2049 NodeItem::new(7_i64, 7_u64),
2050 NodeItem::new(8_i64, 8_u64),
2051 NodeItem::new(9_i64, 9_u64),
2052 NodeItem::new(10_i64, 10_u64),
2053 NodeItem::new(11_i64, 11_u64),
2054 NodeItem::new(12_i64, 12_u64),
2055 NodeItem::new(13_i64, 13_u64),
2056 NodeItem::new(14_i64, 14_u64),
2057 NodeItem::new(15_i64, 15_u64),
2058 NodeItem::new(16_i64, 16_u64),
2059 NodeItem::new(17_i64, 17_u64),
2060 NodeItem::new(18_i64, 18_u64),
2061 ];
2062
2063 let mut offset = 0;
2064 for node in &mut nodes {
2065 node.offset = offset;
2066 offset += NodeItem::<u64>::SERIALIZED_SIZE as u64;
2067 }
2068 let tree = Stree::build(&nodes, 4)?;
2069 let list = tree.find_exact(10)?;
2070 assert_eq!(list.len(), 1);
2071 assert_eq!({ list[0].offset }, nodes[10].offset as usize);
2072
2073 let list = tree.find_exact(0)?;
2074 assert_eq!(list.len(), 1);
2075 assert_eq!({ list[0].offset }, nodes[0].offset as usize);
2076
2077 let list = tree.find_exact(18)?;
2078 assert_eq!(list.len(), 1);
2079 assert_eq!({ list[0].offset }, nodes[18].offset as usize);
2080
2081 let list = tree.find_exact(19)?;
2083 assert_eq!(list.len(), 0);
2084
2085 let list = tree.find_exact(-1)?;
2087 assert_eq!(list.len(), 0);
2088
2089 Ok(())
2090 }
2091
2092 #[test]
2093 fn test_range_search() -> Result<()> {
2094 let mut nodes = vec![
2096 NodeItem::new(0_i64, 0_u64),
2097 NodeItem::new(1_i64, 1_u64),
2098 NodeItem::new(2_i64, 2_u64),
2099 NodeItem::new(3_i64, 3_u64),
2100 NodeItem::new(4_i64, 4_u64),
2101 NodeItem::new(5_i64, 5_u64),
2102 NodeItem::new(6_i64, 6_u64),
2103 NodeItem::new(7_i64, 7_u64),
2104 NodeItem::new(8_i64, 8_u64),
2105 NodeItem::new(9_i64, 9_u64),
2106 NodeItem::new(10_i64, 10_u64),
2107 NodeItem::new(11_i64, 11_u64),
2108 NodeItem::new(12_i64, 12_u64),
2109 NodeItem::new(13_i64, 13_u64),
2110 NodeItem::new(14_i64, 14_u64),
2111 NodeItem::new(15_i64, 15_u64),
2112 NodeItem::new(16_i64, 16_u64),
2113 NodeItem::new(17_i64, 17_u64),
2114 NodeItem::new(18_i64, 18_u64),
2115 ];
2116
2117 let mut offset = 0;
2118 for node in &mut nodes {
2119 node.offset = offset;
2120 offset += NodeItem::<i64>::SERIALIZED_SIZE as u64;
2121 }
2122 let tree = Stree::build(&nodes, 4)?;
2123
2124 let list = tree.find_range(0, 18)?;
2131 assert_eq!(list.len(), 19);
2132
2133 let keys: Vec<i64> = list
2135 .iter()
2136 .map(|item| {
2137 let idx = item.offset / Entry::<i64>::SERIALIZED_SIZE;
2138 idx as i64
2139 })
2140 .collect();
2141
2142 for i in 0..=17 {
2144 assert!(keys.contains(&i));
2145 }
2146
2147 let list = tree.find_range(0, 5)?;
2149 assert_eq!(list.len(), 6);
2150 for item in &list {
2151 assert!(item.index <= 5);
2152 }
2153
2154 let list = tree.find_range(7, 12)?;
2156
2157 assert!(list.len() >= 5);
2160
2161 let found_indices: Vec<usize> = list
2163 .iter()
2164 .map(|item| item.offset / Entry::<i64>::SERIALIZED_SIZE)
2165 .collect();
2166
2167 assert!(found_indices.contains(&7));
2169 assert!(found_indices.contains(&8));
2170 assert!(found_indices.contains(&9));
2171 assert!(found_indices.contains(&10));
2172 assert!(found_indices.contains(&11));
2173
2174 let list = tree.find_range(15, 18)?;
2176
2177 assert!(list.len() >= 3);
2180
2181 let found_indices: Vec<usize> = list
2183 .iter()
2184 .map(|item| item.offset / Entry::<i64>::SERIALIZED_SIZE)
2185 .collect();
2186
2187 assert!(found_indices.contains(&15));
2189 assert!(found_indices.contains(&16));
2190 assert!(found_indices.contains(&17));
2191
2192 let list = tree.find_range(9, 9)?;
2194 assert_eq!(list.len(), 1);
2195 assert_eq!(list[0].index, 9);
2196
2197 let list = tree.find_range(100, 200)?;
2199 assert_eq!(list.len(), 0);
2200
2201 let list = tree.find_range(16, 100)?;
2203 assert_eq!(list.len(), 3); let list = tree.find_range(-10, 2)?;
2207 assert_eq!(list.len(), 3); let list = tree.find_range(10, 5)?;
2211 assert_eq!(list.len(), 0);
2212
2213 Ok(())
2214 }
2215
2216 #[test]
2224 fn test_find_exact_on_max_valued_key() -> Result<()> {
2225 let nodes: Vec<Entry<bool>> = (0..8)
2226 .map(|i| Entry {
2227 key: i % 2 == 0,
2228 offset: (i * 10) as u64,
2229 })
2230 .collect();
2231 let tree = Stree::build(&nodes, 4)?;
2232
2233 let list = tree.find_exact(true)?;
2235 assert!(!list.is_empty(), "Eq(true) found nothing");
2236
2237 let list = tree.find_exact(false)?;
2238 assert!(!list.is_empty(), "Eq(false) found nothing");
2239 Ok(())
2240 }
2241
2242 #[test]
2250 fn test_range_upper_bound_on_separator_key() -> Result<()> {
2251 let nodes: Vec<Entry<i64>> = (0..19)
2252 .map(|i| Entry {
2253 key: i as i64,
2254 offset: (i * 100) as u64,
2255 })
2256 .collect();
2257 let tree = Stree::build(&nodes, 4)?;
2258
2259 for upper in 0..19i64 {
2261 let list = tree.find_range(0, upper)?;
2262 let keys: Vec<i64> = list.iter().map(|r| (r.offset / 100) as i64).collect();
2263 assert!(
2264 keys.contains(&upper),
2265 "find_range(0, {upper}) dropped its inclusive upper bound; got {keys:?}"
2266 );
2267 assert_eq!(
2268 list.len(),
2269 (upper + 1) as usize,
2270 "find_range(0, {upper}) returned the wrong count"
2271 );
2272 }
2273 Ok(())
2274 }
2275
2276 #[test]
2281 fn test_range_strict_bounds_on_separator_keys() -> Result<()> {
2282 let nodes: Vec<Entry<i64>> = (0..19)
2283 .map(|i| Entry {
2284 key: i as i64,
2285 offset: (i * 100) as u64,
2286 })
2287 .collect();
2288 let tree = Stree::build(&nodes, 4)?;
2289
2290 let keys_of = |list: Vec<SearchResultItem>| -> Vec<i64> {
2291 list.iter().map(|r| (r.offset / 100) as i64).collect()
2292 };
2293
2294 for bound in 0..19i64 {
2295 let above = keys_of(tree.find_range_strict(bound, true, 18, false)?);
2297 assert_eq!(
2298 above,
2299 ((bound + 1)..19).collect::<Vec<_>>(),
2300 "find_range_strict(({bound}, 18]) is wrong"
2301 );
2302
2303 let below = keys_of(tree.find_range_strict(0, false, bound, true)?);
2305 assert_eq!(
2306 below,
2307 (0..bound).collect::<Vec<_>>(),
2308 "find_range_strict([0, {bound})) is wrong"
2309 );
2310 }
2311 Ok(())
2312 }
2313
2314 #[test]
2317 fn test_range_strict_degenerate_bounds() -> Result<()> {
2318 let nodes = vec![
2319 NodeItem::new(8_i64, 80_u64),
2320 NodeItem::new(9_i64, 90_u64),
2321 NodeItem::new(9_i64, 91_u64), NodeItem::new(10_i64, 100_u64),
2323 NodeItem::new(11_i64, 110_u64),
2324 NodeItem::new(12_i64, 120_u64),
2325 ];
2326 let tree = Stree::build(&nodes, 4)?;
2327
2328 let offsets = |list: Vec<SearchResultItem>| -> Vec<usize> {
2329 let mut offs: Vec<usize> = list.iter().map(|r| r.offset).collect();
2330 offs.sort_unstable();
2331 offs
2332 };
2333
2334 assert_eq!(
2335 offsets(tree.find_range_strict(9, false, 9, false)?),
2336 offsets(tree.find_exact(9)?),
2337 "[9, 9] must agree with find_exact(9)"
2338 );
2339 assert!(tree.find_range_strict(9, true, 9, false)?.is_empty());
2340 assert!(tree.find_range_strict(9, false, 9, true)?.is_empty());
2341 assert!(tree.find_range_strict(9, true, 9, true)?.is_empty());
2342
2343 assert!(tree.find_range_strict(10, false, 9, false)?.is_empty());
2345 assert!(tree.find_range_strict(10, true, 9, true)?.is_empty());
2346
2347 Ok(())
2348 }
2349
2350 #[test]
2352 fn test_stream_find_range_strict() -> Result<()> {
2353 let nodes = vec![
2354 NodeItem::new(1_i64, 10_u64),
2355 NodeItem::new(1_i64, 20_u64),
2356 NodeItem::new(2_i64, 30_u64),
2357 NodeItem::new(3_i64, 40_u64),
2358 NodeItem::new(4_i64, 50_u64),
2359 NodeItem::new(5_i64, 60_u64),
2360 ];
2361 let tree = Stree::build(&nodes, 3)?;
2362 let mut buf = Vec::new();
2363 tree.stream_write(&mut buf)?;
2364
2365 let scan = |lower, lower_strict, upper, upper_strict| -> Result<Vec<usize>> {
2366 let mut cursor = std::io::Cursor::new(&buf);
2367 let res = Stree::<i64>::stream_find_range_strict(
2368 &mut cursor,
2369 tree.num_leaf_nodes,
2370 3,
2371 lower,
2372 lower_strict,
2373 upper,
2374 upper_strict,
2375 )?;
2376 let mut offs: Vec<usize> = res.iter().map(|r| r.offset).collect();
2377 offs.sort_unstable();
2378 Ok(offs)
2379 };
2380
2381 assert_eq!(scan(1, false, 5, false)?, vec![10, 20, 30, 40, 50, 60]);
2382 assert_eq!(scan(1, true, 5, false)?, vec![30, 40, 50, 60]);
2383 assert_eq!(scan(1, false, 5, true)?, vec![10, 20, 30, 40, 50]);
2384 assert_eq!(scan(1, true, 5, true)?, vec![30, 40, 50]);
2385 assert!(scan(3, true, 3, false)?.is_empty());
2386 assert_eq!(scan(3, false, 3, false)?, vec![40]);
2387
2388 Ok(())
2389 }
2390
2391 #[test]
2392 fn test_string_range_search() -> Result<()> {
2393 let mut nodes = vec![
2394 NodeItem::new(FixedStringKey::<10>::from_str("a"), 0_u64),
2395 NodeItem::new(FixedStringKey::<10>::from_str("b"), 1_u64),
2396 NodeItem::new(FixedStringKey::<10>::from_str("c"), 2_u64),
2397 NodeItem::new(FixedStringKey::<10>::from_str("d"), 3_u64),
2398 NodeItem::new(FixedStringKey::<10>::from_str("e"), 4_u64),
2399 NodeItem::new(FixedStringKey::<10>::from_str("f"), 5_u64),
2400 NodeItem::new(FixedStringKey::<10>::from_str("g"), 6_u64),
2401 NodeItem::new(FixedStringKey::<10>::from_str("h"), 7_u64),
2402 NodeItem::new(FixedStringKey::<10>::from_str("i"), 8_u64),
2403 NodeItem::new(FixedStringKey::<10>::from_str("j"), 9_u64),
2404 NodeItem::new(FixedStringKey::<10>::from_str("k"), 10_u64),
2405 NodeItem::new(FixedStringKey::<10>::from_str("l"), 11_u64),
2406 NodeItem::new(FixedStringKey::<10>::from_str("m"), 12_u64),
2407 NodeItem::new(FixedStringKey::<10>::from_str("n"), 13_u64),
2408 NodeItem::new(FixedStringKey::<10>::from_str("o"), 14_u64),
2409 NodeItem::new(FixedStringKey::<10>::from_str("p"), 15_u64),
2410 NodeItem::new(FixedStringKey::<10>::from_str("q"), 16_u64),
2411 NodeItem::new(FixedStringKey::<10>::from_str("r"), 17_u64),
2412 NodeItem::new(FixedStringKey::<10>::from_str("s"), 18_u64),
2413 ];
2414
2415 let mut offset = 0;
2416 for node in &mut nodes {
2417 node.offset = offset;
2418 offset += NodeItem::<FixedStringKey<10>>::SERIALIZED_SIZE as u64;
2419 }
2420 let tree = Stree::build(&nodes, 3)?;
2421
2422 let list = tree.find_range(
2424 FixedStringKey::<10>::from_str("c"),
2425 FixedStringKey::<10>::from_str("g"),
2426 )?;
2427
2428 assert!(list.len() >= 4); let found_keys: Vec<String> = list
2432 .iter()
2433 .map(|item| {
2434 let idx = item.offset / Entry::<FixedStringKey<10>>::SERIALIZED_SIZE;
2435 let c = (b'a' + idx as u8) as char;
2436 c.to_string()
2437 })
2438 .collect();
2439
2440 assert!(found_keys.iter().any(|k| k == "c"));
2442 assert!(found_keys.iter().any(|k| k == "d"));
2443 assert!(found_keys.iter().any(|k| k == "e"));
2444 assert!(found_keys.iter().any(|k| k == "f"));
2445
2446 let list = tree.find_range(
2448 FixedStringKey::<10>::from_str("t"),
2449 FixedStringKey::<10>::from_str("z"),
2450 )?;
2451
2452 assert_eq!(list.len(), 0);
2453
2454 let list = tree.find_range(
2456 FixedStringKey::<10>::from_str("k"),
2457 FixedStringKey::<10>::from_str("k"),
2458 )?;
2459
2460 if !list.is_empty() {
2461 let idx = list[0].offset / Entry::<FixedStringKey<10>>::SERIALIZED_SIZE;
2463 let found_key = (b'a' + idx as u8) as char;
2464
2465 assert_eq!(found_key, 'k');
2466 } else {
2467 println!("No key 'k' found - this is acceptable with the partition-based approach")
2470 }
2471
2472 Ok(())
2473 }
2474
2475 #[test]
2476 fn tree_generate_nodes() -> Result<()> {
2477 let nodes = vec![
2478 NodeItem::new(0_u64, 0_u64),
2479 NodeItem::new(1_u64, 1_u64),
2480 NodeItem::new(2_u64, 2_u64),
2481 NodeItem::new(3_u64, 3_u64),
2482 NodeItem::new(4_u64, 4_u64),
2483 NodeItem::new(5_u64, 5_u64),
2484 NodeItem::new(6_u64, 6_u64),
2485 NodeItem::new(7_u64, 7_u64),
2486 NodeItem::new(8_u64, 8_u64),
2487 NodeItem::new(9_u64, 9_u64),
2488 NodeItem::new(10_u64, 10_u64),
2489 NodeItem::new(11_u64, 11_u64),
2490 NodeItem::new(12_u64, 12_u64),
2491 NodeItem::new(13_u64, 13_u64),
2492 NodeItem::new(14_u64, 14_u64),
2493 NodeItem::new(15_u64, 15_u64),
2494 NodeItem::new(16_u64, 16_u64),
2495 NodeItem::new(17_u64, 17_u64),
2496 NodeItem::new(18_u64, 18_u64),
2497 ];
2498
2499 let tree = Stree::build(&nodes, 3)?;
2501 let keys = tree
2502 .node_items
2503 .into_iter()
2504 .map(|nodes| nodes.key)
2505 .collect::<Vec<_>>();
2506 let expected = vec![
2507 18,
2508 6,
2509 12,
2510 u64::MAX,
2511 2,
2512 4,
2513 8,
2514 10,
2515 14,
2516 16,
2517 u64::MAX,
2518 0,
2519 1,
2520 2,
2521 3,
2522 4,
2523 5,
2524 6,
2525 7,
2526 8,
2527 9,
2528 10,
2529 11,
2530 12,
2531 13,
2532 14,
2533 15,
2534 16,
2535 17,
2536 18,
2537 ];
2538 assert_eq!(keys, expected);
2539
2540 let tree = Stree::build(&nodes, 4)?;
2542 let keys = tree
2543 .node_items
2544 .into_iter()
2545 .map(|nodes| nodes.key)
2546 .collect::<Vec<_>>();
2547 let expected = vec![
2548 12,
2549 u64::MAX, 3,
2551 6,
2552 9,
2553 15,
2554 18,
2555 0,
2556 1,
2557 2,
2558 3,
2559 4,
2560 5,
2561 6,
2562 7,
2563 8,
2564 9,
2565 10,
2566 11,
2567 12,
2568 13,
2569 14,
2570 15,
2571 16,
2572 17,
2573 18,
2574 ];
2575 assert_eq!(keys, expected);
2576 Ok(())
2577 }
2578
2579 #[test]
2580 fn tree_19items_roundtrip_string() -> Result<()> {
2581 let mut nodes = vec![
2582 NodeItem::new(FixedStringKey::<10>::from_str("a"), 0_u64),
2583 NodeItem::new(FixedStringKey::<10>::from_str("b"), 1_u64),
2584 NodeItem::new(FixedStringKey::<10>::from_str("c"), 2_u64),
2585 NodeItem::new(FixedStringKey::<10>::from_str("d"), 3_u64),
2586 NodeItem::new(FixedStringKey::<10>::from_str("e"), 4_u64),
2587 NodeItem::new(FixedStringKey::<10>::from_str("f"), 5_u64),
2588 NodeItem::new(FixedStringKey::<10>::from_str("g"), 6_u64),
2589 NodeItem::new(FixedStringKey::<10>::from_str("h"), 7_u64),
2590 NodeItem::new(FixedStringKey::<10>::from_str("i"), 8_u64),
2591 NodeItem::new(FixedStringKey::<10>::from_str("j"), 9_u64),
2592 NodeItem::new(FixedStringKey::<10>::from_str("k"), 10_u64),
2593 NodeItem::new(FixedStringKey::<10>::from_str("l"), 11_u64),
2594 NodeItem::new(FixedStringKey::<10>::from_str("m"), 12_u64),
2595 NodeItem::new(FixedStringKey::<10>::from_str("n"), 13_u64),
2596 NodeItem::new(FixedStringKey::<10>::from_str("o"), 14_u64),
2597 NodeItem::new(FixedStringKey::<10>::from_str("p"), 15_u64),
2598 NodeItem::new(FixedStringKey::<10>::from_str("q"), 16_u64),
2599 NodeItem::new(FixedStringKey::<10>::from_str("r"), 17_u64),
2600 NodeItem::new(FixedStringKey::<10>::from_str("s"), 18_u64),
2601 ];
2602
2603 let mut offset = 0;
2604 for node in &mut nodes {
2605 node.offset = offset;
2606 offset += NodeItem::<u64>::SERIALIZED_SIZE as u64;
2607 }
2608 let tree = Stree::build(&nodes, 3)?;
2609 let list = tree.find_exact(FixedStringKey::<10>::from_str("k"))?;
2610 assert_eq!(list.len(), 1);
2611 assert_eq!(list[0].offset, 160);
2614
2615 let list = tree.find_exact(FixedStringKey::<10>::from_str("not exists"))?;
2616 assert_eq!(list.len(), 0);
2617
2618 Ok(())
2619 }
2620 #[test]
2621 fn test_duplicates_exact() -> Result<()> {
2623 let nodes = vec![
2624 NodeItem::new(0, 0),
2625 NodeItem::new(1, 10),
2626 NodeItem::new(1, 20),
2627 NodeItem::new(1, 30),
2628 NodeItem::new(2, 40),
2629 NodeItem::new(2, 50),
2630 NodeItem::new(2, 60),
2631 NodeItem::new(3, 70),
2632 NodeItem::new(3, 80),
2633 NodeItem::new(3, 90),
2634 NodeItem::new(4, 100),
2635 NodeItem::new(5, 110),
2636 NodeItem::new(6, 120),
2637 NodeItem::new(7, 130),
2638 NodeItem::new(8, 140),
2639 NodeItem::new(9, 150),
2640 ];
2641 let tree = Stree::build(&nodes, 2)?;
2642 let res = tree.find_exact(1)?;
2643 assert_eq!(res.len(), 3);
2644 let mut offs: Vec<usize> = res.iter().map(|r| r.offset).collect();
2645 offs.sort_unstable();
2646 assert_eq!(offs, vec![10, 20, 30]);
2647 Ok(())
2648 }
2649
2650 #[test]
2651 fn test_duplicates_range() -> Result<()> {
2653 let nodes = vec![
2654 NodeItem::new(1, 5),
2655 NodeItem::new(1, 6),
2656 NodeItem::new(2, 7),
2657 NodeItem::new(2, 8),
2658 NodeItem::new(3, 9),
2659 ];
2660 let tree = Stree::build(&nodes, 3)?;
2661 let res = tree.find_range(1, 2)?;
2663 assert_eq!(res.len(), 4);
2664 let mut offs: Vec<usize> = res.iter().map(|r| r.offset).collect();
2665 offs.sort_unstable();
2666 assert_eq!(offs, vec![5, 6, 7, 8]);
2667 Ok(())
2668 }
2669
2670 #[test]
2671 fn test_stream_write_payload() -> Result<()> {
2673 let nodes = vec![NodeItem::new(1, 10), NodeItem::new(1, 20)];
2675
2676 let tree = Stree::<i32>::build(&nodes, 2)?;
2677 let mut buf = Vec::new();
2679 let _written = tree.stream_write(&mut buf)?;
2680 let idx_bytes = Stree::<i32>::tree_size(tree.num_items());
2682 let payload = &tree.payload_data;
2684 assert!(!payload.is_empty());
2685 assert_eq!(buf.len(), idx_bytes + payload.len());
2686 assert_eq!(&buf[idx_bytes..], payload);
2687 Ok(())
2688 }
2689
2690 #[test]
2691 fn test_read_write_roundtrip() -> Result<()> {
2693 let nodes = vec![
2695 NodeItem::new(1, 100),
2696 NodeItem::new(1, 200),
2697 NodeItem::new(2, 300),
2698 NodeItem::new(3, 400),
2699 NodeItem::new(3, 500),
2700 ];
2701 let orig = Stree::build(&nodes, 3)?;
2703 let mut buf = Vec::new();
2705 orig.stream_write(&mut buf)?;
2706 let mut cursor = std::io::Cursor::new(&buf);
2708 let restored: Stree<i32> = Stree::from_buf(&mut cursor, orig.num_leaf_nodes, 3)?;
2709 let r1 = restored.find_exact(1)?;
2711 let offs1: Vec<usize> = r1.iter().map(|r| r.offset).collect();
2712 assert_eq!(offs1.len(), 2);
2713 assert!(offs1.contains(&100));
2714 assert!(offs1.contains(&200));
2715 let r2 = restored.find_exact(2)?;
2716 assert_eq!(r2.len(), 1);
2717 assert_eq!(r2[0].offset, 300);
2718 let r3 = restored.find_exact(3)?;
2719 let offs3: Vec<usize> = r3.iter().map(|r| r.offset).collect();
2720 assert_eq!(offs3.len(), 2);
2721 assert!(offs3.contains(&400));
2722 assert!(offs3.contains(&500));
2723 Ok(())
2724 }
2725
2726 #[test]
2727 fn test_stream_find_exact() -> Result<()> {
2729 let nodes = vec![
2730 NodeItem::new(1, 11),
2731 NodeItem::new(1, 22),
2732 NodeItem::new(2, 33),
2733 ];
2734 let tree = Stree::build(&nodes, 2)?;
2735 let mut buf = Vec::new();
2736 tree.stream_write(&mut buf)?;
2737 let mut cursor = std::io::Cursor::new(&buf);
2738 let res = Stree::stream_find_exact(&mut cursor, tree.num_leaf_nodes, 2, 1)?;
2739 assert_eq!(res.len(), 2);
2740 let mut offs: Vec<usize> = res.iter().map(|r| r.offset).collect();
2741 offs.sort_unstable();
2742 assert_eq!(offs, vec![11, 22]);
2743 Ok(())
2744 }
2745
2746 #[test]
2747 fn test_stream_find_range() -> Result<()> {
2749 let nodes = vec![
2750 NodeItem::new(1, 10),
2751 NodeItem::new(1, 20),
2752 NodeItem::new(2, 30),
2753 NodeItem::new(2, 40),
2754 NodeItem::new(3, 50),
2755 NodeItem::new(4, 60),
2756 NodeItem::new(4, 70),
2757 NodeItem::new(5, 80),
2758 NodeItem::new(5, 90),
2759 NodeItem::new(6, 100),
2760 NodeItem::new(6, 110),
2761 NodeItem::new(7, 120),
2762 ];
2763 let tree = Stree::build(&nodes, 3)?;
2764 let _payload_size = tree.payload_data.len();
2765 let mut buf = Vec::new();
2766 tree.stream_write(&mut buf)?;
2767 let mut cursor = std::io::Cursor::new(&buf);
2768 let res: Vec<SearchResultItem> =
2769 Stree::stream_find_range(&mut cursor, tree.num_leaf_nodes, 3, 1, 2)?;
2770 assert_eq!(res.len(), 4);
2771 let mut offs: Vec<usize> = res.iter().map(|r| r.offset).collect();
2772 offs.sort_unstable();
2773 assert_eq!(offs, vec![10, 20, 30, 40]);
2774 Ok(())
2775 }
2776
2777 #[cfg(feature = "http")]
2778 #[tokio::test]
2779 async fn test_http_stream_find_exact() -> Result<()> {
2780 use crate::static_btree::mocked_http_range_client::MockHttpRangeClient;
2781
2782 let nodes = vec![
2783 NodeItem::new(0_i64, 0_u64),
2784 NodeItem::new(1_i64, 1_u64),
2785 NodeItem::new(1_i64, 101_u64),
2786 NodeItem::new(2_i64, 2_u64),
2787 NodeItem::new(3_i64, 3_u64),
2788 NodeItem::new(4_i64, 4_u64),
2789 NodeItem::new(5_i64, 5_u64),
2790 NodeItem::new(6_i64, 6_u64),
2791 NodeItem::new(7_i64, 7_u64),
2792 NodeItem::new(8_i64, 8_u64),
2793 NodeItem::new(9_i64, 9_u64),
2794 NodeItem::new(9_i64, 99_u64),
2795 NodeItem::new(10_i64, 10_u64),
2796 NodeItem::new(11_i64, 11_u64),
2797 NodeItem::new(12_i64, 12_u64),
2798 NodeItem::new(13_i64, 13_u64),
2799 NodeItem::new(14_i64, 14_u64),
2800 NodeItem::new(15_i64, 15_u64),
2801 NodeItem::new(16_i64, 16_u64),
2802 NodeItem::new(17_i64, 17_u64),
2803 NodeItem::new(18_i64, 18_u64),
2804 ];
2805
2806 let test_cases = vec![
2808 ((9_i64, vec![9, 99]), 4),
2823 ((-1_i64, vec![]), 4),
2824 ((1_i64, vec![1, 101]), 4),
2825 ];
2826
2827 for ((query, expected_result), branching_factor) in test_cases {
2828 let tree = Stree::<i64>::build(&nodes, branching_factor)?;
2829 let mut buf: Vec<u8> = Vec::new();
2831 tree.stream_write(&mut buf)?;
2832 let attr_index_size = buf.len();
2833
2834 let mut client = MockHttpRangeClient::new_mock_http_range_client(&buf);
2835
2836 let feature_begin = attr_index_size;
2837
2838 let expected_result = expected_result
2839 .iter()
2840 .map(|item| item + feature_begin)
2841 .collect::<Vec<usize>>();
2842
2843 let res = Stree::<i64>::http_stream_find_exact(
2845 &mut client,
2846 0, feature_begin,
2848 tree.num_leaf_nodes,
2849 branching_factor, query,
2851 256 * 1024, )
2853 .await?;
2854
2855 let mut offs: Vec<usize> = res.iter().map(|item| item.range.start()).collect();
2856 offs.sort_unstable();
2857 println!("query: {query:?}, expected_result: {expected_result:?}, offs: {offs:?}");
2858 assert_eq!(
2859 offs, expected_result,
2860 "expected_result: {expected_result:?}, offs: {offs:?}"
2861 );
2862 }
2863 Ok(())
2864 }
2865
2866 #[cfg(feature = "http")]
2867 #[tokio::test]
2868 async fn test_http_stream_find_partition() -> Result<()> {
2869 use crate::static_btree::mocked_http_range_client::MockHttpRangeClient;
2870 use std::println;
2871
2872 println!("Starting test_http_stream_find_partition");
2873
2874 let nodes = vec![
2876 NodeItem::new(0_i64, 0_u64),
2877 NodeItem::new(1_i64, 1_u64),
2878 NodeItem::new(2_i64, 2_u64),
2879 NodeItem::new(3_i64, 3_u64),
2880 NodeItem::new(4_i64, 4_u64),
2881 NodeItem::new(5_i64, 5_u64),
2882 NodeItem::new(6_i64, 6_u64),
2883 NodeItem::new(7_i64, 7_u64),
2884 NodeItem::new(8_i64, 8_u64),
2885 NodeItem::new(8_i64, 88_u64),
2886 NodeItem::new(9_i64, 9_u64),
2887 NodeItem::new(10_i64, 10_u64),
2888 NodeItem::new(11_i64, 11_u64),
2889 NodeItem::new(12_i64, 12_u64),
2890 NodeItem::new(13_i64, 13_u64),
2891 NodeItem::new(14_i64, 14_u64),
2892 NodeItem::new(15_i64, 15_u64),
2893 NodeItem::new(16_i64, 16_u64),
2894 NodeItem::new(17_i64, 17_u64),
2895 NodeItem::new(18_i64, 18_u64),
2896 ];
2897
2898 println!("Node items (key, offset):");
2900 for (i, node) in nodes.iter().enumerate() {
2901 println!("[{}] = ({}, {})", i, node.key, node.offset);
2902 }
2903
2904 let tree_4 = Stree::<i64>::build(&nodes, 4)?;
2907
2908 let test_cases = vec![
2909 (8_i64, 4, tree_4.find_partition(8_i64)?), (0_i64, 4, tree_4.find_partition(0_i64)?), (18_i64, 4, tree_4.find_partition(18_i64)?), (19_i64, 4, tree_4.find_partition(19_i64)?), (-1_i64, 4, tree_4.find_partition(-1_i64)?), ];
2916
2917 let tree_3 = Stree::<i64>::build(&nodes, 3)?;
2919 let tree_5 = Stree::<i64>::build(&nodes, 5)?;
2920 let tree_6 = Stree::<i64>::build(&nodes, 6)?;
2921
2922 let more_test_cases = vec![
2923 (4_i64, 3, tree_3.find_partition(4_i64)?), (4_i64, 5, tree_5.find_partition(4_i64)?), (4_i64, 6, tree_6.find_partition(4_i64)?), (7_i64, 4, tree_4.find_partition(7_i64)?), (7_i64, 3, tree_3.find_partition(7_i64)?), (7_i64, 5, tree_5.find_partition(7_i64)?), ];
2930
2931 let all_test_cases = [test_cases, more_test_cases].concat();
2933
2934 for (query, branching_factor, expected_position) in all_test_cases {
2935 let tree = Stree::<i64>::build(&nodes, branching_factor)?;
2936
2937 let in_memory_position = tree.find_partition(query)?;
2939
2940 assert_eq!(
2942 in_memory_position, expected_position,
2943 "Unexpected in-memory find_partition result"
2944 );
2945
2946 let mut buf = Vec::new();
2948 tree.stream_write(&mut buf)?;
2949
2950 let mut client = MockHttpRangeClient::new_mock_http_range_client(&buf);
2951
2952 let position = Stree::<i64>::http_stream_find_partition(
2954 &mut client,
2955 0, tree.num_leaf_nodes,
2957 branching_factor, query,
2959 256 * 1024, )
2961 .await?;
2962
2963 assert_eq!(
2965 position, expected_position,
2966 "HTTP version gives {position} but expected {expected_position}"
2967 );
2968 }
2969 Ok(())
2970 }
2971
2972 #[cfg(feature = "http")]
2973 #[tokio::test]
2974 async fn test_http_stream_find_range() -> Result<()> {
2975 use crate::static_btree::mocked_http_range_client::MockHttpRangeClient;
2976 use std::println;
2977
2978 println!("Starting test_http_stream_find_range");
2979
2980 let nodes = vec![
2982 NodeItem::new(0_i64, 0_u64),
2983 NodeItem::new(1_i64, 1_u64),
2984 NodeItem::new(1_i64, 101_u64), NodeItem::new(2_i64, 2_u64),
2986 NodeItem::new(3_i64, 3_u64),
2987 NodeItem::new(4_i64, 4_u64),
2988 NodeItem::new(5_i64, 5_u64),
2989 NodeItem::new(6_i64, 6_u64),
2990 NodeItem::new(7_i64, 7_u64),
2991 NodeItem::new(8_i64, 8_u64),
2992 NodeItem::new(9_i64, 9_u64),
2993 NodeItem::new(9_i64, 99_u64), NodeItem::new(10_i64, 10_u64),
2995 NodeItem::new(11_i64, 11_u64),
2996 NodeItem::new(12_i64, 12_u64),
2997 NodeItem::new(13_i64, 13_u64),
2998 NodeItem::new(14_i64, 14_u64),
2999 NodeItem::new(15_i64, 15_u64),
3000 NodeItem::new(16_i64, 16_u64),
3001 NodeItem::new(17_i64, 17_u64),
3002 NodeItem::new(18_i64, 18_u64),
3003 ];
3004
3005 let test_cases = vec![
3007 (5_i64, 10_i64, 4), (0_i64, 3_i64, 4), (15_i64, 18_i64, 4), (0_i64, 18_i64, 4), (6_i64, 6_i64, 4), (9_i64, 9_i64, 4), (1_i64, 1_i64, 4), (19_i64, 20_i64, 4), (-2_i64, -1_i64, 4), (-1_i64, 2_i64, 4), (17_i64, 20_i64, 4), (7_i64, 12_i64, 3), (7_i64, 12_i64, 5), (7_i64, 12_i64, 6), (10_i64, 5_i64, 4), ];
3024
3025 for (lower, upper, branching_factor) in test_cases {
3026 let tree = Stree::<i64>::build(&nodes, branching_factor)?;
3028
3029 println!("Tree built with num_leaf_nodes: {}", tree.num_leaf_nodes);
3030 println!("Tree level_bounds: {:?}", tree.level_bounds);
3031
3032 let in_memory_results = tree.find_range(lower, upper)?;
3034
3035 println!(
3036 "In-memory range search found {} results",
3037 in_memory_results.len()
3038 );
3039 if !in_memory_results.is_empty() {
3040 println!("First few results (offset, index):");
3041 for (i, item) in in_memory_results.iter().take(5).enumerate() {
3042 println!("[{}] = ({}, {})", i, item.offset, item.index);
3043 }
3044 }
3045
3046 let mut buf = Vec::new();
3048 tree.stream_write(&mut buf)?;
3049 let attr_index_size = buf.len();
3050
3051 let mut client = MockHttpRangeClient::new_mock_http_range_client(&buf);
3052
3053 let feature_begin = attr_index_size; let http_results = Stree::<i64>::http_stream_find_range(
3058 &mut client,
3059 0, attr_index_size,
3061 tree.num_leaf_nodes,
3062 branching_factor,
3063 lower,
3064 upper,
3065 256 * 1024, )
3067 .await?;
3068
3069 println!("HTTP range search found {} results", http_results.len());
3070
3071 let http_comparable_results: Vec<usize> = http_results
3073 .iter()
3074 .map(|item| item.range.start().saturating_sub(feature_begin))
3075 .collect();
3076
3077 let in_memory_comparable_results: Vec<usize> =
3079 in_memory_results.iter().map(|item| item.offset).collect();
3080
3081 if http_results.len() != in_memory_results.len() {
3083 println!(
3084 "Result counts differ! HTTP: {}, In-memory: {}",
3085 http_results.len(),
3086 in_memory_results.len()
3087 );
3088 }
3089
3090 let mut http_sorted = http_comparable_results.clone();
3092 http_sorted.sort_unstable();
3093
3094 let mut in_memory_sorted = in_memory_comparable_results.clone();
3095 in_memory_sorted.sort_unstable();
3096
3097 assert_eq!(
3099 http_sorted, in_memory_sorted,
3100 "HTTP results don't match in-memory results"
3101 );
3102 }
3103
3104 Ok(())
3105 }
3106
3107 }