feat: add view CRUD APIs (#4236)

A view is a named query a database stores and plans on every read. It
holds no rows, which is the whole difference from a materialized view.

## API

| Verb | Route |
| --- | --- |
| `create_view(name, query, namespace_path)` | `POST
/v1/view/{id}/create` |
| `describe_view(name, namespace_path)` | `POST /v1/view/{id}/describe`
|
| `drop_view(name, namespace_path)` | `POST /v1/view/{id}/drop` |
| `list_views(namespace_path)` | `GET /v1/namespace/{id}/view/list` |

On `Connection` and the `Database` trait, with the remote client, Python
(sync and async) and Node bindings. Local databases return
`NotSupported`: the server side is Sophon's, where a view is an object
of the database manifest.

`ViewDescription` carries the defining query, the database *and
namespace path* unqualified names in it resolve against, and the schema
the query resolved to. `create_view` returns one, so a caller has the
schema without a second call.

Both defaults travel with the view because it outlives the session that
declared it: the server re-plans the stored query on every read, so a
reader resolving an unqualified name against its own defaults would read
a different table. `default_namespace_path` crosses the wire as
`default_namespace`, a path like `namespace`, absent for the root.

There is no replace: a name already taken is an error, and changing a
view is a drop followed by a create, each authorized against what it
actually touches.

Querying a view stays SQL's job. There are no rows behind a view, so
there is no `open_view` returning a `Table`.

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 {
@@ -862,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 {