Merge remote-tracking branch 'refs/remotes/origin/main' into gatekeeper/fix-2325-1

# Conflicts:
#	python/python/lancedb/__init__.py

python/CLAUDE.md

@@ -1 +0,0 @@
-AGENTS.md

python/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.40.0-beta.3"
+version = "0.40.0-beta.6"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"

python/python/lancedb/__init__.py

@@ -47,6 +47,7 @@ from .materialized_view import (
     MaterializedView,
     MaterializedViewDefinition,
 )
+from .view import ViewDescription as ViewDescription
 from .table import AsyncTable, CompactionOptions, Table
 from .types import BaseTokenizerType
 from ._lancedb import Session
@@ -582,6 +583,7 @@ __all__ = [
     "AsyncMaterializedView",
     "MaterializedView",
     "MaterializedViewDefinition",
+    "ViewDescription",
     "connect",
     "connect_async",
     "tokenize",

python/python/lancedb/_lancedb.pyi

@@ -153,6 +153,9 @@ class Connection(object):
     async def get_function(self, name: str, version: str) -> str: ...
     async def list_functions(self) -> List[str]: ...
     async def drop_function(self, name: str, version: str) -> bool: ...
+    async def drop_function_async(
+        self, name: str, version: str
+    ) -> Tuple[bool, Job]: ...
     async def create_secret(
         self, name: str, value: str, namespace_path: Optional[List[str]] = None
     ) -> None: ...
@@ -168,6 +171,18 @@ class Connection(object):
     async def describe_secret(
         self, name: str, namespace_path: Optional[List[str]] = None
     ) -> Tuple[str, int, int]: ...
+    async def create_view(
+        self, name: str, query: str, namespace_path: Optional[List[str]] = None
+    ) -> Tuple[str, List[str], str, str, List[str], pa.Schema]: ...
+    async def describe_view(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> Tuple[str, List[str], str, str, List[str], pa.Schema]: ...
+    async def drop_view(
+        self, name: str, namespace_path: Optional[List[str]] = None
+    ) -> None: ...
+    async def list_views(
+        self, namespace_path: Optional[List[str]] = None
+    ) -> List[str]: ...
     async def list_jobs(self) -> List[JobInfo]: ...
     async def cancel_job(self, job_id: str) -> bool: ...
     async def execute_query_async(

python/python/lancedb/db.py

@@ -18,6 +18,7 @@ from typing import (
     Literal,
     Optional,
     Sequence,
+    Tuple,
     Union,
 )
 from uuid import UUID
@@ -74,6 +75,7 @@ from .util import (
     get_uri_scheme,
     validate_table_name,
 )
+from .view import ViewDescription
 
 import deprecation
 
@@ -95,6 +97,28 @@ from .namespace_utils import (
 )
 
 
+def _view_description(
+    described: Tuple[str, List[str], str, str, List[str], "pa.Schema"],
+) -> ViewDescription:
+    """Name the fields the binding returns positionally."""
+    (
+        name,
+        namespace_path,
+        query,
+        default_database,
+        default_namespace_path,
+        schema,
+    ) = described
+    return ViewDescription(
+        name=name,
+        query=query,
+        default_database=default_database,
+        schema=schema,
+        namespace_path=namespace_path,
+        default_namespace_path=default_namespace_path,
+    )
+
+
 class DBConnection(EnforceOverrides):
     """An active LanceDB connection interface."""
 
@@ -828,12 +852,26 @@ class DBConnection(EnforceOverrides):
         )
 
     def drop_function(self, name: str, *, version: str) -> bool:
-        """Remove the current Function name binding from the remote catalog.
+        """Drop a Function name and the object it was bound to.
 
-        The requested version must exist in the currently named object.
-        Object history and existing computed-column references are retained.
-        Returns True when the name was removed and False when it was absent.
-        Local connections raise NotImplementedError.
+        The requested version must exist in the currently named object. Returns
+        True when the name was removed and False when it was absent. The
+        object's content is deleted with it, which may finish after this
+        returns; use :meth:`drop_function_async` to wait for that. Local
+        connections raise NotImplementedError.
+        """
+        raise NotImplementedError(
+            "Function catalog operations are not supported for this connection type"
+        )
+
+    def drop_function_async(self, name: str, *, version: str) -> Tuple[bool, Job]:
+        """Drop a Function name and return its cleanup job.
+
+        The name is unbound before this returns; the object's content may still
+        be being deleted. Call :meth:`Job.wait` to wait for that to finish. When
+        the server deletes inline, or when nothing was bound, the returned job
+        is already finished and has no id. Local connections raise
+        NotImplementedError.
         """
         raise NotImplementedError(
             "Function catalog operations are not supported for this connection type"
@@ -903,6 +941,68 @@ class DBConnection(EnforceOverrides):
             "Secret operations are not supported for this connection type"
         )
 
+    def create_view(
+        self, name: str, query: str, *, namespace_path: Optional[List[str]] = None
+    ) -> ViewDescription:
+        """Create a view: a named query the database plans on every read.
+
+        The query is planned once, at creation, so one that cannot be planned
+        is refused now rather than at the first read. A view holds no rows, and
+        its readers see its sources as they are at read time.
+
+        There is no replace: a name already taken is an error, and changing a
+        view is a drop followed by a create. Local connections raise
+        ``NotImplementedError``.
+
+        >>> import lancedb
+        >>> db = lancedb.connect("db://my_database")  # doctest: +SKIP
+        >>> view = db.create_view(
+        ...     "adults", "SELECT name FROM people WHERE age >= 18"
+        ... )  # doctest: +SKIP
+        >>> view.schema  # doctest: +SKIP
+        name: string
+        """
+        raise NotImplementedError(
+            "View operations are not supported for this connection type"
+        )
+
+    def describe_view(
+        self, name: str, *, namespace_path: Optional[List[str]] = None
+    ) -> ViewDescription:
+        """What this database records about a view: its defining query and the
+        schema that query resolved to.
+
+        The schema is the one recorded at creation; a source altered since then
+        shows up when the view is read. Local connections raise
+        ``NotImplementedError``.
+        """
+        raise NotImplementedError(
+            "View operations are not supported for this connection type"
+        )
+
+    def drop_view(
+        self, name: str, *, namespace_path: Optional[List[str]] = None
+    ) -> None:
+        """Drop a view.
+
+        The tables it reads are untouched: a view holds no rows of its own.
+        Local connections raise ``NotImplementedError``.
+        """
+        raise NotImplementedError(
+            "View operations are not supported for this connection type"
+        )
+
+    def list_views(self, *, namespace_path: Optional[List[str]] = None) -> List[str]:
+        """The names of the views in one namespace.
+
+        Names only; a definition comes from
+        [describe_view][lancedb.db.DBConnection.describe_view]. Local
+        connections raise ``NotImplementedError``.
+        """
+        raise NotImplementedError(
+            "View operations are not supported for this connection type"
+        )
+
     def open_job(self, job_id: str) -> Job:
         """Open a server-side job by id, returning a handle with its record
         already populated.
@@ -1683,6 +1783,11 @@ class LanceDBConnection(DBConnection):
     def drop_function(self, name: str, *, version: str) -> bool:
         return LOOP.run(self._conn.drop_function(name, version=version))
 
+    @override
+    def drop_function_async(self, name: str, *, version: str) -> Tuple[bool, Job]:
+        dropped, job = LOOP.run(self._conn.drop_function_async(name, version=version))
+        return dropped, Job(job)
+
     @override
     def create_secret(
         self, name: str, value: str, *, namespace_path: Optional[List[str]] = None
@@ -1711,6 +1816,30 @@ class LanceDBConnection(DBConnection):
     ) -> SecretInfo:
         return LOOP.run(self._conn.describe_secret(name, namespace_path=namespace_path))
 
+    @override
+    def create_view(
+        self, name: str, query: str, *, namespace_path: Optional[List[str]] = None
+    ) -> ViewDescription:
+        return LOOP.run(
+            self._conn.create_view(name, query, namespace_path=namespace_path)
+        )
+
+    @override
+    def describe_view(
+        self, name: str, *, namespace_path: Optional[List[str]] = None
+    ) -> ViewDescription:
+        return LOOP.run(self._conn.describe_view(name, namespace_path=namespace_path))
+
+    @override
+    def drop_view(
+        self, name: str, *, namespace_path: Optional[List[str]] = None
+    ) -> None:
+        LOOP.run(self._conn.drop_view(name, namespace_path=namespace_path))
+
+    @override
+    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 list_jobs(self) -> List[JobInfo]:
         """List server-side jobs across the database's tables."""
@@ -2596,9 +2725,21 @@ class AsyncConnection(object):
         ]
 
     async def drop_function(self, name: str, *, version: str) -> bool:
-        """Remove the current name binding, retaining the object and its history."""
+        """Drop a Function name and the object it was bound to."""
         return await self._inner.drop_function(name, version)
 
+    async def drop_function_async(
+        self, name: str, *, version: str
+    ) -> Tuple[bool, AsyncJob]:
+        """Drop a Function name and return its cleanup job.
+
+        The name is unbound before this returns; the object's content may still
+        be being deleted. Await :meth:`AsyncJob.wait` to wait for that to
+        finish.
+        """
+        dropped, job = await self._inner.drop_function_async(name, version)
+        return dropped, AsyncJob(job)
+
     async def create_secret(
         self, name: str, value: str, *, namespace_path: Optional[List[str]] = None
     ) -> None:
@@ -2657,6 +2798,38 @@ class AsyncConnection(object):
             updated_at_millis=updated_at_millis,
         )
 
+    async def create_view(
+        self, name: str, query: str, *, namespace_path: Optional[List[str]] = None
+    ) -> ViewDescription:
+        """Create a view: a named query the database plans on every read.
+
+        See
+        [DBConnection.create_view][lancedb.DBConnection.create_view].
+        """
+        return _view_description(
+            await self._inner.create_view(name, query, list(namespace_path or []))
+        )
+
+    async def describe_view(
+        self, name: str, *, namespace_path: Optional[List[str]] = None
+    ) -> ViewDescription:
+        """What this database records about a view: query and schema."""
+        return _view_description(
+            await self._inner.describe_view(name, list(namespace_path or []))
+        )
+
+    async def drop_view(
+        self, name: str, *, namespace_path: Optional[List[str]] = None
+    ) -> None:
+        """Drop a view. The tables it reads are untouched."""
+        await self._inner.drop_view(name, list(namespace_path or []))
+
+    async def list_views(
+        self, *, namespace_path: Optional[List[str]] = None
+    ) -> List[str]:
+        """The names of the views in one namespace."""
+        return await self._inner.list_views(list(namespace_path or []))
+
     async def list_jobs(self) -> List[JobInfo]:
         """List server-side jobs across the database's tables."""
         return await self._inner.list_jobs()

python/python/lancedb/materialized_view.py

@@ -29,9 +29,10 @@ SelectArg = Union[
 ]
 
 
-DEFINITION_FORMAT = 1
-"""The stored layout this version reads: ``{"format": 1, "query": "<SQL>"}``.
-A ``kind`` key beside it is for readers older than the format number."""
+DEFINITION_FORMAT = 2
+"""The newest stored layout this version reads: ``{"format": N, "query": "<SQL>"}``,
+format 2 being a query with ``GROUP BY``. A ``kind`` key beside it is for
+readers older than the format number."""
 
 
 @dataclass
@@ -40,7 +41,7 @@ class MaterializedViewDefinition:
 
         SELECT columns
         FROM [ns.]table [, function(args) AS alias | , UNNEST(column) AS alias]
-        [WHERE predicate] [LIMIT n]
+        [WHERE predicate] [GROUP BY expr, ...] [LIMIT n]
 
     A Function in ``FROM`` position yields one row per element it returns.
     """

python/python/lancedb/remote/db.py

@@ -16,6 +16,7 @@ from typing import (
     List,
     Optional,
     Sequence,
+    Tuple,
     Union,
 )
 from urllib.parse import urlparse
@@ -40,6 +41,7 @@ from ..sql import Query as SqlQuery
 from ..sql import QueryDescription
 from ..materialized_view import MaterializedView, SelectArg
 from ..secrets import EnvVarSecret, SecretInfo
+from ..view import ViewDescription
 
 if TYPE_CHECKING:
     from .._lancedb import JobInfo
@@ -876,6 +878,11 @@ class RemoteDBConnection(DBConnection):
     def drop_function(self, name: str, *, version: str) -> bool:
         return LOOP.run(self._conn.drop_function(name, version=version))
 
+    @override
+    def drop_function_async(self, name: str, *, version: str) -> Tuple[bool, Job]:
+        dropped, job = LOOP.run(self._conn.drop_function_async(name, version=version))
+        return dropped, Job(job)
+
     @override
     def create_secret(
         self, name: str, value: str, *, namespace_path: Optional[List[str]] = None
@@ -904,6 +911,30 @@ class RemoteDBConnection(DBConnection):
     ) -> None:
         LOOP.run(self._conn.drop_secret(name, namespace_path=namespace_path))
 
+    @override
+    def create_view(
+        self, name: str, query: str, *, namespace_path: Optional[List[str]] = None
+    ) -> ViewDescription:
+        return LOOP.run(
+            self._conn.create_view(name, query, namespace_path=namespace_path)
+        )
+
+    @override
+    def describe_view(
+        self, name: str, *, namespace_path: Optional[List[str]] = None
+    ) -> ViewDescription:
+        return LOOP.run(self._conn.describe_view(name, namespace_path=namespace_path))
+
+    @override
+    def drop_view(
+        self, name: str, *, namespace_path: Optional[List[str]] = None
+    ) -> None:
+        LOOP.run(self._conn.drop_view(name, namespace_path=namespace_path))
+
+    @override
+    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 list_jobs(self) -> List["JobInfo"]:
         """List server-side jobs across the database's tables."""

python/python/lancedb/remote/table.py

@@ -984,6 +984,9 @@ class RemoteTable(Table):
         self,
         transforms: Dict[str, str | FunctionApplication]
         | FunctionApplication
+        | pa.Field
+        | List[pa.Field]
+        | pa.Schema
         | None = None,
         *,
         computed: Dict[str, str] | None = None,

python/python/lancedb/table.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 
 import asyncio
 import inspect
+import json
 import deprecation
 import warnings
 from abc import ABC, abstractmethod
@@ -352,12 +353,14 @@ def _into_pyarrow_reader(
 
         # convert to list of dict if data is a bunch of LanceModels
         if isinstance(data[0], LanceModel):
-            schema = data[0].__class__.to_arrow_schema()
+            model_schema = data[0].__class__.to_arrow_schema()
             data = [model_to_dict(d) for d in data]
-            return pa.Table.from_pylist(data, schema=schema).to_reader()
+            data = _serialize_json_values(data, schema or model_schema)
+            return pa.Table.from_pylist(data, schema=model_schema).to_reader()
         elif isinstance(data[0], pa.RecordBatch):
             return pa.Table.from_batches(data).to_reader()
         else:
+            data = _serialize_json_values(data, schema)
             return pa.Table.from_pylist(data).to_reader()
     elif _check_for_pandas(data) and isinstance(data, pd.DataFrame):
         table = pa.Table.from_pandas(data, preserve_index=False)
@@ -705,6 +708,108 @@ def _field_extension_name(field: pa.Field) -> Optional[str]:
     return extension_name
 
 
+def _is_json_field(field: pa.Field) -> bool:
+    return _field_extension_name(field) in ("arrow.json", "lance.json")
+
+
+@dataclass(frozen=True)
+class _JsonSerializationPlan:
+    arrow_field: pa.Field
+    children: Optional[Dict[str, "_JsonSerializationPlan"]] = None
+    item: Optional["_JsonSerializationPlan"] = None
+
+
+def _json_serialization_plan(field: pa.Field) -> Optional[_JsonSerializationPlan]:
+    if _is_json_field(field):
+        return _JsonSerializationPlan(field)
+
+    if pa.types.is_struct(field.type):
+        children: Dict[str, _JsonSerializationPlan] = {}
+        for child_field in field.type:
+            child_plan = _json_serialization_plan(child_field)
+            if child_plan is not None:
+                children[child_field.name] = child_plan
+        if children:
+            return _JsonSerializationPlan(field, children=children)
+
+    if _is_list_like(field.type):
+        item_plan = _json_serialization_plan(field.type.value_field)
+        if item_plan is not None:
+            return _JsonSerializationPlan(field, item=item_plan)
+
+    return None
+
+
+def _json_serialization_plans(
+    schema: pa.Schema,
+) -> Dict[str, _JsonSerializationPlan]:
+    plans: Dict[str, _JsonSerializationPlan] = {}
+    for field in schema:
+        plan = _json_serialization_plan(field)
+        if plan is not None:
+            plans[field.name] = plan
+    return plans
+
+
+def _serialize_json_value(value: Any, plan: _JsonSerializationPlan) -> Any:
+    if value is None or isinstance(value, str):
+        return value
+    if _is_json_field(plan.arrow_field):
+        if isinstance(value, (dict, list)):
+            return json.dumps(value)
+        return value
+
+    if plan.children is not None and isinstance(value, dict):
+        serialized = None
+        for child_name, child_plan in plan.children.items():
+            if child_name not in value:
+                continue
+            child_value = _serialize_json_value(value[child_name], child_plan)
+            if child_value is not value[child_name]:
+                if serialized is None:
+                    serialized = dict(value)
+                serialized[child_name] = child_value
+        return serialized if serialized is not None else value
+
+    if plan.item is not None and isinstance(value, list):
+        serialized = None
+        for index, item in enumerate(value):
+            serialized_item = _serialize_json_value(item, plan.item)
+            if serialized_item is not item:
+                if serialized is None:
+                    serialized = list(value)
+                serialized[index] = serialized_item
+        return serialized if serialized is not None else value
+
+    return value
+
+
+def _serialize_json_values(data: Any, target_schema: Optional[pa.Schema]) -> Any:
+    if target_schema is None or not isinstance(data, list):
+        return data
+
+    plans = _json_serialization_plans(target_schema)
+    if not plans:
+        return data
+
+    serialized_rows = []
+    for row in data:
+        if not isinstance(row, dict):
+            serialized_rows.append(row)
+            continue
+        serialized_row = None
+        for field_name, plan in plans.items():
+            if field_name not in row:
+                continue
+            value = _serialize_json_value(row[field_name], plan)
+            if value is not row[field_name]:
+                if serialized_row is None:
+                    serialized_row = dict(row)
+                serialized_row[field_name] = value
+        serialized_rows.append(serialized_row if serialized_row is not None else row)
+    return serialized_rows
+
+
 def _align_field_types(
     fields: List[pa.Field],
     target_fields: List[pa.Field],
@@ -5819,6 +5924,7 @@ class AsyncTable:
 
         """
         schema = await self.schema()
+        data = _serialize_json_values(data, schema)
         if on_bad_vectors is None:
             on_bad_vectors = "error"
         if fill_value is None:

python/python/lancedb/view.py

@@ -0,0 +1,40 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""Views: a named query a database stores and plans on every read.
+
+A view holds no rows. What it stores is the statement that defines it and the
+schema that statement resolved to, so a reader sees the sources as they are
+now. See ``DBConnection.create_view``.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, List
+
+if TYPE_CHECKING:
+    import pyarrow as pa
+
+
+@dataclass(frozen=True)
+class ViewDescription:
+    """What a database records about one view."""
+
+    name: str
+    """The view's name within its namespace."""
+    query: str
+    """The defining query, as the database stores it."""
+    default_database: str
+    """The database that unqualified table names in ``query`` resolve against."""
+    schema: "pa.Schema"
+    """The schema the defining query resolved to when the view was created."""
+    namespace_path: List[str] = field(default_factory=list)
+    """The namespace holding the view; empty is the root namespace."""
+    default_namespace_path: List[str] = field(default_factory=list)
+    """The namespace path those unqualified names resolve against.
+
+    Recorded with the view because it outlives the session that declared it: a
+    reader resolving the query against its own default namespace could read a
+    different table than the view was defined over.
+    """

python/python/tests/test_materialized_views.py

@@ -522,7 +522,7 @@ def test_stored_queries_and_legacy_layouts_are_read():
 
     # A newer writer's layout is reported, never guessed at.
     for newer in (
-        {"format": 2, "query": query},
+        {"format": 3, "query": query},
         {"kind": "select_v3", "source_table": "people"},
     ):
         with pytest.raises(NotImplementedError, match="cannot refresh"):

python/python/tests/test_remote_db.py

@@ -2747,3 +2747,57 @@ def test_remote_job_handle_reports_its_own_detail():
             "limit": 500,
             "filter": "state = 'claim_complete'",
         }
+
+
+def test_view_crud_addresses_its_own_routes():
+    # The view verbs are their own routes, and the schema comes back in the
+    # namespace spec's JSON encoding, decoded into a pyarrow schema.
+    paths = []
+
+    def handler(request):
+        paths.append((request.command, request.path))
+        if request.path.endswith("/view/list"):
+            body = {"views": ["adults"]}
+        elif request.path.endswith("/drop"):
+            body = {}
+        else:
+            body = {
+                "name": "adults",
+                "namespace": ["analytics"],
+                "query": "SELECT name FROM people",
+                "default_database": "dev",
+                "default_namespace": ["analytics"],
+                "schema": {
+                    "fields": [
+                        {"name": "name", "nullable": True, "type": {"type": "utf8"}}
+                    ]
+                },
+            }
+        request.send_response(200)
+        request.send_header("Content-Type", "application/json")
+        request.end_headers()
+        request.wfile.write(json.dumps(body).encode())
+
+    with mock_lancedb_connection(handler) as db:
+        view = db.create_view(
+            "adults", "SELECT name FROM people", namespace_path=["analytics"]
+        )
+        assert view.name == "adults"
+        assert view.namespace_path == ["analytics"]
+        assert view.query == "SELECT name FROM people"
+        assert view.default_database == "dev"
+        assert view.default_namespace_path == ["analytics"]
+        assert view.schema == pa.schema([pa.field("name", pa.utf8(), nullable=True)])
+
+        described = db.describe_view("adults", namespace_path=["analytics"])
+        assert described.schema == view.schema
+
+        assert db.list_views(namespace_path=["analytics"]) == ["adults"]
+        db.drop_view("adults", namespace_path=["analytics"])
+
+    assert paths == [
+        ("POST", "/v1/view/analytics$adults/create"),
+        ("POST", "/v1/view/analytics$adults/describe"),
+        ("GET", "/v1/namespace/analytics/view/list"),
+        ("POST", "/v1/view/analytics$adults/drop"),
+    ]

python/python/tests/test_table.py

@@ -4,6 +4,7 @@
 
 import ctypes
 import gc
+import json
 import os
 import sys
 import threading
@@ -17,6 +18,7 @@ from typing import Any, List
 from unittest.mock import patch
 
 import lancedb
+from lancedb import table as table_module
 from lancedb.dependencies import _PANDAS_AVAILABLE
 from lancedb.index import BTree, FTS, HnswFlat, HnswPq, HnswSq, IvfPq
 import numpy as np
@@ -3027,6 +3029,236 @@ async def test_add_sanitization_encodes_json(mem_db_async: AsyncConnection):
     assert rows == [{"id": "c", "j": '{"k":3}'}]
 
 
+@pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
+def test_add_python_objects_to_json_column(mem_db: DBConnection):
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int64()),
+            pa.field("payload", pa.json_()),
+            pa.field("metadata", pa.struct([("label", pa.string())])),
+            pa.field("tags", pa.list_(pa.string())),
+        ]
+    )
+    table = mem_db.create_table("json_python_objects_add", schema=schema)
+    table.add(
+        [
+            {
+                "id": 1,
+                "payload": {"foo": "bar", "count": 2},
+                "metadata": {"label": "dict"},
+                "tags": ["a", "b"],
+            },
+            {
+                "id": 2,
+                "payload": {
+                    "name": "alice",
+                    "tags": ["x", "y"],
+                    "nested": {"enabled": True},
+                },
+                "metadata": {"label": "nested"},
+                "tags": ["c"],
+            },
+            {
+                "id": 3,
+                "payload": ["x", "y", "z"],
+                "metadata": {"label": "list"},
+                "tags": ["d", "e"],
+            },
+            {
+                "id": 4,
+                "payload": '{"foo": "bar"}',
+                "metadata": {"label": "string"},
+                "tags": ["f"],
+            },
+            {
+                "id": 5,
+                "payload": None,
+                "metadata": {"label": "null"},
+                "tags": ["g"],
+            },
+        ]
+    )
+
+    rows = {row["id"]: row for row in table.to_arrow().to_pylist()}
+    assert json.loads(rows[1]["payload"]) == {"foo": "bar", "count": 2}
+    assert json.loads(rows[2]["payload"]) == {
+        "name": "alice",
+        "tags": ["x", "y"],
+        "nested": {"enabled": True},
+    }
+    assert json.loads(rows[3]["payload"]) == ["x", "y", "z"]
+    assert json.loads(rows[4]["payload"]) == {"foo": "bar"}
+    assert rows[5]["payload"] is None
+    assert rows[1]["metadata"] == {"label": "dict"}
+    assert rows[1]["tags"] == ["a", "b"]
+
+    matched = table.search().where("json_extract(payload, '$.count') = '2'").to_list()
+    assert [row["id"] for row in matched] == [1]
+
+
+@pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
+def test_merge_insert_python_objects_to_json_column(mem_db: DBConnection):
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int64()),
+            pa.field("payload", pa.json_()),
+            pa.field("metadata", pa.struct([("label", pa.string())])),
+            pa.field("tags", pa.list_(pa.string())),
+        ]
+    )
+    table = mem_db.create_table("json_python_objects_merge", schema=schema)
+    table.add(
+        [
+            {
+                "id": 1,
+                "payload": {"old": True},
+                "metadata": {"label": "old"},
+                "tags": ["old"],
+            }
+        ]
+    )
+
+    table.merge_insert(
+        "id"
+    ).when_matched_update_all().when_not_matched_insert_all().execute(
+        [
+            {
+                "id": 1,
+                "payload": {"foo": "bar", "count": 2},
+                "metadata": {"label": "dict"},
+                "tags": ["a", "b"],
+            },
+            {
+                "id": 2,
+                "payload": {
+                    "name": "alice",
+                    "tags": ["x", "y"],
+                    "nested": {"enabled": True},
+                },
+                "metadata": {"label": "nested"},
+                "tags": ["c"],
+            },
+            {
+                "id": 3,
+                "payload": ["x", "y", "z"],
+                "metadata": {"label": "list"},
+                "tags": ["d", "e"],
+            },
+            {
+                "id": 4,
+                "payload": '{"foo": "bar"}',
+                "metadata": {"label": "string"},
+                "tags": ["f"],
+            },
+            {
+                "id": 5,
+                "payload": None,
+                "metadata": {"label": "null"},
+                "tags": ["g"],
+            },
+        ]
+    )
+
+    rows = {row["id"]: row for row in table.to_arrow().to_pylist()}
+    assert json.loads(rows[1]["payload"]) == {"foo": "bar", "count": 2}
+    assert json.loads(rows[2]["payload"]) == {
+        "name": "alice",
+        "tags": ["x", "y"],
+        "nested": {"enabled": True},
+    }
+    assert json.loads(rows[3]["payload"]) == ["x", "y", "z"]
+    assert json.loads(rows[4]["payload"]) == {"foo": "bar"}
+    assert rows[5]["payload"] is None
+    assert rows[1]["metadata"] == {"label": "dict"}
+    assert rows[1]["tags"] == ["a", "b"]
+
+    matched = table.search().where("json_extract(payload, '$.count') = '2'").to_list()
+    assert [row["id"] for row in matched] == [1]
+
+
+@pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
+def test_add_python_objects_to_nested_json_fields(mem_db: DBConnection):
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int64()),
+            pa.field(
+                "info",
+                pa.struct([pa.field("payload", pa.json_())]),
+            ),
+            pa.field("documents", pa.list_(pa.field("item", pa.json_()))),
+        ]
+    )
+    table = mem_db.create_table("nested_json_python_objects_add", schema=schema)
+
+    table.add(
+        [
+            {
+                "id": 1,
+                "info": {"payload": {"kind": "struct", "value": 1}},
+                "documents": [{"kind": "list", "value": 2}],
+            }
+        ]
+    )
+
+    row = table.to_arrow().to_pylist()[0]
+    assert json.loads(row["info"]["payload"]) == {"kind": "struct", "value": 1}
+    assert json.loads(row["documents"][0]) == {"kind": "list", "value": 2}
+    assert [
+        row["id"]
+        for row in table.search()
+        .where("json_extract(info.payload, '$.value') = '1'")
+        .to_list()
+    ] == [1]
+    assert [
+        row["id"]
+        for row in table.search()
+        .where("json_extract(documents[1], '$.value') = '2'")
+        .to_list()
+    ] == [1]
+
+
+@pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
+def test_json_serialization_plan_skips_non_json_branches():
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int64()),
+            pa.field("vector", pa.list_(pa.float32(), 3)),
+            pa.field("payload", pa.json_()),
+            pa.field(
+                "metadata",
+                pa.struct(
+                    [
+                        pa.field("label", pa.string()),
+                        pa.field("payload", pa.json_()),
+                    ]
+                ),
+            ),
+            pa.field(
+                "documents",
+                pa.list_(
+                    pa.struct(
+                        [
+                            pa.field("title", pa.string()),
+                            pa.field("payload", pa.json_()),
+                        ]
+                    )
+                ),
+            ),
+        ]
+    )
+
+    plans = table_module._json_serialization_plans(schema)
+
+    assert set(plans) == {"payload", "metadata", "documents"}
+    metadata_children = plans["metadata"].children
+    assert metadata_children is not None
+    assert set(metadata_children) == {"payload"}
+    documents_item = plans["documents"].item
+    assert documents_item is not None
+    assert documents_item.children is not None
+    assert set(documents_item.children) == {"payload"}
+
+
 @pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
 @pytest.mark.asyncio
 async def test_add_all_null_json_batch(mem_db_async: AsyncConnection):

python/src/connection.rs

@@ -13,7 +13,11 @@ use crate::{
     runtime::future_into_py,
     table::Table,
 };
-use arrow::{datatypes::Schema, ffi_stream::ArrowArrayStreamReader, pyarrow::FromPyArrow};
+use arrow::{
+    datatypes::Schema,
+    ffi_stream::ArrowArrayStreamReader,
+    pyarrow::{FromPyArrow, ToPyArrow},
+};
 use lancedb::{
     connection::Connection as LanceConnection,
     connection::NamespaceClientPushdownOperation,
@@ -100,6 +104,28 @@ fn parse_default_namespace_path(path: Option<Bound<'_, PyAny>>) -> PyResult<Vec<
     }
 }
 
+/// A view description on its way to Python: name, namespace, query, default
+/// database, and the schema as pyarrow renders it.
+type PyViewDescription = (String, Vec<String>, String, String, Vec<String>, Py<PyAny>);
+
+/// A view description as a plain tuple, with the schema converted to the
+/// pyarrow schema the caller would get from any other lancedb API. The Python
+/// layer names the fields; this keeps the binding free of a class that would
+/// have to be kept in step with the Rust struct.
+fn view_description_to_py(view: lancedb::view::ViewDescription) -> PyResult<PyViewDescription> {
+    Python::attach(|py| {
+        let schema = view.schema.to_pyarrow(py)?.unbind();
+        Ok((
+            view.name,
+            view.namespace_path,
+            view.query,
+            view.default_database,
+            view.default_namespace_path,
+            schema,
+        ))
+    })
+}
+
 #[pymethods]
 impl Connection {
     fn __repr__(&self) -> String {
@@ -768,6 +794,21 @@ impl Connection {
         })
     }
 
+    pub fn drop_function_async(
+        self_: PyRef<'_, Self>,
+        name: String,
+        version: String,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            inner
+                .drop_function_async(name, version)
+                .await
+                .infer_error()
+                .map(|(dropped, job)| (dropped, crate::job::Job::new(job)))
+        })
+    }
+
     #[pyo3(signature = (name, value, namespace_path=None))]
     pub fn create_secret(
         self_: PyRef<'_, Self>,
@@ -847,6 +888,66 @@ impl Connection {
         })
     }
 
+    #[pyo3(signature = (name, query, namespace_path=None))]
+    pub fn create_view(
+        self_: PyRef<'_, Self>,
+        name: String,
+        query: 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 {
+            let view = inner
+                .create_view(name, query, &namespace_path)
+                .await
+                .infer_error()?;
+            view_description_to_py(view)
+        })
+    }
+
+    #[pyo3(signature = (name, namespace_path=None))]
+    pub fn describe_view(
+        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 {
+            let view = inner
+                .describe_view(name, &namespace_path)
+                .await
+                .infer_error()?;
+            view_description_to_py(view)
+        })
+    }
+
+    #[pyo3(signature = (name, namespace_path=None))]
+    pub fn drop_view(
+        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.drop_view(name, &namespace_path).await.infer_error()
+        })
+    }
+
+    #[pyo3(signature = (namespace_path=None))]
+    pub fn list_views(
+        self_: PyRef<'_, Self>,
+        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.list_views(&namespace_path).await.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 {