diff --git a/src/inspector.rs b/src/inspector.rs index bc9b15a..8c1ef55 100644 --- a/src/inspector.rs +++ b/src/inspector.rs @@ -1,5 +1,3 @@ -// src/inspector.rs - //! Tools for inspecting the physical structure of Parcode files. //! //! This module provides the [`ParcodeInspector`] tool, which can analyze a Parcode file diff --git a/src/lib.rs b/src/lib.rs index c6d483b..1df8844 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -267,8 +267,8 @@ pub use api::{Parcode, ParcodeOptions}; pub use error::{ParcodeError, Result}; pub use reader::{ParcodeFile, ParcodeItem, ParcodeNative}; -// Re-export the derive macro so it is accessible as `parcode::ParcodeObject` pub use parcode_derive::ParcodeObject; +pub use rt::{ParcodeCollectionPromise, ParcodeMapPromise, ParcodePromise}; /// Constants used throughout the library. pub mod constants { diff --git a/src/rt.rs b/src/rt.rs index 7d3149a..8603eb2 100644 --- a/src/rt.rs +++ b/src/rt.rs @@ -6,14 +6,13 @@ use crate::format::ChildRef; use crate::graph::{JobConfig, SerializationJob}; use crate::reader::{ChunkNode, ParcodeItem}; use serde::de::DeserializeOwned; +use std::borrow::Cow; use std::collections::HashMap; use std::hash::Hash; use std::io::Cursor; use std::marker::PhantomData; use std::vec::IntoIter; -// --- EXISTING CONFIG WRAPPER --- - /// Wrapper that injects configuration into an existing Job. #[derive(Debug)] pub struct ConfiguredJob<'a, J: ?Sized> { @@ -46,8 +45,6 @@ impl<'a, J: SerializationJob<'a> + ?Sized> SerializationJob<'a> for ConfiguredJo } } -// --- NEW LAZY MIRROR INFRASTRUCTURE --- - /// Trait implemented by types that support Lazy Mirroring. /// /// This trait acts as a bridge between the original type `T` and its generated @@ -123,9 +120,21 @@ impl<'a, T: ParcodeItem + Send + Sync + 'a> ParcodeCollectionPromise<'a, T> { self.node.get(index) } + /// Returns the number of items in the collection. + /// + /// This is an O(1) operation that reads the length from the chunk header. + pub fn len(&self) -> usize { + usize::try_from(self.node.len()).unwrap_or(usize::MAX) + } + + /// Returns true if the collection has no items. + pub fn is_empty(&self) -> bool { + self.node.is_empty() + } + /// Returns a streaming iterator. pub fn iter(&self) -> Result> + 'a> { - self.node.clone().iter() + self.node.clone().iter() // TODO: This must be checked to optimize. } } @@ -162,6 +171,33 @@ where // 4. Read the target item as Lazy T::read_lazy_from_stream(&mut cursor, &mut children_iter) } + + /// Returns the first item as a lazy mirror, or None if empty. + pub fn first(&self) -> Result> { + if self.is_empty() { + Ok(None) + } else { + Ok(Some(self.get_lazy(0)?)) + } + } + + /// Returns the last item as a lazy mirror, or None if empty. + pub fn last(&self) -> Result> { + let len = self.len(); + if len == 0 { + Ok(None) + } else { + Ok(Some(self.get_lazy(len - 1)?)) + } + } + + /// Returns a lazy streaming iterator over the collection. + /// + /// This is highly efficient: it loads shards sequentially and processes items + /// without re-reading bytes. Ideal for scanning large datasets. + pub fn iter_lazy(&self) -> Result> { + ParcodeLazyIterator::new(self.node.clone()) + } } /// A promise for a `HashMap` that supports lazy loading and efficient lookups. @@ -548,6 +584,123 @@ where } } +/// Streaming iterator for lazy collections. +/// +/// This iterator efficiently traverses the shards of a vector, maintaining +/// a cursor position to avoid re-parsing previous items. +#[derive(Debug)] +pub struct ParcodeLazyIterator<'a, T: ParcodeLazyRef<'a>> { + /// The shards (children of the container node). + shards: std::vec::IntoIter>, + + /// Decompressed payload of the active shard. + current_payload: Option>, + /// Current position in the payload buffer (bytes). + current_pos: u64, + /// Iterator over the active shard's children (nested chunks). + current_children: std::vec::IntoIter>, + /// How many items are left to read in this shard. + items_left_in_shard: u64, + + total_items: usize, + items_yielded: usize, + + _marker: PhantomData, +} + +impl<'a, T: ParcodeLazyRef<'a>> ParcodeLazyIterator<'a, T> { + pub fn new(node: ChunkNode<'a>) -> Result { + let total_items = usize::try_from(node.len()).unwrap_or(usize::MAX); + let shards = node.children()?.into_iter(); + + Ok(Self { + shards, + current_payload: None, + current_pos: 0, + current_children: Vec::new().into_iter(), + items_left_in_shard: 0, + total_items, + items_yielded: 0, + _marker: PhantomData, + }) + } + + /// Advances to the next shard if the current one is exhausted. + fn ensure_shard_loaded(&mut self) -> Result { + if self.items_left_in_shard > 0 { + return Ok(true); + } + + // Current shard finished. Load next. + if let Some(shard_node) = self.shards.next() { + // 1. Load Payload + let payload = shard_node.read_raw()?; + + // 2. Load Children + let children = shard_node.children()?; + + // 3. Reset State + // Read length prefix (standard for Parcode slice serialization) + let mut cursor = Cursor::new(payload.as_ref()); + let len: u64 = + bincode::serde::decode_from_std_read(&mut cursor, bincode::config::standard()) + .map_err(|e| crate::ParcodeError::Serialization(e.to_string()))?; + + self.current_pos = cursor.position(); + self.items_left_in_shard = len; + self.current_payload = Some(payload); + self.current_children = children.into_iter(); + + Ok(true) + } else { + Ok(false) // No more shards + } + } +} + +impl<'a, T: ParcodeLazyRef<'a>> Iterator for ParcodeLazyIterator<'a, T> { + type Item = Result; + + fn next(&mut self) -> Option { + // 1. Ensure we have data + match self.ensure_shard_loaded() { + Ok(true) => {} + Ok(false) => return None, // End of iteration + Err(e) => return Some(Err(e)), + } + + // 2. Prepare Cursor at current position + let payload = self + .current_payload + .as_ref() + .expect("ensure_shard_loaded guaranteed payload"); + let mut cursor = Cursor::new(payload.as_ref()); + cursor.set_position(self.current_pos); + + // 3. Deserialize ONE item (Lazy) + // This advances the cursor and the children iterator + let result = T::read_lazy_from_stream(&mut cursor, &mut self.current_children); + + // 4. Update State + self.current_pos = cursor.position(); + self.items_left_in_shard -= 1; + self.items_yielded += 1; + + Some(result) + } + + fn size_hint(&self) -> (usize, Option) { + let remaining = self.total_items - self.items_yielded; + (remaining, Some(remaining)) + } +} + +impl<'a, T: ParcodeLazyRef<'a>> ExactSizeIterator for ParcodeLazyIterator<'a, T> { + fn len(&self) -> usize { + self.total_items - self.items_yielded + } +} + // --- BLANKET IMPLEMENTATIONS FOR PRIMITIVES --- macro_rules! impl_lazy_primitive { @@ -563,10 +716,6 @@ macro_rules! impl_lazy_primitive { _: &mut Cursor<&[u8]>, children: &mut IntoIter>, ) -> Result { - // If a primitive is accessed via read_lazy_from_stream, it MUST be a chunkable field - // (child node), because inline primitives in a Vec are accessed via .get(). - // The macro generates calls to this for fields in `remotes`. - let child_node = children.next().ok_or_else(|| { crate::ParcodeError::Format("Missing child node for chunkable primitive field".into()) })?; @@ -591,10 +740,6 @@ impl<'a, T: ParcodeItem + Send + Sync + 'static> ParcodeLazyRef<'a> for Vec { _: &mut Cursor<&[u8]>, children: &mut IntoIter>, ) -> Result { - // When a Vec is encountered INSIDE another struct's stream (inline), - // it means it's a child node reference. - // We consume one child node from the iterator and wrap it. - let child_node = children.next().ok_or_else(|| { crate::ParcodeError::Format("Missing child node for Vec field".into()) })?; diff --git a/tests/lazy_iterator_tests.rs b/tests/lazy_iterator_tests.rs new file mode 100644 index 0000000..c5ece27 --- /dev/null +++ b/tests/lazy_iterator_tests.rs @@ -0,0 +1,213 @@ +#![allow(missing_docs)] + +use parcode::{Parcode, ParcodeObject}; +use tempfile::NamedTempFile; + +#[derive(ParcodeObject, Debug, PartialEq, Clone, serde::Serialize, serde::Deserialize)] +struct HeavyItem { + id: u32, + #[parcode(chunkable)] + name: String, + #[parcode(chunkable)] + data: Vec, +} + +#[derive(ParcodeObject, Debug, PartialEq, Clone, serde::Serialize, serde::Deserialize)] +struct Root { + #[parcode(chunkable)] + items: Vec, +} + +#[test] +fn test_lazy_iterator_robust() { + // 1. Setup data + let mut items = Vec::new(); + for i in 0u32..100 { + items.push(HeavyItem { + id: i, + name: format!("Item {}", i), + data: vec![u8::try_from(i).expect("i fits in u8"); 10], + }); + } + let root = Root { + items: items.clone(), + }; + + // 2. Serialize to temp file + let temp_file = NamedTempFile::new().expect("Failed to create temp file"); + let path = temp_file.path(); + Parcode::save(path, &root).expect("Serialization failed"); + + // 3. Load + let file = Parcode::open(path).expect("File open failed"); + let loaded_root = file.root::().expect("Lazy load failed"); + let items_promise = &loaded_root.items; + + // 4. Test Collection Promise methods + assert_eq!(items_promise.len(), 100); + assert!(!items_promise.is_empty()); + + // Test first() and last() + let first_lazy = items_promise + .first() + .expect("first() failed") + .expect("first() returned None"); + assert_eq!(first_lazy.id, 0); // Inlined + assert_eq!( + first_lazy.name.load().expect("Failed to load name"), + "Item 0" + ); // Chunkable + assert_eq!(first_lazy.data.len(), 10); // Chunkable collection + + let last_lazy = items_promise + .last() + .expect("last() failed") + .expect("last() returned None"); + assert_eq!(last_lazy.id, 99); + assert_eq!( + last_lazy.name.load().expect("Failed to load name"), + "Item 99" + ); + assert_eq!(last_lazy.data.len(), 10); + + // 5. Test Iterator + let mut iter = items_promise.iter_lazy().expect("iter_lazy() failed"); + + // Initial state + assert_eq!(iter.len(), 100); + let (min, max) = iter.size_hint(); + assert_eq!(min, 100); + assert_eq!(max, Some(100)); + + // Iterate and check size_hint/len at each step + for i in 0..100 { + assert_eq!(iter.len(), 100 - i); + let (min, max) = iter.size_hint(); + assert_eq!(min, 100 - i); + assert_eq!(max, Some(100 - i)); + + let item_lazy = iter + .next() + .expect("Iterator ended prematurely") + .expect("Item load failed"); + assert_eq!(item_lazy.id, u32::try_from(i).expect("i fits in u32")); + assert_eq!( + item_lazy.name.load().expect("Failed to load name"), + format!("Item {}", i) + ); + assert_eq!(item_lazy.data.len(), 10); + } + + // End of iteration + assert_eq!(iter.len(), 0); + assert!(iter.next().is_none()); +} + +#[test] +fn test_empty_collection_lazy() { + let root = Root { items: Vec::new() }; + let temp_file = NamedTempFile::new().expect("Failed to create temp file"); + let path = temp_file.path(); + Parcode::save(path, &root).expect("Serialization failed"); + + let file = Parcode::open(path).expect("File open failed"); + let loaded_root = file.root::().expect("Lazy load failed"); + let items_promise = &loaded_root.items; + + assert_eq!(items_promise.len(), 0); + assert!(items_promise.is_empty()); + assert!(items_promise.first().expect("first() failed").is_none()); + assert!(items_promise.last().expect("last() failed").is_none()); + + let mut iter = items_promise.iter_lazy().expect("iter_lazy() failed"); + assert_eq!(iter.len(), 0); + assert!(iter.next().is_none()); +} + +#[test] +fn test_get_lazy_random_access() { + let mut items = Vec::new(); + for i in 0u32..50 { + items.push(HeavyItem { + id: i, + name: format!("Item {}", i), + data: vec![u8::try_from(i).expect("i fits in u8"); 5], + }); + } + let root = Root { items }; + let temp_file = NamedTempFile::new().expect("Failed to create temp file"); + let path = temp_file.path(); + Parcode::save(path, &root).expect("Serialization failed"); + + let file = Parcode::open(path).expect("File open failed"); + let loaded_root = file.root::().expect("Lazy load failed"); + let items_promise = &loaded_root.items; + + // Random access + let indices = [0, 10, 25, 49]; + for &idx in &indices { + let item_lazy = items_promise.get_lazy(idx).expect("get_lazy failed"); + assert_eq!(item_lazy.id, u32::try_from(idx).expect("idx fits in u32")); + assert_eq!( + item_lazy.name.load().expect("Failed to load name"), + format!("Item {}", idx) + ); + assert_eq!(item_lazy.data.len(), 5); + } +} + +#[test] +fn test_map_lazy_access() { + use std::collections::HashMap; + + #[derive(ParcodeObject, Debug, PartialEq, Clone, serde::Serialize, serde::Deserialize)] + struct Value { + #[parcode(chunkable)] + content: String, + } + + #[derive(ParcodeObject, Debug, PartialEq, Clone, serde::Serialize, serde::Deserialize)] + struct MapRoot { + #[parcode(map)] + data: HashMap, + } + + let mut data = HashMap::new(); + for i in 0..10 { + data.insert( + i, + Value { + content: format!("Value {}", i), + }, + ); + } + let root = MapRoot { data }; + + let temp_file = NamedTempFile::new().expect("Failed to create temp file"); + let path = temp_file.path(); + Parcode::save(path, &root).expect("Serialization failed"); + + let file = Parcode::open(path).expect("File open failed"); + let loaded_root = file.root::().expect("Lazy load failed"); + let map_promise = &loaded_root.data; + + // Test get_lazy on map + for i in 0..10 { + let val_lazy = map_promise + .get_lazy(&i) + .expect("get_lazy failed") + .expect("Value not found"); + assert_eq!( + val_lazy.content.load().expect("Failed to load content"), + format!("Value {}", i) + ); + } + + // Test non-existent key + assert!( + map_promise + .get_lazy(&100) + .expect("get_lazy failed") + .is_none() + ); +}