1use std::path::Path;
6use std::sync::Arc;
7
8use parking_lot::RwLock;
9
10use grafeo_adapters::storage::wal::{WalConfig, WalManager, WalRecord, WalRecovery};
11use grafeo_common::memory::buffer::{BufferManager, BufferManagerConfig};
12use grafeo_common::utils::error::Result;
13use grafeo_core::graph::lpg::LpgStore;
14#[cfg(feature = "rdf")]
15use grafeo_core::graph::rdf::RdfStore;
16
17use crate::config::Config;
18use crate::query::cache::QueryCache;
19use crate::session::Session;
20use crate::transaction::TransactionManager;
21
22pub struct GrafeoDB {
45 config: Config,
47 store: Arc<LpgStore>,
49 #[cfg(feature = "rdf")]
51 rdf_store: Arc<RdfStore>,
52 tx_manager: Arc<TransactionManager>,
54 buffer_manager: Arc<BufferManager>,
56 wal: Option<Arc<WalManager>>,
58 query_cache: Arc<QueryCache>,
60 is_open: RwLock<bool>,
62}
63
64impl GrafeoDB {
65 #[must_use]
81 pub fn new_in_memory() -> Self {
82 Self::with_config(Config::in_memory()).expect("In-memory database creation should not fail")
83 }
84
85 pub fn open(path: impl AsRef<Path>) -> Result<Self> {
104 Self::with_config(Config::persistent(path.as_ref()))
105 }
106
107 pub fn with_config(config: Config) -> Result<Self> {
131 let store = Arc::new(LpgStore::new());
132 #[cfg(feature = "rdf")]
133 let rdf_store = Arc::new(RdfStore::new());
134 let tx_manager = Arc::new(TransactionManager::new());
135
136 let buffer_config = BufferManagerConfig {
138 budget: config.memory_limit.unwrap_or_else(|| {
139 (BufferManagerConfig::detect_system_memory() as f64 * 0.75) as usize
140 }),
141 spill_path: config
142 .spill_path
143 .clone()
144 .or_else(|| config.path.as_ref().map(|p| p.join("spill"))),
145 ..BufferManagerConfig::default()
146 };
147 let buffer_manager = BufferManager::new(buffer_config);
148
149 let wal = if config.wal_enabled {
151 if let Some(ref db_path) = config.path {
152 std::fs::create_dir_all(db_path)?;
154
155 let wal_path = db_path.join("wal");
156
157 if wal_path.exists() {
159 let recovery = WalRecovery::new(&wal_path);
160 let records = recovery.recover()?;
161 Self::apply_wal_records(&store, &records)?;
162 }
163
164 let wal_config = WalConfig::default();
166 let wal_manager = WalManager::with_config(&wal_path, wal_config)?;
167 Some(Arc::new(wal_manager))
168 } else {
169 None
170 }
171 } else {
172 None
173 };
174
175 let query_cache = Arc::new(QueryCache::default());
177
178 Ok(Self {
179 config,
180 store,
181 #[cfg(feature = "rdf")]
182 rdf_store,
183 tx_manager,
184 buffer_manager,
185 wal,
186 query_cache,
187 is_open: RwLock::new(true),
188 })
189 }
190
191 fn apply_wal_records(store: &LpgStore, records: &[WalRecord]) -> Result<()> {
193 for record in records {
194 match record {
195 WalRecord::CreateNode { id, labels } => {
196 let label_refs: Vec<&str> = labels.iter().map(|s| s.as_str()).collect();
197 store.create_node_with_id(*id, &label_refs);
198 }
199 WalRecord::DeleteNode { id } => {
200 store.delete_node(*id);
201 }
202 WalRecord::CreateEdge {
203 id,
204 src,
205 dst,
206 edge_type,
207 } => {
208 store.create_edge_with_id(*id, *src, *dst, edge_type);
209 }
210 WalRecord::DeleteEdge { id } => {
211 store.delete_edge(*id);
212 }
213 WalRecord::SetNodeProperty { id, key, value } => {
214 store.set_node_property(*id, key, value.clone());
215 }
216 WalRecord::SetEdgeProperty { id, key, value } => {
217 store.set_edge_property(*id, key, value.clone());
218 }
219 WalRecord::AddNodeLabel { id, label } => {
220 store.add_label(*id, label);
221 }
222 WalRecord::RemoveNodeLabel { id, label } => {
223 store.remove_label(*id, label);
224 }
225 WalRecord::TxCommit { .. }
226 | WalRecord::TxAbort { .. }
227 | WalRecord::Checkpoint { .. } => {
228 }
231 }
232 }
233 Ok(())
234 }
235
236 #[must_use]
255 pub fn session(&self) -> Session {
256 #[cfg(feature = "rdf")]
257 {
258 Session::with_rdf_store_and_adaptive(
259 Arc::clone(&self.store),
260 Arc::clone(&self.rdf_store),
261 Arc::clone(&self.tx_manager),
262 Arc::clone(&self.query_cache),
263 self.config.adaptive.clone(),
264 self.config.factorized_execution,
265 )
266 }
267 #[cfg(not(feature = "rdf"))]
268 {
269 Session::with_adaptive(
270 Arc::clone(&self.store),
271 Arc::clone(&self.tx_manager),
272 Arc::clone(&self.query_cache),
273 self.config.adaptive.clone(),
274 self.config.factorized_execution,
275 )
276 }
277 }
278
279 #[must_use]
281 pub fn adaptive_config(&self) -> &crate::config::AdaptiveConfig {
282 &self.config.adaptive
283 }
284
285 pub fn execute(&self, query: &str) -> Result<QueryResult> {
295 let session = self.session();
296 session.execute(query)
297 }
298
299 pub fn execute_with_params(
305 &self,
306 query: &str,
307 params: std::collections::HashMap<String, grafeo_common::types::Value>,
308 ) -> Result<QueryResult> {
309 let session = self.session();
310 session.execute_with_params(query, params)
311 }
312
313 #[cfg(feature = "cypher")]
319 pub fn execute_cypher(&self, query: &str) -> Result<QueryResult> {
320 let session = self.session();
321 session.execute_cypher(query)
322 }
323
324 #[cfg(feature = "cypher")]
330 pub fn execute_cypher_with_params(
331 &self,
332 query: &str,
333 params: std::collections::HashMap<String, grafeo_common::types::Value>,
334 ) -> Result<QueryResult> {
335 use crate::query::processor::{QueryLanguage, QueryProcessor};
336
337 let processor = QueryProcessor::for_lpg(Arc::clone(&self.store));
339 processor.process(query, QueryLanguage::Cypher, Some(¶ms))
340 }
341
342 #[cfg(feature = "gremlin")]
348 pub fn execute_gremlin(&self, query: &str) -> Result<QueryResult> {
349 let session = self.session();
350 session.execute_gremlin(query)
351 }
352
353 #[cfg(feature = "gremlin")]
359 pub fn execute_gremlin_with_params(
360 &self,
361 query: &str,
362 params: std::collections::HashMap<String, grafeo_common::types::Value>,
363 ) -> Result<QueryResult> {
364 let session = self.session();
365 session.execute_gremlin_with_params(query, params)
366 }
367
368 #[cfg(feature = "graphql")]
374 pub fn execute_graphql(&self, query: &str) -> Result<QueryResult> {
375 let session = self.session();
376 session.execute_graphql(query)
377 }
378
379 #[cfg(feature = "graphql")]
385 pub fn execute_graphql_with_params(
386 &self,
387 query: &str,
388 params: std::collections::HashMap<String, grafeo_common::types::Value>,
389 ) -> Result<QueryResult> {
390 let session = self.session();
391 session.execute_graphql_with_params(query, params)
392 }
393
394 #[cfg(all(feature = "sparql", feature = "rdf"))]
411 pub fn execute_sparql(&self, query: &str) -> Result<QueryResult> {
412 use crate::query::{
413 Executor, optimizer::Optimizer, planner_rdf::RdfPlanner, sparql_translator,
414 };
415
416 let logical_plan = sparql_translator::translate(query)?;
418
419 let optimizer = Optimizer::new();
421 let optimized_plan = optimizer.optimize(logical_plan)?;
422
423 let planner = RdfPlanner::new(Arc::clone(&self.rdf_store));
425 let mut physical_plan = planner.plan(&optimized_plan)?;
426
427 let executor = Executor::with_columns(physical_plan.columns.clone());
429 executor.execute(physical_plan.operator.as_mut())
430 }
431
432 #[cfg(feature = "rdf")]
436 #[must_use]
437 pub fn rdf_store(&self) -> &Arc<RdfStore> {
438 &self.rdf_store
439 }
440
441 pub fn query_scalar<T: FromValue>(&self, query: &str) -> Result<T> {
447 let result = self.execute(query)?;
448 result.scalar()
449 }
450
451 #[must_use]
453 pub fn config(&self) -> &Config {
454 &self.config
455 }
456
457 #[must_use]
461 pub fn store(&self) -> &Arc<LpgStore> {
462 &self.store
463 }
464
465 #[must_use]
467 pub fn buffer_manager(&self) -> &Arc<BufferManager> {
468 &self.buffer_manager
469 }
470
471 pub fn close(&self) -> Result<()> {
481 let mut is_open = self.is_open.write();
482 if !*is_open {
483 return Ok(());
484 }
485
486 if let Some(ref wal) = self.wal {
488 let epoch = self.store.current_epoch();
489
490 let checkpoint_tx = self.tx_manager.last_assigned_tx_id().unwrap_or_else(|| {
492 self.tx_manager.begin()
494 });
495
496 wal.log(&WalRecord::TxCommit {
498 tx_id: checkpoint_tx,
499 })?;
500
501 wal.checkpoint(checkpoint_tx, epoch)?;
503 wal.sync()?;
504 }
505
506 *is_open = false;
507 Ok(())
508 }
509
510 #[must_use]
512 pub fn wal(&self) -> Option<&Arc<WalManager>> {
513 self.wal.as_ref()
514 }
515
516 fn log_wal(&self, record: &WalRecord) -> Result<()> {
518 if let Some(ref wal) = self.wal {
519 wal.log(record)?;
520 }
521 Ok(())
522 }
523
524 #[must_use]
526 pub fn node_count(&self) -> usize {
527 self.store.node_count()
528 }
529
530 #[must_use]
532 pub fn edge_count(&self) -> usize {
533 self.store.edge_count()
534 }
535
536 #[must_use]
538 pub fn label_count(&self) -> usize {
539 self.store.label_count()
540 }
541
542 #[must_use]
544 pub fn property_key_count(&self) -> usize {
545 self.store.property_key_count()
546 }
547
548 #[must_use]
550 pub fn edge_type_count(&self) -> usize {
551 self.store.edge_type_count()
552 }
553
554 pub fn create_node(&self, labels: &[&str]) -> grafeo_common::types::NodeId {
571 let id = self.store.create_node(labels);
572
573 if let Err(e) = self.log_wal(&WalRecord::CreateNode {
575 id,
576 labels: labels.iter().map(|s| s.to_string()).collect(),
577 }) {
578 tracing::warn!("Failed to log CreateNode to WAL: {}", e);
579 }
580
581 id
582 }
583
584 pub fn create_node_with_props(
588 &self,
589 labels: &[&str],
590 properties: impl IntoIterator<
591 Item = (
592 impl Into<grafeo_common::types::PropertyKey>,
593 impl Into<grafeo_common::types::Value>,
594 ),
595 >,
596 ) -> grafeo_common::types::NodeId {
597 let props: Vec<(
599 grafeo_common::types::PropertyKey,
600 grafeo_common::types::Value,
601 )> = properties
602 .into_iter()
603 .map(|(k, v)| (k.into(), v.into()))
604 .collect();
605
606 let id = self
607 .store
608 .create_node_with_props(labels, props.iter().map(|(k, v)| (k.clone(), v.clone())));
609
610 if let Err(e) = self.log_wal(&WalRecord::CreateNode {
612 id,
613 labels: labels.iter().map(|s| s.to_string()).collect(),
614 }) {
615 tracing::warn!("Failed to log CreateNode to WAL: {}", e);
616 }
617
618 for (key, value) in props {
620 if let Err(e) = self.log_wal(&WalRecord::SetNodeProperty {
621 id,
622 key: key.to_string(),
623 value,
624 }) {
625 tracing::warn!("Failed to log SetNodeProperty to WAL: {}", e);
626 }
627 }
628
629 id
630 }
631
632 #[must_use]
634 pub fn get_node(
635 &self,
636 id: grafeo_common::types::NodeId,
637 ) -> Option<grafeo_core::graph::lpg::Node> {
638 self.store.get_node(id)
639 }
640
641 pub fn delete_node(&self, id: grafeo_common::types::NodeId) -> bool {
645 let result = self.store.delete_node(id);
646
647 if result {
648 if let Err(e) = self.log_wal(&WalRecord::DeleteNode { id }) {
649 tracing::warn!("Failed to log DeleteNode to WAL: {}", e);
650 }
651 }
652
653 result
654 }
655
656 pub fn set_node_property(
660 &self,
661 id: grafeo_common::types::NodeId,
662 key: &str,
663 value: grafeo_common::types::Value,
664 ) {
665 if let Err(e) = self.log_wal(&WalRecord::SetNodeProperty {
667 id,
668 key: key.to_string(),
669 value: value.clone(),
670 }) {
671 tracing::warn!("Failed to log SetNodeProperty to WAL: {}", e);
672 }
673
674 self.store.set_node_property(id, key, value);
675 }
676
677 pub fn add_node_label(&self, id: grafeo_common::types::NodeId, label: &str) -> bool {
695 let result = self.store.add_label(id, label);
696
697 if result {
698 if let Err(e) = self.log_wal(&WalRecord::AddNodeLabel {
700 id,
701 label: label.to_string(),
702 }) {
703 tracing::warn!("Failed to log AddNodeLabel to WAL: {}", e);
704 }
705 }
706
707 result
708 }
709
710 pub fn remove_node_label(&self, id: grafeo_common::types::NodeId, label: &str) -> bool {
728 let result = self.store.remove_label(id, label);
729
730 if result {
731 if let Err(e) = self.log_wal(&WalRecord::RemoveNodeLabel {
733 id,
734 label: label.to_string(),
735 }) {
736 tracing::warn!("Failed to log RemoveNodeLabel to WAL: {}", e);
737 }
738 }
739
740 result
741 }
742
743 #[must_use]
760 pub fn get_node_labels(&self, id: grafeo_common::types::NodeId) -> Option<Vec<String>> {
761 self.store
762 .get_node(id)
763 .map(|node| node.labels.iter().map(|s| s.to_string()).collect())
764 }
765
766 pub fn create_edge(
786 &self,
787 src: grafeo_common::types::NodeId,
788 dst: grafeo_common::types::NodeId,
789 edge_type: &str,
790 ) -> grafeo_common::types::EdgeId {
791 let id = self.store.create_edge(src, dst, edge_type);
792
793 if let Err(e) = self.log_wal(&WalRecord::CreateEdge {
795 id,
796 src,
797 dst,
798 edge_type: edge_type.to_string(),
799 }) {
800 tracing::warn!("Failed to log CreateEdge to WAL: {}", e);
801 }
802
803 id
804 }
805
806 pub fn create_edge_with_props(
810 &self,
811 src: grafeo_common::types::NodeId,
812 dst: grafeo_common::types::NodeId,
813 edge_type: &str,
814 properties: impl IntoIterator<
815 Item = (
816 impl Into<grafeo_common::types::PropertyKey>,
817 impl Into<grafeo_common::types::Value>,
818 ),
819 >,
820 ) -> grafeo_common::types::EdgeId {
821 let props: Vec<(
823 grafeo_common::types::PropertyKey,
824 grafeo_common::types::Value,
825 )> = properties
826 .into_iter()
827 .map(|(k, v)| (k.into(), v.into()))
828 .collect();
829
830 let id = self.store.create_edge_with_props(
831 src,
832 dst,
833 edge_type,
834 props.iter().map(|(k, v)| (k.clone(), v.clone())),
835 );
836
837 if let Err(e) = self.log_wal(&WalRecord::CreateEdge {
839 id,
840 src,
841 dst,
842 edge_type: edge_type.to_string(),
843 }) {
844 tracing::warn!("Failed to log CreateEdge to WAL: {}", e);
845 }
846
847 for (key, value) in props {
849 if let Err(e) = self.log_wal(&WalRecord::SetEdgeProperty {
850 id,
851 key: key.to_string(),
852 value,
853 }) {
854 tracing::warn!("Failed to log SetEdgeProperty to WAL: {}", e);
855 }
856 }
857
858 id
859 }
860
861 #[must_use]
863 pub fn get_edge(
864 &self,
865 id: grafeo_common::types::EdgeId,
866 ) -> Option<grafeo_core::graph::lpg::Edge> {
867 self.store.get_edge(id)
868 }
869
870 pub fn delete_edge(&self, id: grafeo_common::types::EdgeId) -> bool {
874 let result = self.store.delete_edge(id);
875
876 if result {
877 if let Err(e) = self.log_wal(&WalRecord::DeleteEdge { id }) {
878 tracing::warn!("Failed to log DeleteEdge to WAL: {}", e);
879 }
880 }
881
882 result
883 }
884
885 pub fn set_edge_property(
889 &self,
890 id: grafeo_common::types::EdgeId,
891 key: &str,
892 value: grafeo_common::types::Value,
893 ) {
894 if let Err(e) = self.log_wal(&WalRecord::SetEdgeProperty {
896 id,
897 key: key.to_string(),
898 value: value.clone(),
899 }) {
900 tracing::warn!("Failed to log SetEdgeProperty to WAL: {}", e);
901 }
902 self.store.set_edge_property(id, key, value);
903 }
904
905 pub fn remove_node_property(&self, id: grafeo_common::types::NodeId, key: &str) -> bool {
909 self.store.remove_node_property(id, key).is_some()
911 }
912
913 pub fn remove_edge_property(&self, id: grafeo_common::types::EdgeId, key: &str) -> bool {
917 self.store.remove_edge_property(id, key).is_some()
919 }
920
921 pub fn create_property_index(&self, property: &str) {
941 self.store.create_property_index(property);
942 }
943
944 pub fn drop_property_index(&self, property: &str) -> bool {
948 self.store.drop_property_index(property)
949 }
950
951 #[must_use]
953 pub fn has_property_index(&self, property: &str) -> bool {
954 self.store.has_property_index(property)
955 }
956
957 #[must_use]
972 pub fn find_nodes_by_property(
973 &self,
974 property: &str,
975 value: &grafeo_common::types::Value,
976 ) -> Vec<grafeo_common::types::NodeId> {
977 self.store.find_nodes_by_property(property, value)
978 }
979
980 #[must_use]
988 pub fn is_persistent(&self) -> bool {
989 self.config.path.is_some()
990 }
991
992 #[must_use]
996 pub fn path(&self) -> Option<&Path> {
997 self.config.path.as_deref()
998 }
999
1000 #[must_use]
1004 pub fn info(&self) -> crate::admin::DatabaseInfo {
1005 crate::admin::DatabaseInfo {
1006 mode: crate::admin::DatabaseMode::Lpg,
1007 node_count: self.store.node_count(),
1008 edge_count: self.store.edge_count(),
1009 is_persistent: self.is_persistent(),
1010 path: self.config.path.clone(),
1011 wal_enabled: self.config.wal_enabled,
1012 version: env!("CARGO_PKG_VERSION").to_string(),
1013 }
1014 }
1015
1016 #[must_use]
1020 pub fn detailed_stats(&self) -> crate::admin::DatabaseStats {
1021 let disk_bytes = self.config.path.as_ref().and_then(|p| {
1022 if p.exists() {
1023 Self::calculate_disk_usage(p).ok()
1024 } else {
1025 None
1026 }
1027 });
1028
1029 crate::admin::DatabaseStats {
1030 node_count: self.store.node_count(),
1031 edge_count: self.store.edge_count(),
1032 label_count: self.store.label_count(),
1033 edge_type_count: self.store.edge_type_count(),
1034 property_key_count: self.store.property_key_count(),
1035 index_count: 0, memory_bytes: self.buffer_manager.allocated(),
1037 disk_bytes,
1038 }
1039 }
1040
1041 fn calculate_disk_usage(path: &Path) -> Result<usize> {
1043 let mut total = 0usize;
1044 if path.is_dir() {
1045 for entry in std::fs::read_dir(path)? {
1046 let entry = entry?;
1047 let metadata = entry.metadata()?;
1048 if metadata.is_file() {
1049 total += metadata.len() as usize;
1050 } else if metadata.is_dir() {
1051 total += Self::calculate_disk_usage(&entry.path())?;
1052 }
1053 }
1054 }
1055 Ok(total)
1056 }
1057
1058 #[must_use]
1063 pub fn schema(&self) -> crate::admin::SchemaInfo {
1064 let labels = self
1065 .store
1066 .all_labels()
1067 .into_iter()
1068 .map(|name| crate::admin::LabelInfo {
1069 name: name.clone(),
1070 count: self.store.nodes_with_label(&name).count(),
1071 })
1072 .collect();
1073
1074 let edge_types = self
1075 .store
1076 .all_edge_types()
1077 .into_iter()
1078 .map(|name| crate::admin::EdgeTypeInfo {
1079 name: name.clone(),
1080 count: self.store.edges_with_type(&name).count(),
1081 })
1082 .collect();
1083
1084 let property_keys = self.store.all_property_keys();
1085
1086 crate::admin::SchemaInfo::Lpg(crate::admin::LpgSchemaInfo {
1087 labels,
1088 edge_types,
1089 property_keys,
1090 })
1091 }
1092
1093 #[cfg(feature = "rdf")]
1097 #[must_use]
1098 pub fn rdf_schema(&self) -> crate::admin::SchemaInfo {
1099 let stats = self.rdf_store.stats();
1100
1101 let predicates = self
1102 .rdf_store
1103 .predicates()
1104 .into_iter()
1105 .map(|predicate| {
1106 let count = self.rdf_store.triples_with_predicate(&predicate).len();
1107 crate::admin::PredicateInfo {
1108 iri: predicate.to_string(),
1109 count,
1110 }
1111 })
1112 .collect();
1113
1114 crate::admin::SchemaInfo::Rdf(crate::admin::RdfSchemaInfo {
1115 predicates,
1116 named_graphs: Vec::new(), subject_count: stats.subject_count,
1118 object_count: stats.object_count,
1119 })
1120 }
1121
1122 #[must_use]
1130 pub fn validate(&self) -> crate::admin::ValidationResult {
1131 let mut result = crate::admin::ValidationResult::default();
1132
1133 for edge in self.store.all_edges() {
1135 if self.store.get_node(edge.src).is_none() {
1136 result.errors.push(crate::admin::ValidationError {
1137 code: "DANGLING_SRC".to_string(),
1138 message: format!(
1139 "Edge {} references non-existent source node {}",
1140 edge.id.0, edge.src.0
1141 ),
1142 context: Some(format!("edge:{}", edge.id.0)),
1143 });
1144 }
1145 if self.store.get_node(edge.dst).is_none() {
1146 result.errors.push(crate::admin::ValidationError {
1147 code: "DANGLING_DST".to_string(),
1148 message: format!(
1149 "Edge {} references non-existent destination node {}",
1150 edge.id.0, edge.dst.0
1151 ),
1152 context: Some(format!("edge:{}", edge.id.0)),
1153 });
1154 }
1155 }
1156
1157 if self.store.node_count() > 0 && self.store.edge_count() == 0 {
1159 result.warnings.push(crate::admin::ValidationWarning {
1160 code: "NO_EDGES".to_string(),
1161 message: "Database has nodes but no edges".to_string(),
1162 context: None,
1163 });
1164 }
1165
1166 result
1167 }
1168
1169 #[must_use]
1173 pub fn wal_status(&self) -> crate::admin::WalStatus {
1174 if let Some(ref wal) = self.wal {
1175 crate::admin::WalStatus {
1176 enabled: true,
1177 path: self.config.path.as_ref().map(|p| p.join("wal")),
1178 size_bytes: wal.size_bytes(),
1179 record_count: wal.record_count() as usize,
1180 last_checkpoint: wal.last_checkpoint_timestamp(),
1181 current_epoch: self.store.current_epoch().as_u64(),
1182 }
1183 } else {
1184 crate::admin::WalStatus {
1185 enabled: false,
1186 path: None,
1187 size_bytes: 0,
1188 record_count: 0,
1189 last_checkpoint: None,
1190 current_epoch: self.store.current_epoch().as_u64(),
1191 }
1192 }
1193 }
1194
1195 pub fn wal_checkpoint(&self) -> Result<()> {
1203 if let Some(ref wal) = self.wal {
1204 let epoch = self.store.current_epoch();
1205 let tx_id = self
1206 .tx_manager
1207 .last_assigned_tx_id()
1208 .unwrap_or_else(|| self.tx_manager.begin());
1209 wal.checkpoint(tx_id, epoch)?;
1210 wal.sync()?;
1211 }
1212 Ok(())
1213 }
1214
1215 pub fn save(&self, path: impl AsRef<Path>) -> Result<()> {
1230 let path = path.as_ref();
1231
1232 let target_config = Config::persistent(path);
1234 let target = Self::with_config(target_config)?;
1235
1236 for node in self.store.all_nodes() {
1238 let label_refs: Vec<&str> = node.labels.iter().map(|s| &**s).collect();
1239 target.store.create_node_with_id(node.id, &label_refs);
1240
1241 target.log_wal(&WalRecord::CreateNode {
1243 id: node.id,
1244 labels: node.labels.iter().map(|s| s.to_string()).collect(),
1245 })?;
1246
1247 for (key, value) in node.properties {
1249 target
1250 .store
1251 .set_node_property(node.id, key.as_str(), value.clone());
1252 target.log_wal(&WalRecord::SetNodeProperty {
1253 id: node.id,
1254 key: key.to_string(),
1255 value,
1256 })?;
1257 }
1258 }
1259
1260 for edge in self.store.all_edges() {
1262 target
1263 .store
1264 .create_edge_with_id(edge.id, edge.src, edge.dst, &edge.edge_type);
1265
1266 target.log_wal(&WalRecord::CreateEdge {
1268 id: edge.id,
1269 src: edge.src,
1270 dst: edge.dst,
1271 edge_type: edge.edge_type.to_string(),
1272 })?;
1273
1274 for (key, value) in edge.properties {
1276 target
1277 .store
1278 .set_edge_property(edge.id, key.as_str(), value.clone());
1279 target.log_wal(&WalRecord::SetEdgeProperty {
1280 id: edge.id,
1281 key: key.to_string(),
1282 value,
1283 })?;
1284 }
1285 }
1286
1287 target.close()?;
1289
1290 Ok(())
1291 }
1292
1293 pub fn to_memory(&self) -> Result<Self> {
1304 let config = Config::in_memory();
1305 let target = Self::with_config(config)?;
1306
1307 for node in self.store.all_nodes() {
1309 let label_refs: Vec<&str> = node.labels.iter().map(|s| &**s).collect();
1310 target.store.create_node_with_id(node.id, &label_refs);
1311
1312 for (key, value) in node.properties {
1314 target.store.set_node_property(node.id, key.as_str(), value);
1315 }
1316 }
1317
1318 for edge in self.store.all_edges() {
1320 target
1321 .store
1322 .create_edge_with_id(edge.id, edge.src, edge.dst, &edge.edge_type);
1323
1324 for (key, value) in edge.properties {
1326 target.store.set_edge_property(edge.id, key.as_str(), value);
1327 }
1328 }
1329
1330 Ok(target)
1331 }
1332
1333 pub fn open_in_memory(path: impl AsRef<Path>) -> Result<Self> {
1342 let source = Self::open(path)?;
1344
1345 let target = source.to_memory()?;
1347
1348 source.close()?;
1350
1351 Ok(target)
1352 }
1353
1354 pub fn iter_nodes(&self) -> impl Iterator<Item = grafeo_core::graph::lpg::Node> + '_ {
1362 self.store.all_nodes()
1363 }
1364
1365 pub fn iter_edges(&self) -> impl Iterator<Item = grafeo_core::graph::lpg::Edge> + '_ {
1369 self.store.all_edges()
1370 }
1371}
1372
1373impl Drop for GrafeoDB {
1374 fn drop(&mut self) {
1375 if let Err(e) = self.close() {
1376 tracing::error!("Error closing database: {}", e);
1377 }
1378 }
1379}
1380
1381#[derive(Debug)]
1407pub struct QueryResult {
1408 pub columns: Vec<String>,
1410 pub column_types: Vec<grafeo_common::types::LogicalType>,
1412 pub rows: Vec<Vec<grafeo_common::types::Value>>,
1414 pub execution_time_ms: Option<f64>,
1416 pub rows_scanned: Option<u64>,
1418}
1419
1420impl QueryResult {
1421 #[must_use]
1423 pub fn new(columns: Vec<String>) -> Self {
1424 let len = columns.len();
1425 Self {
1426 columns,
1427 column_types: vec![grafeo_common::types::LogicalType::Any; len],
1428 rows: Vec::new(),
1429 execution_time_ms: None,
1430 rows_scanned: None,
1431 }
1432 }
1433
1434 #[must_use]
1436 pub fn with_types(
1437 columns: Vec<String>,
1438 column_types: Vec<grafeo_common::types::LogicalType>,
1439 ) -> Self {
1440 Self {
1441 columns,
1442 column_types,
1443 rows: Vec::new(),
1444 execution_time_ms: None,
1445 rows_scanned: None,
1446 }
1447 }
1448
1449 pub fn with_metrics(mut self, execution_time_ms: f64, rows_scanned: u64) -> Self {
1451 self.execution_time_ms = Some(execution_time_ms);
1452 self.rows_scanned = Some(rows_scanned);
1453 self
1454 }
1455
1456 #[must_use]
1458 pub fn execution_time_ms(&self) -> Option<f64> {
1459 self.execution_time_ms
1460 }
1461
1462 #[must_use]
1464 pub fn rows_scanned(&self) -> Option<u64> {
1465 self.rows_scanned
1466 }
1467
1468 #[must_use]
1470 pub fn row_count(&self) -> usize {
1471 self.rows.len()
1472 }
1473
1474 #[must_use]
1476 pub fn column_count(&self) -> usize {
1477 self.columns.len()
1478 }
1479
1480 #[must_use]
1482 pub fn is_empty(&self) -> bool {
1483 self.rows.is_empty()
1484 }
1485
1486 pub fn scalar<T: FromValue>(&self) -> Result<T> {
1495 if self.rows.len() != 1 || self.columns.len() != 1 {
1496 return Err(grafeo_common::utils::error::Error::InvalidValue(
1497 "Expected single value".to_string(),
1498 ));
1499 }
1500 T::from_value(&self.rows[0][0])
1501 }
1502
1503 pub fn iter(&self) -> impl Iterator<Item = &Vec<grafeo_common::types::Value>> {
1505 self.rows.iter()
1506 }
1507}
1508
1509pub trait FromValue: Sized {
1514 fn from_value(value: &grafeo_common::types::Value) -> Result<Self>;
1516}
1517
1518impl FromValue for i64 {
1519 fn from_value(value: &grafeo_common::types::Value) -> Result<Self> {
1520 value
1521 .as_int64()
1522 .ok_or_else(|| grafeo_common::utils::error::Error::TypeMismatch {
1523 expected: "INT64".to_string(),
1524 found: value.type_name().to_string(),
1525 })
1526 }
1527}
1528
1529impl FromValue for f64 {
1530 fn from_value(value: &grafeo_common::types::Value) -> Result<Self> {
1531 value
1532 .as_float64()
1533 .ok_or_else(|| grafeo_common::utils::error::Error::TypeMismatch {
1534 expected: "FLOAT64".to_string(),
1535 found: value.type_name().to_string(),
1536 })
1537 }
1538}
1539
1540impl FromValue for String {
1541 fn from_value(value: &grafeo_common::types::Value) -> Result<Self> {
1542 value.as_str().map(String::from).ok_or_else(|| {
1543 grafeo_common::utils::error::Error::TypeMismatch {
1544 expected: "STRING".to_string(),
1545 found: value.type_name().to_string(),
1546 }
1547 })
1548 }
1549}
1550
1551impl FromValue for bool {
1552 fn from_value(value: &grafeo_common::types::Value) -> Result<Self> {
1553 value
1554 .as_bool()
1555 .ok_or_else(|| grafeo_common::utils::error::Error::TypeMismatch {
1556 expected: "BOOL".to_string(),
1557 found: value.type_name().to_string(),
1558 })
1559 }
1560}
1561
1562#[cfg(test)]
1563mod tests {
1564 use super::*;
1565
1566 #[test]
1567 fn test_create_in_memory_database() {
1568 let db = GrafeoDB::new_in_memory();
1569 assert_eq!(db.node_count(), 0);
1570 assert_eq!(db.edge_count(), 0);
1571 }
1572
1573 #[test]
1574 fn test_database_config() {
1575 let config = Config::in_memory().with_threads(4).with_query_logging();
1576
1577 let db = GrafeoDB::with_config(config).unwrap();
1578 assert_eq!(db.config().threads, 4);
1579 assert!(db.config().query_logging);
1580 }
1581
1582 #[test]
1583 fn test_database_session() {
1584 let db = GrafeoDB::new_in_memory();
1585 let _session = db.session();
1586 }
1588
1589 #[test]
1590 fn test_persistent_database_recovery() {
1591 use grafeo_common::types::Value;
1592 use tempfile::tempdir;
1593
1594 let dir = tempdir().unwrap();
1595 let db_path = dir.path().join("test_db");
1596
1597 {
1599 let db = GrafeoDB::open(&db_path).unwrap();
1600
1601 let alice = db.create_node(&["Person"]);
1602 db.set_node_property(alice, "name", Value::from("Alice"));
1603
1604 let bob = db.create_node(&["Person"]);
1605 db.set_node_property(bob, "name", Value::from("Bob"));
1606
1607 let _edge = db.create_edge(alice, bob, "KNOWS");
1608
1609 db.close().unwrap();
1611 }
1612
1613 {
1615 let db = GrafeoDB::open(&db_path).unwrap();
1616
1617 assert_eq!(db.node_count(), 2);
1618 assert_eq!(db.edge_count(), 1);
1619
1620 let node0 = db.get_node(grafeo_common::types::NodeId::new(0));
1622 assert!(node0.is_some());
1623
1624 let node1 = db.get_node(grafeo_common::types::NodeId::new(1));
1625 assert!(node1.is_some());
1626 }
1627 }
1628
1629 #[test]
1630 fn test_wal_logging() {
1631 use tempfile::tempdir;
1632
1633 let dir = tempdir().unwrap();
1634 let db_path = dir.path().join("wal_test_db");
1635
1636 let db = GrafeoDB::open(&db_path).unwrap();
1637
1638 let node = db.create_node(&["Test"]);
1640 db.delete_node(node);
1641
1642 if let Some(wal) = db.wal() {
1644 assert!(wal.record_count() > 0);
1645 }
1646
1647 db.close().unwrap();
1648 }
1649
1650 #[test]
1651 fn test_wal_recovery_multiple_sessions() {
1652 use grafeo_common::types::Value;
1654 use tempfile::tempdir;
1655
1656 let dir = tempdir().unwrap();
1657 let db_path = dir.path().join("multi_session_db");
1658
1659 {
1661 let db = GrafeoDB::open(&db_path).unwrap();
1662 let alice = db.create_node(&["Person"]);
1663 db.set_node_property(alice, "name", Value::from("Alice"));
1664 db.close().unwrap();
1665 }
1666
1667 {
1669 let db = GrafeoDB::open(&db_path).unwrap();
1670 assert_eq!(db.node_count(), 1); let bob = db.create_node(&["Person"]);
1672 db.set_node_property(bob, "name", Value::from("Bob"));
1673 db.close().unwrap();
1674 }
1675
1676 {
1678 let db = GrafeoDB::open(&db_path).unwrap();
1679 assert_eq!(db.node_count(), 2);
1680
1681 let node0 = db.get_node(grafeo_common::types::NodeId::new(0)).unwrap();
1683 assert!(node0.labels.iter().any(|l| l.as_str() == "Person"));
1684
1685 let node1 = db.get_node(grafeo_common::types::NodeId::new(1)).unwrap();
1686 assert!(node1.labels.iter().any(|l| l.as_str() == "Person"));
1687 }
1688 }
1689
1690 #[test]
1691 fn test_database_consistency_after_mutations() {
1692 use grafeo_common::types::Value;
1694 use tempfile::tempdir;
1695
1696 let dir = tempdir().unwrap();
1697 let db_path = dir.path().join("consistency_db");
1698
1699 {
1700 let db = GrafeoDB::open(&db_path).unwrap();
1701
1702 let a = db.create_node(&["Node"]);
1704 let b = db.create_node(&["Node"]);
1705 let c = db.create_node(&["Node"]);
1706
1707 let e1 = db.create_edge(a, b, "LINKS");
1709 let _e2 = db.create_edge(b, c, "LINKS");
1710
1711 db.delete_edge(e1);
1713 db.delete_node(b);
1714
1715 db.set_node_property(a, "value", Value::Int64(1));
1717 db.set_node_property(c, "value", Value::Int64(3));
1718
1719 db.close().unwrap();
1720 }
1721
1722 {
1724 let db = GrafeoDB::open(&db_path).unwrap();
1725
1726 let node_a = db.get_node(grafeo_common::types::NodeId::new(0));
1730 assert!(node_a.is_some());
1731
1732 let node_c = db.get_node(grafeo_common::types::NodeId::new(2));
1733 assert!(node_c.is_some());
1734
1735 let node_b = db.get_node(grafeo_common::types::NodeId::new(1));
1737 assert!(node_b.is_none());
1738 }
1739 }
1740
1741 #[test]
1742 fn test_close_is_idempotent() {
1743 use tempfile::tempdir;
1745
1746 let dir = tempdir().unwrap();
1747 let db_path = dir.path().join("close_test_db");
1748
1749 let db = GrafeoDB::open(&db_path).unwrap();
1750 db.create_node(&["Test"]);
1751
1752 assert!(db.close().is_ok());
1754
1755 assert!(db.close().is_ok());
1757 }
1758
1759 #[test]
1760 fn test_query_result_has_metrics() {
1761 let db = GrafeoDB::new_in_memory();
1763 db.create_node(&["Person"]);
1764 db.create_node(&["Person"]);
1765
1766 #[cfg(feature = "gql")]
1767 {
1768 let result = db.execute("MATCH (n:Person) RETURN n").unwrap();
1769
1770 assert!(result.execution_time_ms.is_some());
1772 assert!(result.rows_scanned.is_some());
1773 assert!(result.execution_time_ms.unwrap() >= 0.0);
1774 assert_eq!(result.rows_scanned.unwrap(), 2);
1775 }
1776 }
1777
1778 #[test]
1779 fn test_empty_query_result_metrics() {
1780 let db = GrafeoDB::new_in_memory();
1782 db.create_node(&["Person"]);
1783
1784 #[cfg(feature = "gql")]
1785 {
1786 let result = db.execute("MATCH (n:NonExistent) RETURN n").unwrap();
1788
1789 assert!(result.execution_time_ms.is_some());
1790 assert!(result.rows_scanned.is_some());
1791 assert_eq!(result.rows_scanned.unwrap(), 0);
1792 }
1793 }
1794}