graphar-flight 0.1.2

Apache Arrow Flight SQL service over FalkorDB — Cypher in, Arrow out
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

//! Graph → lakehouse export, end to end against a real FalkorDB/Valkey.
//!
//! Seeds a small graph, exports a Cypher query's results to a GraphAr parquet
//! file, reads it back, and asserts the round trip — the reverse of the ingest
//! bridge.
//!
//! ## Run
//! ```bash
//! cargo test -p graphar-flight --test integration_export -- --ignored --nocapture
//! ```

use std::time::Duration;

use graphar::FileType;
use graphar_flight::{FalkorExecutor, export::export_cypher};
use testcontainers::{GenericImage, ImageExt, core::ContainerPort, runners::AsyncRunner};

fn ensure_docker_host() {
    if std::env::var("DOCKER_HOST").is_err() {
        let uid = std::process::Command::new("id")
            .arg("-u")
            .output()
            .ok()
            .and_then(|o| String::from_utf8(o.stdout).ok())
            .and_then(|s| s.trim().parse::<u32>().ok())
            .unwrap_or(1000);
        let sock = format!("/run/user/{uid}/podman/podman.sock");
        if std::path::Path::new(&sock).exists() {
            unsafe { std::env::set_var("DOCKER_HOST", format!("unix://{sock}")) };
        }
    }
}

#[tokio::test]
#[ignore = "needs podman + a FalkorDB/Valkey image; run with --ignored"]
async fn export_round_trips_graph_to_graphar() {
    ensure_docker_host();

    let image = std::env::var("KNUT_FALKORDB_IMAGE")
        .unwrap_or_else(|_| "localhost/knut-falkordb-valkey9:latest".into());
    let (name, tag) = image.rsplit_once(':').unwrap_or((image.as_str(), "latest"));
    let falkor = GenericImage::new(name, tag)
        .with_exposed_port(ContainerPort::Tcp(6379))
        .with_startup_timeout(Duration::from_secs(120))
        .start()
        .await
        .expect("start falkordb/valkey");
    let port = falkor.get_host_port_ipv4(6379).await.unwrap();
    let url = format!("redis://127.0.0.1:{port}");
    let graph = "export";

    // Seed a tiny graph.
    let client = redis::Client::open(url.clone()).unwrap();
    let mut conn = client.get_multiplexed_async_connection().await.unwrap();
    let _: redis::Value = redis::cmd("GRAPH.QUERY")
        .arg(graph)
        .arg("CREATE (:Person {_gar_id: 0, name: 'alice'}), (:Person {_gar_id: 1, name: 'bob'})")
        .query_async(&mut conn)
        .await
        .unwrap();

    // Export query results → GraphAr parquet.
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("people.parquet");
    let mut exec = FalkorExecutor::connect(&url, graph).await.unwrap();
    let report = export_cypher(
        &mut exec,
        "MATCH (n:Person) RETURN n._gar_id AS id, n.name AS name",
        &path,
        FileType::Parquet,
    )
    .await
    .unwrap();
    assert_eq!(report.rows, 2, "two Person rows exported");
    assert_eq!(report.columns, 2, "id + name");

    // Read it back from the lakehouse side and verify.
    let batches = graphar::io::read_chunk(&path, &FileType::Parquet).unwrap();
    let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
    assert_eq!(rows, 2, "round-tripped 2 rows");
    let names: Vec<String> = batches
        .iter()
        .flat_map(|b| {
            let col = b.column_by_name("name").unwrap();
            let arr = col
                .as_any()
                .downcast_ref::<arrow_array::StringArray>()
                .unwrap();
            use arrow_array::Array;
            (0..arr.len())
                .map(|i| arr.value(i).to_string())
                .collect::<Vec<_>>()
        })
        .collect();
    assert!(names.contains(&"alice".to_string()) && names.contains(&"bob".to_string()));
}