diff --git a/docs/src/python/python.md b/docs/src/python/python.md index 340f44138..9393c57ae 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -204,18 +204,19 @@ listing a storage directory. ## Property Graphs -A property graph is read as nodes and edges, in one of two modes. An -independent graph holds its own rows: define it with node and edge types, and -write it with -[insert_into_property_graph][lancedb.DBConnection.insert_into_property_graph]. -A materialized view graph reads tables: define it with node and edge tables, -and bring it up to them with -[refresh_property_graph][lancedb.DBConnection.refresh_property_graph]. Manage -either with [create_property_graph][lancedb.DBConnection.create_property_graph], -[describe_property_graph][lancedb.DBConnection.describe_property_graph], -[rollback_property_graph][lancedb.DBConnection.rollback_property_graph], -[list_property_graphs][lancedb.DBConnection.list_property_graphs] and -[drop_property_graph][lancedb.DBConnection.drop_property_graph]. +A property graph is read as nodes and edges with SQL/PGQ `GRAPH_TABLE`, and +comes in three kinds that share one namespace of names: + +- A property graph holds its own rows. Define it with node and edge types, and + write it with [PropertyGraph.insert][lancedb.graph.PropertyGraph.insert]. +- A virtual property graph reads tables as they are. Define it with node and + edge tables. +- A materialized virtual property graph has a virtual graph's definition, and + is read as its last refresh built it: bring it up to its tables with + [MaterializedVirtualPropertyGraph.refresh][lancedb.graph.MaterializedVirtualPropertyGraph.refresh]. + +Each kind has its own create, open, describe, list and drop, such as +[create_virtual_property_graph][lancedb.DBConnection.create_virtual_property_graph]. ::: lancedb.graph.NodeType @@ -227,8 +228,18 @@ either with [create_property_graph][lancedb.DBConnection.create_property_graph], ::: lancedb.graph.Endpoint +::: lancedb.graph.PropertyGraph + +::: lancedb.graph.VirtualPropertyGraph + +::: lancedb.graph.MaterializedVirtualPropertyGraph + ::: lancedb.graph.PropertyGraphDescription +::: lancedb.graph.VirtualPropertyGraphDescription + +::: lancedb.graph.MaterializedVirtualPropertyGraphDescription + ## Expressions Type-safe expression builder for filters and projections. Use these instead diff --git a/python/python/lancedb/__init__.py b/python/python/lancedb/__init__.py index ec93cbd5d..9531906f7 100644 --- a/python/python/lancedb/__init__.py +++ b/python/python/lancedb/__init__.py @@ -48,7 +48,14 @@ from .materialized_view import ( MaterializedViewDefinition, ) from .view import ViewDescription as ViewDescription -from .graph import PropertyGraphDescription as PropertyGraphDescription +from .graph import ( + MaterializedVirtualPropertyGraph, + MaterializedVirtualPropertyGraphDescription, + PropertyGraph, + PropertyGraphDescription, + VirtualPropertyGraph, + VirtualPropertyGraphDescription, +) from .table import AsyncTable, Table from .types import BaseTokenizerType from ._lancedb import Session @@ -585,7 +592,12 @@ __all__ = [ "MaterializedView", "MaterializedViewDefinition", "ViewDescription", + "PropertyGraph", "PropertyGraphDescription", + "VirtualPropertyGraph", + "VirtualPropertyGraphDescription", + "MaterializedVirtualPropertyGraph", + "MaterializedVirtualPropertyGraphDescription", "connect", "connect_async", "tokenize", diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index c0cbef484..f9ac29e9e 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -188,20 +188,21 @@ class Connection(object): async def list_views( self, namespace_path: Optional[List[str]] = None ) -> List[str]: ... - async def create_property_graph( + async def create_graph( self, + kind: str, name: str, definition_json: str, namespace_path: Optional[List[str]] = None, ) -> str: ... - async def describe_property_graph( - self, name: str, namespace_path: Optional[List[str]] = None + async def describe_graph( + self, kind: str, name: str, namespace_path: Optional[List[str]] = None ) -> str: ... - async def drop_property_graph( - self, name: str, namespace_path: Optional[List[str]] = None + async def drop_graph( + self, kind: str, name: str, namespace_path: Optional[List[str]] = None ) -> None: ... - async def list_property_graphs( - self, namespace_path: Optional[List[str]] = None + async def list_graphs( + self, kind: str, namespace_path: Optional[List[str]] = None ) -> List[str]: ... async def insert_into_property_graph( self, @@ -210,10 +211,10 @@ class Connection(object): data: Any, namespace_path: Optional[List[str]] = None, ) -> str: ... - async def refresh_property_graph( + async def rollback_property_graph( self, name: str, namespace_path: Optional[List[str]] = None ) -> str: ... - async def rollback_property_graph( + async def refresh_materialized_virtual_property_graph( self, name: str, namespace_path: Optional[List[str]] = None ) -> str: ... async def list_jobs(self) -> List[JobInfo]: ... diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index 7e7330af2..7ea4b45de 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -50,15 +50,25 @@ from . import __version__ from ._lancedb import connect as lancedb_connect # type: ignore from .functions import FunctionVersion, UdfDefinition from .graph import ( + MATERIALIZED_VIRTUAL_PROPERTY_GRAPH, + PROPERTY_GRAPH, + VIRTUAL_PROPERTY_GRAPH, + AsyncMaterializedVirtualPropertyGraph, + AsyncPropertyGraph, + AsyncVirtualPropertyGraph, EdgeTable, EdgeType, + MaterializedVirtualPropertyGraph, + MaterializedVirtualPropertyGraphDescription, NodeTable, NodeType, + PropertyGraph, PropertyGraphDescription, - _definition_json, - _description_from_json, + VirtualPropertyGraph, + VirtualPropertyGraphDescription, + _tables_json, + _types_json, ) -from .scannable import to_scannable from .job import AsyncJob, Job, _typed_job from .sql import AsyncQuery as AsyncSqlQuery from .sql import Query as SqlQuery @@ -1031,81 +1041,25 @@ class DBConnection(EnforceOverrides): def create_property_graph( self, name: str, - nodes: List[Union[NodeTable, NodeType]], - edges: List[Union[EdgeTable, EdgeType]], + nodes: List[NodeType], + edges: List[EdgeType], *, namespace_path: Optional[List[str]] = None, - ) -> PropertyGraphDescription: - """Create a property graph. + ) -> PropertyGraph: + """Create a property graph: a graph that holds its own rows. - With [NodeTable][lancedb.graph.NodeTable] and - [EdgeTable][lancedb.graph.EdgeTable], the graph is a materialized view - of tables in its namespace: each node table names its key column and - label, each edge table its label and the node keys its source and - destination columns reference, and the graph is built from them now. - With [NodeType][lancedb.graph.NodeType] and - [EdgeType][lancedb.graph.EdgeType], the graph is independent: it holds - its own rows, starts empty, and takes rows from - [insert_into_property_graph][lancedb.db.DBConnection.insert_into_property_graph]. - A name already taken is an error. Local connections raise + Each [NodeType][lancedb.graph.NodeType] and + [EdgeType][lancedb.graph.EdgeType] gives one label's schema. The graph + starts empty and takes rows from + [PropertyGraph.insert][lancedb.graph.PropertyGraph.insert]. A name held + by a graph of any kind is taken. Local connections raise ``NotImplementedError``. >>> import lancedb - >>> from lancedb.graph import EdgeTable, Endpoint, NodeTable - >>> db = lancedb.connect("db://my_database") # doctest: +SKIP - >>> graph = db.create_property_graph( - ... "social", - ... nodes=[NodeTable("person", key="person_id", label="Person")], - ... edges=[ - ... EdgeTable( - ... "knows", - ... label="KNOWS", - ... source=Endpoint("src_id", references=("person", "person_id")), - ... destination=Endpoint( - ... "dst_id", references=("person", "person_id") - ... ), - ... ) - ... ], - ... ) # doctest: +SKIP - """ - raise NotImplementedError( - "Property graph operations are not supported for this connection type" - ) - - def describe_property_graph( - self, name: str, *, namespace_path: Optional[List[str]] = None - ) -> PropertyGraphDescription: - """What this database records about a property graph: its mode and - definition, its current and previous commits, its size, and, for a - materialized view, the table versions it was refreshed from. - - Local connections raise ``NotImplementedError``. - """ - raise NotImplementedError( - "Property graph operations are not supported for this connection type" - ) - - def insert_into_property_graph( - self, - name: str, - label: str, - data: DATA, - *, - namespace_path: Optional[List[str]] = None, - ) -> PropertyGraphDescription: - """Insert rows into an independent property graph, as one commit. - - The rows are one node label's or one edge label's. Nodes upsert by key: - a key the graph holds replaces that node's properties. Edges append, - carrying their endpoints' keys in the columns the edge type names; an - edge whose endpoint is not a node of the graph refuses the whole - insert. A materialized view graph is written through its tables and - refreshed instead. Local connections raise ``NotImplementedError``. - >>> import pyarrow as pa >>> from lancedb.graph import EdgeType, NodeType >>> db = lancedb.connect("db://my_database") # doctest: +SKIP - >>> db.create_property_graph( + >>> people = db.create_property_graph( ... "people", ... nodes=[ ... NodeType( @@ -1127,35 +1081,43 @@ class DBConnection(EnforceOverrides): ... ) ... ], ... ) # doctest: +SKIP - >>> db.insert_into_property_graph( - ... "people", "Person", [{"person_id": 1, "name": "Alice"}] + >>> people.insert( + ... "Person", [{"person_id": 1, "name": "Alice"}] ... ) # doctest: +SKIP """ raise NotImplementedError( "Property graph operations are not supported for this connection type" ) - def refresh_property_graph( + def open_property_graph( self, name: str, *, namespace_path: Optional[List[str]] = None - ) -> PropertyGraphDescription: - """Bring a materialized view property graph up to its tables' latest - versions. + ) -> PropertyGraph: + """Open the property graph named ``name``; an error if there is none. - Until it is refreshed, a query over a table that has changed reads the - tables rather than the graph. Local connections raise - ``NotImplementedError``. + Local connections raise ``NotImplementedError``. """ raise NotImplementedError( "Property graph operations are not supported for this connection type" ) - def rollback_property_graph( + def describe_property_graph( self, name: str, *, namespace_path: Optional[List[str]] = None ) -> PropertyGraphDescription: - """Return a property graph to its previous commit. + """What this database records about a property graph: its node and edge + types, its current and previous commits, and its size. - The commit rolled back from is discarded, so a second rollback in a row - is an error. Local connections raise ``NotImplementedError``. + Local connections raise ``NotImplementedError``. + """ + raise NotImplementedError( + "Property graph operations are not supported for this connection type" + ) + + def list_property_graphs( + self, *, namespace_path: Optional[List[str]] = None + ) -> List[str]: + """The names of the property graphs in one namespace. + + Local connections raise ``NotImplementedError``. """ raise NotImplementedError( "Property graph operations are not supported for this connection type" @@ -1166,17 +1128,169 @@ class DBConnection(EnforceOverrides): ) -> None: """Drop a property graph and wait for its data to be deleted. - The tables it reads are untouched. Local connections raise - ``NotImplementedError``. + Local connections raise ``NotImplementedError``. """ raise NotImplementedError( "Property graph operations are not supported for this connection type" ) - def list_property_graphs( + def create_virtual_property_graph( + self, + name: str, + nodes: List[NodeTable], + edges: List[EdgeTable], + *, + namespace_path: Optional[List[str]] = None, + ) -> VirtualPropertyGraph: + """Create a virtual property graph over tables in its namespace. + + Each [NodeTable][lancedb.graph.NodeTable] names the table holding one + label's nodes and its key column, and each + [EdgeTable][lancedb.graph.EdgeTable] the table holding one label's edges + and the node keys its source and destination columns reference. The + graph stores nothing else: a query reads the tables as they are then. A + name held by a graph of any kind is taken. Local connections raise + ``NotImplementedError``. + + >>> import lancedb + >>> from lancedb.graph import EdgeTable, Endpoint, NodeTable + >>> db = lancedb.connect("db://my_database") # doctest: +SKIP + >>> graph = db.create_virtual_property_graph( + ... "social", + ... nodes=[NodeTable("person", key="person_id", label="Person")], + ... edges=[ + ... EdgeTable( + ... "knows", + ... label="KNOWS", + ... source=Endpoint("src_id", references=("person", "person_id")), + ... destination=Endpoint( + ... "dst_id", references=("person", "person_id") + ... ), + ... ) + ... ], + ... ) # doctest: +SKIP + """ + raise NotImplementedError( + "Property graph operations are not supported for this connection type" + ) + + def open_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> VirtualPropertyGraph: + """Open the virtual property graph named ``name``; an error if there is + none. + + Local connections raise ``NotImplementedError``. + """ + raise NotImplementedError( + "Property graph operations are not supported for this connection type" + ) + + def describe_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> VirtualPropertyGraphDescription: + """What this database records about a virtual property graph: its node + and edge tables, with the properties create resolved. + + Local connections raise ``NotImplementedError``. + """ + raise NotImplementedError( + "Property graph operations are not supported for this connection type" + ) + + def list_virtual_property_graphs( self, *, namespace_path: Optional[List[str]] = None ) -> List[str]: - """The names of the property graphs in one namespace. + """The names of the virtual property graphs in one namespace. + + Local connections raise ``NotImplementedError``. + """ + raise NotImplementedError( + "Property graph operations are not supported for this connection type" + ) + + def drop_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> None: + """Drop a virtual property graph and wait for its definition to be + deleted. The tables it reads are untouched. + + Local connections raise ``NotImplementedError``. + """ + raise NotImplementedError( + "Property graph operations are not supported for this connection type" + ) + + def create_materialized_virtual_property_graph( + self, + name: str, + nodes: List[NodeTable], + edges: List[EdgeTable], + *, + namespace_path: Optional[List[str]] = None, + ) -> MaterializedVirtualPropertyGraph: + """Create a materialized virtual property graph over tables in its + namespace. + + The definition is a virtual property graph's (see + [create_virtual_property_graph][lancedb.db.DBConnection.create_virtual_property_graph]). + The graph is built from the tables now, and read as that build until + [refreshed][lancedb.graph.MaterializedVirtualPropertyGraph.refresh]. A + name held by a graph of any kind is taken. Local connections raise + ``NotImplementedError``. + + >>> db = lancedb.connect("db://my_database") # doctest: +SKIP + >>> mvpg = db.create_materialized_virtual_property_graph( + ... "social_mv", nodes, edges + ... ) # doctest: +SKIP + >>> mvpg.refresh() # doctest: +SKIP + """ + raise NotImplementedError( + "Property graph operations are not supported for this connection type" + ) + + def open_materialized_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> MaterializedVirtualPropertyGraph: + """Open the materialized virtual property graph named ``name``; an error + if there is none. + + Local connections raise ``NotImplementedError``. + """ + raise NotImplementedError( + "Property graph operations are not supported for this connection type" + ) + + def describe_materialized_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> MaterializedVirtualPropertyGraphDescription: + """What this database records about a materialized virtual property + graph: its node and edge tables, and what its last refresh built from + them. + + Local connections raise ``NotImplementedError``. + """ + raise NotImplementedError( + "Property graph operations are not supported for this connection type" + ) + + def list_materialized_virtual_property_graphs( + self, *, namespace_path: Optional[List[str]] = None + ) -> List[str]: + """The names of the materialized virtual property graphs in one + namespace. + + Local connections raise ``NotImplementedError``. + """ + raise NotImplementedError( + "Property graph operations are not supported for this connection type" + ) + + def drop_materialized_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> None: + """Drop a materialized virtual property graph and wait for its data to + be deleted. The tables it reads are untouched. Local connections raise ``NotImplementedError``. """ @@ -2049,72 +2163,6 @@ class LanceDBConnection(DBConnection): def list_views(self, *, namespace_path: Optional[List[str]] = None) -> List[str]: return LOOP.run(self._conn.list_views(namespace_path=namespace_path)) - @override - def create_property_graph( - self, - name: str, - nodes: List[Union[NodeTable, NodeType]], - edges: List[Union[EdgeTable, EdgeType]], - *, - namespace_path: Optional[List[str]] = None, - ) -> PropertyGraphDescription: - return LOOP.run( - self._conn.create_property_graph( - name, nodes, edges, namespace_path=namespace_path - ) - ) - - @override - def describe_property_graph( - self, name: str, *, namespace_path: Optional[List[str]] = None - ) -> PropertyGraphDescription: - return LOOP.run( - self._conn.describe_property_graph(name, namespace_path=namespace_path) - ) - - @override - def insert_into_property_graph( - self, - name: str, - label: str, - data: DATA, - *, - namespace_path: Optional[List[str]] = None, - ) -> PropertyGraphDescription: - return LOOP.run( - self._conn.insert_into_property_graph( - name, label, data, namespace_path=namespace_path - ) - ) - - @override - def refresh_property_graph( - self, name: str, *, namespace_path: Optional[List[str]] = None - ) -> PropertyGraphDescription: - return LOOP.run( - self._conn.refresh_property_graph(name, namespace_path=namespace_path) - ) - - @override - def rollback_property_graph( - self, name: str, *, namespace_path: Optional[List[str]] = None - ) -> PropertyGraphDescription: - return LOOP.run( - self._conn.rollback_property_graph(name, namespace_path=namespace_path) - ) - - @override - def drop_property_graph( - self, name: str, *, namespace_path: Optional[List[str]] = None - ) -> None: - LOOP.run(self._conn.drop_property_graph(name, namespace_path=namespace_path)) - - @override - def list_property_graphs( - self, *, namespace_path: Optional[List[str]] = None - ) -> List[str]: - return LOOP.run(self._conn.list_property_graphs(namespace_path=namespace_path)) - @override def list_jobs(self) -> List[JobInfo]: """List server-side jobs across the database's tables.""" @@ -3145,77 +3193,162 @@ class AsyncConnection(object): async def create_property_graph( self, name: str, - nodes: List[Union[NodeTable, NodeType]], - edges: List[Union[EdgeTable, EdgeType]], + nodes: List[NodeType], + edges: List[EdgeType], *, namespace_path: Optional[List[str]] = None, - ) -> PropertyGraphDescription: - """Create a property graph: a materialized view of tables in its - namespace, or an independent graph that holds its own rows. + ) -> AsyncPropertyGraph: + """Create a property graph: a graph that holds its own rows. See [DBConnection.create_property_graph][lancedb.DBConnection.create_property_graph]. """ - return _description_from_json( - await self._inner.create_property_graph( - name, _definition_json(nodes, edges), list(namespace_path or []) - ) + graph = AsyncPropertyGraph(self._inner, name, namespace_path) + await self._inner.create_graph( + PROPERTY_GRAPH, name, _types_json(nodes, edges), graph.namespace_path ) + return graph + + async def open_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> AsyncPropertyGraph: + """Open the property graph named ``name``; an error if there is none.""" + graph = AsyncPropertyGraph(self._inner, name, namespace_path) + await graph.describe() + return graph async def describe_property_graph( self, name: str, *, namespace_path: Optional[List[str]] = None ) -> PropertyGraphDescription: """What this database records about a property graph.""" - return _description_from_json( - await self._inner.describe_property_graph(name, list(namespace_path or [])) - ) - - async def insert_into_property_graph( - self, - name: str, - label: str, - data: DATA, - *, - namespace_path: Optional[List[str]] = None, - ) -> PropertyGraphDescription: - """Insert rows of one label into an independent property graph. - - See - [DBConnection.insert_into_property_graph][lancedb.DBConnection.insert_into_property_graph]. - """ - return _description_from_json( - await self._inner.insert_into_property_graph( - name, label, to_scannable(data), list(namespace_path or []) - ) - ) - - async def refresh_property_graph( - self, name: str, *, namespace_path: Optional[List[str]] = None - ) -> PropertyGraphDescription: - """Bring a materialized view property graph up to its tables.""" - return _description_from_json( - await self._inner.refresh_property_graph(name, list(namespace_path or [])) - ) - - async def rollback_property_graph( - self, name: str, *, namespace_path: Optional[List[str]] = None - ) -> PropertyGraphDescription: - """Return a property graph to its previous commit.""" - return _description_from_json( - await self._inner.rollback_property_graph(name, list(namespace_path or [])) - ) - - async def drop_property_graph( - self, name: str, *, namespace_path: Optional[List[str]] = None - ) -> None: - """Drop a property graph and wait for its data to be deleted.""" - await self._inner.drop_property_graph(name, list(namespace_path or [])) + return await AsyncPropertyGraph(self._inner, name, namespace_path).describe() async def list_property_graphs( self, *, namespace_path: Optional[List[str]] = None ) -> List[str]: """The names of the property graphs in one namespace.""" - return await self._inner.list_property_graphs(list(namespace_path or [])) + return await self._inner.list_graphs(PROPERTY_GRAPH, list(namespace_path or [])) + + async def drop_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> None: + """Drop a property graph and wait for its data to be deleted.""" + await self._inner.drop_graph(PROPERTY_GRAPH, name, list(namespace_path or [])) + + async def create_virtual_property_graph( + self, + name: str, + nodes: List[NodeTable], + edges: List[EdgeTable], + *, + namespace_path: Optional[List[str]] = None, + ) -> AsyncVirtualPropertyGraph: + """Create a virtual property graph over tables in its namespace. + + See + [DBConnection.create_virtual_property_graph][lancedb.DBConnection.create_virtual_property_graph]. + """ + graph = AsyncVirtualPropertyGraph(self._inner, name, namespace_path) + await self._inner.create_graph( + VIRTUAL_PROPERTY_GRAPH, + name, + _tables_json(nodes, edges), + graph.namespace_path, + ) + return graph + + async def open_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> AsyncVirtualPropertyGraph: + """Open the virtual property graph named ``name``; an error if there is + none.""" + graph = AsyncVirtualPropertyGraph(self._inner, name, namespace_path) + await graph.describe() + return graph + + async def describe_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> VirtualPropertyGraphDescription: + """What this database records about a virtual property graph.""" + return await AsyncVirtualPropertyGraph( + self._inner, name, namespace_path + ).describe() + + async def list_virtual_property_graphs( + self, *, namespace_path: Optional[List[str]] = None + ) -> List[str]: + """The names of the virtual property graphs in one namespace.""" + return await self._inner.list_graphs( + VIRTUAL_PROPERTY_GRAPH, list(namespace_path or []) + ) + + async def drop_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> None: + """Drop a virtual property graph and wait for its definition to be + deleted.""" + await self._inner.drop_graph( + VIRTUAL_PROPERTY_GRAPH, name, list(namespace_path or []) + ) + + async def create_materialized_virtual_property_graph( + self, + name: str, + nodes: List[NodeTable], + edges: List[EdgeTable], + *, + namespace_path: Optional[List[str]] = None, + ) -> AsyncMaterializedVirtualPropertyGraph: + """Create a materialized virtual property graph, built from its tables + now. + + See + [DBConnection.create_materialized_virtual_property_graph][lancedb.DBConnection.create_materialized_virtual_property_graph]. + """ + graph = AsyncMaterializedVirtualPropertyGraph(self._inner, name, namespace_path) + await self._inner.create_graph( + MATERIALIZED_VIRTUAL_PROPERTY_GRAPH, + name, + _tables_json(nodes, edges), + graph.namespace_path, + ) + return graph + + async def open_materialized_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> AsyncMaterializedVirtualPropertyGraph: + """Open the materialized virtual property graph named ``name``; an error + if there is none.""" + graph = AsyncMaterializedVirtualPropertyGraph(self._inner, name, namespace_path) + await graph.describe() + return graph + + async def describe_materialized_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> MaterializedVirtualPropertyGraphDescription: + """What this database records about a materialized virtual property + graph.""" + return await AsyncMaterializedVirtualPropertyGraph( + self._inner, name, namespace_path + ).describe() + + async def list_materialized_virtual_property_graphs( + self, *, namespace_path: Optional[List[str]] = None + ) -> List[str]: + """The names of the materialized virtual property graphs in one + namespace.""" + return await self._inner.list_graphs( + MATERIALIZED_VIRTUAL_PROPERTY_GRAPH, list(namespace_path or []) + ) + + async def drop_materialized_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> None: + """Drop a materialized virtual property graph and wait for its data to + be deleted.""" + await self._inner.drop_graph( + MATERIALIZED_VIRTUAL_PROPERTY_GRAPH, name, list(namespace_path or []) + ) async def list_jobs(self) -> List[JobInfo]: """List server-side jobs across the database's tables.""" diff --git a/python/python/lancedb/graph.py b/python/python/lancedb/graph.py index eaf16d2f3..dc5ffc19b 100644 --- a/python/python/lancedb/graph.py +++ b/python/python/lancedb/graph.py @@ -1,31 +1,46 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright The LanceDB Authors -"""Property graphs: catalog objects read as graphs, in one of two modes. +"""Property graphs: catalog objects read as graphs, in three kinds that share +one namespace of names. -An *independent* graph holds its own rows. Its node types (:class:`NodeType`) -and edge types (:class:`EdgeType`) have schemas, and rows are inserted into the -graph itself with ``insert_into_property_graph``. +A *property graph* holds its own rows. Its node types (:class:`NodeType`) and +edge types (:class:`EdgeType`) have schemas, and rows are inserted into the +graph itself with :meth:`PropertyGraph.insert`. -A *materialized view* graph reads tables in its namespace. Its definition says +A *virtual property graph* reads tables in its namespace. Its definition says which tables hold nodes and which hold edges: each node table's key and label (:class:`NodeTable`), each edge table's label and how its source and destination columns reference node keys (:class:`EdgeTable`) -- the shape of -SQL/PGQ's ``CREATE PROPERTY GRAPH``. The tables stay ordinary tables; after -they change, ``refresh_property_graph`` brings the graph up to date. +SQL/PGQ's ``CREATE PROPERTY GRAPH``. A query reads the tables as they are. -See ``DBConnection.create_property_graph``. +A *materialized virtual property graph* has a virtual graph's definition, and +is read as its last refresh built it: after the tables change, +:meth:`MaterializedVirtualPropertyGraph.refresh` brings it up to date. + +See ``DBConnection.create_property_graph``, +``DBConnection.create_virtual_property_graph`` and +``DBConnection.create_materialized_virtual_property_graph``. """ from __future__ import annotations import json from dataclasses import dataclass -from typing import Any, Dict, List, Optional, Sequence, Tuple, Union +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Tuple, Type import pyarrow as pa from ._lancedb import graph_schema_from_json, graph_schema_to_json +from .background_loop import LOOP +from .scannable import to_scannable + +if TYPE_CHECKING: + from .common import DATA + +PROPERTY_GRAPH = "property_graph" +VIRTUAL_PROPERTY_GRAPH = "virtual_property_graph" +MATERIALIZED_VIRTUAL_PROPERTY_GRAPH = "materialized_virtual_property_graph" @dataclass(frozen=True) @@ -72,7 +87,7 @@ class EdgeTable: @dataclass(frozen=True) class NodeType: - """The nodes of one label of an independent graph.""" + """The nodes of one label of a property graph.""" label: str """The label its nodes carry.""" @@ -85,7 +100,7 @@ class NodeType: @dataclass(frozen=True) class EdgeType: - """The edges of one label of an independent graph.""" + """The edges of one label of a property graph.""" label: str """The label its edges carry.""" @@ -107,12 +122,10 @@ class PropertyGraphDescription: """The graph's name within its namespace.""" namespace_path: List[str] """The namespace holding the graph; empty is the root namespace.""" - mode: str - """``"independent"`` or ``"materialized_view"``.""" - nodes: List[Union[NodeTable, NodeType]] - """The node tables (materialized view) or node types (independent).""" - edges: List[Union[EdgeTable, EdgeType]] - """The edge tables (materialized view) or edge types (independent).""" + nodes: List[NodeType] + """The node types.""" + edges: List[EdgeType] + """The edge types.""" commit: str """The graph's current commit.""" committed_at: str @@ -123,52 +136,120 @@ class PropertyGraphDescription: """How many nodes the current commit holds.""" edge_count: int """How many edges the current commit holds.""" + + +@dataclass(frozen=True) +class VirtualPropertyGraphDescription: + """What a database records about one virtual property graph.""" + + name: str + """The graph's name within its namespace.""" + namespace_path: List[str] + """The namespace holding the graph; empty is the root namespace.""" + nodes: List[NodeTable] + """The node tables, each with the properties create resolved.""" + edges: List[EdgeTable] + """The edge tables, each with the properties create resolved.""" + + +@dataclass(frozen=True) +class MaterializedVirtualPropertyGraphDescription: + """What a database records about one materialized virtual property graph.""" + + name: str + """The graph's name within its namespace.""" + namespace_path: List[str] + """The namespace holding the graph; empty is the root namespace.""" + nodes: List[NodeTable] + """The node tables, each with the properties create resolved.""" + edges: List[EdgeTable] + """The edge tables, each with the properties create resolved.""" + commit: str + """The last refresh's commit.""" + committed_at: str + """When the last refresh landed, as RFC 3339.""" + vertex_count: int + """How many nodes the last refresh built.""" + edge_count: int + """How many edges the last refresh built.""" sources: Dict[str, int] - """For a materialized view, the version of each table the current commit - was refreshed from, by table name.""" + """The version of each table the last refresh read, by table name.""" -INDEPENDENT = "independent" -MATERIALIZED_VIEW = "materialized_view" +def _require( + kind: str, + nodes: Sequence[Any], + edges: Sequence[Any], + node_type: Type, + edge_type: Type, +) -> None: + for element, expected in [ + *((node, node_type) for node in nodes), + *((edge, edge_type) for edge in edges), + ]: + if not isinstance(element, expected): + raise TypeError( + f"A {kind}'s nodes are {node_type.__name__} and its edges " + f"{edge_type.__name__}, not {type(element).__name__}" + ) -def _mode_of(nodes: Sequence[Any], edges: Sequence[Any]) -> str: - kinds = {type(element) for element in [*nodes, *edges]} - if kinds <= {NodeType, EdgeType} and kinds: - return INDEPENDENT - if kinds <= {NodeTable, EdgeTable}: - return MATERIALIZED_VIEW - raise ValueError( - "A property graph's nodes and edges are either all tables " - "(NodeTable, EdgeTable) or all types (NodeType, EdgeType)" - ) - - -def _definition_json(nodes: Sequence[Any], edges: Sequence[Any]) -> str: - mode = _mode_of(nodes, edges) +def _types_json(nodes: Sequence[NodeType], edges: Sequence[EdgeType]) -> str: + _require("property graph", nodes, edges, NodeType, EdgeType) return json.dumps( { - "mode": mode, - "nodes": [_node_to_json(node) for node in nodes], - "edges": [_edge_to_json(edge) for edge in edges], + "nodes": [_node_type_to_json(node) for node in nodes], + "edges": [_edge_type_to_json(edge) for edge in edges], } ) -def _description_from_json(text: str) -> PropertyGraphDescription: +def _tables_json(nodes: Sequence[NodeTable], edges: Sequence[EdgeTable]) -> str: + _require("virtual property graph", nodes, edges, NodeTable, EdgeTable) + return json.dumps( + { + "nodes": [_node_table_to_json(node) for node in nodes], + "edges": [_edge_table_to_json(edge) for edge in edges], + } + ) + + +def _property_description(text: str) -> PropertyGraphDescription: described = json.loads(text) - mode = described["mode"] return PropertyGraphDescription( name=described["name"], namespace_path=list(described.get("namespace", [])), - mode=mode, - nodes=[_node_from_json(mode, node) for node in described["nodes"]], - edges=[_edge_from_json(mode, edge) for edge in described.get("edges", [])], + nodes=[_node_type_from_json(node) for node in described["nodes"]], + edges=[_edge_type_from_json(edge) for edge in described.get("edges", [])], commit=described["commit"], committed_at=described["committed_at"], previous_commit=described.get("previous_commit"), vertex_count=described["vertex_count"], edge_count=described["edge_count"], + ) + + +def _virtual_description(text: str) -> VirtualPropertyGraphDescription: + described = json.loads(text) + return VirtualPropertyGraphDescription( + name=described["name"], + namespace_path=list(described.get("namespace", [])), + nodes=[_node_table_from_json(node) for node in described["nodes"]], + edges=[_edge_table_from_json(edge) for edge in described.get("edges", [])], + ) + + +def _materialized_description(text: str) -> MaterializedVirtualPropertyGraphDescription: + described = json.loads(text) + return MaterializedVirtualPropertyGraphDescription( + name=described["name"], + namespace_path=list(described.get("namespace", [])), + nodes=[_node_table_from_json(node) for node in described["nodes"]], + edges=[_edge_table_from_json(edge) for edge in described.get("edges", [])], + commit=described["commit"], + committed_at=described["committed_at"], + vertex_count=described["vertex_count"], + edge_count=described["edge_count"], sources={ source["table"]: source["version"] for source in described.get("sources", []) @@ -184,13 +265,26 @@ def _schema_from_json(schema: Dict[str, Any]) -> pa.Schema: return graph_schema_from_json(json.dumps(schema)) -def _node_to_json(node: Union[NodeTable, NodeType]) -> Dict[str, Any]: - if isinstance(node, NodeType): - return { - "label": node.label, - "key": node.key, - "schema": _schema_to_json(node.schema), - } +def _node_type_to_json(node: NodeType) -> Dict[str, Any]: + return { + "label": node.label, + "key": node.key, + "schema": _schema_to_json(node.schema), + } + + +def _edge_type_to_json(edge: EdgeType) -> Dict[str, Any]: + encoded: Dict[str, Any] = { + "label": edge.label, + "source": {"label": edge.source[0], "column": edge.source[1]}, + "destination": {"label": edge.destination[0], "column": edge.destination[1]}, + } + if edge.schema is not None: + encoded["schema"] = _schema_to_json(edge.schema) + return encoded + + +def _node_table_to_json(node: NodeTable) -> Dict[str, Any]: encoded: Dict[str, Any] = { "table": node.table, "key": node.key, @@ -201,20 +295,8 @@ def _node_to_json(node: Union[NodeTable, NodeType]) -> Dict[str, Any]: return encoded -def _edge_to_json(edge: Union[EdgeTable, EdgeType]) -> Dict[str, Any]: - if isinstance(edge, EdgeType): - encoded: Dict[str, Any] = { - "label": edge.label, - "source": {"label": edge.source[0], "column": edge.source[1]}, - "destination": { - "label": edge.destination[0], - "column": edge.destination[1], - }, - } - if edge.schema is not None: - encoded["schema"] = _schema_to_json(edge.schema) - return encoded - encoded = { +def _edge_table_to_json(edge: EdgeTable) -> Dict[str, Any]: + encoded: Dict[str, Any] = { "table": edge.table, "label": edge.label, "source": _endpoint_to_json(edge.source), @@ -233,13 +315,23 @@ def _endpoint_to_json(endpoint: Endpoint) -> Dict[str, Any]: } -def _node_from_json(mode: str, node: Dict[str, Any]) -> Union[NodeTable, NodeType]: - if mode == INDEPENDENT: - return NodeType( - label=node["label"], - key=node["key"], - schema=_schema_from_json(node["schema"]), - ) +def _node_type_from_json(node: Dict[str, Any]) -> NodeType: + return NodeType( + label=node["label"], key=node["key"], schema=_schema_from_json(node["schema"]) + ) + + +def _edge_type_from_json(edge: Dict[str, Any]) -> EdgeType: + schema = edge.get("schema") + return EdgeType( + label=edge["label"], + source=(edge["source"]["label"], edge["source"]["column"]), + destination=(edge["destination"]["label"], edge["destination"]["column"]), + schema=None if schema is None else _schema_from_json(schema), + ) + + +def _node_table_from_json(node: Dict[str, Any]) -> NodeTable: return NodeTable( table=node["table"], key=node["key"], @@ -248,15 +340,7 @@ def _node_from_json(mode: str, node: Dict[str, Any]) -> Union[NodeTable, NodeTyp ) -def _edge_from_json(mode: str, edge: Dict[str, Any]) -> Union[EdgeTable, EdgeType]: - if mode == INDEPENDENT: - schema = edge.get("schema") - return EdgeType( - label=edge["label"], - source=(edge["source"]["label"], edge["source"]["column"]), - destination=(edge["destination"]["label"], edge["destination"]["column"]), - schema=None if schema is None else _schema_from_json(schema), - ) +def _edge_table_from_json(edge: Dict[str, Any]) -> EdgeTable: return EdgeTable( table=edge["table"], label=edge["label"], @@ -272,3 +356,181 @@ def _endpoint_from_json(endpoint: Dict[str, Any]) -> Endpoint: column=endpoint["column"], references=(references["table"], references["column"]), ) + + +class _AsyncGraph: + """A handle on one graph of a database: its name and namespace, and the + connection it is read and written through.""" + + _kind: str = PROPERTY_GRAPH + + def __init__(self, inner: Any, name: str, namespace_path: Optional[List[str]]): + self._inner = inner + self._name = name + self._namespace_path = list(namespace_path or []) + + def __repr__(self) -> str: + return f"{type(self).__name__}(name={self._name!r})" + + @property + def name(self) -> str: + """The graph's name within its namespace.""" + return self._name + + @property + def namespace_path(self) -> List[str]: + """The namespace holding the graph; empty is the root namespace.""" + return list(self._namespace_path) + + async def _describe(self) -> str: + return await self._inner.describe_graph( + self._kind, self._name, self._namespace_path + ) + + +class AsyncPropertyGraph(_AsyncGraph): + """A property graph: one that holds its own rows. + + Obtained from ``AsyncConnection.create_property_graph`` or + ``AsyncConnection.open_property_graph``. + """ + + _kind = PROPERTY_GRAPH + + async def describe(self) -> PropertyGraphDescription: + """The graph's node and edge types, current commit and size.""" + return _property_description(await self._describe()) + + async def insert(self, label: str, data: "DATA") -> PropertyGraphDescription: + """Insert rows of one node label or edge label, as one commit. + + Nodes upsert by key: a key the graph holds replaces that node's + properties. Edges append, carrying their endpoints' keys in the columns + the edge type names; an edge whose endpoint is not a node of the graph + refuses the whole insert. + """ + return _property_description( + await self._inner.insert_into_property_graph( + self._name, label, to_scannable(data), self._namespace_path + ) + ) + + async def rollback(self) -> PropertyGraphDescription: + """Return the graph to its previous commit. + + The commit rolled back from is discarded, so a second rollback in a row + is an error. + """ + return _property_description( + await self._inner.rollback_property_graph(self._name, self._namespace_path) + ) + + +class AsyncVirtualPropertyGraph(_AsyncGraph): + """A virtual property graph: one defined over tables and read through them. + + Obtained from ``AsyncConnection.create_virtual_property_graph`` or + ``AsyncConnection.open_virtual_property_graph``. + """ + + _kind = VIRTUAL_PROPERTY_GRAPH + + async def describe(self) -> VirtualPropertyGraphDescription: + """The graph's node and edge tables.""" + return _virtual_description(await self._describe()) + + +class AsyncMaterializedVirtualPropertyGraph(_AsyncGraph): + """A materialized virtual property graph: one defined over tables and read + as its last refresh built it. + + Obtained from ``AsyncConnection.create_materialized_virtual_property_graph`` + or ``AsyncConnection.open_materialized_virtual_property_graph``. + """ + + _kind = MATERIALIZED_VIRTUAL_PROPERTY_GRAPH + + async def describe(self) -> MaterializedVirtualPropertyGraphDescription: + """The graph's node and edge tables, and what its last refresh built.""" + return _materialized_description(await self._describe()) + + async def refresh(self) -> MaterializedVirtualPropertyGraphDescription: + """Rebuild the graph from its tables' latest versions. + + When every table's commits since the last refresh left its rows as + they were -- compaction, indexing, metadata -- only the recorded + versions advance; when nothing changed, the commit stays. + """ + return _materialized_description( + await self._inner.refresh_materialized_virtual_property_graph( + self._name, self._namespace_path + ) + ) + + +class _Graph: + """Synchronous variant of a graph handle.""" + + def __init__(self, graph: _AsyncGraph): + self._async = graph + + def __repr__(self) -> str: + return f"{type(self).__name__}(name={self.name!r})" + + @property + def name(self) -> str: + """The graph's name within its namespace.""" + return self._async.name + + @property + def namespace_path(self) -> List[str]: + """The namespace holding the graph; empty is the root namespace.""" + return self._async.namespace_path + + +class PropertyGraph(_Graph): + """Synchronous variant of + [AsyncPropertyGraph][lancedb.graph.AsyncPropertyGraph].""" + + _async: AsyncPropertyGraph + + def describe(self) -> PropertyGraphDescription: + """The graph's node and edge types, current commit and size.""" + return LOOP.run(self._async.describe()) + + def insert(self, label: str, data: "DATA") -> PropertyGraphDescription: + """Insert rows of one node label or edge label, as one commit. See + [AsyncPropertyGraph.insert][lancedb.graph.AsyncPropertyGraph.insert].""" + return LOOP.run(self._async.insert(label, data)) + + def rollback(self) -> PropertyGraphDescription: + """Return the graph to its previous commit. See + [AsyncPropertyGraph.rollback][lancedb.graph.AsyncPropertyGraph.rollback].""" + return LOOP.run(self._async.rollback()) + + +class VirtualPropertyGraph(_Graph): + """Synchronous variant of + [AsyncVirtualPropertyGraph][lancedb.graph.AsyncVirtualPropertyGraph].""" + + _async: AsyncVirtualPropertyGraph + + def describe(self) -> VirtualPropertyGraphDescription: + """The graph's node and edge tables.""" + return LOOP.run(self._async.describe()) + + +class MaterializedVirtualPropertyGraph(_Graph): + """Synchronous variant of + [AsyncMaterializedVirtualPropertyGraph][lancedb.graph.AsyncMaterializedVirtualPropertyGraph].""" + + _async: AsyncMaterializedVirtualPropertyGraph + + def describe(self) -> MaterializedVirtualPropertyGraphDescription: + """The graph's node and edge tables, and what its last refresh built.""" + return LOOP.run(self._async.describe()) + + def refresh(self) -> MaterializedVirtualPropertyGraphDescription: + """Rebuild the graph from its tables' latest versions. See + [AsyncMaterializedVirtualPropertyGraph.refresh][lancedb.graph.AsyncMaterializedVirtualPropertyGraph.refresh].""" + return LOOP.run(self._async.refresh()) diff --git a/python/python/lancedb/remote/db.py b/python/python/lancedb/remote/db.py index 54d155201..fba9d0a27 100644 --- a/python/python/lancedb/remote/db.py +++ b/python/python/lancedb/remote/db.py @@ -36,7 +36,18 @@ import pyarrow as pa from ..common import DATA from ..db import DBConnection, LOOP from ..functions import FunctionVersion, UdfDefinition -from ..graph import EdgeTable, EdgeType, NodeTable, NodeType, PropertyGraphDescription +from ..graph import ( + EdgeTable, + EdgeType, + MaterializedVirtualPropertyGraph, + MaterializedVirtualPropertyGraphDescription, + NodeTable, + NodeType, + PropertyGraph, + PropertyGraphDescription, + VirtualPropertyGraph, + VirtualPropertyGraphDescription, +) from ..job import AsyncJob, Job from ..sql import Query as SqlQuery from ..sql import QueryDescription @@ -947,14 +958,26 @@ class RemoteDBConnection(DBConnection): def create_property_graph( self, name: str, - nodes: List[Union[NodeTable, NodeType]], - edges: List[Union[EdgeTable, EdgeType]], + nodes: List[NodeType], + edges: List[EdgeType], *, namespace_path: Optional[List[str]] = None, - ) -> PropertyGraphDescription: - return LOOP.run( - self._conn.create_property_graph( - name, nodes, edges, namespace_path=namespace_path + ) -> PropertyGraph: + return PropertyGraph( + LOOP.run( + self._conn.create_property_graph( + name, nodes, edges, namespace_path=namespace_path + ) + ) + ) + + @override + def open_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> PropertyGraph: + return PropertyGraph( + LOOP.run( + self._conn.open_property_graph(name, namespace_path=namespace_path) ) ) @@ -967,35 +990,10 @@ class RemoteDBConnection(DBConnection): ) @override - def insert_into_property_graph( - self, - name: str, - label: str, - data: DATA, - *, - namespace_path: Optional[List[str]] = None, - ) -> PropertyGraphDescription: - return LOOP.run( - self._conn.insert_into_property_graph( - name, label, data, namespace_path=namespace_path - ) - ) - - @override - def refresh_property_graph( - self, name: str, *, namespace_path: Optional[List[str]] = None - ) -> PropertyGraphDescription: - return LOOP.run( - self._conn.refresh_property_graph(name, namespace_path=namespace_path) - ) - - @override - def rollback_property_graph( - self, name: str, *, namespace_path: Optional[List[str]] = None - ) -> PropertyGraphDescription: - return LOOP.run( - self._conn.rollback_property_graph(name, namespace_path=namespace_path) - ) + def list_property_graphs( + self, *, namespace_path: Optional[List[str]] = None + ) -> List[str]: + return LOOP.run(self._conn.list_property_graphs(namespace_path=namespace_path)) @override def drop_property_graph( @@ -1004,10 +1002,118 @@ class RemoteDBConnection(DBConnection): LOOP.run(self._conn.drop_property_graph(name, namespace_path=namespace_path)) @override - def list_property_graphs( + def create_virtual_property_graph( + self, + name: str, + nodes: List[NodeTable], + edges: List[EdgeTable], + *, + namespace_path: Optional[List[str]] = None, + ) -> VirtualPropertyGraph: + return VirtualPropertyGraph( + LOOP.run( + self._conn.create_virtual_property_graph( + name, nodes, edges, namespace_path=namespace_path + ) + ) + ) + + @override + def open_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> VirtualPropertyGraph: + return VirtualPropertyGraph( + LOOP.run( + self._conn.open_virtual_property_graph( + name, namespace_path=namespace_path + ) + ) + ) + + @override + def describe_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> VirtualPropertyGraphDescription: + return LOOP.run( + self._conn.describe_virtual_property_graph( + name, namespace_path=namespace_path + ) + ) + + @override + def list_virtual_property_graphs( self, *, namespace_path: Optional[List[str]] = None ) -> List[str]: - return LOOP.run(self._conn.list_property_graphs(namespace_path=namespace_path)) + return LOOP.run( + self._conn.list_virtual_property_graphs(namespace_path=namespace_path) + ) + + @override + def drop_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> None: + LOOP.run( + self._conn.drop_virtual_property_graph(name, namespace_path=namespace_path) + ) + + @override + def create_materialized_virtual_property_graph( + self, + name: str, + nodes: List[NodeTable], + edges: List[EdgeTable], + *, + namespace_path: Optional[List[str]] = None, + ) -> MaterializedVirtualPropertyGraph: + return MaterializedVirtualPropertyGraph( + LOOP.run( + self._conn.create_materialized_virtual_property_graph( + name, nodes, edges, namespace_path=namespace_path + ) + ) + ) + + @override + def open_materialized_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> MaterializedVirtualPropertyGraph: + return MaterializedVirtualPropertyGraph( + LOOP.run( + self._conn.open_materialized_virtual_property_graph( + name, namespace_path=namespace_path + ) + ) + ) + + @override + def describe_materialized_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> MaterializedVirtualPropertyGraphDescription: + return LOOP.run( + self._conn.describe_materialized_virtual_property_graph( + name, namespace_path=namespace_path + ) + ) + + @override + def list_materialized_virtual_property_graphs( + self, *, namespace_path: Optional[List[str]] = None + ) -> List[str]: + return LOOP.run( + self._conn.list_materialized_virtual_property_graphs( + namespace_path=namespace_path + ) + ) + + @override + def drop_materialized_virtual_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> None: + LOOP.run( + self._conn.drop_materialized_virtual_property_graph( + name, namespace_path=namespace_path + ) + ) @override def list_jobs(self) -> List["JobInfo"]: diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index e8ea07dd2..d366bd821 100644 --- a/python/python/tests/test_remote_db.py +++ b/python/python/tests/test_remote_db.py @@ -21,9 +21,13 @@ from lancedb.graph import ( EdgeTable, EdgeType, Endpoint, + MaterializedVirtualPropertyGraph, + MaterializedVirtualPropertyGraphDescription, NodeTable, NodeType, - PropertyGraphDescription, + PropertyGraph, + VirtualPropertyGraph, + VirtualPropertyGraphDescription, ) from lancedb.query import ColumnOrdering from lancedb.remote import ClientConfig @@ -2838,35 +2842,68 @@ def test_view_crud_addresses_its_own_routes(): ] -def test_property_graph_crud_addresses_its_own_routes(): - # The graph verbs are their own routes: the definition is the create body, - # and a description adds the graph's commits, its size and the table - # versions it was refreshed from. +def _graph_handler(calls, respond): + """A mock route handler recording `(method, path, body)` and answering + with `respond(path, body)` as JSON.""" + + def handler(request): + length = int(request.headers.get("Content-Length") or 0) + raw = request.rfile.read(length) if length else b"" + if request.path.split("?")[0].endswith("/insert"): + body = pa.ipc.open_stream(raw).read_all().to_pylist() + else: + body = json.loads(raw) if raw else None + calls.append((request.command, request.path, body)) + request.send_response(200) + request.send_header("Content-Type", "application/json") + request.end_headers() + request.wfile.write(json.dumps(respond(request.path, body)).encode()) + + return handler + + +SOCIAL_NODES = [ + NodeTable("person", key="person_id", label="Person", properties=["name"]) +] +SOCIAL_EDGES = [ + EdgeTable( + "knows", + label="KNOWS", + source=Endpoint("src_id", references=("person", "person_id")), + destination=Endpoint("dst_id", references=("person", "person_id")), + ) +] +SOCIAL_DEFINITION = { + "nodes": [ + { + "table": "person", + "key": "person_id", + "label": "Person", + "properties": ["name"], + } + ], + "edges": [ + { + "table": "knows", + "label": "KNOWS", + "source": { + "column": "src_id", + "references": {"table": "person", "column": "person_id"}, + }, + "destination": { + "column": "dst_id", + "references": {"table": "person", "column": "person_id"}, + }, + } + ], +} + + +def test_virtual_and_materialized_graphs_address_their_own_routes(): + # Each kind of graph has its own routes; the definition is the create + # body, and a materialized graph's description adds its last refresh. calls = [] - person = {"table": "person", "column": "person_id"} - definition = { - "mode": "materialized_view", - "nodes": [ - { - "table": "person", - "key": "person_id", - "label": "Person", - "properties": ["name"], - } - ], - "edges": [ - { - "table": "knows", - "label": "KNOWS", - "source": {"column": "src_id", "references": person}, - "destination": {"column": "dst_id", "references": person}, - } - ], - } - described = { - "name": "social", - "namespace": ["analytics"], - **definition, + refreshed = { "commit": "6f1c2d3e-0000-4000-8000-000000000000", "committed_at": "2026-09-26T17:00:00Z", "vertex_count": 4, @@ -2877,62 +2914,106 @@ def test_property_graph_crud_addresses_its_own_routes(): ], } - def handler(request): - length = int(request.headers.get("Content-Length") or 0) - body = json.loads(request.rfile.read(length)) if length else None - calls.append((request.command, request.path, body)) - if request.path.endswith("/property_graph/list"): - response = {"property_graphs": ["social"], "page_token": None} - elif request.path.endswith("/drop"): - response = {} - else: - response = described - request.send_response(200) - request.send_header("Content-Type", "application/json") - request.end_headers() - request.wfile.write(json.dumps(response).encode()) + def respond(path, body): + if path.endswith("/virtual_property_graph/list"): + return {"virtual_property_graphs": ["social"], "page_token": None} + if path.endswith("/materialized_virtual_property_graph/list"): + return {"materialized_virtual_property_graphs": ["social_mv"]} + if path.endswith("/drop"): + return {} + name = path.split("$")[1].split("/")[0] + described = {"name": name, "namespace": ["analytics"], **SOCIAL_DEFINITION} + if "/materialized_virtual_property_graph/" in path: + described.update(refreshed) + return described - nodes = [NodeTable("person", key="person_id", label="Person", properties=["name"])] - edges = [ - EdgeTable( - "knows", - label="KNOWS", - source=Endpoint("src_id", references=("person", "person_id")), - destination=Endpoint("dst_id", references=("person", "person_id")), + with mock_lancedb_connection(_graph_handler(calls, respond)) as db: + graph = db.create_virtual_property_graph( + "social", SOCIAL_NODES, SOCIAL_EDGES, namespace_path=["analytics"] ) - ] - with mock_lancedb_connection(handler) as db: - graph = db.create_property_graph( - "social", nodes, edges, namespace_path=["analytics"] - ) - assert graph == PropertyGraphDescription( + assert isinstance(graph, VirtualPropertyGraph) + assert (graph.name, graph.namespace_path) == ("social", ["analytics"]) + assert graph.describe() == VirtualPropertyGraphDescription( name="social", namespace_path=["analytics"], - mode="materialized_view", - nodes=nodes, - edges=edges, + nodes=SOCIAL_NODES, + edges=SOCIAL_EDGES, + ) + assert db.list_virtual_property_graphs(namespace_path=["analytics"]) == [ + "social" + ] + db.drop_virtual_property_graph("social", namespace_path=["analytics"]) + + mvpg = db.create_materialized_virtual_property_graph( + "social_mv", SOCIAL_NODES, SOCIAL_EDGES, namespace_path=["analytics"] + ) + assert isinstance(mvpg, MaterializedVirtualPropertyGraph) + expected = MaterializedVirtualPropertyGraphDescription( + name="social_mv", + namespace_path=["analytics"], + nodes=SOCIAL_NODES, + edges=SOCIAL_EDGES, commit="6f1c2d3e-0000-4000-8000-000000000000", committed_at="2026-09-26T17:00:00Z", - previous_commit=None, vertex_count=4, edge_count=5, sources={"person": 3, "knows": 2}, ) - assert db.describe_property_graph("social", namespace_path=["analytics"]) == ( - graph + assert mvpg.refresh() == expected + opened = db.open_materialized_virtual_property_graph( + "social_mv", namespace_path=["analytics"] + ) + assert opened.describe() == expected + assert db.list_materialized_virtual_property_graphs( + namespace_path=["analytics"] + ) == ["social_mv"] + db.drop_materialized_virtual_property_graph( + "social_mv", namespace_path=["analytics"] ) - assert db.list_property_graphs(namespace_path=["analytics"]) == ["social"] - db.drop_property_graph("social", namespace_path=["analytics"]) assert calls == [ - ("POST", "/v1/property_graph/analytics$social/create", definition), - ("POST", "/v1/property_graph/analytics$social/describe", None), - ("GET", "/v1/namespace/analytics/property_graph/list", None), - ("POST", "/v1/property_graph/analytics$social/drop", None), + ( + "POST", + "/v1/virtual_property_graph/analytics$social/create", + SOCIAL_DEFINITION, + ), + ("POST", "/v1/virtual_property_graph/analytics$social/describe", None), + ("GET", "/v1/namespace/analytics/virtual_property_graph/list", None), + ("POST", "/v1/virtual_property_graph/analytics$social/drop", None), + ( + "POST", + "/v1/materialized_virtual_property_graph/analytics$social_mv/create", + SOCIAL_DEFINITION, + ), + ( + "POST", + "/v1/materialized_virtual_property_graph/analytics$social_mv/refresh", + None, + ), + ( + "POST", + "/v1/materialized_virtual_property_graph/analytics$social_mv/describe", + None, + ), + ( + "POST", + "/v1/materialized_virtual_property_graph/analytics$social_mv/describe", + None, + ), + ( + "GET", + "/v1/namespace/analytics/materialized_virtual_property_graph/list", + None, + ), + ( + "POST", + "/v1/materialized_virtual_property_graph/analytics$social_mv/drop", + None, + ), ] -def test_an_independent_graph_carries_schemas_and_takes_arrow_rows(): +def test_a_property_graph_carries_schemas_and_takes_arrow_rows(): # Node and edge types go up as JSON Arrow schemas and come back as pyarrow # schemas; inserted rows go up as an Arrow IPC stream. calls = [] @@ -2954,17 +3035,12 @@ def test_an_independent_graph_carries_schemas_and_takes_arrow_rows(): ) created = {} - def handler(request): - length = int(request.headers.get("Content-Length") or 0) - raw = request.rfile.read(length) if length else b"" - if request.path.split("?")[0].endswith("/insert"): - body = pa.ipc.open_stream(raw).read_all().to_pylist() - else: - body = json.loads(raw) if raw else None - calls.append((request.command, request.path, body)) - if request.path.endswith("/create"): + def respond(path, body): + if path.endswith("/create"): created.update(body) - response = { + if path.endswith("/property_graph/list"): + return {"property_graphs": ["people"]} + return { "name": "people", "namespace": [], **created, @@ -2974,52 +3050,46 @@ def test_an_independent_graph_carries_schemas_and_takes_arrow_rows(): "vertex_count": 1, "edge_count": 0, } - request.send_response(200) - request.send_header("Content-Type", "application/json") - request.end_headers() - request.wfile.write(json.dumps(response).encode()) - with mock_lancedb_connection(handler) as db: + with mock_lancedb_connection(_graph_handler(calls, respond)) as db: graph = db.create_property_graph("people", [person], [knows]) - assert graph.mode == "independent" - assert graph.nodes == [person] and graph.edges == [knows] - assert graph.previous_commit == "c1" - inserted = db.insert_into_property_graph( - "people", "Person", [{"person_id": 1, "tags": ["a"]}] - ) + assert isinstance(graph, PropertyGraph) + described = graph.describe() + assert described.nodes == [person] and described.edges == [knows] + assert described.previous_commit == "c1" + inserted = graph.insert("Person", [{"person_id": 1, "tags": ["a"]}]) assert inserted.vertex_count == 1 - db.refresh_property_graph("people") - db.rollback_property_graph("people") + graph.rollback() + assert db.list_property_graphs() == ["people"] + db.drop_property_graph("people") - create, insert, refresh, rollback = calls + create, describe, insert, rollback, listed, drop = calls assert create[:2] == ("POST", "/v1/property_graph/people/create") - assert create[2]["mode"] == "independent" + assert "mode" not in create[2] assert create[2]["nodes"][0]["schema"]["fields"][0]["name"] == "person_id" assert create[2]["edges"][0]["source"] == {"label": "Person", "column": "src_id"} + assert describe[:2] == ("POST", "/v1/property_graph/people/describe") assert insert == ( "POST", "/v1/property_graph/people/insert?label=Person", [{"person_id": 1, "tags": ["a"]}], ) - assert refresh[:2] == ("POST", "/v1/property_graph/people/refresh") assert rollback[:2] == ("POST", "/v1/property_graph/people/rollback") + assert listed[:2] == ("GET", "/v1/namespace/$/property_graph/list") + assert drop[:2] == ("POST", "/v1/property_graph/people/drop") - with pytest.raises(ValueError, match="either all tables"): - with mock_lancedb_connection(handler) as db: - db.create_property_graph( - "mixed", - [person], - [ - EdgeTable( - "knows", - label="KNOWS", - source=Endpoint("src_id", references=("person", "person_id")), - destination=Endpoint( - "dst_id", references=("person", "person_id") - ), - ) - ], - ) + +def test_a_graph_takes_only_its_own_kind_of_definition(): + person = NodeType( + "Person", + key="person_id", + schema=pa.schema([pa.field("person_id", pa.int64(), nullable=False)]), + ) + with mock_lancedb_connection(_graph_handler([], lambda path, body: {})) as db: + with pytest.raises(TypeError, match="are NodeTable and its edges EdgeTable"): + db.create_virtual_property_graph("mixed", [person], SOCIAL_EDGES) + with pytest.raises(TypeError, match="NodeType"): + db.create_property_graph("mixed", SOCIAL_NODES, []) def test_local_connections_refuse_property_graphs(tmp_path): @@ -3027,4 +3097,6 @@ def test_local_connections_refuse_property_graphs(tmp_path): with pytest.raises(NotImplementedError, match="Property graph operations"): db.list_property_graphs() with pytest.raises(NotImplementedError, match="Property graph operations"): - db.describe_property_graph("social") + db.describe_virtual_property_graph("social") + with pytest.raises(NotImplementedError, match="Property graph operations"): + db.open_materialized_virtual_property_graph("social") diff --git a/python/src/connection.rs b/python/src/connection.rs index f5af99ef6..2f0f7b737 100644 --- a/python/src/connection.rs +++ b/python/src/connection.rs @@ -9,6 +9,7 @@ use std::{ use crate::{ error::PythonErrorExt, + graph::GraphKind, namespace::{create_namespace_storage_options_provider, extract_namespace_arc}, runtime::future_into_py, table::{Table, scannable::PyScannable}, @@ -965,73 +966,144 @@ impl Connection { }) } - #[pyo3(signature = (name, definition_json, namespace_path=None))] - pub fn create_property_graph( + /// Create a graph of `kind` from its definition's JSON, and return its + /// description's JSON. + #[pyo3(signature = (kind, name, definition_json, namespace_path=None))] + pub fn create_graph( self_: PyRef<'_, Self>, + kind: String, name: String, definition_json: String, namespace_path: Option>, ) -> PyResult> { let inner = self_.get_inner()?.clone(); let namespace_path = namespace_path.unwrap_or_default(); - let definition = - lancedb::graph::PropertyGraphDefinition::from_json(&definition_json).infer_error()?; + let kind = GraphKind::parse(&kind)?; future_into_py(self_.py(), async move { - inner - .create_property_graph(name, &definition, &namespace_path) - .await - .infer_error()? - .to_json() - .infer_error() + match kind { + GraphKind::Property => { + let definition = + lancedb::graph::PropertyGraphDefinition::from_json(&definition_json) + .infer_error()?; + inner + .create_property_graph(name, &definition, &namespace_path) + .await + .infer_error()? + .to_json() + } + GraphKind::Virtual => { + let definition = + lancedb::graph::VirtualPropertyGraphDefinition::from_json(&definition_json) + .infer_error()?; + inner + .create_virtual_property_graph(name, &definition, &namespace_path) + .await + .infer_error()? + .to_json() + } + GraphKind::MaterializedVirtual => { + let definition = + lancedb::graph::VirtualPropertyGraphDefinition::from_json(&definition_json) + .infer_error()?; + inner + .create_materialized_virtual_property_graph( + name, + &definition, + &namespace_path, + ) + .await + .infer_error()? + .to_json() + } + } + .infer_error() }) } - #[pyo3(signature = (name, namespace_path=None))] - pub fn describe_property_graph( + /// The JSON description of the graph of `kind` named `name`. + #[pyo3(signature = (kind, name, namespace_path=None))] + pub fn describe_graph( self_: PyRef<'_, Self>, + kind: String, name: String, namespace_path: Option>, ) -> PyResult> { let inner = self_.get_inner()?.clone(); let namespace_path = namespace_path.unwrap_or_default(); + let kind = GraphKind::parse(&kind)?; future_into_py(self_.py(), async move { - inner - .describe_property_graph(name, &namespace_path) - .await - .infer_error()? - .to_json() - .infer_error() + match kind { + GraphKind::Property => inner + .describe_property_graph(name, &namespace_path) + .await + .infer_error()? + .to_json(), + GraphKind::Virtual => inner + .describe_virtual_property_graph(name, &namespace_path) + .await + .infer_error()? + .to_json(), + GraphKind::MaterializedVirtual => inner + .describe_materialized_virtual_property_graph(name, &namespace_path) + .await + .infer_error()? + .to_json(), + } + .infer_error() }) } - #[pyo3(signature = (name, namespace_path=None))] - pub fn drop_property_graph( + /// Drop the graph of `kind` named `name`, waiting for its storage to be + /// deleted. + #[pyo3(signature = (kind, name, namespace_path=None))] + pub fn drop_graph( self_: PyRef<'_, Self>, + kind: String, name: String, namespace_path: Option>, ) -> PyResult> { let inner = self_.get_inner()?.clone(); let namespace_path = namespace_path.unwrap_or_default(); + let kind = GraphKind::parse(&kind)?; future_into_py(self_.py(), async move { - inner - .drop_property_graph(name, &namespace_path) - .await - .infer_error() + match kind { + GraphKind::Property => inner.drop_property_graph(name, &namespace_path).await, + GraphKind::Virtual => { + inner + .drop_virtual_property_graph(name, &namespace_path) + .await + } + GraphKind::MaterializedVirtual => { + inner + .drop_materialized_virtual_property_graph(name, &namespace_path) + .await + } + } + .infer_error() }) } - #[pyo3(signature = (namespace_path=None))] - pub fn list_property_graphs( + /// The names of the graphs of `kind` in one namespace. + #[pyo3(signature = (kind, namespace_path=None))] + pub fn list_graphs( self_: PyRef<'_, Self>, + kind: String, namespace_path: Option>, ) -> PyResult> { let inner = self_.get_inner()?.clone(); let namespace_path = namespace_path.unwrap_or_default(); + let kind = GraphKind::parse(&kind)?; future_into_py(self_.py(), async move { - inner - .list_property_graphs(&namespace_path) - .await - .infer_error() + match kind { + GraphKind::Property => inner.list_property_graphs(&namespace_path).await, + GraphKind::Virtual => inner.list_virtual_property_graphs(&namespace_path).await, + GraphKind::MaterializedVirtual => { + inner + .list_materialized_virtual_property_graphs(&namespace_path) + .await + } + } + .infer_error() }) } @@ -1055,24 +1127,6 @@ impl Connection { }) } - #[pyo3(signature = (name, namespace_path=None))] - pub fn refresh_property_graph( - self_: PyRef<'_, Self>, - name: String, - namespace_path: Option>, - ) -> PyResult> { - let inner = self_.get_inner()?.clone(); - let namespace_path = namespace_path.unwrap_or_default(); - future_into_py(self_.py(), async move { - inner - .refresh_property_graph(name, &namespace_path) - .await - .infer_error()? - .to_json() - .infer_error() - }) - } - #[pyo3(signature = (name, namespace_path=None))] pub fn rollback_property_graph( self_: PyRef<'_, Self>, @@ -1091,6 +1145,24 @@ impl Connection { }) } + #[pyo3(signature = (name, namespace_path=None))] + pub fn refresh_materialized_virtual_property_graph( + self_: PyRef<'_, Self>, + name: String, + namespace_path: Option>, + ) -> PyResult> { + let inner = self_.get_inner()?.clone(); + let namespace_path = namespace_path.unwrap_or_default(); + future_into_py(self_.py(), async move { + inner + .refresh_materialized_virtual_property_graph(name, &namespace_path) + .await + .infer_error()? + .to_json() + .infer_error() + }) + } + pub fn list_jobs(self_: PyRef<'_, Self>) -> PyResult> { let inner = self_.get_inner()?.clone(); future_into_py(self_.py(), async move { diff --git a/python/src/graph.rs b/python/src/graph.rs index 29df02ef6..9e81889a1 100644 --- a/python/src/graph.rs +++ b/python/src/graph.rs @@ -8,6 +8,27 @@ use arrow::{datatypes::Schema, pyarrow::PyArrowType}; use lance_namespace::models::JsonArrowSchema; use pyo3::{PyResult, exceptions::PyValueError, pyfunction}; +/// A kind of graph, spelled as its routes and catalog objects are. +#[derive(Debug, Clone, Copy)] +pub enum GraphKind { + Property, + Virtual, + MaterializedVirtual, +} + +impl GraphKind { + pub fn parse(kind: &str) -> PyResult { + match kind { + "property_graph" => Ok(Self::Property), + "virtual_property_graph" => Ok(Self::Virtual), + "materialized_virtual_property_graph" => Ok(Self::MaterializedVirtual), + other => Err(PyValueError::new_err(format!( + "'{other}' is not a kind of property graph" + ))), + } + } +} + /// A pyarrow schema as the JSON Arrow schema of a node or edge type. #[pyfunction] pub fn graph_schema_to_json(schema: PyArrowType) -> PyResult { diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs index d7bced436..49016db15 100644 --- a/rust/lancedb/src/connection.rs +++ b/rust/lancedb/src/connection.rs @@ -28,7 +28,10 @@ use crate::database::{ }; use crate::embeddings::{EmbeddingRegistry, MemoryRegistry}; use crate::error::{Error, Result}; -use crate::graph::{PropertyGraphDefinition, PropertyGraphDescription}; +use crate::graph::{ + MaterializedVirtualPropertyGraphDescription, PropertyGraphDefinition, PropertyGraphDescription, + VirtualPropertyGraphDefinition, VirtualPropertyGraphDescription, +}; #[cfg(feature = "remote")] use crate::remote::{ client::ClientConfig, @@ -867,40 +870,26 @@ impl Connection { self.internal.list_views(namespace_path).await } - /// Create a property graph. - /// - /// A materialized view graph names the tables in its namespace that hold - /// its nodes, with their keys and labels, and its edges, with the node keys - /// their source and destination columns reference; it is built from them - /// now. An independent graph names its node and edge types with their - /// schemas, and starts empty. A name already taken is an error. Local - /// databases return [`Error::NotSupported`]. + /// Create a property graph: a graph that holds its own rows, typed by its + /// node and edge types, and starts empty. A name held by a graph of any + /// kind is taken. Local databases return [`Error::NotSupported`]. /// /// ```no_run - /// # use lancedb::graph::{EdgeTable, Endpoint, EndpointReference, NodeTable, - /// # PropertyGraphDefinition}; + /// # use arrow_schema::{DataType, Field, Schema}; + /// # use lancedb::graph::{EdgeType, EndpointType, NodeType, PropertyGraphDefinition}; /// # async fn example(connection: &lancedb::Connection) -> lancedb::Result<()> { - /// let person = EndpointReference { - /// table: "person".to_string(), - /// column: "person_id".to_string(), + /// let person = NodeType::new( + /// "Person", + /// "person_id", + /// &Schema::new(vec![Field::new("person_id", DataType::Int64, false)]), + /// )?; + /// let endpoint = |column: &str| EndpointType { + /// label: "Person".to_string(), + /// column: column.to_string(), /// }; - /// let definition = PropertyGraphDefinition::MaterializedView { - /// nodes: vec![NodeTable { - /// table: "person".to_string(), - /// key: "person_id".to_string(), - /// label: "Person".to_string(), - /// properties: None, - /// }], - /// edges: vec![EdgeTable { - /// table: "knows".to_string(), - /// label: "KNOWS".to_string(), - /// source: Endpoint { column: "src_id".to_string(), references: person.clone() }, - /// destination: Endpoint { column: "dst_id".to_string(), references: person }, - /// properties: None, - /// }], - /// }; - /// let graph = connection.create_property_graph("social", &definition, &[]).await?; - /// println!("{} nodes, {} edges", graph.vertex_count, graph.edge_count); + /// let knows = EdgeType::new("KNOWS", endpoint("src_id"), endpoint("dst_id"), None)?; + /// let definition = PropertyGraphDefinition { nodes: vec![person], edges: vec![knows] }; + /// connection.create_property_graph("people", &definition, &[]).await?; /// # Ok(()) /// # } /// ``` @@ -916,10 +905,9 @@ impl Connection { .await } - /// What this database records about one property graph: its mode and - /// definition, its current and previous commits, its size, and, for a - /// materialized view, the table versions it was refreshed from. Local - /// databases return [`Error::NotSupported`]. + /// What this database records about one property graph: its node and edge + /// types, its current and previous commits, and its size. Local databases + /// return [`Error::NotSupported`]. pub async fn describe_property_graph( &self, name: impl AsRef, @@ -931,10 +919,8 @@ impl Connection { .await } - /// Drop a property graph and wait for its data to be deleted. - /// - /// The tables it reads are untouched. Local databases return - /// [`Error::NotSupported`]. + /// Drop a property graph and wait for its data to be deleted. Local + /// databases return [`Error::NotSupported`]. pub async fn drop_property_graph( &self, name: impl AsRef, @@ -955,15 +941,13 @@ impl Connection { self.internal.list_property_graphs(namespace_path).await } - /// Insert rows into an independent property graph, as one commit. + /// Insert rows into a property graph, as one commit. /// /// The rows are one node label's or one edge label's. Nodes upsert by key: /// a key the graph holds replaces that node's properties. Edges append, /// carrying their endpoints' keys in the columns the edge type names; an /// edge whose endpoint is not a node of the graph refuses the whole - /// insert. A materialized view graph is written through its tables and - /// [refreshed](Self::refresh_property_graph) instead. Local databases - /// return [`Error::NotSupported`]. + /// insert. Local databases return [`Error::NotSupported`]. pub async fn insert_into_property_graph( &self, name: impl AsRef, @@ -982,23 +966,6 @@ impl Connection { .await } - /// Bring a materialized view property graph up to its tables' latest - /// versions. - /// - /// Until it is refreshed, a query over a table that has changed reads the - /// tables rather than the graph. Local databases return - /// [`Error::NotSupported`]. - pub async fn refresh_property_graph( - &self, - name: impl AsRef, - namespace_path: &[String], - ) -> Result { - validate_property_graph_reference(name.as_ref(), namespace_path)?; - self.internal - .refresh_property_graph(name.as_ref(), namespace_path) - .await - } - /// Return a property graph to its previous commit. /// /// The commit rolled back from is discarded, so a second rollback in a row @@ -1014,6 +981,166 @@ impl Connection { .await } + /// Create a virtual property graph over tables in its namespace. + /// + /// The definition names the tables that hold its nodes, with their keys + /// and labels, and its edges, with the node keys their source and + /// destination columns reference. The graph stores nothing else: a query + /// reads the tables as they are then. A name held by a graph of any kind + /// is taken. Local databases return [`Error::NotSupported`]. + /// + /// ```no_run + /// # use lancedb::graph::{EdgeTable, Endpoint, EndpointReference, NodeTable, + /// # VirtualPropertyGraphDefinition}; + /// # async fn example(connection: &lancedb::Connection) -> lancedb::Result<()> { + /// let person = EndpointReference { + /// table: "person".to_string(), + /// column: "person_id".to_string(), + /// }; + /// let definition = VirtualPropertyGraphDefinition { + /// nodes: vec![NodeTable { + /// table: "person".to_string(), + /// key: "person_id".to_string(), + /// label: "Person".to_string(), + /// properties: None, + /// }], + /// edges: vec![EdgeTable { + /// table: "knows".to_string(), + /// label: "KNOWS".to_string(), + /// source: Endpoint { column: "src_id".to_string(), references: person.clone() }, + /// destination: Endpoint { column: "dst_id".to_string(), references: person }, + /// properties: None, + /// }], + /// }; + /// connection.create_virtual_property_graph("social", &definition, &[]).await?; + /// # Ok(()) + /// # } + /// ``` + pub async fn create_virtual_property_graph( + &self, + name: impl AsRef, + definition: &VirtualPropertyGraphDefinition, + namespace_path: &[String], + ) -> Result { + validate_property_graph_reference(name.as_ref(), namespace_path)?; + self.internal + .create_virtual_property_graph(name.as_ref(), definition, namespace_path) + .await + } + + /// What this database records about one virtual property graph: its node + /// and edge tables. Local databases return [`Error::NotSupported`]. + pub async fn describe_virtual_property_graph( + &self, + name: impl AsRef, + namespace_path: &[String], + ) -> Result { + validate_property_graph_reference(name.as_ref(), namespace_path)?; + self.internal + .describe_virtual_property_graph(name.as_ref(), namespace_path) + .await + } + + /// Drop a virtual property graph and wait for its definition to be + /// deleted. The tables it reads are untouched. Local databases return + /// [`Error::NotSupported`]. + pub async fn drop_virtual_property_graph( + &self, + name: impl AsRef, + namespace_path: &[String], + ) -> Result<()> { + validate_property_graph_reference(name.as_ref(), namespace_path)?; + self.internal + .drop_virtual_property_graph(name.as_ref(), namespace_path) + .await + } + + /// The names of the virtual property graphs in one namespace. Local + /// databases return [`Error::NotSupported`]. + pub async fn list_virtual_property_graphs( + &self, + namespace_path: &[String], + ) -> Result> { + validate_namespace(namespace_path)?; + self.internal + .list_virtual_property_graphs(namespace_path) + .await + } + + /// Create a materialized virtual property graph: a virtual property + /// graph's definition, built now from the tables' latest versions and + /// read as that build until + /// [refreshed](Self::refresh_materialized_virtual_property_graph). Local + /// databases return [`Error::NotSupported`]. + pub async fn create_materialized_virtual_property_graph( + &self, + name: impl AsRef, + definition: &VirtualPropertyGraphDefinition, + namespace_path: &[String], + ) -> Result { + validate_property_graph_reference(name.as_ref(), namespace_path)?; + self.internal + .create_materialized_virtual_property_graph(name.as_ref(), definition, namespace_path) + .await + } + + /// What this database records about one materialized virtual property + /// graph: its node and edge tables, and what its last refresh built from + /// them. Local databases return [`Error::NotSupported`]. + pub async fn describe_materialized_virtual_property_graph( + &self, + name: impl AsRef, + namespace_path: &[String], + ) -> Result { + validate_property_graph_reference(name.as_ref(), namespace_path)?; + self.internal + .describe_materialized_virtual_property_graph(name.as_ref(), namespace_path) + .await + } + + /// Drop a materialized virtual property graph and wait for its data to be + /// deleted. The tables it reads are untouched. Local databases return + /// [`Error::NotSupported`]. + pub async fn drop_materialized_virtual_property_graph( + &self, + name: impl AsRef, + namespace_path: &[String], + ) -> Result<()> { + validate_property_graph_reference(name.as_ref(), namespace_path)?; + self.internal + .drop_materialized_virtual_property_graph(name.as_ref(), namespace_path) + .await + } + + /// The names of the materialized virtual property graphs in one + /// namespace. Local databases return [`Error::NotSupported`]. + pub async fn list_materialized_virtual_property_graphs( + &self, + namespace_path: &[String], + ) -> Result> { + validate_namespace(namespace_path)?; + self.internal + .list_materialized_virtual_property_graphs(namespace_path) + .await + } + + /// Rebuild a materialized virtual property graph from its tables' latest + /// versions. + /// + /// When every table's commits since the last refresh left its rows as + /// they were, only the recorded versions advance. Local databases return + /// [`Error::NotSupported`]. + pub async fn refresh_materialized_virtual_property_graph( + &self, + name: impl AsRef, + namespace_path: &[String], + ) -> Result { + validate_property_graph_reference(name.as_ref(), namespace_path)?; + self.internal + .refresh_materialized_virtual_property_graph(name.as_ref(), namespace_path) + .await + } + /// Rename a table in the database. /// /// This is only supported in LanceDB Cloud. diff --git a/rust/lancedb/src/database.rs b/rust/lancedb/src/database.rs index de76fb367..1e718c71c 100644 --- a/rust/lancedb/src/database.rs +++ b/rust/lancedb/src/database.rs @@ -28,7 +28,10 @@ use lance_namespace::models::{ use crate::data::scannable::Scannable; use crate::error::Result; -use crate::graph::{PropertyGraphDefinition, PropertyGraphDescription}; +use crate::graph::{ + MaterializedVirtualPropertyGraphDescription, PropertyGraphDefinition, PropertyGraphDescription, + VirtualPropertyGraphDefinition, VirtualPropertyGraphDescription, +}; use crate::job::Job; use crate::materialized_view::CreateMaterializedViewRequest; use crate::secrets::SecretInfo; @@ -514,7 +517,7 @@ pub trait Database: async fn list_views(&self, _namespace_path: &[String]) -> Result> { view_ops_not_supported() } - /// Create a property graph over tables in its namespace. + /// Create a property graph, empty. async fn create_property_graph( &self, _name: &str, @@ -531,8 +534,7 @@ pub trait Database: ) -> Result { property_graph_ops_not_supported() } - /// Drop a property graph and wait for its data to be deleted. The tables it - /// reads are untouched. + /// Drop a property graph and wait for its data to be deleted. async fn drop_property_graph(&self, _name: &str, _namespace_path: &[String]) -> Result<()> { property_graph_ops_not_supported() } @@ -550,8 +552,8 @@ pub trait Database: async fn list_property_graphs(&self, _namespace_path: &[String]) -> Result> { property_graph_ops_not_supported() } - /// Insert rows of one node label or edge label into an independent - /// property graph, as one commit. + /// Insert rows of one node label or edge label into a property graph, as + /// one commit. async fn insert_into_property_graph( &self, _name: &str, @@ -561,14 +563,6 @@ pub trait Database: ) -> Result { property_graph_ops_not_supported() } - /// Bring a materialized view property graph up to its tables. - async fn refresh_property_graph( - &self, - _name: &str, - _namespace_path: &[String], - ) -> Result { - property_graph_ops_not_supported() - } /// Make a property graph's previous commit current again. async fn rollback_property_graph( &self, @@ -577,6 +571,103 @@ pub trait Database: ) -> Result { property_graph_ops_not_supported() } + /// Create a virtual property graph over tables in its namespace. + async fn create_virtual_property_graph( + &self, + _name: &str, + _definition: &VirtualPropertyGraphDefinition, + _namespace_path: &[String], + ) -> Result { + property_graph_ops_not_supported() + } + /// What the database records about one virtual property graph. + async fn describe_virtual_property_graph( + &self, + _name: &str, + _namespace_path: &[String], + ) -> Result { + property_graph_ops_not_supported() + } + /// Drop a virtual property graph and wait for its definition to be + /// deleted. The tables it reads are untouched. + async fn drop_virtual_property_graph( + &self, + _name: &str, + _namespace_path: &[String], + ) -> Result<()> { + property_graph_ops_not_supported() + } + /// Drop a virtual property graph and return the job deleting its + /// definition, without waiting. + #[doc(hidden)] + async fn drop_virtual_property_graph_async( + &self, + _name: &str, + _namespace_path: &[String], + ) -> Result { + property_graph_ops_not_supported() + } + /// The names of the virtual property graphs in one namespace. + async fn list_virtual_property_graphs( + &self, + _namespace_path: &[String], + ) -> Result> { + property_graph_ops_not_supported() + } + /// Create a materialized virtual property graph over tables in its + /// namespace, built from them now. + async fn create_materialized_virtual_property_graph( + &self, + _name: &str, + _definition: &VirtualPropertyGraphDefinition, + _namespace_path: &[String], + ) -> Result { + property_graph_ops_not_supported() + } + /// What the database records about one materialized virtual property + /// graph. + async fn describe_materialized_virtual_property_graph( + &self, + _name: &str, + _namespace_path: &[String], + ) -> Result { + property_graph_ops_not_supported() + } + /// Drop a materialized virtual property graph and wait for its data to be + /// deleted. The tables it reads are untouched. + async fn drop_materialized_virtual_property_graph( + &self, + _name: &str, + _namespace_path: &[String], + ) -> Result<()> { + property_graph_ops_not_supported() + } + /// Drop a materialized virtual property graph and return the job deleting + /// its data, without waiting. + #[doc(hidden)] + async fn drop_materialized_virtual_property_graph_async( + &self, + _name: &str, + _namespace_path: &[String], + ) -> Result { + property_graph_ops_not_supported() + } + /// The names of the materialized virtual property graphs in one namespace. + async fn list_materialized_virtual_property_graphs( + &self, + _namespace_path: &[String], + ) -> Result> { + property_graph_ops_not_supported() + } + /// Rebuild a materialized virtual property graph from its tables' latest + /// versions. + async fn refresh_materialized_virtual_property_graph( + &self, + _name: &str, + _namespace_path: &[String], + ) -> Result { + property_graph_ops_not_supported() + } /// Open a job by id, returning a handle with its record already /// populated. Fails with [`crate::Error::JobNotFound`] when the server has /// no such job. diff --git a/rust/lancedb/src/graph.rs b/rust/lancedb/src/graph.rs index 64901bc24..b811ac404 100644 --- a/rust/lancedb/src/graph.rs +++ b/rust/lancedb/src/graph.rs @@ -1,15 +1,18 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright The LanceDB Authors -//! Property graphs: catalog objects read as graphs, in one of two modes. +//! Property graphs: catalog objects read as graphs, in three kinds that share +//! one namespace of names. //! -//! - Independent: the graph holds its own rows. Its node and edge types have +//! - A property graph holds its own rows. Its node and edge types have //! schemas, and rows are inserted into the graph itself. -//! - Materialized view: the graph reads tables in its namespace -- which hold +//! - A virtual property graph reads tables in its namespace -- which hold //! nodes and which hold edges, each node table's key and label, each edge //! table's label and how its source and destination columns reference node -//! keys, the shape of SQL/PGQ's `CREATE PROPERTY GRAPH` -- and is refreshed -//! from them. The tables stay ordinary tables. +//! keys, the shape of SQL/PGQ's `CREATE PROPERTY GRAPH` -- as they are when +//! a query reads them. +//! - A materialized virtual property graph has a virtual graph's definition, +//! and is read as its last refresh from the tables built it. //! //! The verbs live on [`crate::connection::Connection`]. @@ -19,38 +22,54 @@ use serde::{Deserialize, Serialize}; use crate::error::{Error, Result}; -/// A graph's definition, tagged with its mode. +fn from_json(json: &str) -> Result { + serde_json::from_str(json).map_err(|source| Error::InvalidInput { + message: format!("invalid property graph definition: {source}"), + }) +} + +fn to_json(value: &impl Serialize) -> Result { + serde_json::to_string(value).map_err(|source| Error::Runtime { + message: format!("could not encode a property graph description: {source}"), + }) +} + +/// A property graph's node and edge types. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(tag = "mode", rename_all = "snake_case")] -pub enum PropertyGraphDefinition { - /// The graph holds its own rows, inserted into it. - Independent { - /// The node types, one per label. - nodes: Vec, - /// The edge types, one per label. - #[serde(default)] - edges: Vec, - }, - /// The graph reads tables and is refreshed from them. - MaterializedView { - /// The node tables, one per label. - nodes: Vec, - /// The edge tables, one per label. - #[serde(default)] - edges: Vec, - }, +pub struct PropertyGraphDefinition { + /// The node types, one per label. + pub nodes: Vec, + /// The edge types, one per label. + #[serde(default)] + pub edges: Vec, } impl PropertyGraphDefinition { /// Parse a definition from its JSON form. pub fn from_json(json: &str) -> Result { - serde_json::from_str(json).map_err(|source| Error::InvalidInput { - message: format!("invalid property graph definition: {source}"), - }) + from_json(json) } } -/// The nodes of one label of an independent graph. +/// The node and edge tables of a virtual property graph, which a materialized +/// virtual property graph shares. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct VirtualPropertyGraphDefinition { + /// The node tables, one per label. + pub nodes: Vec, + /// The edge tables, one per label. + #[serde(default)] + pub edges: Vec, +} + +impl VirtualPropertyGraphDefinition { + /// Parse a definition from its JSON form. + pub fn from_json(json: &str) -> Result { + from_json(json) + } +} + +/// The nodes of one label of a property graph. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct NodeType { /// The label its nodes carry. @@ -79,7 +98,7 @@ impl NodeType { } } -/// The edges of one label of an independent graph. +/// The edges of one label of a property graph. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct EdgeType { /// The label its edges carry. @@ -180,7 +199,7 @@ pub struct EndpointReference { pub column: String, } -/// The version of one table a graph's current commit was refreshed from. +/// The version of one table a materialized graph's last refresh read. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct GraphSourceVersion { /// The table. @@ -192,7 +211,7 @@ pub struct GraphSourceVersion { /// What a database records about one property graph. /// /// Returned by [`crate::connection::Connection::describe_property_graph`], and -/// by every call that writes a graph. +/// by every call that writes one. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct PropertyGraphDescription { /// The graph's name within its namespace. @@ -200,7 +219,7 @@ pub struct PropertyGraphDescription { /// The namespace holding the graph; empty is the root namespace. #[serde(rename = "namespace", default)] pub namespace_path: Vec, - /// The graph's mode and definition. + /// The graph's node and edge types. #[serde(flatten)] pub definition: PropertyGraphDefinition, /// The graph's current commit. @@ -214,18 +233,65 @@ pub struct PropertyGraphDescription { pub vertex_count: u64, /// How many edges the current commit holds. pub edge_count: u64, - /// The table versions a materialized view's current commit was refreshed - /// from; empty for an independent graph. - #[serde(default)] - pub sources: Vec, } impl PropertyGraphDescription { /// The description in its JSON form. pub fn to_json(&self) -> Result { - serde_json::to_string(self).map_err(|source| Error::Runtime { - message: format!("could not encode a property graph description: {source}"), - }) + to_json(self) + } +} + +/// What a database records about one virtual property graph: its node and +/// edge tables, with each table's properties as create resolved them. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct VirtualPropertyGraphDescription { + /// The graph's name within its namespace. + pub name: String, + /// The namespace holding the graph; empty is the root namespace. + #[serde(rename = "namespace", default)] + pub namespace_path: Vec, + /// The graph's node and edge tables. + #[serde(flatten)] + pub definition: VirtualPropertyGraphDefinition, +} + +impl VirtualPropertyGraphDescription { + /// The description in its JSON form. + pub fn to_json(&self) -> Result { + to_json(self) + } +} + +/// What a database records about one materialized virtual property graph: +/// its node and edge tables, and what its last refresh built from them. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct MaterializedVirtualPropertyGraphDescription { + /// The graph's name within its namespace. + pub name: String, + /// The namespace holding the graph; empty is the root namespace. + #[serde(rename = "namespace", default)] + pub namespace_path: Vec, + /// The graph's node and edge tables. + #[serde(flatten)] + pub definition: VirtualPropertyGraphDefinition, + /// The last refresh's commit. + pub commit: String, + /// When the last refresh landed, as RFC 3339. + pub committed_at: String, + /// How many nodes the last refresh built. + pub vertex_count: u64, + /// How many edges the last refresh built. + pub edge_count: u64, + /// The version of each table the last refresh read. + #[serde(default)] + pub sources: Vec, +} + +impl MaterializedVirtualPropertyGraphDescription { + /// The description in its JSON form. + pub fn to_json(&self) -> Result { + to_json(self) } } @@ -235,12 +301,12 @@ mod tests { use super::*; - fn social() -> PropertyGraphDefinition { + fn social() -> VirtualPropertyGraphDefinition { let person = EndpointReference { table: "person".to_string(), column: "person_id".to_string(), }; - PropertyGraphDefinition::MaterializedView { + VirtualPropertyGraphDefinition { nodes: vec![NodeTable { table: "person".to_string(), key: "person_id".to_string(), @@ -264,12 +330,11 @@ mod tests { } #[test] - fn test_definition_json_is_the_sql_pgq_shape() { + fn test_a_virtual_definition_is_the_sql_pgq_shape() { let json = serde_json::to_value(social()).unwrap(); assert_eq!( json, serde_json::json!({ - "mode": "materialized_view", "nodes": [{"table": "person", "key": "person_id", "label": "Person", "properties": ["name"]}], "edges": [{"table": "knows", "label": "KNOWS", @@ -281,14 +346,14 @@ mod tests { }) ); assert_eq!( - PropertyGraphDefinition::from_json(&json.to_string()).unwrap(), + VirtualPropertyGraphDefinition::from_json(&json.to_string()).unwrap(), social() ); - assert!(PropertyGraphDefinition::from_json(r#"{"nodes": []}"#).is_err()); + assert!(VirtualPropertyGraphDefinition::from_json(r#"{"edges": []}"#).is_err()); } #[test] - fn test_an_independent_definition_carries_its_schemas() { + fn test_a_property_graph_definition_carries_its_schemas() { let person = NodeType::new( "Person", "person_id", @@ -311,12 +376,12 @@ mod tests { None, ) .unwrap(); - let definition = PropertyGraphDefinition::Independent { + let definition = PropertyGraphDefinition { nodes: vec![person.clone()], edges: vec![knows], }; let json = serde_json::to_value(&definition).unwrap(); - assert_eq!(json["mode"], "independent"); + assert!(json.get("mode").is_none()); assert_eq!(json["nodes"][0]["schema"]["fields"][0]["name"], "person_id"); assert!(json["edges"][0].get("schema").is_none()); assert_eq!( @@ -329,6 +394,29 @@ mod tests { ); } + #[test] + fn test_a_materialized_description_carries_its_definition_at_the_top_level() { + let description = MaterializedVirtualPropertyGraphDescription { + name: "social".to_string(), + namespace_path: vec!["analytics".to_string()], + definition: social(), + commit: "c1".to_string(), + committed_at: "2026-09-26T17:00:00Z".to_string(), + vertex_count: 4, + edge_count: 5, + sources: vec![GraphSourceVersion { + table: "person".to_string(), + version: 3, + }], + }; + let json = serde_json::to_value(&description).unwrap(); + assert_eq!(json["namespace"], serde_json::json!(["analytics"])); + assert_eq!(json["nodes"][0]["table"], "person"); + let decoded: MaterializedVirtualPropertyGraphDescription = + serde_json::from_value(json).unwrap(); + assert_eq!(decoded, description); + } + #[tokio::test] async fn test_local_databases_refuse_property_graphs() { let dir = tempfile::tempdir().unwrap(); @@ -343,27 +431,45 @@ mod tests { "{error}" ); }; + let people = PropertyGraphDefinition { + nodes: vec![], + edges: vec![], + }; refused( - conn.create_property_graph("social", &social(), &[]) + conn.create_property_graph("people", &people, &[]) .await .unwrap_err(), ); refused( - conn.describe_property_graph("social", &[]) + conn.create_virtual_property_graph("social", &social(), &[]) .await .unwrap_err(), ); refused( - conn.refresh_property_graph("social", &[]) + conn.create_materialized_virtual_property_graph("social", &social(), &[]) .await .unwrap_err(), ); refused( - conn.rollback_property_graph("social", &[]) + conn.describe_virtual_property_graph("social", &[]) + .await + .unwrap_err(), + ); + refused( + conn.refresh_materialized_virtual_property_graph("social", &[]) + .await + .unwrap_err(), + ); + refused( + conn.rollback_property_graph("people", &[]) + .await + .unwrap_err(), + ); + refused(conn.drop_property_graph("people", &[]).await.unwrap_err()); + refused( + conn.list_materialized_virtual_property_graphs(&[]) .await .unwrap_err(), ); - refused(conn.drop_property_graph("social", &[]).await.unwrap_err()); - refused(conn.list_property_graphs(&[]).await.unwrap_err()); } } diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index 2a6bba747..4cd4da13a 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -29,7 +29,10 @@ use crate::function::{ FunctionArtifactRequest, FunctionRegistrationRequest, FunctionSignature, FunctionVersion, PythonRuntimeSpec, }; -use crate::graph::{PropertyGraphDefinition, PropertyGraphDescription}; +use crate::graph::{ + MaterializedVirtualPropertyGraphDescription, PropertyGraphDefinition, PropertyGraphDescription, + VirtualPropertyGraphDefinition, VirtualPropertyGraphDescription, +}; use crate::job::Job; use crate::materialized_view::CreateMaterializedViewRequest; use crate::remote::job::{PauseJobResponse, RemoteJob, ResumeJobResponse, job_state_to_client}; @@ -841,12 +844,119 @@ struct RemoteListViewsResponse { page_token: Option, } -#[derive(serde::Deserialize)] -struct RemoteListPropertyGraphsResponse { - #[serde(default)] - property_graphs: Vec, - #[serde(default)] - page_token: Option, +/// The route prefix of each kind of graph. +const PROPERTY_GRAPH: &str = "property_graph"; +const VIRTUAL_PROPERTY_GRAPH: &str = "virtual_property_graph"; +const MATERIALIZED_VIRTUAL_PROPERTY_GRAPH: &str = "materialized_virtual_property_graph"; + +/// A graph definition as a create request's body. +fn graph_body(definition: &impl serde::Serialize) -> Result { + serde_json::to_value(definition).map_err(|source| Error::InvalidInput { + message: format!("could not encode a property graph definition: {source}"), + }) +} + +impl RemoteDatabase { + /// POST one route of a graph of `kind` and decode what it answers. + async fn graph_request( + &self, + kind: &str, + name: &str, + namespace_path: &[String], + verb: &str, + body: Option, + ) -> Result { + let graph_id = build_object_identifier("Property graph name", name, namespace_path)?; + let mut req = self.client.post(&format!("/v1/{kind}/{graph_id}/{verb}")); + if let Some(body) = body { + req = req.json(&body); + } + let (request_id, response) = self.client.send(req).await?; + let response = self.client.check_response(&request_id, response).await?; + response.json().await.err_to_http(request_id) + } + + /// Drop a graph of `kind` and return the job deleting its storage. + async fn drop_graph_async( + &self, + kind: &str, + name: &str, + namespace_path: &[String], + ) -> Result { + let graph_id = build_object_identifier("Property graph name", name, namespace_path)?; + let req = self.client.post(&format!("/v1/{kind}/{graph_id}/drop")); + let (request_id, response) = self.client.send(req).await?; + let response = self.client.check_response(&request_id, response).await?; + let status = response.status(); + let body = response.text().await.err_to_http(request_id.clone())?; + match status { + // Nothing was bound to the name, so nothing is being deleted. + StatusCode::OK => Ok(Job::new_done()), + StatusCode::ACCEPTED => { + let job_id = extract_job_id(&body).ok_or_else(|| Error::Http { + source: "property graph drop response did not contain a valid job_id".into(), + request_id, + status_code: Some(status), + })?; + Ok(Job::new(Box::new(RemoteJob::new( + self.client.clone(), + job_id, + )))) + } + _ => Err(Error::Http { + source: "property graph drop must return 200 OK or 202 Accepted".into(), + request_id, + status_code: Some(status), + }), + } + } + + /// The names of the graphs of `kind` in one namespace, every page of them. + async fn list_graphs(&self, kind: &str, namespace_path: &[String]) -> Result> { + let namespace_id = build_namespace_identifier(namespace_path)?; + let path = format!("/v1/namespace/{namespace_id}/{kind}/list"); + // A listing names its field after the kind: `property_graphs`, ... + let field = format!("{kind}s"); + let mut graphs = Vec::new(); + let mut page_token: Option = None; + let mut seen_page_tokens = HashSet::new(); + loop { + let mut req = self.client.get(&path); + if let Some(token) = &page_token { + req = req.query(&[("page_token", token)]); + } + let (request_id, response) = self.client.send(req).await?; + let response = self.client.check_response(&request_id, response).await?; + let status = response.status(); + let mut page: serde_json::Map = + response.json().await.err_to_http(request_id.clone())?; + let names: Vec = match page.remove(&field) { + Some(names) => serde_json::from_value(names).map_err(|source| Error::Http { + source: format!("{kind} listing response is malformed: {source}").into(), + request_id: request_id.clone(), + status_code: Some(status), + })?, + None => Vec::new(), + }; + graphs.extend(names); + let Some(next_page_token) = page + .remove("page_token") + .and_then(|token| token.as_str().map(str::to_string)) + .filter(|token| !token.is_empty()) + else { + break; + }; + if !seen_page_tokens.insert(next_page_token.clone()) { + return Err(Error::Http { + source: "Property graph listing response repeated a page_token".into(), + request_id, + status_code: Some(status), + }); + } + page_token = Some(next_page_token); + } + Ok(graphs) + } } /// Bound on `list_jobs` page walking; a warning is logged when the listing @@ -1294,14 +1404,9 @@ impl Database for RemoteDatabase { definition: &PropertyGraphDefinition, namespace_path: &[String], ) -> Result { - let graph_id = build_object_identifier("Property graph name", name, namespace_path)?; - let req = self - .client - .post(&format!("/v1/property_graph/{graph_id}/create")) - .json(definition); - let (request_id, response) = self.client.send(req).await?; - let response = self.client.check_response(&request_id, response).await?; - response.json().await.err_to_http(request_id) + let body = graph_body(definition)?; + self.graph_request(PROPERTY_GRAPH, name, namespace_path, "create", Some(body)) + .await } async fn describe_property_graph( @@ -1309,13 +1414,8 @@ impl Database for RemoteDatabase { name: &str, namespace_path: &[String], ) -> Result { - let graph_id = build_object_identifier("Property graph name", name, namespace_path)?; - let req = self - .client - .post(&format!("/v1/property_graph/{graph_id}/describe")); - let (request_id, response) = self.client.send(req).await?; - let response = self.client.check_response(&request_id, response).await?; - response.json().await.err_to_http(request_id) + self.graph_request(PROPERTY_GRAPH, name, namespace_path, "describe", None) + .await } async fn drop_property_graph(&self, name: &str, namespace_path: &[String]) -> Result<()> { @@ -1330,34 +1430,12 @@ impl Database for RemoteDatabase { name: &str, namespace_path: &[String], ) -> Result { - let graph_id = build_object_identifier("Property graph name", name, namespace_path)?; - let req = self - .client - .post(&format!("/v1/property_graph/{graph_id}/drop")); - let (request_id, response) = self.client.send(req).await?; - let response = self.client.check_response(&request_id, response).await?; - let status = response.status(); - let body = response.text().await.err_to_http(request_id.clone())?; - match status { - // Nothing was bound to the name, so nothing is being deleted. - StatusCode::OK => Ok(Job::new_done()), - StatusCode::ACCEPTED => { - let job_id = extract_job_id(&body).ok_or_else(|| Error::Http { - source: "property graph drop response did not contain a valid job_id".into(), - request_id, - status_code: Some(status), - })?; - Ok(Job::new(Box::new(RemoteJob::new( - self.client.clone(), - job_id, - )))) - } - _ => Err(Error::Http { - source: "property graph drop must return 200 OK or 202 Accepted".into(), - request_id, - status_code: Some(status), - }), - } + self.drop_graph_async(PROPERTY_GRAPH, name, namespace_path) + .await + } + + async fn list_property_graphs(&self, namespace_path: &[String]) -> Result> { + self.list_graphs(PROPERTY_GRAPH, namespace_path).await } async fn insert_into_property_graph( @@ -1381,7 +1459,7 @@ impl Database for RemoteDatabase { } let req = self .client - .post(&format!("/v1/property_graph/{graph_id}/insert")) + .post(&format!("/v1/{PROPERTY_GRAPH}/{graph_id}/insert")) .query(&[("label", label)]) .body(body) .header(CONTENT_TYPE, ARROW_STREAM_CONTENT_TYPE); @@ -1390,65 +1468,145 @@ impl Database for RemoteDatabase { response.json().await.err_to_http(request_id) } - async fn refresh_property_graph( - &self, - name: &str, - namespace_path: &[String], - ) -> Result { - let graph_id = build_object_identifier("Property graph name", name, namespace_path)?; - let req = self - .client - .post(&format!("/v1/property_graph/{graph_id}/refresh")); - let (request_id, response) = self.client.send(req).await?; - let response = self.client.check_response(&request_id, response).await?; - response.json().await.err_to_http(request_id) - } - async fn rollback_property_graph( &self, name: &str, namespace_path: &[String], ) -> Result { - let graph_id = build_object_identifier("Property graph name", name, namespace_path)?; - let req = self - .client - .post(&format!("/v1/property_graph/{graph_id}/rollback")); - let (request_id, response) = self.client.send(req).await?; - let response = self.client.check_response(&request_id, response).await?; - response.json().await.err_to_http(request_id) + self.graph_request(PROPERTY_GRAPH, name, namespace_path, "rollback", None) + .await } - async fn list_property_graphs(&self, namespace_path: &[String]) -> Result> { - let namespace_id = build_namespace_identifier(namespace_path)?; - let path = format!("/v1/namespace/{namespace_id}/property_graph/list"); - let mut graphs = Vec::new(); - let mut page_token: Option = None; - let mut seen_page_tokens = HashSet::new(); - loop { - let mut req = self.client.get(&path); - if let Some(token) = &page_token { - req = req.query(&[("page_token", token)]); - } - let (request_id, response) = self.client.send(req).await?; - let response = self.client.check_response(&request_id, response).await?; - let status = response.status(); - let response: RemoteListPropertyGraphsResponse = - response.json().await.err_to_http(request_id.clone())?; - graphs.extend(response.property_graphs); - let Some(next_page_token) = response.page_token.filter(|token| !token.is_empty()) - else { - break; - }; - if !seen_page_tokens.insert(next_page_token.clone()) { - return Err(Error::Http { - source: "Property graph listing response repeated a page_token".into(), - request_id, - status_code: Some(status), - }); - } - page_token = Some(next_page_token); - } - Ok(graphs) + async fn create_virtual_property_graph( + &self, + name: &str, + definition: &VirtualPropertyGraphDefinition, + namespace_path: &[String], + ) -> Result { + let body = graph_body(definition)?; + self.graph_request( + VIRTUAL_PROPERTY_GRAPH, + name, + namespace_path, + "create", + Some(body), + ) + .await + } + + async fn describe_virtual_property_graph( + &self, + name: &str, + namespace_path: &[String], + ) -> Result { + self.graph_request( + VIRTUAL_PROPERTY_GRAPH, + name, + namespace_path, + "describe", + None, + ) + .await + } + + async fn drop_virtual_property_graph( + &self, + name: &str, + namespace_path: &[String], + ) -> Result<()> { + self.drop_virtual_property_graph_async(name, namespace_path) + .await? + .wait() + .await + } + + async fn drop_virtual_property_graph_async( + &self, + name: &str, + namespace_path: &[String], + ) -> Result { + self.drop_graph_async(VIRTUAL_PROPERTY_GRAPH, name, namespace_path) + .await + } + + async fn list_virtual_property_graphs(&self, namespace_path: &[String]) -> Result> { + self.list_graphs(VIRTUAL_PROPERTY_GRAPH, namespace_path) + .await + } + + async fn create_materialized_virtual_property_graph( + &self, + name: &str, + definition: &VirtualPropertyGraphDefinition, + namespace_path: &[String], + ) -> Result { + let body = graph_body(definition)?; + self.graph_request( + MATERIALIZED_VIRTUAL_PROPERTY_GRAPH, + name, + namespace_path, + "create", + Some(body), + ) + .await + } + + async fn describe_materialized_virtual_property_graph( + &self, + name: &str, + namespace_path: &[String], + ) -> Result { + self.graph_request( + MATERIALIZED_VIRTUAL_PROPERTY_GRAPH, + name, + namespace_path, + "describe", + None, + ) + .await + } + + async fn drop_materialized_virtual_property_graph( + &self, + name: &str, + namespace_path: &[String], + ) -> Result<()> { + self.drop_materialized_virtual_property_graph_async(name, namespace_path) + .await? + .wait() + .await + } + + async fn drop_materialized_virtual_property_graph_async( + &self, + name: &str, + namespace_path: &[String], + ) -> Result { + self.drop_graph_async(MATERIALIZED_VIRTUAL_PROPERTY_GRAPH, name, namespace_path) + .await + } + + async fn list_materialized_virtual_property_graphs( + &self, + namespace_path: &[String], + ) -> Result> { + self.list_graphs(MATERIALIZED_VIRTUAL_PROPERTY_GRAPH, namespace_path) + .await + } + + async fn refresh_materialized_virtual_property_graph( + &self, + name: &str, + namespace_path: &[String], + ) -> Result { + self.graph_request( + MATERIALIZED_VIRTUAL_PROPERTY_GRAPH, + name, + namespace_path, + "refresh", + None, + ) + .await } async fn open_job(&self, job_id: &str) -> Result { @@ -4453,10 +4611,9 @@ mod tests { } } - fn social_graph() -> crate::graph::PropertyGraphDefinition { - crate::graph::PropertyGraphDefinition::from_json( - r#"{"mode": "materialized_view", - "nodes": [{"table": "person", "key": "person_id", "label": "Person"}], + fn social_graph() -> crate::graph::VirtualPropertyGraphDefinition { + crate::graph::VirtualPropertyGraphDefinition::from_json( + r#"{"nodes": [{"table": "person", "key": "person_id", "label": "Person"}], "edges": [{"table": "knows", "label": "KNOWS", "source": {"column": "src_id", "references": {"table": "person", "column": "person_id"}}, @@ -4468,50 +4625,75 @@ mod tests { .unwrap() } - fn property_graph_description_body(name: &str, namespace: &[&str]) -> String { - let mut body = serde_json::to_value(social_graph()).unwrap(); + fn people_graph() -> crate::graph::PropertyGraphDefinition { + crate::graph::PropertyGraphDefinition::from_json( + r#"{"nodes": [{"label": "Person", "key": "person_id", "schema": {"fields": [ + {"name": "person_id", "type": {"type": "int64"}, "nullable": false}]}}], + "edges": [{"label": "KNOWS", + "source": {"label": "Person", "column": "src_id"}, + "destination": {"label": "Person", "column": "dst_id"}}]}"#, + ) + .unwrap() + } + + /// A description of `definition`, with `extra` fields beside it. + fn graph_description_body( + definition: serde_json::Value, + name: &str, + namespace: &[&str], + extra: serde_json::Value, + ) -> String { + let mut body = definition; let fields = body.as_object_mut().unwrap(); fields.insert("name".into(), name.into()); fields.insert("namespace".into(), namespace.into()); - fields.insert( - "commit".into(), - "6f1c2d3e-0000-4000-8000-000000000000".into(), - ); - fields.insert("committed_at".into(), "2026-09-26T17:00:00Z".into()); - fields.insert("vertex_count".into(), 4.into()); - fields.insert("edge_count".into(), 5.into()); - fields.insert( - "sources".into(), - serde_json::json!([{"table": "person", "version": 3}, {"table": "knows", "version": 2}]), - ); + fields.extend(extra.as_object().unwrap().clone()); body.to_string() } + fn materialized_body(name: &str, namespace: &[&str]) -> String { + graph_description_body( + serde_json::to_value(social_graph()).unwrap(), + name, + namespace, + serde_json::json!({ + "commit": "6f1c2d3e-0000-4000-8000-000000000000", + "committed_at": "2026-09-26T17:00:00Z", + "vertex_count": 4, + "edge_count": 5, + "sources": [{"table": "person", "version": 3}, {"table": "knows", "version": 2}], + }), + ) + } + #[tokio::test] - async fn test_create_property_graph_posts_the_definition() { + async fn test_create_materialized_virtual_property_graph_posts_the_definition() { let conn = Connection::new_with_handler(|request| { assert_eq!(request.method(), &reqwest::Method::POST); assert_eq!( request.url().path(), - "/v1/property_graph/analytics$social/create" + "/v1/materialized_virtual_property_graph/analytics$social/create" ); let body: serde_json::Value = serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap(); assert_eq!(body, serde_json::to_value(social_graph()).unwrap()); http::Response::builder() .status(200) - .body(property_graph_description_body("social", &["analytics"])) + .body(materialized_body("social", &["analytics"])) .unwrap() }); let graph = conn - .create_property_graph("social", &social_graph(), &["analytics".into()]) + .create_materialized_virtual_property_graph( + "social", + &social_graph(), + &["analytics".into()], + ) .await .unwrap(); assert_eq!(graph.name, "social"); assert_eq!(graph.namespace_path, vec!["analytics".to_string()]); assert_eq!(graph.definition, social_graph()); assert_eq!(graph.commit, "6f1c2d3e-0000-4000-8000-000000000000"); - assert_eq!(graph.previous_commit, None); assert_eq!((graph.vertex_count, graph.edge_count), (4, 5)); assert_eq!( graph.sources[1], @@ -4523,29 +4705,97 @@ mod tests { } #[tokio::test] - async fn test_graph_writes_post_to_their_routes() { + async fn test_each_kind_of_graph_has_its_own_routes() { let conn = Connection::new_with_handler(|request| { assert_eq!(request.method(), &reqwest::Method::POST); - match request.url().path() { - "/v1/property_graph/analytics$people/insert" => { - assert_eq!(request.url().query(), Some("label=Person")); - assert_eq!( - request.headers().get("Content-Type").unwrap(), - ARROW_STREAM_CONTENT_TYPE - ); - } - "/v1/property_graph/analytics$people/refresh" - | "/v1/property_graph/analytics$people/rollback" => { + let body = match request.url().path() { + "/v1/virtual_property_graph/social/create" + | "/v1/virtual_property_graph/social/describe" => graph_description_body( + serde_json::to_value(social_graph()).unwrap(), + "social", + &[], + serde_json::json!({}), + ), + "/v1/materialized_virtual_property_graph/social_mv/describe" + | "/v1/materialized_virtual_property_graph/social_mv/refresh" => { assert!(request.body().is_none(), "{:?}", request.body()); + materialized_body("social_mv", &[]) } + "/v1/property_graph/people/create" + | "/v1/property_graph/people/describe" + | "/v1/property_graph/people/rollback" => graph_description_body( + serde_json::to_value(people_graph()).unwrap(), + "people", + &[], + serde_json::json!({ + "commit": "c2", + "committed_at": "2026-09-26T17:00:00Z", + "previous_commit": "c1", + "vertex_count": 0, + "edge_count": 0, + }), + ), other => panic!("unexpected path {other}"), - } + }; + http::Response::builder().status(200).body(body).unwrap() + }); + let virtual_graph = conn + .create_virtual_property_graph("social", &social_graph(), &[]) + .await + .unwrap(); + assert_eq!(virtual_graph.definition, social_graph()); + assert_eq!( + conn.describe_virtual_property_graph("social", &[]) + .await + .unwrap(), + virtual_graph + ); + let refreshed = conn + .refresh_materialized_virtual_property_graph("social_mv", &[]) + .await + .unwrap(); + assert_eq!(refreshed.edge_count, 5); + conn.describe_materialized_virtual_property_graph("social_mv", &[]) + .await + .unwrap(); + let people = conn + .create_property_graph("people", &people_graph(), &[]) + .await + .unwrap(); + assert_eq!(people.definition, people_graph()); + assert_eq!(people.previous_commit.as_deref(), Some("c1")); + conn.describe_property_graph("people", &[]).await.unwrap(); + conn.rollback_property_graph("people", &[]).await.unwrap(); + } + + #[tokio::test] + async fn test_insert_into_property_graph_sends_arrow_rows() { + let conn = Connection::new_with_handler(|request| { + assert_eq!(request.method(), &reqwest::Method::POST); + assert_eq!( + request.url().path(), + "/v1/property_graph/analytics$people/insert" + ); + assert_eq!(request.url().query(), Some("label=Person")); + assert_eq!( + request.headers().get("Content-Type").unwrap(), + ARROW_STREAM_CONTENT_TYPE + ); http::Response::builder() .status(200) - .body(property_graph_description_body("people", &["analytics"])) + .body(graph_description_body( + serde_json::to_value(people_graph()).unwrap(), + "people", + &["analytics"], + serde_json::json!({ + "commit": "c2", + "committed_at": "2026-09-26T17:00:00Z", + "vertex_count": 2, + "edge_count": 0, + }), + )) .unwrap() }); - let namespace = ["analytics".to_string()]; let rows = RecordBatch::try_new( Arc::new(Schema::new(vec![Field::new( "person_id", @@ -4556,67 +4806,47 @@ mod tests { ) .unwrap(); let inserted = conn - .insert_into_property_graph("people", "Person", rows, &namespace) - .await - .unwrap(); - assert_eq!(inserted.name, "people"); - conn.refresh_property_graph("people", &namespace) - .await - .unwrap(); - conn.rollback_property_graph("people", &namespace) + .insert_into_property_graph("people", "Person", rows, &["analytics".to_string()]) .await .unwrap(); + assert_eq!(inserted.vertex_count, 2); } #[tokio::test] - async fn test_describe_and_drop_address_the_property_graph_in_the_path() { - let conn = Connection::new_with_handler(|request| { - assert_eq!(request.method(), &reqwest::Method::POST); - assert_eq!(request.url().path(), "/v1/property_graph/social/describe"); - assert!(request.body().is_none(), "{:?}", request.body()); - http::Response::builder() - .status(200) - .body(property_graph_description_body("social", &[])) - .unwrap() - }); - let graph = conn.describe_property_graph("social", &[]).await.unwrap(); - assert!(graph.namespace_path.is_empty()); - let crate::graph::PropertyGraphDefinition::MaterializedView { edges, .. } = - graph.definition - else { - panic!("expected a materialized view graph") - }; - assert_eq!(edges[0].properties, Some(vec!["since".to_string()])); - - let conn = Connection::new_with_handler(|request| { - assert_eq!(request.method(), &reqwest::Method::POST); - assert_eq!( - request.url().path(), - "/v1/property_graph/analytics$social/drop" - ); - assert!(request.body().is_none(), "{:?}", request.body()); - http::Response::builder().status(200).body("{}").unwrap() - }); - conn.drop_property_graph("social", &["analytics".into()]) - .await + async fn test_drop_graph_async_reports_the_cleanup_job() { + for kind in [ + "property_graph", + "virtual_property_graph", + "materialized_virtual_property_graph", + ] { + let db = super::RemoteDatabase::new_mock(move |request| { + assert_eq!(request.url().path(), format!("/v1/{kind}/social/drop")); + http::Response::builder() + .status(202) + .body(r#"{"job_id":"j1-do-graph"}"#) + .unwrap() + }); + let job = match kind { + "property_graph" => db.drop_property_graph_async("social", &[]).await, + "virtual_property_graph" => { + db.drop_virtual_property_graph_async("social", &[]).await + } + _ => { + db.drop_materialized_virtual_property_graph_async("social", &[]) + .await + } + } .unwrap(); - } - - #[tokio::test] - async fn test_drop_property_graph_async_reports_the_cleanup_job() { - let db = super::RemoteDatabase::new_mock(|_| { - http::Response::builder() - .status(202) - .body(r#"{"job_id":"j1-do-graph"}"#) - .unwrap() - }); - let job = db.drop_property_graph_async("social", &[]).await.unwrap(); - assert_eq!(job.id(), Some("j1-do-graph")); + assert_eq!(job.id(), Some("j1-do-graph"), "{kind}"); + } let db = super::RemoteDatabase::new_mock(|_| { http::Response::builder().status(200).body("{}").unwrap() }); - let job = db.drop_property_graph_async("social", &[]).await.unwrap(); + let job = db + .drop_virtual_property_graph_async("social", &[]) + .await + .unwrap(); assert_eq!(job.id(), None); job.wait().await.unwrap(); @@ -4637,12 +4867,12 @@ mod tests { } #[tokio::test] - async fn test_list_property_graphs_walks_pages() { + async fn test_list_graphs_walks_pages_of_their_kind() { let conn = Connection::new_with_handler(|request| { assert_eq!(request.method(), &reqwest::Method::GET); assert_eq!( request.url().path(), - "/v1/namespace/analytics/property_graph/list" + "/v1/namespace/analytics/materialized_virtual_property_graph/list" ); let page = request .url() @@ -4650,23 +4880,40 @@ mod tests { .find(|(key, _)| key == "page_token") .map(|(_, value)| value.into_owned()); let body = match page.as_deref() { - None => r#"{"property_graphs":["social"],"page_token":"p2"}"#, - Some("p2") => r#"{"property_graphs":["payments"],"page_token":null}"#, + None => r#"{"materialized_virtual_property_graphs":["social"],"page_token":"p2"}"#, + Some("p2") => { + r#"{"materialized_virtual_property_graphs":["payments"],"page_token":null}"# + } Some(other) => panic!("unexpected page token: {other}"), }; http::Response::builder().status(200).body(body).unwrap() }); assert_eq!( - conn.list_property_graphs(&["analytics".into()]) + conn.list_materialized_virtual_property_graphs(&["analytics".into()]) .await .unwrap(), vec!["social".to_string(), "payments".to_string()] ); + let conn = Connection::new_with_handler(|request| { + assert_eq!( + request.url().path(), + "/v1/namespace/$/virtual_property_graph/list" + ); + http::Response::builder() + .status(200) + .body(r#"{"virtual_property_graphs":["social"]}"#) + .unwrap() + }); + assert_eq!( + conn.list_virtual_property_graphs(&[]).await.unwrap(), + vec!["social".to_string()] + ); + let conn = Connection::new_with_handler(|_| { http::Response::builder() .status(200) - .body(r#"{"property_graphs":["social"],"page_token":"same"}"#) + .body(r#"{"property_graphs":["people"],"page_token":"same"}"#) .unwrap() }); let error = conn.list_property_graphs(&[]).await.unwrap_err(); @@ -4689,7 +4936,7 @@ mod tests { ("", vec![]), ("..", vec![]), ] { - let error = match conn.describe_property_graph(name, &namespace).await { + let error = match conn.describe_virtual_property_graph(name, &namespace).await { Ok(_) => panic!("accepted {name:?} in {namespace:?}"), Err(error) => error.to_string(), };