feat(sql): support SQLEngine to run graph patterns on the same CSR

crates/lance-csr-catalog/Cargo.toml

@@ -7,7 +7,7 @@ authors = ["The lance-csr Authors"]
 repository = "https://github.com/hyperdimensionalcomputingai/lance-csr"
 homepage = "https://lance-csr.hdclabs.ai/"
 readme = "README.md"
-description = "Catalog and namespace utilities for Lance graph"
+description = "Table catalog and directory namespace for the lance-csr graph query engine"
 keywords = ["lance", "graph", "catalog", "namespace"]
 categories = ["database", "data-structures", "science"]
 

crates/lance-csr-catalog/README.md

@@ -1,6 +1,34 @@
-# Lance Graph Catalog
+# lance-csr-catalog
 
-Catalog and namespace utilities shared by the Lance graph query engine.
+The table lookups behind [lance-csr](https://crates.io/crates/lance-csr), a Cypher graph query engine for
+[Lance](https://lancedb.github.io/lance/) datasets built on Apache DataFusion. lance-csr reads a graph from
+ordinary tables: one per node label and one per relationship type. This crate answers the two questions that
+raises: which DataFusion table holds a given label or type, and where a named Lance dataset lives.
 
-> **Experimental.** The Rust API is not stable yet: breaking changes are possible over the next few
-> weeks while the CSR work lands. Pin an exact version (`lance-csr-catalog = "=x.y.z"`) if you depend on it.
+- `GraphSourceCatalog` is the trait the lance-csr planner asks for a node label's or relationship type's
+  DataFusion `TableSource`, so that it can build scans while planning. Lookups ignore case.
+  `InMemoryCatalog` is a map-backed implementation, and `SimpleTableSource` is a table source with a fixed
+  schema and no data, for planning and tests.
+- `DirNamespace` implements Lance's `LanceNamespace` for a directory of datasets: the table `person` under
+  `s3://bucket/graph` is `s3://bucket/graph/person.lance`. It works with any URI Lance can open.
+
+lance-csr re-exports all four types, so most programs only depend on `lance-csr`. Depend on this crate directly to
+implement your own catalog, for example over a metadata service, without pulling in the query engine.
+
+```rust
+use std::sync::Arc;
+
+use arrow_schema::{DataType, Field, Schema};
+use lance_csr_catalog::{GraphSourceCatalog, InMemoryCatalog, SimpleTableSource};
+
+let person = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
+let catalog = InMemoryCatalog::new()
+    .with_node_source("Person", Arc::new(SimpleTableSource::new(person)));
+assert!(catalog.node_source("person").is_some());
+```
+
+The [Rust guide](https://lance-csr.hdclabs.ai/rust/) covers the engine that uses these types.
+
+## License
+
+Apache-2.0. See `LICENSE` and `NOTICE`.

crates/lance-csr-catalog/src/lib.rs

@@ -1,10 +1,34 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The Lance Authors
 
-//! Catalog and namespace utilities for Lance Graph.
+//! The table lookups behind [lance-csr](https://docs.rs/lance-csr), a Cypher graph query
+//! engine for Lance datasets built on Apache DataFusion.
 //!
-//! **Experimental.** The Rust API is not stable yet: breaking changes are possible over the
-//! next few weeks while the CSR work lands. Pin an exact version if you depend on it.
+//! lance-csr reads a graph from ordinary tables: one per node label and one per relationship
+//! type. This crate answers the two questions that raises:
+//!
+//! - [`GraphSourceCatalog`]: which DataFusion `TableSource` holds a node label or relationship
+//!   type, so the planner can build scans. Lookups ignore case. [`InMemoryCatalog`] is a
+//!   map-backed implementation, and [`SimpleTableSource`] is a table source with a fixed schema
+//!   and no data, for planning and tests.
+//! - [`DirNamespace`]: where a named Lance dataset lives, as Lance's `LanceNamespace` for a
+//!   directory of datasets. The table `person` under `s3://bucket/graph` is
+//!   `s3://bucket/graph/person.lance`.
+//!
+//! lance-csr re-exports all four types, so most programs only depend on `lance-csr`. Depend on
+//! this crate directly to implement your own catalog without pulling in the query engine.
+//!
+//! ```
+//! use std::sync::Arc;
+//!
+//! use arrow_schema::{DataType, Field, Schema};
+//! use lance_csr_catalog::{GraphSourceCatalog, InMemoryCatalog, SimpleTableSource};
+//!
+//! let person = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
+//! let catalog = InMemoryCatalog::new()
+//!     .with_node_source("Person", Arc::new(SimpleTableSource::new(person)));
+//! assert!(catalog.node_source("person").is_some());
+//! ```
 
 pub mod namespace;
 pub mod source_catalog;

crates/lance-csr-python/src/graph.rs

@@ -21,15 +21,13 @@ use arrow::compute::concat_batches;
 use arrow::ffi_stream::ArrowArrayStreamReader;
 use arrow_array::{RecordBatch, RecordBatchReader};
 use arrow_schema::Schema;
-use datafusion::datasource::MemTable;
-use datafusion::execution::context::SessionContext;
 use futures::StreamExt;
 use lance_csr::{
     ast::{DistanceMetric as RustDistanceMetric, GraphPattern, ReadingClause},
     engine::{CypherEngine as RustCypherEngine, EngineOptions, EngineSource, MAX_TARGET_PARTITIONS},
     CypherQuery as RustCypherQuery, ExecutionStrategy as RustExecutionStrategy,
     GraphConfig as RustGraphConfig, GraphError as RustGraphError, SqlDialect as RustSqlDialect,
-    SqlQuery as RustSqlQuery, VectorSearch as RustVectorSearch,
+    SqlEngine as RustSqlEngine, SqlQuery as RustSqlQuery, VectorSearch as RustVectorSearch,
 };
 use pyo3::{
     exceptions::{PyNotImplementedError, PyRuntimeError, PyTypeError, PyValueError},
@@ -1009,7 +1007,7 @@ fn reader_with_deadline(
 
 #[pyclass(name = "CypherEngine", module = "lance_csr.graph")]
 pub struct CypherEngine {
-    inner: RustCypherEngine,
+    inner: Arc<RustCypherEngine>,
 }
 
 #[pymethods]
@@ -1066,7 +1064,9 @@ impl CypherEngine {
                 ),
             )?
             .map_err(graph_error_to_pyerr)?;
-        Ok(Self { inner })
+        Ok(Self {
+            inner: Arc::new(inner),
+        })
     }
 
     #[pyo3(signature = (query, params = None, *, timeout = None))]
@@ -1251,54 +1251,87 @@ impl SqlQuery {
 
 #[pyclass(name = "SqlEngine", module = "lance_csr.graph")]
 pub struct SqlEngine {
-    context: Arc<SessionContext>,
+    inner: RustSqlEngine,
 }
 
 #[pymethods]
 impl SqlEngine {
     #[new]
-    fn new(datasets: &Bound<'_, PyDict>) -> PyResult<Self> {
-        let arrow_datasets = python_datasets_to_batches(datasets)?;
-
-        if arrow_datasets.is_empty() {
-            return Err(PyValueError::new_err("No input datasets provided"));
+    #[pyo3(signature = (datasets, config = None, use_csr = false))]
+    fn new(
+        py: Python,
+        datasets: &Bound<'_, PyDict>,
+        config: Option<&GraphConfig>,
+        use_csr: bool,
+    ) -> PyResult<Self> {
+        if use_csr && config.is_none() {
+            return Err(PyValueError::new_err("use_csr requires a config"));
         }
-
-        let ctx = SessionContext::new();
-
-        for (name, batch) in &arrow_datasets {
-            let mem_table = Arc::new(
-                MemTable::try_new(batch.schema(), vec![vec![batch.clone()]]).map_err(|e| {
-                    PyRuntimeError::new_err(format!(
-                        "Failed to create MemTable for {}: {}",
-                        name, e
-                    ))
-                })?,
-            );
-
-            let normalized_name = name.to_lowercase();
-            ctx.register_table(&normalized_name, mem_table)
-                .map_err(|e| {
-                    PyRuntimeError::new_err(format!("Failed to register table {}: {}", name, e))
-                })?;
+        let mut sources = HashMap::new();
+        for (key, value) in datasets.iter() {
+            let name: String = key.extract()?;
+            // A Lance dataset opened through a namespace can't be reopened by URI, so
+            // without a graph it is read into memory, as SqlEngine always did.
+            let source = match python_value_to_source(py, &name, &value) {
+                Err(_) if config.is_none() && is_lance_dataset(&value)? => {
+                    EngineSource::Arrow(python_dataset_to_batch(&value)?)
+                }
+                source => source?,
+            };
+            sources.insert(name, source);
         }
-
-        Ok(Self {
-            context: Arc::new(ctx),
-        })
+        let config = config.map(|c| c.inner.clone()).unwrap_or_default();
+        let options = EngineOptions {
+            use_csr,
+            ..Default::default()
+        };
+        let inner = RT
+            .block_on(Some(py), RustSqlEngine::with_graph(config, sources, options))?
+            .map_err(graph_error_to_pyerr)?;
+        Ok(Self { inner })
     }
 
-    fn execute(&self, py: Python, sql: &str) -> PyResult<Py<PyAny>> {
-        let query = RustSqlQuery::new(sql);
-        let context = self.context.as_ref().clone();
+    #[staticmethod]
+    fn from_engine(engine: &CypherEngine) -> Self {
+        Self {
+            inner: RustSqlEngine::from_engine(engine.inner.clone()),
+        }
+    }
 
+    #[pyo3(signature = (sql, params = None, *, timeout = None))]
+    fn execute(
+        &self,
+        py: Python,
+        sql: &str,
+        params: Option<&Bound<'_, PyDict>>,
+        timeout: Option<f64>,
+    ) -> PyResult<Py<PyAny>> {
+        let deadline = QueryDeadline::new(timeout)?;
+        deadline.check()?;
+        let params = python_params(params)?;
         let result_batch = RT
-            .block_on(Some(py), query.execute_with_context(context))?
+            .block_on_until(Some(py), deadline, self.inner.execute(sql, params))?
             .map_err(graph_error_to_pyerr)?;
-
         record_batch_to_python_table(py, &result_batch)
     }
 
+    #[pyo3(signature = (sql, params = None))]
+    fn explain(
+        &self,
+        py: Python,
+        sql: &str,
+        params: Option<&Bound<'_, PyDict>>,
+    ) -> PyResult<String> {
+        let params = python_params(params)?;
+        RT.block_on(Some(py), self.inner.explain(sql, params))?
+            .map_err(graph_error_to_pyerr)
+    }
+
+    fn checkout_latest(&self, py: Python) -> PyResult<()> {
+        RT.block_on(Some(py), self.inner.checkout_latest())?
+            .map_err(graph_error_to_pyerr)
+    }
+
     fn __repr__(&self) -> String {
         "SqlEngine(...)".to_string()
     }

crates/lance-csr/README.md

@@ -5,16 +5,14 @@ with a CSR (compressed sparse row) adjacency index for fast multi-hop traversal.
 Arrow tables as property graphs. The [lance-csr Python package](https://pypi.org/project/lance-csr/) wraps this
 crate.
 
-> **Experimental.** The Rust API is not stable yet: breaking changes are possible over the next few
-> weeks while the CSR work lands. Pin an exact version (`lance-csr = "=x.y.z"`) if you depend on it.
-
 ## Features
 
 - Cypher query parsing and AST construction
 - Graph configuration for mapping Lance tables to nodes and relationships
 - Semantic validation with typed `GraphError` diagnostics
 - Pluggable execution strategies (DataFusion planner by default, Lance Native placeholder)
 - `CypherEngine`: register datasets once, optionally build the CSR, and serve many queries
+- `SqlEngine`: SQL over the same datasets, with Cypher patterns inside a `GRAPH_TABLE` clause
 - Lance datasets read lazily and pinned to a version until `checkout_latest()`
 - Async query execution that returns Arrow `RecordBatch` results or a stream
 - JSON-serializable parameter binding for reusable query templates
@@ -154,7 +152,8 @@ clauses, patterns, expressions and functions the engine runs.
 - `logical_plan` – Builders for graph logical plans.
 - `datafusion_planner` – DataFusion-based execution planning.
 - `config` – Graph configuration types and builders.
-- `engine` – `CypherEngine`: cached snapshot, CSR adjacencies, `checkout_latest()`.
+- `engine` – `CypherEngine` and `SqlEngine`: cached snapshot, CSR adjacencies, `checkout_latest()`.
+- `graph_table` – The `GRAPH_TABLE` clause and the `graph_table` table function.
 - `query` – High level `CypherQuery` API and runtime.
 - `csr_index`, `csr_expand` – The CSR adjacency index and the traversal operators that use it.
 - `error` – `GraphError` and result helpers.