diff --git a/docs/src/python/python.md b/docs/src/python/python.md index 6efb278fb..340f44138 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -204,12 +204,23 @@ listing a storage directory. ## Property Graphs -A property graph reads tables as the nodes and edges of a graph. Manage one -with [create_property_graph][lancedb.DBConnection.create_property_graph], +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]. +::: lancedb.graph.NodeType + +::: lancedb.graph.EdgeType + ::: lancedb.graph.NodeTable ::: lancedb.graph.EdgeTable diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 1df17d503..c0cbef484 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -49,6 +49,8 @@ class MetricDescription: description: str def register_lancedb_metrics_recorder() -> bool: ... +def graph_schema_to_json(schema: pa.Schema) -> str: ... +def graph_schema_from_json(json: str) -> pa.Schema: ... def lancedb_metrics_catalog() -> List[MetricDescription]: ... def snapshot_lancedb_metrics() -> List[MetricPoint]: ... def tokenize( @@ -201,6 +203,19 @@ class Connection(object): async def list_property_graphs( self, namespace_path: Optional[List[str]] = None ) -> List[str]: ... + async def insert_into_property_graph( + self, + name: str, + label: str, + data: Any, + namespace_path: Optional[List[str]] = None, + ) -> str: ... + async def refresh_property_graph( + self, name: str, namespace_path: Optional[List[str]] = None + ) -> str: ... + async def rollback_property_graph( + self, name: str, namespace_path: Optional[List[str]] = None + ) -> str: ... async def list_jobs(self) -> List[JobInfo]: ... async def cancel_job(self, job_id: str) -> bool: ... async def pause_job(self, job_id: str) -> str: ... diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index 045a67e72..7e7330af2 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -51,11 +51,14 @@ from ._lancedb import connect as lancedb_connect # type: ignore from .functions import FunctionVersion, UdfDefinition from .graph import ( EdgeTable, + EdgeType, NodeTable, + NodeType, PropertyGraphDescription, _definition_json, _description_from_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 @@ -1028,17 +1031,23 @@ class DBConnection(EnforceOverrides): def create_property_graph( self, name: str, - nodes: List[NodeTable], - edges: List[EdgeTable], + nodes: List[Union[NodeTable, NodeType]], + edges: List[Union[EdgeTable, EdgeType]], *, namespace_path: Optional[List[str]] = None, ) -> PropertyGraphDescription: - """Create a property graph: tables in its namespace read as the nodes - and edges of a graph. + """Create a property graph. - Each node table names its key column and label. Each edge table names - its label and the node keys its source and destination columns - reference. A name already taken is an error. Local connections raise + 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 ``NotImplementedError``. >>> import lancedb @@ -1066,8 +1075,9 @@ class DBConnection(EnforceOverrides): def describe_property_graph( self, name: str, *, namespace_path: Optional[List[str]] = None ) -> PropertyGraphDescription: - """What this database records about a property graph: its node and - edge tables, its size, and the table versions it was built from. + """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``. """ @@ -1075,6 +1085,82 @@ class DBConnection(EnforceOverrides): "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", + ... nodes=[ + ... NodeType( + ... "Person", + ... key="person_id", + ... schema=pa.schema( + ... [ + ... pa.field("person_id", pa.int64(), False), + ... ("name", pa.string()), + ... ] + ... ), + ... ) + ... ], + ... edges=[ + ... EdgeType( + ... "KNOWS", + ... source=("Person", "src_id"), + ... destination=("Person", "dst_id"), + ... ) + ... ], + ... ) # doctest: +SKIP + >>> db.insert_into_property_graph( + ... "people", "Person", [{"person_id": 1, "name": "Alice"}] + ... ) # doctest: +SKIP + """ + raise NotImplementedError( + "Property graph operations are not supported for this connection type" + ) + + def refresh_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> PropertyGraphDescription: + """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 connections raise + ``NotImplementedError``. + """ + raise NotImplementedError( + "Property graph operations are not supported for this connection type" + ) + + def rollback_property_graph( + self, name: str, *, namespace_path: Optional[List[str]] = None + ) -> PropertyGraphDescription: + """Return a property graph to its previous commit. + + The commit rolled back from is discarded, so a second rollback in a row + is an error. Local connections raise ``NotImplementedError``. + """ + raise NotImplementedError( + "Property graph operations are not supported for this connection type" + ) + def drop_property_graph( self, name: str, *, namespace_path: Optional[List[str]] = None ) -> None: @@ -1967,8 +2053,8 @@ class LanceDBConnection(DBConnection): def create_property_graph( self, name: str, - nodes: List[NodeTable], - edges: List[EdgeTable], + nodes: List[Union[NodeTable, NodeType]], + edges: List[Union[EdgeTable, EdgeType]], *, namespace_path: Optional[List[str]] = None, ) -> PropertyGraphDescription: @@ -1986,6 +2072,37 @@ class LanceDBConnection(DBConnection): 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 @@ -3028,12 +3145,13 @@ class AsyncConnection(object): async def create_property_graph( self, name: str, - nodes: List[NodeTable], - edges: List[EdgeTable], + nodes: List[Union[NodeTable, NodeType]], + edges: List[Union[EdgeTable, EdgeType]], *, namespace_path: Optional[List[str]] = None, ) -> PropertyGraphDescription: - """Create a property graph over tables in its namespace. + """Create a property graph: a materialized view of tables in its + namespace, or an independent graph that holds its own rows. See [DBConnection.create_property_graph][lancedb.DBConnection.create_property_graph]. @@ -3052,6 +3170,41 @@ class AsyncConnection(object): 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: diff --git a/python/python/lancedb/graph.py b/python/python/lancedb/graph.py index 0d6c8bd8c..eaf16d2f3 100644 --- a/python/python/lancedb/graph.py +++ b/python/python/lancedb/graph.py @@ -1,21 +1,31 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright The LanceDB Authors -"""Property graphs: tables read as the nodes and edges of a graph. +"""Property graphs: catalog objects read as graphs, in one of two modes. -A property graph is a catalog object in a namespace, next to the tables it -reads. Its definition says which tables 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``. The tables stay ordinary tables. See -``DBConnection.create_property_graph``. +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 *materialized view* 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. + +See ``DBConnection.create_property_graph``. """ from __future__ import annotations import json from dataclasses import dataclass -from typing import Any, Dict, List, Optional, Sequence, Tuple +from typing import Any, Dict, List, Optional, Sequence, Tuple, Union + +import pyarrow as pa + +from ._lancedb import graph_schema_from_json, graph_schema_to_json @dataclass(frozen=True) @@ -29,7 +39,8 @@ class NodeTable: label: str """The label its nodes carry.""" properties: Optional[List[str]] = None - """The columns exposed as node properties; every column when ``None``.""" + """The columns exposed as node properties; every column when ``None``. A + created graph reports the list it resolved to.""" @dataclass(frozen=True) @@ -55,7 +66,37 @@ class EdgeTable: destination: Endpoint """The column naming each edge's destination node.""" properties: Optional[List[str]] = None - """The columns exposed as edge properties; every column when ``None``.""" + """The columns exposed as edge properties; every column when ``None``. A + created graph reports the list it resolved to.""" + + +@dataclass(frozen=True) +class NodeType: + """The nodes of one label of an independent graph.""" + + label: str + """The label its nodes carry.""" + key: str + """The field that identifies a node: a non-nullable integer, string or + binary field of ``schema``.""" + schema: pa.Schema + """Every property, the key among them.""" + + +@dataclass(frozen=True) +class EdgeType: + """The edges of one label of an independent graph.""" + + label: str + """The label its edges carry.""" + source: Tuple[str, str] + """The label of each edge's source node, and the column inserted rows carry + its key in, as ``(label, column)``.""" + destination: Tuple[str, str] + """The label of each edge's destination node, and the column inserted rows + carry its key in, as ``(label, column)``.""" + schema: Optional[pa.Schema] = None + """The edge's properties, when it has any.""" @dataclass(frozen=True) @@ -66,21 +107,48 @@ class PropertyGraphDescription: """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.""" - edges: List[EdgeTable] - """The edge tables.""" + 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).""" + commit: str + """The graph's current commit.""" + committed_at: str + """When the current commit landed, as RFC 3339.""" + previous_commit: Optional[str] + """The commit a rollback returns to, if there is one.""" vertex_count: int - """How many nodes the graph held when it was built.""" + """How many nodes the current commit holds.""" edge_count: int - """How many edges the graph held when it was built.""" + """How many edges the current commit holds.""" sources: Dict[str, int] - """The version of each table the graph was built from, by table name.""" + """For a materialized view, the version of each table the current commit + was refreshed from, by table name.""" -def _definition_json(nodes: Sequence[NodeTable], edges: Sequence[EdgeTable]) -> str: +INDEPENDENT = "independent" +MATERIALIZED_VIEW = "materialized_view" + + +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) return json.dumps( { + "mode": mode, "nodes": [_node_to_json(node) for node in nodes], "edges": [_edge_to_json(edge) for edge in edges], } @@ -89,11 +157,16 @@ def _definition_json(nodes: Sequence[NodeTable], edges: Sequence[EdgeTable]) -> def _description_from_json(text: str) -> PropertyGraphDescription: described = json.loads(text) + mode = described["mode"] return PropertyGraphDescription( name=described["name"], namespace_path=list(described.get("namespace", [])), - nodes=[_node_from_json(node) for node in described["nodes"]], - edges=[_edge_from_json(edge) for edge in described.get("edges", [])], + 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", [])], + commit=described["commit"], + committed_at=described["committed_at"], + previous_commit=described.get("previous_commit"), vertex_count=described["vertex_count"], edge_count=described["edge_count"], sources={ @@ -103,7 +176,21 @@ def _description_from_json(text: str) -> PropertyGraphDescription: ) -def _node_to_json(node: NodeTable) -> Dict[str, Any]: +def _schema_to_json(schema: pa.Schema) -> Dict[str, Any]: + return json.loads(graph_schema_to_json(schema)) + + +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), + } encoded: Dict[str, Any] = { "table": node.table, "key": node.key, @@ -114,8 +201,20 @@ def _node_to_json(node: NodeTable) -> Dict[str, Any]: return encoded -def _edge_to_json(edge: EdgeTable) -> Dict[str, Any]: - encoded: Dict[str, Any] = { +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 = { "table": edge.table, "label": edge.label, "source": _endpoint_to_json(edge.source), @@ -134,7 +233,13 @@ def _endpoint_to_json(endpoint: Endpoint) -> Dict[str, Any]: } -def _node_from_json(node: Dict[str, Any]) -> NodeTable: +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"]), + ) return NodeTable( table=node["table"], key=node["key"], @@ -143,7 +248,15 @@ def _node_from_json(node: Dict[str, Any]) -> NodeTable: ) -def _edge_from_json(edge: Dict[str, Any]) -> EdgeTable: +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), + ) return EdgeTable( table=edge["table"], label=edge["label"], diff --git a/python/python/lancedb/remote/db.py b/python/python/lancedb/remote/db.py index a657e77bb..54d155201 100644 --- a/python/python/lancedb/remote/db.py +++ b/python/python/lancedb/remote/db.py @@ -36,7 +36,7 @@ import pyarrow as pa from ..common import DATA from ..db import DBConnection, LOOP from ..functions import FunctionVersion, UdfDefinition -from ..graph import EdgeTable, NodeTable, PropertyGraphDescription +from ..graph import EdgeTable, EdgeType, NodeTable, NodeType, PropertyGraphDescription from ..job import AsyncJob, Job from ..sql import Query as SqlQuery from ..sql import QueryDescription @@ -947,8 +947,8 @@ class RemoteDBConnection(DBConnection): def create_property_graph( self, name: str, - nodes: List[NodeTable], - edges: List[EdgeTable], + nodes: List[Union[NodeTable, NodeType]], + edges: List[Union[EdgeTable, EdgeType]], *, namespace_path: Optional[List[str]] = None, ) -> PropertyGraphDescription: @@ -966,6 +966,37 @@ class RemoteDBConnection(DBConnection): 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 diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index ed4b3b60c..e8ea07dd2 100644 --- a/python/python/tests/test_remote_db.py +++ b/python/python/tests/test_remote_db.py @@ -17,7 +17,14 @@ from packaging.version import Version import lancedb from lancedb.conftest import MockTextEmbeddingFunction -from lancedb.graph import EdgeTable, Endpoint, NodeTable, PropertyGraphDescription +from lancedb.graph import ( + EdgeTable, + EdgeType, + Endpoint, + NodeTable, + NodeType, + PropertyGraphDescription, +) from lancedb.query import ColumnOrdering from lancedb.remote import ClientConfig from lancedb.remote.errors import HttpError, RetryError @@ -2833,11 +2840,12 @@ 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 size and the table versions it was - # built from. + # and a description adds the graph's commits, its size and the table + # versions it was refreshed from. calls = [] person = {"table": "person", "column": "person_id"} definition = { + "mode": "materialized_view", "nodes": [ { "table": "person", @@ -2859,6 +2867,8 @@ def test_property_graph_crud_addresses_its_own_routes(): "name": "social", "namespace": ["analytics"], **definition, + "commit": "6f1c2d3e-0000-4000-8000-000000000000", + "committed_at": "2026-09-26T17:00:00Z", "vertex_count": 4, "edge_count": 5, "sources": [ @@ -2898,8 +2908,12 @@ def test_property_graph_crud_addresses_its_own_routes(): assert graph == PropertyGraphDescription( name="social", namespace_path=["analytics"], + mode="materialized_view", nodes=nodes, edges=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}, @@ -2918,6 +2932,96 @@ def test_property_graph_crud_addresses_its_own_routes(): ] +def test_an_independent_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 = [] + person = NodeType( + "Person", + key="person_id", + schema=pa.schema( + [ + pa.field("person_id", pa.int64(), nullable=False), + ("tags", pa.list_(pa.string())), + ] + ), + ) + knows = EdgeType( + "KNOWS", + source=("Person", "src_id"), + destination=("Person", "dst_id"), + schema=pa.schema([("since", pa.int32())]), + ) + 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"): + created.update(body) + response = { + "name": "people", + "namespace": [], + **created, + "commit": "c2", + "committed_at": "2026-09-26T17:00:00Z", + "previous_commit": "c1", + "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: + 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 inserted.vertex_count == 1 + db.refresh_property_graph("people") + db.rollback_property_graph("people") + + create, insert, refresh, rollback = calls + assert create[:2] == ("POST", "/v1/property_graph/people/create") + assert create[2]["mode"] == "independent" + assert create[2]["nodes"][0]["schema"]["fields"][0]["name"] == "person_id" + assert create[2]["edges"][0]["source"] == {"label": "Person", "column": "src_id"} + 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") + + 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_local_connections_refuse_property_graphs(tmp_path): db = lancedb.connect(tmp_path) with pytest.raises(NotImplementedError, match="Property graph operations"): diff --git a/python/src/connection.rs b/python/src/connection.rs index 707bb1e16..f5af99ef6 100644 --- a/python/src/connection.rs +++ b/python/src/connection.rs @@ -11,7 +11,7 @@ use crate::{ error::PythonErrorExt, namespace::{create_namespace_storage_options_provider, extract_namespace_arc}, runtime::future_into_py, - table::Table, + table::{Table, scannable::PyScannable}, }; use arrow::{ datatypes::Schema, @@ -1035,6 +1035,62 @@ impl Connection { }) } + #[pyo3(signature = (name, label, data, namespace_path=None))] + pub fn insert_into_property_graph( + self_: PyRef<'_, Self>, + name: String, + label: String, + data: PyScannable, + 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 + .insert_into_property_graph(name, label, data, &namespace_path) + .await + .infer_error()? + .to_json() + .infer_error() + }) + } + + #[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>, + 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 + .rollback_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 new file mode 100644 index 000000000..29df02ef6 --- /dev/null +++ b/python/src/graph.rs @@ -0,0 +1,27 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Property graph schemas, between pyarrow and the JSON form a graph's +//! definition carries. + +use arrow::{datatypes::Schema, pyarrow::PyArrowType}; +use lance_namespace::models::JsonArrowSchema; +use pyo3::{PyResult, exceptions::PyValueError, pyfunction}; + +/// A pyarrow schema as the JSON Arrow schema of a node or edge type. +#[pyfunction] +pub fn graph_schema_to_json(schema: PyArrowType) -> PyResult { + let json = lance_namespace::schema::arrow_schema_to_json(&schema.0) + .map_err(|error| PyValueError::new_err(error.to_string()))?; + serde_json::to_string(&json).map_err(|error| PyValueError::new_err(error.to_string())) +} + +/// A node or edge type's JSON Arrow schema as a pyarrow schema. +#[pyfunction] +pub fn graph_schema_from_json(json: String) -> PyResult> { + let json: JsonArrowSchema = + serde_json::from_str(&json).map_err(|error| PyValueError::new_err(error.to_string()))?; + let schema = lance_namespace::schema::convert_json_arrow_schema(&json) + .map_err(|error| PyValueError::new_err(error.to_string()))?; + Ok(PyArrowType(schema)) +} diff --git a/python/src/lib.rs b/python/src/lib.rs index ab04663c0..38adbe999 100644 --- a/python/src/lib.rs +++ b/python/src/lib.rs @@ -26,6 +26,7 @@ pub mod catalog; pub mod connection; pub mod error; pub mod expr; +pub mod graph; pub mod header; pub mod index; pub mod job; @@ -65,6 +66,8 @@ pub fn _lancedb(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_function(wrap_pyfunction!(catalog::connect_catalog, m)?)?; + m.add_function(wrap_pyfunction!(graph::graph_schema_to_json, m)?)?; + m.add_function(wrap_pyfunction!(graph::graph_schema_from_json, m)?)?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/python/src/table.rs b/python/src/table.rs index e7b6f1c6f..ae5a901df 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -37,7 +37,7 @@ use pyo3::{ types::{IntoPyDict, PyAnyMethods, PyBytes, PyDict, PyDictMethods, PyList, PyListMethods}, }; -mod scannable; +pub(crate) mod scannable; /// Convert `LsmStats` to a Python dict, preserving the per-bucket list. /// diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs index 5134f1bc8..d7bced436 100644 --- a/rust/lancedb/src/connection.rs +++ b/rust/lancedb/src/connection.rs @@ -867,13 +867,14 @@ impl Connection { self.internal.list_views(namespace_path).await } - /// Create a property graph: tables in its namespace read as the nodes and - /// edges of a graph. + /// Create a property graph. /// - /// The definition names the node tables with their keys and labels, and - /// the edge tables with the node keys their source and destination columns - /// reference. A name already taken is an error. Local databases return - /// [`Error::NotSupported`]. + /// 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`]. /// /// ```no_run /// # use lancedb::graph::{EdgeTable, Endpoint, EndpointReference, NodeTable, @@ -883,7 +884,7 @@ impl Connection { /// table: "person".to_string(), /// column: "person_id".to_string(), /// }; - /// let definition = PropertyGraphDefinition { + /// let definition = PropertyGraphDefinition::MaterializedView { /// nodes: vec![NodeTable { /// table: "person".to_string(), /// key: "person_id".to_string(), @@ -915,9 +916,10 @@ impl Connection { .await } - /// What this database records about one property graph: its definition, - /// its size, and the table versions it was built from. Local databases - /// return [`Error::NotSupported`]. + /// 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`]. pub async fn describe_property_graph( &self, name: impl AsRef, @@ -953,6 +955,65 @@ impl Connection { self.internal.list_property_graphs(namespace_path).await } + /// 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](Self::refresh_property_graph) instead. Local databases + /// return [`Error::NotSupported`]. + pub async fn insert_into_property_graph( + &self, + name: impl AsRef, + label: impl AsRef, + data: T, + namespace_path: &[String], + ) -> Result { + validate_property_graph_reference(name.as_ref(), namespace_path)?; + self.internal + .insert_into_property_graph( + name.as_ref(), + label.as_ref(), + Box::new(data), + namespace_path, + ) + .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 + /// is an error. Local databases return [`Error::NotSupported`]. + pub async fn rollback_property_graph( + &self, + name: impl AsRef, + namespace_path: &[String], + ) -> Result { + validate_property_graph_reference(name.as_ref(), namespace_path)?; + self.internal + .rollback_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 c1de55756..de76fb367 100644 --- a/rust/lancedb/src/database.rs +++ b/rust/lancedb/src/database.rs @@ -550,6 +550,33 @@ 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. + async fn insert_into_property_graph( + &self, + _name: &str, + _label: &str, + _data: Box, + _namespace_path: &[String], + ) -> 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, + _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 7b82b8a0b..64901bc24 100644 --- a/rust/lancedb/src/graph.rs +++ b/rust/lancedb/src/graph.rs @@ -1,20 +1,135 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright The LanceDB Authors -//! Property graphs: tables read as the nodes and edges of a graph. +//! Property graphs: catalog objects read as graphs, in one of two modes. //! -//! A property graph is a catalog object in a namespace, next to the tables it -//! reads. Its definition says which tables 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`. The tables themselves stay ordinary tables. +//! - Independent: the 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 +//! 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. //! //! The verbs live on [`crate::connection::Connection`]. +use arrow_schema::Schema; +use lance_namespace::models::JsonArrowSchema; use serde::{Deserialize, Serialize}; use crate::error::{Error, Result}; +/// A graph's definition, tagged with its mode. +#[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, + }, +} + +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}"), + }) + } +} + +/// The nodes of one label of an independent graph. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct NodeType { + /// The label its nodes carry. + pub label: String, + /// The field that identifies a node: a non-nullable integer, string or + /// binary field of `schema`. + pub key: String, + /// Every property, the key among them. + pub schema: JsonArrowSchema, +} + +impl NodeType { + pub fn new(label: impl Into, key: impl Into, schema: &Schema) -> Result { + Ok(Self { + label: label.into(), + key: key.into(), + schema: lance_namespace::schema::arrow_schema_to_json(schema)?, + }) + } + + /// The properties as an Arrow schema. + pub fn arrow_schema(&self) -> Result { + Ok(lance_namespace::schema::convert_json_arrow_schema( + &self.schema, + )?) + } +} + +/// The edges of one label of an independent graph. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct EdgeType { + /// The label its edges carry. + pub label: String, + /// The label of each edge's source node, and the column inserted rows + /// carry its key in. + pub source: EndpointType, + /// The label of each edge's destination node, and the column inserted + /// rows carry its key in. + pub destination: EndpointType, + /// The edge's properties, when it has any. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub schema: Option, +} + +impl EdgeType { + pub fn new( + label: impl Into, + source: EndpointType, + destination: EndpointType, + schema: Option<&Schema>, + ) -> Result { + Ok(Self { + label: label.into(), + source, + destination, + schema: schema + .map(lance_namespace::schema::arrow_schema_to_json) + .transpose()?, + }) + } + + /// The properties as an Arrow schema; empty when there are none. + pub fn arrow_schema(&self) -> Result { + match &self.schema { + Some(schema) => Ok(lance_namespace::schema::convert_json_arrow_schema(schema)?), + None => Ok(Schema::empty()), + } + } +} + +/// One end of an edge type. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct EndpointType { + /// The label of the node at this end. + pub label: String, + /// The column of inserted edge rows holding that node's key. + pub column: String, +} + /// A table whose rows are the nodes of one label. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct NodeTable { @@ -24,7 +139,8 @@ pub struct NodeTable { pub key: String, /// The label its nodes carry. pub label: String, - /// The columns exposed as node properties; every column when absent. + /// The columns exposed as node properties; every column when absent. A + /// created graph reports the list it resolved to. #[serde(default, skip_serializing_if = "Option::is_none")] pub properties: Option>, } @@ -40,7 +156,8 @@ pub struct EdgeTable { pub source: Endpoint, /// The column naming each edge's destination node. pub destination: Endpoint, - /// The columns exposed as edge properties; every column when absent. + /// The columns exposed as edge properties; every column when absent. A + /// created graph reports the list it resolved to. #[serde(default, skip_serializing_if = "Option::is_none")] pub properties: Option>, } @@ -63,55 +180,42 @@ pub struct EndpointReference { pub column: String, } -/// The node and edge tables that make up a property graph. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct PropertyGraphDefinition { - /// The node tables, one per label. - pub nodes: Vec, - /// The edge tables, 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}"), - }) - } -} - -/// The version of one table a graph was built from. +/// The version of one table a graph's current commit was refreshed from. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct GraphSourceVersion { /// The table. pub table: String, - /// Its version when the graph was built. + /// Its version when the graph was refreshed. pub version: u64, } /// What a database records about one property graph. /// /// Returned by [`crate::connection::Connection::describe_property_graph`], and -/// by `create_property_graph`. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +/// by every call that writes a graph. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct PropertyGraphDescription { /// 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 node tables. - pub nodes: Vec, - /// The edge tables. - #[serde(default)] - pub edges: Vec, - /// How many nodes the graph held when it was built. + /// The graph's mode and definition. + #[serde(flatten)] + pub definition: PropertyGraphDefinition, + /// The graph's current commit. + pub commit: String, + /// When the current commit landed, as RFC 3339. + pub committed_at: String, + /// The commit a rollback returns to. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub previous_commit: Option, + /// How many nodes the current commit holds. pub vertex_count: u64, - /// How many edges the graph held when it was built. + /// How many edges the current commit holds. pub edge_count: u64, - /// The table versions the graph was built from. + /// The table versions a materialized view's current commit was refreshed + /// from; empty for an independent graph. #[serde(default)] pub sources: Vec, } @@ -127,6 +231,8 @@ impl PropertyGraphDescription { #[cfg(test)] mod tests { + use arrow_schema::{DataType, Field}; + use super::*; fn social() -> PropertyGraphDefinition { @@ -134,7 +240,7 @@ mod tests { table: "person".to_string(), column: "person_id".to_string(), }; - PropertyGraphDefinition { + PropertyGraphDefinition::MaterializedView { nodes: vec![NodeTable { table: "person".to_string(), key: "person_id".to_string(), @@ -163,6 +269,7 @@ mod tests { assert_eq!( json, serde_json::json!({ + "mode": "materialized_view", "nodes": [{"table": "person", "key": "person_id", "label": "Person", "properties": ["name"]}], "edges": [{"table": "knows", "label": "KNOWS", @@ -177,7 +284,49 @@ mod tests { PropertyGraphDefinition::from_json(&json.to_string()).unwrap(), social() ); - assert!(PropertyGraphDefinition::from_json(r#"{"edges": []}"#).is_err()); + assert!(PropertyGraphDefinition::from_json(r#"{"nodes": []}"#).is_err()); + } + + #[test] + fn test_an_independent_definition_carries_its_schemas() { + let person = NodeType::new( + "Person", + "person_id", + &Schema::new(vec![ + Field::new("person_id", DataType::Int64, false), + Field::new("tags", DataType::new_list(DataType::Utf8, true), true), + ]), + ) + .unwrap(); + let knows = EdgeType::new( + "KNOWS", + EndpointType { + label: "Person".to_string(), + column: "src_id".to_string(), + }, + EndpointType { + label: "Person".to_string(), + column: "dst_id".to_string(), + }, + None, + ) + .unwrap(); + let definition = PropertyGraphDefinition::Independent { + nodes: vec![person.clone()], + edges: vec![knows], + }; + let json = serde_json::to_value(&definition).unwrap(); + assert_eq!(json["mode"], "independent"); + assert_eq!(json["nodes"][0]["schema"]["fields"][0]["name"], "person_id"); + assert!(json["edges"][0].get("schema").is_none()); + assert_eq!( + PropertyGraphDefinition::from_json(&json.to_string()).unwrap(), + definition + ); + assert_eq!( + person.arrow_schema().unwrap().field(1).data_type(), + &DataType::new_list(DataType::Utf8, true) + ); } #[tokio::test] @@ -204,6 +353,16 @@ mod tests { .await .unwrap_err(), ); + refused( + conn.refresh_property_graph("social", &[]) + .await + .unwrap_err(), + ); + refused( + conn.rollback_property_graph("social", &[]) + .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 ecbc410d8..50d1b7840 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -19,6 +19,7 @@ use lance_namespace::models::{ }; use crate::Error; +use crate::data::scannable::Scannable; use crate::database::{ CloneTableRequest, CreateTableMode, CreateTableRequest, Database, DatabaseOptions, JobInfo, OpenTableRequest, ReadConsistency, TableNamesRequest, @@ -1359,6 +1360,54 @@ impl Database for RemoteDatabase { } } + async fn insert_into_property_graph( + &self, + name: &str, + label: &str, + mut data: Box, + namespace_path: &[String], + ) -> Result { + let graph_id = build_object_identifier("Property graph name", name, namespace_path)?; + let body = stream_as_body(data.scan_as_stream())?; + let req = self + .client + .post(&format!("/v1/property_graph/{graph_id}/insert")) + .query(&[("label", label)]) + .body(body) + .header(CONTENT_TYPE, ARROW_STREAM_CONTENT_TYPE); + 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 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) + } + 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"); @@ -4396,7 +4445,8 @@ mod tests { fn social_graph() -> crate::graph::PropertyGraphDefinition { crate::graph::PropertyGraphDefinition::from_json( - r#"{"nodes": [{"table": "person", "key": "person_id", "label": "Person"}], + r#"{"mode": "materialized_view", + "nodes": [{"table": "person", "key": "person_id", "label": "Person"}], "edges": [{"table": "knows", "label": "KNOWS", "source": {"column": "src_id", "references": {"table": "person", "column": "person_id"}}, @@ -4413,6 +4463,11 @@ mod tests { 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( @@ -4444,8 +4499,9 @@ mod tests { .unwrap(); assert_eq!(graph.name, "social"); assert_eq!(graph.namespace_path, vec!["analytics".to_string()]); - assert_eq!(graph.nodes, social_graph().nodes); - assert_eq!(graph.edges, social_graph().edges); + 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], @@ -4456,6 +4512,52 @@ mod tests { ); } + #[tokio::test] + async fn test_graph_writes_post_to_their_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" => { + assert!(request.body().is_none(), "{:?}", request.body()); + } + other => panic!("unexpected path {other}"), + } + http::Response::builder() + .status(200) + .body(property_graph_description_body("people", &["analytics"])) + .unwrap() + }); + let namespace = ["analytics".to_string()]; + let rows = RecordBatch::try_new( + Arc::new(Schema::new(vec![Field::new( + "person_id", + DataType::Int32, + false, + )])), + vec![Arc::new(Int32Array::from(vec![1, 2]))], + ) + .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) + .await + .unwrap(); + } + #[tokio::test] async fn test_describe_and_drop_address_the_property_graph_in_the_path() { let conn = Connection::new_with_handler(|request| { @@ -4469,7 +4571,12 @@ mod tests { }); let graph = conn.describe_property_graph("social", &[]).await.unwrap(); assert!(graph.namespace_path.is_empty()); - assert_eq!(graph.edges[0].properties, Some(vec!["since".to_string()])); + 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);