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

feat: property, virtual and materialized virtual property graphs in the client

Browse Source 
Each kind has its own create, open, describe, list and drop. Python create and open return handles: PropertyGraph.insert and rollback, and MaterializedVirtualPropertyGraph.refresh.
This commit is contained in:
Jack Ye
2026-09-29 01:00:11 -05:00
parent 6f68be2030
commit 973a793d18
13 changed files with 2092 additions and 831 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/python/lancedb/__init__.py
+13 -1
View File
@@ -48,7 +48,14 @@ from .materialized_view import (
MaterializedViewDefinition,
)
from .view import ViewDescription as ViewDescription
from .graph import PropertyGraphDescription as PropertyGraphDescription
from .graph import (
MaterializedVirtualPropertyGraph,
MaterializedVirtualPropertyGraphDescription,
PropertyGraph,
PropertyGraphDescription,
VirtualPropertyGraph,
VirtualPropertyGraphDescription,
)
from .table import AsyncTable, Table
from .types import BaseTokenizerType
from ._lancedb import Session
@@ -585,7 +592,12 @@ __all__ = [
"MaterializedView",
"MaterializedViewDefinition",
"ViewDescription",
"PropertyGraph",
"PropertyGraphDescription",
"VirtualPropertyGraph",
"VirtualPropertyGraphDescription",
"MaterializedVirtualPropertyGraph",
"MaterializedVirtualPropertyGraphDescription",
"connect",
"connect_async",
"tokenize",
python/python/lancedb/_lancedb.pyi
+10 -9
View File
@@ -188,20 +188,21 @@ class Connection(object):
async def list_views(
self, namespace_path: Optional[List[str]] = None
) -> List[str]: ...
async def create_property_graph(
async def create_graph(
self,
kind: str,
name: str,
definition_json: str,
namespace_path: Optional[List[str]] = None,
) -> str: ...
async def describe_property_graph(
self, name: str, namespace_path: Optional[List[str]] = None
async def describe_graph(
self, kind: str, name: str, namespace_path: Optional[List[str]] = None
) -> str: ...
async def drop_property_graph(
self, name: str, namespace_path: Optional[List[str]] = None
async def drop_graph(
self, kind: str, name: str, namespace_path: Optional[List[str]] = None
) -> None: ...
async def list_property_graphs(
self, namespace_path: Optional[List[str]] = None
async def list_graphs(
self, kind: str, namespace_path: Optional[List[str]] = None
) -> List[str]: ...
async def insert_into_property_graph(
self,
@@ -210,10 +211,10 @@ class Connection(object):
data: Any,
namespace_path: Optional[List[str]] = None,
) -> str: ...
async def refresh_property_graph(
async def rollback_property_graph(
self, name: str, namespace_path: Optional[List[str]] = None
) -> str: ...
async def rollback_property_graph(
async def refresh_materialized_virtual_property_graph(
self, name: str, namespace_path: Optional[List[str]] = None
) -> str: ...
async def list_jobs(self) -> List[JobInfo]: ...
python/python/lancedb/db.py
+339 -206
View File
@@ -50,15 +50,25 @@ from . import __version__
from ._lancedb import connect as lancedb_connect # type: ignore
from .functions import FunctionVersion, UdfDefinition
from .graph import (
MATERIALIZED_VIRTUAL_PROPERTY_GRAPH,
PROPERTY_GRAPH,
VIRTUAL_PROPERTY_GRAPH,
AsyncMaterializedVirtualPropertyGraph,
AsyncPropertyGraph,
AsyncVirtualPropertyGraph,
EdgeTable,
EdgeType,
MaterializedVirtualPropertyGraph,
MaterializedVirtualPropertyGraphDescription,
NodeTable,
NodeType,
PropertyGraph,
PropertyGraphDescription,
_definition_json,
_description_from_json,
VirtualPropertyGraph,
VirtualPropertyGraphDescription,
_tables_json,
_types_json,
)
from .scannable import to_scannable
from .job import AsyncJob, Job, _typed_job
from .sql import AsyncQuery as AsyncSqlQuery
from .sql import Query as SqlQuery
@@ -1031,81 +1041,25 @@ class DBConnection(EnforceOverrides):
def create_property_graph(
self,
name: str,
nodes: List[Union[NodeTable, NodeType]],
edges: List[Union[EdgeTable, EdgeType]],
nodes: List[NodeType],
edges: List[EdgeType],
*,
namespace_path: Optional[List[str]] = None,
) -> PropertyGraphDescription:
"""Create a property graph.
) -> PropertyGraph:
"""Create a property graph: a graph that holds its own rows.
With [NodeTable][lancedb.graph.NodeTable] and
[EdgeTable][lancedb.graph.EdgeTable], the graph is a materialized view
of tables in its namespace: each node table names its key column and
label, each edge table its label and the node keys its source and
destination columns reference, and the graph is built from them now.
With [NodeType][lancedb.graph.NodeType] and
[EdgeType][lancedb.graph.EdgeType], the graph is independent: it holds
its own rows, starts empty, and takes rows from
[insert_into_property_graph][lancedb.db.DBConnection.insert_into_property_graph].
A name already taken is an error. Local connections raise
Each [NodeType][lancedb.graph.NodeType] and
[EdgeType][lancedb.graph.EdgeType] gives one label's schema. The graph
starts empty and takes rows from
[PropertyGraph.insert][lancedb.graph.PropertyGraph.insert]. A name held
by a graph of any kind is taken. Local connections raise
``NotImplementedError``.
>>> import lancedb
>>> from lancedb.graph import EdgeTable, Endpoint, NodeTable
>>> db = lancedb.connect("db://my_database") # doctest: +SKIP
>>> graph = db.create_property_graph(
... "social",
... nodes=[NodeTable("person", key="person_id", label="Person")],
... edges=[
... EdgeTable(
... "knows",
... label="KNOWS",
... source=Endpoint("src_id", references=("person", "person_id")),
... destination=Endpoint(
... "dst_id", references=("person", "person_id")
... ),
... )
... ],
... ) # doctest: +SKIP
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def describe_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraphDescription:
"""What this database records about a property graph: its mode and
definition, its current and previous commits, its size, and, for a
materialized view, the table versions it was refreshed from.
Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def insert_into_property_graph(
self,
name: str,
label: str,
data: DATA,
*,
namespace_path: Optional[List[str]] = None,
) -> PropertyGraphDescription:
"""Insert rows into an independent property graph, as one commit.
The rows are one node label's or one edge label's. Nodes upsert by key:
a key the graph holds replaces that node's properties. Edges append,
carrying their endpoints' keys in the columns the edge type names; an
edge whose endpoint is not a node of the graph refuses the whole
insert. A materialized view graph is written through its tables and
refreshed instead. Local connections raise ``NotImplementedError``.
>>> import pyarrow as pa
>>> from lancedb.graph import EdgeType, NodeType
>>> db = lancedb.connect("db://my_database") # doctest: +SKIP
>>> db.create_property_graph(
>>> people = db.create_property_graph(
... "people",
... nodes=[
... NodeType(
@@ -1127,35 +1081,43 @@ class DBConnection(EnforceOverrides):
... )
... ],
... ) # doctest: +SKIP
>>> db.insert_into_property_graph(
... "people", "Person", [{"person_id": 1, "name": "Alice"}]
>>> people.insert(
... "Person", [{"person_id": 1, "name": "Alice"}]
... ) # doctest: +SKIP
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def refresh_property_graph(
def open_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraphDescription:
"""Bring a materialized view property graph up to its tables' latest
versions.
) -> PropertyGraph:
"""Open the property graph named ``name``; an error if there is none.
Until it is refreshed, a query over a table that has changed reads the
tables rather than the graph. Local connections raise
``NotImplementedError``.
Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def rollback_property_graph(
def describe_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraphDescription:
"""Return a property graph to its previous commit.
"""What this database records about a property graph: its node and edge
types, its current and previous commits, and its size.
The commit rolled back from is discarded, so a second rollback in a row
is an error. Local connections raise ``NotImplementedError``.
Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def list_property_graphs(
self, *, namespace_path: Optional[List[str]] = None
) -> List[str]:
"""The names of the property graphs in one namespace.
Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
@@ -1166,17 +1128,169 @@ class DBConnection(EnforceOverrides):
) -> None:
"""Drop a property graph and wait for its data to be deleted.
The tables it reads are untouched. Local connections raise
``NotImplementedError``.
Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def list_property_graphs(
def create_virtual_property_graph(
self,
name: str,
nodes: List[NodeTable],
edges: List[EdgeTable],
*,
namespace_path: Optional[List[str]] = None,
) -> VirtualPropertyGraph:
"""Create a virtual property graph over tables in its namespace.
Each [NodeTable][lancedb.graph.NodeTable] names the table holding one
label's nodes and its key column, and each
[EdgeTable][lancedb.graph.EdgeTable] the table holding one label's edges
and the node keys its source and destination columns reference. The
graph stores nothing else: a query reads the tables as they are then. A
name held by a graph of any kind is taken. Local connections raise
``NotImplementedError``.
>>> import lancedb
>>> from lancedb.graph import EdgeTable, Endpoint, NodeTable
>>> db = lancedb.connect("db://my_database") # doctest: +SKIP
>>> graph = db.create_virtual_property_graph(
... "social",
... nodes=[NodeTable("person", key="person_id", label="Person")],
... edges=[
... EdgeTable(
... "knows",
... label="KNOWS",
... source=Endpoint("src_id", references=("person", "person_id")),
... destination=Endpoint(
... "dst_id", references=("person", "person_id")
... ),
... )
... ],
... ) # doctest: +SKIP
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def open_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> VirtualPropertyGraph:
"""Open the virtual property graph named ``name``; an error if there is
none.
Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def describe_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> VirtualPropertyGraphDescription:
"""What this database records about a virtual property graph: its node
and edge tables, with the properties create resolved.
Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def list_virtual_property_graphs(
self, *, namespace_path: Optional[List[str]] = None
) -> List[str]:
"""The names of the property graphs in one namespace.
"""The names of the virtual property graphs in one namespace.
Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def drop_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> None:
"""Drop a virtual property graph and wait for its definition to be
deleted. The tables it reads are untouched.
Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def create_materialized_virtual_property_graph(
self,
name: str,
nodes: List[NodeTable],
edges: List[EdgeTable],
*,
namespace_path: Optional[List[str]] = None,
) -> MaterializedVirtualPropertyGraph:
"""Create a materialized virtual property graph over tables in its
namespace.
The definition is a virtual property graph's (see
[create_virtual_property_graph][lancedb.db.DBConnection.create_virtual_property_graph]).
The graph is built from the tables now, and read as that build until
[refreshed][lancedb.graph.MaterializedVirtualPropertyGraph.refresh]. A
name held by a graph of any kind is taken. Local connections raise
``NotImplementedError``.
>>> db = lancedb.connect("db://my_database") # doctest: +SKIP
>>> mvpg = db.create_materialized_virtual_property_graph(
... "social_mv", nodes, edges
... ) # doctest: +SKIP
>>> mvpg.refresh() # doctest: +SKIP
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def open_materialized_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> MaterializedVirtualPropertyGraph:
"""Open the materialized virtual property graph named ``name``; an error
if there is none.
Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def describe_materialized_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> MaterializedVirtualPropertyGraphDescription:
"""What this database records about a materialized virtual property
graph: its node and edge tables, and what its last refresh built from
them.
Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def list_materialized_virtual_property_graphs(
self, *, namespace_path: Optional[List[str]] = None
) -> List[str]:
"""The names of the materialized virtual property graphs in one
namespace.
Local connections raise ``NotImplementedError``.
"""
raise NotImplementedError(
"Property graph operations are not supported for this connection type"
)
def drop_materialized_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> None:
"""Drop a materialized virtual property graph and wait for its data to
be deleted. The tables it reads are untouched.
Local connections raise ``NotImplementedError``.
"""
@@ -2049,72 +2163,6 @@ class LanceDBConnection(DBConnection):
def list_views(self, *, namespace_path: Optional[List[str]] = None) -> List[str]:
return LOOP.run(self._conn.list_views(namespace_path=namespace_path))
@override
def create_property_graph(
self,
name: str,
nodes: List[Union[NodeTable, NodeType]],
edges: List[Union[EdgeTable, EdgeType]],
*,
namespace_path: Optional[List[str]] = None,
) -> PropertyGraphDescription:
return LOOP.run(
self._conn.create_property_graph(
name, nodes, edges, namespace_path=namespace_path
)
)
@override
def describe_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraphDescription:
return LOOP.run(
self._conn.describe_property_graph(name, namespace_path=namespace_path)
)
@override
def insert_into_property_graph(
self,
name: str,
label: str,
data: DATA,
*,
namespace_path: Optional[List[str]] = None,
) -> PropertyGraphDescription:
return LOOP.run(
self._conn.insert_into_property_graph(
name, label, data, namespace_path=namespace_path
)
)
@override
def refresh_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraphDescription:
return LOOP.run(
self._conn.refresh_property_graph(name, namespace_path=namespace_path)
)
@override
def rollback_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraphDescription:
return LOOP.run(
self._conn.rollback_property_graph(name, namespace_path=namespace_path)
)
@override
def drop_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> None:
LOOP.run(self._conn.drop_property_graph(name, namespace_path=namespace_path))
@override
def list_property_graphs(
self, *, namespace_path: Optional[List[str]] = None
) -> List[str]:
return LOOP.run(self._conn.list_property_graphs(namespace_path=namespace_path))
@override
def list_jobs(self) -> List[JobInfo]:
"""List server-side jobs across the database's tables."""
@@ -3145,77 +3193,162 @@ class AsyncConnection(object):
async def create_property_graph(
self,
name: str,
nodes: List[Union[NodeTable, NodeType]],
edges: List[Union[EdgeTable, EdgeType]],
nodes: List[NodeType],
edges: List[EdgeType],
*,
namespace_path: Optional[List[str]] = None,
) -> PropertyGraphDescription:
"""Create a property graph: a materialized view of tables in its
namespace, or an independent graph that holds its own rows.
) -> AsyncPropertyGraph:
"""Create a property graph: a graph that holds its own rows.
See
[DBConnection.create_property_graph][lancedb.DBConnection.create_property_graph].
"""
return _description_from_json(
await self._inner.create_property_graph(
name, _definition_json(nodes, edges), list(namespace_path or [])
)
graph = AsyncPropertyGraph(self._inner, name, namespace_path)
await self._inner.create_graph(
PROPERTY_GRAPH, name, _types_json(nodes, edges), graph.namespace_path
)
return graph
async def open_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> AsyncPropertyGraph:
"""Open the property graph named ``name``; an error if there is none."""
graph = AsyncPropertyGraph(self._inner, name, namespace_path)
await graph.describe()
return graph
async def describe_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraphDescription:
"""What this database records about a property graph."""
return _description_from_json(
await self._inner.describe_property_graph(name, list(namespace_path or []))
)
async def insert_into_property_graph(
self,
name: str,
label: str,
data: DATA,
*,
namespace_path: Optional[List[str]] = None,
) -> PropertyGraphDescription:
"""Insert rows of one label into an independent property graph.
See
[DBConnection.insert_into_property_graph][lancedb.DBConnection.insert_into_property_graph].
"""
return _description_from_json(
await self._inner.insert_into_property_graph(
name, label, to_scannable(data), list(namespace_path or [])
)
)
async def refresh_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraphDescription:
"""Bring a materialized view property graph up to its tables."""
return _description_from_json(
await self._inner.refresh_property_graph(name, list(namespace_path or []))
)
async def rollback_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraphDescription:
"""Return a property graph to its previous commit."""
return _description_from_json(
await self._inner.rollback_property_graph(name, list(namespace_path or []))
)
async def drop_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> None:
"""Drop a property graph and wait for its data to be deleted."""
await self._inner.drop_property_graph(name, list(namespace_path or []))
return await AsyncPropertyGraph(self._inner, name, namespace_path).describe()
async def list_property_graphs(
self, *, namespace_path: Optional[List[str]] = None
) -> List[str]:
"""The names of the property graphs in one namespace."""
return await self._inner.list_property_graphs(list(namespace_path or []))
return await self._inner.list_graphs(PROPERTY_GRAPH, list(namespace_path or []))
async def drop_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> None:
"""Drop a property graph and wait for its data to be deleted."""
await self._inner.drop_graph(PROPERTY_GRAPH, name, list(namespace_path or []))
async def create_virtual_property_graph(
self,
name: str,
nodes: List[NodeTable],
edges: List[EdgeTable],
*,
namespace_path: Optional[List[str]] = None,
) -> AsyncVirtualPropertyGraph:
"""Create a virtual property graph over tables in its namespace.
See
[DBConnection.create_virtual_property_graph][lancedb.DBConnection.create_virtual_property_graph].
"""
graph = AsyncVirtualPropertyGraph(self._inner, name, namespace_path)
await self._inner.create_graph(
VIRTUAL_PROPERTY_GRAPH,
name,
_tables_json(nodes, edges),
graph.namespace_path,
)
return graph
async def open_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> AsyncVirtualPropertyGraph:
"""Open the virtual property graph named ``name``; an error if there is
none."""
graph = AsyncVirtualPropertyGraph(self._inner, name, namespace_path)
await graph.describe()
return graph
async def describe_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> VirtualPropertyGraphDescription:
"""What this database records about a virtual property graph."""
return await AsyncVirtualPropertyGraph(
self._inner, name, namespace_path
).describe()
async def list_virtual_property_graphs(
self, *, namespace_path: Optional[List[str]] = None
) -> List[str]:
"""The names of the virtual property graphs in one namespace."""
return await self._inner.list_graphs(
VIRTUAL_PROPERTY_GRAPH, list(namespace_path or [])
)
async def drop_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> None:
"""Drop a virtual property graph and wait for its definition to be
deleted."""
await self._inner.drop_graph(
VIRTUAL_PROPERTY_GRAPH, name, list(namespace_path or [])
)
async def create_materialized_virtual_property_graph(
self,
name: str,
nodes: List[NodeTable],
edges: List[EdgeTable],
*,
namespace_path: Optional[List[str]] = None,
) -> AsyncMaterializedVirtualPropertyGraph:
"""Create a materialized virtual property graph, built from its tables
now.
See
[DBConnection.create_materialized_virtual_property_graph][lancedb.DBConnection.create_materialized_virtual_property_graph].
"""
graph = AsyncMaterializedVirtualPropertyGraph(self._inner, name, namespace_path)
await self._inner.create_graph(
MATERIALIZED_VIRTUAL_PROPERTY_GRAPH,
name,
_tables_json(nodes, edges),
graph.namespace_path,
)
return graph
async def open_materialized_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> AsyncMaterializedVirtualPropertyGraph:
"""Open the materialized virtual property graph named ``name``; an error
if there is none."""
graph = AsyncMaterializedVirtualPropertyGraph(self._inner, name, namespace_path)
await graph.describe()
return graph
async def describe_materialized_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> MaterializedVirtualPropertyGraphDescription:
"""What this database records about a materialized virtual property
graph."""
return await AsyncMaterializedVirtualPropertyGraph(
self._inner, name, namespace_path
).describe()
async def list_materialized_virtual_property_graphs(
self, *, namespace_path: Optional[List[str]] = None
) -> List[str]:
"""The names of the materialized virtual property graphs in one
namespace."""
return await self._inner.list_graphs(
MATERIALIZED_VIRTUAL_PROPERTY_GRAPH, list(namespace_path or [])
)
async def drop_materialized_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> None:
"""Drop a materialized virtual property graph and wait for its data to
be deleted."""
await self._inner.drop_graph(
MATERIALIZED_VIRTUAL_PROPERTY_GRAPH, name, list(namespace_path or [])
)
async def list_jobs(self) -> List[JobInfo]:
"""List server-side jobs across the database's tables."""
python/python/lancedb/graph.py
+342 -80
View File
@@ -1,31 +1,46 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""Property graphs: catalog objects read as graphs, in one of two modes.
"""Property graphs: catalog objects read as graphs, in three kinds that share
one namespace of names.
An *independent* graph holds its own rows. Its node types (:class:`NodeType`)
and edge types (:class:`EdgeType`) have schemas, and rows are inserted into the
graph itself with ``insert_into_property_graph``.
A *property graph* holds its own rows. Its node types (:class:`NodeType`) and
edge types (:class:`EdgeType`) have schemas, and rows are inserted into the
graph itself with :meth:`PropertyGraph.insert`.
A *materialized view* graph reads tables in its namespace. Its definition says
A *virtual property graph* reads tables in its namespace. Its definition says
which tables hold nodes and which hold edges: each node table's key and label
(:class:`NodeTable`), each edge table's label and how its source and
destination columns reference node keys (:class:`EdgeTable`) -- the shape of
SQL/PGQ's ``CREATE PROPERTY GRAPH``. The tables stay ordinary tables; after
they change, ``refresh_property_graph`` brings the graph up to date.
SQL/PGQ's ``CREATE PROPERTY GRAPH``. A query reads the tables as they are.
See ``DBConnection.create_property_graph``.
A *materialized virtual property graph* has a virtual graph's definition, and
is read as its last refresh built it: after the tables change,
:meth:`MaterializedVirtualPropertyGraph.refresh` brings it up to date.
See ``DBConnection.create_property_graph``,
``DBConnection.create_virtual_property_graph`` and
``DBConnection.create_materialized_virtual_property_graph``.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Tuple, Type
import pyarrow as pa
from ._lancedb import graph_schema_from_json, graph_schema_to_json
from .background_loop import LOOP
from .scannable import to_scannable
if TYPE_CHECKING:
from .common import DATA
PROPERTY_GRAPH = "property_graph"
VIRTUAL_PROPERTY_GRAPH = "virtual_property_graph"
MATERIALIZED_VIRTUAL_PROPERTY_GRAPH = "materialized_virtual_property_graph"
@dataclass(frozen=True)
@@ -72,7 +87,7 @@ class EdgeTable:
@dataclass(frozen=True)
class NodeType:
"""The nodes of one label of an independent graph."""
"""The nodes of one label of a property graph."""
label: str
"""The label its nodes carry."""
@@ -85,7 +100,7 @@ class NodeType:
@dataclass(frozen=True)
class EdgeType:
"""The edges of one label of an independent graph."""
"""The edges of one label of a property graph."""
label: str
"""The label its edges carry."""
@@ -107,12 +122,10 @@ class PropertyGraphDescription:
"""The graph's name within its namespace."""
namespace_path: List[str]
"""The namespace holding the graph; empty is the root namespace."""
mode: str
"""``"independent"`` or ``"materialized_view"``."""
nodes: List[Union[NodeTable, NodeType]]
"""The node tables (materialized view) or node types (independent)."""
edges: List[Union[EdgeTable, EdgeType]]
"""The edge tables (materialized view) or edge types (independent)."""
nodes: List[NodeType]
"""The node types."""
edges: List[EdgeType]
"""The edge types."""
commit: str
"""The graph's current commit."""
committed_at: str
@@ -123,52 +136,120 @@ class PropertyGraphDescription:
"""How many nodes the current commit holds."""
edge_count: int
"""How many edges the current commit holds."""
@dataclass(frozen=True)
class VirtualPropertyGraphDescription:
"""What a database records about one virtual property graph."""
name: str
"""The graph's name within its namespace."""
namespace_path: List[str]
"""The namespace holding the graph; empty is the root namespace."""
nodes: List[NodeTable]
"""The node tables, each with the properties create resolved."""
edges: List[EdgeTable]
"""The edge tables, each with the properties create resolved."""
@dataclass(frozen=True)
class MaterializedVirtualPropertyGraphDescription:
"""What a database records about one materialized virtual property graph."""
name: str
"""The graph's name within its namespace."""
namespace_path: List[str]
"""The namespace holding the graph; empty is the root namespace."""
nodes: List[NodeTable]
"""The node tables, each with the properties create resolved."""
edges: List[EdgeTable]
"""The edge tables, each with the properties create resolved."""
commit: str
"""The last refresh's commit."""
committed_at: str
"""When the last refresh landed, as RFC 3339."""
vertex_count: int
"""How many nodes the last refresh built."""
edge_count: int
"""How many edges the last refresh built."""
sources: Dict[str, int]
"""For a materialized view, the version of each table the current commit
was refreshed from, by table name."""
"""The version of each table the last refresh read, by table name."""
INDEPENDENT = "independent"
MATERIALIZED_VIEW = "materialized_view"
def _require(
kind: str,
nodes: Sequence[Any],
edges: Sequence[Any],
node_type: Type,
edge_type: Type,
) -> None:
for element, expected in [
*((node, node_type) for node in nodes),
*((edge, edge_type) for edge in edges),
]:
if not isinstance(element, expected):
raise TypeError(
f"A {kind}'s nodes are {node_type.__name__} and its edges "
f"{edge_type.__name__}, not {type(element).__name__}"
)
def _mode_of(nodes: Sequence[Any], edges: Sequence[Any]) -> str:
kinds = {type(element) for element in [*nodes, *edges]}
if kinds <= {NodeType, EdgeType} and kinds:
return INDEPENDENT
if kinds <= {NodeTable, EdgeTable}:
return MATERIALIZED_VIEW
raise ValueError(
"A property graph's nodes and edges are either all tables "
"(NodeTable, EdgeTable) or all types (NodeType, EdgeType)"
)
def _definition_json(nodes: Sequence[Any], edges: Sequence[Any]) -> str:
mode = _mode_of(nodes, edges)
def _types_json(nodes: Sequence[NodeType], edges: Sequence[EdgeType]) -> str:
_require("property graph", nodes, edges, NodeType, EdgeType)
return json.dumps(
{
"mode": mode,
"nodes": [_node_to_json(node) for node in nodes],
"edges": [_edge_to_json(edge) for edge in edges],
"nodes": [_node_type_to_json(node) for node in nodes],
"edges": [_edge_type_to_json(edge) for edge in edges],
}
)
def _description_from_json(text: str) -> PropertyGraphDescription:
def _tables_json(nodes: Sequence[NodeTable], edges: Sequence[EdgeTable]) -> str:
_require("virtual property graph", nodes, edges, NodeTable, EdgeTable)
return json.dumps(
{
"nodes": [_node_table_to_json(node) for node in nodes],
"edges": [_edge_table_to_json(edge) for edge in edges],
}
)
def _property_description(text: str) -> PropertyGraphDescription:
described = json.loads(text)
mode = described["mode"]
return PropertyGraphDescription(
name=described["name"],
namespace_path=list(described.get("namespace", [])),
mode=mode,
nodes=[_node_from_json(mode, node) for node in described["nodes"]],
edges=[_edge_from_json(mode, edge) for edge in described.get("edges", [])],
nodes=[_node_type_from_json(node) for node in described["nodes"]],
edges=[_edge_type_from_json(edge) for edge in described.get("edges", [])],
commit=described["commit"],
committed_at=described["committed_at"],
previous_commit=described.get("previous_commit"),
vertex_count=described["vertex_count"],
edge_count=described["edge_count"],
)
def _virtual_description(text: str) -> VirtualPropertyGraphDescription:
described = json.loads(text)
return VirtualPropertyGraphDescription(
name=described["name"],
namespace_path=list(described.get("namespace", [])),
nodes=[_node_table_from_json(node) for node in described["nodes"]],
edges=[_edge_table_from_json(edge) for edge in described.get("edges", [])],
)
def _materialized_description(text: str) -> MaterializedVirtualPropertyGraphDescription:
described = json.loads(text)
return MaterializedVirtualPropertyGraphDescription(
name=described["name"],
namespace_path=list(described.get("namespace", [])),
nodes=[_node_table_from_json(node) for node in described["nodes"]],
edges=[_edge_table_from_json(edge) for edge in described.get("edges", [])],
commit=described["commit"],
committed_at=described["committed_at"],
vertex_count=described["vertex_count"],
edge_count=described["edge_count"],
sources={
source["table"]: source["version"]
for source in described.get("sources", [])
@@ -184,13 +265,26 @@ def _schema_from_json(schema: Dict[str, Any]) -> pa.Schema:
return graph_schema_from_json(json.dumps(schema))
def _node_to_json(node: Union[NodeTable, NodeType]) -> Dict[str, Any]:
if isinstance(node, NodeType):
return {
"label": node.label,
"key": node.key,
"schema": _schema_to_json(node.schema),
}
def _node_type_to_json(node: NodeType) -> Dict[str, Any]:
return {
"label": node.label,
"key": node.key,
"schema": _schema_to_json(node.schema),
}
def _edge_type_to_json(edge: EdgeType) -> Dict[str, Any]:
encoded: Dict[str, Any] = {
"label": edge.label,
"source": {"label": edge.source[0], "column": edge.source[1]},
"destination": {"label": edge.destination[0], "column": edge.destination[1]},
}
if edge.schema is not None:
encoded["schema"] = _schema_to_json(edge.schema)
return encoded
def _node_table_to_json(node: NodeTable) -> Dict[str, Any]:
encoded: Dict[str, Any] = {
"table": node.table,
"key": node.key,
@@ -201,20 +295,8 @@ def _node_to_json(node: Union[NodeTable, NodeType]) -> Dict[str, Any]:
return encoded
def _edge_to_json(edge: Union[EdgeTable, EdgeType]) -> Dict[str, Any]:
if isinstance(edge, EdgeType):
encoded: Dict[str, Any] = {
"label": edge.label,
"source": {"label": edge.source[0], "column": edge.source[1]},
"destination": {
"label": edge.destination[0],
"column": edge.destination[1],
},
}
if edge.schema is not None:
encoded["schema"] = _schema_to_json(edge.schema)
return encoded
encoded = {
def _edge_table_to_json(edge: EdgeTable) -> Dict[str, Any]:
encoded: Dict[str, Any] = {
"table": edge.table,
"label": edge.label,
"source": _endpoint_to_json(edge.source),
@@ -233,13 +315,23 @@ def _endpoint_to_json(endpoint: Endpoint) -> Dict[str, Any]:
}
def _node_from_json(mode: str, node: Dict[str, Any]) -> Union[NodeTable, NodeType]:
if mode == INDEPENDENT:
return NodeType(
label=node["label"],
key=node["key"],
schema=_schema_from_json(node["schema"]),
)
def _node_type_from_json(node: Dict[str, Any]) -> NodeType:
return NodeType(
label=node["label"], key=node["key"], schema=_schema_from_json(node["schema"])
)
def _edge_type_from_json(edge: Dict[str, Any]) -> EdgeType:
schema = edge.get("schema")
return EdgeType(
label=edge["label"],
source=(edge["source"]["label"], edge["source"]["column"]),
destination=(edge["destination"]["label"], edge["destination"]["column"]),
schema=None if schema is None else _schema_from_json(schema),
)
def _node_table_from_json(node: Dict[str, Any]) -> NodeTable:
return NodeTable(
table=node["table"],
key=node["key"],
@@ -248,15 +340,7 @@ def _node_from_json(mode: str, node: Dict[str, Any]) -> Union[NodeTable, NodeTyp
)
def _edge_from_json(mode: str, edge: Dict[str, Any]) -> Union[EdgeTable, EdgeType]:
if mode == INDEPENDENT:
schema = edge.get("schema")
return EdgeType(
label=edge["label"],
source=(edge["source"]["label"], edge["source"]["column"]),
destination=(edge["destination"]["label"], edge["destination"]["column"]),
schema=None if schema is None else _schema_from_json(schema),
)
def _edge_table_from_json(edge: Dict[str, Any]) -> EdgeTable:
return EdgeTable(
table=edge["table"],
label=edge["label"],
@@ -272,3 +356,181 @@ def _endpoint_from_json(endpoint: Dict[str, Any]) -> Endpoint:
column=endpoint["column"],
references=(references["table"], references["column"]),
)
class _AsyncGraph:
"""A handle on one graph of a database: its name and namespace, and the
connection it is read and written through."""
_kind: str = PROPERTY_GRAPH
def __init__(self, inner: Any, name: str, namespace_path: Optional[List[str]]):
self._inner = inner
self._name = name
self._namespace_path = list(namespace_path or [])
def __repr__(self) -> str:
return f"{type(self).__name__}(name={self._name!r})"
@property
def name(self) -> str:
"""The graph's name within its namespace."""
return self._name
@property
def namespace_path(self) -> List[str]:
"""The namespace holding the graph; empty is the root namespace."""
return list(self._namespace_path)
async def _describe(self) -> str:
return await self._inner.describe_graph(
self._kind, self._name, self._namespace_path
)
class AsyncPropertyGraph(_AsyncGraph):
"""A property graph: one that holds its own rows.
Obtained from ``AsyncConnection.create_property_graph`` or
``AsyncConnection.open_property_graph``.
"""
_kind = PROPERTY_GRAPH
async def describe(self) -> PropertyGraphDescription:
"""The graph's node and edge types, current commit and size."""
return _property_description(await self._describe())
async def insert(self, label: str, data: "DATA") -> PropertyGraphDescription:
"""Insert rows of one node label or edge label, as one commit.
Nodes upsert by key: a key the graph holds replaces that node's
properties. Edges append, carrying their endpoints' keys in the columns
the edge type names; an edge whose endpoint is not a node of the graph
refuses the whole insert.
"""
return _property_description(
await self._inner.insert_into_property_graph(
self._name, label, to_scannable(data), self._namespace_path
)
)
async def rollback(self) -> PropertyGraphDescription:
"""Return the graph to its previous commit.
The commit rolled back from is discarded, so a second rollback in a row
is an error.
"""
return _property_description(
await self._inner.rollback_property_graph(self._name, self._namespace_path)
)
class AsyncVirtualPropertyGraph(_AsyncGraph):
"""A virtual property graph: one defined over tables and read through them.
Obtained from ``AsyncConnection.create_virtual_property_graph`` or
``AsyncConnection.open_virtual_property_graph``.
"""
_kind = VIRTUAL_PROPERTY_GRAPH
async def describe(self) -> VirtualPropertyGraphDescription:
"""The graph's node and edge tables."""
return _virtual_description(await self._describe())
class AsyncMaterializedVirtualPropertyGraph(_AsyncGraph):
"""A materialized virtual property graph: one defined over tables and read
as its last refresh built it.
Obtained from ``AsyncConnection.create_materialized_virtual_property_graph``
or ``AsyncConnection.open_materialized_virtual_property_graph``.
"""
_kind = MATERIALIZED_VIRTUAL_PROPERTY_GRAPH
async def describe(self) -> MaterializedVirtualPropertyGraphDescription:
"""The graph's node and edge tables, and what its last refresh built."""
return _materialized_description(await self._describe())
async def refresh(self) -> MaterializedVirtualPropertyGraphDescription:
"""Rebuild the graph from its tables' latest versions.
When every table's commits since the last refresh left its rows as
they were -- compaction, indexing, metadata -- only the recorded
versions advance; when nothing changed, the commit stays.
"""
return _materialized_description(
await self._inner.refresh_materialized_virtual_property_graph(
self._name, self._namespace_path
)
)
class _Graph:
"""Synchronous variant of a graph handle."""
def __init__(self, graph: _AsyncGraph):
self._async = graph
def __repr__(self) -> str:
return f"{type(self).__name__}(name={self.name!r})"
@property
def name(self) -> str:
"""The graph's name within its namespace."""
return self._async.name
@property
def namespace_path(self) -> List[str]:
"""The namespace holding the graph; empty is the root namespace."""
return self._async.namespace_path
class PropertyGraph(_Graph):
"""Synchronous variant of
[AsyncPropertyGraph][lancedb.graph.AsyncPropertyGraph]."""
_async: AsyncPropertyGraph
def describe(self) -> PropertyGraphDescription:
"""The graph's node and edge types, current commit and size."""
return LOOP.run(self._async.describe())
def insert(self, label: str, data: "DATA") -> PropertyGraphDescription:
"""Insert rows of one node label or edge label, as one commit. See
[AsyncPropertyGraph.insert][lancedb.graph.AsyncPropertyGraph.insert]."""
return LOOP.run(self._async.insert(label, data))
def rollback(self) -> PropertyGraphDescription:
"""Return the graph to its previous commit. See
[AsyncPropertyGraph.rollback][lancedb.graph.AsyncPropertyGraph.rollback]."""
return LOOP.run(self._async.rollback())
class VirtualPropertyGraph(_Graph):
"""Synchronous variant of
[AsyncVirtualPropertyGraph][lancedb.graph.AsyncVirtualPropertyGraph]."""
_async: AsyncVirtualPropertyGraph
def describe(self) -> VirtualPropertyGraphDescription:
"""The graph's node and edge tables."""
return LOOP.run(self._async.describe())
class MaterializedVirtualPropertyGraph(_Graph):
"""Synchronous variant of
[AsyncMaterializedVirtualPropertyGraph][lancedb.graph.AsyncMaterializedVirtualPropertyGraph]."""
_async: AsyncMaterializedVirtualPropertyGraph
def describe(self) -> MaterializedVirtualPropertyGraphDescription:
"""The graph's node and edge tables, and what its last refresh built."""
return LOOP.run(self._async.describe())
def refresh(self) -> MaterializedVirtualPropertyGraphDescription:
"""Rebuild the graph from its tables' latest versions. See
[AsyncMaterializedVirtualPropertyGraph.refresh][lancedb.graph.AsyncMaterializedVirtualPropertyGraph.refresh]."""
return LOOP.run(self._async.refresh())
python/python/lancedb/remote/db.py
+144 -38
View File
@@ -36,7 +36,18 @@ import pyarrow as pa
from ..common import DATA
from ..db import DBConnection, LOOP
from ..functions import FunctionVersion, UdfDefinition
from ..graph import EdgeTable, EdgeType, NodeTable, NodeType, PropertyGraphDescription
from ..graph import (
EdgeTable,
EdgeType,
MaterializedVirtualPropertyGraph,
MaterializedVirtualPropertyGraphDescription,
NodeTable,
NodeType,
PropertyGraph,
PropertyGraphDescription,
VirtualPropertyGraph,
VirtualPropertyGraphDescription,
)
from ..job import AsyncJob, Job
from ..sql import Query as SqlQuery
from ..sql import QueryDescription
@@ -947,14 +958,26 @@ class RemoteDBConnection(DBConnection):
def create_property_graph(
self,
name: str,
nodes: List[Union[NodeTable, NodeType]],
edges: List[Union[EdgeTable, EdgeType]],
nodes: List[NodeType],
edges: List[EdgeType],
*,
namespace_path: Optional[List[str]] = None,
) -> PropertyGraphDescription:
return LOOP.run(
self._conn.create_property_graph(
name, nodes, edges, namespace_path=namespace_path
) -> PropertyGraph:
return PropertyGraph(
LOOP.run(
self._conn.create_property_graph(
name, nodes, edges, namespace_path=namespace_path
)
)
)
@override
def open_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraph:
return PropertyGraph(
LOOP.run(
self._conn.open_property_graph(name, namespace_path=namespace_path)
)
)
@@ -967,35 +990,10 @@ class RemoteDBConnection(DBConnection):
)
@override
def insert_into_property_graph(
self,
name: str,
label: str,
data: DATA,
*,
namespace_path: Optional[List[str]] = None,
) -> PropertyGraphDescription:
return LOOP.run(
self._conn.insert_into_property_graph(
name, label, data, namespace_path=namespace_path
)
)
@override
def refresh_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraphDescription:
return LOOP.run(
self._conn.refresh_property_graph(name, namespace_path=namespace_path)
)
@override
def rollback_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> PropertyGraphDescription:
return LOOP.run(
self._conn.rollback_property_graph(name, namespace_path=namespace_path)
)
def list_property_graphs(
self, *, namespace_path: Optional[List[str]] = None
) -> List[str]:
return LOOP.run(self._conn.list_property_graphs(namespace_path=namespace_path))
@override
def drop_property_graph(
@@ -1004,10 +1002,118 @@ class RemoteDBConnection(DBConnection):
LOOP.run(self._conn.drop_property_graph(name, namespace_path=namespace_path))
@override
def list_property_graphs(
def create_virtual_property_graph(
self,
name: str,
nodes: List[NodeTable],
edges: List[EdgeTable],
*,
namespace_path: Optional[List[str]] = None,
) -> VirtualPropertyGraph:
return VirtualPropertyGraph(
LOOP.run(
self._conn.create_virtual_property_graph(
name, nodes, edges, namespace_path=namespace_path
)
)
)
@override
def open_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> VirtualPropertyGraph:
return VirtualPropertyGraph(
LOOP.run(
self._conn.open_virtual_property_graph(
name, namespace_path=namespace_path
)
)
)
@override
def describe_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> VirtualPropertyGraphDescription:
return LOOP.run(
self._conn.describe_virtual_property_graph(
name, namespace_path=namespace_path
)
)
@override
def list_virtual_property_graphs(
self, *, namespace_path: Optional[List[str]] = None
) -> List[str]:
return LOOP.run(self._conn.list_property_graphs(namespace_path=namespace_path))
return LOOP.run(
self._conn.list_virtual_property_graphs(namespace_path=namespace_path)
)
@override
def drop_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> None:
LOOP.run(
self._conn.drop_virtual_property_graph(name, namespace_path=namespace_path)
)
@override
def create_materialized_virtual_property_graph(
self,
name: str,
nodes: List[NodeTable],
edges: List[EdgeTable],
*,
namespace_path: Optional[List[str]] = None,
) -> MaterializedVirtualPropertyGraph:
return MaterializedVirtualPropertyGraph(
LOOP.run(
self._conn.create_materialized_virtual_property_graph(
name, nodes, edges, namespace_path=namespace_path
)
)
)
@override
def open_materialized_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> MaterializedVirtualPropertyGraph:
return MaterializedVirtualPropertyGraph(
LOOP.run(
self._conn.open_materialized_virtual_property_graph(
name, namespace_path=namespace_path
)
)
)
@override
def describe_materialized_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> MaterializedVirtualPropertyGraphDescription:
return LOOP.run(
self._conn.describe_materialized_virtual_property_graph(
name, namespace_path=namespace_path
)
)
@override
def list_materialized_virtual_property_graphs(
self, *, namespace_path: Optional[List[str]] = None
) -> List[str]:
return LOOP.run(
self._conn.list_materialized_virtual_property_graphs(
namespace_path=namespace_path
)
)
@override
def drop_materialized_virtual_property_graph(
self, name: str, *, namespace_path: Optional[List[str]] = None
) -> None:
LOOP.run(
self._conn.drop_materialized_virtual_property_graph(
name, namespace_path=namespace_path
)
)
@override
def list_jobs(self) -> List["JobInfo"]:
python/python/tests/test_remote_db.py
+184 -112
View File
@@ -21,9 +21,13 @@ from lancedb.graph import (
EdgeTable,
EdgeType,
Endpoint,
MaterializedVirtualPropertyGraph,
MaterializedVirtualPropertyGraphDescription,
NodeTable,
NodeType,
PropertyGraphDescription,
PropertyGraph,
VirtualPropertyGraph,
VirtualPropertyGraphDescription,
)
from lancedb.query import ColumnOrdering
from lancedb.remote import ClientConfig
@@ -2838,35 +2842,68 @@ def test_view_crud_addresses_its_own_routes():
]
def test_property_graph_crud_addresses_its_own_routes():
# The graph verbs are their own routes: the definition is the create body,
# and a description adds the graph's commits, its size and the table
# versions it was refreshed from.
def _graph_handler(calls, respond):
"""A mock route handler recording `(method, path, body)` and answering
with `respond(path, body)` as JSON."""
def handler(request):
length = int(request.headers.get("Content-Length") or 0)
raw = request.rfile.read(length) if length else b""
if request.path.split("?")[0].endswith("/insert"):
body = pa.ipc.open_stream(raw).read_all().to_pylist()
else:
body = json.loads(raw) if raw else None
calls.append((request.command, request.path, body))
request.send_response(200)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(json.dumps(respond(request.path, body)).encode())
return handler
SOCIAL_NODES = [
NodeTable("person", key="person_id", label="Person", properties=["name"])
]
SOCIAL_EDGES = [
EdgeTable(
"knows",
label="KNOWS",
source=Endpoint("src_id", references=("person", "person_id")),
destination=Endpoint("dst_id", references=("person", "person_id")),
)
]
SOCIAL_DEFINITION = {
"nodes": [
{
"table": "person",
"key": "person_id",
"label": "Person",
"properties": ["name"],
}
],
"edges": [
{
"table": "knows",
"label": "KNOWS",
"source": {
"column": "src_id",
"references": {"table": "person", "column": "person_id"},
},
"destination": {
"column": "dst_id",
"references": {"table": "person", "column": "person_id"},
},
}
],
}
def test_virtual_and_materialized_graphs_address_their_own_routes():
# Each kind of graph has its own routes; the definition is the create
# body, and a materialized graph's description adds its last refresh.
calls = []
person = {"table": "person", "column": "person_id"}
definition = {
"mode": "materialized_view",
"nodes": [
{
"table": "person",
"key": "person_id",
"label": "Person",
"properties": ["name"],
}
],
"edges": [
{
"table": "knows",
"label": "KNOWS",
"source": {"column": "src_id", "references": person},
"destination": {"column": "dst_id", "references": person},
}
],
}
described = {
"name": "social",
"namespace": ["analytics"],
**definition,
refreshed = {
"commit": "6f1c2d3e-0000-4000-8000-000000000000",
"committed_at": "2026-09-26T17:00:00Z",
"vertex_count": 4,
@@ -2877,62 +2914,106 @@ def test_property_graph_crud_addresses_its_own_routes():
],
}
def handler(request):
length = int(request.headers.get("Content-Length") or 0)
body = json.loads(request.rfile.read(length)) if length else None
calls.append((request.command, request.path, body))
if request.path.endswith("/property_graph/list"):
response = {"property_graphs": ["social"], "page_token": None}
elif request.path.endswith("/drop"):
response = {}
else:
response = described
request.send_response(200)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(json.dumps(response).encode())
def respond(path, body):
if path.endswith("/virtual_property_graph/list"):
return {"virtual_property_graphs": ["social"], "page_token": None}
if path.endswith("/materialized_virtual_property_graph/list"):
return {"materialized_virtual_property_graphs": ["social_mv"]}
if path.endswith("/drop"):
return {}
name = path.split("$")[1].split("/")[0]
described = {"name": name, "namespace": ["analytics"], **SOCIAL_DEFINITION}
if "/materialized_virtual_property_graph/" in path:
described.update(refreshed)
return described
nodes = [NodeTable("person", key="person_id", label="Person", properties=["name"])]
edges = [
EdgeTable(
"knows",
label="KNOWS",
source=Endpoint("src_id", references=("person", "person_id")),
destination=Endpoint("dst_id", references=("person", "person_id")),
with mock_lancedb_connection(_graph_handler(calls, respond)) as db:
graph = db.create_virtual_property_graph(
"social", SOCIAL_NODES, SOCIAL_EDGES, namespace_path=["analytics"]
)
]
with mock_lancedb_connection(handler) as db:
graph = db.create_property_graph(
"social", nodes, edges, namespace_path=["analytics"]
)
assert graph == PropertyGraphDescription(
assert isinstance(graph, VirtualPropertyGraph)
assert (graph.name, graph.namespace_path) == ("social", ["analytics"])
assert graph.describe() == VirtualPropertyGraphDescription(
name="social",
namespace_path=["analytics"],
mode="materialized_view",
nodes=nodes,
edges=edges,
nodes=SOCIAL_NODES,
edges=SOCIAL_EDGES,
)
assert db.list_virtual_property_graphs(namespace_path=["analytics"]) == [
"social"
]
db.drop_virtual_property_graph("social", namespace_path=["analytics"])
mvpg = db.create_materialized_virtual_property_graph(
"social_mv", SOCIAL_NODES, SOCIAL_EDGES, namespace_path=["analytics"]
)
assert isinstance(mvpg, MaterializedVirtualPropertyGraph)
expected = MaterializedVirtualPropertyGraphDescription(
name="social_mv",
namespace_path=["analytics"],
nodes=SOCIAL_NODES,
edges=SOCIAL_EDGES,
commit="6f1c2d3e-0000-4000-8000-000000000000",
committed_at="2026-09-26T17:00:00Z",
previous_commit=None,
vertex_count=4,
edge_count=5,
sources={"person": 3, "knows": 2},
)
assert db.describe_property_graph("social", namespace_path=["analytics"]) == (
graph
assert mvpg.refresh() == expected
opened = db.open_materialized_virtual_property_graph(
"social_mv", namespace_path=["analytics"]
)
assert opened.describe() == expected
assert db.list_materialized_virtual_property_graphs(
namespace_path=["analytics"]
) == ["social_mv"]
db.drop_materialized_virtual_property_graph(
"social_mv", namespace_path=["analytics"]
)
assert db.list_property_graphs(namespace_path=["analytics"]) == ["social"]
db.drop_property_graph("social", namespace_path=["analytics"])
assert calls == [
("POST", "/v1/property_graph/analytics$social/create", definition),
("POST", "/v1/property_graph/analytics$social/describe", None),
("GET", "/v1/namespace/analytics/property_graph/list", None),
("POST", "/v1/property_graph/analytics$social/drop", None),
(
"POST",
"/v1/virtual_property_graph/analytics$social/create",
SOCIAL_DEFINITION,
),
("POST", "/v1/virtual_property_graph/analytics$social/describe", None),
("GET", "/v1/namespace/analytics/virtual_property_graph/list", None),
("POST", "/v1/virtual_property_graph/analytics$social/drop", None),
(
"POST",
"/v1/materialized_virtual_property_graph/analytics$social_mv/create",
SOCIAL_DEFINITION,
),
(
"POST",
"/v1/materialized_virtual_property_graph/analytics$social_mv/refresh",
None,
),
(
"POST",
"/v1/materialized_virtual_property_graph/analytics$social_mv/describe",
None,
),
(
"POST",
"/v1/materialized_virtual_property_graph/analytics$social_mv/describe",
None,
),
(
"GET",
"/v1/namespace/analytics/materialized_virtual_property_graph/list",
None,
),
(
"POST",
"/v1/materialized_virtual_property_graph/analytics$social_mv/drop",
None,
),
]
def test_an_independent_graph_carries_schemas_and_takes_arrow_rows():
def test_a_property_graph_carries_schemas_and_takes_arrow_rows():
# Node and edge types go up as JSON Arrow schemas and come back as pyarrow
# schemas; inserted rows go up as an Arrow IPC stream.
calls = []
@@ -2954,17 +3035,12 @@ def test_an_independent_graph_carries_schemas_and_takes_arrow_rows():
)
created = {}
def handler(request):
length = int(request.headers.get("Content-Length") or 0)
raw = request.rfile.read(length) if length else b""
if request.path.split("?")[0].endswith("/insert"):
body = pa.ipc.open_stream(raw).read_all().to_pylist()
else:
body = json.loads(raw) if raw else None
calls.append((request.command, request.path, body))
if request.path.endswith("/create"):
def respond(path, body):
if path.endswith("/create"):
created.update(body)
response = {
if path.endswith("/property_graph/list"):
return {"property_graphs": ["people"]}
return {
"name": "people",
"namespace": [],
**created,
@@ -2974,52 +3050,46 @@ def test_an_independent_graph_carries_schemas_and_takes_arrow_rows():
"vertex_count": 1,
"edge_count": 0,
}
request.send_response(200)
request.send_header("Content-Type", "application/json")
request.end_headers()
request.wfile.write(json.dumps(response).encode())
with mock_lancedb_connection(handler) as db:
with mock_lancedb_connection(_graph_handler(calls, respond)) as db:
graph = db.create_property_graph("people", [person], [knows])
assert graph.mode == "independent"
assert graph.nodes == [person] and graph.edges == [knows]
assert graph.previous_commit == "c1"
inserted = db.insert_into_property_graph(
"people", "Person", [{"person_id": 1, "tags": ["a"]}]
)
assert isinstance(graph, PropertyGraph)
described = graph.describe()
assert described.nodes == [person] and described.edges == [knows]
assert described.previous_commit == "c1"
inserted = graph.insert("Person", [{"person_id": 1, "tags": ["a"]}])
assert inserted.vertex_count == 1
db.refresh_property_graph("people")
db.rollback_property_graph("people")
graph.rollback()
assert db.list_property_graphs() == ["people"]
db.drop_property_graph("people")
create, insert, refresh, rollback = calls
create, describe, insert, rollback, listed, drop = calls
assert create[:2] == ("POST", "/v1/property_graph/people/create")
assert create[2]["mode"] == "independent"
assert "mode" not in create[2]
assert create[2]["nodes"][0]["schema"]["fields"][0]["name"] == "person_id"
assert create[2]["edges"][0]["source"] == {"label": "Person", "column": "src_id"}
assert describe[:2] == ("POST", "/v1/property_graph/people/describe")
assert insert == (
"POST",
"/v1/property_graph/people/insert?label=Person",
[{"person_id": 1, "tags": ["a"]}],
)
assert refresh[:2] == ("POST", "/v1/property_graph/people/refresh")
assert rollback[:2] == ("POST", "/v1/property_graph/people/rollback")
assert listed[:2] == ("GET", "/v1/namespace/$/property_graph/list")
assert drop[:2] == ("POST", "/v1/property_graph/people/drop")
with pytest.raises(ValueError, match="either all tables"):
with mock_lancedb_connection(handler) as db:
db.create_property_graph(
"mixed",
[person],
[
EdgeTable(
"knows",
label="KNOWS",
source=Endpoint("src_id", references=("person", "person_id")),
destination=Endpoint(
"dst_id", references=("person", "person_id")
),
)
],
)
def test_a_graph_takes_only_its_own_kind_of_definition():
person = NodeType(
"Person",
key="person_id",
schema=pa.schema([pa.field("person_id", pa.int64(), nullable=False)]),
)
with mock_lancedb_connection(_graph_handler([], lambda path, body: {})) as db:
with pytest.raises(TypeError, match="are NodeTable and its edges EdgeTable"):
db.create_virtual_property_graph("mixed", [person], SOCIAL_EDGES)
with pytest.raises(TypeError, match="NodeType"):
db.create_property_graph("mixed", SOCIAL_NODES, [])
def test_local_connections_refuse_property_graphs(tmp_path):
@@ -3027,4 +3097,6 @@ def test_local_connections_refuse_property_graphs(tmp_path):
with pytest.raises(NotImplementedError, match="Property graph operations"):
db.list_property_graphs()
with pytest.raises(NotImplementedError, match="Property graph operations"):
db.describe_property_graph("social")
db.describe_virtual_property_graph("social")
with pytest.raises(NotImplementedError, match="Property graph operations"):
db.open_materialized_virtual_property_graph("social")
python/src/connection.rs
+120 -48
View File
@@ -9,6 +9,7 @@ use std::{
use crate::{
error::PythonErrorExt,
graph::GraphKind,
namespace::{create_namespace_storage_options_provider, extract_namespace_arc},
runtime::future_into_py,
table::{Table, scannable::PyScannable},
@@ -965,73 +966,144 @@ impl Connection {
})
}
#[pyo3(signature = (name, definition_json, namespace_path=None))]
pub fn create_property_graph(
/// Create a graph of `kind` from its definition's JSON, and return its
/// description's JSON.
#[pyo3(signature = (kind, name, definition_json, namespace_path=None))]
pub fn create_graph(
self_: PyRef<'_, Self>,
kind: String,
name: String,
definition_json: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
let definition =
lancedb::graph::PropertyGraphDefinition::from_json(&definition_json).infer_error()?;
let kind = GraphKind::parse(&kind)?;
future_into_py(self_.py(), async move {
inner
.create_property_graph(name, &definition, &namespace_path)
.await
.infer_error()?
.to_json()
.infer_error()
match kind {
GraphKind::Property => {
let definition =
lancedb::graph::PropertyGraphDefinition::from_json(&definition_json)
.infer_error()?;
inner
.create_property_graph(name, &definition, &namespace_path)
.await
.infer_error()?
.to_json()
}
GraphKind::Virtual => {
let definition =
lancedb::graph::VirtualPropertyGraphDefinition::from_json(&definition_json)
.infer_error()?;
inner
.create_virtual_property_graph(name, &definition, &namespace_path)
.await
.infer_error()?
.to_json()
}
GraphKind::MaterializedVirtual => {
let definition =
lancedb::graph::VirtualPropertyGraphDefinition::from_json(&definition_json)
.infer_error()?;
inner
.create_materialized_virtual_property_graph(
name,
&definition,
&namespace_path,
)
.await
.infer_error()?
.to_json()
}
}
.infer_error()
})
}
#[pyo3(signature = (name, namespace_path=None))]
pub fn describe_property_graph(
/// The JSON description of the graph of `kind` named `name`.
#[pyo3(signature = (kind, name, namespace_path=None))]
pub fn describe_graph(
self_: PyRef<'_, Self>,
kind: String,
name: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
let kind = GraphKind::parse(&kind)?;
future_into_py(self_.py(), async move {
inner
.describe_property_graph(name, &namespace_path)
.await
.infer_error()?
.to_json()
.infer_error()
match kind {
GraphKind::Property => inner
.describe_property_graph(name, &namespace_path)
.await
.infer_error()?
.to_json(),
GraphKind::Virtual => inner
.describe_virtual_property_graph(name, &namespace_path)
.await
.infer_error()?
.to_json(),
GraphKind::MaterializedVirtual => inner
.describe_materialized_virtual_property_graph(name, &namespace_path)
.await
.infer_error()?
.to_json(),
}
.infer_error()
})
}
#[pyo3(signature = (name, namespace_path=None))]
pub fn drop_property_graph(
/// Drop the graph of `kind` named `name`, waiting for its storage to be
/// deleted.
#[pyo3(signature = (kind, name, namespace_path=None))]
pub fn drop_graph(
self_: PyRef<'_, Self>,
kind: String,
name: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
let kind = GraphKind::parse(&kind)?;
future_into_py(self_.py(), async move {
inner
.drop_property_graph(name, &namespace_path)
.await
.infer_error()
match kind {
GraphKind::Property => inner.drop_property_graph(name, &namespace_path).await,
GraphKind::Virtual => {
inner
.drop_virtual_property_graph(name, &namespace_path)
.await
}
GraphKind::MaterializedVirtual => {
inner
.drop_materialized_virtual_property_graph(name, &namespace_path)
.await
}
}
.infer_error()
})
}
#[pyo3(signature = (namespace_path=None))]
pub fn list_property_graphs(
/// The names of the graphs of `kind` in one namespace.
#[pyo3(signature = (kind, namespace_path=None))]
pub fn list_graphs(
self_: PyRef<'_, Self>,
kind: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
let kind = GraphKind::parse(&kind)?;
future_into_py(self_.py(), async move {
inner
.list_property_graphs(&namespace_path)
.await
.infer_error()
match kind {
GraphKind::Property => inner.list_property_graphs(&namespace_path).await,
GraphKind::Virtual => inner.list_virtual_property_graphs(&namespace_path).await,
GraphKind::MaterializedVirtual => {
inner
.list_materialized_virtual_property_graphs(&namespace_path)
.await
}
}
.infer_error()
})
}
@@ -1055,24 +1127,6 @@ impl Connection {
})
}
#[pyo3(signature = (name, namespace_path=None))]
pub fn refresh_property_graph(
self_: PyRef<'_, Self>,
name: String,
namespace_path: Option<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>,
@@ -1091,6 +1145,24 @@ impl Connection {
})
}
#[pyo3(signature = (name, namespace_path=None))]
pub fn refresh_materialized_virtual_property_graph(
self_: PyRef<'_, Self>,
name: String,
namespace_path: Option<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_materialized_virtual_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
+21
View File
@@ -8,6 +8,27 @@ use arrow::{datatypes::Schema, pyarrow::PyArrowType};
use lance_namespace::models::JsonArrowSchema;
use pyo3::{PyResult, exceptions::PyValueError, pyfunction};
/// A kind of graph, spelled as its routes and catalog objects are.
#[derive(Debug, Clone, Copy)]
pub enum GraphKind {
Property,
Virtual,
MaterializedVirtual,
}
impl GraphKind {
pub fn parse(kind: &str) -> PyResult<Self> {
match kind {
"property_graph" => Ok(Self::Property),
"virtual_property_graph" => Ok(Self::Virtual),
"materialized_virtual_property_graph" => Ok(Self::MaterializedVirtual),
other => Err(PyValueError::new_err(format!(
"'{other}' is not a kind of property graph"
))),
}
}
}
/// A pyarrow schema as the JSON Arrow schema of a node or edge type.
#[pyfunction]
pub fn graph_schema_to_json(schema: PyArrowType<Schema>) -> PyResult<String> {