grafeo_engine/database/
admin.rs

1//! Admin, introspection, and diagnostic operations for GrafeoDB.
2
3use std::path::Path;
4
5use grafeo_common::utils::error::Result;
6
7impl super::GrafeoDB {
8    // =========================================================================
9    // ADMIN API: Counts
10    // =========================================================================
11
12    /// Returns the number of nodes in the database.
13    #[must_use]
14    pub fn node_count(&self) -> usize {
15        self.store.node_count()
16    }
17
18    /// Returns the number of edges in the database.
19    #[must_use]
20    pub fn edge_count(&self) -> usize {
21        self.store.edge_count()
22    }
23
24    /// Returns the number of distinct labels in the database.
25    #[must_use]
26    pub fn label_count(&self) -> usize {
27        self.store.label_count()
28    }
29
30    /// Returns the number of distinct property keys in the database.
31    #[must_use]
32    pub fn property_key_count(&self) -> usize {
33        self.store.property_key_count()
34    }
35
36    /// Returns the number of distinct edge types in the database.
37    #[must_use]
38    pub fn edge_type_count(&self) -> usize {
39        self.store.edge_type_count()
40    }
41
42    // =========================================================================
43    // ADMIN API: Introspection
44    // =========================================================================
45
46    /// Returns true if this database is backed by a file (persistent).
47    ///
48    /// In-memory databases return false.
49    #[must_use]
50    pub fn is_persistent(&self) -> bool {
51        self.config.path.is_some()
52    }
53
54    /// Returns the database file path, if persistent.
55    ///
56    /// In-memory databases return None.
57    #[must_use]
58    pub fn path(&self) -> Option<&Path> {
59        self.config.path.as_deref()
60    }
61
62    /// Returns high-level database information.
63    ///
64    /// Includes node/edge counts, persistence status, and mode (LPG/RDF).
65    #[must_use]
66    pub fn info(&self) -> crate::admin::DatabaseInfo {
67        crate::admin::DatabaseInfo {
68            mode: crate::admin::DatabaseMode::Lpg,
69            node_count: self.store.node_count(),
70            edge_count: self.store.edge_count(),
71            is_persistent: self.is_persistent(),
72            path: self.config.path.clone(),
73            wal_enabled: self.config.wal_enabled,
74            version: env!("CARGO_PKG_VERSION").to_string(),
75        }
76    }
77
78    /// Returns a hierarchical memory usage breakdown.
79    ///
80    /// Walks all internal structures (store, indexes, MVCC chains, caches,
81    /// string pools, buffer manager) and returns estimated heap bytes for each.
82    /// Safe to call concurrently with queries.
83    #[must_use]
84    pub fn memory_usage(&self) -> crate::memory_usage::MemoryUsage {
85        use crate::memory_usage::{BufferManagerMemory, CacheMemory, MemoryUsage};
86        use grafeo_common::memory::MemoryRegion;
87
88        let (store, indexes, mvcc, string_pool) = self.store.memory_breakdown();
89
90        let (parsed_bytes, optimized_bytes, cached_plan_count) =
91            self.query_cache.heap_memory_bytes();
92        let mut caches = CacheMemory {
93            parsed_plan_cache_bytes: parsed_bytes,
94            optimized_plan_cache_bytes: optimized_bytes,
95            cached_plan_count,
96            ..Default::default()
97        };
98        caches.compute_total();
99
100        let bm_stats = self.buffer_manager.stats();
101        let buffer_manager = BufferManagerMemory {
102            budget_bytes: bm_stats.budget,
103            allocated_bytes: bm_stats.total_allocated,
104            graph_storage_bytes: bm_stats.region_usage(MemoryRegion::GraphStorage),
105            index_buffers_bytes: bm_stats.region_usage(MemoryRegion::IndexBuffers),
106            execution_buffers_bytes: bm_stats.region_usage(MemoryRegion::ExecutionBuffers),
107            spill_staging_bytes: bm_stats.region_usage(MemoryRegion::SpillStaging),
108        };
109
110        let mut usage = MemoryUsage {
111            store,
112            indexes,
113            mvcc,
114            caches,
115            string_pool,
116            buffer_manager,
117            ..Default::default()
118        };
119        usage.compute_total();
120        usage
121    }
122
123    /// Returns detailed database statistics.
124    ///
125    /// Includes counts, memory usage, and index information.
126    #[must_use]
127    pub fn detailed_stats(&self) -> crate::admin::DatabaseStats {
128        #[cfg(feature = "wal")]
129        let disk_bytes = self.config.path.as_ref().and_then(|p| {
130            if p.exists() {
131                Self::calculate_disk_usage(p).ok()
132            } else {
133                None
134            }
135        });
136        #[cfg(not(feature = "wal"))]
137        let disk_bytes: Option<usize> = None;
138
139        crate::admin::DatabaseStats {
140            node_count: self.store.node_count(),
141            edge_count: self.store.edge_count(),
142            label_count: self.store.label_count(),
143            edge_type_count: self.store.edge_type_count(),
144            property_key_count: self.store.property_key_count(),
145            index_count: self.catalog.index_count(),
146            memory_bytes: self.buffer_manager.allocated(),
147            disk_bytes,
148        }
149    }
150
151    /// Calculates total disk usage for the database directory.
152    #[cfg(feature = "wal")]
153    fn calculate_disk_usage(path: &Path) -> Result<usize> {
154        let mut total = 0usize;
155        if path.is_dir() {
156            for entry in std::fs::read_dir(path)? {
157                let entry = entry?;
158                let metadata = entry.metadata()?;
159                if metadata.is_file() {
160                    total += metadata.len() as usize;
161                } else if metadata.is_dir() {
162                    total += Self::calculate_disk_usage(&entry.path())?;
163                }
164            }
165        }
166        Ok(total)
167    }
168
169    /// Returns schema information (labels, edge types, property keys).
170    ///
171    /// For LPG mode, returns label and edge type information.
172    /// For RDF mode, returns predicate and named graph information.
173    #[must_use]
174    pub fn schema(&self) -> crate::admin::SchemaInfo {
175        let labels = self
176            .store
177            .all_labels()
178            .into_iter()
179            .map(|name| crate::admin::LabelInfo {
180                name: name.clone(),
181                count: self.store.nodes_with_label(&name).count(),
182            })
183            .collect();
184
185        let edge_types = self
186            .store
187            .all_edge_types()
188            .into_iter()
189            .map(|name| crate::admin::EdgeTypeInfo {
190                name: name.clone(),
191                count: self.store.edges_with_type(&name).count(),
192            })
193            .collect();
194
195        let property_keys = self.store.all_property_keys();
196
197        crate::admin::SchemaInfo::Lpg(crate::admin::LpgSchemaInfo {
198            labels,
199            edge_types,
200            property_keys,
201        })
202    }
203
204    /// Returns detailed information about all indexes.
205    #[must_use]
206    pub fn list_indexes(&self) -> Vec<crate::admin::IndexInfo> {
207        self.catalog
208            .all_indexes()
209            .into_iter()
210            .map(|def| {
211                let label_name = self
212                    .catalog
213                    .get_label_name(def.label)
214                    .unwrap_or_else(|| "?".into());
215                let prop_name = self
216                    .catalog
217                    .get_property_key_name(def.property_key)
218                    .unwrap_or_else(|| "?".into());
219                crate::admin::IndexInfo {
220                    name: format!("idx_{}_{}", label_name, prop_name),
221                    index_type: format!("{:?}", def.index_type),
222                    target: format!("{}:{}", label_name, prop_name),
223                    unique: false,
224                    cardinality: None,
225                    size_bytes: None,
226                }
227            })
228            .collect()
229    }
230
231    /// Validates database integrity.
232    ///
233    /// Checks for:
234    /// - Dangling edge references (edges pointing to non-existent nodes)
235    /// - Internal index consistency
236    ///
237    /// Returns a list of errors and warnings. Empty errors = valid.
238    #[must_use]
239    pub fn validate(&self) -> crate::admin::ValidationResult {
240        let mut result = crate::admin::ValidationResult::default();
241
242        // Check for dangling edge references
243        for edge in self.store.all_edges() {
244            if self.store.get_node(edge.src).is_none() {
245                result.errors.push(crate::admin::ValidationError {
246                    code: "DANGLING_SRC".to_string(),
247                    message: format!(
248                        "Edge {} references non-existent source node {}",
249                        edge.id.0, edge.src.0
250                    ),
251                    context: Some(format!("edge:{}", edge.id.0)),
252                });
253            }
254            if self.store.get_node(edge.dst).is_none() {
255                result.errors.push(crate::admin::ValidationError {
256                    code: "DANGLING_DST".to_string(),
257                    message: format!(
258                        "Edge {} references non-existent destination node {}",
259                        edge.id.0, edge.dst.0
260                    ),
261                    context: Some(format!("edge:{}", edge.id.0)),
262                });
263            }
264        }
265
266        // Add warnings for potential issues
267        if self.store.node_count() > 0 && self.store.edge_count() == 0 {
268            result.warnings.push(crate::admin::ValidationWarning {
269                code: "NO_EDGES".to_string(),
270                message: "Database has nodes but no edges".to_string(),
271                context: None,
272            });
273        }
274
275        result
276    }
277
278    /// Returns WAL (Write-Ahead Log) status.
279    ///
280    /// Returns None if WAL is not enabled.
281    #[must_use]
282    pub fn wal_status(&self) -> crate::admin::WalStatus {
283        #[cfg(feature = "wal")]
284        if let Some(ref wal) = self.wal {
285            return crate::admin::WalStatus {
286                enabled: true,
287                path: self.config.path.as_ref().map(|p| p.join("wal")),
288                size_bytes: wal.size_bytes(),
289                record_count: wal.record_count() as usize,
290                last_checkpoint: wal.last_checkpoint_timestamp(),
291                current_epoch: self.store.current_epoch().as_u64(),
292            };
293        }
294
295        crate::admin::WalStatus {
296            enabled: false,
297            path: None,
298            size_bytes: 0,
299            record_count: 0,
300            last_checkpoint: None,
301            current_epoch: self.store.current_epoch().as_u64(),
302        }
303    }
304
305    /// Forces a WAL checkpoint.
306    ///
307    /// Flushes all pending WAL records to the main storage.
308    ///
309    /// # Errors
310    ///
311    /// Returns an error if the checkpoint fails.
312    pub fn wal_checkpoint(&self) -> Result<()> {
313        #[cfg(feature = "wal")]
314        if let Some(ref wal) = self.wal {
315            let epoch = self.store.current_epoch();
316            let transaction_id = self
317                .transaction_manager
318                .last_assigned_transaction_id()
319                .unwrap_or_else(|| self.transaction_manager.begin());
320            wal.checkpoint(transaction_id, epoch)?;
321            wal.sync()?;
322        }
323
324        // For single-file format: flush snapshot to .grafeo file
325        #[cfg(feature = "grafeo-file")]
326        if let Some(ref fm) = self.file_manager {
327            self.checkpoint_to_file(fm)?;
328        }
329
330        Ok(())
331    }
332
333    // =========================================================================
334    // ADMIN API: Change Data Capture
335    // =========================================================================
336
337    /// Returns the full change history for an entity (node or edge).
338    ///
339    /// Events are ordered chronologically by epoch.
340    ///
341    /// # Errors
342    ///
343    /// Returns an error if the CDC feature is not enabled.
344    #[cfg(feature = "cdc")]
345    pub fn history(
346        &self,
347        entity_id: impl Into<crate::cdc::EntityId>,
348    ) -> Result<Vec<crate::cdc::ChangeEvent>> {
349        Ok(self.cdc_log.history(entity_id.into()))
350    }
351
352    /// Returns change events for an entity since the given epoch.
353    #[cfg(feature = "cdc")]
354    pub fn history_since(
355        &self,
356        entity_id: impl Into<crate::cdc::EntityId>,
357        since_epoch: grafeo_common::types::EpochId,
358    ) -> Result<Vec<crate::cdc::ChangeEvent>> {
359        Ok(self.cdc_log.history_since(entity_id.into(), since_epoch))
360    }
361
362    /// Returns all change events across all entities in an epoch range.
363    #[cfg(feature = "cdc")]
364    pub fn changes_between(
365        &self,
366        start_epoch: grafeo_common::types::EpochId,
367        end_epoch: grafeo_common::types::EpochId,
368    ) -> Result<Vec<crate::cdc::ChangeEvent>> {
369        Ok(self.cdc_log.changes_between(start_epoch, end_epoch))
370    }
371}
grafeo_engine/database/admin.rs

grafeo_engine/database/
admin.rs
