lora_database/database/
profile.rs

1//! `Database::profile` — execute a query and report runtime metrics.
2//!
3//! Unlike `explain`, `profile` runs the query against the live database.
4//! Mutating queries (CREATE / MERGE / SET / DELETE / REMOVE) are
5//! persisted exactly as they would be from `execute()`. Callers who
6//! want to inspect a mutating plan without running it should use
7//! `explain` instead.
8//!
9//! v1 surfaces coarse metrics (total elapsed time, total rows produced,
10//! whether mutations occurred). Per-operator instrumentation is
11//! reserved for a future phase and lives in
12//! [`crate::explain::ProfileMetrics::per_operator`] as an empty map.
13
14use std::any::Any;
15use std::collections::BTreeMap;
16use std::sync::Arc;
17use std::time::Instant;
18
19use lora_compiler::plan_tree_from_compiled;
20use lora_executor::{
21    classify_stream, collect_compiled, plan_result_columns, CollectorGuard, LoraValue,
22    MetricsCollector, StreamShape,
23};
24use lora_store::{GraphStorage, GraphStorageMut};
25
26use crate::database::Database;
27use crate::error::LoraError;
28use crate::explain::{OperatorMetrics, PlanShape, ProfileMetrics, QueryPlan, QueryProfile};
29
30impl<S> Database<S>
31where
32    S: GraphStorage + GraphStorageMut + Any + Clone + Send + Sync + 'static,
33{
34    /// Execute `query` and return the plan plus runtime metrics.
35    ///
36    /// **PROFILE executes the query for real.** Mutating queries
37    /// (CREATE / MERGE / SET / DELETE / REMOVE) produce exactly the
38    /// same side effects as `execute()` — the WAL is written, snapshots
39    /// observe the commit, and the live store advances. Use
40    /// [`Database::explain`] to inspect a plan without running it.
41    pub fn profile(
42        &self,
43        query: &str,
44        params: Option<BTreeMap<String, LoraValue>>,
45    ) -> Result<QueryProfile, LoraError> {
46        let params = params.unwrap_or_default();
47
48        // Compile up front so we can attach the plan to the profile
49        // result whether execution succeeds or not. (Errors during
50        // compile flow through the same `LoraError` path as
51        // `execute()`, so the caller sees consistent error codes.)
52        let (store, store_epoch) = self.read_store_with_epoch_deadline(None)?;
53        let compiled = self
54            .compile_query_cached(query, &*store, store_epoch)
55            .map_err(LoraError::from_anyhow)?;
56        let tree = plan_tree_from_compiled(&compiled);
57        let shape: PlanShape = classify_stream(&compiled).into();
58        let result_columns = plan_result_columns(&compiled.physical);
59        drop(store);
60
61        let plan = QueryPlan {
62            query: query.to_string(),
63            tree,
64            shape,
65            result_columns,
66        };
67
68        let collector = Arc::new(MetricsCollector::new());
69        let _guard = CollectorGuard::install(collector.clone());
70
71        // For read-only queries route through the streaming pull
72        // executor so the metrics collector sees every operator's
73        // `next_row` call. The `execute()` fast path uses a buffered
74        // executor when there's no early LIMIT; that path skips
75        // `build_streaming`, so we'd get only top-level totals back.
76        // Profile accepts the small perf cost in exchange for per-op
77        // timing.
78        let started = Instant::now();
79        let rows = if matches!(classify_stream(&compiled), StreamShape::ReadOnly) {
80            let snapshot = self.read_store();
81            let res = collect_compiled(&*snapshot, params, &compiled)
82                .map_err(|e| LoraError::from_anyhow(e.into()));
83            drop(snapshot);
84            res?
85        } else {
86            self.execute_rows_with_params(query, params)
87                .map_err(|e| LoraError::from_anyhow(e.into()))?
88        };
89        let total_elapsed_ns = started.elapsed().as_nanos() as u64;
90
91        // Drop the guard before reading the snapshot so the
92        // thread-local is cleared even if `snapshot` panics. (`Arc`
93        // ensures the collector outlives the guard.)
94        drop(_guard);
95        let per_operator = collector
96            .snapshot()
97            .into_iter()
98            .map(|(id, op)| {
99                (
100                    id,
101                    OperatorMetrics {
102                        rows: op.rows,
103                        elapsed_ns: op.elapsed_ns,
104                        next_calls: op.next_calls,
105                        // db_hits is reserved for a future phase.
106                        db_hits: 0,
107                    },
108                )
109            })
110            .collect();
111
112        let metrics = ProfileMetrics {
113            total_elapsed_ns,
114            total_rows: rows.len() as u64,
115            mutated: shape.is_mutating(),
116            per_operator,
117        };
118
119        Ok(QueryProfile { plan, metrics })
120    }
121}
lora_database/database/profile.rs

lora_database/database/
profile.rs
