graphar_flight/
falkor.rs

1use arrow_array::RecordBatch;
2use arrow_schema::SchemaRef;
3use redis::aio::MultiplexedConnection;
4
5use crate::{
6    convert::{response_to_batch, response_to_batch_auto},
7    error::Result,
8};
9
10/// Executes Cypher queries against FalkorDB and converts results to Arrow.
11///
12/// `MultiplexedConnection` is `Clone + Send + Sync`, so this struct is cheap
13/// to clone per-request — no mutex needed.
14#[derive(Clone)]
15pub struct FalkorExecutor {
16    conn: MultiplexedConnection,
17    graph: String,
18}
19
20impl FalkorExecutor {
21    pub async fn connect(url: &str, graph: impl Into<String>) -> Result<Self> {
22        let client = redis::Client::open(url)?;
23        let conn = client.get_multiplexed_async_connection().await?;
24        Ok(Self {
25            conn,
26            graph: graph.into(),
27        })
28    }
29
30    /// Run a read Cypher query; convert the response using the given schema.
31    pub async fn query(&mut self, cypher: &str, schema: &SchemaRef) -> Result<RecordBatch> {
32        let raw: redis::Value = redis::cmd("GRAPH.QUERY")
33            .arg(&self.graph)
34            .arg(cypher)
35            .query_async(&mut self.conn)
36            .await?;
37        response_to_batch(raw, schema)
38    }
39
40    /// Run a read Cypher query using the auto-string schema fallback.
41    ///
42    /// Column names are taken from the FalkorDB response header; all values
43    /// are returned as `Utf8`. No schema pre-registration is required.
44    pub async fn query_auto(&mut self, cypher: &str) -> Result<RecordBatch> {
45        let raw: redis::Value = redis::cmd("GRAPH.QUERY")
46            .arg(&self.graph)
47            .arg(cypher)
48            .query_async(&mut self.conn)
49            .await?;
50        response_to_batch_auto(raw)
51    }
52
53    /// Run a write Cypher query (CREATE / MERGE / DELETE …) with no result rows.
54    /// Returns the number of nodes/relationships affected (from the stats line).
55    pub async fn execute(&mut self, cypher: &str) -> Result<u64> {
56        let raw: redis::Value = redis::cmd("GRAPH.QUERY")
57            .arg(&self.graph)
58            .arg(cypher)
59            .query_async(&mut self.conn)
60            .await?;
61        Ok(parse_affected_count(raw))
62    }
63
64    pub fn graph(&self) -> &str {
65        &self.graph
66    }
67}
68
69/// Pull a rough "nodes created + relationships created" count from the stats
70/// array that FalkorDB appends to every response.
71fn parse_affected_count(response: redis::Value) -> u64 {
72    let outer = match response {
73        redis::Value::Array(v) => v,
74        _ => return 0,
75    };
76    // Stats are in the last element.
77    let stats = match outer.last() {
78        Some(redis::Value::Array(s)) => s,
79        _ => return 0,
80    };
81    let mut total = 0u64;
82    for stat in stats {
83        if let redis::Value::BulkString(b) = stat {
84            let s = String::from_utf8_lossy(b);
85            // e.g. "Nodes created: 3" or "Relationships created: 1"
86            if (s.contains("created:") || s.contains("deleted:"))
87                && let Some(n) = s
88                    .split(':')
89                    .nth(1)
90                    .and_then(|p| p.trim().parse::<u64>().ok())
91            {
92                total += n;
93            }
94        }
95    }
96    total
97}
graphar_flight/falkor.rs

graphar_flight/
falkor.rs
