somatize_core/
virtual_value.rs

1//! Virtual values — lazy references to data that can be materialized on demand.
2//!
3//! Instead of eagerly loading all intermediate results, [`VirtualValue`]
4//! keeps references (Materialized, Cached, Deferred, Stream) and only
5//! materializes when a filter actually needs the data.
6
7use crate::cache::CacheKey;
8use crate::schema::Schema;
9use crate::value::Value;
10use serde::{Deserialize, Serialize};
11use std::fmt;
12
13/// A lazy reference to a value that can be materialized on demand.
14///
15/// This is the core of Soma's data virtualization. Instead of computing
16/// and storing every intermediate result, values are represented as
17/// references that carry enough information to:
18///
19/// - Inspect schema without loading data
20/// - Check whether the data is already available
21/// - Compute it when needed
22///
23/// Like Denodo's data virtualization, but for computation rather than SQL.
24#[derive(Debug, Clone, Serialize, Deserialize)]
25#[non_exhaustive]
26pub enum VirtualValue {
27    /// Already computed and in memory. Ready to use.
28    Materialized {
29        /// The concrete value.
30        value: Value,
31        /// Its schema (inferred or declared).
32        schema: Schema,
33    },
34
35    /// Stored in cache (K/V store). Can be loaded on demand.
36    Cached {
37        /// Key to load the value from the cache store.
38        key: CacheKey,
39        /// Schema, known without loading the data.
40        schema: Schema,
41    },
42
43    /// Not computed yet. Carries the "recipe" to produce it:
44    /// which node produces it, and what its cache key would be.
45    Deferred {
46        /// The node whose execution would produce this value.
47        producer_node_id: String,
48        /// Where the value will land once computed.
49        cache_key: CacheKey,
50        /// Expected schema of the eventual value.
51        schema: Schema,
52    },
53
54    /// A stream that materializes chunk by chunk.
55    Stream {
56        /// Identifier of the stream source producing the chunks.
57        source_id: String,
58        /// Schema of each chunk.
59        schema: Schema,
60    },
61}
62
63/// Status of a VirtualValue without inspecting the actual data.
64#[derive(Debug, Clone, Copy, PartialEq, Eq)]
65pub enum ValueStatus {
66    /// In memory, ready to use.
67    InMemory,
68    /// On disk/cache, needs loading.
69    OnDisk,
70    /// Not computed, needs execution.
71    NotComputed,
72    /// Streaming, partial data.
73    Streaming,
74}
75
76impl VirtualValue {
77    /// Create a materialized value with auto-inferred schema.
78    pub fn materialized(value: Value) -> Self {
79        let schema = Self::infer_schema(&value);
80        Self::Materialized { value, schema }
81    }
82
83    /// Create a materialized value with explicit schema.
84    pub fn materialized_with_schema(value: Value, schema: Schema) -> Self {
85        Self::Materialized { value, schema }
86    }
87
88    /// Create a cached reference.
89    pub fn cached(key: CacheKey, schema: Schema) -> Self {
90        Self::Cached { key, schema }
91    }
92
93    /// Create a deferred (not yet computed) reference.
94    pub fn deferred(
95        producer_node_id: impl Into<String>,
96        cache_key: CacheKey,
97        schema: Schema,
98    ) -> Self {
99        Self::Deferred {
100            producer_node_id: producer_node_id.into(),
101            cache_key,
102            schema,
103        }
104    }
105
106    /// Get the schema without materializing the value.
107    pub fn schema(&self) -> &Schema {
108        match self {
109            Self::Materialized { schema, .. }
110            | Self::Cached { schema, .. }
111            | Self::Deferred { schema, .. }
112            | Self::Stream { schema, .. } => schema,
113        }
114    }
115
116    /// Get the current status.
117    pub fn status(&self) -> ValueStatus {
118        match self {
119            Self::Materialized { .. } => ValueStatus::InMemory,
120            Self::Cached { .. } => ValueStatus::OnDisk,
121            Self::Deferred { .. } => ValueStatus::NotComputed,
122            Self::Stream { .. } => ValueStatus::Streaming,
123        }
124    }
125
126    /// Get the materialized value if already in memory.
127    pub fn as_value(&self) -> Option<&Value> {
128        match self {
129            Self::Materialized { value, .. } => Some(value),
130            _ => None,
131        }
132    }
133
134    /// Get the cache key (if this value is cached or deferred).
135    pub fn cache_key(&self) -> Option<&CacheKey> {
136        match self {
137            Self::Cached { key, .. } | Self::Deferred { cache_key: key, .. } => Some(key),
138            _ => None,
139        }
140    }
141
142    /// Materialize from a cache store. Returns the value if found, None if not cached.
143    pub fn try_load(
144        &self,
145        cache: &dyn crate::cache::CacheStore,
146    ) -> crate::error::Result<Option<Value>> {
147        match self {
148            Self::Materialized { value, .. } => Ok(Some(value.clone())),
149            Self::Cached { key, .. } => cache.get(key),
150            Self::Deferred { cache_key, .. } => cache.get(cache_key),
151            _ => Ok(None),
152        }
153    }
154
155    /// Upgrade this reference: if Deferred, check cache; if Cached, load.
156    /// Returns a new VirtualValue that may be closer to Materialized.
157    pub fn resolve(&self, cache: &dyn crate::cache::CacheStore) -> crate::error::Result<Self> {
158        match self {
159            Self::Materialized { .. } => Ok(self.clone()),
160            Self::Cached { key, schema } => {
161                if let Some(value) = cache.get(key)? {
162                    Ok(Self::Materialized {
163                        value,
164                        schema: schema.clone(),
165                    })
166                } else {
167                    Ok(self.clone()) // still cached but value missing
168                }
169            }
170            Self::Deferred {
171                cache_key, schema, ..
172            } => {
173                if let Some(value) = cache.get(cache_key)? {
174                    Ok(Self::Materialized {
175                        value,
176                        schema: schema.clone(),
177                    })
178                } else {
179                    Ok(self.clone()) // still deferred
180                }
181            }
182            _ => Ok(self.clone()),
183        }
184    }
185
186    /// Infer schema from a concrete Value.
187    fn infer_schema(value: &Value) -> Schema {
188        match value {
189            Value::Tensor { values: _, shape } => Schema {
190                dtype: crate::schema::DataType::Float64,
191                shape: Some(
192                    shape
193                        .iter()
194                        .map(|&d| crate::schema::Dimension::Fixed(d))
195                        .collect(),
196                ),
197            },
198            Value::Text(_) => Schema::text(),
199            Value::Json(_) => Schema::json(),
200            Value::Bytes(_) | Value::Object(_) => Schema::bytes(),
201            Value::Empty => Schema::dynamic(crate::schema::DataType::Float64),
202        }
203    }
204}
205
206impl fmt::Display for VirtualValue {
207    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
208        match self {
209            Self::Materialized { schema, value } => {
210                write!(f, "Materialized({schema}, size={})", value.size())
211            }
212            Self::Cached { key, schema } => {
213                write!(f, "Cached({schema}, key={key})")
214            }
215            Self::Deferred {
216                producer_node_id,
217                schema,
218                ..
219            } => {
220                write!(f, "Deferred({schema}, producer={producer_node_id})")
221            }
222            Self::Stream { source_id, schema } => {
223                write!(f, "Stream({schema}, source={source_id})")
224            }
225        }
226    }
227}
228
229impl From<Value> for VirtualValue {
230    fn from(value: Value) -> Self {
231        Self::materialized(value)
232    }
233}
234
235#[cfg(test)]
236mod tests {
237    use super::*;
238    use crate::cache::CacheKey;
239    use crate::schema::{DataType, Schema};
240
241    #[test]
242    fn materialized_from_value() {
243        let val = Value::tensor(vec![1.0, 2.0, 3.0], vec![3]);
244        let vv = VirtualValue::materialized(val.clone());
245
246        assert_eq!(vv.status(), ValueStatus::InMemory);
247        assert_eq!(vv.as_value(), Some(&val));
248        assert_eq!(vv.schema().dtype, DataType::Float64);
249        assert_eq!(vv.schema().rank(), Some(1));
250    }
251
252    #[test]
253    fn cached_reference() {
254        let key = CacheKey::hash_data(b"test");
255        let schema = Schema::vector(DataType::Float64, 100);
256        let vv = VirtualValue::cached(key.clone(), schema.clone());
257
258        assert_eq!(vv.status(), ValueStatus::OnDisk);
259        assert_eq!(vv.cache_key(), Some(&key));
260        assert_eq!(vv.schema(), &schema);
261        assert!(vv.as_value().is_none());
262    }
263
264    #[test]
265    fn deferred_reference() {
266        let key = CacheKey::hash_data(b"deferred");
267        let schema = Schema::batched(DataType::Float64, &[128]);
268        let vv = VirtualValue::deferred("my_node", key.clone(), schema.clone());
269
270        assert_eq!(vv.status(), ValueStatus::NotComputed);
271        assert_eq!(vv.cache_key(), Some(&key));
272        assert_eq!(vv.schema(), &schema);
273    }
274
275    #[test]
276    fn schema_inferred_from_tensor() {
277        let val = Value::tensor(vec![0.0; 12], vec![3, 4]);
278        let vv = VirtualValue::materialized(val);
279        assert_eq!(vv.schema().dtype, DataType::Float64);
280        assert_eq!(vv.schema().rank(), Some(2));
281    }
282
283    #[test]
284    fn schema_inferred_from_json() {
285        let val = Value::json(serde_json::json!({"a": 1}));
286        let vv = VirtualValue::materialized(val);
287        assert_eq!(vv.schema().dtype, DataType::Json);
288    }
289
290    #[test]
291    fn display_formatting() {
292        let vv = VirtualValue::materialized(Value::tensor(vec![1.0], vec![1]));
293        assert!(vv.to_string().contains("Materialized"));
294
295        let vv = VirtualValue::cached(CacheKey::hash_data(b"k"), Schema::json());
296        assert!(vv.to_string().contains("Cached"));
297
298        let vv = VirtualValue::deferred("node_1", CacheKey::hash_data(b"k"), Schema::json());
299        assert!(vv.to_string().contains("Deferred"));
300    }
301
302    #[test]
303    fn from_value_conversion() {
304        let val = Value::tensor(vec![1.0, 2.0], vec![2]);
305        let vv: VirtualValue = val.clone().into();
306        assert_eq!(vv.status(), ValueStatus::InMemory);
307        assert_eq!(vv.as_value(), Some(&val));
308    }
309
310    #[test]
311    fn resolve_materialized_stays_materialized() {
312        use crate::cache::CacheStore;
313
314        // Simple mock cache
315        struct EmptyCache;
316        impl CacheStore for EmptyCache {
317            fn get(&self, _: &CacheKey) -> crate::error::Result<Option<Value>> {
318                Ok(None)
319            }
320            fn put(&self, _: &CacheKey, _: &Value) -> crate::error::Result<()> {
321                Ok(())
322            }
323            fn exists(&self, _: &CacheKey) -> crate::error::Result<bool> {
324                Ok(false)
325            }
326            fn remove(&self, _: &CacheKey) -> crate::error::Result<()> {
327                Ok(())
328            }
329            fn metadata(
330                &self,
331                _: &CacheKey,
332            ) -> crate::error::Result<Option<crate::cache::EntryMeta>> {
333                Ok(None)
334            }
335        }
336
337        let val = Value::tensor(vec![1.0], vec![1]);
338        let vv = VirtualValue::materialized(val);
339        let resolved = vv.resolve(&EmptyCache).unwrap();
340        assert_eq!(resolved.status(), ValueStatus::InMemory);
341    }
342
343    #[test]
344    fn resolve_deferred_checks_cache() {
345        use crate::cache::{CacheStore, EntryMeta};
346        use std::collections::HashMap;
347        use std::sync::Mutex;
348
349        struct TestCache {
350            store: Mutex<HashMap<CacheKey, Value>>,
351        }
352        impl TestCache {
353            fn with(key: CacheKey, value: Value) -> Self {
354                let mut store = HashMap::new();
355                store.insert(key, value);
356                Self {
357                    store: Mutex::new(store),
358                }
359            }
360        }
361        impl CacheStore for TestCache {
362            fn get(&self, key: &CacheKey) -> crate::error::Result<Option<Value>> {
363                Ok(self.store.lock().unwrap().get(key).cloned())
364            }
365            fn put(&self, _: &CacheKey, _: &Value) -> crate::error::Result<()> {
366                Ok(())
367            }
368            fn exists(&self, key: &CacheKey) -> crate::error::Result<bool> {
369                Ok(self.store.lock().unwrap().contains_key(key))
370            }
371            fn remove(&self, _: &CacheKey) -> crate::error::Result<()> {
372                Ok(())
373            }
374            fn metadata(&self, _: &CacheKey) -> crate::error::Result<Option<EntryMeta>> {
375                Ok(None)
376            }
377        }
378
379        let key = CacheKey::hash_data(b"cached_value");
380        let expected = Value::tensor(vec![42.0], vec![1]);
381        let cache = TestCache::with(key.clone(), expected.clone());
382
383        // Deferred → resolve → Materialized (if found in cache)
384        let vv = VirtualValue::deferred("producer", key, Schema::vector(DataType::Float64, 1));
385        assert_eq!(vv.status(), ValueStatus::NotComputed);
386
387        let resolved = vv.resolve(&cache).unwrap();
388        assert_eq!(resolved.status(), ValueStatus::InMemory);
389        assert_eq!(resolved.as_value(), Some(&expected));
390    }
391
392    #[test]
393    fn serde_roundtrip() {
394        let values = vec![
395            VirtualValue::materialized(Value::tensor(vec![1.0], vec![1])),
396            VirtualValue::cached(CacheKey::hash_data(b"k"), Schema::json()),
397            VirtualValue::deferred(
398                "n",
399                CacheKey::hash_data(b"d"),
400                Schema::vector(DataType::Float64, 10),
401            ),
402        ];
403        for vv in values {
404            let json = serde_json::to_string(&vv).unwrap();
405            let deserialized: VirtualValue = serde_json::from_str(&json).unwrap();
406            assert_eq!(vv.status(), deserialized.status());
407            assert_eq!(vv.schema(), deserialized.schema());
408        }
409    }
410}
somatize_core/virtual_value.rs

somatize_core/
virtual_value.rs
