use crate::{
client::blocking::FalkorSyncClientInner,
graph::{generate_create_index_query, generate_drop_index_query, HasGraphSchema},
Constraint, ConstraintType, EntityType, ExecutionPlan, FalkorIndex, FalkorResult, GraphSchema,
IndexType, LazyResultSet, ProcedureQueryBuilder, QueryBuilder, QueryResult, SlowlogEntry,
};
use std::{collections::HashMap, fmt::Display, sync::Arc};
#[derive(Clone)]
pub struct SyncGraph {
client: Arc<FalkorSyncClientInner>,
graph_name: String,
graph_schema: GraphSchema,
}
impl SyncGraph {
pub(crate) fn new<T: ToString>(
client: Arc<FalkorSyncClientInner>,
graph_name: T,
) -> Self {
Self {
graph_name: graph_name.to_string(),
graph_schema: GraphSchema::new(graph_name, client.clone()), client,
}
}
pub fn graph_name(&self) -> &str {
self.graph_name.as_str()
}
pub(crate) fn get_client(&self) -> &Arc<FalkorSyncClientInner> {
&self.client
}
#[cfg_attr(
feature = "tracing",
tracing::instrument(name = "Graph Execute Command", skip_all, level = "info")
)]
fn execute_command(
&self,
command: &str,
subcommand: Option<&str>,
params: Option<&[&str]>,
) -> FalkorResult<redis::Value> {
self.client
.borrow_connection(self.client.clone())?
.execute_command(Some(self.graph_name.as_str()), command, subcommand, params)
}
#[cfg_attr(
feature = "tracing",
tracing::instrument(name = "Delete Graph", skip_all, level = "info")
)]
pub fn delete(&mut self) -> FalkorResult<()> {
self.execute_command("GRAPH.DELETE", None, None)?;
self.graph_schema.clear();
Ok(())
}
#[cfg_attr(
feature = "tracing",
tracing::instrument(name = "Get Graph Slowlog", skip_all, level = "info")
)]
pub fn slowlog(&self) -> FalkorResult<Vec<SlowlogEntry>> {
self.execute_command("GRAPH.SLOWLOG", None, None)
.and_then(crate::response::slowlog_entry::parse_slowlog)
}
#[cfg_attr(
feature = "tracing",
tracing::instrument(name = "Reset Graph Slowlog", skip_all, level = "info")
)]
pub fn slowlog_reset(&self) -> FalkorResult<redis::Value> {
self.execute_command("GRAPH.SLOWLOG", None, Some(&["RESET"]))
}
pub fn profile<'a>(
&'a mut self,
query_string: &'a str,
) -> QueryBuilder<'a, ExecutionPlan, &'a str, Self> {
QueryBuilder::<'a>::new(self, "GRAPH.PROFILE", query_string)
}
pub fn explain<'a>(
&'a mut self,
query_string: &'a str,
) -> QueryBuilder<'a, ExecutionPlan, &'a str, Self> {
QueryBuilder::new(self, "GRAPH.EXPLAIN", query_string)
}
pub fn query<T: Display>(
&mut self,
query_string: T,
) -> QueryBuilder<QueryResult<LazyResultSet>, T, Self> {
QueryBuilder::new(self, "GRAPH.QUERY", query_string)
}
pub fn ro_query<'a>(
&'a mut self,
query_string: &'a str,
) -> QueryBuilder<'a, QueryResult<LazyResultSet<'a>>, &'a str, Self> {
QueryBuilder::new(self, "GRAPH.RO_QUERY", query_string)
}
pub fn batch(&mut self) -> crate::BatchBuilder<'_, Self> {
crate::BatchBuilder::new(self)
}
pub fn call_procedure<'a, P>(
&'a mut self,
procedure_name: &'a str,
) -> ProcedureQueryBuilder<'a, P, Self> {
ProcedureQueryBuilder::new(self, procedure_name)
}
pub fn call_procedure_ro<'a, P>(
&'a mut self,
procedure_name: &'a str,
) -> ProcedureQueryBuilder<'a, P, Self> {
ProcedureQueryBuilder::new_readonly(self, procedure_name)
}
#[cfg_attr(
feature = "tracing",
tracing::instrument(name = "List Graph Indices", skip_all, level = "info")
)]
pub fn list_indices(&mut self) -> FalkorResult<QueryResult<Vec<FalkorIndex>>> {
ProcedureQueryBuilder::<QueryResult<Vec<FalkorIndex>>, Self>::new(self, "DB.INDEXES")
.read_only()
.execute()
}
#[cfg_attr(
feature = "tracing",
tracing::instrument(name = "Graph Create Index", skip_all, level = "info")
)]
pub fn create_index<P: Display>(
&mut self,
index_field_type: IndexType,
entity_type: EntityType,
label: &str,
properties: &[P],
options: Option<&HashMap<String, String>>,
) -> FalkorResult<QueryResult<LazyResultSet>> {
let query_str =
generate_create_index_query(index_field_type, entity_type, label, properties, options);
QueryBuilder::<QueryResult<LazyResultSet>, String, Self>::new(
self,
"GRAPH.QUERY",
query_str,
)
.execute()
}
#[cfg_attr(
feature = "tracing",
tracing::instrument(name = "Graph Drop Index", skip_all, level = "info")
)]
pub fn drop_index<P: Display>(
&mut self,
index_field_type: IndexType,
entity_type: EntityType,
label: &str,
properties: &[P],
) -> FalkorResult<QueryResult<LazyResultSet>> {
let query_str = generate_drop_index_query(index_field_type, entity_type, label, properties);
self.query(query_str).execute()
}
#[cfg_attr(
feature = "tracing",
tracing::instrument(name = "List Graph Constraints", skip_all, level = "info")
)]
pub fn list_constraints(&mut self) -> FalkorResult<QueryResult<Vec<Constraint>>> {
ProcedureQueryBuilder::<QueryResult<Vec<Constraint>>, Self>::new(self, "DB.CONSTRAINTS")
.read_only()
.execute()
}
#[cfg_attr(
feature = "tracing",
tracing::instrument(name = "Create Graph Mandatory Constraint", skip_all, level = "info")
)]
pub fn create_mandatory_constraint(
&self,
entity_type: EntityType,
label: &str,
properties: &[&str],
) -> FalkorResult<redis::Value> {
let entity_type = entity_type.to_string();
let properties_count = properties.len().to_string();
let mut params = Vec::with_capacity(5 + properties.len());
params.extend([
"MANDATORY",
entity_type.as_str(),
label,
"PROPERTIES",
properties_count.as_str(),
]);
params.extend(properties);
self.execute_command("GRAPH.CONSTRAINT", Some("CREATE"), Some(params.as_slice()))
}
#[cfg_attr(
feature = "tracing",
tracing::instrument(name = "Create Graph Unique Constraint", skip_all, level = "info")
)]
pub fn create_unique_constraint(
&mut self,
entity_type: EntityType,
label: String,
properties: &[&str],
) -> FalkorResult<redis::Value> {
self.create_index(
IndexType::Range,
entity_type,
label.as_str(),
properties,
None,
)?;
let entity_type = entity_type.to_string();
let properties_count = properties.len().to_string();
let mut params: Vec<&str> = Vec::with_capacity(5 + properties.len());
params.extend([
"UNIQUE",
entity_type.as_str(),
label.as_str(),
"PROPERTIES",
properties_count.as_str(),
]);
params.extend(properties);
self.execute_command("GRAPH.CONSTRAINT", Some("CREATE"), Some(params.as_slice()))
}
#[cfg_attr(
feature = "tracing",
tracing::instrument(name = "Drop Graph Constraint", skip_all, level = "info")
)]
pub fn drop_constraint(
&self,
constraint_type: ConstraintType,
entity_type: EntityType,
label: &str,
properties: &[&str],
) -> FalkorResult<redis::Value> {
let constraint_type = constraint_type.to_string();
let entity_type = entity_type.to_string();
let properties_count = properties.len().to_string();
let mut params = Vec::with_capacity(5 + properties.len());
params.extend([
constraint_type.as_str(),
entity_type.as_str(),
label,
"PROPERTIES",
properties_count.as_str(),
]);
params.extend(properties);
self.execute_command("GRAPH.CONSTRAINT", Some("DROP"), Some(params.as_slice()))
}
}
impl HasGraphSchema for SyncGraph {
fn get_graph_schema_mut(&mut self) -> &mut GraphSchema {
&mut self.graph_schema
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{
test_utils::{create_test_client, imdb_test_client, open_empty_test_graph, retry_until},
FalkorDBError, IndexStatus, IndexType, WaitOptions,
};
#[test]
fn test_call_procedure_ro_routes_read_only() {
let mut graph = create_test_client().select_graph("imdb");
let result = graph
.call_procedure_ro::<QueryResult<Vec<FalkorIndex>>>("DB.INDEXES")
.execute();
assert!(result.is_ok());
}
#[test]
fn test_create_drop_index() {
let mut graph = open_empty_test_graph("test_create_drop_index");
let indices = graph
.inner
.create_index(
IndexType::Fulltext,
EntityType::Node,
"actor",
&["Hello"],
None,
)
.expect("Could not create index");
assert_eq!(indices.get_indices_created(), Some(1));
let indices = retry_until(
|| graph.inner.list_indices().expect("Could not list indices"),
|indices| indices.data.len() == 1,
);
assert_eq!(indices.data.len(), 1);
assert_eq!(
indices.data[0].field_types["Hello"],
vec![IndexType::Fulltext]
);
let indices = graph
.inner
.drop_index(IndexType::Fulltext, EntityType::Node, "actor", &["Hello"])
.expect("Could not drop index");
assert_eq!(indices.get_indices_deleted(), Some(1));
}
#[test]
fn test_invalid_cypher_query_syntax_returns_error() {
let mut graph = create_test_client().select_graph("imdb");
let res = graph.query("not a cypher query").execute();
assert!(matches!(res, Err(FalkorDBError::RedisError(_))));
}
#[test]
fn test_list_indices() {
let mut graph = imdb_test_client().select_graph("imdb");
let indices = graph.list_indices().expect("Could not list indices");
assert_eq!(indices.data.len(), 1);
assert_eq!(indices.data[0].entity_type, EntityType::Node);
assert_eq!(indices.data[0].index_label, "actor".to_string());
assert_eq!(indices.data[0].field_types.len(), 2);
assert_eq!(
indices.data[0].field_types,
HashMap::from([
("name".to_string(), vec![IndexType::Fulltext]),
("age".to_string(), vec![IndexType::Range])
])
);
}
#[test]
fn test_create_drop_mandatory_constraint() {
let graph = open_empty_test_graph("test_mandatory_constraint");
graph
.inner
.create_mandatory_constraint(EntityType::Edge, "act", &["hello", "goodbye"])
.expect("Could not create constraint");
graph
.inner
.drop_constraint(
ConstraintType::Mandatory,
EntityType::Edge,
"act",
&["hello", "goodbye"],
)
.expect("Could not drop constraint");
}
#[test]
fn test_create_drop_unique_constraint() {
let mut graph = open_empty_test_graph("test_unique_constraint");
graph
.inner
.create_unique_constraint(
EntityType::Node,
"actor".to_string(),
&["first_name", "last_name"],
)
.expect("Could not create constraint");
graph
.inner
.drop_constraint(
ConstraintType::Unique,
EntityType::Node,
"actor",
&["first_name", "last_name"],
)
.expect("Could not drop constraint");
}
#[test]
fn test_list_constraints() {
let mut graph = open_empty_test_graph("test_list_constraints");
graph
.inner
.create_unique_constraint(
EntityType::Node,
"actor".to_string(),
&["first_name", "last_name"],
)
.expect("Could not create constraint");
let res = retry_until(
|| {
graph
.inner
.list_constraints()
.expect("Could not list constraints")
},
|res| res.data.len() == 1,
);
assert_eq!(res.data.len(), 1);
}
#[test]
fn test_create_index_op_execute_is_non_blocking() {
let mut graph = open_empty_test_graph("test_create_index_op_execute");
let res = graph
.inner
.create_index_op(
IndexType::Fulltext,
EntityType::Node,
"actor",
&["name"],
None,
)
.execute()
.expect("Could not create index");
assert_eq!(res.get_indices_created(), Some(1));
retry_until(
|| {
graph
.inner
.list_indices()
.expect("Could not list indices")
.data
},
|indices| {
indices.iter().any(|index| {
index.index_label == "actor"
&& index
.field_types
.get("name")
.is_some_and(|types| types.contains(&IndexType::Fulltext))
})
},
);
let res = graph
.inner
.drop_index_op(IndexType::Fulltext, EntityType::Node, "actor", &["name"])
.execute()
.expect("Could not drop index");
assert_eq!(res.get_indices_deleted(), Some(1));
}
#[test]
fn test_create_drop_index_op_wait() {
let mut graph = open_empty_test_graph("test_create_index_op_wait");
graph
.inner
.create_index_op(IndexType::Range, EntityType::Node, "person", &["age"], None)
.wait()
.expect("Index did not become operational");
let indices = graph
.inner
.list_indices()
.expect("Could not list indices")
.data;
assert!(indices.iter().any(|index| {
index.index_label == "person"
&& index.status == IndexStatus::Active
&& index
.field_types
.get("age")
.is_some_and(|types| types.contains(&IndexType::Range))
}));
graph
.inner
.drop_index_op(IndexType::Range, EntityType::Node, "person", &["age"])
.wait()
.expect("Index was not dropped");
let indices = graph
.inner
.list_indices()
.expect("Could not list indices")
.data;
assert!(indices.is_empty());
}
#[test]
fn test_mandatory_constraint_op_wait() {
let mut graph = open_empty_test_graph("test_mandatory_constraint_op_wait");
graph
.inner
.create_mandatory_constraint_op(EntityType::Node, "person", &["name"])
.wait()
.expect("Constraint did not become operational");
graph
.inner
.drop_constraint_op(
ConstraintType::Mandatory,
EntityType::Node,
"person",
&["name"],
)
.wait()
.expect("Constraint was not dropped");
}
#[test]
fn test_constraint_op_execute_is_non_blocking() {
let mut graph = open_empty_test_graph("test_constraint_op_execute");
graph
.inner
.create_mandatory_constraint_op(EntityType::Node, "person", &["name"])
.execute()
.expect("Could not create constraint");
}
#[test]
fn test_drop_index_op_wait_errors_when_missing() {
let mut graph = open_empty_test_graph("test_drop_index_op_missing");
let result = graph
.inner
.drop_index_op(IndexType::Range, EntityType::Node, "person", &["age"])
.wait();
assert!(result.is_err());
}
#[test]
fn test_unique_constraint_op_wait() {
let mut graph = open_empty_test_graph("test_unique_constraint_op_wait");
graph
.inner
.create_unique_constraint_op(EntityType::Node, "person", &["email"])
.wait()
.expect("Constraint did not become operational");
graph
.inner
.drop_constraint_op(
ConstraintType::Unique,
EntityType::Node,
"person",
&["email"],
)
.wait()
.expect("Constraint was not dropped");
}
#[test]
fn test_unique_constraint_op_wait_reports_failure() {
let mut graph = open_empty_test_graph("test_unique_constraint_op_failed");
graph
.inner
.query("CREATE (:person {email: 'dup'}), (:person {email: 'dup'})")
.execute()
.expect("Could not seed conflicting data");
let result = graph
.inner
.create_unique_constraint_op(EntityType::Node, "person", &["email"])
.wait_with(WaitOptions::with_timeout(std::time::Duration::from_secs(
10,
)));
assert_eq!(
result,
Err(FalkorDBError::ConstraintFailed {
label: "person".to_string(),
properties: vec!["email".to_string()],
constraint_type: ConstraintType::Unique,
})
);
}
#[test]
#[ignore] fn test_slowlog() {
let mut graph = open_empty_test_graph("test_slowlog");
graph
.inner
.query("UNWIND range(0, 500) AS x RETURN x")
.execute()
.expect("Could not generate the fast query");
graph
.inner
.query("UNWIND range(0, 100000) AS x RETURN x")
.execute()
.expect("Could not generate the slow query");
let slowlog = graph
.inner
.slowlog()
.expect("Could not get slowlog entries");
assert_eq!(slowlog.len(), 2);
assert_eq!(
slowlog[0].arguments,
"UNWIND range(0, 500) AS x RETURN x".to_string()
);
assert_eq!(
slowlog[1].arguments,
"UNWIND range(0, 100000) AS x RETURN x".to_string()
);
graph
.inner
.slowlog_reset()
.expect("Could not reset slowlog memory");
let slowlog_after_reset = graph
.inner
.slowlog()
.expect("Could not get slowlog entries after reset");
assert!(slowlog_after_reset.is_empty());
}
#[test]
fn test_explain() {
let mut graph = imdb_test_client().select_graph("imdb");
let execution_plan = graph.explain("MATCH (a:actor) WITH a MATCH (b:actor) WHERE a.age = b.age AND a <> b RETURN a, collect(b) LIMIT 100").execute().expect("Could not create execution plan");
assert_eq!(execution_plan.plan().len(), 7);
assert!(execution_plan.operations().get("Aggregate").is_some());
assert_eq!(execution_plan.operations()["Aggregate"].len(), 1);
assert_eq!(
execution_plan.string_representation(),
"\nResults\n Limit\n Aggregate\n Filter\n Node By Index Scan | (b:actor)\n Project\n Node By Label Scan | (a:actor)"
);
}
#[test]
fn test_profile() {
let mut graph = open_empty_test_graph("test_profile");
let execution_plan = graph
.inner
.profile("UNWIND range(0, 1000) AS x RETURN x")
.execute()
.expect("Could not generate the query");
assert_eq!(execution_plan.plan().len(), 3);
let expected = vec!["Results", "Project", "Unwind"];
let mut current_rc = execution_plan.operation_tree().clone();
for step in expected {
assert_eq!(current_rc.name, step);
if step != "Unwind" {
current_rc = current_rc.children[0].clone();
}
}
}
}