feat: Add APOC export procedures (Neo4j-compatible) (#182)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: genezhang

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ### 🚀 Features
 
+- **APOC Export Procedures**: Neo4j-compatible `CALL apoc.export.{csv|json|parquet}.query(cypher, destination, config)` for exporting query results. Supports local files, S3, GCS, Azure, and HTTP destinations. Works in HTTP server, Bolt protocol, and embedded mode.
+  - **Destination resolver**: Maps URI schemes to ClickHouse `INSERT INTO FUNCTION` table functions (`file()`, `s3()`, `url()`, `azureBlobStorage()`)
+  - **Parser fix**: Standalone CALL with positional args now correctly parsed even when inner Cypher contains RETURN/UNION keywords
+  - **Config**: Parquet compression codecs (snappy, gzip, lz4, zstd, brotli)
+
 - **Embedded mode** (PR #179): Run Cypher graph queries entirely in-process via [chdb](https://github.com/chdb-io/chdb) — no external ClickHouse server required. Supports Parquet, CSV, Iceberg, Delta Lake, and S3-compatible storage.
   - **`QueryExecutor` trait**: Abstracts SQL execution; `RemoteClickHouseExecutor` (existing) and `ChdbExecutor` (new) are the two backends. Default behaviour is unchanged.
   - **`clickgraph-embedded` crate**: Kuzu-compatible Rust library API — `Database::new(schema, config)`, `Connection::new(&db)`, `conn.query(cypher)`, `result.next()` → `Row`.

STATUS.md

@@ -54,6 +54,7 @@ Supports both remote ClickHouse and embedded (in-process) mode via chdb.
 - **GraphRAG structured output**: `format: "Graph"` returns deduplicated nodes, edges, and stats
 - **ClickHouse cluster load balancing**: `CLICKHOUSE_CLUSTER` for auto-discovery and load balancing
 - **Embedded mode** (`--features embedded`): `QueryExecutor` trait + `ChdbExecutor` + `clickgraph-embedded` crate — run Cypher queries in-process over Parquet/Iceberg/Delta/S3 without a ClickHouse server. Kuzu-compatible Rust API (`Database`, `Connection`, `QueryResult`). `source:` URI field in YAML schema. S3/GCS/Azure credential support via `StorageCredentials`.
+- **APOC Export Procedures**: Neo4j-compatible `CALL apoc.export.{csv|json|parquet}.query(cypher, destination, config)` — translates inner Cypher to SQL, resolves destination URI (local file, S3, GCS, Azure, HTTP), wraps in `INSERT INTO FUNCTION`. Works in server mode (HTTP + Bolt) and embedded mode.
 
 ## Current Limitations
 

clickgraph-embedded/src/connection.rs

@@ -161,7 +161,107 @@ impl<'db> Connection<'db> {
         .await
     }
 
+    /// Handle `CALL apoc.export.{csv|json|parquet}.query(...)` in embedded mode.
+    ///
+    /// Parses arguments, translates inner Cypher → SQL, builds export SQL, executes.
+    /// Returns a single-row result with export status.
+    async fn handle_export_call(&self, cypher: &str) -> Result<QueryResult, EmbeddedError> {
+        use clickgraph::clickhouse_query_generator::cypher_to_sql;
+        use clickgraph::open_cypher_parser;
+        use clickgraph::open_cypher_parser::ast::CypherStatement;
+        use clickgraph::procedures::apoc_export;
+        use clickgraph::server::query_context::{
+            set_current_schema, with_query_context, QueryContext,
+        };
+
+        let schema = Arc::clone(&self.schema);
+        let executor = Arc::clone(&self.executor);
+        let cypher = cypher.to_string();
+
+        with_query_context(QueryContext::new(None), async move {
+            set_current_schema(Arc::clone(&schema));
+
+            // Parse the CALL statement
+            let (_, stmt) = open_cypher_parser::parse_cypher_statement(&cypher)
+                .map_err(|e| EmbeddedError::Query(format!("Parse error: {}", e)))?;
+
+            // Extract procedure name and arguments
+            let (proc_name, expressions): (String, Vec<_>) = match &stmt {
+                CypherStatement::ProcedureCall(pc) => {
+                    (pc.procedure_name.to_string(), pc.arguments.iter().collect())
+                }
+                CypherStatement::Query { query, .. } => {
+                    let cc = query
+                        .call_clause
+                        .as_ref()
+                        .ok_or_else(|| EmbeddedError::Query("No CALL clause found".to_string()))?;
+                    (
+                        cc.procedure_name.to_string(),
+                        cc.arguments.iter().map(|a| &a.value).collect(),
+                    )
+                }
+            };
+
+            let ch_format = apoc_export::format_from_procedure_name(&proc_name)
+                .map_err(EmbeddedError::Query)?;
+
+            let args =
+                apoc_export::parse_export_call(&expressions).map_err(EmbeddedError::Query)?;
+
+            // Translate inner Cypher → SQL
+            let inner_sql =
+                cypher_to_sql(&args.cypher_query, &schema, 100).map_err(EmbeddedError::Query)?;
+
+            // Build export SQL using the full destination resolver
+            let export_sql = apoc_export::build_export_sql(
+                &inner_sql,
+                &args.destination,
+                ch_format,
+                &args.config,
+            )
+            .map_err(EmbeddedError::Query)?;
+
+            // Execute
+            executor
+                .execute_text(&export_sql, "TabSeparated", None)
+                .await
+                .map_err(EmbeddedError::from)?;
+
+            // Return status as a single-row result
+            let columns = vec![
+                "file".to_string(),
+                "format".to_string(),
+                "source".to_string(),
+            ];
+            let rows = vec![vec![
+                Value::String(args.destination),
+                Value::String(ch_format.to_string()),
+                Value::String(args.cypher_query),
+            ]];
+            Ok(QueryResult::new(columns, rows))
+        })
+        .await
+    }
+
     async fn query_async(&self, cypher: &str) -> Result<QueryResult, EmbeddedError> {
+        // Intercept CALL apoc.export.* — parse first to avoid false positives
+        if let Ok((_, stmt)) = clickgraph::open_cypher_parser::parse_cypher_statement(cypher) {
+            let proc_name = match &stmt {
+                clickgraph::open_cypher_parser::ast::CypherStatement::ProcedureCall(pc) => {
+                    Some(pc.procedure_name.to_string())
+                }
+                clickgraph::open_cypher_parser::ast::CypherStatement::Query { query, .. } => query
+                    .call_clause
+                    .as_ref()
+                    .map(|cc| cc.procedure_name.to_string()),
+            };
+            if let Some(name) = proc_name {
+                if clickgraph::procedures::apoc_export::is_export_procedure(&name) {
+                    return self.handle_export_call(cypher).await;
+                }
+            }
+        }
+
         use clickgraph::clickhouse_query_generator::cypher_to_sql;
         use clickgraph::server::query_context::{
             set_current_schema, with_query_context, QueryContext,
@@ -386,4 +486,43 @@ graph_schema:
             "unknown extension without format should error"
         );
     }
+
+    #[test]
+    fn test_call_export_via_query() {
+        // Verify that CALL apoc.export.*.query() is intercepted and routed
+        // to the export handler (returns a status result, not SQL error)
+        let db = make_stub_db();
+        let conn = Connection::new(&db).unwrap();
+        let result = conn.query(
+            r#"CALL apoc.export.parquet.query("MATCH (u:User) RETURN u.name", "/tmp/users.parquet", {})"#,
+        );
+        // With stub executor, this should succeed (StubExecutor returns empty string)
+        assert!(
+            result.is_ok(),
+            "CALL export should be handled: {:?}",
+            result.err()
+        );
+        let qr = result.unwrap();
+        assert_eq!(qr.get_column_names(), &["file", "format", "source"]);
+    }
+
+    #[test]
+    fn test_call_export_csv_via_query() {
+        let db = make_stub_db();
+        let conn = Connection::new(&db).unwrap();
+        let result = conn.query(
+            r#"CALL apoc.export.csv.query("MATCH (u:User) RETURN u.name", "/tmp/users.csv", {})"#,
+        );
+        assert!(result.is_ok(), "CSV export should work: {:?}", result.err());
+    }
+
+    #[test]
+    fn test_call_export_s3_destination() {
+        let db = make_stub_db();
+        let conn = Connection::new(&db).unwrap();
+        let result = conn.query(
+            r#"CALL apoc.export.json.query("MATCH (u:User) RETURN u.name", "s3://mybucket/users.json", {})"#,
+        );
+        assert!(result.is_ok(), "S3 export should work: {:?}", result.err());
+    }
 }

docs/wiki/Cypher-Language-Reference.md

@@ -32,6 +32,7 @@ Complete syntax reference for Cypher queries supported by ClickGraph.
 - [System Procedures](#system-procedures) ⭐ **NEW**
   - [Schema Metadata](#schema-metadata-procedures)
   - [Graph Algorithms](#graph-algorithms)
+  - [Export Procedures (APOC-Compatible)](#export-procedures-apoc-compatible)
 - [Query Examples](#query-examples)
 
 ---
@@ -2231,7 +2232,52 @@ LIMIT 10
 
 ---
 
-## Query Examples
+### Export Procedures (APOC-Compatible)
+
+Export query results to files or cloud storage using Neo4j APOC-compatible syntax.
+
+**Syntax:**
+```cypher
+CALL apoc.export.{format}.query(cypher_query, destination, config)
+```
+
+**Formats:**
+| Procedure | ClickHouse Format |
+|-----------|------------------|
+| `apoc.export.csv.query` | CSVWithNames |
+| `apoc.export.json.query` | JSONEachRow |
+| `apoc.export.parquet.query` | Parquet |
+
+**Destination URIs:**
+| Scheme | Example |
+|--------|---------|
+| Local file | `"/tmp/users.parquet"` or `"./output.csv"` |
+| S3 | `"s3://bucket/path/file.parquet"` |
+| GCS | `"gs://bucket/path/file.json"` |
+| Azure | `"azure://container/path/file.csv"` |
+| HTTP | `"https://webhook.example.com/ingest"` |
+
+**Examples:**
+```cypher
+-- Export to local Parquet file
+CALL apoc.export.parquet.query("MATCH (u:User) RETURN u.name, u.email", "/tmp/users.parquet", {})
+
+-- Export to CSV
+CALL apoc.export.csv.query("MATCH (u:User)-[:FOLLOWS]->(f:User) RETURN u.name, f.name", "follows.csv", {})
+
+-- Export to S3 with compression
+CALL apoc.export.parquet.query("MATCH (u:User) RETURN u.name", "s3://my-bucket/users.parquet", {compression: "zstd"})
+
+-- Export to JSON
+CALL apoc.export.json.query("MATCH (u:User) WHERE u.country = 'US' RETURN u", "us_users.json", {})
+```
+
+**Config options:**
+- `compression`: Parquet codec — `"none"`, `"snappy"`, `"gzip"`, `"lz4"`, `"zstd"`, `"brotli"`
+
+> **Note**: Works in all modes — HTTP server, Bolt protocol, and embedded mode.
+
+---
 
 ### Simple Queries
 

docs/wiki/Embedded-Mode.md

@@ -177,6 +177,30 @@ print(conn.export_to_sql("MATCH (u:User) RETURN u.name", "users.parquet"))
 
 Supported file extensions: `.parquet` / `.pq`, `.csv`, `.tsv`, `.json`, `.ndjson` / `.jsonl`
 
+**APOC Export Procedures (Neo4j-compatible):**
+
+Use Neo4j APOC-style `CALL` syntax for exports with flexible destinations:
+
+```python
+conn = db.connect()
+
+# Export to local file
+conn.query('CALL apoc.export.parquet.query("MATCH (u:User) RETURN u.name", "/tmp/users.parquet", {})')
+
+# Export to CSV
+conn.query('CALL apoc.export.csv.query("MATCH (u:User) RETURN u.name, u.email", "users.csv", {})')
+
+# Export to S3
+conn.query('CALL apoc.export.json.query("MATCH (u:User) RETURN u", "s3://bucket/users.json", {})')
+
+# Export with Parquet compression
+conn.query('CALL apoc.export.parquet.query("MATCH (u:User) RETURN u.name", "output.parquet", {compression: "zstd"})')
+```
+
+Supported destinations: local files, `s3://`, `gs://`, `azure://`, `http://`, `https://`
+
+The APOC export syntax also works in server mode (HTTP and Bolt), providing a unified export interface across all deployment modes.
+
 ---
 
 ## Schema Configuration

src/open_cypher_parser/mod.rs

@@ -42,23 +42,20 @@ pub fn parse_cypher_statement(
 ) -> IResult<&'_ str, CypherStatement<'_>, OpenCypherParsingError<'_>> {
     let (input, _) = multispace0.parse(input)?;
 
-    // Check if this looks like a CALL with UNION or RETURN
-    // If so, parse as Query, not standalone procedure call
-    let input_upper = input.to_uppercase();
-    let has_union_in_query = input_upper.contains("UNION");
-    let has_return_in_query = input_upper.contains("RETURN");
-
-    if !has_union_in_query && !has_return_in_query {
-        // Try to parse as standalone procedure call
-        if let Ok((remaining, procedure_call)) =
-            standalone_procedure_call::parse_standalone_procedure_call(input)
-        {
-            // Optional trailing semicolon
+    // Try standalone procedure call first.
+    // Only use it if parsing consumed the entire input (modulo whitespace/semicolons).
+    // If the standalone parser succeeds but leaves unconsumed tokens (e.g., RETURN, UNION),
+    // fall through to the query parser which handles CALL as part of a larger query.
+    if let Ok((remaining, procedure_call)) =
+        standalone_procedure_call::parse_standalone_procedure_call(input)
+    {
+        let check = remaining.trim();
+        if check.is_empty() || check == ";" {
             let (input, _) = opt(ws(tag(";"))).parse(remaining)?;
             return Ok((input, CypherStatement::ProcedureCall(procedure_call)));
         }
     }
-    // Has UNION or RETURN (or standalone parse failed) - fall through to regular query parser
+    // Standalone parse didn't consume everything - fall through to regular query parser
 
     // Parse the first query
     let (input, first_query) = parse_query_with_nom.parse(input)?;