Db v4 schema layer

db4-storage/src/api/edges.rs

@@ -193,4 +193,7 @@ pub trait EdgeRefOps<'a>: Copy + Clone + Send + Sync {
 
     fn edge_id(&self) -> EID;
     fn edge_ref(self, dir: Dir) -> Option<EdgeRef>;
+
+    /// Graph-wide edge `Meta`
+    fn edge_meta(&self) -> &Arc<Meta>;
 }

db4-storage/src/pages/edge_page/writer.rs

@@ -57,6 +57,14 @@ impl<'a, MP: DerefMut<Target = MemEdgeSegment> + std::fmt::Debug, ES: EdgeSegmen
         layer_id: LayerId,
     ) -> LocalPOS {
         self.graph_stats.update_time(t.t());
+        // Mirror each (layer, prop_id) into the per-layer property presence bitset in Meta.
+        // `.inspect` runs once per emitted item as the iterator is consumed in `insert_edge_internal`
+        // without needing to pass Meta to insert_edge_internal
+        let meta = self.writer.edge_meta().clone();
+        let props = props.into_iter().inspect(move |(id, _)| {
+            meta.temporal_prop_mapper()
+                .mark_prop_in_layer(layer_id, *id);
+        });
         if self
             .writer
             .insert_edge_internal(t, edge_pos, src, dst, layer_id, props)
@@ -132,6 +140,15 @@ impl<'a, MP: DerefMut<Target = MemEdgeSegment> + std::fmt::Debug, ES: EdgeSegmen
             }
         }
 
+        // Mirror each (layer, prop_id) into the per-layer property presence bitset in Meta.
+        // `.inspect` runs once per emitted item as the iterator is consumed in `insert_edge_internal`
+        let meta = self.writer.edge_meta().clone();
+        let t_meta = meta.as_ref();
+        let t_props = t_props.into_iter().inspect(move |(id, _)| {
+            t_meta
+                .temporal_prop_mapper()
+                .mark_prop_in_layer(layer_id, *id);
+        });
         if self
             .writer
             .insert_edge_internal(t, edge_pos, src, dst, layer_id, t_props)
@@ -142,6 +159,9 @@ impl<'a, MP: DerefMut<Target = MemEdgeSegment> + std::fmt::Debug, ES: EdgeSegmen
 
         self.graph_stats.update_time(t.t());
 
+        let c_props = c_props.into_iter().inspect(move |(id, _)| {
+            meta.metadata_mapper().mark_prop_in_layer(layer_id, *id);
+        });
         self.writer
             .update_const_properties(edge_pos, src, dst, layer_id, c_props);
     }
@@ -213,6 +233,12 @@ impl<'a, MP: DerefMut<Target = MemEdgeSegment> + std::fmt::Debug, ES: EdgeSegmen
         if !existing_edge {
             self.increment_layer_num_edges(layer_id);
         }
+        // Mirror each (layer, prop_id) into the per-layer property presence bitset in Meta.
+        // `.inspect` runs once per emitted item as the iterator is consumed (in update_const_properties)
+        let meta = self.writer.edge_meta().clone();
+        let props = props.into_iter().inspect(move |(id, _)| {
+            meta.metadata_mapper().mark_prop_in_layer(layer_id, *id);
+        });
         self.writer
             .update_const_properties(edge_pos, src, dst, layer_id, props);
     }

db4-storage/src/pages/node_page/writer.rs

@@ -155,6 +155,13 @@ impl<'a, MP: DerefMut<Target = MemNodeSegment> + 'a, NS: NodeSegmentOps> NodeWri
         props: impl IntoIterator<Item = (usize, P)>,
     ) {
         self.l_counter.update_time(t.t());
+        // Mirror each (layer, prop_id) into the per-layer property presence bitset in Meta.
+        // `.inspect` runs once per emitted item as the iterator is consumed in add_props
+        let meta = self.mut_segment.node_meta().clone();
+        let props = props.into_iter().inspect(move |(id, _)| {
+            meta.temporal_prop_mapper()
+                .mark_prop_in_layer(layer_id, *id);
+        });
         let (is_new_node, add) = self.mut_segment.add_props(t, pos, layer_id, props);
         self.mut_segment.increment_est_size(add);
         if is_new_node && !self.page.has_node(pos, layer_id) {
@@ -178,6 +185,12 @@ impl<'a, MP: DerefMut<Target = MemNodeSegment> + 'a, NS: NodeSegmentOps> NodeWri
         layer_id: LayerId,
         props: impl IntoIterator<Item = (usize, P)>,
     ) {
+        // Mirror each (layer, prop_id) into the per-layer property presence bitset in Meta.
+        // `.inspect` runs once per emitted item as the iterator is consumed in update_metadata
+        let meta = self.mut_segment.node_meta().clone();
+        let props = props.into_iter().inspect(move |(id, _)| {
+            meta.metadata_mapper().mark_prop_in_layer(layer_id, *id);
+        });
         let (is_new_node, add) = self.mut_segment.update_metadata(pos, layer_id, props);
         self.mut_segment.increment_est_size(add);
         if is_new_node && !self.page.has_node(pos, layer_id) {

db4-storage/src/pages/node_store.rs

@@ -52,6 +52,10 @@ pub struct ReadLockedNodeStorage<NS: NodeSegmentOps<Extension = EXT>, EXT> {
 impl<NS: NodeSegmentOps<Extension = EXT>, EXT: PersistenceStrategy<NS = NS>>
     ReadLockedNodeStorage<NS, EXT>
 {
+    pub fn storage(&self) -> &NodeStorageInner<NS, EXT> {
+        &self.storage
+    }
+
     pub fn node_ref(
         &self,
         node: impl Into<VID>,

db4-storage/src/properties/mod.rs

@@ -23,6 +23,7 @@ use std::sync::Arc;
 
 pub mod props_meta_writer;
 
+// Held by SegmentContainer, which corresponds to one layer in a Mem(Edge/Node)Segment
 #[derive(Debug, Default)]
 pub struct Properties {
     c_properties: Vec<PropColumn>,

db4-storage/src/segments/edge/entry.rs

@@ -5,7 +5,11 @@ use crate::{
     generic_time_ops::{AdditionCellsRef, DeletionCellsRef, WithTimeCells},
     segments::{additions::MemAdditions, edge::segment::MemEdgeSegment},
 };
-use raphtory_api::core::entities::{LayerId, edges::edge_ref::Dir, properties::prop::Prop};
+use raphtory_api::core::entities::{
+    LayerId,
+    edges::edge_ref::Dir,
+    properties::{meta::Meta, prop::Prop},
+};
 use raphtory_core::{
     entities::{
         EID, Multiple, VID,
@@ -14,7 +18,7 @@ use raphtory_core::{
     },
     storage::timeindex::{EventTime, TimeIndexOps},
 };
-use std::marker::PhantomData;
+use std::{marker::PhantomData, sync::Arc};
 
 #[derive(Debug)]
 pub struct MemEdgeEntry<'a, MES> {
@@ -227,4 +231,9 @@ impl<'a> EdgeRefOps<'a> for MemEdgeRef<'a> {
     fn internal_num_layers(self) -> usize {
         self.es.as_ref().len()
     }
+
+    #[inline]
+    fn edge_meta(&self) -> &Arc<Meta> {
+        self.es.edge_meta()
+    }
 }

raphtory-api/src/core/entities/properties/meta.rs

@@ -235,6 +235,18 @@ impl Meta {
             self.temporal_prop_mapper.get_name(prop_id)
         }
     }
+
+    /// O(1) check: has temporal-prop `prop_id` been observed in `layer_id`?
+    #[inline]
+    pub fn temporal_layer_has(&self, layer_id: LayerId, prop_id: usize) -> bool {
+        self.temporal_prop_mapper.layer_has(layer_id, prop_id)
+    }
+
+    /// O(1) check: has metadata-prop `prop_id` been observed in `layer_id`?
+    #[inline]
+    pub fn metadata_layer_has(&self, layer_id: LayerId, prop_id: usize) -> bool {
+        self.metadata_mapper.layer_has(layer_id, prop_id)
+    }
 }
 
 /// Manages the mapping of property names to their IDs and types.
@@ -248,6 +260,10 @@ pub struct PropMapper {
 
     /// Estimated size in bytes of a single row of properties maintained by this mapper.
     row_size: AtomicUsize,
+
+    /// Per-layer property presence bitset; `layer_prop_presence[layer_id][prop_id]`
+    /// is true iff this property has been observed in this layer
+    layer_prop_presence: Arc<RwLock<Vec<Vec<bool>>>>,
 }
 
 impl Debug for PropMapper {
@@ -285,7 +301,31 @@ impl PropMapper {
             id_mapper: DictMapper::new_with_private_fields(fields),
             row_size: AtomicUsize::new(row_size),
             dtypes: Arc::new(RwLock::new(dtypes)),
+            layer_prop_presence: Arc::new(RwLock::new(Vec::new())),
+        }
+    }
+
+    /// O(1) check: has property `prop_id` ever been observed in `layer_id`?
+    /// `false` is authoritative; callers can safely skip column reads for
+    /// this (layer, prop). `true` means at least one entity in `layer_id`
+    /// has prop `prop_id`.
+    #[inline]
+    pub fn layer_has(&self, layer_id: LayerId, prop_id: usize) -> bool {
+        self.layer_prop_presence
+            .read_recursive()
+            .get(layer_id.0)
+            .and_then(|row| row.get(prop_id))
+            .copied()
+            .unwrap_or(false)
+    }
+
+    /// Mark `prop_id` as present in `layer_id`. only takes the write lock once per (layer, prop)
+    pub fn mark_prop_in_layer(&self, layer_id: LayerId, prop_id: usize) {
+        if self.layer_has(layer_id, prop_id) {
+            return;
         }
+        let mut guard = self.layer_prop_presence.write();
+        ensure_and_set(&mut guard, layer_id.0, prop_id);
     }
 
     pub fn d_types(&self) -> impl Deref<Target = Vec<PropType>> + '_ {
@@ -294,10 +334,12 @@ impl PropMapper {
 
     pub fn deep_clone(&self) -> Self {
         let dtypes = self.dtypes.read_recursive().clone();
+        let layer_presence = self.layer_prop_presence.read_recursive().clone();
         Self {
             id_mapper: self.id_mapper.deep_clone(),
             row_size: AtomicUsize::new(self.row_size.load(std::sync::atomic::Ordering::Relaxed)),
             dtypes: Arc::new(RwLock::new(dtypes)),
+            layer_prop_presence: Arc::new(RwLock::new(layer_presence)),
         }
     }
 
@@ -413,10 +455,23 @@ impl PropMapper {
             dict_mapper: self.id_mapper.write(),
             d_types: self.dtypes.write(),
             row_size: &self.row_size,
+            layer_presence: self.layer_prop_presence.write(),
         }
     }
 }
 
+#[inline]
+fn ensure_and_set(presence: &mut Vec<Vec<bool>>, layer_idx: usize, prop_id: usize) {
+    if presence.len() <= layer_idx {
+        presence.resize_with(layer_idx + 1, Vec::new);
+    }
+    let row = &mut presence[layer_idx];
+    if row.len() <= prop_id {
+        row.resize(prop_id + 1, false);
+    }
+    row[prop_id] = true;
+}
+
 /// Write-locked view of a [`PropMapper`].
 pub struct WriteLockedPropMapper<'a> {
     /// Maps property names to their IDs.
@@ -427,6 +482,9 @@ pub struct WriteLockedPropMapper<'a> {
 
     /// Estimated size in bytes of a single row of properties maintained by this mapper.
     row_size: &'a AtomicUsize,
+
+    /// Per-layer property presence bitset
+    layer_presence: RwLockWriteGuard<'a, Vec<Vec<bool>>>,
 }
 
 impl<'a> WriteLockedPropMapper<'a> {
@@ -520,6 +578,11 @@ impl<'a> WriteLockedPropMapper<'a> {
     ) -> Result<Option<Either<usize, usize>>, PropError> {
         fast_proptype_check(self.dict_mapper.map(), &self.d_types, prop, dtype)
     }
+
+    /// Mark `prop_id` as present in `layer_id`
+    pub fn mark_in_layer(&mut self, layer_id: LayerId, prop_id: usize) {
+        ensure_and_set(&mut *self.layer_presence, layer_id.0, prop_id);
+    }
 }
 
 pub struct LockedPropMapper<'a> {

raphtory-graphql/src/data.rs

@@ -13,6 +13,7 @@ use crate::{
             namespaced_item::NamespacedItem,
             vectorised_graph::GqlVectorisedGraph,
         },
+        schema::cache::SchemaCache,
     },
     paths::{
         mark_dirty, ExistingGraphFolder, InternalPathValidationError, PathValidationError,
@@ -753,11 +754,16 @@ impl Data {
         path: &str,
         perm: GraphPermission,
         graph_type: Option<GqlGraphType>,
-    ) -> async_graphql::Result<(UnlockedGraphFolder, DynamicGraph)> {
+    ) -> async_graphql::Result<(UnlockedGraphFolder, DynamicGraph, Option<Arc<SchemaCache>>)> {
         let gwv = self.get_graph(path).await?;
+        // Cache can't be used if we change the graph type; a type override produces a different view
+        let mut cacheable = graph_type.is_none();
         let typed_graph = match graph_type {
             Some(GqlGraphType::Event) => match gwv.graph() {
-                MaterializedGraph::EventGraph(g) => MaterializedGraph::EventGraph(g.clone()),
+                MaterializedGraph::EventGraph(g) => {
+                    cacheable = true;
+                    MaterializedGraph::EventGraph(g.clone())
+                }
                 MaterializedGraph::PersistentGraph(g) => {
                     MaterializedGraph::EventGraph(g.event_graph())
                 }
@@ -767,6 +773,7 @@ impl Data {
                     MaterializedGraph::PersistentGraph(g.persistent_graph())
                 }
                 MaterializedGraph::PersistentGraph(g) => {
+                    cacheable = true;
                     MaterializedGraph::PersistentGraph(g.clone())
                 }
             },
@@ -777,11 +784,14 @@ impl Data {
             filter: Some(ref f),
         } = perm
         {
+            // a redacted view hides keys, and a cached schema could leak redacted information/properties
+            cacheable = false;
             apply_access_filter(raw, f).await?
         } else {
             raw
         };
-        Ok((gwv.folder().clone(), graph))
+        let schema_cache = cacheable.then(|| gwv.schema_cache());
+        Ok((gwv.folder().clone(), graph, schema_cache))
     }
 
     /// For the `graph()` resolver: permission denial → `Ok(None)` (null to client, hides
@@ -791,7 +801,8 @@ impl Data {
         ctx: &Context<'_>,
         path: &str,
         graph_type: Option<GqlGraphType>,
-    ) -> async_graphql::Result<Option<(UnlockedGraphFolder, DynamicGraph)>> {
+    ) -> async_graphql::Result<Option<(UnlockedGraphFolder, DynamicGraph, Option<Arc<SchemaCache>>)>>
+    {
         match require_at_least_read(ctx, &self.auth_policy, path) {
             Ok(perm) => self.load_and_filter(path, perm, graph_type).await.map(Some),
             Err(_) => Ok(None),
@@ -809,7 +820,8 @@ impl Data {
         graph_type: Option<GqlGraphType>,
     ) -> async_graphql::Result<(UnlockedGraphFolder, DynamicGraph)> {
         let perm = require_at_least_read(ctx, &self.auth_policy, path)?;
-        self.load_and_filter(path, perm, graph_type).await
+        let (folder, graph, _cache) = self.load_and_filter(path, perm, graph_type).await?;
+        Ok((folder, graph))
     }
 
     /// Checks read permission then returns the raw `GraphWithVectors` (unfiltered).

raphtory-graphql/src/graph.rs

@@ -1,4 +1,5 @@
 use crate::{
+    model::schema::cache::SchemaCache,
     paths::{ExistingGraphFolder, UnlockedGraphFolder, ValidGraphPaths},
     rayon::blocking_compute,
 };
@@ -46,6 +47,9 @@ pub struct GraphWithVectorsInner {
     pub folder: UnlockedGraphFolder,
     pub is_dirty: AtomicBool,
     pub is_flushing: AtomicBool,
+    /// Cache of computed edge schemas for the unfiltered base view of this graph.
+    /// Cleared on every mutation
+    pub schema_cache: Arc<SchemaCache>,
 }
 
 impl GraphWithVectors {
@@ -60,6 +64,7 @@ impl GraphWithVectors {
             folder: folder.unlock(),
             is_dirty: AtomicBool::new(false),
             is_flushing: AtomicBool::new(false),
+            schema_cache: Arc::new(SchemaCache::new()),
         });
         Self { inner }
     }
@@ -96,6 +101,17 @@ impl GraphWithVectors {
     pub fn folder(&self) -> &UnlockedGraphFolder {
         &self.inner.folder
     }
+
+    /// Handle to this graph's schema cache.
+    pub fn schema_cache(&self) -> Arc<SchemaCache> {
+        self.inner.schema_cache.clone()
+    }
+
+    /// Drop all cached schemas. Called after every mutation.
+    pub fn invalidate_schema_cache(&self) {
+        self.inner.schema_cache.invalidate();
+    }
+
     pub fn set_dirty(&self, is_dirty: bool) {
         self.inner.is_dirty.store(is_dirty, Ordering::Release);
     }