diff --git a/.bumpversion.toml b/.bumpversion.toml index d712ca2..dde9a9b 100644 --- a/.bumpversion.toml +++ b/.bumpversion.toml @@ -33,11 +33,6 @@ filename = "crates/lance-csr/Cargo.toml" search = 'version = "{current_version}"' replace = 'version = "{new_version}"' -[[tool.bumpversion.files]] -filename = "crates/lance-csr-catalog/Cargo.toml" -search = 'version = "{current_version}"' -replace = 'version = "{new_version}"' - [[tool.bumpversion.files]] filename = "crates/lance-csr-benches/Cargo.toml" search = 'version = "{current_version}"' diff --git a/.github/workflows/python-publish.yml b/.github/workflows/python-publish.yml index bf58610..1eb73cf 100644 --- a/.github/workflows/python-publish.yml +++ b/.github/workflows/python-publish.yml @@ -217,8 +217,8 @@ jobs: publish-crates: # After PyPI, so the two registries never disagree. If only this job fails, rerun it on - # its own; the wheels don't need rebuilding. The crates.io trusted publishers for both - # crates are registered for this workflow and environment `crates-io`. + # its own; the wheels don't need rebuilding. The crates.io trusted publisher for the + # crate is registered for this workflow and environment `crates-io`. needs: publish runs-on: ubuntu-24.04 timeout-minutes: 60 @@ -237,14 +237,13 @@ jobs: sudo apt-get install -y protobuf-compiler # Verifying compiles lance, so do it before getting the token, which lasts 30 minutes. - # Cargo packages lance-csr-catalog first, since lance-csr depends on it. - - name: Package and verify the crates - run: cargo package --locked -p lance-csr-catalog -p lance-csr + - name: Package and verify the crate + run: cargo package --locked -p lance-csr - uses: rust-lang/crates-io-auth-action@v1 id: auth - name: Publish to crates.io - run: cargo publish --locked --no-verify -p lance-csr-catalog -p lance-csr + run: cargo publish --locked --no-verify -p lance-csr env: CARGO_REGISTRY_TOKEN: ${{ steps.auth.outputs.token }} diff --git a/.github/workflows/python-test.yml b/.github/workflows/python-test.yml index f3c0a51..a503760 100644 --- a/.github/workflows/python-test.yml +++ b/.github/workflows/python-test.yml @@ -46,8 +46,7 @@ jobs: shared-key: "lance-csr-python" # maturin builds the bindings crate, which sits outside the root # workspace. The root workspace is listed too so the cache key hashes - # the manifests of lance-csr and lance-csr-catalog, which the bindings - # use by path; otherwise a dependency or feature change there keeps + # the manifest of lance-csr, which the bindings use by path; otherwise a dependency or feature change there keeps # restoring a stale cache. Its target directory is never built here. workspaces: | . -> target diff --git a/AGENTS.md b/AGENTS.md index 89a533c..f508dad 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -2,7 +2,6 @@ ## Project Structure & Module Organization - `crates/lance-csr/` hosts the Rust Cypher engine; keep new modules under `src/` and co-locate helpers inside `query/` or feature-specific submodules. -- `crates/lance-csr-catalog/` provides the graph source catalog and `DirNamespace` (Unity Catalog and Delta Lake support were removed). - `crates/lance-csr-benches/` contains performance benchmarks in a dedicated unpublished crate; add new benchmarks here rather than in the main crate. - `crates/lance-csr-python/src/` contains the PyO3 bridge; `python/python/lance_csr/` holds the pure-Python facade and packaging metadata. - `python/python/tests/` stores functional tests; mirror new features with targeted cases here and in the corresponding Rust module. diff --git a/Cargo.lock b/Cargo.lock index ddc6dd0..f6ac25b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3962,7 +3962,6 @@ dependencies = [ "futures", "lance", "lance-arrow", - "lance-csr-catalog", "lance-linalg", "lance-namespace", "lance-table", @@ -3991,20 +3990,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "lance-csr-catalog" -version = "0.7.1" -dependencies = [ - "arrow-schema", - "async-trait", - "datafusion", - "lance-namespace", - "serde", - "serde_json", - "snafu 0.8.9", - "tokio", -] - [[package]] name = "lance-datafusion" version = "13.0.0" diff --git a/Cargo.toml b/Cargo.toml index 72a5d29..64b54d5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,7 +1,6 @@ [workspace] members = [ "crates/lance-csr", - "crates/lance-csr-catalog", "crates/lance-csr-benches", ] # Python package needs to be built by maturin. diff --git a/RELEASE.md b/RELEASE.md index 96aab8a..7d0990d 100644 --- a/RELEASE.md +++ b/RELEASE.md @@ -1,7 +1,7 @@ # Releasing lance-csr A release publishes the Python package to [PyPI](https://pypi.org/project/lance-csr/) and the -`lance-csr-catalog` and `lance-csr` crates to [crates.io](https://crates.io/crates/lance-csr). PyPI publishing is +`lance-csr` crate to [crates.io](https://crates.io/crates/lance-csr). PyPI publishing is automated from a GitHub release, and the crates follow once PyPI has the new version. The whole release takes about 50 minutes: the macOS wheel build, then the crates. Don't add steps that build the @@ -26,9 +26,9 @@ cargo update --workspace --manifest-path crates/lance-csr-python/Cargo.toml (cd python && uv lock) ``` -`bump-my-version` updates `.bumpversion.toml`, the four crate `Cargo.toml` files and `python/pyproject.toml`. It +`bump-my-version` updates `.bumpversion.toml`, the three crate `Cargo.toml` files and `python/pyproject.toml`. It doesn't update the three lockfiles that also carry the version, which is what the other three commands are for. -The diff should be 9 files and 15 lines each way, touching only version lines. If `uv lock` or `cargo update` +The diff should be 8 files and 11 lines each way, touching only version lines. If `uv lock` or `cargo update` moves a dependency, undo that and update only the workspace packages. Add `--allow-dirty` to `bump-my-version` if untracked files in your checkout stop it. @@ -68,12 +68,12 @@ Publishing it starts `.github/workflows/python-publish.yml`: | `build-macos` | about 35 min | macOS arm64 wheel, built natively on `macos-15` | | `test` | under 1 min each | each wheel on Python 3.12, 3.13 and 3.14 | | `publish` | seconds | checks the tag matches `python/pyproject.toml`, then uploads through PyPI trusted publishing | -| `publish-crates` | about 10 to 15 min | after `publish`: packages and verifies `lance-csr-catalog` and `lance-csr`, then uploads them through crates.io trusted publishing | +| `publish-crates` | about 10 to 15 min | after `publish`: packages and verifies `lance-csr`, then uploads it through crates.io trusted publishing | The build jobs run in parallel. Nothing reaches PyPI unless every build and test job passes, and nothing reaches crates.io unless PyPI publishing succeeds. The `pypi` and `crates-io` environments only accept `v*` tags. The PyPI trusted publisher is registered for workflow `python-publish.yml` and environment `pypi`, and -the crates.io trusted publishers for both crates for the same workflow and environment `crates-io`, so don't +the crates.io trusted publisher for `lance-csr` for the same workflow and environment `crates-io`, so don't rename any of them. **Don't run a separate `mode=dry_run` before a release.** The release run builds and tests the same wheels @@ -94,11 +94,8 @@ If a build or test job failed, nothing has been published, so nothing needs undo PyPI already has the release, so don't release again. Rerun just that job with `gh run rerun --failed`; the wheels aren't rebuilt. -Cargo publishes `lance-csr-catalog` first, since `lance-csr` depends on it. If it published the catalog and -then failed on `lance-csr`, a rerun fails because the catalog version already exists. Publish the rest by hand -from the release tag: a crate owner runs `cargo publish -p lance-csr`. If **Require trusted publishing** is on -for the crate, crates.io rejects API tokens, so turn it off for the manual publish and back on afterwards. -`lance-csr-python` and `lance-csr-benches` are `publish = false`. +`lance-csr-python` and `lance-csr-benches` are `publish = false`. Up to 0.7.1, the catalog types shipped +in a separate `lance-csr-catalog` crate; it is no longer published. ## 4. Check the release diff --git a/crates/lance-csr-catalog/Cargo.toml b/crates/lance-csr-catalog/Cargo.toml deleted file mode 100644 index 3c0d758..0000000 --- a/crates/lance-csr-catalog/Cargo.toml +++ /dev/null @@ -1,24 +0,0 @@ -[package] -name = "lance-csr-catalog" -version = "0.7.1" -edition = "2021" -license = "Apache-2.0" -authors = ["The lance-csr Authors"] -repository = "https://github.com/hyperdimensionalcomputingai/lance-csr" -homepage = "https://lance-csr.hdclabs.ai/" -readme = "README.md" -description = "Table catalog and directory namespace for the lance-csr graph query engine" -keywords = ["lance", "graph", "catalog", "namespace"] -categories = ["database", "data-structures", "science"] - -[dependencies] -arrow-schema = "58" -async-trait = "0.1" -datafusion = { version = "54", default-features = false } -lance-namespace = "13.0.0" -serde = { version = "1", features = ["derive"] } -serde_json = "1" -snafu = "0.8" - -[dev-dependencies] -tokio = { version = "1.37", features = ["macros", "rt-multi-thread"] } diff --git a/crates/lance-csr-catalog/LICENSE b/crates/lance-csr-catalog/LICENSE deleted file mode 120000 index 30cff74..0000000 --- a/crates/lance-csr-catalog/LICENSE +++ /dev/null @@ -1 +0,0 @@ -../../LICENSE \ No newline at end of file diff --git a/crates/lance-csr-catalog/NOTICE b/crates/lance-csr-catalog/NOTICE deleted file mode 120000 index fb376cf..0000000 --- a/crates/lance-csr-catalog/NOTICE +++ /dev/null @@ -1 +0,0 @@ -../../NOTICE \ No newline at end of file diff --git a/crates/lance-csr-catalog/README.md b/crates/lance-csr-catalog/README.md deleted file mode 100644 index ae7d607..0000000 --- a/crates/lance-csr-catalog/README.md +++ /dev/null @@ -1,34 +0,0 @@ -# lance-csr-catalog - -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. - -- `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`. diff --git a/crates/lance-csr-catalog/src/lib.rs b/crates/lance-csr-catalog/src/lib.rs deleted file mode 100644 index d543687..0000000 --- a/crates/lance-csr-catalog/src/lib.rs +++ /dev/null @@ -1,38 +0,0 @@ -// SPDX-License-Identifier: Apache-2.0 -// SPDX-FileCopyrightText: Copyright The Lance Authors - -//! The table lookups behind [lance-csr](https://docs.rs/lance-csr), a Cypher graph query -//! engine for 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: -//! -//! - [`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; - -// Existing exports -pub use namespace::DirNamespace; -pub use source_catalog::{GraphSourceCatalog, InMemoryCatalog, SimpleTableSource}; diff --git a/crates/lance-csr-catalog/src/namespace/mod.rs b/crates/lance-csr-catalog/src/namespace/mod.rs deleted file mode 100644 index fe6c89c..0000000 --- a/crates/lance-csr-catalog/src/namespace/mod.rs +++ /dev/null @@ -1,3 +0,0 @@ -pub mod directory; - -pub use directory::DirNamespace; diff --git a/crates/lance-csr-python/Cargo.lock b/crates/lance-csr-python/Cargo.lock index 9703c3b..aa69b03 100644 --- a/crates/lance-csr-python/Cargo.lock +++ b/crates/lance-csr-python/Cargo.lock @@ -3846,7 +3846,6 @@ dependencies = [ "datafusion-sql", "futures", "lance", - "lance-csr-catalog", "lance-linalg", "lance-namespace", "lance-table", @@ -3859,19 +3858,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "lance-csr-catalog" -version = "0.7.1" -dependencies = [ - "arrow-schema", - "async-trait", - "datafusion", - "lance-namespace", - "serde", - "serde_json", - "snafu 0.8.9", -] - [[package]] name = "lance-csr-python" version = "0.7.1" diff --git a/crates/lance-csr/Cargo.toml b/crates/lance-csr/Cargo.toml index 2740757..8347849 100644 --- a/crates/lance-csr/Cargo.toml +++ b/crates/lance-csr/Cargo.toml @@ -33,7 +33,6 @@ datafusion-expr = "54" datafusion-sql = "54" datafusion-functions-aggregate = "54" futures = "0.3" -lance-csr-catalog = { path = "../lance-csr-catalog", version = "0.7.1" } lance = "13.0.0" lance-linalg = "13.0.0" lance-namespace = "13.0.0" diff --git a/crates/lance-csr/README.md b/crates/lance-csr/README.md index 633db59..23753e1 100644 --- a/crates/lance-csr/README.md +++ b/crates/lance-csr/README.md @@ -157,12 +157,13 @@ clauses, patterns, expressions and functions the engine runs. - `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. -- `namespace` – Namespace helpers (re-exported from `lance-csr-catalog`). -- `source_catalog` – Catalog helpers for looking up table metadata (re-exported from `lance-csr-catalog`). +- `mapping` – `GraphMapping`, the JSON graph descriptor that maps labels to tables, and + `GraphSnapshot`, which pins each table to a Lance version and opens them for `CypherEngine`. +- `namespace` – `DirNamespace`, a Lance namespace for a directory of datasets. +- `source_catalog` – `GraphSourceCatalog`, which resolves labels to DataFusion table sources. -`lance-csr` re-exports the catalog and namespace types from the `lance-csr-catalog` crate for -API compatibility. You can depend on `lance-csr-catalog` directly if you only need catalog or -namespace utilities. +Up to 0.7.1 the `namespace` and `source_catalog` types lived in a separate `lance-csr-catalog` crate. +They now live here, at the same root paths (`lance_csr::DirNamespace` and so on). ## Error Handling diff --git a/crates/lance-csr/src/apply.rs b/crates/lance-csr/src/apply.rs index 274dc7a..8a46439 100644 --- a/crates/lance-csr/src/apply.rs +++ b/crates/lance-csr/src/apply.rs @@ -506,7 +506,7 @@ mod tests { #[test] fn call_bypasses_snapshot_templates_in_every_reading_stage() { let (config, _) = graph(); - let catalog = Arc::new(lance_csr_catalog::InMemoryCatalog::new()); + let catalog = Arc::new(crate::InMemoryCatalog::new()); for text in [ "CALL { RETURN 1 AS n } RETURN n", "WITH 1 AS x CALL { WITH x RETURN x AS n } RETURN n", diff --git a/crates/lance-csr/src/datafusion_planner/config_helpers.rs b/crates/lance-csr/src/datafusion_planner/config_helpers.rs index 9f4fcfb..99e8b4c 100644 --- a/crates/lance-csr/src/datafusion_planner/config_helpers.rs +++ b/crates/lance-csr/src/datafusion_planner/config_helpers.rs @@ -10,7 +10,7 @@ use super::analysis::PlanningContext; use super::DataFusionPlanner; use crate::config::{NodeMapping, RelationshipMapping}; use crate::error::Result; -use lance_csr_catalog::GraphSourceCatalog; +use crate::GraphSourceCatalog; use std::sync::Arc; impl DataFusionPlanner { diff --git a/crates/lance-csr/src/datafusion_planner/join_ops.rs b/crates/lance-csr/src/datafusion_planner/join_ops.rs index 790b586..930b60c 100644 --- a/crates/lance-csr/src/datafusion_planner/join_ops.rs +++ b/crates/lance-csr/src/datafusion_planner/join_ops.rs @@ -14,11 +14,11 @@ use crate::ast::{PropertyValue, RelationshipDirection}; use crate::case_insensitive::qualify_column; use crate::config::{NodeMapping, RelationshipMapping}; use crate::error::Result; +use crate::GraphSourceCatalog; use datafusion::common::Column; use datafusion::logical_expr::{ col, BinaryExpr, Expr, JoinType, LogicalPlan, LogicalPlanBuilder, Operator, TableSource, }; -use lance_csr_catalog::GraphSourceCatalog; use std::collections::HashMap; use std::sync::Arc; diff --git a/crates/lance-csr/src/datafusion_planner/mod.rs b/crates/lance-csr/src/datafusion_planner/mod.rs index 7864ab7..bfbb510 100644 --- a/crates/lance-csr/src/datafusion_planner/mod.rs +++ b/crates/lance-csr/src/datafusion_planner/mod.rs @@ -37,8 +37,8 @@ pub(crate) use scalar_functions::sql_expr; use crate::config::GraphConfig; use crate::error::Result; use crate::logical_plan::LogicalOperator; +use crate::GraphSourceCatalog; use datafusion::logical_expr::LogicalPlan; -use lance_csr_catalog::GraphSourceCatalog; use std::sync::Arc; /// Planner abstraction for graph-to-physical planning diff --git a/crates/lance-csr/src/datafusion_planner/scan_ops.rs b/crates/lance-csr/src/datafusion_planner/scan_ops.rs index ef5cf1b..6e54651 100644 --- a/crates/lance-csr/src/datafusion_planner/scan_ops.rs +++ b/crates/lance-csr/src/datafusion_planner/scan_ops.rs @@ -118,7 +118,7 @@ impl DataFusionPlanner { // No catalog attached - create empty source fallback for flexibility // This allows planners created with DataFusionPlanner::new() to work // without requiring a catalog, though they won't have actual data sources - let empty_source = Arc::new(lance_csr_catalog::SimpleTableSource::empty()); + let empty_source = Arc::new(crate::SimpleTableSource::empty()); let normalized_label = label.to_lowercase(); let builder = LogicalPlanBuilder::scan(&normalized_label, empty_source, None).map_err(|e| { diff --git a/crates/lance-csr/src/datafusion_planner/test_fixtures.rs b/crates/lance-csr/src/datafusion_planner/test_fixtures.rs index 90bda75..97a063b 100644 --- a/crates/lance-csr/src/datafusion_planner/test_fixtures.rs +++ b/crates/lance-csr/src/datafusion_planner/test_fixtures.rs @@ -5,8 +5,8 @@ use crate::config::GraphConfig; use crate::logical_plan::LogicalOperator; +use crate::{InMemoryCatalog, SimpleTableSource}; use arrow_schema::{DataType, Field, Schema}; -use lance_csr_catalog::{InMemoryCatalog, SimpleTableSource}; use std::sync::Arc; pub fn person_schema() -> Arc { @@ -17,7 +17,7 @@ pub fn person_schema() -> Arc { ])) } -pub fn make_catalog() -> Arc { +pub fn make_catalog() -> Arc { let person_src = Arc::new(SimpleTableSource::new(person_schema())); let knows_schema = Arc::new(Schema::new(vec![ Field::new("src_person_id", DataType::Int64, false), diff --git a/crates/lance-csr/src/engine.rs b/crates/lance-csr/src/engine.rs index 1638f03..514e5c1 100644 --- a/crates/lance-csr/src/engine.rs +++ b/crates/lance-csr/src/engine.rs @@ -511,8 +511,9 @@ fn config_error(message: String) -> GraphError { } } -/// Open a Lance dataset. `version = None` means the latest version. -async fn open_lance_dataset( +/// Open the Lance dataset at `uri`, pinned to `version`, or at its latest version when +/// `version` is `None`. Storage options go to the object store, as in Lance itself. +pub async fn open_lance_dataset( uri: &str, version: Option, storage_options: &HashMap, diff --git a/crates/lance-csr/src/lib.rs b/crates/lance-csr/src/lib.rs index 8a72708..d54bc3c 100644 --- a/crates/lance-csr/src/lib.rs +++ b/crates/lance-csr/src/lib.rs @@ -58,6 +58,8 @@ pub mod graph_table; pub mod key_statistics; pub mod lance_vector_search; pub mod logical_plan; +pub mod mapping; +pub mod namespace; pub mod parameter_substitution; pub mod parser; mod path_inventory; @@ -68,6 +70,7 @@ mod pattern_order; pub mod query; pub mod semantic; pub mod shortest_path; +pub mod source_catalog; pub mod spark_dialect; pub mod sql_query; @@ -76,9 +79,11 @@ pub const MAX_VARIABLE_LENGTH_HOPS: u32 = 20; pub use config::{GraphConfig, NodeMapping, RelationshipMapping}; pub use csr_index::{CsrIndex, CsrIndexBuilder}; -pub use engine::{CypherEngine, EngineSource, SqlEngine}; +pub use engine::{open_lance_dataset, CypherEngine, EngineSource, SqlEngine}; pub use error::{GraphError, Result}; -pub use lance_csr_catalog::{DirNamespace, GraphSourceCatalog, InMemoryCatalog, SimpleTableSource}; pub use lance_vector_search::VectorSearch; +pub use mapping::{EdgeTableMapping, GraphMapping, GraphSnapshot, NodeTableMapping, PinnedTable}; +pub use namespace::DirNamespace; pub use query::{CypherQuery, ExecutionStrategy, SqlDialect}; +pub use source_catalog::{GraphSourceCatalog, InMemoryCatalog, SimpleTableSource}; pub use sql_query::SqlQuery; diff --git a/crates/lance-csr/src/mapping.rs b/crates/lance-csr/src/mapping.rs new file mode 100644 index 0000000..e91668f --- /dev/null +++ b/crates/lance-csr/src/mapping.rs @@ -0,0 +1,517 @@ +// SPDX-License-Identifier: Apache-2.0 + +//! A graph mapping says which table holds each node label and relationship type, and a +//! [`GraphSnapshot`] pins each of those tables to one Lance version. +//! +//! Applications that store graphs as plain Lance tables, one per label, share these types +//! instead of keeping their own copies. Their JSON is the version 1 graph descriptor: +//! +//! ```json +//! { +//! "version": 1, +//! "nodes": [{"label": "Person", "table": "people", "key": "id", "display": "name"}], +//! "edges": [{"label": "KNOWS", "table": "knows", "key": "edge_id", "source": "src", +//! "target": "dst", "sourceType": "Person", "targetType": "Person"}] +//! } +//! ``` +//! +//! A snapshot adds `tables`, each a `name`, `path` and `version`. It ignores other fields, so +//! it reads a publication manifest that carries row counts or schemas as well. +//! +//! ```no_run +//! use std::collections::HashMap; +//! use lance_csr::{CypherEngine, GraphSnapshot, Result}; +//! +//! # async fn example(json: &str) -> Result<()> { +//! let snapshot: GraphSnapshot = serde_json::from_str(json).expect("valid snapshot"); +//! let (config, sources) = snapshot.open(&HashMap::new()).await?; +//! let engine = CypherEngine::new(config, sources, true).await?; +//! # Ok(()) +//! # } +//! ``` + +use std::collections::{HashMap, HashSet}; + +use arrow_schema::{DataType, Schema, SchemaRef}; +use serde::{Deserialize, Serialize}; + +use crate::config::{GraphConfig, NodeMapping, RelationshipMapping}; +use crate::engine::{open_lance_dataset, EngineSource}; +use crate::error::{GraphError, Result}; + +/// The only mapping format version so far. +pub const GRAPH_MAPPING_VERSION: u32 = 1; +/// The most node labels and relationship types one mapping may hold. +pub const MAX_MAPPED_TYPES: usize = 64; +/// The longest label, in bytes. +pub const MAX_LABEL_LEN: usize = 80; + +/// A node label backed by one table, with `key` as its unique id column. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct NodeTableMapping { + pub label: String, + pub table: String, + pub key: String, + /// The column applications show as the node's name. + pub display: Option, +} + +/// A relationship type backed by one table. `source` and `target` hold keys of the +/// `source_type` and `target_type` node labels. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct EdgeTableMapping { + pub label: String, + pub table: String, + pub key: String, + pub source: String, + pub target: String, + pub source_type: String, + pub target_type: String, +} + +/// Which table holds each node label and relationship type. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct GraphMapping { + pub version: u32, + pub nodes: Vec, + pub edges: Vec, +} + +/// One table at one Lance version. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct PinnedTable { + pub name: String, + /// The dataset URI. Resolve a relative path against its root before calling + /// [`GraphSnapshot::open`]. + pub path: String, + pub version: u64, +} + +/// A set of pinned tables and the mapping that reads them as a graph. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct GraphSnapshot { + pub tables: Vec, + /// `None` while the tables have no graph mapping yet. + #[serde(default)] + pub mapping: Option, +} + +fn invalid(message: impl Into) -> GraphError { + GraphError::ConfigError { + message: message.into(), + location: snafu::Location::new(file!(), line!(), column!()), + } +} + +fn check_label(label: &str) -> Result<()> { + let valid = !label.is_empty() + && label.len() <= MAX_LABEL_LEN + && label + .chars() + .enumerate() + .all(|(i, c)| c == '_' || c.is_ascii_alphabetic() || (i > 0 && c.is_ascii_digit())); + if valid { + Ok(()) + } else { + Err(invalid(format!( + "Graph labels must be ASCII identifiers of at most {MAX_LABEL_LEN} characters: '{label}'" + ))) + } +} + +impl GraphMapping { + /// Check the rules that need no table data: a supported version, at least one node label, + /// at most [`MAX_MAPPED_TYPES`] labels, labels that are ASCII identifiers and unique + /// ignoring case, one graph role per table, and relationship endpoints that name a mapped + /// node label. + pub fn validate(&self) -> Result<()> { + if self.version != GRAPH_MAPPING_VERSION { + return Err(invalid(format!( + "Unsupported graph mapping version {}; expected {GRAPH_MAPPING_VERSION}", + self.version + ))); + } + if self.nodes.is_empty() { + return Err(invalid("Map at least one node table")); + } + if self.nodes.len() + self.edges.len() > MAX_MAPPED_TYPES { + return Err(invalid(format!( + "At most {MAX_MAPPED_TYPES} node labels and relationship types are supported" + ))); + } + let mut labels = HashSet::new(); + let mut tables = HashSet::new(); + let roles = self + .nodes + .iter() + .map(|n| (&n.label, &n.table)) + .chain(self.edges.iter().map(|e| (&e.label, &e.table))); + for (label, table) in roles { + check_label(label)?; + if !labels.insert(label.to_lowercase()) { + return Err(invalid(format!( + "Duplicate graph label, ignoring case: '{label}'" + ))); + } + if !tables.insert(table) { + return Err(invalid(format!( + "Table '{table}' has more than one graph role" + ))); + } + } + for edge in &self.edges { + for end in [&edge.source_type, &edge.target_type] { + if self.node(end).is_none() { + return Err(invalid(format!( + "Relationship '{}' connects '{end}', which is not a mapped node label", + edge.label + ))); + } + } + } + Ok(()) + } + + fn node(&self, label: &str) -> Option<&NodeTableMapping> { + self.nodes + .iter() + .find(|n| n.label.eq_ignore_ascii_case(label)) + } + + /// Build the [`GraphConfig`] for this mapping, given the Arrow schema of each mapped table, + /// keyed by table name. + /// + /// Besides [`Self::validate`], this checks each mapped table: its columns don't collide + /// ignoring case, the key, display and endpoint columns exist, keys are scalar, and each + /// endpoint column has the type of the node key it refers to. Node properties are every + /// column but the key; relationship properties are every column. + pub fn graph_config(&self, schemas: &HashMap) -> Result { + self.validate()?; + let schema = |table: &str| -> Result<&Schema> { + let schema = schemas + .get(table) + .ok_or_else(|| invalid(format!("Mapped table '{table}' has no schema")))?; + let mut names = HashSet::new(); + if !schema + .fields() + .iter() + .all(|f| names.insert(f.name().to_lowercase())) + { + return Err(invalid(format!( + "Table '{table}' has columns whose names differ only in case" + ))); + } + Ok(schema) + }; + let column = |schema: &Schema, table: &str, name: &str| -> Result { + schema + .field_with_name(name) + .map(|f| f.data_type().clone()) + .map_err(|_| invalid(format!("Table '{table}' has no column '{name}'"))) + }; + let key = |schema: &Schema, table: &str, name: &str| -> Result { + let data_type = column(schema, table, name)?; + if data_type.is_nested() || data_type == DataType::Null { + return Err(invalid(format!( + "Key column '{table}.{name}' must have a scalar type, not {data_type}" + ))); + } + Ok(data_type) + }; + let columns = |schema: &Schema, skip: Option<&str>| -> Vec { + schema + .fields() + .iter() + .map(|f| f.name().clone()) + .filter(|name| Some(name.as_str()) != skip) + .collect() + }; + + let mut builder = GraphConfig::builder(); + for node in &self.nodes { + let s = schema(&node.table)?; + key(s, &node.table, &node.key)?; + if let Some(display) = &node.display { + column(s, &node.table, display)?; + } + builder = builder.with_node_mapping( + NodeMapping::new(node.label.as_str(), node.key.as_str()) + .with_properties(columns(s, Some(&node.key))), + ); + } + for edge in &self.edges { + let s = schema(&edge.table)?; + key(s, &edge.table, &edge.key)?; + for (end, end_type) in [ + (&edge.source, &edge.source_type), + (&edge.target, &edge.target_type), + ] { + let node = self.node(end_type).expect("validated endpoint"); + let expected = key(schema(&node.table)?, &node.table, &node.key)?; + let actual = key(s, &edge.table, end)?; + if actual != expected { + return Err(invalid(format!( + "Endpoint '{}.{end}' is {actual}, but node key '{}.{}' is {expected}", + edge.table, node.table, node.key + ))); + } + } + builder = builder.with_relationship_mapping( + RelationshipMapping::new( + edge.label.as_str(), + edge.source.as_str(), + edge.target.as_str(), + ) + .with_endpoints(edge.source_type.as_str(), edge.target_type.as_str()) + .with_properties(columns(s, None)), + ); + } + builder.build() + } +} + +impl GraphSnapshot { + /// Open every mapped table at its pinned version and build the graph configuration. + /// + /// Returns what [`crate::CypherEngine::new`] takes: the configuration and one source per + /// label, keyed by the lowercased label. Tables the mapping doesn't use stay closed. + pub async fn open( + &self, + storage_options: &HashMap, + ) -> Result<(GraphConfig, HashMap)> { + let mapping = self + .mapping + .as_ref() + .ok_or_else(|| invalid("The snapshot has no graph mapping"))?; + mapping.validate()?; + let roles = mapping + .nodes + .iter() + .map(|n| (&n.label, &n.table)) + .chain(mapping.edges.iter().map(|e| (&e.label, &e.table))); + let mut schemas = HashMap::new(); + let mut sources = HashMap::new(); + for (label, table) in roles { + let pinned = self + .tables + .iter() + .find(|t| &t.name == table) + .ok_or_else(|| invalid(format!("Mapped table '{table}' is not in the snapshot")))?; + let dataset = + open_lance_dataset(&pinned.path, Some(pinned.version), storage_options).await?; + schemas.insert(table.clone(), SchemaRef::new(dataset.schema().into())); + sources.insert( + label.to_lowercase(), + EngineSource::Lance { + uri: pinned.path.clone(), + storage_options: storage_options.clone(), + dataset: std::sync::Arc::new(dataset), + }, + ); + } + Ok((mapping.graph_config(&schemas)?, sources)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use arrow_schema::Field; + + const DESCRIPTOR: &str = r#"{ + "version": 1, + "nodes": [{"label": "Person", "table": "people", "key": "id", "display": "name"}], + "edges": [{"label": "KNOWS", "table": "knows", "key": "edge_id", "source": "src", + "target": "dst", "sourceType": "Person", "targetType": "Person"}] + }"#; + + fn mapping() -> GraphMapping { + serde_json::from_str(DESCRIPTOR).unwrap() + } + + fn schemas(endpoint: DataType) -> HashMap { + let people = Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new("name", DataType::Utf8, true), + ]); + let knows = Schema::new(vec![ + Field::new("edge_id", DataType::Int64, false), + Field::new("src", endpoint.clone(), false), + Field::new("dst", endpoint, false), + ]); + HashMap::from([ + ("people".to_string(), SchemaRef::new(people)), + ("knows".to_string(), SchemaRef::new(knows)), + ]) + } + + fn error(result: Result) -> String { + result.err().expect("an error").to_string() + } + + #[test] + fn reads_the_descriptor_and_round_trips() { + let m = mapping(); + m.validate().unwrap(); + assert_eq!(m.nodes[0].display.as_deref(), Some("name")); + let back: GraphMapping = serde_json::from_str(&serde_json::to_string(&m).unwrap()).unwrap(); + assert_eq!(back, m); + // An unknown field is a different format, not a typo to skip. + let extra = DESCRIPTOR.replacen("\"version\"", "\"extra\": 1, \"version\"", 1); + assert!(serde_json::from_str::(&extra).is_err()); + } + + #[test] + fn snapshot_reads_a_publication_manifest() { + let manifest = format!( + r#"{{"format": "example-publication", "version": 1, "name": "demo", + "tables": [{{"name": "people", "path": "data/people.lance", "version": 3, + "rows": 4, "fields": []}}], + "mapping": {DESCRIPTOR}, "sample": null}}"# + ); + let snapshot: GraphSnapshot = serde_json::from_str(&manifest).unwrap(); + assert_eq!(snapshot.tables[0].version, 3); + assert_eq!(snapshot.mapping, Some(mapping())); + let unmapped: GraphSnapshot = serde_json::from_str(r#"{"tables": []}"#).unwrap(); + assert!(unmapped.mapping.is_none()); + } + + #[test] + fn validate_rejects_structural_errors() { + let mut m = mapping(); + m.version = 2; + assert!(error(m.validate()).contains("version")); + + let mut m = mapping(); + m.nodes.clear(); + assert!(error(m.validate()).contains("at least one node")); + + let mut m = mapping(); + m.edges[0].label = "person".into(); + assert!(error(m.validate()).contains("Duplicate graph label")); + + let mut m = mapping(); + m.edges[0].table = "people".into(); + assert!(error(m.validate()).contains("more than one graph role")); + + let mut m = mapping(); + m.nodes[0].label = "Pérson".into(); + assert!(error(m.validate()).contains("ASCII identifiers")); + + let mut m = mapping(); + m.edges[0].target_type = "Company".into(); + assert!(error(m.validate()).contains("not a mapped node label")); + + let mut m = mapping(); + m.edges[0].source_type = "PERSON".into(); + m.validate().unwrap(); + } + + #[test] + fn graph_config_checks_columns_and_sets_properties() { + let config = mapping().graph_config(&schemas(DataType::Int64)).unwrap(); + let person = config.get_node_mapping("Person").unwrap(); + assert_eq!(person.property_fields, ["name"]); + let knows = config.get_relationship_mapping("KNOWS").unwrap(); + assert_eq!(knows.source_label.as_deref(), Some("Person")); + + assert!(error(mapping().graph_config(&schemas(DataType::Utf8))).contains("node key")); + + let mut m = mapping(); + m.nodes[0].display = Some("nickname".into()); + assert!(error(m.graph_config(&schemas(DataType::Int64))).contains("no column 'nickname'")); + + let mut s = schemas(DataType::Int64); + s.insert( + "people".into(), + SchemaRef::new(Schema::new(vec![ + Field::new("id", DataType::Int64, false), + Field::new("ID", DataType::Int64, false), + ])), + ); + assert!(error(mapping().graph_config(&s)).contains("differ only in case")); + } + + #[tokio::test] + async fn open_reads_the_pinned_version() { + use arrow_array::{Int64Array, RecordBatch, RecordBatchIterator, StringArray}; + use lance::dataset::{Dataset, WriteMode, WriteParams}; + + async fn write(path: &str, batch: RecordBatch, mode: WriteMode) { + let schema = batch.schema(); + let reader = RecordBatchIterator::new(vec![Ok(batch)], schema); + let params = WriteParams { + mode, + ..Default::default() + }; + Dataset::write(reader, path, Some(params)).await.unwrap(); + } + let s = schemas(DataType::Int64); + let people = |ids: Vec| { + let names: Vec = ids.iter().map(|i| format!("p{i}")).collect(); + RecordBatch::try_new( + s["people"].clone(), + vec![ + std::sync::Arc::new(Int64Array::from(ids)), + std::sync::Arc::new(StringArray::from(names)), + ], + ) + .unwrap() + }; + let knows = RecordBatch::try_new( + s["knows"].clone(), + vec![ + std::sync::Arc::new(Int64Array::from(vec![10])), + std::sync::Arc::new(Int64Array::from(vec![1])), + std::sync::Arc::new(Int64Array::from(vec![2])), + ], + ) + .unwrap(); + + let dir = tempfile::tempdir().unwrap(); + let path = |name: &str| dir.path().join(name).to_string_lossy().into_owned(); + write(&path("people.lance"), people(vec![1, 2]), WriteMode::Create).await; + write(&path("people.lance"), people(vec![3]), WriteMode::Append).await; + write(&path("knows.lance"), knows, WriteMode::Create).await; + + let snapshot = GraphSnapshot { + tables: vec![ + PinnedTable { + name: "people".into(), + path: path("people.lance"), + version: 1, + }, + PinnedTable { + name: "knows".into(), + path: path("knows.lance"), + version: 1, + }, + ], + mapping: Some(mapping()), + }; + let (config, sources) = snapshot.open(&HashMap::new()).await.unwrap(); + let engine = crate::CypherEngine::new(config, sources, true) + .await + .unwrap(); + let batch = engine + .execute("MATCH (p:Person) RETURN count(p) AS n", HashMap::new()) + .await + .unwrap(); + let n = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0); + assert_eq!(n, 2); + + let mut missing = snapshot.clone(); + missing.tables.pop(); + assert!(error(missing.open(&HashMap::new()).await).contains("not in the snapshot")); + } +} diff --git a/crates/lance-csr-catalog/src/namespace/directory.rs b/crates/lance-csr/src/namespace.rs similarity index 88% rename from crates/lance-csr-catalog/src/namespace/directory.rs rename to crates/lance-csr/src/namespace.rs index b905416..a3bb7a4 100644 --- a/crates/lance-csr-catalog/src/namespace/directory.rs +++ b/crates/lance-csr/src/namespace.rs @@ -1,5 +1,12 @@ // SPDX-License-Identifier: Apache-2.0 +//! [`DirNamespace`], a Lance namespace for a directory of datasets: the table `person` under +//! `s3://bucket/graph` is `s3://bucket/graph/person.lance`. +//! +//! [`crate::CypherQuery::execute_with_namespace`] accepts any `LanceNamespace`, so Lance's own +//! `DirectoryNamespace` and REST namespaces work as well. Those can also pin a table version and +//! supply storage options, which this one never does. + use async_trait::async_trait; use lance_namespace::models::{DescribeTableRequest, DescribeTableResponse}; use lance_namespace::{Error as NamespaceError, LanceNamespace, Result}; diff --git a/crates/lance-csr/src/query.rs b/crates/lance-csr/src/query.rs index af56063..f809887 100644 --- a/crates/lance-csr/src/query.rs +++ b/crates/lance-csr/src/query.rs @@ -20,7 +20,6 @@ use datafusion_sql::unparser::dialect::{ CustomDialect, DefaultDialect, MySqlDialect, PostgreSqlDialect, SqliteDialect, }; use datafusion_sql::unparser::Unparser; -use lance_csr_catalog::DirNamespace; use lance_namespace::models::DescribeTableRequest; use std::collections::{HashMap, HashSet}; use std::sync::Arc; @@ -248,7 +247,7 @@ pub struct CypherQuery { struct CachedPlan { graph: crate::logical_plan::LogicalOperator, datafusion: datafusion::logical_expr::LogicalPlan, - catalog: Arc, + catalog: Arc, parameter_types: HashMap, } @@ -346,7 +345,7 @@ impl CypherQuery { #[doc(hidden)] pub fn prepare_cached_plan( &mut self, - catalog: Arc, + catalog: Arc, ) -> Result { // Rich path properties live in an extension-owned spec, and its inventory is // query-local. Keep ordinary substitution and a fresh inventory for each query. @@ -463,12 +462,14 @@ impl CypherQuery { } } - /// Execute the query using a namespace-backed table resolver. + /// Execute the query, resolving each node label and relationship type to a table through a + /// Lance namespace, such as [`crate::DirNamespace`] or Lance's own `DirectoryNamespace`. /// - /// The namespace is provided by value and will be shared internally as needed. + /// Each table opens at the version and with the storage options the namespace's + /// `describe_table` returns, or at its latest version when the namespace returns none. pub async fn execute_with_namespace( &self, - namespace: DirNamespace, + namespace: impl lance_namespace::LanceNamespace + 'static, strategy: Option, ) -> Result { self.execute_with_namespace_arc(std::sync::Arc::new(namespace), strategy) @@ -478,18 +479,16 @@ impl CypherQuery { /// Execute the query using a shared namespace instance. pub async fn execute_with_namespace_arc( &self, - namespace: std::sync::Arc, + namespace: std::sync::Arc, strategy: Option, ) -> Result { - let namespace_trait: std::sync::Arc = - namespace; - self.execute_with_namespace_internal(namespace_trait, strategy) + self.execute_with_namespace_internal(namespace, strategy) .await } async fn execute_with_namespace_internal( &self, - namespace: std::sync::Arc, + namespace: std::sync::Arc, strategy: Option, ) -> Result { let strategy = strategy.unwrap_or_default(); @@ -729,8 +728,8 @@ impl CypherQuery { &self, ctx: datafusion::execution::context::SessionContext, ) -> Result { + use crate::InMemoryCatalog; use datafusion::datasource::DefaultTableSource; - use lance_csr_catalog::InMemoryCatalog; use std::sync::Arc; let config = self.require_config()?; @@ -815,7 +814,7 @@ impl CypherQuery { /// ``` pub async fn execute_with_catalog_and_context( &self, - catalog: std::sync::Arc, + catalog: std::sync::Arc, ctx: datafusion::execution::context::SessionContext, ) -> Result { let df = self @@ -831,7 +830,7 @@ impl CypherQuery { /// operators (scans and filters stream; joins, sorts and aggregates may buffer). pub async fn execute_stream_with_catalog_and_context( &self, - catalog: std::sync::Arc, + catalog: std::sync::Arc, ctx: datafusion::execution::context::SessionContext, ) -> Result { let df = self @@ -848,7 +847,7 @@ impl CypherQuery { /// Explain the plan with an explicit catalog and session context. pub async fn explain_with_catalog_and_context( &self, - catalog: std::sync::Arc, + catalog: std::sync::Arc, ctx: datafusion::execution::context::SessionContext, ) -> Result { self.explain_internal(catalog, ctx).await @@ -857,7 +856,7 @@ impl CypherQuery { /// Plan the query (phases 1-3) and hand it to DataFusion (phase 4) without executing. async fn dataframe_with_catalog_and_context( &self, - catalog: std::sync::Arc, + catalog: std::sync::Arc, ctx: datafusion::execution::context::SessionContext, ) -> Result { let (_logical_plan, df_logical_plan) = self.create_logical_plans(catalog)?; @@ -867,7 +866,7 @@ impl CypherQuery { /// The unoptimized DataFusion plan (phases 1-3), for embedding in a larger plan. pub(crate) fn datafusion_plan( &self, - catalog: std::sync::Arc, + catalog: std::sync::Arc, ) -> Result { Ok(self.create_logical_plans(catalog)?.1) } @@ -908,12 +907,12 @@ impl CypherQuery { &self, datasets: HashMap, ) -> Result<( - lance_csr_catalog::InMemoryCatalog, + crate::InMemoryCatalog, datafusion::execution::context::SessionContext, )> { + use crate::InMemoryCatalog; use datafusion::datasource::{DefaultTableSource, MemTable}; use datafusion::execution::context::SessionContext; - use lance_csr_catalog::InMemoryCatalog; use std::sync::Arc; if datasets.is_empty() { @@ -973,15 +972,15 @@ impl CypherQuery { /// Helper to build catalog and context using a namespace resolver async fn build_catalog_and_context_from_namespace( &self, - namespace: std::sync::Arc, + namespace: std::sync::Arc, ) -> Result<( - lance_csr_catalog::InMemoryCatalog, + crate::InMemoryCatalog, datafusion::execution::context::SessionContext, )> { + use crate::InMemoryCatalog; use datafusion::datasource::{DefaultTableSource, TableProvider}; use datafusion::execution::context::SessionContext; use lance::datafusion::LanceTableProvider; - use lance_csr_catalog::InMemoryCatalog; use std::sync::Arc; let config = self.require_config()?; @@ -1042,7 +1041,19 @@ impl CypherQuery { location: snafu::Location::new(file!(), line!(), column!()), })?; - let dataset = lance::dataset::Dataset::open(&location) + let version = response + .version + .map(u64::try_from) + .transpose() + .map_err(|_| GraphError::ConfigError { + message: format!( + "Namespace returned a negative version for table '{}'", + table_name + ), + location: snafu::Location::new(file!(), line!(), column!()), + })?; + let storage_options = response.storage_options.unwrap_or_default(); + let dataset = crate::engine::open_lance_dataset(&location, version, &storage_options) .await .map_err(|e| GraphError::ConfigError { message: format!("Failed to open dataset for table '{}': {}", table_name, e), @@ -1104,7 +1115,7 @@ impl CypherQuery { /// Internal helper to explain the query execution plan with explicit catalog and session context async fn explain_internal( &self, - catalog: std::sync::Arc, + catalog: std::sync::Arc, ctx: datafusion::execution::context::SessionContext, ) -> Result { // Create all plans (phases 1-4) @@ -1121,7 +1132,7 @@ impl CypherQuery { /// DataFusion logical planning) without creating the physical plan. fn create_logical_plans( &self, - catalog: std::sync::Arc, + catalog: std::sync::Arc, ) -> Result<( crate::logical_plan::LogicalOperator, datafusion::logical_expr::LogicalPlan, @@ -1151,7 +1162,7 @@ impl CypherQuery { fn build_logical_plans( &self, - catalog: Arc, + catalog: Arc, parameterized: bool, ) -> Result<( crate::logical_plan::LogicalOperator, @@ -1203,7 +1214,7 @@ impl CypherQuery { /// Helper to create all plans (graph logical, DataFusion logical, physical) async fn create_plans( &self, - catalog: std::sync::Arc, + catalog: std::sync::Arc, ctx: &datafusion::execution::context::SessionContext, ) -> Result<( crate::logical_plan::LogicalOperator, @@ -1689,11 +1700,11 @@ mod tests { #[tokio::test] async fn cartesian_stream_yields_and_drops_before_full_collection() { + use crate::InMemoryCatalog; use arrow_array::{ArrayRef, Int64Array}; use datafusion::datasource::{DefaultTableSource, MemTable}; use datafusion::execution::context::{SessionConfig, SessionContext}; use futures::TryStreamExt; - use lance_csr_catalog::InMemoryCatalog; use std::sync::atomic::{AtomicUsize, Ordering}; let batch = RecordBatch::try_from_iter([( @@ -1867,7 +1878,7 @@ mod tests { .build_catalog_and_context_from_datasets(datasets.clone()) .await .unwrap(); - let catalog: Arc = Arc::new(catalog); + let catalog: Arc = Arc::new(catalog); assert!( template.prepare_cached_plan(catalog.clone()).unwrap(), "{text}" @@ -2011,7 +2022,7 @@ mod tests { .build_catalog_and_context_from_datasets(data) .await .unwrap(); - let catalog: Arc = Arc::new(catalog); + let catalog: Arc = Arc::new(catalog); template.prepare_cached_plan(catalog.clone()).unwrap(); let state = context.state(); let mut totals = [std::time::Duration::ZERO; 8]; @@ -2950,7 +2961,7 @@ mod tests { .expect("query parses") .with_config(config); - let namespace = DirNamespace::new(tmp_dir.path().to_string_lossy().into_owned()); + let namespace = crate::DirNamespace::new(tmp_dir.path().to_string_lossy().into_owned()); let result = query .execute_with_namespace(namespace.clone(), None) @@ -2971,6 +2982,81 @@ mod tests { assert_eq!(values, vec!["Bob".to_string(), "David".to_string()]); } + /// Pins every table to version 1, as Lance's `DirectoryNamespace` does for a tagged table. + #[derive(Debug)] + struct PinnedNamespace(crate::DirNamespace); + + #[async_trait::async_trait] + impl lance_namespace::LanceNamespace for PinnedNamespace { + fn namespace_id(&self) -> String { + "pinned".to_string() + } + + async fn describe_table( + &self, + request: DescribeTableRequest, + ) -> lance_namespace::Result { + let mut response = self.0.describe_table(request).await?; + response.version = Some(1); + Ok(response) + } + } + + #[tokio::test] + async fn namespace_execution_opens_the_described_version() { + use arrow_array::{Array, RecordBatchIterator, StringArray}; + use lance::dataset::{Dataset, WriteMode, WriteParams}; + use tempfile::tempdir; + + let tmp_dir = tempdir().unwrap(); + let person = tmp_dir.path().join("Person.lance"); + write_lance_dataset(&person, build_people_batch()).await; + write_lance_dataset( + &tmp_dir.path().join("FRIEND_OF.lance"), + build_friendship_batch(), + ) + .await; + // Version 2 keeps only Alice, who is under 30. + let alice = build_people_batch().slice(0, 1); + let reader = RecordBatchIterator::new(vec![Ok(alice.clone())], alice.schema()); + let params = WriteParams { + mode: WriteMode::Overwrite, + ..Default::default() + }; + Dataset::write(reader, person.to_str().unwrap(), Some(params)) + .await + .unwrap(); + + let config = GraphConfig::builder() + .node_table("Person", "person_id") + .rel_table("FRIEND_OF", "Person", "Person", "person1_id", "person2_id") + .build() + .unwrap(); + let query = CypherQuery::new("MATCH (p:Person) WHERE p.age > 30 RETURN p.name") + .unwrap() + .with_config(config); + let root = tmp_dir.path().to_string_lossy().into_owned(); + + let latest = query + .execute_with_namespace(crate::DirNamespace::new(root.clone()), None) + .await + .unwrap(); + assert_eq!(latest.num_rows(), 0); + + let pinned = query + .execute_with_namespace(PinnedNamespace(crate::DirNamespace::new(root)), None) + .await + .unwrap(); + let names = pinned + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let mut values: Vec<&str> = (0..names.len()).map(|i| names.value(i)).collect(); + values.sort(); + assert_eq!(values, ["Bob", "David"]); + } + #[tokio::test] async fn test_execute_fails_on_semantic_error() { use arrow_array::RecordBatch; diff --git a/crates/lance-csr/src/semantic.rs b/crates/lance-csr/src/semantic.rs index 3d63f13..ebdd71d 100644 --- a/crates/lance-csr/src/semantic.rs +++ b/crates/lance-csr/src/semantic.rs @@ -25,7 +25,7 @@ pub struct SemanticAnalyzer { /// Variables whose scope a WITH ended; still type-checked at the end. retired: Vec, current_scope: ScopeType, - catalog: Option>, + catalog: Option>, expression_types: HashMap, call_body: bool, } @@ -84,10 +84,7 @@ impl SemanticAnalyzer { } /// Use the execution snapshot's schemas for checks that configuration alone cannot make. - pub(crate) fn with_catalog( - mut self, - catalog: Arc, - ) -> Self { + pub(crate) fn with_catalog(mut self, catalog: Arc) -> Self { self.catalog = Some(catalog); self } diff --git a/crates/lance-csr/src/semantic/types.rs b/crates/lance-csr/src/semantic/types.rs index 5718e4d..e70a034 100644 --- a/crates/lance-csr/src/semantic/types.rs +++ b/crates/lance-csr/src/semantic/types.rs @@ -324,8 +324,8 @@ pub(super) fn comparison_operator(operator: &ComparisonOperator) -> Operator { #[cfg(test)] mod tests { use super::*; + use crate::{InMemoryCatalog, SimpleTableSource}; use arrow_schema::{Field, Schema}; - use lance_csr_catalog::{InMemoryCatalog, SimpleTableSource}; fn analyze(query: &str) -> SemanticResult { let schema = Arc::new(Schema::new(vec![ diff --git a/crates/lance-csr-catalog/src/source_catalog.rs b/crates/lance-csr/src/source_catalog.rs similarity index 100% rename from crates/lance-csr-catalog/src/source_catalog.rs rename to crates/lance-csr/src/source_catalog.rs diff --git a/docs/project_structure.md b/docs/project_structure.md index 554881b..37a315d 100644 --- a/docs/project_structure.md +++ b/docs/project_structure.md @@ -1,15 +1,14 @@ # Project structure -lance-csr is a Cargo workspace with Python bindings. The Rust crates are named `lance-csr*`; the engine and -catalog crates are published on crates.io. +lance-csr is a Cargo workspace with Python bindings. The Rust crates are named `lance-csr*`; only the engine +crate, `lance-csr`, is published on crates.io. ``` . -├── Cargo.toml # Workspace: lance-csr, lance-csr-catalog, lance-csr-benches +├── Cargo.toml # Workspace: lance-csr, lance-csr-benches ├── crates/ │ ├── lance-csr/ # The engine: Cypher parser, semantic analysis, logical planner, │ │ # DataFusion planner, CSR index and traversal operators -│ ├── lance-csr-catalog/ # Graph source catalog and DirNamespace │ ├── lance-csr-python/ # PyO3 bindings (built by maturin; outside the workspace) │ └── lance-csr-benches/ # Criterion benchmarks (not published) ├── python/ diff --git a/docs/rust.md b/docs/rust.md index 3bcd5f4..b4a4423 100644 --- a/docs/rust.md +++ b/docs/rust.md @@ -7,11 +7,9 @@ different: installing the crate, the async API, and the Rust types. !!! info "Rust API reference" - Every public Rust type and function is documented on docs.rs: - [`lance-csr`](https://docs.rs/lance-csr) for the engine, and - [`lance-csr-catalog`](https://docs.rs/lance-csr-catalog) for `DirNamespace` and the catalog types. - Both crates are published on crates.io as [`lance-csr`](https://crates.io/crates/lance-csr) and - [`lance-csr-catalog`](https://crates.io/crates/lance-csr-catalog). + Every public Rust type and function is documented on docs.rs under + [`lance-csr`](https://docs.rs/lance-csr), including `DirNamespace` and the catalog types, which + lived in a separate `lance-csr-catalog` crate up to 0.7.1. Every Rust snippet on this page comes from [`examples/embedded_query.rs`](https://github.com/hyperdimensionalcomputingai/lance-csr/blob/main/crates/lance-csr/examples/embedded_query.rs), @@ -132,6 +130,42 @@ checked, so also race the query against the same deadline with `tokio::time::tim The deadline applies only to work inside the call; it isn't stored in the engine. A timed-out query surfaces as `Err(Elapsed)` from `timeout_at`, or as an execution error from a join checkpoint; treat both as a timeout. +## Graph mappings and pinned snapshots + +An application that keeps a graph as plain Lance tables, one per node label and relationship type, has to +record which table plays which role and which version of each table it read. The `mapping` module holds +both, in a JSON format that applications can store next to their tables or in a publication manifest: + +- `GraphMapping` is the graph descriptor: per label, its table and key column, and per relationship type, + its source and target columns and the node labels they point to. `validate()` checks what needs no + table data: version 1, at least one node label, at most 64 labels, labels that are ASCII identifiers + and unique ignoring case, and one graph role per table. +- `GraphSnapshot` pins each table, by `name`, to a `path` and Lance `version`, and carries the mapping. + It ignores fields it doesn't know, so it reads a manifest that also lists row counts or schemas. + +`GraphSnapshot::open` opens each mapped table at its pinned version and returns the `GraphConfig` and +sources that `CypherEngine::new` takes. Building the configuration also checks the tables themselves: the +key, display and endpoint columns exist, keys are scalar, and each endpoint column has the type of the node +key it refers to. + +```rust +use std::collections::HashMap; +use lance_csr::{CypherEngine, GraphSnapshot}; + +let snapshot: GraphSnapshot = serde_json::from_str(&manifest)?; +let (config, sources) = snapshot.open(&HashMap::new()).await?; +let engine = CypherEngine::new(config, sources, true).await?; +``` + +A relative `path` in a manifest is relative to wherever the application keeps its data, so resolve it +before calling `open`. To open one table at a pinned version without a mapping, use +`lance_csr::open_lance_dataset(uri, Some(version), &storage_options)`. + +`CypherQuery::execute_with_namespace` accepts any Lance namespace, such as Lance's own `DirectoryNamespace` +as well as `DirNamespace`. It opens each table at the version and with the storage options that the +namespace's `describe_table` returns, and at the latest version when it returns none, which is always the +case for `DirNamespace`. + ## Python and Rust side by side | Python | Rust |