Skip to main content

grafeo_engine/
session.rs

1//! Lightweight handles for database interaction.
2//!
3//! A session is your conversation with the database. Each session can have
4//! its own transaction state, so concurrent sessions don't interfere with
5//! each other. Sessions are cheap to create - spin up as many as you need.
6
7use std::sync::Arc;
8use std::sync::atomic::{AtomicUsize, Ordering};
9use std::time::{Duration, Instant};
10
11use grafeo_common::types::{EdgeId, EpochId, NodeId, TransactionId, Value};
12use grafeo_common::utils::error::Result;
13use grafeo_core::graph::Direction;
14use grafeo_core::graph::GraphStoreMut;
15use grafeo_core::graph::lpg::{Edge, LpgStore, Node};
16#[cfg(feature = "rdf")]
17use grafeo_core::graph::rdf::RdfStore;
18
19use crate::catalog::{Catalog, CatalogConstraintValidator};
20use crate::config::{AdaptiveConfig, GraphModel};
21use crate::database::QueryResult;
22use crate::query::cache::QueryCache;
23use crate::transaction::TransactionManager;
24
25/// Your handle to the database - execute queries and manage transactions.
26///
27/// Get one from [`GrafeoDB::session()`](crate::GrafeoDB::session). Each session
28/// tracks its own transaction state, so you can have multiple concurrent
29/// sessions without them interfering.
30pub struct Session {
31    /// The underlying store.
32    store: Arc<LpgStore>,
33    /// Graph store trait object for pluggable storage backends.
34    graph_store: Arc<dyn GraphStoreMut>,
35    /// Schema and metadata catalog shared across sessions.
36    catalog: Arc<Catalog>,
37    /// RDF triple store (if RDF feature is enabled).
38    #[cfg(feature = "rdf")]
39    rdf_store: Arc<RdfStore>,
40    /// Transaction manager.
41    transaction_manager: Arc<TransactionManager>,
42    /// Query cache shared across sessions.
43    query_cache: Arc<QueryCache>,
44    /// Current transaction ID (if any). Behind a Mutex so that GQL commands
45    /// (`START TRANSACTION`, `COMMIT`, `ROLLBACK`) can manage transactions
46    /// from within `execute(&self)`.
47    current_transaction: parking_lot::Mutex<Option<TransactionId>>,
48    /// Whether the current transaction is read-only (blocks mutations).
49    read_only_tx: parking_lot::Mutex<bool>,
50    /// Whether the session is in auto-commit mode.
51    auto_commit: bool,
52    /// Adaptive execution configuration.
53    #[allow(dead_code)] // Stored for future adaptive re-optimization during execution
54    adaptive_config: AdaptiveConfig,
55    /// Whether to use factorized execution for multi-hop queries.
56    factorized_execution: bool,
57    /// The graph data model this session operates on.
58    graph_model: GraphModel,
59    /// Maximum time a query may run before being cancelled.
60    query_timeout: Option<Duration>,
61    /// Shared commit counter for triggering auto-GC.
62    commit_counter: Arc<AtomicUsize>,
63    /// GC every N commits (0 = disabled).
64    gc_interval: usize,
65    /// Node count at the start of the current transaction (for PreparedCommit stats).
66    transaction_start_node_count: AtomicUsize,
67    /// Edge count at the start of the current transaction (for PreparedCommit stats).
68    transaction_start_edge_count: AtomicUsize,
69    /// WAL for logging schema changes.
70    #[cfg(feature = "wal")]
71    wal: Option<Arc<grafeo_adapters::storage::wal::LpgWal>>,
72    /// CDC log for change tracking.
73    #[cfg(feature = "cdc")]
74    cdc_log: Arc<crate::cdc::CdcLog>,
75    /// Current graph name (for multi-graph USE GRAPH support). None = default graph.
76    current_graph: parking_lot::Mutex<Option<String>>,
77    /// Session time zone override.
78    time_zone: parking_lot::Mutex<Option<String>>,
79    /// Session-level parameters (SET PARAMETER).
80    session_params:
81        parking_lot::Mutex<std::collections::HashMap<String, grafeo_common::types::Value>>,
82    /// Override epoch for time-travel queries (None = use transaction/current epoch).
83    viewing_epoch_override: parking_lot::Mutex<Option<EpochId>>,
84    /// Savepoints within the current transaction: name -> (next_node_id, next_edge_id) snapshot.
85    savepoints: parking_lot::Mutex<Vec<(String, u64, u64)>>,
86    /// Nesting depth for nested transactions (0 = outermost).
87    /// Nested `START TRANSACTION` creates an auto-savepoint; nested `COMMIT`
88    /// releases it, nested `ROLLBACK` rolls back to it.
89    transaction_nesting_depth: parking_lot::Mutex<u32>,
90}
91
92impl Session {
93    /// Creates a new session with adaptive execution configuration.
94    #[allow(dead_code, clippy::too_many_arguments)]
95    pub(crate) fn with_adaptive(
96        store: Arc<LpgStore>,
97        transaction_manager: Arc<TransactionManager>,
98        query_cache: Arc<QueryCache>,
99        catalog: Arc<Catalog>,
100        adaptive_config: AdaptiveConfig,
101        factorized_execution: bool,
102        graph_model: GraphModel,
103        query_timeout: Option<Duration>,
104        commit_counter: Arc<AtomicUsize>,
105        gc_interval: usize,
106    ) -> Self {
107        let graph_store = Arc::clone(&store) as Arc<dyn GraphStoreMut>;
108        Self {
109            store,
110            graph_store,
111            catalog,
112            #[cfg(feature = "rdf")]
113            rdf_store: Arc::new(RdfStore::new()),
114            transaction_manager,
115            query_cache,
116            current_transaction: parking_lot::Mutex::new(None),
117            read_only_tx: parking_lot::Mutex::new(false),
118            auto_commit: true,
119            adaptive_config,
120            factorized_execution,
121            graph_model,
122            query_timeout,
123            commit_counter,
124            gc_interval,
125            transaction_start_node_count: AtomicUsize::new(0),
126            transaction_start_edge_count: AtomicUsize::new(0),
127            #[cfg(feature = "wal")]
128            wal: None,
129            #[cfg(feature = "cdc")]
130            cdc_log: Arc::new(crate::cdc::CdcLog::new()),
131            current_graph: parking_lot::Mutex::new(None),
132            time_zone: parking_lot::Mutex::new(None),
133            session_params: parking_lot::Mutex::new(std::collections::HashMap::new()),
134            viewing_epoch_override: parking_lot::Mutex::new(None),
135            savepoints: parking_lot::Mutex::new(Vec::new()),
136            transaction_nesting_depth: parking_lot::Mutex::new(0),
137        }
138    }
139
140    /// Sets the WAL for this session (shared with the database).
141    ///
142    /// This also wraps `graph_store` in a [`WalGraphStore`] so that mutation
143    /// operators (INSERT, DELETE, SET via queries) log to the WAL.
144    #[cfg(feature = "wal")]
145    pub(crate) fn set_wal(&mut self, wal: Arc<grafeo_adapters::storage::wal::LpgWal>) {
146        // Wrap the graph store so query-engine mutations are WAL-logged
147        self.graph_store = Arc::new(crate::database::wal_store::WalGraphStore::new(
148            Arc::clone(&self.store),
149            Arc::clone(&wal),
150        ));
151        self.wal = Some(wal);
152    }
153
154    /// Sets the CDC log for this session (shared with the database).
155    #[cfg(feature = "cdc")]
156    pub(crate) fn set_cdc_log(&mut self, cdc_log: Arc<crate::cdc::CdcLog>) {
157        self.cdc_log = cdc_log;
158    }
159
160    /// Creates a new session with RDF store and adaptive configuration.
161    #[cfg(feature = "rdf")]
162    #[allow(clippy::too_many_arguments)]
163    pub(crate) fn with_rdf_store_and_adaptive(
164        store: Arc<LpgStore>,
165        rdf_store: Arc<RdfStore>,
166        transaction_manager: Arc<TransactionManager>,
167        query_cache: Arc<QueryCache>,
168        catalog: Arc<Catalog>,
169        adaptive_config: AdaptiveConfig,
170        factorized_execution: bool,
171        graph_model: GraphModel,
172        query_timeout: Option<Duration>,
173        commit_counter: Arc<AtomicUsize>,
174        gc_interval: usize,
175    ) -> Self {
176        let graph_store = Arc::clone(&store) as Arc<dyn GraphStoreMut>;
177        Self {
178            store,
179            graph_store,
180            catalog,
181            rdf_store,
182            transaction_manager,
183            query_cache,
184            current_transaction: parking_lot::Mutex::new(None),
185            read_only_tx: parking_lot::Mutex::new(false),
186            auto_commit: true,
187            adaptive_config,
188            factorized_execution,
189            graph_model,
190            query_timeout,
191            commit_counter,
192            gc_interval,
193            transaction_start_node_count: AtomicUsize::new(0),
194            transaction_start_edge_count: AtomicUsize::new(0),
195            #[cfg(feature = "wal")]
196            wal: None,
197            #[cfg(feature = "cdc")]
198            cdc_log: Arc::new(crate::cdc::CdcLog::new()),
199            current_graph: parking_lot::Mutex::new(None),
200            time_zone: parking_lot::Mutex::new(None),
201            session_params: parking_lot::Mutex::new(std::collections::HashMap::new()),
202            viewing_epoch_override: parking_lot::Mutex::new(None),
203            savepoints: parking_lot::Mutex::new(Vec::new()),
204            transaction_nesting_depth: parking_lot::Mutex::new(0),
205        }
206    }
207
208    /// Creates a session backed by an external graph store.
209    ///
210    /// The external store handles all data operations. Transaction management
211    /// (begin/commit/rollback) is not supported for external stores.
212    #[allow(clippy::too_many_arguments)]
213    pub(crate) fn with_external_store(
214        store: Arc<dyn GraphStoreMut>,
215        transaction_manager: Arc<TransactionManager>,
216        query_cache: Arc<QueryCache>,
217        catalog: Arc<Catalog>,
218        adaptive_config: AdaptiveConfig,
219        factorized_execution: bool,
220        graph_model: GraphModel,
221        query_timeout: Option<Duration>,
222        commit_counter: Arc<AtomicUsize>,
223        gc_interval: usize,
224    ) -> Self {
225        Self {
226            store: Arc::new(LpgStore::new().expect("arena allocation for dummy LpgStore")), // dummy for LpgStore-specific ops
227            graph_store: store,
228            catalog,
229            #[cfg(feature = "rdf")]
230            rdf_store: Arc::new(RdfStore::new()),
231            transaction_manager,
232            query_cache,
233            current_transaction: parking_lot::Mutex::new(None),
234            read_only_tx: parking_lot::Mutex::new(false),
235            auto_commit: true,
236            adaptive_config,
237            factorized_execution,
238            graph_model,
239            query_timeout,
240            commit_counter,
241            gc_interval,
242            transaction_start_node_count: AtomicUsize::new(0),
243            transaction_start_edge_count: AtomicUsize::new(0),
244            #[cfg(feature = "wal")]
245            wal: None,
246            #[cfg(feature = "cdc")]
247            cdc_log: Arc::new(crate::cdc::CdcLog::new()),
248            current_graph: parking_lot::Mutex::new(None),
249            time_zone: parking_lot::Mutex::new(None),
250            session_params: parking_lot::Mutex::new(std::collections::HashMap::new()),
251            viewing_epoch_override: parking_lot::Mutex::new(None),
252            savepoints: parking_lot::Mutex::new(Vec::new()),
253            transaction_nesting_depth: parking_lot::Mutex::new(0),
254        }
255    }
256
257    /// Returns the graph model this session operates on.
258    #[must_use]
259    pub fn graph_model(&self) -> GraphModel {
260        self.graph_model
261    }
262
263    // === Session State Management ===
264
265    /// Sets the current graph for this session (USE GRAPH).
266    pub fn use_graph(&self, name: &str) {
267        *self.current_graph.lock() = Some(name.to_string());
268    }
269
270    /// Returns the current graph name, if any.
271    #[must_use]
272    pub fn current_graph(&self) -> Option<String> {
273        self.current_graph.lock().clone()
274    }
275
276    /// Sets the session time zone.
277    pub fn set_time_zone(&self, tz: &str) {
278        *self.time_zone.lock() = Some(tz.to_string());
279    }
280
281    /// Returns the session time zone, if set.
282    #[must_use]
283    pub fn time_zone(&self) -> Option<String> {
284        self.time_zone.lock().clone()
285    }
286
287    /// Sets a session parameter.
288    pub fn set_parameter(&self, key: &str, value: grafeo_common::types::Value) {
289        self.session_params.lock().insert(key.to_string(), value);
290    }
291
292    /// Gets a session parameter by cloning it.
293    #[must_use]
294    pub fn get_parameter(&self, key: &str) -> Option<grafeo_common::types::Value> {
295        self.session_params.lock().get(key).cloned()
296    }
297
298    /// Resets all session state to defaults.
299    pub fn reset_session(&self) {
300        *self.current_graph.lock() = None;
301        *self.time_zone.lock() = None;
302        self.session_params.lock().clear();
303        *self.viewing_epoch_override.lock() = None;
304    }
305
306    // --- Time-travel API ---
307
308    /// Sets a viewing epoch override for time-travel queries.
309    ///
310    /// While set, all queries on this session see the database as it existed
311    /// at the given epoch. Use [`clear_viewing_epoch`](Self::clear_viewing_epoch)
312    /// to return to normal behavior.
313    pub fn set_viewing_epoch(&self, epoch: EpochId) {
314        *self.viewing_epoch_override.lock() = Some(epoch);
315    }
316
317    /// Clears the viewing epoch override, returning to normal behavior.
318    pub fn clear_viewing_epoch(&self) {
319        *self.viewing_epoch_override.lock() = None;
320    }
321
322    /// Returns the current viewing epoch override, if any.
323    #[must_use]
324    pub fn viewing_epoch(&self) -> Option<EpochId> {
325        *self.viewing_epoch_override.lock()
326    }
327
328    /// Returns all versions of a node with their creation/deletion epochs.
329    ///
330    /// Properties and labels reflect the current state (not versioned per-epoch).
331    #[must_use]
332    pub fn get_node_history(&self, id: NodeId) -> Vec<(EpochId, Option<EpochId>, Node)> {
333        self.store.get_node_history(id)
334    }
335
336    /// Returns all versions of an edge with their creation/deletion epochs.
337    ///
338    /// Properties reflect the current state (not versioned per-epoch).
339    #[must_use]
340    pub fn get_edge_history(&self, id: EdgeId) -> Vec<(EpochId, Option<EpochId>, Edge)> {
341        self.store.get_edge_history(id)
342    }
343
344    /// Checks that the session's graph model supports LPG operations.
345    fn require_lpg(&self, language: &str) -> Result<()> {
346        if self.graph_model == GraphModel::Rdf {
347            return Err(grafeo_common::utils::error::Error::Internal(format!(
348                "This is an RDF database. {language} queries require an LPG database."
349            )));
350        }
351        Ok(())
352    }
353
354    /// Executes a session or transaction command, returning an empty result.
355    #[cfg(feature = "gql")]
356    fn execute_session_command(
357        &self,
358        cmd: grafeo_adapters::query::gql::ast::SessionCommand,
359    ) -> Result<QueryResult> {
360        use grafeo_adapters::query::gql::ast::{SessionCommand, TransactionIsolationLevel};
361        use grafeo_common::utils::error::{Error, QueryError, QueryErrorKind};
362
363        match cmd {
364            SessionCommand::CreateGraph {
365                name,
366                if_not_exists,
367                typed,
368                like_graph,
369                copy_of,
370                open: _,
371            } => {
372                // Validate source graph exists for LIKE / AS COPY OF
373                if let Some(ref src) = like_graph
374                    && self.store.graph(src).is_none()
375                {
376                    return Err(Error::Query(QueryError::new(
377                        QueryErrorKind::Semantic,
378                        format!("Source graph '{src}' does not exist"),
379                    )));
380                }
381                if let Some(ref src) = copy_of
382                    && self.store.graph(src).is_none()
383                {
384                    return Err(Error::Query(QueryError::new(
385                        QueryErrorKind::Semantic,
386                        format!("Source graph '{src}' does not exist"),
387                    )));
388                }
389
390                let created = self
391                    .store
392                    .create_graph(&name)
393                    .map_err(|e| Error::Internal(e.to_string()))?;
394                if !created && !if_not_exists {
395                    return Err(Error::Query(QueryError::new(
396                        QueryErrorKind::Semantic,
397                        format!("Graph '{name}' already exists"),
398                    )));
399                }
400
401                // AS COPY OF: copy data from source graph
402                if let Some(ref src) = copy_of {
403                    self.store
404                        .copy_graph(Some(src), Some(&name))
405                        .map_err(|e| Error::Internal(e.to_string()))?;
406                }
407
408                // Bind to graph type if specified
409                if let Some(type_name) = typed
410                    && let Err(e) = self.catalog.bind_graph_type(&name, type_name.clone())
411                {
412                    return Err(Error::Query(QueryError::new(
413                        QueryErrorKind::Semantic,
414                        e.to_string(),
415                    )));
416                }
417
418                // LIKE: copy graph type binding from source
419                if let Some(ref src) = like_graph
420                    && let Some(src_type) = self.catalog.get_graph_type_binding(src)
421                {
422                    let _ = self.catalog.bind_graph_type(&name, src_type);
423                }
424
425                Ok(QueryResult::empty())
426            }
427            SessionCommand::DropGraph { name, if_exists } => {
428                let dropped = self.store.drop_graph(&name);
429                if !dropped && !if_exists {
430                    return Err(Error::Query(QueryError::new(
431                        QueryErrorKind::Semantic,
432                        format!("Graph '{name}' does not exist"),
433                    )));
434                }
435                Ok(QueryResult::empty())
436            }
437            SessionCommand::UseGraph(name) => {
438                // Verify graph exists (default graph is always valid)
439                if !name.eq_ignore_ascii_case("default") && self.store.graph(&name).is_none() {
440                    return Err(Error::Query(QueryError::new(
441                        QueryErrorKind::Semantic,
442                        format!("Graph '{name}' does not exist"),
443                    )));
444                }
445                self.use_graph(&name);
446                Ok(QueryResult::empty())
447            }
448            SessionCommand::SessionSetGraph(name) => {
449                self.use_graph(&name);
450                Ok(QueryResult::empty())
451            }
452            SessionCommand::SessionSetTimeZone(tz) => {
453                self.set_time_zone(&tz);
454                Ok(QueryResult::empty())
455            }
456            SessionCommand::SessionSetParameter(key, expr) => {
457                if key.eq_ignore_ascii_case("viewing_epoch") {
458                    match Self::eval_integer_literal(&expr) {
459                        Some(n) if n >= 0 => {
460                            self.set_viewing_epoch(EpochId::new(n as u64));
461                            Ok(QueryResult::status(format!("Set viewing_epoch to {n}")))
462                        }
463                        _ => Err(Error::Query(QueryError::new(
464                            QueryErrorKind::Semantic,
465                            "viewing_epoch must be a non-negative integer literal",
466                        ))),
467                    }
468                } else {
469                    // For now, store parameter name with Null value.
470                    // Full expression evaluation would require building and executing a plan.
471                    self.set_parameter(&key, Value::Null);
472                    Ok(QueryResult::empty())
473                }
474            }
475            SessionCommand::SessionReset => {
476                self.reset_session();
477                Ok(QueryResult::empty())
478            }
479            SessionCommand::SessionClose => {
480                self.reset_session();
481                Ok(QueryResult::empty())
482            }
483            SessionCommand::StartTransaction {
484                read_only,
485                isolation_level,
486            } => {
487                let engine_level = isolation_level.map(|l| match l {
488                    TransactionIsolationLevel::ReadCommitted => {
489                        crate::transaction::IsolationLevel::ReadCommitted
490                    }
491                    TransactionIsolationLevel::SnapshotIsolation => {
492                        crate::transaction::IsolationLevel::SnapshotIsolation
493                    }
494                    TransactionIsolationLevel::Serializable => {
495                        crate::transaction::IsolationLevel::Serializable
496                    }
497                });
498                self.begin_transaction_inner(read_only, engine_level)?;
499                Ok(QueryResult::status("Transaction started"))
500            }
501            SessionCommand::Commit => {
502                self.commit_inner()?;
503                Ok(QueryResult::status("Transaction committed"))
504            }
505            SessionCommand::Rollback => {
506                self.rollback_inner()?;
507                Ok(QueryResult::status("Transaction rolled back"))
508            }
509            SessionCommand::Savepoint(name) => {
510                self.savepoint(&name)?;
511                Ok(QueryResult::status(format!("Savepoint '{name}' created")))
512            }
513            SessionCommand::RollbackToSavepoint(name) => {
514                self.rollback_to_savepoint(&name)?;
515                Ok(QueryResult::status(format!(
516                    "Rolled back to savepoint '{name}'"
517                )))
518            }
519            SessionCommand::ReleaseSavepoint(name) => {
520                self.release_savepoint(&name)?;
521                Ok(QueryResult::status(format!("Savepoint '{name}' released")))
522            }
523        }
524    }
525
526    /// Logs a WAL record for a schema change (no-op if WAL is not enabled).
527    #[cfg(feature = "wal")]
528    fn log_schema_wal(&self, record: &grafeo_adapters::storage::wal::WalRecord) {
529        if let Some(ref wal) = self.wal
530            && let Err(e) = wal.log(record)
531        {
532            tracing::warn!("Failed to log schema change to WAL: {}", e);
533        }
534    }
535
536    /// Executes a schema DDL command, returning a status result.
537    #[cfg(feature = "gql")]
538    fn execute_schema_command(
539        &self,
540        cmd: grafeo_adapters::query::gql::ast::SchemaStatement,
541    ) -> Result<QueryResult> {
542        use crate::catalog::{
543            EdgeTypeDefinition, NodeTypeDefinition, PropertyDataType, TypedProperty,
544        };
545        use grafeo_adapters::query::gql::ast::SchemaStatement;
546        #[cfg(feature = "wal")]
547        use grafeo_adapters::storage::wal::WalRecord;
548        use grafeo_common::utils::error::{Error, QueryError, QueryErrorKind};
549
550        /// Logs a WAL record for schema changes. Compiles to nothing without `wal`.
551        macro_rules! wal_log {
552            ($self:expr, $record:expr) => {
553                #[cfg(feature = "wal")]
554                $self.log_schema_wal(&$record);
555            };
556        }
557
558        let result = match cmd {
559            SchemaStatement::CreateNodeType(stmt) => {
560                #[cfg(feature = "wal")]
561                let props_for_wal: Vec<(String, String, bool)> = stmt
562                    .properties
563                    .iter()
564                    .map(|p| (p.name.clone(), p.data_type.clone(), p.nullable))
565                    .collect();
566                let def = NodeTypeDefinition {
567                    name: stmt.name.clone(),
568                    properties: stmt
569                        .properties
570                        .iter()
571                        .map(|p| TypedProperty {
572                            name: p.name.clone(),
573                            data_type: PropertyDataType::from_type_name(&p.data_type),
574                            nullable: p.nullable,
575                            default_value: None,
576                        })
577                        .collect(),
578                    constraints: Vec::new(),
579                };
580                let result = if stmt.or_replace {
581                    let _ = self.catalog.drop_node_type(&stmt.name);
582                    self.catalog.register_node_type(def)
583                } else {
584                    self.catalog.register_node_type(def)
585                };
586                match result {
587                    Ok(()) => {
588                        wal_log!(
589                            self,
590                            WalRecord::CreateNodeType {
591                                name: stmt.name.clone(),
592                                properties: props_for_wal,
593                                constraints: Vec::new(),
594                            }
595                        );
596                        Ok(QueryResult::status(format!(
597                            "Created node type '{}'",
598                            stmt.name
599                        )))
600                    }
601                    Err(e) if stmt.if_not_exists => {
602                        let _ = e;
603                        Ok(QueryResult::status("No change"))
604                    }
605                    Err(e) => Err(Error::Query(QueryError::new(
606                        QueryErrorKind::Semantic,
607                        e.to_string(),
608                    ))),
609                }
610            }
611            SchemaStatement::CreateEdgeType(stmt) => {
612                #[cfg(feature = "wal")]
613                let props_for_wal: Vec<(String, String, bool)> = stmt
614                    .properties
615                    .iter()
616                    .map(|p| (p.name.clone(), p.data_type.clone(), p.nullable))
617                    .collect();
618                let def = EdgeTypeDefinition {
619                    name: stmt.name.clone(),
620                    properties: stmt
621                        .properties
622                        .iter()
623                        .map(|p| TypedProperty {
624                            name: p.name.clone(),
625                            data_type: PropertyDataType::from_type_name(&p.data_type),
626                            nullable: p.nullable,
627                            default_value: None,
628                        })
629                        .collect(),
630                    constraints: Vec::new(),
631                };
632                let result = if stmt.or_replace {
633                    let _ = self.catalog.drop_edge_type_def(&stmt.name);
634                    self.catalog.register_edge_type_def(def)
635                } else {
636                    self.catalog.register_edge_type_def(def)
637                };
638                match result {
639                    Ok(()) => {
640                        wal_log!(
641                            self,
642                            WalRecord::CreateEdgeType {
643                                name: stmt.name.clone(),
644                                properties: props_for_wal,
645                                constraints: Vec::new(),
646                            }
647                        );
648                        Ok(QueryResult::status(format!(
649                            "Created edge type '{}'",
650                            stmt.name
651                        )))
652                    }
653                    Err(e) if stmt.if_not_exists => {
654                        let _ = e;
655                        Ok(QueryResult::status("No change"))
656                    }
657                    Err(e) => Err(Error::Query(QueryError::new(
658                        QueryErrorKind::Semantic,
659                        e.to_string(),
660                    ))),
661                }
662            }
663            SchemaStatement::CreateVectorIndex(stmt) => {
664                Self::create_vector_index_on_store(
665                    &self.store,
666                    &stmt.node_label,
667                    &stmt.property,
668                    stmt.dimensions,
669                    stmt.metric.as_deref(),
670                )?;
671                wal_log!(
672                    self,
673                    WalRecord::CreateIndex {
674                        name: stmt.name.clone(),
675                        label: stmt.node_label.clone(),
676                        property: stmt.property.clone(),
677                        index_type: "vector".to_string(),
678                    }
679                );
680                Ok(QueryResult::status(format!(
681                    "Created vector index '{}'",
682                    stmt.name
683                )))
684            }
685            SchemaStatement::DropNodeType { name, if_exists } => {
686                match self.catalog.drop_node_type(&name) {
687                    Ok(()) => {
688                        wal_log!(self, WalRecord::DropNodeType { name: name.clone() });
689                        Ok(QueryResult::status(format!("Dropped node type '{name}'")))
690                    }
691                    Err(e) if if_exists => {
692                        let _ = e;
693                        Ok(QueryResult::status("No change"))
694                    }
695                    Err(e) => Err(Error::Query(QueryError::new(
696                        QueryErrorKind::Semantic,
697                        e.to_string(),
698                    ))),
699                }
700            }
701            SchemaStatement::DropEdgeType { name, if_exists } => {
702                match self.catalog.drop_edge_type_def(&name) {
703                    Ok(()) => {
704                        wal_log!(self, WalRecord::DropEdgeType { name: name.clone() });
705                        Ok(QueryResult::status(format!("Dropped edge type '{name}'")))
706                    }
707                    Err(e) if if_exists => {
708                        let _ = e;
709                        Ok(QueryResult::status("No change"))
710                    }
711                    Err(e) => Err(Error::Query(QueryError::new(
712                        QueryErrorKind::Semantic,
713                        e.to_string(),
714                    ))),
715                }
716            }
717            SchemaStatement::CreateIndex(stmt) => {
718                use grafeo_adapters::query::gql::ast::IndexKind;
719                let index_type_str = match stmt.index_kind {
720                    IndexKind::Property => "property",
721                    IndexKind::BTree => "btree",
722                    IndexKind::Text => "text",
723                    IndexKind::Vector => "vector",
724                };
725                match stmt.index_kind {
726                    IndexKind::Property | IndexKind::BTree => {
727                        for prop in &stmt.properties {
728                            self.store.create_property_index(prop);
729                        }
730                    }
731                    IndexKind::Text => {
732                        for prop in &stmt.properties {
733                            Self::create_text_index_on_store(&self.store, &stmt.label, prop)?;
734                        }
735                    }
736                    IndexKind::Vector => {
737                        for prop in &stmt.properties {
738                            Self::create_vector_index_on_store(
739                                &self.store,
740                                &stmt.label,
741                                prop,
742                                stmt.options.dimensions,
743                                stmt.options.metric.as_deref(),
744                            )?;
745                        }
746                    }
747                }
748                #[cfg(feature = "wal")]
749                for prop in &stmt.properties {
750                    wal_log!(
751                        self,
752                        WalRecord::CreateIndex {
753                            name: stmt.name.clone(),
754                            label: stmt.label.clone(),
755                            property: prop.clone(),
756                            index_type: index_type_str.to_string(),
757                        }
758                    );
759                }
760                Ok(QueryResult::status(format!(
761                    "Created {} index '{}'",
762                    index_type_str, stmt.name
763                )))
764            }
765            SchemaStatement::DropIndex { name, if_exists } => {
766                // Try to drop property index by name
767                let dropped = self.store.drop_property_index(&name);
768                if dropped || if_exists {
769                    if dropped {
770                        wal_log!(self, WalRecord::DropIndex { name: name.clone() });
771                    }
772                    Ok(QueryResult::status(if dropped {
773                        format!("Dropped index '{name}'")
774                    } else {
775                        "No change".to_string()
776                    }))
777                } else {
778                    Err(Error::Query(QueryError::new(
779                        QueryErrorKind::Semantic,
780                        format!("Index '{name}' does not exist"),
781                    )))
782                }
783            }
784            SchemaStatement::CreateConstraint(stmt) => {
785                use grafeo_adapters::query::gql::ast::ConstraintKind;
786                let kind_str = match stmt.constraint_kind {
787                    ConstraintKind::Unique => "unique",
788                    ConstraintKind::NodeKey => "node_key",
789                    ConstraintKind::NotNull => "not_null",
790                    ConstraintKind::Exists => "exists",
791                };
792                let constraint_name = stmt
793                    .name
794                    .clone()
795                    .unwrap_or_else(|| format!("{}_{kind_str}", stmt.label));
796                wal_log!(
797                    self,
798                    WalRecord::CreateConstraint {
799                        name: constraint_name.clone(),
800                        label: stmt.label.clone(),
801                        properties: stmt.properties.clone(),
802                        kind: kind_str.to_string(),
803                    }
804                );
805                Ok(QueryResult::status(format!(
806                    "Created {kind_str} constraint '{constraint_name}'"
807                )))
808            }
809            SchemaStatement::DropConstraint { name, if_exists } => {
810                let _ = if_exists;
811                wal_log!(self, WalRecord::DropConstraint { name: name.clone() });
812                Ok(QueryResult::status(format!("Dropped constraint '{name}'")))
813            }
814            SchemaStatement::CreateGraphType(stmt) => {
815                use crate::catalog::GraphTypeDefinition;
816                use grafeo_adapters::query::gql::ast::InlineElementType;
817
818                // GG04: LIKE clause copies type from existing graph
819                let (mut node_types, mut edge_types, open) =
820                    if let Some(ref like_graph) = stmt.like_graph {
821                        // Infer types from the graph's bound type, or use its existing types
822                        if let Some(type_name) = self.catalog.get_graph_type_binding(like_graph) {
823                            if let Some(existing) = self
824                                .catalog
825                                .schema()
826                                .and_then(|s| s.get_graph_type(&type_name))
827                            {
828                                (
829                                    existing.allowed_node_types.clone(),
830                                    existing.allowed_edge_types.clone(),
831                                    existing.open,
832                                )
833                            } else {
834                                (Vec::new(), Vec::new(), true)
835                            }
836                        } else {
837                            // GG22: Infer from graph data (labels used in graph)
838                            let nt = self.catalog.all_node_type_names();
839                            let et = self.catalog.all_edge_type_names();
840                            if nt.is_empty() && et.is_empty() {
841                                (Vec::new(), Vec::new(), true)
842                            } else {
843                                (nt, et, false)
844                            }
845                        }
846                    } else {
847                        (stmt.node_types.clone(), stmt.edge_types.clone(), stmt.open)
848                    };
849
850                // GG03: Register inline element types and add their names
851                for inline in &stmt.inline_types {
852                    match inline {
853                        InlineElementType::Node {
854                            name, properties, ..
855                        } => {
856                            let def = NodeTypeDefinition {
857                                name: name.clone(),
858                                properties: properties
859                                    .iter()
860                                    .map(|p| TypedProperty {
861                                        name: p.name.clone(),
862                                        data_type: PropertyDataType::from_type_name(&p.data_type),
863                                        nullable: p.nullable,
864                                        default_value: None,
865                                    })
866                                    .collect(),
867                                constraints: Vec::new(),
868                            };
869                            // Register or replace so inline defs override existing
870                            self.catalog.register_or_replace_node_type(def);
871                            #[cfg(feature = "wal")]
872                            {
873                                let props_for_wal: Vec<(String, String, bool)> = properties
874                                    .iter()
875                                    .map(|p| (p.name.clone(), p.data_type.clone(), p.nullable))
876                                    .collect();
877                                self.log_schema_wal(&WalRecord::CreateNodeType {
878                                    name: name.clone(),
879                                    properties: props_for_wal,
880                                    constraints: Vec::new(),
881                                });
882                            }
883                            if !node_types.contains(name) {
884                                node_types.push(name.clone());
885                            }
886                        }
887                        InlineElementType::Edge {
888                            name, properties, ..
889                        } => {
890                            let def = EdgeTypeDefinition {
891                                name: name.clone(),
892                                properties: properties
893                                    .iter()
894                                    .map(|p| TypedProperty {
895                                        name: p.name.clone(),
896                                        data_type: PropertyDataType::from_type_name(&p.data_type),
897                                        nullable: p.nullable,
898                                        default_value: None,
899                                    })
900                                    .collect(),
901                                constraints: Vec::new(),
902                            };
903                            self.catalog.register_or_replace_edge_type_def(def);
904                            #[cfg(feature = "wal")]
905                            {
906                                let props_for_wal: Vec<(String, String, bool)> = properties
907                                    .iter()
908                                    .map(|p| (p.name.clone(), p.data_type.clone(), p.nullable))
909                                    .collect();
910                                self.log_schema_wal(&WalRecord::CreateEdgeType {
911                                    name: name.clone(),
912                                    properties: props_for_wal,
913                                    constraints: Vec::new(),
914                                });
915                            }
916                            if !edge_types.contains(name) {
917                                edge_types.push(name.clone());
918                            }
919                        }
920                    }
921                }
922
923                let def = GraphTypeDefinition {
924                    name: stmt.name.clone(),
925                    allowed_node_types: node_types.clone(),
926                    allowed_edge_types: edge_types.clone(),
927                    open,
928                };
929                let result = if stmt.or_replace {
930                    // Drop existing first, ignore error if not found
931                    let _ = self.catalog.drop_graph_type(&stmt.name);
932                    self.catalog.register_graph_type(def)
933                } else {
934                    self.catalog.register_graph_type(def)
935                };
936                match result {
937                    Ok(()) => {
938                        wal_log!(
939                            self,
940                            WalRecord::CreateGraphType {
941                                name: stmt.name.clone(),
942                                node_types,
943                                edge_types,
944                                open,
945                            }
946                        );
947                        Ok(QueryResult::status(format!(
948                            "Created graph type '{}'",
949                            stmt.name
950                        )))
951                    }
952                    Err(e) if stmt.if_not_exists => {
953                        let _ = e;
954                        Ok(QueryResult::status("No change"))
955                    }
956                    Err(e) => Err(Error::Query(QueryError::new(
957                        QueryErrorKind::Semantic,
958                        e.to_string(),
959                    ))),
960                }
961            }
962            SchemaStatement::DropGraphType { name, if_exists } => {
963                match self.catalog.drop_graph_type(&name) {
964                    Ok(()) => {
965                        wal_log!(self, WalRecord::DropGraphType { name: name.clone() });
966                        Ok(QueryResult::status(format!("Dropped graph type '{name}'")))
967                    }
968                    Err(e) if if_exists => {
969                        let _ = e;
970                        Ok(QueryResult::status("No change"))
971                    }
972                    Err(e) => Err(Error::Query(QueryError::new(
973                        QueryErrorKind::Semantic,
974                        e.to_string(),
975                    ))),
976                }
977            }
978            SchemaStatement::CreateSchema {
979                name,
980                if_not_exists,
981            } => match self.catalog.register_schema_namespace(name.clone()) {
982                Ok(()) => {
983                    wal_log!(self, WalRecord::CreateSchema { name: name.clone() });
984                    Ok(QueryResult::status(format!("Created schema '{name}'")))
985                }
986                Err(e) if if_not_exists => {
987                    let _ = e;
988                    Ok(QueryResult::status("No change"))
989                }
990                Err(e) => Err(Error::Query(QueryError::new(
991                    QueryErrorKind::Semantic,
992                    e.to_string(),
993                ))),
994            },
995            SchemaStatement::DropSchema { name, if_exists } => {
996                match self.catalog.drop_schema_namespace(&name) {
997                    Ok(()) => {
998                        wal_log!(self, WalRecord::DropSchema { name: name.clone() });
999                        Ok(QueryResult::status(format!("Dropped schema '{name}'")))
1000                    }
1001                    Err(e) if if_exists => {
1002                        let _ = e;
1003                        Ok(QueryResult::status("No change"))
1004                    }
1005                    Err(e) => Err(Error::Query(QueryError::new(
1006                        QueryErrorKind::Semantic,
1007                        e.to_string(),
1008                    ))),
1009                }
1010            }
1011            SchemaStatement::AlterNodeType(stmt) => {
1012                use grafeo_adapters::query::gql::ast::TypeAlteration;
1013                let mut wal_alts = Vec::new();
1014                for alt in &stmt.alterations {
1015                    match alt {
1016                        TypeAlteration::AddProperty(prop) => {
1017                            let typed = TypedProperty {
1018                                name: prop.name.clone(),
1019                                data_type: PropertyDataType::from_type_name(&prop.data_type),
1020                                nullable: prop.nullable,
1021                                default_value: None,
1022                            };
1023                            self.catalog
1024                                .alter_node_type_add_property(&stmt.name, typed)
1025                                .map_err(|e| {
1026                                    Error::Query(QueryError::new(
1027                                        QueryErrorKind::Semantic,
1028                                        e.to_string(),
1029                                    ))
1030                                })?;
1031                            wal_alts.push((
1032                                "add".to_string(),
1033                                prop.name.clone(),
1034                                prop.data_type.clone(),
1035                                prop.nullable,
1036                            ));
1037                        }
1038                        TypeAlteration::DropProperty(name) => {
1039                            self.catalog
1040                                .alter_node_type_drop_property(&stmt.name, name)
1041                                .map_err(|e| {
1042                                    Error::Query(QueryError::new(
1043                                        QueryErrorKind::Semantic,
1044                                        e.to_string(),
1045                                    ))
1046                                })?;
1047                            wal_alts.push(("drop".to_string(), name.clone(), String::new(), false));
1048                        }
1049                    }
1050                }
1051                wal_log!(
1052                    self,
1053                    WalRecord::AlterNodeType {
1054                        name: stmt.name.clone(),
1055                        alterations: wal_alts,
1056                    }
1057                );
1058                Ok(QueryResult::status(format!(
1059                    "Altered node type '{}'",
1060                    stmt.name
1061                )))
1062            }
1063            SchemaStatement::AlterEdgeType(stmt) => {
1064                use grafeo_adapters::query::gql::ast::TypeAlteration;
1065                let mut wal_alts = Vec::new();
1066                for alt in &stmt.alterations {
1067                    match alt {
1068                        TypeAlteration::AddProperty(prop) => {
1069                            let typed = TypedProperty {
1070                                name: prop.name.clone(),
1071                                data_type: PropertyDataType::from_type_name(&prop.data_type),
1072                                nullable: prop.nullable,
1073                                default_value: None,
1074                            };
1075                            self.catalog
1076                                .alter_edge_type_add_property(&stmt.name, typed)
1077                                .map_err(|e| {
1078                                    Error::Query(QueryError::new(
1079                                        QueryErrorKind::Semantic,
1080                                        e.to_string(),
1081                                    ))
1082                                })?;
1083                            wal_alts.push((
1084                                "add".to_string(),
1085                                prop.name.clone(),
1086                                prop.data_type.clone(),
1087                                prop.nullable,
1088                            ));
1089                        }
1090                        TypeAlteration::DropProperty(name) => {
1091                            self.catalog
1092                                .alter_edge_type_drop_property(&stmt.name, name)
1093                                .map_err(|e| {
1094                                    Error::Query(QueryError::new(
1095                                        QueryErrorKind::Semantic,
1096                                        e.to_string(),
1097                                    ))
1098                                })?;
1099                            wal_alts.push(("drop".to_string(), name.clone(), String::new(), false));
1100                        }
1101                    }
1102                }
1103                wal_log!(
1104                    self,
1105                    WalRecord::AlterEdgeType {
1106                        name: stmt.name.clone(),
1107                        alterations: wal_alts,
1108                    }
1109                );
1110                Ok(QueryResult::status(format!(
1111                    "Altered edge type '{}'",
1112                    stmt.name
1113                )))
1114            }
1115            SchemaStatement::AlterGraphType(stmt) => {
1116                use grafeo_adapters::query::gql::ast::GraphTypeAlteration;
1117                let mut wal_alts = Vec::new();
1118                for alt in &stmt.alterations {
1119                    match alt {
1120                        GraphTypeAlteration::AddNodeType(name) => {
1121                            self.catalog
1122                                .alter_graph_type_add_node_type(&stmt.name, name.clone())
1123                                .map_err(|e| {
1124                                    Error::Query(QueryError::new(
1125                                        QueryErrorKind::Semantic,
1126                                        e.to_string(),
1127                                    ))
1128                                })?;
1129                            wal_alts.push(("add_node_type".to_string(), name.clone()));
1130                        }
1131                        GraphTypeAlteration::DropNodeType(name) => {
1132                            self.catalog
1133                                .alter_graph_type_drop_node_type(&stmt.name, name)
1134                                .map_err(|e| {
1135                                    Error::Query(QueryError::new(
1136                                        QueryErrorKind::Semantic,
1137                                        e.to_string(),
1138                                    ))
1139                                })?;
1140                            wal_alts.push(("drop_node_type".to_string(), name.clone()));
1141                        }
1142                        GraphTypeAlteration::AddEdgeType(name) => {
1143                            self.catalog
1144                                .alter_graph_type_add_edge_type(&stmt.name, name.clone())
1145                                .map_err(|e| {
1146                                    Error::Query(QueryError::new(
1147                                        QueryErrorKind::Semantic,
1148                                        e.to_string(),
1149                                    ))
1150                                })?;
1151                            wal_alts.push(("add_edge_type".to_string(), name.clone()));
1152                        }
1153                        GraphTypeAlteration::DropEdgeType(name) => {
1154                            self.catalog
1155                                .alter_graph_type_drop_edge_type(&stmt.name, name)
1156                                .map_err(|e| {
1157                                    Error::Query(QueryError::new(
1158                                        QueryErrorKind::Semantic,
1159                                        e.to_string(),
1160                                    ))
1161                                })?;
1162                            wal_alts.push(("drop_edge_type".to_string(), name.clone()));
1163                        }
1164                    }
1165                }
1166                wal_log!(
1167                    self,
1168                    WalRecord::AlterGraphType {
1169                        name: stmt.name.clone(),
1170                        alterations: wal_alts,
1171                    }
1172                );
1173                Ok(QueryResult::status(format!(
1174                    "Altered graph type '{}'",
1175                    stmt.name
1176                )))
1177            }
1178            SchemaStatement::CreateProcedure(stmt) => {
1179                use crate::catalog::ProcedureDefinition;
1180
1181                let def = ProcedureDefinition {
1182                    name: stmt.name.clone(),
1183                    params: stmt
1184                        .params
1185                        .iter()
1186                        .map(|p| (p.name.clone(), p.param_type.clone()))
1187                        .collect(),
1188                    returns: stmt
1189                        .returns
1190                        .iter()
1191                        .map(|r| (r.name.clone(), r.return_type.clone()))
1192                        .collect(),
1193                    body: stmt.body.clone(),
1194                };
1195
1196                if stmt.or_replace {
1197                    self.catalog.replace_procedure(def).map_err(|e| {
1198                        Error::Query(QueryError::new(QueryErrorKind::Semantic, e.to_string()))
1199                    })?;
1200                } else {
1201                    match self.catalog.register_procedure(def) {
1202                        Ok(()) => {}
1203                        Err(_) if stmt.if_not_exists => {
1204                            return Ok(QueryResult::empty());
1205                        }
1206                        Err(e) => {
1207                            return Err(Error::Query(QueryError::new(
1208                                QueryErrorKind::Semantic,
1209                                e.to_string(),
1210                            )));
1211                        }
1212                    }
1213                }
1214
1215                wal_log!(
1216                    self,
1217                    WalRecord::CreateProcedure {
1218                        name: stmt.name.clone(),
1219                        params: stmt
1220                            .params
1221                            .iter()
1222                            .map(|p| (p.name.clone(), p.param_type.clone()))
1223                            .collect(),
1224                        returns: stmt
1225                            .returns
1226                            .iter()
1227                            .map(|r| (r.name.clone(), r.return_type.clone()))
1228                            .collect(),
1229                        body: stmt.body,
1230                    }
1231                );
1232                Ok(QueryResult::status(format!(
1233                    "Created procedure '{}'",
1234                    stmt.name
1235                )))
1236            }
1237            SchemaStatement::DropProcedure { name, if_exists } => {
1238                match self.catalog.drop_procedure(&name) {
1239                    Ok(()) => {}
1240                    Err(_) if if_exists => {
1241                        return Ok(QueryResult::empty());
1242                    }
1243                    Err(e) => {
1244                        return Err(Error::Query(QueryError::new(
1245                            QueryErrorKind::Semantic,
1246                            e.to_string(),
1247                        )));
1248                    }
1249                }
1250                wal_log!(self, WalRecord::DropProcedure { name: name.clone() });
1251                Ok(QueryResult::status(format!("Dropped procedure '{name}'")))
1252            }
1253        };
1254
1255        // Invalidate all cached query plans after any successful DDL change.
1256        // DDL is rare, so clearing the entire cache is cheap and correct.
1257        if result.is_ok() {
1258            self.query_cache.clear();
1259        }
1260
1261        result
1262    }
1263
1264    /// Creates a vector index on the store by scanning existing nodes.
1265    #[cfg(all(feature = "gql", feature = "vector-index"))]
1266    fn create_vector_index_on_store(
1267        store: &LpgStore,
1268        label: &str,
1269        property: &str,
1270        dimensions: Option<usize>,
1271        metric: Option<&str>,
1272    ) -> Result<()> {
1273        use grafeo_common::types::{PropertyKey, Value};
1274        use grafeo_common::utils::error::Error;
1275        use grafeo_core::index::vector::{DistanceMetric, HnswConfig, HnswIndex};
1276
1277        let metric = match metric {
1278            Some(m) => DistanceMetric::from_str(m).ok_or_else(|| {
1279                Error::Internal(format!(
1280                    "Unknown distance metric '{m}'. Use: cosine, euclidean, dot_product, manhattan"
1281                ))
1282            })?,
1283            None => DistanceMetric::Cosine,
1284        };
1285
1286        let prop_key = PropertyKey::new(property);
1287        let mut found_dims: Option<usize> = dimensions;
1288        let mut vectors: Vec<(grafeo_common::types::NodeId, Vec<f32>)> = Vec::new();
1289
1290        for node in store.nodes_with_label(label) {
1291            if let Some(Value::Vector(v)) = node.properties.get(&prop_key) {
1292                if let Some(expected) = found_dims {
1293                    if v.len() != expected {
1294                        return Err(Error::Internal(format!(
1295                            "Vector dimension mismatch: expected {expected}, found {} on node {}",
1296                            v.len(),
1297                            node.id.0
1298                        )));
1299                    }
1300                } else {
1301                    found_dims = Some(v.len());
1302                }
1303                vectors.push((node.id, v.to_vec()));
1304            }
1305        }
1306
1307        let Some(dims) = found_dims else {
1308            return Err(Error::Internal(format!(
1309                "No vector properties found on :{label}({property}) and no dimensions specified"
1310            )));
1311        };
1312
1313        let config = HnswConfig::new(dims, metric);
1314        let index = HnswIndex::with_capacity(config, vectors.len());
1315        let accessor = grafeo_core::index::vector::PropertyVectorAccessor::new(store, property);
1316        for (node_id, vec) in &vectors {
1317            index.insert(*node_id, vec, &accessor);
1318        }
1319
1320        store.add_vector_index(label, property, Arc::new(index));
1321        Ok(())
1322    }
1323
1324    /// Stub for when vector-index feature is not enabled.
1325    #[cfg(all(feature = "gql", not(feature = "vector-index")))]
1326    fn create_vector_index_on_store(
1327        _store: &LpgStore,
1328        _label: &str,
1329        _property: &str,
1330        _dimensions: Option<usize>,
1331        _metric: Option<&str>,
1332    ) -> Result<()> {
1333        Err(grafeo_common::utils::error::Error::Internal(
1334            "Vector index support requires the 'vector-index' feature".to_string(),
1335        ))
1336    }
1337
1338    /// Creates a text index on the store by scanning existing nodes.
1339    #[cfg(all(feature = "gql", feature = "text-index"))]
1340    fn create_text_index_on_store(store: &LpgStore, label: &str, property: &str) -> Result<()> {
1341        use grafeo_common::types::{PropertyKey, Value};
1342        use grafeo_core::index::text::{BM25Config, InvertedIndex};
1343
1344        let mut index = InvertedIndex::new(BM25Config::default());
1345        let prop_key = PropertyKey::new(property);
1346
1347        let nodes = store.nodes_by_label(label);
1348        for node_id in nodes {
1349            if let Some(Value::String(text)) = store.get_node_property(node_id, &prop_key) {
1350                index.insert(node_id, text.as_str());
1351            }
1352        }
1353
1354        store.add_text_index(label, property, Arc::new(parking_lot::RwLock::new(index)));
1355        Ok(())
1356    }
1357
1358    /// Stub for when text-index feature is not enabled.
1359    #[cfg(all(feature = "gql", not(feature = "text-index")))]
1360    fn create_text_index_on_store(_store: &LpgStore, _label: &str, _property: &str) -> Result<()> {
1361        Err(grafeo_common::utils::error::Error::Internal(
1362            "Text index support requires the 'text-index' feature".to_string(),
1363        ))
1364    }
1365
1366    /// Returns a table of all indexes from the catalog.
1367    fn execute_show_indexes(&self) -> Result<QueryResult> {
1368        let indexes = self.catalog.all_indexes();
1369        let columns = vec![
1370            "name".to_string(),
1371            "type".to_string(),
1372            "label".to_string(),
1373            "property".to_string(),
1374        ];
1375        let rows: Vec<Vec<Value>> = indexes
1376            .into_iter()
1377            .map(|def| {
1378                let label_name = self
1379                    .catalog
1380                    .get_label_name(def.label)
1381                    .unwrap_or_else(|| "?".into());
1382                let prop_name = self
1383                    .catalog
1384                    .get_property_key_name(def.property_key)
1385                    .unwrap_or_else(|| "?".into());
1386                vec![
1387                    Value::from(format!("idx_{}_{}", label_name, prop_name)),
1388                    Value::from(format!("{:?}", def.index_type)),
1389                    Value::from(&*label_name),
1390                    Value::from(&*prop_name),
1391                ]
1392            })
1393            .collect();
1394        Ok(QueryResult {
1395            columns,
1396            column_types: Vec::new(),
1397            rows,
1398            ..QueryResult::empty()
1399        })
1400    }
1401
1402    /// Returns a table of all constraints (currently metadata-only).
1403    fn execute_show_constraints(&self) -> Result<QueryResult> {
1404        // Constraints are tracked in WAL but not yet in a queryable catalog.
1405        // Return an empty table with the expected schema.
1406        Ok(QueryResult {
1407            columns: vec![
1408                "name".to_string(),
1409                "type".to_string(),
1410                "label".to_string(),
1411                "properties".to_string(),
1412            ],
1413            column_types: Vec::new(),
1414            rows: Vec::new(),
1415            ..QueryResult::empty()
1416        })
1417    }
1418
1419    /// Executes a GQL query.
1420    ///
1421    /// # Errors
1422    ///
1423    /// Returns an error if the query fails to parse or execute.
1424    ///
1425    /// # Examples
1426    ///
1427    /// ```no_run
1428    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
1429    /// use grafeo_engine::GrafeoDB;
1430    ///
1431    /// let db = GrafeoDB::new_in_memory();
1432    /// let session = db.session();
1433    ///
1434    /// // Create a node
1435    /// session.execute("INSERT (:Person {name: 'Alix', age: 30})")?;
1436    ///
1437    /// // Query nodes
1438    /// let result = session.execute("MATCH (n:Person) RETURN n.name, n.age")?;
1439    /// for row in &result.rows {
1440    ///     println!("{:?}", row);
1441    /// }
1442    /// # Ok(())
1443    /// # }
1444    /// ```
1445    #[cfg(feature = "gql")]
1446    pub fn execute(&self, query: &str) -> Result<QueryResult> {
1447        self.require_lpg("GQL")?;
1448
1449        use crate::query::{
1450            Executor, binder::Binder, cache::CacheKey, optimizer::Optimizer,
1451            processor::QueryLanguage, translators::gql,
1452        };
1453
1454        #[cfg(not(target_arch = "wasm32"))]
1455        let start_time = std::time::Instant::now();
1456
1457        // Parse and translate, checking for session/schema commands first
1458        let translation = gql::translate_full(query)?;
1459        let logical_plan = match translation {
1460            gql::GqlTranslationResult::SessionCommand(cmd) => {
1461                return self.execute_session_command(cmd);
1462            }
1463            gql::GqlTranslationResult::SchemaCommand(cmd) => {
1464                // All DDL is a write operation
1465                if *self.read_only_tx.lock() {
1466                    return Err(grafeo_common::utils::error::Error::Transaction(
1467                        grafeo_common::utils::error::TransactionError::ReadOnly,
1468                    ));
1469                }
1470                return self.execute_schema_command(cmd);
1471            }
1472            gql::GqlTranslationResult::Plan(plan) => {
1473                // Block mutations in read-only transactions
1474                if *self.read_only_tx.lock() && plan.root.has_mutations() {
1475                    return Err(grafeo_common::utils::error::Error::Transaction(
1476                        grafeo_common::utils::error::TransactionError::ReadOnly,
1477                    ));
1478                }
1479                plan
1480            }
1481        };
1482
1483        // Create cache key for this query
1484        let cache_key = CacheKey::new(query, QueryLanguage::Gql);
1485
1486        // Try to get cached optimized plan, or use the plan we just translated
1487        let optimized_plan = if let Some(cached_plan) = self.query_cache.get_optimized(&cache_key) {
1488            cached_plan
1489        } else {
1490            // Semantic validation
1491            let mut binder = Binder::new();
1492            let _binding_context = binder.bind(&logical_plan)?;
1493
1494            // Optimize the plan
1495            let optimizer = Optimizer::from_graph_store(&*self.graph_store);
1496            let plan = optimizer.optimize(logical_plan)?;
1497
1498            // Cache the optimized plan for future use
1499            self.query_cache.put_optimized(cache_key, plan.clone());
1500
1501            plan
1502        };
1503
1504        // EXPLAIN: annotate pushdown hints and return the plan tree
1505        if optimized_plan.explain {
1506            use crate::query::processor::{annotate_pushdown_hints, explain_result};
1507            let mut plan = optimized_plan;
1508            annotate_pushdown_hints(&mut plan.root, self.graph_store.as_ref());
1509            return Ok(explain_result(&plan));
1510        }
1511
1512        // PROFILE: execute with per-operator instrumentation
1513        if optimized_plan.profile {
1514            let has_mutations = optimized_plan.root.has_mutations();
1515            return self.with_auto_commit(has_mutations, || {
1516                let (viewing_epoch, transaction_id) = self.get_transaction_context();
1517                let planner = self.create_planner(viewing_epoch, transaction_id);
1518                let (mut physical_plan, entries) = planner.plan_profiled(&optimized_plan)?;
1519
1520                let executor = Executor::with_columns(physical_plan.columns.clone())
1521                    .with_deadline(self.query_deadline());
1522                let _result = executor.execute(physical_plan.operator.as_mut())?;
1523
1524                let total_time_ms;
1525                #[cfg(not(target_arch = "wasm32"))]
1526                {
1527                    total_time_ms = start_time.elapsed().as_secs_f64() * 1000.0;
1528                }
1529                #[cfg(target_arch = "wasm32")]
1530                {
1531                    total_time_ms = 0.0;
1532                }
1533
1534                let profile_tree = crate::query::profile::build_profile_tree(
1535                    &optimized_plan.root,
1536                    &mut entries.into_iter(),
1537                );
1538                Ok(crate::query::profile::profile_result(
1539                    &profile_tree,
1540                    total_time_ms,
1541                ))
1542            });
1543        }
1544
1545        let has_mutations = optimized_plan.root.has_mutations();
1546
1547        self.with_auto_commit(has_mutations, || {
1548            // Get transaction context for MVCC visibility
1549            let (viewing_epoch, transaction_id) = self.get_transaction_context();
1550
1551            // Convert to physical plan with transaction context
1552            // (Physical planning cannot be cached as it depends on transaction state)
1553            let planner = self.create_planner(viewing_epoch, transaction_id);
1554            let mut physical_plan = planner.plan(&optimized_plan)?;
1555
1556            // Execute the plan
1557            let executor = Executor::with_columns(physical_plan.columns.clone())
1558                .with_deadline(self.query_deadline());
1559            let mut result = executor.execute(physical_plan.operator.as_mut())?;
1560
1561            // Add execution metrics
1562            let rows_scanned = result.rows.len() as u64;
1563            #[cfg(not(target_arch = "wasm32"))]
1564            {
1565                let elapsed_ms = start_time.elapsed().as_secs_f64() * 1000.0;
1566                result.execution_time_ms = Some(elapsed_ms);
1567            }
1568            result.rows_scanned = Some(rows_scanned);
1569
1570            Ok(result)
1571        })
1572    }
1573
1574    /// Executes a GQL query with visibility at the specified epoch.
1575    ///
1576    /// This enables time-travel queries: the query sees the database
1577    /// as it existed at the given epoch.
1578    ///
1579    /// # Errors
1580    ///
1581    /// Returns an error if parsing or execution fails.
1582    #[cfg(feature = "gql")]
1583    pub fn execute_at_epoch(&self, query: &str, epoch: EpochId) -> Result<QueryResult> {
1584        let previous = self.viewing_epoch_override.lock().replace(epoch);
1585        let result = self.execute(query);
1586        *self.viewing_epoch_override.lock() = previous;
1587        result
1588    }
1589
1590    /// Executes a GQL query with parameters.
1591    ///
1592    /// # Errors
1593    ///
1594    /// Returns an error if the query fails to parse or execute.
1595    #[cfg(feature = "gql")]
1596    pub fn execute_with_params(
1597        &self,
1598        query: &str,
1599        params: std::collections::HashMap<String, Value>,
1600    ) -> Result<QueryResult> {
1601        self.require_lpg("GQL")?;
1602
1603        use crate::query::processor::{QueryLanguage, QueryProcessor};
1604
1605        let has_mutations = Self::query_looks_like_mutation(query);
1606
1607        self.with_auto_commit(has_mutations, || {
1608            // Get transaction context for MVCC visibility
1609            let (viewing_epoch, transaction_id) = self.get_transaction_context();
1610
1611            // Create processor with transaction context
1612            let processor = QueryProcessor::for_graph_store_with_transaction(
1613                Arc::clone(&self.graph_store),
1614                Arc::clone(&self.transaction_manager),
1615            );
1616
1617            // Apply transaction context if in a transaction
1618            let processor = if let Some(transaction_id) = transaction_id {
1619                processor.with_transaction_context(viewing_epoch, transaction_id)
1620            } else {
1621                processor
1622            };
1623
1624            processor.process(query, QueryLanguage::Gql, Some(&params))
1625        })
1626    }
1627
1628    /// Executes a GQL query with parameters.
1629    ///
1630    /// # Errors
1631    ///
1632    /// Returns an error if no query language is enabled.
1633    #[cfg(not(any(feature = "gql", feature = "cypher")))]
1634    pub fn execute_with_params(
1635        &self,
1636        _query: &str,
1637        _params: std::collections::HashMap<String, Value>,
1638    ) -> Result<QueryResult> {
1639        Err(grafeo_common::utils::error::Error::Internal(
1640            "No query language enabled".to_string(),
1641        ))
1642    }
1643
1644    /// Executes a GQL query.
1645    ///
1646    /// # Errors
1647    ///
1648    /// Returns an error if no query language is enabled.
1649    #[cfg(not(any(feature = "gql", feature = "cypher")))]
1650    pub fn execute(&self, _query: &str) -> Result<QueryResult> {
1651        Err(grafeo_common::utils::error::Error::Internal(
1652            "No query language enabled".to_string(),
1653        ))
1654    }
1655
1656    /// Executes a Cypher query.
1657    ///
1658    /// # Errors
1659    ///
1660    /// Returns an error if the query fails to parse or execute.
1661    #[cfg(feature = "cypher")]
1662    pub fn execute_cypher(&self, query: &str) -> Result<QueryResult> {
1663        use crate::query::{
1664            Executor, binder::Binder, cache::CacheKey, optimizer::Optimizer,
1665            processor::QueryLanguage, translators::cypher,
1666        };
1667        use grafeo_common::utils::error::{Error as GrafeoError, QueryError, QueryErrorKind};
1668
1669        // Handle schema DDL and SHOW commands before the normal query path
1670        let translation = cypher::translate_full(query)?;
1671        match translation {
1672            cypher::CypherTranslationResult::SchemaCommand(cmd) => {
1673                if *self.read_only_tx.lock() {
1674                    return Err(GrafeoError::Query(QueryError::new(
1675                        QueryErrorKind::Semantic,
1676                        "Cannot execute schema DDL in a read-only transaction",
1677                    )));
1678                }
1679                return self.execute_schema_command(cmd);
1680            }
1681            cypher::CypherTranslationResult::ShowIndexes => {
1682                return self.execute_show_indexes();
1683            }
1684            cypher::CypherTranslationResult::ShowConstraints => {
1685                return self.execute_show_constraints();
1686            }
1687            cypher::CypherTranslationResult::Plan(_) => {
1688                // Fall through to normal execution below
1689            }
1690        }
1691
1692        #[cfg(not(target_arch = "wasm32"))]
1693        let start_time = std::time::Instant::now();
1694
1695        // Create cache key for this query
1696        let cache_key = CacheKey::new(query, QueryLanguage::Cypher);
1697
1698        // Try to get cached optimized plan
1699        let optimized_plan = if let Some(cached_plan) = self.query_cache.get_optimized(&cache_key) {
1700            cached_plan
1701        } else {
1702            // Parse and translate the query to a logical plan
1703            let logical_plan = cypher::translate(query)?;
1704
1705            // Semantic validation
1706            let mut binder = Binder::new();
1707            let _binding_context = binder.bind(&logical_plan)?;
1708
1709            // Optimize the plan
1710            let optimizer = Optimizer::from_graph_store(&*self.graph_store);
1711            let plan = optimizer.optimize(logical_plan)?;
1712
1713            // Cache the optimized plan
1714            self.query_cache.put_optimized(cache_key, plan.clone());
1715
1716            plan
1717        };
1718
1719        // EXPLAIN
1720        if optimized_plan.explain {
1721            use crate::query::processor::{annotate_pushdown_hints, explain_result};
1722            let mut plan = optimized_plan;
1723            annotate_pushdown_hints(&mut plan.root, self.graph_store.as_ref());
1724            return Ok(explain_result(&plan));
1725        }
1726
1727        // PROFILE
1728        if optimized_plan.profile {
1729            let has_mutations = optimized_plan.root.has_mutations();
1730            return self.with_auto_commit(has_mutations, || {
1731                let (viewing_epoch, transaction_id) = self.get_transaction_context();
1732                let planner = self.create_planner(viewing_epoch, transaction_id);
1733                let (mut physical_plan, entries) = planner.plan_profiled(&optimized_plan)?;
1734
1735                let executor = Executor::with_columns(physical_plan.columns.clone())
1736                    .with_deadline(self.query_deadline());
1737                let _result = executor.execute(physical_plan.operator.as_mut())?;
1738
1739                let total_time_ms;
1740                #[cfg(not(target_arch = "wasm32"))]
1741                {
1742                    total_time_ms = start_time.elapsed().as_secs_f64() * 1000.0;
1743                }
1744                #[cfg(target_arch = "wasm32")]
1745                {
1746                    total_time_ms = 0.0;
1747                }
1748
1749                let profile_tree = crate::query::profile::build_profile_tree(
1750                    &optimized_plan.root,
1751                    &mut entries.into_iter(),
1752                );
1753                Ok(crate::query::profile::profile_result(
1754                    &profile_tree,
1755                    total_time_ms,
1756                ))
1757            });
1758        }
1759
1760        let has_mutations = optimized_plan.root.has_mutations();
1761
1762        self.with_auto_commit(has_mutations, || {
1763            // Get transaction context for MVCC visibility
1764            let (viewing_epoch, transaction_id) = self.get_transaction_context();
1765
1766            // Convert to physical plan with transaction context
1767            let planner = self.create_planner(viewing_epoch, transaction_id);
1768            let mut physical_plan = planner.plan(&optimized_plan)?;
1769
1770            // Execute the plan
1771            let executor = Executor::with_columns(physical_plan.columns.clone())
1772                .with_deadline(self.query_deadline());
1773            executor.execute(physical_plan.operator.as_mut())
1774        })
1775    }
1776
1777    /// Executes a Gremlin query.
1778    ///
1779    /// # Errors
1780    ///
1781    /// Returns an error if the query fails to parse or execute.
1782    ///
1783    /// # Examples
1784    ///
1785    /// ```no_run
1786    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
1787    /// use grafeo_engine::GrafeoDB;
1788    ///
1789    /// let db = GrafeoDB::new_in_memory();
1790    /// let session = db.session();
1791    ///
1792    /// // Create some nodes first
1793    /// session.create_node(&["Person"]);
1794    ///
1795    /// // Query using Gremlin
1796    /// let result = session.execute_gremlin("g.V().hasLabel('Person')")?;
1797    /// # Ok(())
1798    /// # }
1799    /// ```
1800    #[cfg(feature = "gremlin")]
1801    pub fn execute_gremlin(&self, query: &str) -> Result<QueryResult> {
1802        use crate::query::{Executor, binder::Binder, optimizer::Optimizer, translators::gremlin};
1803
1804        // Parse and translate the query to a logical plan
1805        let logical_plan = gremlin::translate(query)?;
1806
1807        // Semantic validation
1808        let mut binder = Binder::new();
1809        let _binding_context = binder.bind(&logical_plan)?;
1810
1811        // Optimize the plan
1812        let optimizer = Optimizer::from_graph_store(&*self.graph_store);
1813        let optimized_plan = optimizer.optimize(logical_plan)?;
1814
1815        let has_mutations = optimized_plan.root.has_mutations();
1816
1817        self.with_auto_commit(has_mutations, || {
1818            // Get transaction context for MVCC visibility
1819            let (viewing_epoch, transaction_id) = self.get_transaction_context();
1820
1821            // Convert to physical plan with transaction context
1822            let planner = self.create_planner(viewing_epoch, transaction_id);
1823            let mut physical_plan = planner.plan(&optimized_plan)?;
1824
1825            // Execute the plan
1826            let executor = Executor::with_columns(physical_plan.columns.clone())
1827                .with_deadline(self.query_deadline());
1828            executor.execute(physical_plan.operator.as_mut())
1829        })
1830    }
1831
1832    /// Executes a Gremlin query with parameters.
1833    ///
1834    /// # Errors
1835    ///
1836    /// Returns an error if the query fails to parse or execute.
1837    #[cfg(feature = "gremlin")]
1838    pub fn execute_gremlin_with_params(
1839        &self,
1840        query: &str,
1841        params: std::collections::HashMap<String, Value>,
1842    ) -> Result<QueryResult> {
1843        use crate::query::processor::{QueryLanguage, QueryProcessor};
1844
1845        let has_mutations = Self::query_looks_like_mutation(query);
1846
1847        self.with_auto_commit(has_mutations, || {
1848            // Get transaction context for MVCC visibility
1849            let (viewing_epoch, transaction_id) = self.get_transaction_context();
1850
1851            // Create processor with transaction context
1852            let processor = QueryProcessor::for_graph_store_with_transaction(
1853                Arc::clone(&self.graph_store),
1854                Arc::clone(&self.transaction_manager),
1855            );
1856
1857            // Apply transaction context if in a transaction
1858            let processor = if let Some(transaction_id) = transaction_id {
1859                processor.with_transaction_context(viewing_epoch, transaction_id)
1860            } else {
1861                processor
1862            };
1863
1864            processor.process(query, QueryLanguage::Gremlin, Some(&params))
1865        })
1866    }
1867
1868    /// Executes a GraphQL query against the LPG store.
1869    ///
1870    /// # Errors
1871    ///
1872    /// Returns an error if the query fails to parse or execute.
1873    ///
1874    /// # Examples
1875    ///
1876    /// ```no_run
1877    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
1878    /// use grafeo_engine::GrafeoDB;
1879    ///
1880    /// let db = GrafeoDB::new_in_memory();
1881    /// let session = db.session();
1882    ///
1883    /// // Create some nodes first
1884    /// session.create_node(&["User"]);
1885    ///
1886    /// // Query using GraphQL
1887    /// let result = session.execute_graphql("query { user { id name } }")?;
1888    /// # Ok(())
1889    /// # }
1890    /// ```
1891    #[cfg(feature = "graphql")]
1892    pub fn execute_graphql(&self, query: &str) -> Result<QueryResult> {
1893        use crate::query::{Executor, binder::Binder, optimizer::Optimizer, translators::graphql};
1894
1895        // Parse and translate the query to a logical plan
1896        let logical_plan = graphql::translate(query)?;
1897
1898        // Semantic validation
1899        let mut binder = Binder::new();
1900        let _binding_context = binder.bind(&logical_plan)?;
1901
1902        // Optimize the plan
1903        let optimizer = Optimizer::from_graph_store(&*self.graph_store);
1904        let optimized_plan = optimizer.optimize(logical_plan)?;
1905
1906        let has_mutations = optimized_plan.root.has_mutations();
1907
1908        self.with_auto_commit(has_mutations, || {
1909            // Get transaction context for MVCC visibility
1910            let (viewing_epoch, transaction_id) = self.get_transaction_context();
1911
1912            // Convert to physical plan with transaction context
1913            let planner = self.create_planner(viewing_epoch, transaction_id);
1914            let mut physical_plan = planner.plan(&optimized_plan)?;
1915
1916            // Execute the plan
1917            let executor = Executor::with_columns(physical_plan.columns.clone())
1918                .with_deadline(self.query_deadline());
1919            executor.execute(physical_plan.operator.as_mut())
1920        })
1921    }
1922
1923    /// Executes a GraphQL query with parameters.
1924    ///
1925    /// # Errors
1926    ///
1927    /// Returns an error if the query fails to parse or execute.
1928    #[cfg(feature = "graphql")]
1929    pub fn execute_graphql_with_params(
1930        &self,
1931        query: &str,
1932        params: std::collections::HashMap<String, Value>,
1933    ) -> Result<QueryResult> {
1934        use crate::query::processor::{QueryLanguage, QueryProcessor};
1935
1936        let has_mutations = Self::query_looks_like_mutation(query);
1937
1938        self.with_auto_commit(has_mutations, || {
1939            // Get transaction context for MVCC visibility
1940            let (viewing_epoch, transaction_id) = self.get_transaction_context();
1941
1942            // Create processor with transaction context
1943            let processor = QueryProcessor::for_graph_store_with_transaction(
1944                Arc::clone(&self.graph_store),
1945                Arc::clone(&self.transaction_manager),
1946            );
1947
1948            // Apply transaction context if in a transaction
1949            let processor = if let Some(transaction_id) = transaction_id {
1950                processor.with_transaction_context(viewing_epoch, transaction_id)
1951            } else {
1952                processor
1953            };
1954
1955            processor.process(query, QueryLanguage::GraphQL, Some(&params))
1956        })
1957    }
1958
1959    /// Executes a SQL/PGQ query (SQL:2023 GRAPH_TABLE).
1960    ///
1961    /// # Errors
1962    ///
1963    /// Returns an error if the query fails to parse or execute.
1964    ///
1965    /// # Examples
1966    ///
1967    /// ```no_run
1968    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
1969    /// use grafeo_engine::GrafeoDB;
1970    ///
1971    /// let db = GrafeoDB::new_in_memory();
1972    /// let session = db.session();
1973    ///
1974    /// let result = session.execute_sql(
1975    ///     "SELECT * FROM GRAPH_TABLE (
1976    ///         MATCH (n:Person)
1977    ///         COLUMNS (n.name AS name)
1978    ///     )"
1979    /// )?;
1980    /// # Ok(())
1981    /// # }
1982    /// ```
1983    #[cfg(feature = "sql-pgq")]
1984    pub fn execute_sql(&self, query: &str) -> Result<QueryResult> {
1985        use crate::query::{
1986            Executor, binder::Binder, cache::CacheKey, optimizer::Optimizer, plan::LogicalOperator,
1987            processor::QueryLanguage, translators::sql_pgq,
1988        };
1989
1990        // Parse and translate (always needed to check for DDL)
1991        let logical_plan = sql_pgq::translate(query)?;
1992
1993        // Handle DDL statements directly (they don't go through the query pipeline)
1994        if let LogicalOperator::CreatePropertyGraph(ref cpg) = logical_plan.root {
1995            return Ok(QueryResult {
1996                columns: vec!["status".into()],
1997                column_types: vec![grafeo_common::types::LogicalType::String],
1998                rows: vec![vec![Value::from(format!(
1999                    "Property graph '{}' created ({} node tables, {} edge tables)",
2000                    cpg.name,
2001                    cpg.node_tables.len(),
2002                    cpg.edge_tables.len()
2003                ))]],
2004                execution_time_ms: None,
2005                rows_scanned: None,
2006                status_message: None,
2007                gql_status: grafeo_common::utils::GqlStatus::SUCCESS,
2008            });
2009        }
2010
2011        // Create cache key for query plans
2012        let cache_key = CacheKey::new(query, QueryLanguage::SqlPgq);
2013
2014        // Try to get cached optimized plan
2015        let optimized_plan = if let Some(cached_plan) = self.query_cache.get_optimized(&cache_key) {
2016            cached_plan
2017        } else {
2018            // Semantic validation
2019            let mut binder = Binder::new();
2020            let _binding_context = binder.bind(&logical_plan)?;
2021
2022            // Optimize the plan
2023            let optimizer = Optimizer::from_graph_store(&*self.graph_store);
2024            let plan = optimizer.optimize(logical_plan)?;
2025
2026            // Cache the optimized plan
2027            self.query_cache.put_optimized(cache_key, plan.clone());
2028
2029            plan
2030        };
2031
2032        let has_mutations = optimized_plan.root.has_mutations();
2033
2034        self.with_auto_commit(has_mutations, || {
2035            // Get transaction context for MVCC visibility
2036            let (viewing_epoch, transaction_id) = self.get_transaction_context();
2037
2038            // Convert to physical plan with transaction context
2039            let planner = self.create_planner(viewing_epoch, transaction_id);
2040            let mut physical_plan = planner.plan(&optimized_plan)?;
2041
2042            // Execute the plan
2043            let executor = Executor::with_columns(physical_plan.columns.clone())
2044                .with_deadline(self.query_deadline());
2045            executor.execute(physical_plan.operator.as_mut())
2046        })
2047    }
2048
2049    /// Executes a SQL/PGQ query with parameters.
2050    ///
2051    /// # Errors
2052    ///
2053    /// Returns an error if the query fails to parse or execute.
2054    #[cfg(feature = "sql-pgq")]
2055    pub fn execute_sql_with_params(
2056        &self,
2057        query: &str,
2058        params: std::collections::HashMap<String, Value>,
2059    ) -> Result<QueryResult> {
2060        use crate::query::processor::{QueryLanguage, QueryProcessor};
2061
2062        let has_mutations = Self::query_looks_like_mutation(query);
2063
2064        self.with_auto_commit(has_mutations, || {
2065            // Get transaction context for MVCC visibility
2066            let (viewing_epoch, transaction_id) = self.get_transaction_context();
2067
2068            // Create processor with transaction context
2069            let processor = QueryProcessor::for_graph_store_with_transaction(
2070                Arc::clone(&self.graph_store),
2071                Arc::clone(&self.transaction_manager),
2072            );
2073
2074            // Apply transaction context if in a transaction
2075            let processor = if let Some(transaction_id) = transaction_id {
2076                processor.with_transaction_context(viewing_epoch, transaction_id)
2077            } else {
2078                processor
2079            };
2080
2081            processor.process(query, QueryLanguage::SqlPgq, Some(&params))
2082        })
2083    }
2084
2085    /// Executes a SPARQL query.
2086    ///
2087    /// # Errors
2088    ///
2089    /// Returns an error if the query fails to parse or execute.
2090    #[cfg(all(feature = "sparql", feature = "rdf"))]
2091    pub fn execute_sparql(&self, query: &str) -> Result<QueryResult> {
2092        use crate::query::{
2093            Executor, optimizer::Optimizer, planner::rdf::RdfPlanner, translators::sparql,
2094        };
2095
2096        // Parse and translate the SPARQL query to a logical plan
2097        let logical_plan = sparql::translate(query)?;
2098
2099        // Optimize the plan
2100        let optimizer = Optimizer::from_graph_store(&*self.graph_store);
2101        let optimized_plan = optimizer.optimize(logical_plan)?;
2102
2103        // Convert to physical plan using RDF planner
2104        let planner = RdfPlanner::new(Arc::clone(&self.rdf_store))
2105            .with_transaction_id(*self.current_transaction.lock());
2106        let mut physical_plan = planner.plan(&optimized_plan)?;
2107
2108        // Execute the plan
2109        let executor = Executor::with_columns(physical_plan.columns.clone())
2110            .with_deadline(self.query_deadline());
2111        executor.execute(physical_plan.operator.as_mut())
2112    }
2113
2114    /// Executes a SPARQL query with parameters.
2115    ///
2116    /// # Errors
2117    ///
2118    /// Returns an error if the query fails to parse or execute.
2119    #[cfg(all(feature = "sparql", feature = "rdf"))]
2120    pub fn execute_sparql_with_params(
2121        &self,
2122        query: &str,
2123        params: std::collections::HashMap<String, Value>,
2124    ) -> Result<QueryResult> {
2125        use crate::query::{
2126            Executor, optimizer::Optimizer, planner::rdf::RdfPlanner, processor::substitute_params,
2127            translators::sparql,
2128        };
2129
2130        let mut logical_plan = sparql::translate(query)?;
2131
2132        substitute_params(&mut logical_plan, &params)?;
2133
2134        let optimizer = Optimizer::from_graph_store(&*self.graph_store);
2135        let optimized_plan = optimizer.optimize(logical_plan)?;
2136
2137        let planner = RdfPlanner::new(Arc::clone(&self.rdf_store))
2138            .with_transaction_id(*self.current_transaction.lock());
2139        let mut physical_plan = planner.plan(&optimized_plan)?;
2140
2141        let executor = Executor::with_columns(physical_plan.columns.clone())
2142            .with_deadline(self.query_deadline());
2143        executor.execute(physical_plan.operator.as_mut())
2144    }
2145
2146    /// Executes a query in the specified language by name.
2147    ///
2148    /// Supported language names: `"gql"`, `"cypher"`, `"gremlin"`, `"graphql"`,
2149    /// `"sparql"`, `"sql"`. Each requires the corresponding feature flag.
2150    ///
2151    /// # Errors
2152    ///
2153    /// Returns an error if the language is unknown/disabled or the query fails.
2154    pub fn execute_language(
2155        &self,
2156        query: &str,
2157        language: &str,
2158        params: Option<std::collections::HashMap<String, Value>>,
2159    ) -> Result<QueryResult> {
2160        match language {
2161            "gql" => {
2162                if let Some(p) = params {
2163                    self.execute_with_params(query, p)
2164                } else {
2165                    self.execute(query)
2166                }
2167            }
2168            #[cfg(feature = "cypher")]
2169            "cypher" => {
2170                if let Some(p) = params {
2171                    use crate::query::processor::{QueryLanguage, QueryProcessor};
2172                    let has_mutations = Self::query_looks_like_mutation(query);
2173                    self.with_auto_commit(has_mutations, || {
2174                        let processor = QueryProcessor::for_graph_store_with_transaction(
2175                            Arc::clone(&self.graph_store),
2176                            Arc::clone(&self.transaction_manager),
2177                        );
2178                        let (viewing_epoch, transaction_id) = self.get_transaction_context();
2179                        let processor = if let Some(transaction_id) = transaction_id {
2180                            processor.with_transaction_context(viewing_epoch, transaction_id)
2181                        } else {
2182                            processor
2183                        };
2184                        processor.process(query, QueryLanguage::Cypher, Some(&p))
2185                    })
2186                } else {
2187                    self.execute_cypher(query)
2188                }
2189            }
2190            #[cfg(feature = "gremlin")]
2191            "gremlin" => {
2192                if let Some(p) = params {
2193                    self.execute_gremlin_with_params(query, p)
2194                } else {
2195                    self.execute_gremlin(query)
2196                }
2197            }
2198            #[cfg(feature = "graphql")]
2199            "graphql" => {
2200                if let Some(p) = params {
2201                    self.execute_graphql_with_params(query, p)
2202                } else {
2203                    self.execute_graphql(query)
2204                }
2205            }
2206            #[cfg(feature = "sql-pgq")]
2207            "sql" | "sql-pgq" => {
2208                if let Some(p) = params {
2209                    self.execute_sql_with_params(query, p)
2210                } else {
2211                    self.execute_sql(query)
2212                }
2213            }
2214            #[cfg(all(feature = "sparql", feature = "rdf"))]
2215            "sparql" => {
2216                if let Some(p) = params {
2217                    self.execute_sparql_with_params(query, p)
2218                } else {
2219                    self.execute_sparql(query)
2220                }
2221            }
2222            other => Err(grafeo_common::utils::error::Error::Query(
2223                grafeo_common::utils::error::QueryError::new(
2224                    grafeo_common::utils::error::QueryErrorKind::Semantic,
2225                    format!("Unknown query language: '{other}'"),
2226                ),
2227            )),
2228        }
2229    }
2230
2231    /// Begins a new transaction.
2232    ///
2233    /// # Errors
2234    ///
2235    /// Returns an error if a transaction is already active.
2236    ///
2237    /// # Examples
2238    ///
2239    /// ```no_run
2240    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
2241    /// use grafeo_engine::GrafeoDB;
2242    ///
2243    /// let db = GrafeoDB::new_in_memory();
2244    /// let mut session = db.session();
2245    ///
2246    /// session.begin_transaction()?;
2247    /// session.execute("INSERT (:Person {name: 'Alix'})")?;
2248    /// session.execute("INSERT (:Person {name: 'Gus'})")?;
2249    /// session.commit()?; // Both inserts committed atomically
2250    /// # Ok(())
2251    /// # }
2252    /// ```
2253    /// Clears all cached query plans.
2254    ///
2255    /// The plan cache is shared across all sessions on the same database,
2256    /// so clearing from one session affects all sessions.
2257    pub fn clear_plan_cache(&self) {
2258        self.query_cache.clear();
2259    }
2260
2261    /// Begins a new transaction on this session.
2262    ///
2263    /// Uses the default isolation level (`SnapshotIsolation`).
2264    ///
2265    /// # Errors
2266    ///
2267    /// Returns an error if a transaction is already active.
2268    pub fn begin_transaction(&mut self) -> Result<()> {
2269        self.begin_transaction_inner(false, None)
2270    }
2271
2272    /// Begins a transaction with a specific isolation level.
2273    ///
2274    /// See [`begin_transaction`](Self::begin_transaction) for the default (`SnapshotIsolation`).
2275    ///
2276    /// # Errors
2277    ///
2278    /// Returns an error if a transaction is already active.
2279    pub fn begin_transaction_with_isolation(
2280        &mut self,
2281        isolation_level: crate::transaction::IsolationLevel,
2282    ) -> Result<()> {
2283        self.begin_transaction_inner(false, Some(isolation_level))
2284    }
2285
2286    /// Core transaction begin logic, usable from both `&mut self` and `&self` paths.
2287    fn begin_transaction_inner(
2288        &self,
2289        read_only: bool,
2290        isolation_level: Option<crate::transaction::IsolationLevel>,
2291    ) -> Result<()> {
2292        let mut current = self.current_transaction.lock();
2293        if current.is_some() {
2294            // Nested transaction: create an auto-savepoint instead of a new tx.
2295            drop(current);
2296            let mut depth = self.transaction_nesting_depth.lock();
2297            *depth += 1;
2298            let sp_name = format!("_nested_tx_{}", *depth);
2299            self.savepoint(&sp_name)?;
2300            return Ok(());
2301        }
2302
2303        self.transaction_start_node_count
2304            .store(self.store.node_count(), Ordering::Relaxed);
2305        self.transaction_start_edge_count
2306            .store(self.store.edge_count(), Ordering::Relaxed);
2307        let transaction_id = if let Some(level) = isolation_level {
2308            self.transaction_manager.begin_with_isolation(level)
2309        } else {
2310            self.transaction_manager.begin()
2311        };
2312        *current = Some(transaction_id);
2313        *self.read_only_tx.lock() = read_only;
2314        Ok(())
2315    }
2316
2317    /// Commits the current transaction.
2318    ///
2319    /// Makes all changes since [`begin_transaction`](Self::begin_transaction) permanent.
2320    ///
2321    /// # Errors
2322    ///
2323    /// Returns an error if no transaction is active.
2324    pub fn commit(&mut self) -> Result<()> {
2325        self.commit_inner()
2326    }
2327
2328    /// Core commit logic, usable from both `&mut self` and `&self` paths.
2329    fn commit_inner(&self) -> Result<()> {
2330        // Nested transaction: release the auto-savepoint (changes are preserved).
2331        {
2332            let mut depth = self.transaction_nesting_depth.lock();
2333            if *depth > 0 {
2334                let sp_name = format!("_nested_tx_{depth}");
2335                *depth -= 1;
2336                drop(depth);
2337                return self.release_savepoint(&sp_name);
2338            }
2339        }
2340
2341        let transaction_id = self.current_transaction.lock().take().ok_or_else(|| {
2342            grafeo_common::utils::error::Error::Transaction(
2343                grafeo_common::utils::error::TransactionError::InvalidState(
2344                    "No active transaction".to_string(),
2345                ),
2346            )
2347        })?;
2348
2349        // Commit RDF store pending operations
2350        #[cfg(feature = "rdf")]
2351        self.rdf_store.commit_transaction(transaction_id);
2352
2353        self.transaction_manager.commit(transaction_id)?;
2354
2355        // Sync the LpgStore epoch with the TxManager so that
2356        // convenience lookups (edge_type, get_edge, get_node) that use
2357        // store.current_epoch() can see versions created at the latest epoch.
2358        self.store
2359            .sync_epoch(self.transaction_manager.current_epoch());
2360
2361        // Reset read-only flag and clear savepoints
2362        *self.read_only_tx.lock() = false;
2363        self.savepoints.lock().clear();
2364
2365        // Auto-GC: periodically prune old MVCC versions
2366        if self.gc_interval > 0 {
2367            let count = self.commit_counter.fetch_add(1, Ordering::Relaxed) + 1;
2368            if count.is_multiple_of(self.gc_interval) {
2369                let min_epoch = self.transaction_manager.min_active_epoch();
2370                self.store.gc_versions(min_epoch);
2371                self.transaction_manager.gc();
2372            }
2373        }
2374
2375        Ok(())
2376    }
2377
2378    /// Aborts the current transaction.
2379    ///
2380    /// Discards all changes since [`begin_transaction`](Self::begin_transaction).
2381    ///
2382    /// # Errors
2383    ///
2384    /// Returns an error if no transaction is active.
2385    ///
2386    /// # Examples
2387    ///
2388    /// ```no_run
2389    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
2390    /// use grafeo_engine::GrafeoDB;
2391    ///
2392    /// let db = GrafeoDB::new_in_memory();
2393    /// let mut session = db.session();
2394    ///
2395    /// session.begin_transaction()?;
2396    /// session.execute("INSERT (:Person {name: 'Alix'})")?;
2397    /// session.rollback()?; // Insert is discarded
2398    /// # Ok(())
2399    /// # }
2400    /// ```
2401    pub fn rollback(&mut self) -> Result<()> {
2402        self.rollback_inner()
2403    }
2404
2405    /// Core rollback logic, usable from both `&mut self` and `&self` paths.
2406    fn rollback_inner(&self) -> Result<()> {
2407        // Nested transaction: rollback to the auto-savepoint.
2408        {
2409            let mut depth = self.transaction_nesting_depth.lock();
2410            if *depth > 0 {
2411                let sp_name = format!("_nested_tx_{depth}");
2412                *depth -= 1;
2413                drop(depth);
2414                return self.rollback_to_savepoint(&sp_name);
2415            }
2416        }
2417
2418        let transaction_id = self.current_transaction.lock().take().ok_or_else(|| {
2419            grafeo_common::utils::error::Error::Transaction(
2420                grafeo_common::utils::error::TransactionError::InvalidState(
2421                    "No active transaction".to_string(),
2422                ),
2423            )
2424        })?;
2425
2426        // Reset read-only flag
2427        *self.read_only_tx.lock() = false;
2428
2429        // Discard uncommitted versions in the LPG store
2430        self.store.discard_uncommitted_versions(transaction_id);
2431
2432        // Discard pending operations in the RDF store
2433        #[cfg(feature = "rdf")]
2434        self.rdf_store.rollback_transaction(transaction_id);
2435
2436        // Clear savepoints
2437        self.savepoints.lock().clear();
2438
2439        // Mark transaction as aborted in the manager
2440        self.transaction_manager.abort(transaction_id)
2441    }
2442
2443    /// Creates a named savepoint within the current transaction.
2444    ///
2445    /// The savepoint captures the current node/edge ID counters so that
2446    /// [`rollback_to_savepoint`](Self::rollback_to_savepoint) can discard
2447    /// entities created after this point.
2448    ///
2449    /// # Errors
2450    ///
2451    /// Returns an error if no transaction is active.
2452    pub fn savepoint(&self, name: &str) -> Result<()> {
2453        let _tx_id = self.current_transaction.lock().ok_or_else(|| {
2454            grafeo_common::utils::error::Error::Transaction(
2455                grafeo_common::utils::error::TransactionError::InvalidState(
2456                    "No active transaction".to_string(),
2457                ),
2458            )
2459        })?;
2460
2461        let next_node = self.store.peek_next_node_id();
2462        let next_edge = self.store.peek_next_edge_id();
2463        self.savepoints
2464            .lock()
2465            .push((name.to_string(), next_node, next_edge));
2466        Ok(())
2467    }
2468
2469    /// Rolls back to a named savepoint, undoing all writes made after it.
2470    ///
2471    /// The savepoint and any savepoints created after it are removed.
2472    /// Entities with IDs >= the savepoint snapshot are discarded.
2473    ///
2474    /// # Errors
2475    ///
2476    /// Returns an error if no transaction is active or the savepoint does not exist.
2477    pub fn rollback_to_savepoint(&self, name: &str) -> Result<()> {
2478        let transaction_id = self.current_transaction.lock().ok_or_else(|| {
2479            grafeo_common::utils::error::Error::Transaction(
2480                grafeo_common::utils::error::TransactionError::InvalidState(
2481                    "No active transaction".to_string(),
2482                ),
2483            )
2484        })?;
2485
2486        let mut savepoints = self.savepoints.lock();
2487
2488        // Find the savepoint by name (search from the end for nested savepoints)
2489        let pos = savepoints
2490            .iter()
2491            .rposition(|(n, _, _)| n == name)
2492            .ok_or_else(|| {
2493                grafeo_common::utils::error::Error::Transaction(
2494                    grafeo_common::utils::error::TransactionError::InvalidState(format!(
2495                        "Savepoint '{name}' not found"
2496                    )),
2497                )
2498            })?;
2499
2500        let (_, sp_next_node, sp_next_edge) = savepoints[pos].clone();
2501
2502        // Remove this savepoint and all later ones
2503        savepoints.truncate(pos);
2504        drop(savepoints);
2505
2506        // Discard all nodes with ID >= sp_next_node and edges with ID >= sp_next_edge
2507        let current_next_node = self.store.peek_next_node_id();
2508        let current_next_edge = self.store.peek_next_edge_id();
2509
2510        let node_ids: Vec<NodeId> = (sp_next_node..current_next_node).map(NodeId::new).collect();
2511        let edge_ids: Vec<EdgeId> = (sp_next_edge..current_next_edge).map(EdgeId::new).collect();
2512
2513        if !node_ids.is_empty() || !edge_ids.is_empty() {
2514            self.store
2515                .discard_entities_by_id(transaction_id, &node_ids, &edge_ids);
2516        }
2517
2518        Ok(())
2519    }
2520
2521    /// Releases (removes) a named savepoint without rolling back.
2522    ///
2523    /// # Errors
2524    ///
2525    /// Returns an error if no transaction is active or the savepoint does not exist.
2526    pub fn release_savepoint(&self, name: &str) -> Result<()> {
2527        let _tx_id = self.current_transaction.lock().ok_or_else(|| {
2528            grafeo_common::utils::error::Error::Transaction(
2529                grafeo_common::utils::error::TransactionError::InvalidState(
2530                    "No active transaction".to_string(),
2531                ),
2532            )
2533        })?;
2534
2535        let mut savepoints = self.savepoints.lock();
2536        let pos = savepoints
2537            .iter()
2538            .rposition(|(n, _, _)| n == name)
2539            .ok_or_else(|| {
2540                grafeo_common::utils::error::Error::Transaction(
2541                    grafeo_common::utils::error::TransactionError::InvalidState(format!(
2542                        "Savepoint '{name}' not found"
2543                    )),
2544                )
2545            })?;
2546        savepoints.remove(pos);
2547        Ok(())
2548    }
2549
2550    /// Returns whether a transaction is active.
2551    #[must_use]
2552    pub fn in_transaction(&self) -> bool {
2553        self.current_transaction.lock().is_some()
2554    }
2555
2556    /// Returns the current transaction ID, if any.
2557    #[must_use]
2558    pub(crate) fn current_transaction_id(&self) -> Option<TransactionId> {
2559        *self.current_transaction.lock()
2560    }
2561
2562    /// Returns a reference to the transaction manager.
2563    #[must_use]
2564    pub(crate) fn transaction_manager(&self) -> &TransactionManager {
2565        &self.transaction_manager
2566    }
2567
2568    /// Returns the store's current node count and the count at transaction start.
2569    #[must_use]
2570    pub(crate) fn node_count_delta(&self) -> (usize, usize) {
2571        (
2572            self.transaction_start_node_count.load(Ordering::Relaxed),
2573            self.store.node_count(),
2574        )
2575    }
2576
2577    /// Returns the store's current edge count and the count at transaction start.
2578    #[must_use]
2579    pub(crate) fn edge_count_delta(&self) -> (usize, usize) {
2580        (
2581            self.transaction_start_edge_count.load(Ordering::Relaxed),
2582            self.store.edge_count(),
2583        )
2584    }
2585
2586    /// Prepares the current transaction for a two-phase commit.
2587    ///
2588    /// Returns a [`PreparedCommit`](crate::transaction::PreparedCommit) that
2589    /// lets you inspect pending changes and attach metadata before finalizing.
2590    /// The mutable borrow prevents concurrent operations while the commit is
2591    /// pending.
2592    ///
2593    /// If the `PreparedCommit` is dropped without calling `commit()` or
2594    /// `abort()`, the transaction is automatically rolled back.
2595    ///
2596    /// # Errors
2597    ///
2598    /// Returns an error if no transaction is active.
2599    ///
2600    /// # Examples
2601    ///
2602    /// ```no_run
2603    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
2604    /// use grafeo_engine::GrafeoDB;
2605    ///
2606    /// let db = GrafeoDB::new_in_memory();
2607    /// let mut session = db.session();
2608    ///
2609    /// session.begin_transaction()?;
2610    /// session.execute("INSERT (:Person {name: 'Alix'})")?;
2611    ///
2612    /// let mut prepared = session.prepare_commit()?;
2613    /// println!("Nodes written: {}", prepared.info().nodes_written);
2614    /// prepared.set_metadata("audit_user", "admin");
2615    /// prepared.commit()?;
2616    /// # Ok(())
2617    /// # }
2618    /// ```
2619    pub fn prepare_commit(&mut self) -> Result<crate::transaction::PreparedCommit<'_>> {
2620        crate::transaction::PreparedCommit::new(self)
2621    }
2622
2623    /// Sets auto-commit mode.
2624    pub fn set_auto_commit(&mut self, auto_commit: bool) {
2625        self.auto_commit = auto_commit;
2626    }
2627
2628    /// Returns whether auto-commit is enabled.
2629    #[must_use]
2630    pub fn auto_commit(&self) -> bool {
2631        self.auto_commit
2632    }
2633
2634    /// Returns `true` if auto-commit should wrap this execution.
2635    ///
2636    /// Auto-commit kicks in when: the session is in auto-commit mode,
2637    /// no explicit transaction is active, and the query mutates data.
2638    fn needs_auto_commit(&self, has_mutations: bool) -> bool {
2639        self.auto_commit && has_mutations && self.current_transaction.lock().is_none()
2640    }
2641
2642    /// Wraps `body` in an automatic begin/commit when [`needs_auto_commit`]
2643    /// returns `true`. On error the transaction is rolled back.
2644    fn with_auto_commit<F>(&self, has_mutations: bool, body: F) -> Result<QueryResult>
2645    where
2646        F: FnOnce() -> Result<QueryResult>,
2647    {
2648        if self.needs_auto_commit(has_mutations) {
2649            self.begin_transaction_inner(false, None)?;
2650            match body() {
2651                Ok(result) => {
2652                    self.commit_inner()?;
2653                    Ok(result)
2654                }
2655                Err(e) => {
2656                    let _ = self.rollback_inner();
2657                    Err(e)
2658                }
2659            }
2660        } else {
2661            body()
2662        }
2663    }
2664
2665    /// Quick heuristic: returns `true` when the query text looks like it
2666    /// performs a mutation. Used by `_with_params` paths that go through the
2667    /// `QueryProcessor` (where the logical plan isn't available before
2668    /// execution). False negatives are harmless: the data just won't be
2669    /// auto-committed, which matches the prior behaviour.
2670    fn query_looks_like_mutation(query: &str) -> bool {
2671        let upper = query.to_ascii_uppercase();
2672        upper.contains("INSERT")
2673            || upper.contains("CREATE")
2674            || upper.contains("DELETE")
2675            || upper.contains("MERGE")
2676            || upper.contains("SET")
2677            || upper.contains("REMOVE")
2678            || upper.contains("DROP")
2679            || upper.contains("ALTER")
2680    }
2681
2682    /// Computes the wall-clock deadline for query execution.
2683    #[must_use]
2684    fn query_deadline(&self) -> Option<Instant> {
2685        #[cfg(not(target_arch = "wasm32"))]
2686        {
2687            self.query_timeout.map(|d| Instant::now() + d)
2688        }
2689        #[cfg(target_arch = "wasm32")]
2690        {
2691            let _ = &self.query_timeout;
2692            None
2693        }
2694    }
2695
2696    /// Evaluates a simple integer literal from a session parameter expression.
2697    fn eval_integer_literal(expr: &grafeo_adapters::query::gql::ast::Expression) -> Option<i64> {
2698        use grafeo_adapters::query::gql::ast::{Expression, Literal};
2699        match expr {
2700            Expression::Literal(Literal::Integer(n)) => Some(*n),
2701            _ => None,
2702        }
2703    }
2704
2705    /// Returns the current transaction context for MVCC visibility.
2706    ///
2707    /// Returns `(viewing_epoch, transaction_id)` where:
2708    /// - `viewing_epoch` is the epoch at which to check version visibility
2709    /// - `transaction_id` is the current transaction ID (if in a transaction)
2710    #[must_use]
2711    fn get_transaction_context(&self) -> (EpochId, Option<TransactionId>) {
2712        // Time-travel override takes precedence (read-only, no tx context)
2713        if let Some(epoch) = *self.viewing_epoch_override.lock() {
2714            return (epoch, None);
2715        }
2716
2717        if let Some(transaction_id) = *self.current_transaction.lock() {
2718            // In a transaction: use the transaction's start epoch
2719            let epoch = self
2720                .transaction_manager
2721                .start_epoch(transaction_id)
2722                .unwrap_or_else(|| self.transaction_manager.current_epoch());
2723            (epoch, Some(transaction_id))
2724        } else {
2725            // No transaction: use current epoch
2726            (self.transaction_manager.current_epoch(), None)
2727        }
2728    }
2729
2730    /// Creates a planner with transaction context and constraint validator.
2731    fn create_planner(
2732        &self,
2733        viewing_epoch: EpochId,
2734        transaction_id: Option<TransactionId>,
2735    ) -> crate::query::Planner {
2736        use crate::query::Planner;
2737
2738        let mut planner = Planner::with_context(
2739            Arc::clone(&self.graph_store),
2740            Arc::clone(&self.transaction_manager),
2741            transaction_id,
2742            viewing_epoch,
2743        )
2744        .with_factorized_execution(self.factorized_execution)
2745        .with_catalog(Arc::clone(&self.catalog));
2746
2747        // Attach the constraint validator for schema enforcement
2748        let validator = CatalogConstraintValidator::new(Arc::clone(&self.catalog));
2749        planner = planner.with_validator(Arc::new(validator));
2750
2751        planner
2752    }
2753
2754    /// Creates a node directly (bypassing query execution).
2755    ///
2756    /// This is a low-level API for testing and direct manipulation.
2757    /// If a transaction is active, the node will be versioned with the transaction ID.
2758    pub fn create_node(&self, labels: &[&str]) -> NodeId {
2759        let (epoch, transaction_id) = self.get_transaction_context();
2760        self.store.create_node_versioned(
2761            labels,
2762            epoch,
2763            transaction_id.unwrap_or(TransactionId::SYSTEM),
2764        )
2765    }
2766
2767    /// Creates a node with properties.
2768    ///
2769    /// If a transaction is active, the node will be versioned with the transaction ID.
2770    pub fn create_node_with_props<'a>(
2771        &self,
2772        labels: &[&str],
2773        properties: impl IntoIterator<Item = (&'a str, Value)>,
2774    ) -> NodeId {
2775        let (epoch, transaction_id) = self.get_transaction_context();
2776        self.store.create_node_with_props_versioned(
2777            labels,
2778            properties,
2779            epoch,
2780            transaction_id.unwrap_or(TransactionId::SYSTEM),
2781        )
2782    }
2783
2784    /// Creates an edge between two nodes.
2785    ///
2786    /// This is a low-level API for testing and direct manipulation.
2787    /// If a transaction is active, the edge will be versioned with the transaction ID.
2788    pub fn create_edge(
2789        &self,
2790        src: NodeId,
2791        dst: NodeId,
2792        edge_type: &str,
2793    ) -> grafeo_common::types::EdgeId {
2794        let (epoch, transaction_id) = self.get_transaction_context();
2795        self.store.create_edge_versioned(
2796            src,
2797            dst,
2798            edge_type,
2799            epoch,
2800            transaction_id.unwrap_or(TransactionId::SYSTEM),
2801        )
2802    }
2803
2804    // =========================================================================
2805    // Direct Lookup APIs (bypass query planning for O(1) point reads)
2806    // =========================================================================
2807
2808    /// Gets a node by ID directly, bypassing query planning.
2809    ///
2810    /// This is the fastest way to retrieve a single node when you know its ID.
2811    /// Skips parsing, binding, optimization, and physical planning entirely.
2812    ///
2813    /// # Performance
2814    ///
2815    /// - Time complexity: O(1) average case
2816    /// - No lock contention (uses DashMap internally)
2817    /// - ~20-30x faster than equivalent MATCH query
2818    ///
2819    /// # Example
2820    ///
2821    /// ```no_run
2822    /// # use grafeo_engine::GrafeoDB;
2823    /// # let db = GrafeoDB::new_in_memory();
2824    /// let session = db.session();
2825    /// let node_id = session.create_node(&["Person"]);
2826    ///
2827    /// // Direct lookup - O(1), no query planning
2828    /// let node = session.get_node(node_id);
2829    /// assert!(node.is_some());
2830    /// ```
2831    #[must_use]
2832    pub fn get_node(&self, id: NodeId) -> Option<Node> {
2833        let (epoch, transaction_id) = self.get_transaction_context();
2834        self.store
2835            .get_node_versioned(id, epoch, transaction_id.unwrap_or(TransactionId::SYSTEM))
2836    }
2837
2838    /// Gets a single property from a node by ID, bypassing query planning.
2839    ///
2840    /// More efficient than `get_node()` when you only need one property,
2841    /// as it avoids loading the full node with all properties.
2842    ///
2843    /// # Performance
2844    ///
2845    /// - Time complexity: O(1) average case
2846    /// - No query planning overhead
2847    ///
2848    /// # Example
2849    ///
2850    /// ```no_run
2851    /// # use grafeo_engine::GrafeoDB;
2852    /// # use grafeo_common::types::Value;
2853    /// # let db = GrafeoDB::new_in_memory();
2854    /// let session = db.session();
2855    /// let id = session.create_node_with_props(&["Person"], [("name", "Alix".into())]);
2856    ///
2857    /// // Direct property access - O(1)
2858    /// let name = session.get_node_property(id, "name");
2859    /// assert_eq!(name, Some(Value::String("Alix".into())));
2860    /// ```
2861    #[must_use]
2862    pub fn get_node_property(&self, id: NodeId, key: &str) -> Option<Value> {
2863        self.get_node(id)
2864            .and_then(|node| node.get_property(key).cloned())
2865    }
2866
2867    /// Gets an edge by ID directly, bypassing query planning.
2868    ///
2869    /// # Performance
2870    ///
2871    /// - Time complexity: O(1) average case
2872    /// - No lock contention
2873    #[must_use]
2874    pub fn get_edge(&self, id: EdgeId) -> Option<Edge> {
2875        let (epoch, transaction_id) = self.get_transaction_context();
2876        self.store
2877            .get_edge_versioned(id, epoch, transaction_id.unwrap_or(TransactionId::SYSTEM))
2878    }
2879
2880    /// Gets outgoing neighbors of a node directly, bypassing query planning.
2881    ///
2882    /// Returns (neighbor_id, edge_id) pairs for all outgoing edges.
2883    ///
2884    /// # Performance
2885    ///
2886    /// - Time complexity: O(degree) where degree is the number of outgoing edges
2887    /// - Uses adjacency index for direct access
2888    /// - ~10-20x faster than equivalent MATCH query
2889    ///
2890    /// # Example
2891    ///
2892    /// ```no_run
2893    /// # use grafeo_engine::GrafeoDB;
2894    /// # let db = GrafeoDB::new_in_memory();
2895    /// let session = db.session();
2896    /// let alix = session.create_node(&["Person"]);
2897    /// let gus = session.create_node(&["Person"]);
2898    /// session.create_edge(alix, gus, "KNOWS");
2899    ///
2900    /// // Direct neighbor lookup - O(degree)
2901    /// let neighbors = session.get_neighbors_outgoing(alix);
2902    /// assert_eq!(neighbors.len(), 1);
2903    /// assert_eq!(neighbors[0].0, gus);
2904    /// ```
2905    #[must_use]
2906    pub fn get_neighbors_outgoing(&self, node: NodeId) -> Vec<(NodeId, EdgeId)> {
2907        self.store.edges_from(node, Direction::Outgoing).collect()
2908    }
2909
2910    /// Gets incoming neighbors of a node directly, bypassing query planning.
2911    ///
2912    /// Returns (neighbor_id, edge_id) pairs for all incoming edges.
2913    ///
2914    /// # Performance
2915    ///
2916    /// - Time complexity: O(degree) where degree is the number of incoming edges
2917    /// - Uses backward adjacency index for direct access
2918    #[must_use]
2919    pub fn get_neighbors_incoming(&self, node: NodeId) -> Vec<(NodeId, EdgeId)> {
2920        self.store.edges_from(node, Direction::Incoming).collect()
2921    }
2922
2923    /// Gets outgoing neighbors filtered by edge type, bypassing query planning.
2924    ///
2925    /// # Example
2926    ///
2927    /// ```no_run
2928    /// # use grafeo_engine::GrafeoDB;
2929    /// # let db = GrafeoDB::new_in_memory();
2930    /// # let session = db.session();
2931    /// # let alix = session.create_node(&["Person"]);
2932    /// let neighbors = session.get_neighbors_outgoing_by_type(alix, "KNOWS");
2933    /// ```
2934    #[must_use]
2935    pub fn get_neighbors_outgoing_by_type(
2936        &self,
2937        node: NodeId,
2938        edge_type: &str,
2939    ) -> Vec<(NodeId, EdgeId)> {
2940        self.store
2941            .edges_from(node, Direction::Outgoing)
2942            .filter(|(_, edge_id)| {
2943                self.get_edge(*edge_id)
2944                    .is_some_and(|e| e.edge_type.as_str() == edge_type)
2945            })
2946            .collect()
2947    }
2948
2949    /// Checks if a node exists, bypassing query planning.
2950    ///
2951    /// # Performance
2952    ///
2953    /// - Time complexity: O(1)
2954    /// - Fastest existence check available
2955    #[must_use]
2956    pub fn node_exists(&self, id: NodeId) -> bool {
2957        self.get_node(id).is_some()
2958    }
2959
2960    /// Checks if an edge exists, bypassing query planning.
2961    #[must_use]
2962    pub fn edge_exists(&self, id: EdgeId) -> bool {
2963        self.get_edge(id).is_some()
2964    }
2965
2966    /// Gets the degree (number of edges) of a node.
2967    ///
2968    /// Returns (outgoing_degree, incoming_degree).
2969    #[must_use]
2970    pub fn get_degree(&self, node: NodeId) -> (usize, usize) {
2971        let out = self.store.out_degree(node);
2972        let in_degree = self.store.in_degree(node);
2973        (out, in_degree)
2974    }
2975
2976    /// Batch lookup of multiple nodes by ID.
2977    ///
2978    /// More efficient than calling `get_node()` in a loop because it
2979    /// amortizes overhead.
2980    ///
2981    /// # Performance
2982    ///
2983    /// - Time complexity: O(n) where n is the number of IDs
2984    /// - Better cache utilization than individual lookups
2985    #[must_use]
2986    pub fn get_nodes_batch(&self, ids: &[NodeId]) -> Vec<Option<Node>> {
2987        let (epoch, transaction_id) = self.get_transaction_context();
2988        let tx = transaction_id.unwrap_or(TransactionId::SYSTEM);
2989        ids.iter()
2990            .map(|&id| self.store.get_node_versioned(id, epoch, tx))
2991            .collect()
2992    }
2993
2994    // ── Change Data Capture ─────────────────────────────────────────────
2995
2996    /// Returns the full change history for an entity (node or edge).
2997    #[cfg(feature = "cdc")]
2998    pub fn history(
2999        &self,
3000        entity_id: impl Into<crate::cdc::EntityId>,
3001    ) -> Result<Vec<crate::cdc::ChangeEvent>> {
3002        Ok(self.cdc_log.history(entity_id.into()))
3003    }
3004
3005    /// Returns change events for an entity since the given epoch.
3006    #[cfg(feature = "cdc")]
3007    pub fn history_since(
3008        &self,
3009        entity_id: impl Into<crate::cdc::EntityId>,
3010        since_epoch: EpochId,
3011    ) -> Result<Vec<crate::cdc::ChangeEvent>> {
3012        Ok(self.cdc_log.history_since(entity_id.into(), since_epoch))
3013    }
3014
3015    /// Returns all change events across all entities in an epoch range.
3016    #[cfg(feature = "cdc")]
3017    pub fn changes_between(
3018        &self,
3019        start_epoch: EpochId,
3020        end_epoch: EpochId,
3021    ) -> Result<Vec<crate::cdc::ChangeEvent>> {
3022        Ok(self.cdc_log.changes_between(start_epoch, end_epoch))
3023    }
3024}
3025
3026#[cfg(test)]
3027mod tests {
3028    use crate::database::GrafeoDB;
3029
3030    #[test]
3031    fn test_session_create_node() {
3032        let db = GrafeoDB::new_in_memory();
3033        let session = db.session();
3034
3035        let id = session.create_node(&["Person"]);
3036        assert!(id.is_valid());
3037        assert_eq!(db.node_count(), 1);
3038    }
3039
3040    #[test]
3041    fn test_session_transaction() {
3042        let db = GrafeoDB::new_in_memory();
3043        let mut session = db.session();
3044
3045        assert!(!session.in_transaction());
3046
3047        session.begin_transaction().unwrap();
3048        assert!(session.in_transaction());
3049
3050        session.commit().unwrap();
3051        assert!(!session.in_transaction());
3052    }
3053
3054    #[test]
3055    fn test_session_transaction_context() {
3056        let db = GrafeoDB::new_in_memory();
3057        let mut session = db.session();
3058
3059        // Without transaction - context should have current epoch and no transaction_id
3060        let (_epoch1, transaction_id1) = session.get_transaction_context();
3061        assert!(transaction_id1.is_none());
3062
3063        // Start a transaction
3064        session.begin_transaction().unwrap();
3065        let (epoch2, transaction_id2) = session.get_transaction_context();
3066        assert!(transaction_id2.is_some());
3067        // Transaction should have a valid epoch
3068        let _ = epoch2; // Use the variable
3069
3070        // Commit and verify
3071        session.commit().unwrap();
3072        let (epoch3, tx_id3) = session.get_transaction_context();
3073        assert!(tx_id3.is_none());
3074        // Epoch should have advanced after commit
3075        assert!(epoch3.as_u64() >= epoch2.as_u64());
3076    }
3077
3078    #[test]
3079    fn test_session_rollback() {
3080        let db = GrafeoDB::new_in_memory();
3081        let mut session = db.session();
3082
3083        session.begin_transaction().unwrap();
3084        session.rollback().unwrap();
3085        assert!(!session.in_transaction());
3086    }
3087
3088    #[test]
3089    fn test_session_rollback_discards_versions() {
3090        use grafeo_common::types::TransactionId;
3091
3092        let db = GrafeoDB::new_in_memory();
3093
3094        // Create a node outside of any transaction (at system level)
3095        let node_before = db.store().create_node(&["Person"]);
3096        assert!(node_before.is_valid());
3097        assert_eq!(db.node_count(), 1, "Should have 1 node before transaction");
3098
3099        // Start a transaction
3100        let mut session = db.session();
3101        session.begin_transaction().unwrap();
3102        let transaction_id = session.current_transaction.lock().unwrap();
3103
3104        // Create a node versioned with the transaction's ID
3105        let epoch = db.store().current_epoch();
3106        let node_in_tx = db
3107            .store()
3108            .create_node_versioned(&["Person"], epoch, transaction_id);
3109        assert!(node_in_tx.is_valid());
3110
3111        // Should see 2 nodes at this point
3112        assert_eq!(db.node_count(), 2, "Should have 2 nodes during transaction");
3113
3114        // Rollback the transaction
3115        session.rollback().unwrap();
3116        assert!(!session.in_transaction());
3117
3118        // The node created in the transaction should be discarded
3119        // Only the first node should remain visible
3120        let count_after = db.node_count();
3121        assert_eq!(
3122            count_after, 1,
3123            "Rollback should discard uncommitted node, but got {count_after}"
3124        );
3125
3126        // The original node should still be accessible
3127        let current_epoch = db.store().current_epoch();
3128        assert!(
3129            db.store()
3130                .get_node_versioned(node_before, current_epoch, TransactionId::SYSTEM)
3131                .is_some(),
3132            "Original node should still exist"
3133        );
3134
3135        // The node created in the transaction should not be accessible
3136        assert!(
3137            db.store()
3138                .get_node_versioned(node_in_tx, current_epoch, TransactionId::SYSTEM)
3139                .is_none(),
3140            "Transaction node should be gone"
3141        );
3142    }
3143
3144    #[test]
3145    fn test_session_create_node_in_transaction() {
3146        // Test that session.create_node() is transaction-aware
3147        let db = GrafeoDB::new_in_memory();
3148
3149        // Create a node outside of any transaction
3150        let node_before = db.create_node(&["Person"]);
3151        assert!(node_before.is_valid());
3152        assert_eq!(db.node_count(), 1, "Should have 1 node before transaction");
3153
3154        // Start a transaction and create a node through the session
3155        let mut session = db.session();
3156        session.begin_transaction().unwrap();
3157
3158        // Create a node through session.create_node() - should be versioned with tx
3159        let node_in_tx = session.create_node(&["Person"]);
3160        assert!(node_in_tx.is_valid());
3161
3162        // Should see 2 nodes at this point
3163        assert_eq!(db.node_count(), 2, "Should have 2 nodes during transaction");
3164
3165        // Rollback the transaction
3166        session.rollback().unwrap();
3167
3168        // The node created via session.create_node() should be discarded
3169        let count_after = db.node_count();
3170        assert_eq!(
3171            count_after, 1,
3172            "Rollback should discard node created via session.create_node(), but got {count_after}"
3173        );
3174    }
3175
3176    #[test]
3177    fn test_session_create_node_with_props_in_transaction() {
3178        use grafeo_common::types::Value;
3179
3180        // Test that session.create_node_with_props() is transaction-aware
3181        let db = GrafeoDB::new_in_memory();
3182
3183        // Create a node outside of any transaction
3184        db.create_node(&["Person"]);
3185        assert_eq!(db.node_count(), 1, "Should have 1 node before transaction");
3186
3187        // Start a transaction and create a node with properties
3188        let mut session = db.session();
3189        session.begin_transaction().unwrap();
3190
3191        let node_in_tx =
3192            session.create_node_with_props(&["Person"], [("name", Value::String("Alix".into()))]);
3193        assert!(node_in_tx.is_valid());
3194
3195        // Should see 2 nodes
3196        assert_eq!(db.node_count(), 2, "Should have 2 nodes during transaction");
3197
3198        // Rollback the transaction
3199        session.rollback().unwrap();
3200
3201        // The node should be discarded
3202        let count_after = db.node_count();
3203        assert_eq!(
3204            count_after, 1,
3205            "Rollback should discard node created via session.create_node_with_props()"
3206        );
3207    }
3208
3209    #[cfg(feature = "gql")]
3210    mod gql_tests {
3211        use super::*;
3212
3213        #[test]
3214        fn test_gql_query_execution() {
3215            let db = GrafeoDB::new_in_memory();
3216            let session = db.session();
3217
3218            // Create some test data
3219            session.create_node(&["Person"]);
3220            session.create_node(&["Person"]);
3221            session.create_node(&["Animal"]);
3222
3223            // Execute a GQL query
3224            let result = session.execute("MATCH (n:Person) RETURN n").unwrap();
3225
3226            // Should return 2 Person nodes
3227            assert_eq!(result.row_count(), 2);
3228            assert_eq!(result.column_count(), 1);
3229            assert_eq!(result.columns[0], "n");
3230        }
3231
3232        #[test]
3233        fn test_gql_empty_result() {
3234            let db = GrafeoDB::new_in_memory();
3235            let session = db.session();
3236
3237            // No data in database
3238            let result = session.execute("MATCH (n:Person) RETURN n").unwrap();
3239
3240            assert_eq!(result.row_count(), 0);
3241        }
3242
3243        #[test]
3244        fn test_gql_parse_error() {
3245            let db = GrafeoDB::new_in_memory();
3246            let session = db.session();
3247
3248            // Invalid GQL syntax
3249            let result = session.execute("MATCH (n RETURN n");
3250
3251            assert!(result.is_err());
3252        }
3253
3254        #[test]
3255        fn test_gql_relationship_traversal() {
3256            let db = GrafeoDB::new_in_memory();
3257            let session = db.session();
3258
3259            // Create a graph: Alix -> Gus, Alix -> Vincent
3260            let alix = session.create_node(&["Person"]);
3261            let gus = session.create_node(&["Person"]);
3262            let vincent = session.create_node(&["Person"]);
3263
3264            session.create_edge(alix, gus, "KNOWS");
3265            session.create_edge(alix, vincent, "KNOWS");
3266
3267            // Execute a path query: MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b
3268            let result = session
3269                .execute("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b")
3270                .unwrap();
3271
3272            // Should return 2 rows (Alix->Gus, Alix->Vincent)
3273            assert_eq!(result.row_count(), 2);
3274            assert_eq!(result.column_count(), 2);
3275            assert_eq!(result.columns[0], "a");
3276            assert_eq!(result.columns[1], "b");
3277        }
3278
3279        #[test]
3280        fn test_gql_relationship_with_type_filter() {
3281            let db = GrafeoDB::new_in_memory();
3282            let session = db.session();
3283
3284            // Create a graph: Alix -KNOWS-> Gus, Alix -WORKS_WITH-> Vincent
3285            let alix = session.create_node(&["Person"]);
3286            let gus = session.create_node(&["Person"]);
3287            let vincent = session.create_node(&["Person"]);
3288
3289            session.create_edge(alix, gus, "KNOWS");
3290            session.create_edge(alix, vincent, "WORKS_WITH");
3291
3292            // Query only KNOWS relationships
3293            let result = session
3294                .execute("MATCH (a:Person)-[:KNOWS]->(b:Person) RETURN a, b")
3295                .unwrap();
3296
3297            // Should return only 1 row (Alix->Gus)
3298            assert_eq!(result.row_count(), 1);
3299        }
3300
3301        #[test]
3302        fn test_gql_semantic_error_undefined_variable() {
3303            let db = GrafeoDB::new_in_memory();
3304            let session = db.session();
3305
3306            // Reference undefined variable 'x' in RETURN
3307            let result = session.execute("MATCH (n:Person) RETURN x");
3308
3309            // Should fail with semantic error
3310            assert!(result.is_err());
3311            let Err(err) = result else {
3312                panic!("Expected error")
3313            };
3314            assert!(
3315                err.to_string().contains("Undefined variable"),
3316                "Expected undefined variable error, got: {}",
3317                err
3318            );
3319        }
3320
3321        #[test]
3322        fn test_gql_where_clause_property_filter() {
3323            use grafeo_common::types::Value;
3324
3325            let db = GrafeoDB::new_in_memory();
3326            let session = db.session();
3327
3328            // Create people with ages
3329            session.create_node_with_props(&["Person"], [("age", Value::Int64(25))]);
3330            session.create_node_with_props(&["Person"], [("age", Value::Int64(35))]);
3331            session.create_node_with_props(&["Person"], [("age", Value::Int64(45))]);
3332
3333            // Query with WHERE clause: age > 30
3334            let result = session
3335                .execute("MATCH (n:Person) WHERE n.age > 30 RETURN n")
3336                .unwrap();
3337
3338            // Should return 2 people (ages 35 and 45)
3339            assert_eq!(result.row_count(), 2);
3340        }
3341
3342        #[test]
3343        fn test_gql_where_clause_equality() {
3344            use grafeo_common::types::Value;
3345
3346            let db = GrafeoDB::new_in_memory();
3347            let session = db.session();
3348
3349            // Create people with names
3350            session.create_node_with_props(&["Person"], [("name", Value::String("Alix".into()))]);
3351            session.create_node_with_props(&["Person"], [("name", Value::String("Gus".into()))]);
3352            session.create_node_with_props(&["Person"], [("name", Value::String("Alix".into()))]);
3353
3354            // Query with WHERE clause: name = "Alix"
3355            let result = session
3356                .execute("MATCH (n:Person) WHERE n.name = \"Alix\" RETURN n")
3357                .unwrap();
3358
3359            // Should return 2 people named Alix
3360            assert_eq!(result.row_count(), 2);
3361        }
3362
3363        #[test]
3364        fn test_gql_return_property_access() {
3365            use grafeo_common::types::Value;
3366
3367            let db = GrafeoDB::new_in_memory();
3368            let session = db.session();
3369
3370            // Create people with names and ages
3371            session.create_node_with_props(
3372                &["Person"],
3373                [
3374                    ("name", Value::String("Alix".into())),
3375                    ("age", Value::Int64(30)),
3376                ],
3377            );
3378            session.create_node_with_props(
3379                &["Person"],
3380                [
3381                    ("name", Value::String("Gus".into())),
3382                    ("age", Value::Int64(25)),
3383                ],
3384            );
3385
3386            // Query returning properties
3387            let result = session
3388                .execute("MATCH (n:Person) RETURN n.name, n.age")
3389                .unwrap();
3390
3391            // Should return 2 rows with name and age columns
3392            assert_eq!(result.row_count(), 2);
3393            assert_eq!(result.column_count(), 2);
3394            assert_eq!(result.columns[0], "n.name");
3395            assert_eq!(result.columns[1], "n.age");
3396
3397            // Check that we get actual values
3398            let names: Vec<&Value> = result.rows.iter().map(|r| &r[0]).collect();
3399            assert!(names.contains(&&Value::String("Alix".into())));
3400            assert!(names.contains(&&Value::String("Gus".into())));
3401        }
3402
3403        #[test]
3404        fn test_gql_return_mixed_expressions() {
3405            use grafeo_common::types::Value;
3406
3407            let db = GrafeoDB::new_in_memory();
3408            let session = db.session();
3409
3410            // Create a person
3411            session.create_node_with_props(&["Person"], [("name", Value::String("Alix".into()))]);
3412
3413            // Query returning both node and property
3414            let result = session
3415                .execute("MATCH (n:Person) RETURN n, n.name")
3416                .unwrap();
3417
3418            assert_eq!(result.row_count(), 1);
3419            assert_eq!(result.column_count(), 2);
3420            assert_eq!(result.columns[0], "n");
3421            assert_eq!(result.columns[1], "n.name");
3422
3423            // Second column should be the name
3424            assert_eq!(result.rows[0][1], Value::String("Alix".into()));
3425        }
3426    }
3427
3428    #[cfg(feature = "cypher")]
3429    mod cypher_tests {
3430        use super::*;
3431
3432        #[test]
3433        fn test_cypher_query_execution() {
3434            let db = GrafeoDB::new_in_memory();
3435            let session = db.session();
3436
3437            // Create some test data
3438            session.create_node(&["Person"]);
3439            session.create_node(&["Person"]);
3440            session.create_node(&["Animal"]);
3441
3442            // Execute a Cypher query
3443            let result = session.execute_cypher("MATCH (n:Person) RETURN n").unwrap();
3444
3445            // Should return 2 Person nodes
3446            assert_eq!(result.row_count(), 2);
3447            assert_eq!(result.column_count(), 1);
3448            assert_eq!(result.columns[0], "n");
3449        }
3450
3451        #[test]
3452        fn test_cypher_empty_result() {
3453            let db = GrafeoDB::new_in_memory();
3454            let session = db.session();
3455
3456            // No data in database
3457            let result = session.execute_cypher("MATCH (n:Person) RETURN n").unwrap();
3458
3459            assert_eq!(result.row_count(), 0);
3460        }
3461
3462        #[test]
3463        fn test_cypher_parse_error() {
3464            let db = GrafeoDB::new_in_memory();
3465            let session = db.session();
3466
3467            // Invalid Cypher syntax
3468            let result = session.execute_cypher("MATCH (n RETURN n");
3469
3470            assert!(result.is_err());
3471        }
3472    }
3473
3474    // ==================== Direct Lookup API Tests ====================
3475
3476    mod direct_lookup_tests {
3477        use super::*;
3478        use grafeo_common::types::Value;
3479
3480        #[test]
3481        fn test_get_node() {
3482            let db = GrafeoDB::new_in_memory();
3483            let session = db.session();
3484
3485            let id = session.create_node(&["Person"]);
3486            let node = session.get_node(id);
3487
3488            assert!(node.is_some());
3489            let node = node.unwrap();
3490            assert_eq!(node.id, id);
3491        }
3492
3493        #[test]
3494        fn test_get_node_not_found() {
3495            use grafeo_common::types::NodeId;
3496
3497            let db = GrafeoDB::new_in_memory();
3498            let session = db.session();
3499
3500            // Try to get a non-existent node
3501            let node = session.get_node(NodeId::new(9999));
3502            assert!(node.is_none());
3503        }
3504
3505        #[test]
3506        fn test_get_node_property() {
3507            let db = GrafeoDB::new_in_memory();
3508            let session = db.session();
3509
3510            let id = session
3511                .create_node_with_props(&["Person"], [("name", Value::String("Alix".into()))]);
3512
3513            let name = session.get_node_property(id, "name");
3514            assert_eq!(name, Some(Value::String("Alix".into())));
3515
3516            // Non-existent property
3517            let missing = session.get_node_property(id, "missing");
3518            assert!(missing.is_none());
3519        }
3520
3521        #[test]
3522        fn test_get_edge() {
3523            let db = GrafeoDB::new_in_memory();
3524            let session = db.session();
3525
3526            let alix = session.create_node(&["Person"]);
3527            let gus = session.create_node(&["Person"]);
3528            let edge_id = session.create_edge(alix, gus, "KNOWS");
3529
3530            let edge = session.get_edge(edge_id);
3531            assert!(edge.is_some());
3532            let edge = edge.unwrap();
3533            assert_eq!(edge.id, edge_id);
3534            assert_eq!(edge.src, alix);
3535            assert_eq!(edge.dst, gus);
3536        }
3537
3538        #[test]
3539        fn test_get_edge_not_found() {
3540            use grafeo_common::types::EdgeId;
3541
3542            let db = GrafeoDB::new_in_memory();
3543            let session = db.session();
3544
3545            let edge = session.get_edge(EdgeId::new(9999));
3546            assert!(edge.is_none());
3547        }
3548
3549        #[test]
3550        fn test_get_neighbors_outgoing() {
3551            let db = GrafeoDB::new_in_memory();
3552            let session = db.session();
3553
3554            let alix = session.create_node(&["Person"]);
3555            let gus = session.create_node(&["Person"]);
3556            let harm = session.create_node(&["Person"]);
3557
3558            session.create_edge(alix, gus, "KNOWS");
3559            session.create_edge(alix, harm, "KNOWS");
3560
3561            let neighbors = session.get_neighbors_outgoing(alix);
3562            assert_eq!(neighbors.len(), 2);
3563
3564            let neighbor_ids: Vec<_> = neighbors.iter().map(|(node_id, _)| *node_id).collect();
3565            assert!(neighbor_ids.contains(&gus));
3566            assert!(neighbor_ids.contains(&harm));
3567        }
3568
3569        #[test]
3570        fn test_get_neighbors_incoming() {
3571            let db = GrafeoDB::new_in_memory();
3572            let session = db.session();
3573
3574            let alix = session.create_node(&["Person"]);
3575            let gus = session.create_node(&["Person"]);
3576            let harm = session.create_node(&["Person"]);
3577
3578            session.create_edge(gus, alix, "KNOWS");
3579            session.create_edge(harm, alix, "KNOWS");
3580
3581            let neighbors = session.get_neighbors_incoming(alix);
3582            assert_eq!(neighbors.len(), 2);
3583
3584            let neighbor_ids: Vec<_> = neighbors.iter().map(|(node_id, _)| *node_id).collect();
3585            assert!(neighbor_ids.contains(&gus));
3586            assert!(neighbor_ids.contains(&harm));
3587        }
3588
3589        #[test]
3590        fn test_get_neighbors_outgoing_by_type() {
3591            let db = GrafeoDB::new_in_memory();
3592            let session = db.session();
3593
3594            let alix = session.create_node(&["Person"]);
3595            let gus = session.create_node(&["Person"]);
3596            let company = session.create_node(&["Company"]);
3597
3598            session.create_edge(alix, gus, "KNOWS");
3599            session.create_edge(alix, company, "WORKS_AT");
3600
3601            let knows_neighbors = session.get_neighbors_outgoing_by_type(alix, "KNOWS");
3602            assert_eq!(knows_neighbors.len(), 1);
3603            assert_eq!(knows_neighbors[0].0, gus);
3604
3605            let works_neighbors = session.get_neighbors_outgoing_by_type(alix, "WORKS_AT");
3606            assert_eq!(works_neighbors.len(), 1);
3607            assert_eq!(works_neighbors[0].0, company);
3608
3609            // No edges of this type
3610            let no_neighbors = session.get_neighbors_outgoing_by_type(alix, "LIKES");
3611            assert!(no_neighbors.is_empty());
3612        }
3613
3614        #[test]
3615        fn test_node_exists() {
3616            use grafeo_common::types::NodeId;
3617
3618            let db = GrafeoDB::new_in_memory();
3619            let session = db.session();
3620
3621            let id = session.create_node(&["Person"]);
3622
3623            assert!(session.node_exists(id));
3624            assert!(!session.node_exists(NodeId::new(9999)));
3625        }
3626
3627        #[test]
3628        fn test_edge_exists() {
3629            use grafeo_common::types::EdgeId;
3630
3631            let db = GrafeoDB::new_in_memory();
3632            let session = db.session();
3633
3634            let alix = session.create_node(&["Person"]);
3635            let gus = session.create_node(&["Person"]);
3636            let edge_id = session.create_edge(alix, gus, "KNOWS");
3637
3638            assert!(session.edge_exists(edge_id));
3639            assert!(!session.edge_exists(EdgeId::new(9999)));
3640        }
3641
3642        #[test]
3643        fn test_get_degree() {
3644            let db = GrafeoDB::new_in_memory();
3645            let session = db.session();
3646
3647            let alix = session.create_node(&["Person"]);
3648            let gus = session.create_node(&["Person"]);
3649            let harm = session.create_node(&["Person"]);
3650
3651            // Alix knows Gus and Harm (2 outgoing)
3652            session.create_edge(alix, gus, "KNOWS");
3653            session.create_edge(alix, harm, "KNOWS");
3654            // Gus knows Alix (1 incoming for Alix)
3655            session.create_edge(gus, alix, "KNOWS");
3656
3657            let (out_degree, in_degree) = session.get_degree(alix);
3658            assert_eq!(out_degree, 2);
3659            assert_eq!(in_degree, 1);
3660
3661            // Node with no edges
3662            let lonely = session.create_node(&["Person"]);
3663            let (out, in_deg) = session.get_degree(lonely);
3664            assert_eq!(out, 0);
3665            assert_eq!(in_deg, 0);
3666        }
3667
3668        #[test]
3669        fn test_get_nodes_batch() {
3670            let db = GrafeoDB::new_in_memory();
3671            let session = db.session();
3672
3673            let alix = session.create_node(&["Person"]);
3674            let gus = session.create_node(&["Person"]);
3675            let harm = session.create_node(&["Person"]);
3676
3677            let nodes = session.get_nodes_batch(&[alix, gus, harm]);
3678            assert_eq!(nodes.len(), 3);
3679            assert!(nodes[0].is_some());
3680            assert!(nodes[1].is_some());
3681            assert!(nodes[2].is_some());
3682
3683            // With non-existent node
3684            use grafeo_common::types::NodeId;
3685            let nodes_with_missing = session.get_nodes_batch(&[alix, NodeId::new(9999), harm]);
3686            assert_eq!(nodes_with_missing.len(), 3);
3687            assert!(nodes_with_missing[0].is_some());
3688            assert!(nodes_with_missing[1].is_none()); // Missing node
3689            assert!(nodes_with_missing[2].is_some());
3690        }
3691
3692        #[test]
3693        fn test_auto_commit_setting() {
3694            let db = GrafeoDB::new_in_memory();
3695            let mut session = db.session();
3696
3697            // Default is auto-commit enabled
3698            assert!(session.auto_commit());
3699
3700            session.set_auto_commit(false);
3701            assert!(!session.auto_commit());
3702
3703            session.set_auto_commit(true);
3704            assert!(session.auto_commit());
3705        }
3706
3707        #[test]
3708        fn test_transaction_double_begin_nests() {
3709            let db = GrafeoDB::new_in_memory();
3710            let mut session = db.session();
3711
3712            session.begin_transaction().unwrap();
3713            // Second begin_transaction creates a nested transaction (auto-savepoint)
3714            let result = session.begin_transaction();
3715            assert!(result.is_ok());
3716            // Commit the inner (releases savepoint)
3717            session.commit().unwrap();
3718            // Commit the outer
3719            session.commit().unwrap();
3720        }
3721
3722        #[test]
3723        fn test_commit_without_transaction_error() {
3724            let db = GrafeoDB::new_in_memory();
3725            let mut session = db.session();
3726
3727            let result = session.commit();
3728            assert!(result.is_err());
3729        }
3730
3731        #[test]
3732        fn test_rollback_without_transaction_error() {
3733            let db = GrafeoDB::new_in_memory();
3734            let mut session = db.session();
3735
3736            let result = session.rollback();
3737            assert!(result.is_err());
3738        }
3739
3740        #[test]
3741        fn test_create_edge_in_transaction() {
3742            let db = GrafeoDB::new_in_memory();
3743            let mut session = db.session();
3744
3745            // Create nodes outside transaction
3746            let alix = session.create_node(&["Person"]);
3747            let gus = session.create_node(&["Person"]);
3748
3749            // Create edge in transaction
3750            session.begin_transaction().unwrap();
3751            let edge_id = session.create_edge(alix, gus, "KNOWS");
3752
3753            // Edge should be visible in the transaction
3754            assert!(session.edge_exists(edge_id));
3755
3756            // Commit
3757            session.commit().unwrap();
3758
3759            // Edge should still be visible
3760            assert!(session.edge_exists(edge_id));
3761        }
3762
3763        #[test]
3764        fn test_neighbors_empty_node() {
3765            let db = GrafeoDB::new_in_memory();
3766            let session = db.session();
3767
3768            let lonely = session.create_node(&["Person"]);
3769
3770            assert!(session.get_neighbors_outgoing(lonely).is_empty());
3771            assert!(session.get_neighbors_incoming(lonely).is_empty());
3772            assert!(
3773                session
3774                    .get_neighbors_outgoing_by_type(lonely, "KNOWS")
3775                    .is_empty()
3776            );
3777        }
3778    }
3779
3780    #[test]
3781    fn test_auto_gc_triggers_on_commit_interval() {
3782        use crate::config::Config;
3783
3784        let config = Config::in_memory().with_gc_interval(2);
3785        let db = GrafeoDB::with_config(config).unwrap();
3786        let mut session = db.session();
3787
3788        // First commit: counter = 1, no GC (not a multiple of 2)
3789        session.begin_transaction().unwrap();
3790        session.create_node(&["A"]);
3791        session.commit().unwrap();
3792
3793        // Second commit: counter = 2, GC should trigger (multiple of 2)
3794        session.begin_transaction().unwrap();
3795        session.create_node(&["B"]);
3796        session.commit().unwrap();
3797
3798        // Verify the database is still functional after GC
3799        assert_eq!(db.node_count(), 2);
3800    }
3801
3802    #[test]
3803    fn test_query_timeout_config_propagates_to_session() {
3804        use crate::config::Config;
3805        use std::time::Duration;
3806
3807        let config = Config::in_memory().with_query_timeout(Duration::from_secs(5));
3808        let db = GrafeoDB::with_config(config).unwrap();
3809        let session = db.session();
3810
3811        // Verify the session has a query deadline (timeout was set)
3812        assert!(session.query_deadline().is_some());
3813    }
3814
3815    #[test]
3816    fn test_no_query_timeout_returns_no_deadline() {
3817        let db = GrafeoDB::new_in_memory();
3818        let session = db.session();
3819
3820        // Default config has no timeout
3821        assert!(session.query_deadline().is_none());
3822    }
3823
3824    #[test]
3825    fn test_graph_model_accessor() {
3826        use crate::config::GraphModel;
3827
3828        let db = GrafeoDB::new_in_memory();
3829        let session = db.session();
3830
3831        assert_eq!(session.graph_model(), GraphModel::Lpg);
3832    }
3833
3834    #[cfg(feature = "gql")]
3835    #[test]
3836    fn test_external_store_session() {
3837        use grafeo_core::graph::GraphStoreMut;
3838        use std::sync::Arc;
3839
3840        let config = crate::config::Config::in_memory();
3841        let store =
3842            Arc::new(grafeo_core::graph::lpg::LpgStore::new().unwrap()) as Arc<dyn GraphStoreMut>;
3843        let db = GrafeoDB::with_store(store, config).unwrap();
3844
3845        let session = db.session();
3846
3847        // Create data through a query (goes through the external graph_store)
3848        session.execute("INSERT (:Test {name: 'hello'})").unwrap();
3849
3850        // Verify we can query through it
3851        let result = session.execute("MATCH (n:Test) RETURN n.name").unwrap();
3852        assert_eq!(result.row_count(), 1);
3853    }
3854
3855    // ==================== Session Command Tests ====================
3856
3857    #[cfg(feature = "gql")]
3858    mod session_command_tests {
3859        use super::*;
3860
3861        #[test]
3862        fn test_use_graph_sets_current_graph() {
3863            let db = GrafeoDB::new_in_memory();
3864            let session = db.session();
3865
3866            // Create the graph first, then USE it
3867            session.execute("CREATE GRAPH mydb").unwrap();
3868            session.execute("USE GRAPH mydb").unwrap();
3869
3870            assert_eq!(session.current_graph(), Some("mydb".to_string()));
3871        }
3872
3873        #[test]
3874        fn test_use_graph_nonexistent_errors() {
3875            let db = GrafeoDB::new_in_memory();
3876            let session = db.session();
3877
3878            let result = session.execute("USE GRAPH doesnotexist");
3879            assert!(result.is_err());
3880            let err = result.unwrap_err().to_string();
3881            assert!(
3882                err.contains("does not exist"),
3883                "Expected 'does not exist' error, got: {err}"
3884            );
3885        }
3886
3887        #[test]
3888        fn test_use_graph_default_always_valid() {
3889            let db = GrafeoDB::new_in_memory();
3890            let session = db.session();
3891
3892            // "default" is always valid, even without CREATE GRAPH
3893            session.execute("USE GRAPH default").unwrap();
3894            assert_eq!(session.current_graph(), Some("default".to_string()));
3895        }
3896
3897        #[test]
3898        fn test_session_set_graph() {
3899            let db = GrafeoDB::new_in_memory();
3900            let session = db.session();
3901
3902            // SESSION SET GRAPH does not verify existence
3903            session.execute("SESSION SET GRAPH analytics").unwrap();
3904            assert_eq!(session.current_graph(), Some("analytics".to_string()));
3905        }
3906
3907        #[test]
3908        fn test_session_set_time_zone() {
3909            let db = GrafeoDB::new_in_memory();
3910            let session = db.session();
3911
3912            assert_eq!(session.time_zone(), None);
3913
3914            session.execute("SESSION SET TIME ZONE 'UTC'").unwrap();
3915            assert_eq!(session.time_zone(), Some("UTC".to_string()));
3916
3917            session
3918                .execute("SESSION SET TIME ZONE 'America/New_York'")
3919                .unwrap();
3920            assert_eq!(session.time_zone(), Some("America/New_York".to_string()));
3921        }
3922
3923        #[test]
3924        fn test_session_set_parameter() {
3925            let db = GrafeoDB::new_in_memory();
3926            let session = db.session();
3927
3928            session
3929                .execute("SESSION SET PARAMETER $timeout = 30")
3930                .unwrap();
3931
3932            // Parameter is stored (value is Null for now, since expression
3933            // evaluation is not yet wired up)
3934            assert!(session.get_parameter("timeout").is_some());
3935        }
3936
3937        #[test]
3938        fn test_session_reset_clears_all_state() {
3939            let db = GrafeoDB::new_in_memory();
3940            let session = db.session();
3941
3942            // Set various session state
3943            session.execute("SESSION SET GRAPH analytics").unwrap();
3944            session.execute("SESSION SET TIME ZONE 'UTC'").unwrap();
3945            session
3946                .execute("SESSION SET PARAMETER $limit = 100")
3947                .unwrap();
3948
3949            // Verify state was set
3950            assert!(session.current_graph().is_some());
3951            assert!(session.time_zone().is_some());
3952            assert!(session.get_parameter("limit").is_some());
3953
3954            // Reset everything
3955            session.execute("SESSION RESET").unwrap();
3956
3957            assert_eq!(session.current_graph(), None);
3958            assert_eq!(session.time_zone(), None);
3959            assert!(session.get_parameter("limit").is_none());
3960        }
3961
3962        #[test]
3963        fn test_session_close_clears_state() {
3964            let db = GrafeoDB::new_in_memory();
3965            let session = db.session();
3966
3967            session.execute("SESSION SET GRAPH analytics").unwrap();
3968            session.execute("SESSION SET TIME ZONE 'UTC'").unwrap();
3969
3970            session.execute("SESSION CLOSE").unwrap();
3971
3972            assert_eq!(session.current_graph(), None);
3973            assert_eq!(session.time_zone(), None);
3974        }
3975
3976        #[test]
3977        fn test_create_graph() {
3978            let db = GrafeoDB::new_in_memory();
3979            let session = db.session();
3980
3981            session.execute("CREATE GRAPH mydb").unwrap();
3982
3983            // Should be able to USE it now
3984            session.execute("USE GRAPH mydb").unwrap();
3985            assert_eq!(session.current_graph(), Some("mydb".to_string()));
3986        }
3987
3988        #[test]
3989        fn test_create_graph_duplicate_errors() {
3990            let db = GrafeoDB::new_in_memory();
3991            let session = db.session();
3992
3993            session.execute("CREATE GRAPH mydb").unwrap();
3994            let result = session.execute("CREATE GRAPH mydb");
3995
3996            assert!(result.is_err());
3997            let err = result.unwrap_err().to_string();
3998            assert!(
3999                err.contains("already exists"),
4000                "Expected 'already exists' error, got: {err}"
4001            );
4002        }
4003
4004        #[test]
4005        fn test_create_graph_if_not_exists() {
4006            let db = GrafeoDB::new_in_memory();
4007            let session = db.session();
4008
4009            session.execute("CREATE GRAPH mydb").unwrap();
4010            // Should succeed silently with IF NOT EXISTS
4011            session.execute("CREATE GRAPH IF NOT EXISTS mydb").unwrap();
4012        }
4013
4014        #[test]
4015        fn test_drop_graph() {
4016            let db = GrafeoDB::new_in_memory();
4017            let session = db.session();
4018
4019            session.execute("CREATE GRAPH mydb").unwrap();
4020            session.execute("DROP GRAPH mydb").unwrap();
4021
4022            // Should no longer be usable
4023            let result = session.execute("USE GRAPH mydb");
4024            assert!(result.is_err());
4025        }
4026
4027        #[test]
4028        fn test_drop_graph_nonexistent_errors() {
4029            let db = GrafeoDB::new_in_memory();
4030            let session = db.session();
4031
4032            let result = session.execute("DROP GRAPH nosuchgraph");
4033            assert!(result.is_err());
4034            let err = result.unwrap_err().to_string();
4035            assert!(
4036                err.contains("does not exist"),
4037                "Expected 'does not exist' error, got: {err}"
4038            );
4039        }
4040
4041        #[test]
4042        fn test_drop_graph_if_exists() {
4043            let db = GrafeoDB::new_in_memory();
4044            let session = db.session();
4045
4046            // Should succeed silently with IF EXISTS
4047            session.execute("DROP GRAPH IF EXISTS nosuchgraph").unwrap();
4048        }
4049
4050        #[test]
4051        fn test_start_transaction_via_gql() {
4052            let db = GrafeoDB::new_in_memory();
4053            let session = db.session();
4054
4055            session.execute("START TRANSACTION").unwrap();
4056            assert!(session.in_transaction());
4057            session.execute("INSERT (:Person {name: 'Alix'})").unwrap();
4058            session.execute("COMMIT").unwrap();
4059            assert!(!session.in_transaction());
4060
4061            let result = session.execute("MATCH (n:Person) RETURN n.name").unwrap();
4062            assert_eq!(result.rows.len(), 1);
4063        }
4064
4065        #[test]
4066        fn test_start_transaction_read_only_blocks_insert() {
4067            let db = GrafeoDB::new_in_memory();
4068            let session = db.session();
4069
4070            session.execute("START TRANSACTION READ ONLY").unwrap();
4071            let result = session.execute("INSERT (:Person {name: 'Alix'})");
4072            assert!(result.is_err());
4073            let err = result.unwrap_err().to_string();
4074            assert!(
4075                err.contains("read-only"),
4076                "Expected read-only error, got: {err}"
4077            );
4078            session.execute("ROLLBACK").unwrap();
4079        }
4080
4081        #[test]
4082        fn test_start_transaction_read_only_allows_reads() {
4083            let db = GrafeoDB::new_in_memory();
4084            let mut session = db.session();
4085            session.begin_transaction().unwrap();
4086            session.execute("INSERT (:Person {name: 'Alix'})").unwrap();
4087            session.commit().unwrap();
4088
4089            session.execute("START TRANSACTION READ ONLY").unwrap();
4090            let result = session.execute("MATCH (n:Person) RETURN n.name").unwrap();
4091            assert_eq!(result.rows.len(), 1);
4092            session.execute("COMMIT").unwrap();
4093        }
4094
4095        #[test]
4096        fn test_rollback_via_gql() {
4097            let db = GrafeoDB::new_in_memory();
4098            let session = db.session();
4099
4100            session.execute("START TRANSACTION").unwrap();
4101            session.execute("INSERT (:Person {name: 'Alix'})").unwrap();
4102            session.execute("ROLLBACK").unwrap();
4103
4104            let result = session.execute("MATCH (n:Person) RETURN n.name").unwrap();
4105            assert!(result.rows.is_empty());
4106        }
4107
4108        #[test]
4109        fn test_start_transaction_with_isolation_level() {
4110            let db = GrafeoDB::new_in_memory();
4111            let session = db.session();
4112
4113            session
4114                .execute("START TRANSACTION ISOLATION LEVEL SERIALIZABLE")
4115                .unwrap();
4116            assert!(session.in_transaction());
4117            session.execute("ROLLBACK").unwrap();
4118        }
4119
4120        #[test]
4121        fn test_session_commands_return_empty_result() {
4122            let db = GrafeoDB::new_in_memory();
4123            let session = db.session();
4124
4125            let result = session.execute("SESSION SET GRAPH test").unwrap();
4126            assert_eq!(result.row_count(), 0);
4127            assert_eq!(result.column_count(), 0);
4128        }
4129
4130        #[test]
4131        fn test_current_graph_default_is_none() {
4132            let db = GrafeoDB::new_in_memory();
4133            let session = db.session();
4134
4135            assert_eq!(session.current_graph(), None);
4136        }
4137
4138        #[test]
4139        fn test_time_zone_default_is_none() {
4140            let db = GrafeoDB::new_in_memory();
4141            let session = db.session();
4142
4143            assert_eq!(session.time_zone(), None);
4144        }
4145
4146        #[test]
4147        fn test_session_state_independent_across_sessions() {
4148            let db = GrafeoDB::new_in_memory();
4149            let session1 = db.session();
4150            let session2 = db.session();
4151
4152            session1.execute("SESSION SET GRAPH first").unwrap();
4153            session2.execute("SESSION SET GRAPH second").unwrap();
4154
4155            assert_eq!(session1.current_graph(), Some("first".to_string()));
4156            assert_eq!(session2.current_graph(), Some("second".to_string()));
4157        }
4158    }
4159}