use super::{AcidTransaction, IsolationLevel, TransactionId, TransactionState};
use crate::model::{GraphName, NamedNode, Object, Predicate, Quad, Subject};
use crate::OxirsError;
use ahash::{AHashMap, AHashSet};
use std::sync::{Arc, RwLock};
pub struct NamedGraphTransaction {
inner: AcidTransaction,
graph_operations: AHashMap<GraphName, GraphOperations>,
graph_locks: Arc<RwLock<AHashSet<GraphName>>>,
}
#[derive(Debug, Clone, Default)]
struct GraphOperations {
inserts: Vec<Quad>,
deletes: Vec<Quad>,
cleared: bool,
created: bool,
dropped: bool,
}
impl NamedGraphTransaction {
pub(super) fn new(
inner: AcidTransaction,
graph_locks: Arc<RwLock<AHashSet<GraphName>>>,
) -> Self {
Self {
inner,
graph_operations: AHashMap::new(),
graph_locks,
}
}
pub fn id(&self) -> TransactionId {
self.inner.id()
}
pub fn state(&self) -> TransactionState {
self.inner.state()
}
pub fn isolation(&self) -> IsolationLevel {
self.inner.isolation()
}
pub fn insert_into_graph(
&mut self,
graph: GraphName,
subject: Subject,
predicate: Predicate,
object: Object,
) -> Result<bool, OxirsError> {
let quad = Quad::new(subject, predicate, object, graph.clone());
let ops = self.graph_operations.entry(graph).or_default();
ops.inserts.push(quad.clone());
self.inner.insert(quad)
}
pub fn delete_from_graph(
&mut self,
graph: GraphName,
subject: Subject,
predicate: Predicate,
object: Object,
) -> Result<bool, OxirsError> {
let quad = Quad::new(subject, predicate, object, graph.clone());
let ops = self.graph_operations.entry(graph).or_default();
ops.deletes.push(quad.clone());
self.inner.delete(quad)
}
pub fn clear_graph(&mut self, graph: GraphName) -> Result<usize, OxirsError> {
let ops = self.graph_operations.entry(graph.clone()).or_default();
ops.cleared = true;
Ok(0)
}
pub fn create_graph(&mut self, graph: NamedNode) -> Result<(), OxirsError> {
let graph_name = GraphName::NamedNode(graph);
let ops = self.graph_operations.entry(graph_name).or_default();
if ops.dropped {
return Err(OxirsError::Store(
"Cannot create a graph that was dropped in the same transaction".to_string(),
));
}
ops.created = true;
Ok(())
}
pub fn drop_graph(&mut self, graph: NamedNode) -> Result<(), OxirsError> {
let graph_name = GraphName::NamedNode(graph);
let should_remove = if let Some(ops) = self.graph_operations.get(&graph_name) {
ops.created
} else {
false
};
if should_remove {
self.graph_operations.remove(&graph_name);
} else {
let ops = self.graph_operations.entry(graph_name).or_default();
ops.dropped = true;
ops.cleared = true; }
Ok(())
}
pub fn copy_graph(
&mut self,
_source: GraphName,
_destination: GraphName,
) -> Result<usize, OxirsError> {
Ok(0)
}
pub fn move_graph(
&mut self,
source: GraphName,
destination: GraphName,
) -> Result<usize, OxirsError> {
let count = self.copy_graph(source.clone(), destination)?;
if let GraphName::NamedNode(node) = source {
self.drop_graph(node)?;
}
Ok(count)
}
pub fn add_graph(
&mut self,
_source: GraphName,
_destination: GraphName,
) -> Result<usize, OxirsError> {
Ok(0)
}
pub fn graph_stats(&self, graph: &GraphName) -> Option<GraphStats> {
self.graph_operations.get(graph).map(|ops| GraphStats {
inserts: ops.inserts.len(),
deletes: ops.deletes.len(),
cleared: ops.cleared,
created: ops.created,
dropped: ops.dropped,
})
}
pub fn modified_graphs(&self) -> Vec<GraphName> {
self.graph_operations.keys().cloned().collect()
}
pub fn lock_graph(&mut self, graph: GraphName) -> Result<(), OxirsError> {
if self.isolation() == IsolationLevel::Serializable {
let mut locks = self.graph_locks.write().map_err(|e| {
OxirsError::ConcurrencyError(format!("Failed to acquire graph lock: {}", e))
})?;
if !locks.insert(graph) {
return Err(OxirsError::ConcurrencyError(
"Graph is already locked by another transaction".to_string(),
));
}
}
Ok(())
}
fn release_locks(&self) -> Result<(), OxirsError> {
let mut locks = self.graph_locks.write().map_err(|e| {
OxirsError::ConcurrencyError(format!("Failed to release graph locks: {}", e))
})?;
for graph in self.graph_operations.keys() {
locks.remove(graph);
}
Ok(())
}
pub fn commit(self) -> Result<(), OxirsError> {
self.release_locks()?;
self.inner.commit()
}
pub fn rollback(self) -> Result<(), OxirsError> {
self.release_locks()?;
self.inner.abort()
}
}
#[derive(Debug, Clone)]
pub struct GraphStats {
pub inserts: usize,
pub deletes: usize,
pub cleared: bool,
pub created: bool,
pub dropped: bool,
}
#[cfg(test)]
mod tests {
use super::*;
use crate::model::{Literal, NamedNode};
fn create_test_quad(graph: GraphName) -> Quad {
Quad::new(
Subject::NamedNode(NamedNode::new("http://example.org/subject").expect("valid IRI")),
Predicate::from(NamedNode::new("http://example.org/predicate").expect("valid IRI")),
Object::Literal(Literal::new("test")),
graph,
)
}
#[test]
fn test_graph_operations_tracking() {
let graph =
GraphName::NamedNode(NamedNode::new("http://example.org/graph1").expect("valid IRI"));
let mut ops = GraphOperations::default();
ops.inserts.push(create_test_quad(graph.clone()));
assert_eq!(ops.inserts.len(), 1);
assert_eq!(ops.deletes.len(), 0);
assert!(!ops.cleared);
}
#[test]
fn test_graph_stats() {
let mut ops = GraphOperations::default();
ops.inserts.push(create_test_quad(GraphName::DefaultGraph));
ops.cleared = true;
assert_eq!(ops.inserts.len(), 1);
assert!(ops.cleared);
}
}