rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-10-13 07:12:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: independent property graphs and graph writes from the client

Browse Source 
A property graph definition now carries its mode. Independent graphs are defined by node and edge types with Arrow schemas and take rows through insert_into_property_graph; materialized view graphs are defined over tables as before and brought up to them with refresh_property_graph. rollback_property_graph returns either to its previous commit, and descriptions report the mode and the current and previous commits.
This commit is contained in:
Jack Ye
2026-09-26 17:19:41 -07:00
parent b961b1695c
commit 0b1e274e05
14 changed files with 971 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/src/python/python.md
+13 -2
View File
@@ -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
python/python/lancedb/_lancedb.pyi
+15
View File
@@ -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: ...
python/python/lancedb/db.py
+167 -14
View File
@@ -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:
python/python/lancedb/graph.py
+138 -25
View File
@@ -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"],
python/python/lancedb/remote/db.py
+34 -3
View File
@@ -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
python/python/tests/test_remote_db.py
+107 -3
View File
@@ -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"):
python/src/connection.rs
+57 -1
View File
@@ -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<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
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<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
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<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
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<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
future_into_py(self_.py(), async move {
python/src/graph.rs
+27
View File
@@ -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<Schema>) -> PyResult<String> {
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<PyArrowType<Schema>> {
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))
}
python/src/lib.rs
+3
View File
@@ -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::<Connection>()?;
m.add_class::<catalog::Catalog>()?;
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::<Session>()?;
m.add_class::<Table>()?;
m.add_class::<crate::oauth::PyOAuthSession>()?;
python/src/table.rs
+1 -1
View File
@@ -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.
///
rust/lancedb/src/connection.rs
+71 -10
View File
@@ -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<str>,
@@ -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<T: Scannable + 'static>(
&self,
name: impl AsRef<str>,
label: impl AsRef<str>,
data: T,
namespace_path: &[String],
) -> Result<PropertyGraphDescription> {
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<str>,
namespace_path: &[String],
) -> Result<PropertyGraphDescription> {
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<str>,
namespace_path: &[String],
) -> Result<PropertyGraphDescription> {
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.
rust/lancedb/src/database.rs
+27
View File
@@ -550,6 +550,33 @@ pub trait Database:
async fn list_property_graphs(&self, _namespace_path: &[String]) -> Result<Vec<String>> {
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<dyn Scannable>,
_namespace_path: &[String],
) -> Result<PropertyGraphDescription> {
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<PropertyGraphDescription> {
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<PropertyGraphDescription> {
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.
rust/lancedb/src/graph.rs
+200 -41
View File
@@ -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<NodeType>,
/// The edge types, one per label.
#[serde(default)]
edges: Vec<EdgeType>,
},
/// The graph reads tables and is refreshed from them.
MaterializedView {
/// The node tables, one per label.
nodes: Vec<NodeTable>,
/// The edge tables, one per label.
#[serde(default)]
edges: Vec<EdgeTable>,
},
}
impl PropertyGraphDefinition {
/// Parse a definition from its JSON form.
pub fn from_json(json: &str) -> Result<Self> {
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<String>, key: impl Into<String>, schema: &Schema) -> Result<Self> {
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<Schema> {
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<JsonArrowSchema>,
}
impl EdgeType {
pub fn new(
label: impl Into<String>,
source: EndpointType,
destination: EndpointType,
schema: Option<&Schema>,
) -> Result<Self> {
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<Schema> {
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<Vec<String>>,
}
@@ -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<Vec<String>>,
}
@@ -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<NodeTable>,
/// The edge tables, one per label.
#[serde(default)]
pub edges: Vec<EdgeTable>,
}
impl PropertyGraphDefinition {
/// Parse a definition from its JSON form.
pub fn from_json(json: &str) -> Result<Self> {
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<String>,
/// The node tables.
pub nodes: Vec<NodeTable>,
/// The edge tables.
#[serde(default)]
pub edges: Vec<EdgeTable>,
/// 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<String>,
/// 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<GraphSourceVersion>,
}
@@ -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());
}
rust/lancedb/src/remote/db.rs
+111 -4
View File
@@ -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<S: HttpSend> Database for RemoteDatabase<S> {
}
}
async fn insert_into_property_graph(
&self,
name: &str,
label: &str,
mut data: Box<dyn Scannable>,
namespace_path: &[String],
) -> Result<PropertyGraphDescription> {
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<PropertyGraphDescription> {
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<PropertyGraphDescription> {
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<Vec<String>> {
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);