diff --git a/db4-storage/src/api/edges.rs b/db4-storage/src/api/edges.rs index 12b93af1ec..56bd112270 100644 --- a/db4-storage/src/api/edges.rs +++ b/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; + + /// Graph-wide edge `Meta` + fn edge_meta(&self) -> &Arc; } diff --git a/db4-storage/src/pages/edge_page/writer.rs b/db4-storage/src/pages/edge_page/writer.rs index 55d8fa19a6..9c613d7a2f 100644 --- a/db4-storage/src/pages/edge_page/writer.rs +++ b/db4-storage/src/pages/edge_page/writer.rs @@ -57,6 +57,13 @@ impl<'a, MP: DerefMut + std::fmt::Debug, ES: EdgeSegmen layer_id: LayerId, ) -> LocalPOS { self.graph_stats.update_time(t.t()); + // Update 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 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 +139,15 @@ impl<'a, MP: DerefMut + std::fmt::Debug, ES: EdgeSegmen } } + // Update 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 +158,9 @@ impl<'a, MP: DerefMut + 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 +232,12 @@ impl<'a, MP: DerefMut + std::fmt::Debug, ES: EdgeSegmen if !existing_edge { self.increment_layer_num_edges(layer_id); } + // Update 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); } diff --git a/db4-storage/src/pages/node_page/writer.rs b/db4-storage/src/pages/node_page/writer.rs index 4b07c8b634..c79a813541 100644 --- a/db4-storage/src/pages/node_page/writer.rs +++ b/db4-storage/src/pages/node_page/writer.rs @@ -155,6 +155,13 @@ impl<'a, MP: DerefMut + 'a, NS: NodeSegmentOps> NodeWri props: impl IntoIterator, ) { self.l_counter.update_time(t.t()); + // Update 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 + 'a, NS: NodeSegmentOps> NodeWri layer_id: LayerId, props: impl IntoIterator, ) { + // Update 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) { diff --git a/db4-storage/src/pages/node_store.rs b/db4-storage/src/pages/node_store.rs index 8811ce3472..24627d69e4 100644 --- a/db4-storage/src/pages/node_store.rs +++ b/db4-storage/src/pages/node_store.rs @@ -52,6 +52,10 @@ pub struct ReadLockedNodeStorage, EXT> { impl, EXT: PersistenceStrategy> ReadLockedNodeStorage { + pub fn storage(&self) -> &NodeStorageInner { + &self.storage + } + pub fn node_ref( &self, node: impl Into, diff --git a/db4-storage/src/properties/mod.rs b/db4-storage/src/properties/mod.rs index 456d6f199f..75ee01c654 100644 --- a/db4-storage/src/properties/mod.rs +++ b/db4-storage/src/properties/mod.rs @@ -23,6 +23,7 @@ use std::sync::Arc; pub mod props_meta_writer; +/// Contains the properties of a `SegmentContainer` (which corresponds to one layer in a Mem(Edge/Node)Segment) #[derive(Debug, Default)] pub struct Properties { c_properties: Vec, diff --git a/db4-storage/src/segments/edge/entry.rs b/db4-storage/src/segments/edge/entry.rs index ffc604b2d1..cd320a7af0 100644 --- a/db4-storage/src/segments/edge/entry.rs +++ b/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 { + self.es.edge_meta() + } } diff --git a/db4-storage/src/segments/mod.rs b/db4-storage/src/segments/mod.rs index 222c8b2d71..6c08d54b17 100644 --- a/db4-storage/src/segments/mod.rs +++ b/db4-storage/src/segments/mod.rs @@ -153,6 +153,7 @@ impl SparseVec { } } +/// A single layer's data in a Mem(Edge/Node)Segment. #[derive(Debug)] pub struct SegmentContainer { segment_id: usize, diff --git a/docs/reference/graphql/graphql_API.md b/docs/reference/graphql/graphql_API.md index 3f85bd9b5c..1a6b438895 100644 --- a/docs/reference/graphql/graphql_API.md +++ b/docs/reference/graphql/graphql_API.md @@ -1505,61 +1505,6 @@ Composite edge filter (by property, layer, src/dst, etc.). -### EdgeSchema - -Describes edges between a specific pair of node types — the property and -metadata keys seen on such edges, along with their observed value types. -One `EdgeSchema` per `(srcType, dstType)` pair per layer. - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
FieldArgumentTypeDescription
srcTypeString! - -Returns the type of source for these edges - -
dstTypeString! - -Returns the type of destination for these edges - -
properties[PropertySchema!]! - -Returns the list of property schemas for edges connecting these types of nodes - -
metadata[PropertySchema!]! - -Returns the list of metadata schemas for edges connecting these types of nodes - -
- ### EdgeWindowSet A lazy sequence of per-window views of a single edge, produced by @@ -4297,8 +4242,8 @@ Compute the minimum interval between consecutive timestamps. Returns None if few ### LayerSchema -Describes a single edge layer — its name and the per `(srcType, dstType)` -edge schemas observed within it. +Describes a single edge layer: its name and the edge property/metadata keys +present in it, with observed variants (property values) on small graphs @@ -4320,11 +4265,20 @@ Returns the name of the layer with this schema - - + + + + + + + diff --git a/python/tests/test_base_install/test_graphql/test_schema.py b/python/tests/test_base_install/test_graphql/test_schema.py index 6c67fd58b3..7d7ba45cc5 100644 --- a/python/tests/test_base_install/test_graphql/test_schema.py +++ b/python/tests/test_base_install/test_graphql/test_schema.py @@ -21,11 +21,6 @@ def sort_lists_in_structure(key_main, obj): (sort_lists_in_structure(None, item) for item in obj), key=lambda x: x["typeName"], ) - if key_main == "edges": - return sorted( - (sort_lists_in_structure(None, item) for item in obj), - key=lambda x: (x["srcType"], x["dstType"]), - ) if key_main == "properties": return sorted( (sort_lists_in_structure(None, item) for item in obj), @@ -66,19 +61,15 @@ def test_node_edge_properties_schema(): schema { layers { name - edges { - srcType - dstType - properties { - key - propertyType - variants - } - metadata { - key - propertyType - variants - } + properties { + key + propertyType + variants + } + metadata { + key + propertyType + variants } } nodes { @@ -92,7 +83,7 @@ def test_node_edge_properties_schema(): key propertyType variants - } + } } } } @@ -105,135 +96,92 @@ def test_node_edge_properties_schema(): "layers": [ { "name": "0", - "edges": [ + "properties": [ + { + "key": "prop1", + "propertyType": "I64", + "variants": ["1"], + }, + { + "key": "prop2", + "propertyType": "F64", + "variants": ["9.8"], + }, { - "srcType": "a", - "dstType": "None", - "properties": [ - { - "key": "prop1", - "propertyType": "I64", - "variants": ["1"], - }, - { - "key": "prop3", - "propertyType": "Str", - "variants": ["test"], - }, - ], - "metadata": [ - { - "key": "static", - "propertyType": "Str", - "variants": ["test"], - }, - ], + "key": "prop3", + "propertyType": "Str", + "variants": ["test"], }, + ], + "metadata": [ { - "srcType": "None", - "dstType": "b", - "properties": [ - { - "key": "prop1", - "propertyType": "I64", - "variants": ["1"], - }, - { - "key": "prop2", - "propertyType": "F64", - "variants": ["9.8"], - }, - ], - "metadata": [], + "key": "static", + "propertyType": "Str", + "variants": ["test"], }, ], }, { "name": "1", - "edges": [ + "properties": [ { - "srcType": "b", - "dstType": "b", - "properties": [ - { - "key": "prop4", - "propertyType": "I64", - "variants": ["1"], - }, - { - "key": "prop5", - "propertyType": "F64", - "variants": ["9.8"], - }, - { - "key": "prop6", - "propertyType": "Map{ data: Str }", - "variants": ['{"data": "map"}'], - }, - ], - "metadata": [], + "key": "prop1", + "propertyType": "I64", + "variants": ["1"], }, { - "srcType": "b", - "dstType": "None", - "properties": [ - { - "key": "prop1", - "propertyType": "I64", - "variants": ["1"], - }, - { - "key": "propArray", - "propertyType": "List", - "variants": ["[1, 2, 3]"], - }, - { - "key": "prop2", - "propertyType": "F64", - "variants": ["9.8"], - }, - ], - "metadata": [], + "key": "prop2", + "propertyType": "F64", + "variants": ["9.8"], + }, + { + "key": "prop4", + "propertyType": "I64", + "variants": ["1"], + }, + { + "key": "prop5", + "propertyType": "F64", + "variants": ["9.8"], + }, + { + "key": "prop6", + "propertyType": "Map{ data: Str }", + "variants": ['{"data": "map"}'], + }, + { + "key": "propArray", + "propertyType": "List", + "variants": ["[1, 2, 3]"], }, ], + "metadata": [], }, { "name": "2", - "edges": [ + "properties": [ { - "srcType": "None", - "dstType": "None", - "properties": [ - { - "key": "prop1", - "propertyType": "I64", - "variants": ["1"], - }, - { - "key": "prop2", - "propertyType": "F64", - "variants": ["9.8"], - }, - { - "key": "prop3", - "propertyType": "Str", - "variants": ["test"], - }, - ], - "metadata": [], - } + "key": "prop1", + "propertyType": "I64", + "variants": ["1"], + }, + { + "key": "prop2", + "propertyType": "F64", + "variants": ["9.8"], + }, + { + "key": "prop3", + "propertyType": "Str", + "variants": ["test"], + }, ], + "metadata": [], }, { "name": "3", - "edges": [ - { - "srcType": "None", - "dstType": "None", - "properties": [], - "metadata": [], - } - ], + "properties": [], + "metadata": [], }, ], "nodes": [ diff --git a/raphtory-api/src/core/entities/properties/meta.rs b/raphtory-api/src/core/entities/properties/meta.rs index 6b2f644435..c0cbecb4a0 100644 --- a/raphtory-api/src/core/entities/properties/meta.rs +++ b/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>>>, } 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> + '_ { @@ -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>, 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>>, } impl<'a> WriteLockedPropMapper<'a> { @@ -520,6 +578,11 @@ impl<'a> WriteLockedPropMapper<'a> { ) -> Result>, 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> { diff --git a/raphtory-graphql/schema.graphql b/raphtory-graphql/schema.graphql index dfaf911edc..7f63354f72 100644 --- a/raphtory-graphql/schema.graphql +++ b/raphtory-graphql/schema.graphql @@ -720,30 +720,6 @@ input EdgeLayersExpr { expr: EdgeFilter! } -""" -Describes edges between a specific pair of node types — the property and -metadata keys seen on such edges, along with their observed value types. -One `EdgeSchema` per `(srcType, dstType)` pair per layer. -""" -type EdgeSchema { - """ - Returns the type of source for these edges - """ - srcType: String! - """ - Returns the type of destination for these edges - """ - dstType: String! - """ - Returns the list of property schemas for edges connecting these types of nodes - """ - properties: [PropertySchema!]! - """ - Returns the list of metadata schemas for edges connecting these types of nodes - """ - metadata: [PropertySchema!]! -} - input EdgeSortBy { """ Reverse order @@ -2547,8 +2523,8 @@ type Intervals { } """ -Describes a single edge layer — its name and the per `(srcType, dstType)` -edge schemas observed within it. +Describes a single edge layer: its name and the edge property/metadata keys +present in it, with observed variants (property values) on small graphs """ type LayerSchema { """ @@ -2556,9 +2532,13 @@ type LayerSchema { """ name: String! """ - Returns the list of edge schemas for this edge layer + Returns the list of property schemas present on edges in this layer """ - edges: [EdgeSchema!]! + properties: [PropertySchema!]! + """ + Returns the list of metadata schemas present on edges in this layer + """ + metadata: [PropertySchema!]! } """ diff --git a/raphtory-graphql/src/data.rs b/raphtory-graphql/src/data.rs index cb3221798f..071226f16d 100644 --- a/raphtory-graphql/src/data.rs +++ b/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, @@ -751,11 +752,16 @@ impl Data { path: &str, perm: GraphPermission, graph_type: Option, - ) -> async_graphql::Result<(UnlockedGraphFolder, DynamicGraph)> { + ) -> async_graphql::Result<(UnlockedGraphFolder, DynamicGraph, Option>)> { 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()) } @@ -765,6 +771,7 @@ impl Data { MaterializedGraph::PersistentGraph(g.persistent_graph()) } MaterializedGraph::PersistentGraph(g) => { + cacheable = true; MaterializedGraph::PersistentGraph(g.clone()) } }, @@ -775,11 +782,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 @@ -789,7 +799,8 @@ impl Data { ctx: &Context<'_>, path: &str, graph_type: Option, - ) -> async_graphql::Result> { + ) -> async_graphql::Result>)>> + { 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), @@ -807,7 +818,8 @@ impl Data { graph_type: Option, ) -> 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). diff --git a/raphtory-graphql/src/graph.rs b/raphtory-graphql/src/graph.rs index 88d259b170..816e60a154 100644 --- a/raphtory-graphql/src/graph.rs +++ b/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(crate) schema_cache: Arc, } 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(crate) fn schema_cache(&self) -> Arc { + 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); } diff --git a/raphtory-graphql/src/lib.rs b/raphtory-graphql/src/lib.rs index 8d74e5b71a..aaf5087f54 100644 --- a/raphtory-graphql/src/lib.rs +++ b/raphtory-graphql/src/lib.rs @@ -390,17 +390,15 @@ mod graphql_test { graph(path: "graph") { schema { layers { - edges { - properties { - key - propertyType - variants - } - metadata { - key - propertyType - variants - } + properties { + key + propertyType + variants + } + metadata { + key + propertyType + variants } } nodes { @@ -451,7 +449,7 @@ mod graphql_test { } if let Value::Array(mut edge_properties) = - data["graph"]["schema"]["layers"][0]["edges"][0]["properties"].clone() + data["graph"]["schema"]["layers"][0]["properties"].clone() { sort_properties(&mut edge_properties); @@ -461,7 +459,7 @@ mod graphql_test { } if let Value::Array(mut edge_metadata) = - data["graph"]["schema"]["layers"][0]["edges"][0]["metadata"].clone() + data["graph"]["schema"]["layers"][0]["metadata"].clone() { sort_properties(&mut edge_metadata); diff --git a/raphtory-graphql/src/model/graph/graph.rs b/raphtory-graphql/src/model/graph/graph.rs index f6f3cd2a79..59b02c2694 100644 --- a/raphtory-graphql/src/model/graph/graph.rs +++ b/raphtory-graphql/src/model/graph/graph.rs @@ -16,7 +16,7 @@ use crate::{ GqlAlignmentUnit, WindowDuration, }, plugins::graph_algorithm_plugin::GraphAlgorithmPlugin, - schema::graph_schema::GraphSchema, + schema::{cache::SchemaCache, graph_schema::GraphSchema}, }, paths::{ExistingGraphFolder, PathValidationError, UnlockedGraphFolder, ValidGraphPaths}, rayon::blocking_compute, @@ -62,6 +62,9 @@ use std::{ pub(crate) struct GqlGraph { path: UnlockedGraphFolder, graph: DynamicGraph, + /// Shared schema cache, set only for the unfiltered native base graph. + /// Any derived view (window/layer/filter/...) drops it so it recomputes. + schema_cache: Option>, } impl From for GqlGraph { @@ -75,6 +78,21 @@ impl GqlGraph { Self { path, graph: graph.into_dynamic(), + schema_cache: None, + } + } + + /// Carries the base graph's schema cache. Used only for the top-level, unfiltered (e.g. materialized) graph view + /// `cache` is `None` when the view is redacted or type-overridden. + pub fn new_cached( + path: UnlockedGraphFolder, + graph: G, + cache: Option>, + ) -> Self { + Self { + path, + graph: graph.into_dynamic(), + schema_cache: cache, } } @@ -86,6 +104,8 @@ impl GqlGraph { Self { path: self.path.clone(), graph: graph_operation(&self.graph).into_dynamic(), + // a derived view's schema differs from the base, so don't use the cache + schema_cache: None, } } } @@ -631,7 +651,10 @@ impl GqlGraph { /// Returns the graph schema. async fn schema(&self) -> Result { let self_clone = self.clone(); - Ok(blocking_compute(move || GraphSchema::new(&self_clone.graph)).await) + Ok(blocking_compute(move || { + GraphSchema::new(&self_clone.graph, self_clone.schema_cache.clone()) + }) + .await) } /// Access registered graph algorithms (PageRank, shortest path, etc.) for this diff --git a/raphtory-graphql/src/model/graph/mutable_graph.rs b/raphtory-graphql/src/model/graph/mutable_graph.rs index b8d8effa8b..49bd4a5432 100644 --- a/raphtory-graphql/src/model/graph/mutable_graph.rs +++ b/raphtory-graphql/src/model/graph/mutable_graph.rs @@ -563,6 +563,7 @@ impl GqlMutableGraph { /// Post mutation operations. async fn post_mutation_ops(&self) { self.graph.set_dirty(true); + self.graph.invalidate_schema_cache(); } } @@ -689,6 +690,7 @@ impl GqlMutableNode { /// Post mutation operations. async fn post_mutation_ops(&self) { self.node.graph.set_dirty(true); + self.node.graph.invalidate_schema_cache(); } } @@ -850,6 +852,7 @@ impl GqlMutableEdge { /// Post mutation operations. async fn post_mutation_ops(&self) { self.edge.graph.set_dirty(true); + self.edge.graph.invalidate_schema_cache(); } } diff --git a/raphtory-graphql/src/model/mod.rs b/raphtory-graphql/src/model/mod.rs index 82e7788f30..c16346c09a 100644 --- a/raphtory-graphql/src/model/mod.rs +++ b/raphtory-graphql/src/model/mod.rs @@ -192,13 +192,13 @@ impl QueryRoot { ) -> Result> { let data = ctx.data_unchecked::(); // Ok(None) = permission denied (hides existence/access level); Err = load failed. - let Some((folder, graph)) = data + let Some((folder, graph, schema_cache)) = data .get_graph_with_read_permission(ctx, path, graph_type) .await? else { return Ok(None); }; - Ok(Some(GqlGraph::new(folder, graph))) + Ok(Some(GqlGraph::new_cached(folder, graph, schema_cache))) } /// Returns lightweight metadata for a graph (node/edge counts, timestamps) without loading it. diff --git a/raphtory-graphql/src/model/schema/cache.rs b/raphtory-graphql/src/model/schema/cache.rs new file mode 100644 index 0000000000..08218ed3b2 --- /dev/null +++ b/raphtory-graphql/src/model/schema/cache.rs @@ -0,0 +1,93 @@ +use crate::model::schema::property_schema::PropertySchema; +use raphtory_api::core::entities::LayerId; +use std::{collections::HashMap, sync::RwLock}; + +/// Root per-graph schema cache. Lives on the in-memory `GraphWithVectors` and is +/// shared with the base `GqlGraph` view. +#[derive(Default)] +pub(crate) struct SchemaCache { + node: NodeSchemaCache, + layer: LayerSchemaCache, +} + +impl SchemaCache { + pub(crate) fn new() -> Self { + Self::default() + } + + pub(crate) fn node(&self) -> &NodeSchemaCache { + &self.node + } + + pub(crate) fn layer(&self) -> &LayerSchemaCache { + &self.layer + } + + /// Drop all cached schema data. Called on any graph mutation. + pub(crate) fn invalidate(&self) { + self.node.invalidate(); + self.layer.invalidate(); + } +} + +/// Cache of computed node schema results, keyed by node type id. +#[derive(Default)] +pub(crate) struct NodeSchemaCache { + properties: RwLock>>, + metadata: RwLock>>, +} + +impl NodeSchemaCache { + pub(crate) fn get_properties(&self, type_id: usize) -> Option> { + self.properties.read().unwrap().get(&type_id).cloned() + } + + pub(crate) fn store_properties(&self, type_id: usize, value: Vec) { + self.properties.write().unwrap().insert(type_id, value); + } + + pub(crate) fn get_metadata(&self, type_id: usize) -> Option> { + self.metadata.read().unwrap().get(&type_id).cloned() + } + + pub(crate) fn store_metadata(&self, type_id: usize, value: Vec) { + self.metadata.write().unwrap().insert(type_id, value); + } + + fn invalidate(&self) { + self.properties.write().unwrap().clear(); + self.metadata.write().unwrap().clear(); + } +} + +/// Cache of computed layer schema results, keyed by layer id. Caches the +/// resolved property/metadata lists for a layer whether or not variants were +/// collected, so a repeated `schema` query reuses them until the next mutation. +#[derive(Default)] +pub(crate) struct LayerSchemaCache { + properties: RwLock>>, + metadata: RwLock>>, +} + +impl LayerSchemaCache { + pub(crate) fn get_properties(&self, key: &LayerId) -> Option> { + self.properties.read().unwrap().get(key).cloned() + } + + pub(crate) fn store_properties(&self, key: LayerId, value: Vec) { + self.properties.write().unwrap().insert(key, value); + } + + pub(crate) fn get_metadata(&self, key: &LayerId) -> Option> { + self.metadata.read().unwrap().get(key).cloned() + } + + pub(crate) fn store_metadata(&self, key: LayerId, value: Vec) { + self.metadata.write().unwrap().insert(key, value); + } + + fn invalidate(&self) { + self.properties.write().unwrap().clear(); + self.metadata.write().unwrap().clear(); + } +} diff --git a/raphtory-graphql/src/model/schema/edge_schema.rs b/raphtory-graphql/src/model/schema/edge_schema.rs deleted file mode 100644 index 463980db89..0000000000 --- a/raphtory-graphql/src/model/schema/edge_schema.rs +++ /dev/null @@ -1,116 +0,0 @@ -use crate::{ - model::schema::{ - get_node_type, merge_schemas, property_schema::PropertySchema, SchemaAggregate, - }, - rayon::blocking_compute, -}; -use dynamic_graphql::{ResolvedObject, ResolvedObjectFields}; -use itertools::Itertools; -use raphtory::{ - db::{api::view::StaticGraphViewOps, graph::edge::EdgeView}, - prelude::*, -}; -use raphtory_api::core::entities::properties::meta::PropMapper; -use std::collections::HashSet; - -/// Describes edges between a specific pair of node types — the property and -/// metadata keys seen on such edges, along with their observed value types. -/// One `EdgeSchema` per `(srcType, dstType)` pair per layer. -#[derive(Clone, ResolvedObject)] -pub(crate) struct EdgeSchema { - graph: G, - src_type: String, - dst_type: String, -} - -impl EdgeSchema { - pub fn new(graph: G, src_type: String, dst_type: String) -> Self { - Self { - graph, - src_type, - dst_type, - } - } - - fn edges(&self) -> impl Iterator> { - (&&self.graph).edges().into_iter().filter(|&edge| { - let src_type = get_node_type(edge.src()); - let dst_type = get_node_type(edge.dst()); - src_type == self.src_type && dst_type == self.dst_type - }) - } -} - -#[ResolvedObjectFields] -impl EdgeSchema { - /// Returns the type of source for these edges - async fn src_type(&self) -> String { - self.src_type.clone() - } - - /// Returns the type of destination for these edges - async fn dst_type(&self) -> String { - self.dst_type.clone() - } - - /// Returns the list of property schemas for edges connecting these types of nodes - async fn properties(&self) -> Vec { - let cloned = self.clone(); - blocking_compute(move || { - let schema: SchemaAggregate = cloned - .edges() - .map(collect_edge_property_schema) - .reduce(merge_schemas) - .unwrap_or_default(); - schema.into_iter().map(|prop| prop.into()).collect_vec() - }) - .await - } - /// Returns the list of metadata schemas for edges connecting these types of nodes - async fn metadata(&self) -> Vec { - let cloned = self.clone(); - blocking_compute(move || { - let schema: SchemaAggregate = cloned - .edges() - .map(collect_edge_metadata_schema) - .reduce(merge_schemas) - .unwrap_or_default(); - schema.into_iter().map(|prop| prop.into()).collect_vec() - }) - .await - } -} - -fn collect_schema(props: P, mapper: &PropMapper) -> SchemaAggregate { - props - .iter() - .zip(props.ids()) - .filter_map(|((key, value), id)| { - let value = value?; - let key_with_prop_type = ( - key.to_string(), - mapper - .get_dtype(id) - .expect("type for internal id should always exist") - .to_string(), - ); - Some((key_with_prop_type, HashSet::from([value.to_string()]))) - }) - .collect() -} - -fn collect_edge_property_schema<'graph, G: GraphViewOps<'graph>>( - edge: EdgeView, -) -> SchemaAggregate { - let props = edge.properties(); - let mapper = edge.graph.edge_meta().temporal_prop_mapper(); - collect_schema(props, mapper) -} - -fn collect_edge_metadata_schema<'graph, G: GraphViewOps<'graph>>( - edge: EdgeView, -) -> SchemaAggregate { - let props = edge.metadata(); - let mapper = edge.graph.edge_meta().metadata_mapper(); - collect_schema(props, mapper) -} diff --git a/raphtory-graphql/src/model/schema/graph_schema.rs b/raphtory-graphql/src/model/schema/graph_schema.rs index f0c007ae39..7b231268e3 100644 --- a/raphtory-graphql/src/model/schema/graph_schema.rs +++ b/raphtory-graphql/src/model/schema/graph_schema.rs @@ -1,8 +1,11 @@ -use crate::model::schema::{layer_schema::LayerSchema, node_schema::NodeSchema}; +use crate::model::schema::{ + cache::SchemaCache, layer_schema::LayerSchema, node_schema::NodeSchema, +}; use dynamic_graphql::SimpleObject; use itertools::Itertools; use raphtory::{db::api::view::DynamicGraph, prelude::*}; use raphtory_storage::core_ops::CoreGraphOps; +use std::sync::Arc; #[derive(SimpleObject)] pub(crate) struct GraphSchema { @@ -11,15 +14,20 @@ pub(crate) struct GraphSchema { } impl GraphSchema { - pub fn new(graph: &DynamicGraph) -> Self { + /// `cache` is `Some` only for the unfiltered, base graph. + /// Other views (such as filtered, layered, windowed) pass `None` and recompute every time. + pub fn new(graph: &DynamicGraph, cache: Option>) -> Self { let node_types = graph.node_meta().node_type_meta().ids(); let nodes = node_types - .map(|node_type| NodeSchema::new(node_type, graph.clone())) + .map(|node_type| NodeSchema::new(node_type, graph.clone(), cache.clone())) .collect(); let layers = graph .unique_layers() - .map(|layer_name| graph.layers(layer_name).unwrap().into()) + .filter_map(|layer| { + let layer_id = graph.get_layer_id(layer.as_ref())?; + Some(LayerSchema::new(graph.clone(), layer_id, cache.clone())) + }) .collect_vec(); GraphSchema { nodes, layers } diff --git a/raphtory-graphql/src/model/schema/layer_schema.rs b/raphtory-graphql/src/model/schema/layer_schema.rs index 3cfbfa5c3e..ea2475871b 100644 --- a/raphtory-graphql/src/model/schema/layer_schema.rs +++ b/raphtory-graphql/src/model/schema/layer_schema.rs @@ -1,21 +1,35 @@ -use crate::model::schema::{edge_schema::EdgeSchema, get_node_type}; +use crate::{ + model::schema::{ + cache::SchemaCache, property_schema::PropertySchema, ENUM_BOUNDARY, + MAX_DETAILED_SCHEMA_ENTITIES, + }, + rayon::blocking_compute, +}; use dynamic_graphql::{ResolvedObject, ResolvedObjectFields}; -use itertools::Itertools; -use raphtory::{ - db::{api::view::StaticGraphViewOps, graph::views::layer_graph::LayeredGraph}, - prelude::*, +use raphtory::{db::api::view::StaticGraphViewOps, prelude::*}; +use raphtory_api::core::entities::{properties::meta::PropMapper, LayerId, LayerIds}; +use std::{ + collections::{hash_map::Entry, HashMap, HashSet}, + sync::Arc, }; -/// Describes a single edge layer — its name and the per `(srcType, dstType)` -/// edge schemas observed within it. -#[derive(ResolvedObject)] +/// Describes a single edge layer: its name and the edge property/metadata keys +/// present in it, with observed variants (property values) on small graphs +#[derive(Clone, ResolvedObject)] pub(crate) struct LayerSchema { - graph: LayeredGraph, + graph: G, + layer_id: LayerId, + // schema cache for the base graph, `None` for filtered/derived views + cache: Option>, } -impl From> for LayerSchema { - fn from(value: LayeredGraph) -> Self { - Self { graph: value } +impl LayerSchema { + pub fn new(graph: G, layer_id: LayerId, cache: Option>) -> Self { + Self { + graph, + layer_id, + cache, + } } } @@ -23,26 +37,123 @@ impl From> for LayerSchema { impl LayerSchema { /// Returns the name of the layer with this schema async fn name(&self) -> String { - let mut layers = self.graph.unique_layers(); - let layer = layers.next().expect("Layered graph has a layer"); - debug_assert!( - layers.next().is_none(), - "Layered graph outputted more than one layer name" - ); - layer.into() + self.graph.get_layer_name(self.layer_id).to_string() + } + + /// Returns the list of property schemas present on edges in this layer + async fn properties(&self) -> Vec { + if let Some(cache) = &self.cache { + if let Some(hit) = cache.layer().get_properties(&self.layer_id) { + return hit; + } + } + let graph = self.graph.clone(); + let layer_id = self.layer_id; + let result = blocking_compute(move || { + if graph.unfiltered_num_edges(&LayerIds::One(layer_id)) > MAX_DETAILED_SCHEMA_ENTITIES { + // too many edges to collect values: keys/types only, no variants + return collect_layer_schema(&graph, layer_id, false); + } + let layer = graph.get_layer_name(layer_id); + let layered = graph.valid_layers(layer); + let mapper = graph.edge_meta().temporal_prop_mapper(); + collect_variants(layered.edges().into_iter().map(|e| e.properties()), mapper) + }) + .await; + if let Some(cache) = &self.cache { + cache + .layer() + .store_properties(self.layer_id, result.clone()); + } + result + } + + /// Returns the list of metadata schemas present on edges in this layer + async fn metadata(&self) -> Vec { + if let Some(cache) = &self.cache { + if let Some(hit) = cache.layer().get_metadata(&self.layer_id) { + return hit; + } + } + let graph = self.graph.clone(); + let layer_id = self.layer_id; + let result = blocking_compute(move || { + if graph.unfiltered_num_edges(&LayerIds::One(layer_id)) > MAX_DETAILED_SCHEMA_ENTITIES { + return collect_layer_schema(&graph, layer_id, true); + } + let layer = graph.get_layer_name(layer_id); + let layered = graph.valid_layers(layer); + let mapper = graph.edge_meta().metadata_mapper(); + collect_variants(layered.edges().into_iter().map(|e| e.metadata()), mapper) + }) + .await; + if let Some(cache) = &self.cache { + cache.layer().store_metadata(self.layer_id, result.clone()); + } + result } - /// Returns the list of edge schemas for this edge layer - async fn edges(&self) -> Vec>> { - self.graph - .edges() - .into_iter() - .map(|edge| { - let src_type = get_node_type(edge.src()); - let dst_type = get_node_type(edge.dst()); - (src_type, dst_type) - }) - .unique() - .map(|(src_type, dst_type)| EdgeSchema::new(self.graph.clone(), src_type, dst_type)) - .collect_vec() +} + +/// Get edge property/metadata keys and types using bitset without collecting values. +/// Redacted properties are handled by the GraphView (in edge_layer_has_*). +fn collect_layer_schema( + graph: &G, + layer_id: LayerId, + metadata: bool, +) -> Vec { + let meta = graph.edge_meta(); + let mapper = if metadata { + meta.metadata_mapper() + } else { + meta.temporal_prop_mapper() + }; + mapper + .locked() + .iter_ids_and_types() + .filter(|(id, _, _)| { + if metadata { + graph.edge_layer_has_metadata(layer_id, *id) + } else { + graph.edge_layer_has_temporal_prop(layer_id, *id) + } + }) + .map(|(_, name, dtype)| PropertySchema::new(name.to_string(), dtype.to_string(), vec![])) + .collect() +} + +/// Collect distinct property values for `(key, dtype)` pairs. If there are too many values, we stop. +fn collect_variants( + props_per_edge: impl Iterator, + mapper: &PropMapper, +) -> Vec { + let mut schema: HashMap<(String, String), HashSet> = HashMap::new(); + for props in props_per_edge { + for ((key, value), id) in props.iter().zip(props.ids()) { + let Some(value) = value else { continue }; + let key_with_prop_type = ( + key.to_string(), + mapper + .get_dtype(id) + .expect("type for internal id should always exist") + .to_string(), + ); + match schema.entry(key_with_prop_type) { + Entry::Vacant(entry) => { + entry.insert(HashSet::from([value.to_string()])); + } + Entry::Occupied(mut entry) => { + let variants = entry.get_mut(); + // An empty set means "too many variants", so we skip + // Otherwise, there should always be at least 1 value in the set + if !variants.is_empty() { + variants.insert(value.to_string()); + if variants.len() > ENUM_BOUNDARY { + variants.clear(); + } + } + } + } + } } + schema.into_iter().map(|prop| prop.into()).collect() } diff --git a/raphtory-graphql/src/model/schema/mod.rs b/raphtory-graphql/src/model/schema/mod.rs index 877fb8d1c7..4910769977 100644 --- a/raphtory-graphql/src/model/schema/mod.rs +++ b/raphtory-graphql/src/model/schema/mod.rs @@ -1,42 +1,15 @@ -use raphtory::{ - db::graph::node::NodeView, - prelude::{GraphViewOps, NodeViewOps}, -}; -use rustc_hash::FxHashMap; -use std::collections::HashSet; - -pub(crate) mod edge_schema; +pub(crate) mod cache; pub(crate) mod graph_schema; pub(crate) mod layer_schema; pub(crate) mod node_schema; pub(crate) mod property_schema; +/// Maximum number of distinct values collected per property key. More than that and we don't report any. const ENUM_BOUNDARY: usize = 20; -const DEFAULT_NODE_TYPE: &'static str = "None"; - -fn get_node_type<'graph, G: GraphViewOps<'graph>>(node: NodeView<'graph, G>) -> String { - match node.node_type() { - None => "None".into(), - Some(n) => n.to_string(), - } -} +/// Above this many entities (nodes graph-wide for `NodeSchema`, edges in the +/// layer for `LayerSchema`), schema resolvers skip collecting property values +/// (variants) and only return keys and types +const MAX_DETAILED_SCHEMA_ENTITIES: usize = 1000; -type SchemaAggregate = FxHashMap<(String, String), HashSet>; - -fn merge_schemas(mut s1: SchemaAggregate, s2: SchemaAggregate) -> SchemaAggregate { - for ((key, prop_type), set2) in s2 { - if let Some(set1) = s1.get_mut(&(key.clone(), prop_type.clone())) { - // Here, an empty set means: too many values to be interpreted as an enumerated type - if set1.len() > 0 && set2.len() > 0 { - set1.extend(set2); - } - if set1.len() > ENUM_BOUNDARY { - set1.clear(); - } - } else { - s1.insert((key, prop_type), set2); - } - } - s1 -} +const DEFAULT_NODE_TYPE: &'static str = "None"; diff --git a/raphtory-graphql/src/model/schema/node_schema.rs b/raphtory-graphql/src/model/schema/node_schema.rs index 6f8e5f61c3..c555d6809e 100644 --- a/raphtory-graphql/src/model/schema/node_schema.rs +++ b/raphtory-graphql/src/model/schema/node_schema.rs @@ -1,4 +1,7 @@ -use crate::model::schema::{property_schema::PropertySchema, DEFAULT_NODE_TYPE}; +use crate::model::schema::{ + cache::SchemaCache, property_schema::PropertySchema, DEFAULT_NODE_TYPE, + MAX_DETAILED_SCHEMA_ENTITIES, +}; use dynamic_graphql::{ResolvedObject, ResolvedObjectFields}; use raphtory::{ db::{ @@ -14,21 +17,26 @@ use raphtory::{ use raphtory_api::core::entities::LayerIds; use raphtory_storage::core_ops::CoreGraphOps; use rayon::prelude::*; +use std::sync::Arc; /// Describes nodes of a specific type in a graph — its property keys and /// observed value types (and, for string-valued properties, the set of /// distinct values seen). One `NodeSchema` per node type. -#[derive(ResolvedObject)] +#[derive(Clone, ResolvedObject)] pub(crate) struct NodeSchema { - pub(crate) type_id: usize, + /// node type id; also the schema-cache lookup key + type_id: usize, graph: DynamicGraph, + // schema cache for the base graph, `None` for filtered/derived views + cache: Option>, } impl NodeSchema { - pub fn new(node_type: usize, graph: DynamicGraph) -> Self { + pub fn new(node_type: usize, graph: DynamicGraph, cache: Option>) -> Self { Self { type_id: node_type, graph, + cache, } } } @@ -45,13 +53,33 @@ impl NodeSchema { /// ever set on a node of this type, with its observed `PropertyType` and (for /// string-valued properties) the set of distinct values. async fn properties(&self) -> Vec { - self.properties_inner() + if let Some(cache) = &self.cache { + if let Some(hit) = cache.node().get_properties(self.type_id) { + return hit; + } + } + // not found in cache, so calculate it and cache it + let result = self.properties_inner(); + if let Some(cache) = &self.cache { + cache.node().store_properties(self.type_id, result.clone()); + } + result } /// Metadata schemas seen on nodes of this type — like `properties`, but /// covering metadata fields rather than temporal properties. async fn metadata(&self) -> Vec { - self.metadata_inner() + if let Some(cache) = &self.cache { + if let Some(hit) = cache.node().get_metadata(self.type_id) { + return hit; + } + } + // not found in cache, so calculate it and cache it + let result = self.metadata_inner(); + if let Some(cache) = &self.cache { + cache.node().store_metadata(self.type_id, result.clone()); + } + result } } @@ -76,7 +104,7 @@ impl NodeSchema { .map(|(_, name, dtype)| (name.to_string(), dtype.to_string())) .unzip(); - if self.graph.unfiltered_num_nodes(&LayerIds::All) > 1000 { + if self.graph.unfiltered_num_nodes(&LayerIds::All) > MAX_DETAILED_SCHEMA_ENTITIES { // large graph, do not collect detailed schema as it is expensive keys.into_iter() .zip(property_types) @@ -126,7 +154,7 @@ impl NodeSchema { .map(|(_, name, dtype)| (name.to_string(), dtype.to_string())) .unzip(); - if self.graph.unfiltered_num_nodes(&LayerIds::All) > 1000 { + if self.graph.unfiltered_num_nodes(&LayerIds::All) > MAX_DETAILED_SCHEMA_ENTITIES { // large graph, do not collect detailed schema as it is expensive keys.into_iter() .zip(property_types) @@ -222,7 +250,7 @@ mod test { } fn check_schema(g: &Graph) { - let gs = GraphSchema::new(&g.clone().into_dynamic()); + let gs = GraphSchema::new(&g.clone().into_dynamic(), None); let mut actual: Vec<(String, Vec, Vec)> = gs .nodes diff --git a/raphtory-graphql/src/model/schema/property_schema.rs b/raphtory-graphql/src/model/schema/property_schema.rs index 21e61bc697..7f5d311730 100644 --- a/raphtory-graphql/src/model/schema/property_schema.rs +++ b/raphtory-graphql/src/model/schema/property_schema.rs @@ -1,6 +1,6 @@ use dynamic_graphql::SimpleObject; -#[derive(SimpleObject, Debug, PartialEq, PartialOrd, Ord, Eq)] +#[derive(SimpleObject, Clone, Debug, PartialEq, PartialOrd, Ord, Eq)] pub(crate) struct PropertySchema { key: String, property_type: String, diff --git a/raphtory-storage/src/graph/edges/edge_storage_ops.rs b/raphtory-storage/src/graph/edges/edge_storage_ops.rs index c6b9cc7cea..2ecd2559c6 100644 --- a/raphtory-storage/src/graph/edges/edge_storage_ops.rs +++ b/raphtory-storage/src/graph/edges/edge_storage_ops.rs @@ -237,4 +237,28 @@ impl<'a> EdgeStorageOps<'a> for storage::EdgeEntryRef<'a> { fn metadata_layer(self, layer_id: LayerId, prop_id: usize) -> Option { EdgeRefOps::c_prop(self, layer_id, prop_id) } + + // Layer-skip override: drop layers where we know the property isn't present + fn temporal_prop_iter( + self, + layer_ids: &'a LayerIds, + prop_id: usize, + ) -> impl Iterator)> + 'a { + let meta = EdgeRefOps::edge_meta(&self).clone(); + self.layer_ids_iter(layer_ids) + .filter(move |&layer_id| meta.temporal_layer_has(layer_id, prop_id)) + .map(move |id| (id, EdgeStorageOps::temporal_prop_layer(self, id, prop_id))) + } + + // Layer-skip override: same shape as `temporal_prop_iter` but against the metadata mapper + fn metadata_iter( + self, + layer_ids: &'a LayerIds, + prop_id: usize, + ) -> impl Iterator + 'a { + let meta = EdgeRefOps::edge_meta(&self).clone(); + self.layer_ids_iter(layer_ids) + .filter(move |&layer_id| meta.metadata_layer_has(layer_id, prop_id)) + .filter_map(move |id| Some((id, EdgeStorageOps::metadata_layer(self, id, prop_id)?))) + } } diff --git a/raphtory-storage/src/graph/edges/edges.rs b/raphtory-storage/src/graph/edges/edges.rs index 1d96f8792f..a1f0004da7 100644 --- a/raphtory-storage/src/graph/edges/edges.rs +++ b/raphtory-storage/src/graph/edges/edges.rs @@ -1,6 +1,8 @@ use super::{edge_entry::EdgeStorageEntry, unlocked::UnlockedEdges}; use either::Either; -use raphtory_api::core::entities::{properties::meta::STATIC_GRAPH_LAYER_ID, LayerIds, EID}; +use raphtory_api::core::entities::{ + properties::meta::STATIC_GRAPH_LAYER_ID, LayerId, LayerIds, EID, +}; use raphtory_core::entities::edges::edge_ref::EdgeRef; use rayon::iter::ParallelIterator; use std::sync::Arc; @@ -145,4 +147,30 @@ impl<'a> EdgesStorageRef<'a> { EdgesStorageRef::Unlocked(storage) => storage.storage().num_edges(), } } + + /// O(1) check: has property `prop_id` ever been observed in `layer_id`? + /// `false` is authoritative i.e. callers can skip column reads for `(layer_id, prop_id)`. + pub fn layer_has_temporal_prop(&self, layer_id: LayerId, prop_id: usize) -> bool { + let inner = match self { + EdgesStorageRef::Mem(storage) => storage.storage(), + EdgesStorageRef::Unlocked(storage) => storage.storage(), + }; + inner + .edge_meta() + .temporal_prop_mapper() + .layer_has(layer_id, prop_id) + } + + /// O(1) check: has metadata `prop_id` ever been observed in `layer_id`? + /// `false` is authoritative i.e. callers can skip column reads for `(layer_id, prop_id)`. + pub fn layer_has_metadata(&self, layer_id: LayerId, prop_id: usize) -> bool { + let inner = match self { + EdgesStorageRef::Mem(storage) => storage.storage(), + EdgesStorageRef::Unlocked(storage) => storage.storage(), + }; + inner + .edge_meta() + .metadata_mapper() + .layer_has(layer_id, prop_id) + } } diff --git a/raphtory-storage/src/graph/mod.rs b/raphtory-storage/src/graph/mod.rs index 13af057137..2c8788edff 100644 --- a/raphtory-storage/src/graph/mod.rs +++ b/raphtory-storage/src/graph/mod.rs @@ -3,3 +3,36 @@ pub mod graph; pub mod locked; pub mod nodes; pub mod variants; + +use raphtory_api::core::entities::{properties::meta::STATIC_GRAPH_LAYER_ID, LayerId, LayerIds}; +use raphtory_core::entities::LayerVariants; +use storage::utils::Iter3; + +/// Build an iterator over all layer ids we should visit when reading +/// node-style temporal data: `STATIC_GRAPH_LAYER_ID` is always included +/// (unlayered entities must be visible in every view), plus the layers +/// requested in `layer_ids`. +pub fn layer_ids_with_static( + num_layers: usize, + layer_ids: &LayerIds, +) -> impl Iterator + Send + Sync + 'static { + match layer_ids { + LayerIds::None => LayerVariants::None(std::iter::once(STATIC_GRAPH_LAYER_ID)), + LayerIds::All => LayerVariants::All((0..num_layers).map(LayerId)), + LayerIds::One(id) => { + if *id == STATIC_GRAPH_LAYER_ID { + LayerVariants::One(std::iter::once(*id)) + } else { + LayerVariants::Multiple(Iter3::I([STATIC_GRAPH_LAYER_ID, *id].into_iter())) + } + } + LayerIds::Multiple(ids) => { + if ids.contains(STATIC_GRAPH_LAYER_ID) { + LayerVariants::Multiple(Iter3::J(ids.clone().into_iter())) + } else { + let v = std::iter::once(STATIC_GRAPH_LAYER_ID).chain(ids.clone().into_iter()); + LayerVariants::Multiple(Iter3::K(v)) + } + } + } +} diff --git a/raphtory-storage/src/graph/nodes/node_storage_ops.rs b/raphtory-storage/src/graph/nodes/node_storage_ops.rs index 589dc368dd..be4183c608 100644 --- a/raphtory-storage/src/graph/nodes/node_storage_ops.rs +++ b/raphtory-storage/src/graph/nodes/node_storage_ops.rs @@ -1,3 +1,4 @@ +use crate::graph::layer_ids_with_static; use raphtory_api::core::{ entities::{ edges::edge_ref::EdgeRef, @@ -12,7 +13,7 @@ use raphtory_api::core::{ }; use raphtory_core::{entities::LayerVariants, storage::timeindex::EventTime}; use std::{borrow::Cow, ops::Range, sync::Arc}; -use storage::{api::nodes::NodeRefOps, generic_time_ops::LayerIter, utils::Iter3, NodeEntryRef}; +use storage::{api::nodes::NodeRefOps, generic_time_ops::LayerIter, NodeEntryRef}; pub trait NodeStorageOps<'a>: Copy + Sized + Send + Sync + 'a { fn degree(self, layers: &LayerIds, dir: Direction) -> usize; @@ -63,7 +64,7 @@ pub trait NodeStorageOps<'a>: Copy + Sized + Send + Sync + 'a { prop_id: usize, ) -> impl Iterator)> + 'a { self.layer_ids_iter(layer_ids) - .map(move |id| (id, self.temporal_prop_layer(id, prop_id))) + .map(move |id| (id, NodeStorageOps::temporal_prop_layer(self, id, prop_id))) } fn tprop(self, prop_id: usize) -> storage::NodeTProps<'a>; @@ -73,33 +74,14 @@ pub trait NodeStorageOps<'a>: Copy + Sized + Send + Sync + 'a { /// Iterate over `NodeTProps` for each layer specified by `layer_ids`, always /// including `STATIC_GRAPH_LAYER_ID` (the layer for nodes added without an - /// explicit layer name). This mirrors the behaviour of `layer_ids_with_static` - /// used for node additions: unlayered nodes must be visible in every view. + /// explicit layer name). Unlayered nodes must be visible in every view. fn tprop_iter_layers( self, layer_ids: &LayerIds, prop_id: usize, ) -> impl Iterator> + Send + Sync + 'a { - let layers = match layer_ids { - LayerIds::None => LayerVariants::None(std::iter::once(STATIC_GRAPH_LAYER_ID)), - LayerIds::All => LayerVariants::All((0..self.num_layers()).map(LayerId)), - LayerIds::One(id) => { - if *id == STATIC_GRAPH_LAYER_ID { - LayerVariants::One(std::iter::once(*id)) - } else { - LayerVariants::Multiple(Iter3::I([STATIC_GRAPH_LAYER_ID, *id].into_iter())) - } - } - LayerIds::Multiple(ids) => { - if ids.contains(STATIC_GRAPH_LAYER_ID) { - LayerVariants::Multiple(Iter3::J(ids.clone().into_iter())) - } else { - let v = std::iter::once(STATIC_GRAPH_LAYER_ID).chain(ids.clone().into_iter()); - LayerVariants::Multiple(Iter3::K(v)) - } - } - }; - layers.map(move |id| self.temporal_prop_layer(id, prop_id)) + layer_ids_with_static(self.num_layers(), layer_ids) + .map(move |id| self.temporal_prop_layer(id, prop_id)) } fn constant_prop_layer(self, layer_id: LayerId, prop_id: usize) -> Option; @@ -109,8 +91,9 @@ pub trait NodeStorageOps<'a>: Copy + Sized + Send + Sync + 'a { layer_ids: &'a LayerIds, prop_id: usize, ) -> impl Iterator + 'a { - self.layer_ids_iter(layer_ids) - .filter_map(move |id| Some((id, self.constant_prop_layer(id, prop_id)?))) + self.layer_ids_iter(layer_ids).filter_map(move |id| { + Some((id, NodeStorageOps::constant_prop_layer(self, id, prop_id)?)) + }) } fn temp_prop_rows_range( @@ -206,6 +189,53 @@ impl<'a> NodeStorageOps<'a> for NodeEntryRef<'a> { NodeRefOps::c_prop(self, layer_id, prop_id) } + // Layer-skip override: drop layers whose per-layer property presence bitset in + // Meta says `prop_id` has never been written + fn temporal_prop_iter( + self, + layer_ids: &'a LayerIds, + prop_id: usize, + ) -> impl Iterator)> + 'a { + let meta = NodeRefOps::node_meta(&self).clone(); + self.layer_ids_iter(layer_ids) + .filter(move |&layer_id| meta.temporal_layer_has(layer_id, prop_id)) + .map(move |layer_id| { + ( + layer_id, + NodeStorageOps::temporal_prop_layer(self, layer_id, prop_id), + ) + }) + } + + // Layer-skip override: same shape as `temporal_prop_iter` but for the metadata mapper. + fn constant_prop_iter( + self, + layer_ids: &'a LayerIds, + prop_id: usize, + ) -> impl Iterator + 'a { + let meta = NodeRefOps::node_meta(&self).clone(); + self.layer_ids_iter(layer_ids) + .filter(move |&layer_id| meta.metadata_layer_has(layer_id, prop_id)) + .filter_map(move |layer_id| { + Some(( + layer_id, + NodeStorageOps::constant_prop_layer(self, layer_id, prop_id)?, + )) + }) + } + + // Layer-skip override: drops layers that the per-layer bitset says don't carry `prop_id`. + fn tprop_iter_layers( + self, + layer_ids: &LayerIds, + prop_id: usize, + ) -> impl Iterator> + Send + Sync + 'a { + let meta = NodeRefOps::node_meta(&self).clone(); + layer_ids_with_static(self.num_layers(), layer_ids) + .filter(move |&id| meta.temporal_layer_has(id, prop_id)) + .map(move |id| NodeStorageOps::temporal_prop_layer(self, id, prop_id)) + } + fn temp_prop_rows_range( self, w: Option>, diff --git a/raphtory-storage/src/graph/nodes/nodes_ref.rs b/raphtory-storage/src/graph/nodes/nodes_ref.rs index f170f8dafd..ef211e0a9c 100644 --- a/raphtory-storage/src/graph/nodes/nodes_ref.rs +++ b/raphtory-storage/src/graph/nodes/nodes_ref.rs @@ -1,6 +1,6 @@ use super::node_ref::NodeStorageRef; use crate::graph::variants::storage_variants3::StorageVariants3; -use raphtory_api::core::entities::VID; +use raphtory_api::core::entities::{LayerId, VID}; use rayon::iter::ParallelIterator; use storage::{Extension, ReadLockedNodes}; @@ -45,6 +45,30 @@ impl<'a> NodesStorageEntry<'a> { for_all_variants!(self, nodes => nodes.iter()) } + /// O(1) check + pub fn layer_has_temporal_prop(&self, layer_id: LayerId, prop_id: usize) -> bool { + let inner = match self { + NodesStorageEntry::Mem(nodes) => nodes.storage(), + NodesStorageEntry::Unlocked(nodes) => nodes.storage(), + }; + inner + .prop_meta() + .temporal_prop_mapper() + .layer_has(layer_id, prop_id) + } + + /// O(1) check + pub fn layer_has_metadata(&self, layer_id: LayerId, prop_id: usize) -> bool { + let inner = match self { + NodesStorageEntry::Mem(nodes) => nodes.storage(), + NodesStorageEntry::Unlocked(nodes) => nodes.storage(), + }; + inner + .prop_meta() + .metadata_mapper() + .layer_has(layer_id, prop_id) + } + /// Returns a parallel iterator over nodes row groups /// the (usize) part is the row group not the segment pub fn row_groups_par_iter( diff --git a/raphtory/src/db/api/properties/internal.rs b/raphtory/src/db/api/properties/internal.rs index 9f61e55c86..b763d71fa5 100644 --- a/raphtory/src/db/api/properties/internal.rs +++ b/raphtory/src/db/api/properties/internal.rs @@ -1,7 +1,10 @@ use crate::db::api::view::BoxedLIter; use raphtory_api::{ core::{ - entities::properties::prop::{Prop, PropType}, + entities::{ + properties::prop::{Prop, PropType}, + LayerId, + }, storage::{arc_str::ArcStr, timeindex::EventTime}, }, inherit::Base, @@ -20,6 +23,10 @@ pub trait NodePropertySchemaOps: Send + Sync { fn node_visible_metadata_id(&self, name: &str) -> Option; /// Returns `None` if `id` is not visible in this view (e.g. redacted). fn node_visible_metadata_name(&self, id: usize) -> Option; + /// O(1) check: is temporal-prop `prop_id` present on any node in `layer_id`? + fn node_layer_has_temporal_prop(&self, layer_id: LayerId, prop_id: usize) -> bool; + /// O(1) check: is metadata-prop `prop_id` present on any node in `layer_id`? + fn node_layer_has_metadata(&self, layer_id: LayerId, prop_id: usize) -> bool; } /// Same as `NodePropertySchemaOps` but for edge properties. @@ -32,6 +39,10 @@ pub trait EdgePropertySchemaOps: Send + Sync { fn edge_visible_metadata_id(&self, name: &str) -> Option; /// Returns `None` if `id` is not visible in this view (e.g. redacted). fn edge_visible_metadata_name(&self, id: usize) -> Option; + /// O(1) check: is temporal-prop `prop_id` present on any edge in `layer_id`? + fn edge_layer_has_temporal_prop(&self, layer_id: LayerId, prop_id: usize) -> bool; + /// O(1) check: is metadata-prop `prop_id` present on any edge in `layer_id`? + fn edge_layer_has_metadata(&self, layer_id: LayerId, prop_id: usize) -> bool; } /// Marker: delegate `NodePropertySchemaOps` through `Base`. @@ -67,6 +78,14 @@ where fn node_visible_metadata_name(&self, id: usize) -> Option { self.base().node_visible_metadata_name(id) } + #[inline] + fn node_layer_has_temporal_prop(&self, layer_id: LayerId, prop_id: usize) -> bool { + self.base().node_layer_has_temporal_prop(layer_id, prop_id) + } + #[inline] + fn node_layer_has_metadata(&self, layer_id: LayerId, prop_id: usize) -> bool { + self.base().node_layer_has_metadata(layer_id, prop_id) + } } impl EdgePropertySchemaOps for G @@ -97,6 +116,14 @@ where fn edge_visible_metadata_name(&self, id: usize) -> Option { self.base().edge_visible_metadata_name(id) } + #[inline] + fn edge_layer_has_temporal_prop(&self, layer_id: LayerId, prop_id: usize) -> bool { + self.base().edge_layer_has_temporal_prop(layer_id, prop_id) + } + #[inline] + fn edge_layer_has_metadata(&self, layer_id: LayerId, prop_id: usize) -> bool { + self.base().edge_layer_has_metadata(layer_id, prop_id) + } } pub trait InternalTemporalPropertyViewOps { diff --git a/raphtory/src/db/api/storage/graph/storage_ops/property_schema.rs b/raphtory/src/db/api/storage/graph/storage_ops/property_schema.rs index 94f44793bc..3919f8cd75 100644 --- a/raphtory/src/db/api/storage/graph/storage_ops/property_schema.rs +++ b/raphtory/src/db/api/storage/graph/storage_ops/property_schema.rs @@ -3,7 +3,10 @@ use crate::db::api::{ properties::internal::{EdgePropertySchemaOps, NodePropertySchemaOps}, view::BoxedLIter, }; -use raphtory_api::{core::storage::arc_str::ArcStr, iter::IntoDynBoxed}; +use raphtory_api::{ + core::{entities::LayerId, storage::arc_str::ArcStr}, + iter::IntoDynBoxed, +}; impl NodePropertySchemaOps for GraphStorage { fn node_visible_temporal_prop_ids(&self) -> BoxedLIter<'_, usize> { @@ -27,6 +30,12 @@ impl NodePropertySchemaOps for GraphStorage { fn node_visible_metadata_name(&self, id: usize) -> Option { Some(self.node_meta().metadata_mapper().get_name(id).clone()) } + fn node_layer_has_temporal_prop(&self, layer_id: LayerId, prop_id: usize) -> bool { + self.nodes().layer_has_temporal_prop(layer_id, prop_id) + } + fn node_layer_has_metadata(&self, layer_id: LayerId, prop_id: usize) -> bool { + self.nodes().layer_has_metadata(layer_id, prop_id) + } } impl EdgePropertySchemaOps for GraphStorage { @@ -51,4 +60,10 @@ impl EdgePropertySchemaOps for GraphStorage { fn edge_visible_metadata_name(&self, id: usize) -> Option { Some(self.edge_meta().metadata_mapper().get_name(id).clone()) } + fn edge_layer_has_temporal_prop(&self, layer_id: LayerId, prop_id: usize) -> bool { + self.edges().layer_has_temporal_prop(layer_id, prop_id) + } + fn edge_layer_has_metadata(&self, layer_id: LayerId, prop_id: usize) -> bool { + self.edges().layer_has_metadata(layer_id, prop_id) + } } diff --git a/raphtory/src/db/api/view/graph.rs b/raphtory/src/db/api/view/graph.rs index e424b4f969..2b0cf39ed3 100644 --- a/raphtory/src/db/api/view/graph.rs +++ b/raphtory/src/db/api/view/graph.rs @@ -409,7 +409,6 @@ pub fn materialize_impl( let stream_capacity = 10; let (tx, rx) = crossbeam_channel::bounded::(stream_capacity); - // let mut scope_result = Ok(()); // Use std::thread::scope rather than rayon::scope so the producer runs on its own OS thread. // With rayon::scope on a single-thread pool, the main thread blocking on rx.recv() would starve the spawned producer. std::thread::scope(|scope| { diff --git a/raphtory/src/db/api/view/internal/time_semantics/filtered_edge.rs b/raphtory/src/db/api/view/internal/time_semantics/filtered_edge.rs index a97d4e4fa5..3ad8bd6704 100644 --- a/raphtory/src/db/api/view/internal/time_semantics/filtered_edge.rs +++ b/raphtory/src/db/api/view/internal/time_semantics/filtered_edge.rs @@ -421,7 +421,10 @@ impl<'a> FilteredEdgeStorageOps<'a> for EdgeEntryRef<'a> { view: G, layer_ids: &'a LayerIds, ) -> impl Iterator)> + 'a { + let prune_view = view.clone(); self.filtered_layer_ids_iter(view.clone(), layer_ids) + // skip layers we know don't have any edges with the property + .filter(move |&layer_id| prune_view.edge_layer_has_temporal_prop(layer_id, prop_id)) .map(move |layer_id| { ( layer_id, @@ -439,7 +442,10 @@ impl<'a> FilteredEdgeStorageOps<'a> for EdgeEntryRef<'a> { let layer_ids = view.layer_ids(); let mut values = self .metadata_iter(layer_ids, prop_id) - .filter(|(layer, _)| layer_filter(*layer)); + // skip layers we know don't have any edge with the property + .filter(|(layer, _)| { + view.edge_layer_has_metadata(*layer, prop_id) && layer_filter(*layer) + }); if view.num_layers() > 1 { let mut values = values.peekable(); if values.peek().is_some() { diff --git a/raphtory/src/db/graph/views/property_redacted_graph.rs b/raphtory/src/db/graph/views/property_redacted_graph.rs index c06aa001d7..80e58e21e6 100644 --- a/raphtory/src/db/graph/views/property_redacted_graph.rs +++ b/raphtory/src/db/graph/views/property_redacted_graph.rs @@ -13,7 +13,10 @@ use crate::db::api::{ }, }; use raphtory_api::{ - core::{entities::properties::prop::Prop, storage::arc_str::ArcStr}, + core::{ + entities::{properties::prop::Prop, LayerId}, + storage::arc_str::ArcStr, + }, inherit::Base, }; use raphtory_storage::layer_ops::InheritLayerOps; @@ -247,6 +250,16 @@ impl NodePropertySchemaOps for PropertyRedactedGraph { } self.graph.node_visible_metadata_name(id) } + + fn node_layer_has_temporal_prop(&self, layer_id: LayerId, prop_id: usize) -> bool { + is_visible(&self.redaction.node_props_visible, prop_id) + && self.graph.node_layer_has_temporal_prop(layer_id, prop_id) + } + + fn node_layer_has_metadata(&self, layer_id: LayerId, prop_id: usize) -> bool { + is_visible(&self.redaction.node_meta_visible, prop_id) + && self.graph.node_layer_has_metadata(layer_id, prop_id) + } } impl EdgePropertySchemaOps for PropertyRedactedGraph { @@ -289,6 +302,16 @@ impl EdgePropertySchemaOps for PropertyRedactedGraph { } self.graph.edge_visible_metadata_name(id) } + + fn edge_layer_has_temporal_prop(&self, layer_id: LayerId, prop_id: usize) -> bool { + is_visible(&self.redaction.edge_props_visible, prop_id) + && self.graph.edge_layer_has_temporal_prop(layer_id, prop_id) + } + + fn edge_layer_has_metadata(&self, layer_id: LayerId, prop_id: usize) -> bool { + is_visible(&self.redaction.edge_meta_visible, prop_id) + && self.graph.edge_layer_has_metadata(layer_id, prop_id) + } } // Graph-level property redaction: override InternalTemporalPropertiesOps and InternalMetadataOps
edges[EdgeSchema!]!properties[PropertySchema!]! + +Returns the list of property schemas present on edges in this layer + +
metadata[PropertySchema!]! -Returns the list of edge schemas for this edge layer +Returns the list of metadata schemas present on edges in this layer