python: create_index returns Job / AsyncJob (ENT-1966)

Table.create_index, create_scalar_index, create_fts_index, and the materialized-view delegates now return a Job (AsyncJob on AsyncTable). When the server defers the build (pending vector index), the returned job tracks it through the platform jobs API; synchronous builds (scalar, FTS, native tables, GPU-accelerated local paths) return a pre-completed job whose status()/wait() report finished immediately and whose cancel() is a no-op. AsyncTable now carries its owning connection so the async handles can reach the jobs API. wait_timeout keeps working; docs steer new code to job.wait(). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
rust: create_index returns the server-minted job id
2026-08-26 07:58:31 +00:00 · 2026-07-18 21:02:29 -07:00 · 2026-07-18 20:52:29 -07:00 · 2026-07-18 17:56:54 -07:00 · 2026-07-18 17:14:44 -07:00 · 2026-07-18 17:14:44 -07:00
25 changed files with 4536 additions and 137 deletions
@@ -3421,8 +3421,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"

 [[package]]
 name = "fsst"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow-array",
 "rand 0.9.5",
@@ -4777,8 +4777,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"

 [[package]]
 name = "lance"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -4852,8 +4852,8 @@ dependencies = [

 [[package]]
 name = "lance-arrow"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4875,7 +4875,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-scalar"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4889,7 +4889,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-stats"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -4898,8 +4898,8 @@ dependencies = [

 [[package]]
 name = "lance-bitpacking"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrayref",
 "crunchy",
@@ -4909,8 +4909,8 @@ dependencies = [

 [[package]]
 name = "lance-core"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -4948,8 +4948,8 @@ dependencies = [

 [[package]]
 name = "lance-datafusion"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -4979,8 +4979,8 @@ dependencies = [

 [[package]]
 name = "lance-datagen"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -4997,8 +4997,8 @@ dependencies = [

 [[package]]
 name = "lance-derive"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5007,8 +5007,8 @@ dependencies = [

 [[package]]
 name = "lance-encoding"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5043,8 +5043,8 @@ dependencies = [

 [[package]]
 name = "lance-file"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow-arith",
 "arrow-array",
@@ -5074,8 +5074,8 @@ dependencies = [

 [[package]]
 name = "lance-index"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arc-swap",
 "arrow",
@@ -5141,8 +5141,8 @@ dependencies = [

 [[package]]
 name = "lance-index-core"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5164,8 +5164,8 @@ dependencies = [

 [[package]]
 name = "lance-io"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow",
 "arrow-arith",
@@ -5208,8 +5208,8 @@ dependencies = [

 [[package]]
 name = "lance-linalg"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5225,8 +5225,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow",
 "async-trait",
@@ -5238,8 +5238,8 @@ dependencies = [

 [[package]]
 name = "lance-namespace-impls"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow",
 "arrow-ipc",
@@ -5293,8 +5293,8 @@ dependencies = [

 [[package]]
 name = "lance-select"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow-array",
 "arrow-buffer",
@@ -5309,8 +5309,8 @@ dependencies = [

 [[package]]
 name = "lance-table"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -5349,8 +5349,8 @@ dependencies = [

 [[package]]
 name = "lance-testing"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "arrow-array",
 "arrow-schema",
@@ -5363,8 +5363,8 @@ dependencies = [

 [[package]]
 name = "lance-tokenizer"
-version = "9.0.0-rc.1"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-rc.1#cec0b7dffe2d85c7e66dbe9d1f3891c297903a1d"
+version = "9.1.0-beta.2"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.2#9139486ed371dd6caeb9c708b39fba7f5649a9b0"
 dependencies = [
 "icu_segmenter",
 "jieba-rs",
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"

 [workspace.dependencies]
-lance = { "version" = "=9.0.0-rc.1", default-features = false, "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=9.0.0-rc.1", "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=9.0.0-rc.1", "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=9.0.0-rc.1", "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=9.0.0-rc.1", default-features = false, "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=9.0.0-rc.1", "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=9.0.0-rc.1", "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=9.0.0-rc.1", "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=9.0.0-rc.1", default-features = false, "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=9.0.0-rc.1", "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=9.0.0-rc.1", "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=9.0.0-rc.1", "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=9.0.0-rc.1", "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=9.0.0-rc.1", "tag" = "v9.0.0-rc.1", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=9.1.0-beta.2", default-features = false, "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=9.1.0-beta.2", "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=9.1.0-beta.2", "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=9.1.0-beta.2", "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=9.1.0-beta.2", default-features = false, "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=9.1.0-beta.2", "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=9.1.0-beta.2", "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=9.1.0-beta.2", "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=9.1.0-beta.2", default-features = false, "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=9.1.0-beta.2", "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=9.1.0-beta.2", "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=9.1.0-beta.2", "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=9.1.0-beta.2", "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=9.1.0-beta.2", "tag" = "v9.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
 ahash = "0.8"
 # Note that this one does not include pyarrow
 arrow = { version = "58.0.0", optional = false }
@@ -28,7 +28,7 @@
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>9.0.0-rc.1</lance-core.version>
+        <lance-core.version>9.1.0-beta.2</lance-core.version>
        <spotless.skip>false</spotless.skip>
        <spotless.version>2.30.0</spotless.version>
        <spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
@@ -165,7 +165,7 @@ impl Table {
        if let Some(train) = train {
            builder = builder.train(train);
        }
-        builder.execute().await.default_error()
+        builder.execute().await.default_error().map(|_| ())
    }

    #[napi(catch_unwind)]
@@ -19,6 +19,17 @@ from .db import AsyncConnection, DBConnection, LanceDBConnection
 from .remote import ClientConfig
 from .remote.db import RemoteDBConnection
 from .expr import Expr, col, lit, func
+from .udf import (
+    udf,
+    table_udf,
+    Udf,
+    Job,
+    JobFailedError,
+    MaterializedView,
+    AsyncJob,
+    AsyncMaterializedView,
+)
+from .lineage import Lineage, Node, Edge, FunctionRef
 from .schema import blob, vector, BlobType
 from .table import AsyncTable, Table
 from .types import BaseTokenizerType
@@ -491,6 +502,18 @@ async def connect_async(


 __all__ = [
+    "udf",
+    "table_udf",
+    "Udf",
+    "Job",
+    "JobFailedError",
+    "MaterializedView",
+    "AsyncJob",
+    "AsyncMaterializedView",
+    "Lineage",
+    "Node",
+    "Edge",
+    "FunctionRef",
    "connect",
    "connect_async",
    "tokenize",
@@ -65,6 +65,7 @@ if TYPE_CHECKING:
    from .common import DATA, URI
    from .embeddings import EmbeddingFunctionConfig
    from ._lancedb import Session
+    from .udf import MaterializedView, AsyncMaterializedView

 from .namespace_utils import (
    _normalize_create_namespace_mode,
@@ -562,6 +563,277 @@ class DBConnection(EnforceOverrides):
        """
        raise NotImplementedError("serialize is not supported for this connection type")

+    # -- Derived compute: functions, materialized views, jobs -------------
+    # Server-backed features (LanceDB Enterprise / Cloud); local
+    # connections raise NotImplementedError for now.
+
+    def create_function(
+        self,
+        name,
+        language: str = "python",
+        return_type: Optional[str] = None,
+        body: Optional[str] = None,
+        options: Optional[Dict[str, str]] = None,
+        *,
+        replace: bool = False,
+    ):
+        """Register a UDF (CREATE FUNCTION).
+
+        Pass a ``@udf`` / ``@table_udf``-decorated function (preferred):
+
+            db.create_function(embed)
+
+        or the explicit fields:
+
+        Parameters
+        ----------
+        name: str or Udf
+            A decorated UDF object, or the function name.
+        language: str
+            Implementation language (currently "python").
+        return_type: str
+            SQL return type, e.g. "FLOAT", "FLOAT[1536]",
+            "STRUCT(a FLOAT, b VARCHAR)", "TABLE(chunk VARCHAR, idx INT)".
+        body: str
+            Function body: source text, or base64 cloudpickle bytes when
+            options["body_format"] == "cloudpickle".
+        options: dict, optional
+            input_columns, pip, num_gpus, batch_size, timeout,
+            error_policy, docker_image, body_format, ...
+        replace: bool
+            Drop an existing function of the same name first.
+        """
+        from .udf import Udf
+
+        if isinstance(name, Udf):
+            req = name.create_request()
+            name, language, return_type, body, options = (
+                req["name"],
+                req["language"],
+                req["return_type"],
+                req["body"],
+                req["options"],
+            )
+        if replace:
+            try:
+                self.drop_function(name)
+            except Exception:
+                pass
+        LOOP.run(self._conn.create_function(name, language, return_type, body, options))
+
+    def list_functions(self):
+        """List registered functions (SHOW FUNCTIONS)."""
+        return LOOP.run(self._conn.list_functions())
+
+    def drop_function(self, name: str):
+        """Drop a registered function (DROP FUNCTION)."""
+        LOOP.run(self._conn.drop_function(name))
+
+    def create_materialized_view(
+        self,
+        name: str,
+        source=None,
+        select=None,
+        *,
+        query: Optional[str] = None,
+        where: Optional[str] = None,
+        auto_refresh: bool = False,
+        with_no_data: bool = False,
+        replace: bool = False,
+        partition_by: Optional[str] = None,
+    ) -> "MaterializedView":
+        """Create a materialized view (CREATE MATERIALIZED VIEW); returns a
+        `MaterializedView` handle (``.wait()`` blocks until it is populated).
+
+        Two ways to specify the view body:
+
+        - ergonomic: pass ``source`` (a table name or table) and ``select``
+          items -- column names, expression strings ("embed(body)"),
+          (alias, expression) tuples, or ``@udf`` / ``@table_udf`` objects.
+          The SELECT is assembled and parsed server-side (one parser, shared
+          with SQL).
+        - raw: pass ``query=`` with a full SELECT, e.g.
+          "SELECT id, embed(body) AS vec FROM articles WHERE id > 1".
+
+        `partition_by` partitions the view's (single) table function on a source
+        column. If that column has an IVF vector index the server partitions by
+        its index clusters (image-dedup style); otherwise it groups by distinct
+        value. (Geneva's `partition_by` and `partition_by_indexed_column` unify
+        here -- the engine picks the strategy from the column.)
+        """
+        from .udf import build_view_query, MaterializedView
+
+        if query is None:
+            if source is None or select is None:
+                raise ValueError(
+                    "create_materialized_view needs either query= or both "
+                    "source and select"
+                )
+            query = build_view_query(source, select)
+        if where:
+            query += f" WHERE {where}"
+        if replace:
+            self._drop_view_if_exists(name)
+        job_id = LOOP.run(
+            self._conn.create_materialized_view(
+                name,
+                query=query,
+                auto_refresh=auto_refresh,
+                with_no_data=with_no_data,
+                partition_by=partition_by,
+            )
+        )
+        return MaterializedView(self, name, job_id=job_id)
+
+    def _drop_view_if_exists(self, name: str) -> None:
+        # `replace=True` is "drop if present"; only a not-found error is
+        # benign here. Anything else (perms, server fault) must surface rather
+        # than be masked by a later create failure.
+        try:
+            self.drop_materialized_view(name)
+        except Exception as e:
+            msg = str(e).lower()
+            if "not found" not in msg and "does not exist" not in msg:
+                raise
+
+    def job(self, job_id: str):
+        """A `Job` for reconnecting to an inflight job by id -- e.g. an
+        id you stored, or one returned from the SQL / REST surface. Submit
+        methods (`refresh_column`, `MaterializedView.refresh`) already return a
+        handle directly, so you do not need this to wait on a fresh submission."""
+        from .udf import Job
+
+        return Job(self, job_id)
+
+    def lineage(
+        self,
+        table: str,
+        column: Optional[str] = None,
+        *,
+        direction: Optional[str] = None,
+        depth: Optional[int] = None,
+    ):
+        """Derived-compute lineage of a table/view, or one of its columns:
+        upstream sources, downstream dependents, and the function version +
+        location that produced each derived column (with a drift flag). Returns
+        a `Lineage`. `direction` is "upstream" | "downstream" | "both" (server
+        default both); `depth` limits column-hops (transitive when omitted)."""
+        # `self._conn` is the AsyncConnection; drive its async `lineage`
+        # (which parses the JSON) on the loop, mirroring create_materialized_view.
+        return LOOP.run(
+            self._conn.lineage(table, column, direction=direction, depth=depth)
+        )
+
+    def _refresh_materialized_view(
+        self,
+        name: str,
+        *,
+        full: bool = False,
+        src_version: Optional[int] = None,
+        num_workers: Optional[int] = None,
+        max_workers: Optional[int] = None,
+    ) -> str:
+        """Internal: submit a materialized-view refresh, return the job id.
+        The public surface is ``MaterializedView.refresh()`` (which returns a
+        `Job`); this stays private so refresh is only reached through the
+        handle.
+
+        ``full=True`` forces a full rebuild (recompute and replace every row)
+        instead of the default incremental refresh.
+        """
+        return LOOP.run(
+            self._conn._refresh_materialized_view(
+                name,
+                full=full,
+                src_version=src_version,
+                num_workers=num_workers,
+                max_workers=max_workers,
+            )
+        )
+
+    def explain_refresh_materialized_view(
+        self,
+        name: str,
+        *,
+        full: bool = False,
+        src_version: Optional[int] = None,
+    ):
+        """Plan a refresh without running it (EXPLAIN REFRESH). Returns a
+        plan with .has_work / .source_version / .last_refreshed_version /
+        .full_refresh / .rebuild / .units_total. `full=True` plans a full
+        rebuild (incremental planning needs stable row IDs on the source)."""
+        return LOOP.run(
+            self._conn.explain_refresh_materialized_view(
+                name, full=full, src_version=src_version
+            )
+        )
+
+    def alter_materialized_view(self, name: str, *, auto_refresh: bool):
+        """Update a materialized view's options (ALTER MATERIALIZED VIEW)."""
+        LOOP.run(self._conn.alter_materialized_view(name, auto_refresh=auto_refresh))
+
+    def drop_materialized_view(self, name: str):
+        """Drop a materialized view definition (DROP MATERIALIZED VIEW)."""
+        LOOP.run(self._conn.drop_materialized_view(name))
+
+    def list_materialized_views(self):
+        """List registered materialized view definitions."""
+        return LOOP.run(self._conn.list_materialized_views())
+
+    def list_jobs(self):
+        """List inflight server-side jobs across the database's tables."""
+        return LOOP.run(self._conn.list_jobs())
+
+    def get_job(self, job_id: str, table: "str | None" = None):
+        """Look up one server-side job by id (the wait()/status poll path).
+
+        Passing ``table`` (the job's table) lets the server answer with an O(1)
+        single-node read instead of scanning the database's active jobs.
+        Returns the job's status, or None if it's unknown or no longer active.
+        """
+        return LOOP.run(self._conn.get_job(job_id, table))
+
+    def cancel_job(self, job_id: str) -> bool:
+        """Cancel an inflight server-side job by id (CANCEL JOB).
+
+        Returns True if a matching inflight job was found and flagged for
+        cancellation, False if none was inflight (already finished or
+        unknown id) -- cancellation is best-effort.
+        """
+        return LOOP.run(self._conn.cancel_job(job_id))
+
+    def describe_platform_job(self, platform_job_id: str):
+        """Describe a platform job (POST /v1/jobs/describe): registry-backed
+        lifecycle state plus the owner-written status payload. None when the
+        registry has no such job."""
+        return LOOP.run(self._conn.describe_platform_job(platform_job_id))
+
+    def resolve_platform_job_id(
+        self, manifest_job_id: str, table: "str | None" = None
+    ):
+        """Resolve a submission (manifest) job id to its platform job id.
+        None until the job has registered (dispatch is async)."""
+        return LOOP.run(self._conn.resolve_platform_job_id(manifest_job_id, table))
+
+    def cancel_platform_job(self, platform_job_id: str) -> None:
+        """Cancel a platform job (POST /v1/jobs/cancel). Idempotent on
+        already-terminal jobs."""
+        return LOOP.run(self._conn.cancel_platform_job(platform_job_id))
+
+    def job_history(self, job_id: "str | None" = None):
+        """Durable history of completed server-side jobs (SHOW JOB HISTORY).
+
+        Pass ``job_id`` to narrow to a single job. Unlike :meth:`list_jobs`
+        (live, inflight) these are the terminal records.
+        """
+        return LOOP.run(self._conn.job_history(job_id))
+
+    def errors(self, job_id: "str | None" = None, table: "str | None" = None):
+        """Per-row UDF errors recorded by ``error_policy=skip`` (SHOW ERRORS),
+        optionally filtered by ``job_id`` and/or ``table``.
+        """
+        return LOOP.run(self._conn.errors(job_id, table))
+

 class LanceDBConnection(DBConnection):
    """
@@ -1655,7 +1927,7 @@ class AsyncConnection(object):
                namespace_client=namespace_client,
            )

-        return AsyncTable(new_table)
+        return AsyncTable(new_table, conn=self)

    async def open_table(
        self,
@@ -1728,7 +2000,7 @@ class AsyncConnection(object):
            namespace_client=namespace_client,
            managed_versioning=managed_versioning,
        )
-        tbl = AsyncTable(table)
+        tbl = AsyncTable(table, conn=self)
        # "main" is the default branch, so treat it as no branch: remote rejects
        # every branch checkout (even "main"), and the version still applies.
        if branch is not None and branch != "main":
@@ -1785,7 +2057,218 @@ class AsyncConnection(object):
            source_tag=source_tag,
            is_shallow=is_shallow,
        )
-        return AsyncTable(table)
+        return AsyncTable(table, conn=self)
+
+    # -- Derived compute: functions, materialized views, jobs -------------
+    # Server-backed features (LanceDB Enterprise / Cloud); local
+    # connections raise NotImplementedError for now.
+
+    async def create_function(
+        self,
+        name,
+        language: str = "python",
+        return_type: Optional[str] = None,
+        body: Optional[str] = None,
+        options: Optional[Dict[str, str]] = None,
+        *,
+        replace: bool = False,
+    ):
+        """Register a UDF (CREATE FUNCTION). Accepts a ``@udf``/``@table_udf``
+        object (preferred) or the explicit (name, language, return_type, body,
+        options)."""
+        from .udf import Udf
+
+        if isinstance(name, Udf):
+            req = name.create_request()
+            name, language, return_type, body, options = (
+                req["name"],
+                req["language"],
+                req["return_type"],
+                req["body"],
+                req["options"],
+            )
+        if replace:
+            try:
+                await self.drop_function(name)
+            except Exception:
+                pass
+        await self._inner.create_function(name, language, return_type, body, options)
+
+    async def list_functions(self):
+        """List registered functions (SHOW FUNCTIONS)."""
+        return await self._inner.list_functions()
+
+    async def drop_function(self, name: str):
+        """Drop a registered function (DROP FUNCTION)."""
+        await self._inner.drop_function(name)
+
+    async def create_materialized_view(
+        self,
+        name: str,
+        source=None,
+        select=None,
+        *,
+        query: Optional[str] = None,
+        where: Optional[str] = None,
+        auto_refresh: bool = False,
+        with_no_data: bool = False,
+        replace: bool = False,
+        partition_by: Optional[str] = None,
+    ) -> "AsyncMaterializedView":
+        """Create a materialized view; returns an `AsyncMaterializedView`
+        handle (``.wait()`` blocks until populated). Pass either ``query=`` (a
+        full SELECT) or ``source`` + ``select`` items; `partition_by`
+        partitions the view's table function on a source column (index-cluster
+        if the column is IVF-indexed, else distinct-value). See the sync
+        method for the select grammar."""
+        from .udf import build_view_query, AsyncMaterializedView
+
+        if query is None:
+            if source is None or select is None:
+                raise ValueError(
+                    "create_materialized_view needs either query= or both "
+                    "source and select"
+                )
+            query = build_view_query(source, select)
+        if where:
+            query += f" WHERE {where}"
+        if replace:
+            try:
+                await self.drop_materialized_view(name)
+            except Exception as e:
+                msg = str(e).lower()
+                if "not found" not in msg and "does not exist" not in msg:
+                    raise
+        job_id = await self._inner.create_materialized_view(
+            name,
+            query,
+            auto_refresh=auto_refresh,
+            with_no_data=with_no_data,
+            partition_by=partition_by,
+        )
+        return AsyncMaterializedView(self, name, job_id=job_id)
+
+    def job(self, job_id: str):
+        """An `AsyncJob` for reconnecting to an inflight job by id (a
+        stored id, or one from the SQL / REST surface). Submit methods already
+        return a handle, so this is only needed to re-attach to an existing
+        job."""
+        from .udf import AsyncJob
+
+        return AsyncJob(self, job_id)
+
+    async def lineage(
+        self,
+        table: str,
+        column: Optional[str] = None,
+        *,
+        direction: Optional[str] = None,
+        depth: Optional[int] = None,
+    ):
+        """Derived-compute lineage of a table/view (or column). See the sync
+        `Connection.lineage`. Returns a `Lineage`."""
+        from .lineage import Lineage
+
+        raw = await self._inner.table_lineage(table, column, direction, depth)
+        return Lineage.from_json(raw)
+
+    async def _refresh_materialized_view(
+        self,
+        name: str,
+        *,
+        full: bool = False,
+        src_version: Optional[int] = None,
+        num_workers: Optional[int] = None,
+        max_workers: Optional[int] = None,
+    ) -> str:
+        """Internal: submit a refresh, return the job id. The public surface is
+        ``AsyncMaterializedView.refresh()`` (returns an `AsyncJob`).
+
+        ``full=True`` forces a full rebuild (recompute and replace every row)
+        instead of the default incremental refresh.
+        """
+        return await self._inner.refresh_materialized_view(
+            name,
+            full=full,
+            src_version=src_version,
+            num_workers=num_workers,
+            max_workers=max_workers,
+        )
+
+    async def explain_refresh_materialized_view(
+        self,
+        name: str,
+        *,
+        full: bool = False,
+        src_version: Optional[int] = None,
+    ):
+        """Plan a refresh without running it (EXPLAIN REFRESH)."""
+        return await self._inner.explain_refresh_materialized_view(
+            name, full=full, src_version=src_version
+        )
+
+    async def alter_materialized_view(self, name: str, *, auto_refresh: bool):
+        """Update a materialized view's options."""
+        await self._inner.alter_materialized_view(name, auto_refresh)
+
+    async def drop_materialized_view(self, name: str):
+        """Drop a materialized view definition."""
+        await self._inner.drop_materialized_view(name)
+
+    async def list_materialized_views(self):
+        """List registered materialized view definitions."""
+        return await self._inner.list_materialized_views()
+
+    async def list_jobs(self):
+        """List inflight server-side jobs across the database's tables."""
+        return await self._inner.list_jobs()
+
+    async def get_job(self, job_id: str, table: "str | None" = None):
+        """Look up one server-side job by id (the wait()/status poll path).
+        ``table`` (the job's table) enables an O(1) server-side lookup.
+        Returns the job's status, or None if unknown / no longer active."""
+        return await self._inner.get_job(job_id, table)
+
+    async def cancel_job(self, job_id: str) -> bool:
+        """Cancel an inflight server-side job by id (CANCEL JOB).
+
+        Returns True if a matching inflight job was found and flagged for
+        cancellation, False otherwise (best-effort).
+        """
+        return await self._inner.cancel_job(job_id)
+
+    async def describe_platform_job(self, platform_job_id: str):
+        """Describe a platform job: registry-backed lifecycle state plus the
+        owner-written status payload. None when the registry has no such
+        job."""
+        return await self._inner.describe_platform_job(platform_job_id)
+
+    async def resolve_platform_job_id(
+        self, manifest_job_id: str, table: "str | None" = None
+    ):
+        """Resolve a submission (manifest) job id to its platform job id.
+        None until the job has registered (dispatch is async)."""
+        return await self._inner.resolve_platform_job_id(manifest_job_id, table)
+
+    async def cancel_platform_job(self, platform_job_id: str) -> None:
+        """Cancel a platform job. Idempotent on already-terminal jobs."""
+        return await self._inner.cancel_platform_job(platform_job_id)
+
+    async def job_history(self, job_id: "str | None" = None):
+        """Durable history of completed server-side jobs (SHOW JOB HISTORY).
+
+        Reads each table's durable job-history store. Pass ``job_id`` to narrow
+        to a single job. Unlike :meth:`list_jobs` (live, inflight) these are the
+        terminal records, with created/updated/completed timestamps.
+        """
+        return await self._inner.job_history(job_id)
+
+    async def errors(self, job_id: "str | None" = None, table: "str | None" = None):
+        """Per-row UDF errors recorded by ``error_policy=skip`` (SHOW ERRORS).
+
+        Optionally filtered by ``job_id`` and/or ``table``.
+        """
+        return await self._inner.errors(job_id, table)

    async def rename_table(
        self,
@@ -0,0 +1,177 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+"""Client-side model of derived-compute lineage.
+
+`Connection.lineage()` / `Table.lineage()` / `MaterializedView.lineage()` return
+a `Lineage`: the graph of what a column or materialized view derives from
+(upstream), what derives from it (downstream), and -- for each derived column --
+the function that produced it, the version it was produced with, and whether
+that is stale relative to the function the registry now holds.
+
+The server returns this as JSON (the wire contract); these classes deserialize
+it. Nothing here talks to the server.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from typing import List, Optional, Union
+
+
+@dataclass
+class FunctionRef:
+    """The function that produced a derived column, with version + location."""
+
+    name: str
+    #: Version that produced the data (stamped at compute time), if known.
+    as_computed_version: Optional[str] = None
+    #: Version the registry currently holds for this function name.
+    current_version: Optional[str] = None
+    #: True when the column was produced by an older function than the registry
+    #: now holds -- i.e. silently stale; re-refresh to catch up.
+    stale_vs_current: bool = False
+    language: Optional[str] = None
+    docker_image: Optional[str] = None
+    env_digest: Optional[str] = None
+    code_uri: Optional[str] = None
+
+    @classmethod
+    def _from(cls, d: dict) -> "FunctionRef":
+        return cls(
+            name=d["name"],
+            as_computed_version=d.get("as_computed_version"),
+            current_version=d.get("current_version"),
+            stale_vs_current=d.get("stale_vs_current", False),
+            language=d.get("language"),
+            docker_image=d.get("docker_image"),
+            env_digest=d.get("env_digest"),
+            code_uri=d.get("code_uri"),
+        )
+
+
+@dataclass
+class Node:
+    """A lineage node: a table, view, column, or function."""
+
+    kind: str  # "table" | "view" | "column" | "function"
+    id: str  # "table", "table.column", or "fn:name@version"
+    table: Optional[str] = None
+    function: Optional[FunctionRef] = None
+
+    @classmethod
+    def _from(cls, d: dict) -> "Node":
+        fn = d.get("function")
+        return cls(
+            kind=d["kind"],
+            id=d["id"],
+            table=d.get("table"),
+            function=FunctionRef._from(fn) if fn else None,
+        )
+
+
+@dataclass
+class Edge:
+    """`downstream` depends on `upstream`, produced by `via` (a function name,
+    or None for a passthrough)."""
+
+    downstream: str
+    upstream: str
+    via: Optional[str] = None
+
+    @classmethod
+    def _from(cls, d: dict) -> "Edge":
+        return cls(downstream=d["downstream"], upstream=d["upstream"], via=d.get("via"))
+
+
+@dataclass
+class Lineage:
+    """A derived-compute lineage graph (nodes + labeled edges)."""
+
+    target: str
+    nodes: List[Node] = field(default_factory=list)
+    edges: List[Edge] = field(default_factory=list)
+
+    @classmethod
+    def from_json(cls, raw: Union[str, bytes, dict]) -> "Lineage":
+        d = json.loads(raw) if isinstance(raw, (str, bytes)) else raw
+        return cls(
+            target=d.get("target", ""),
+            nodes=[Node._from(n) for n in d.get("nodes", [])],
+            edges=[Edge._from(e) for e in d.get("edges", [])],
+        )
+
+    def functions(self) -> List[FunctionRef]:
+        """The function nodes in the graph."""
+        return [n.function for n in self.nodes if n.function is not None]
+
+    def stale(self) -> List[FunctionRef]:
+        """Functions whose as-computed version is behind the current registry
+        version -- the columns they produced are silently out of date."""
+        return [f for f in self.functions() if f.stale_vs_current]
+
+    def to_dict(self) -> dict:
+        def prune(d: dict) -> dict:
+            return {k: v for k, v in d.items() if v is not None}
+
+        return {
+            "target": self.target,
+            "nodes": [
+                prune(
+                    {
+                        "kind": n.kind,
+                        "id": n.id,
+                        "table": n.table,
+                        "function": prune(vars(n.function)) if n.function else None,
+                    }
+                )
+                for n in self.nodes
+            ],
+            "edges": [prune(vars(e)) for e in self.edges],
+        }
+
+    def to_graphviz(self) -> str:
+        """Graphviz DOT for the lineage DAG: columns/tables as nodes, function
+        names on edges, drift edges dashed + red."""
+        stale_names = {f.name for f in self.stale()}
+        out = [
+            "digraph lineage {",
+            "  rankdir=LR;",
+            '  node [fontname="monospace"];',
+        ]
+        for n in self.nodes:
+            if n.kind == "function":
+                continue
+            shape = "ellipse" if n.kind in ("table", "view") else "box"
+            out.append(f'  "{n.id}" [shape={shape}];')
+        for e in self.edges:
+            attrs = ""
+            if e.via:
+                if e.via in stale_names:
+                    attrs = f' [label="{e.via}" color=red style=dashed]'
+                else:
+                    attrs = f' [label="{e.via}"]'
+            out.append(f'  "{e.upstream}" -> "{e.downstream}"{attrs};')
+        out.append("}")
+        return "\n".join(out)
+
+    def _repr_html_(self) -> str:
+        warn = ""
+        drift = self.stale()
+        if drift:
+            names = ", ".join(sorted({f.name for f in drift}))
+            warn = (
+                f'<p style="color:#b00000"><b>stale vs current:</b> {names} '
+                "(re-refresh to catch up)</p>"
+            )
+        rows = "".join(
+            f"<tr><td><code>{e.downstream}</code></td>"
+            f"<td>&larr; {e.via or ''}</td>"
+            f"<td><code>{e.upstream}</code></td></tr>"
+            for e in self.edges
+        )
+        return (
+            f"<b>lineage: <code>{self.target}</code></b>{warn}"
+            "<table><tr><th>derived</th><th>via</th><th>from</th></tr>"
+            f"{rows}</table>"
+        )
@@ -3875,18 +3875,18 @@ class AsyncHybridQuery(AsyncStandardQuery, AsyncVectorQueryBase):
        >>> asyncio.run(doctest_example()) # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
        RRFReranker(K=60)
            ProjectionExec: expr=[vector@0 as vector, text@3 as text, _distance@2 as _distance]
-              Take: columns="vector, _rowid, _distance, (text)"
-                CoalesceBatchesExec: target_batch_size=1024
-                  GlobalLimitExec: skip=0, fetch=10
-                    FilterExec: _distance@2 IS NOT NULL
-                      SortExec: TopK(fetch=10), expr=[_distance@2 ASC NULLS LAST, _rowid@1 ASC NULLS LAST], preserve_partitioning=[false]
-                        KNNVectorDistance: metric=l2
-                          LanceRead: uri=..., projection=[vector], ...
+            Take: columns="vector, _rowid, _distance, (text)"
+              CoalesceBatchesExec: target_batch_size=1024
+                GlobalLimitExec: skip=0, fetch=10
+                  FilterExec: _distance@2 IS NOT NULL
+                    SortExec: TopK(fetch=10), expr=[_distance@2 ASC NULLS LAST, _rowid@1 ASC NULLS LAST], preserve_partitioning=[false]
+                      KNNVectorDistance: metric=l2
+                        LanceRead: uri=..., projection=[vector], ...
            ProjectionExec: expr=[vector@2 as vector, text@3 as text, _score@1 as _score]
-              Take: columns="_rowid, _score, (vector), (text)"
-                CoalesceBatchesExec: target_batch_size=1024
-                  GlobalLimitExec: skip=0, fetch=10
-                    MatchQuery: column=text, query=hello
+            Take: columns="_rowid, _score, (vector), (text)"
+              CoalesceBatchesExec: target_batch_size=1024
+                GlobalLimitExec: skip=0, fetch=10
+                  MatchQuery: column=text, query=[hello]

        Parameters
        ----------
@@ -13,10 +13,14 @@ from typing import (
    Iterable,
    List,
    Optional,
+    TYPE_CHECKING,
    Union,
    Literal,
    overload,
 )
+
+if TYPE_CHECKING:
+    from ..udf import Job
 import warnings

 from lancedb import __version__
@@ -914,8 +918,142 @@ class RemoteTable(Table):
    def count_rows(self, filter: Optional[str] = None) -> int:
        return LOOP.run(self._table.count_rows(filter))

-    def add_columns(self, transforms: Dict[str, str]) -> AddColumnsResult:
-        return LOOP.run(self._table.add_columns(transforms))
+    def add_columns(
+        self,
+        transforms: Optional[Dict[str, str]] = None,
+        *,
+        computed: Optional[Dict[str, tuple]] = None,
+    ) -> Optional[AddColumnsResult]:
+        result = None
+        if transforms is not None:
+            result = LOOP.run(self._table.add_columns(transforms))
+        if computed:
+            LOOP.run(self._table.add_columns(computed=computed))
+        return result
+
+    def refresh_column(
+        self,
+        columns,
+        *,
+        where: Optional[str] = None,
+        num_workers: Optional[int] = None,
+        max_workers: Optional[int] = None,
+        batch_size: Optional[int] = None,
+        priority: Optional[str] = None,
+    ) -> "Job":
+        """Trigger recompute of computed columns (REFRESH COLUMN).
+
+        The expression is resolved server-side from each column's stored
+        binding; columns bound to the same struct-returning function
+        refresh together. Returns a `Job` to wait on, poll, or cancel
+        (``tbl.refresh_column("c").wait()``). Server-backed feature
+        (LanceDB Enterprise / Cloud).
+
+        num_workers / max_workers / batch_size / priority are per-refresh
+        scheduling knobs (how to run THIS refresh) and override any default
+        the function carries. `priority` is a Kueue tier
+        (training | interactive | backfill).
+        """
+        from ..udf import Job
+
+        if isinstance(columns, str):
+            columns = [columns]
+        job_id = LOOP.run(
+            self._table.refresh_column(
+                list(columns),
+                where=where,
+                num_workers=num_workers,
+                max_workers=max_workers,
+                batch_size=batch_size,
+                priority=priority,
+            )
+        )
+        return Job(self._job_conn(), job_id)
+
+    def lineage(self, column=None, *, direction=None, depth=None):
+        """Derived-compute lineage of this table, or one of its columns:
+        upstream sources, downstream dependents, and the function version +
+        location that produced each derived column (with a drift flag). Returns
+        a `Lineage`. See `Connection.lineage`."""
+        return self._job_conn().lineage(
+            self._name, column, direction=direction, depth=depth
+        )
+
+    def _job_conn(self):
+        """A client connection for polling jobs this table spawns. Built lazily
+        from the table's serialized connection state and cached (not pickled --
+        a forked/unpickled table rebuilds it on next use)."""
+        from lancedb import deserialize_conn
+
+        conn = getattr(self, "_job_conn_cache", None)
+        if conn is None:
+            conn = deserialize_conn(self._serialized_connection_state())
+            self._job_conn_cache = conn
+        return conn
+
+    def load_columns(
+        self,
+        source: Union[str, Iterable[str]],
+        pk: str,
+        columns: Union[Iterable[str], Dict[str, str]],
+        *,
+        source_format: str = "parquet",
+        source_pk: Optional[str] = None,
+        on_missing: str = "carry",
+        source_storage_options: Optional[Dict[str, str]] = None,
+        num_workers: Optional[int] = None,
+        max_workers: Optional[int] = None,
+        batch_size: Optional[int] = None,
+        commit_granularity: Optional[int] = None,
+        priority: Optional[str] = None,
+    ) -> str:
+        """Fill existing columns from an external source by primary-key join.
+
+        The distributed-job equivalent of Geneva's ``Table.load_columns()``:
+        imports precomputed values (e.g. embeddings) from Parquet/Lance/IPC into
+        this table, matching on a primary key. Returns the load job id.
+        Server-backed feature (LanceDB Enterprise / Cloud).
+
+        Parameters
+        ----------
+        source: str | list[str]
+            One source URI or a list of URIs.
+        pk: str
+            Destination primary-key column. Also the source key unless
+            ``source_pk`` is given.
+        columns: list[str] | dict[str, str]
+            Value columns to load. A list loads same-named columns; a dict maps
+            ``{target: source}``.
+        source_format: str
+            ``"parquet"`` (default), ``"lance"``, or ``"ipc"``.
+        source_pk: str, optional
+            Source primary-key column when it differs from ``pk``.
+        on_missing: str
+            Behavior for destination rows with no source match:
+            ``"carry"`` (default, keep existing), ``"null"``, or ``"error"``.
+        """
+        if isinstance(source, str):
+            source = [source]
+        if isinstance(columns, dict):
+            mappings = [(target, src) for target, src in columns.items()]
+        else:
+            mappings = [(c, None) for c in columns]
+        return LOOP.run(
+            self._table.load_columns(
+                list(source),
+                source_format,
+                pk,
+                mappings,
+                source_key=source_pk,
+                source_storage_options=source_storage_options,
+                on_missing=on_missing,
+                num_workers=num_workers,
+                max_workers=max_workers,
+                batch_size=batch_size,
+                commit_granularity=commit_granularity,
+                priority=priority,
+            )
+        )

    def alter_columns(
        self, *alterations: Iterable[Dict[str, str]]
@@ -162,6 +162,7 @@ def _maybe_add_fts_error_note(

 if TYPE_CHECKING:
    from .db import LanceDBConnection
+    from .udf import Job
    from ._lancedb import (
        Table as LanceDBTable,
        OptimizeStats,
@@ -702,6 +703,24 @@ def _normalize_progress(progress):
    return progress, False


+def _computed_groups(computed):
+    """Group computed columns by expression, preserving declaration order
+    (struct-returning functions need their columns adjacent so schema order
+    matches field order). Accepts the ergonomic forms -- `fn("col")` values
+    and tuple keys for struct fan-out -- via `_normalize_computed`."""
+    from .udf import _normalize_computed
+
+    groups = []
+    for name, (sql_type, expression) in _normalize_computed(computed).items():
+        for expr, cols in groups:
+            if expr == expression:
+                cols.append((name, sql_type))
+                break
+        else:
+            groups.append((expression, [(name, sql_type)]))
+    return groups
+
+
 class Table(ABC):
    """
    A Table is a collection of Records in a LanceDB Database.
@@ -807,6 +826,59 @@ class Table(ABC):
        """The number of rows in this Table"""
        return self.count_rows(None)

+    def add_computed_column(
+        self,
+        columns,
+        fn,
+        args: Optional[List[str]] = None,
+        types=None,
+    ) -> None:
+        """Declare computed column(s) bound to a UDF -- no compute happens
+        here (the agent fills them lazily, or refresh_column() triggers a run).
+
+        .. deprecated::
+            A computed column is an expression over a registered function, so
+            bind it as one: ``add_columns(computed={"vec": embed("data")})``.
+            ``embed("data")`` applies the function to the `data` column and
+            infers the type from the function's return signature -- the
+            function never couples to a particular column. Prefer that form.
+        """
+        import warnings
+
+        warnings.warn(
+            "add_computed_column is deprecated; use add_columns(computed="
+            '{"vec": embed("data")}).',
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        from .udf import Udf, struct_field_types
+
+        multi = isinstance(columns, (tuple, list))
+        if isinstance(fn, Udf):
+            expr = fn.expression(*(args or []))
+            if types is None:
+                if multi:
+                    if not fn.returns.upper().startswith("STRUCT"):
+                        raise ValueError(
+                            "several columns need a STRUCT-returning function"
+                        )
+                    types = struct_field_types(fn.returns)
+                else:
+                    types = fn.returns
+        else:
+            if types is None:
+                raise ValueError("pass types= when fn is a name string")
+            expr = f"{fn}({', '.join(args or [])})"
+        if multi:
+            if len(types) != len(columns):
+                raise ValueError(
+                    f"{len(columns)} columns but {len(types)} output types"
+                )
+            computed = {c: (t, expr) for c, t in zip(columns, types)}
+        else:
+            computed = {columns: (types, expr)}
+        self.add_columns(computed=computed)
+
    @property
    @abstractmethod
    def embedding_functions(self) -> Dict[str, EmbeddingFunctionConfig]:
@@ -883,7 +955,7 @@ class Table(ABC):
        wait_timeout: Optional[timedelta] = ...,
        name: Optional[str] = ...,
        train: bool = ...,
-    ) -> None: ...
+    ) -> "Job": ...

    # Legacy API overload (deprecated)
    @overload
@@ -907,7 +979,7 @@ class Table(ABC):
        name: Optional[str] = ...,
        train: bool = ...,
        target_partition_size: Optional[int] = ...,
-    ) -> None: ...
+    ) -> "Job": ...

    def create_index(
        self,
@@ -958,6 +1030,14 @@ class Table(ABC):
        train : bool, default True
            Whether to train the index with existing data.

+        Returns
+        -------
+        Job
+            A handle on the index build. When the server defers the build to a
+            background job, ``job.wait()`` blocks until it completes; when the
+            build finished within this call, the job is already ``finished``.
+            Prefer ``job.wait()`` over the deprecated ``wait_timeout``.
+
        Examples
        --------
        New API (recommended):
@@ -2125,8 +2205,8 @@ class LanceTable(Table):
    def from_inner(cls, tbl: LanceDBTable):
        from .db import LanceDBConnection

-        async_tbl = AsyncTable(tbl)
        conn = LanceDBConnection.from_inner(tbl.database())
+        async_tbl = AsyncTable(tbl, conn=conn._conn)
        return cls(
            conn,
            async_tbl.name,
@@ -2528,7 +2608,7 @@ class LanceTable(Table):
        wait_timeout: Optional[timedelta] = ...,
        name: Optional[str] = ...,
        train: bool = ...,
-    ) -> None: ...
+    ) -> "Job": ...

    # Legacy API overload (deprecated)
    @overload
@@ -2554,7 +2634,7 @@ class LanceTable(Table):
        name: Optional[str] = ...,
        train: bool = ...,
        target_partition_size: Optional[int] = ...,
-    ) -> None: ...
+    ) -> "Job": ...

    def create_index(
        self,
@@ -2613,6 +2693,14 @@ class LanceTable(Table):
        train : bool, default True
            Whether to train the index with existing data.

+        Returns
+        -------
+        Job
+            A handle on the index build. When the server defers the build to a
+            background job, ``job.wait()`` blocks until it completes; when the
+            build finished within this call, the job is already ``finished``.
+            Prefer ``job.wait()`` over the deprecated ``wait_timeout``.
+
        Examples
        --------
        New API (recommended):
@@ -2688,7 +2776,7 @@ class LanceTable(Table):
                    target_partition_size=target_partition_size,
                )
                self.checkout_latest()
-                return
+                return self._sync_job(None)
        else:
            # New API: metric is the column name
            column = metric
@@ -2725,19 +2813,30 @@ class LanceTable(Table):
                        ),
                    )
                    self.checkout_latest()
-                    return
+                    return self._sync_job(None)

-        return LOOP.run(
-            self._table.create_index(
-                column,
-                replace=replace,
-                config=config,
-                wait_timeout=wait_timeout,
-                name=name,
-                train=train,
+        return self._sync_job(
+            LOOP.run(
+                self._table.create_index(
+                    column,
+                    replace=replace,
+                    config=config,
+                    wait_timeout=wait_timeout,
+                    name=name,
+                    train=train,
+                )
            )
        )

+    def _sync_job(self, ajob) -> "Job":
+        """Convert an AsyncJob (or None for work done in-process) into a sync
+        Job bound to this table's connection."""
+        from .udf import Job
+
+        if ajob is not None and ajob.id:
+            return Job(self._conn, ajob.id, table=self.name)
+        return Job._completed(self._conn, table=self.name)
+
    def _is_legacy_create_index_call(
        self,
        first_arg: str,
@@ -2988,8 +3087,12 @@ class LanceTable(Table):
            config = LabelList()
        else:
            raise ValueError(f"Unknown index type {index_type}")
-        return LOOP.run(
-            self._table.create_index(column, replace=replace, config=config, name=name)
+        return self._sync_job(
+            LOOP.run(
+                self._table.create_index(
+                    column, replace=replace, config=config, name=name
+                )
+            )
        )

    @deprecation.deprecated(
@@ -3072,7 +3175,7 @@ class LanceTable(Table):
        )

        try:
-            LOOP.run(
+            ajob = LOOP.run(
                self._table.create_index(
                    field_names,
                    replace=replace,
@@ -3087,6 +3190,7 @@ class LanceTable(Table):
                language=config.language,
            )
            raise e
+        return self._sync_job(ajob)

    @staticmethod
    def infer_tokenizer_configs(tokenizer_name: str) -> dict:
@@ -3815,9 +3919,68 @@ class LanceTable(Table):
        return LOOP.run(self._table.index_stats(index_name))

    def add_columns(
-        self, transforms: Dict[str, str] | pa.field | List[pa.field] | pa.Schema
-    ) -> AddColumnsResult:
-        return LOOP.run(self._table.add_columns(transforms))
+        self,
+        transforms: Dict[str, str]
+        | pa.field
+        | List[pa.field]
+        | pa.Schema
+        | None = None,
+        *,
+        computed: Optional[Dict] = None,
+    ) -> Optional[AddColumnsResult]:
+        result = None
+        if transforms is not None:
+            result = LOOP.run(self._table.add_columns(transforms))
+        if computed:
+            # computed binds an expression over a registered function to a
+            # column: {col: fn("input_col")} -- fn("input_col") yields the
+            # expression and carries the inferred type; a tuple key fans a
+            # STRUCT return out to several columns. Declares the binding only;
+            # the server fills the values (server-backed). The legacy
+            # {col: (sql_type, expression)} tuple form is still accepted.
+            result_unused = LOOP.run(self._table.add_columns(computed=computed))
+            del result_unused
+        return result
+
+    def refresh_column(
+        self,
+        columns,
+        *,
+        where: Optional[str] = None,
+        num_workers: Optional[int] = None,
+        max_workers: Optional[int] = None,
+        batch_size: Optional[int] = None,
+        priority: Optional[str] = None,
+    ) -> "Job":
+        """Trigger recompute of computed columns (REFRESH COLUMN).
+
+        The expression is resolved server-side from each column's stored
+        binding; columns bound to the same struct-returning function
+        refresh together. Returns a `Job` to wait on, poll, or cancel
+        (``tbl.refresh_column("col").wait()``) -- mirrors
+        `MaterializedView.refresh()`. Server-backed feature (LanceDB
+        Enterprise / Cloud).
+
+        num_workers / max_workers / batch_size / priority are per-refresh
+        scheduling knobs (how to run THIS refresh) and override any default
+        the function carries. `priority` is a Kueue tier
+        (training | interactive | backfill).
+        """
+        from .udf import Job
+
+        if isinstance(columns, str):
+            columns = [columns]
+        job_id = LOOP.run(
+            self._table.refresh_column(
+                list(columns),
+                where=where,
+                num_workers=num_workers,
+                max_workers=max_workers,
+                batch_size=batch_size,
+                priority=priority,
+            )
+        )
+        return Job(self._conn, job_id, table=self.name)

    def alter_columns(
        self, *alterations: Iterable[Dict[str, str]]
@@ -4433,6 +4596,7 @@ class AsyncTable:
        self,
        table: LanceDBTable,
        *,
+        conn: Optional[Any] = None,
        namespace_path: Optional[List[str]] = None,
        namespace_client: Optional[Any] = None,
        pushdown_operations: Optional[set] = None,
@@ -4446,6 +4610,9 @@ class AsyncTable:
        [AsyncConnection.open_table][lancedb.AsyncConnection.open_table] to obtain
        Table objects."""
        self._inner = table
+        #: The owning AsyncConnection, when known -- lets index/refresh calls
+        #: hand back AsyncJob handles that can reach the platform jobs API.
+        self._conn = conn
        self._namespace_path = namespace_path or []
        self._namespace_client = namespace_client
        self._pushdown_operations = pushdown_operations or set()
@@ -4753,6 +4920,14 @@ class AsyncTable:
        train: bool, default True
            Whether to train the index with existing data. Vector indices always train
            with existing data.
+
+        Returns
+        -------
+        AsyncJob
+            A handle on the index build. When the server defers the build to a
+            background job, ``await job.wait()`` blocks until it completes;
+            when the build finished within this call, the job is already
+            ``finished``. Prefer ``await job.wait()`` over ``wait_timeout``.
        """
        if config is not None:
            if not isinstance(
@@ -4778,7 +4953,7 @@ class AsyncTable:
                    + str(type(config))
                )
        try:
-            await self._inner.create_index(
+            job_id = await self._inner.create_index(
                column,
                index=config,
                replace=replace,
@@ -4795,6 +4970,12 @@ class AsyncTable:
                )
            raise e

+        from .udf import AsyncJob
+
+        if job_id:
+            return AsyncJob(self._conn, job_id, table=self.name)
+        return AsyncJob._completed(self._conn, table=self.name)
+
    async def drop_index(self, name: str) -> None:
        """
        Drop an index from the table.
@@ -5570,9 +5751,44 @@ class AsyncTable:

        return await self._inner.update(updates_sql, where)

+    async def refresh_column(
+        self,
+        columns,
+        *,
+        where: Optional[str] = None,
+        num_workers: Optional[int] = None,
+        max_workers: Optional[int] = None,
+        batch_size: Optional[int] = None,
+        priority: Optional[str] = None,
+    ) -> str:
+        """Trigger recompute of computed columns (REFRESH COLUMN).
+        Returns the refresh job id. Server-backed feature.
+
+        num_workers / max_workers / batch_size / priority are per-refresh
+        scheduling knobs (how to run THIS refresh); they override any default
+        the function carries. `priority` is a Kueue tier
+        (training | interactive | backfill)."""
+        if isinstance(columns, str):
+            columns = [columns]
+        return await self._inner.refresh_column(
+            list(columns),
+            where_clause=where,
+            num_workers=num_workers,
+            max_workers=max_workers,
+            batch_size=batch_size,
+            priority=priority,
+        )
+
    async def add_columns(
-        self, transforms: dict[str, str] | pa.field | List[pa.field] | pa.Schema
-    ) -> AddColumnsResult:
+        self,
+        transforms: dict[str, str]
+        | pa.field
+        | List[pa.field]
+        | pa.Schema
+        | None = None,
+        *,
+        computed: Optional[Dict] = None,
+    ) -> Optional[AddColumnsResult]:
        """
        Add new columns with defined values.

@@ -5591,6 +5807,7 @@ class AsyncTable:
            version: the new version number of the table after adding columns.

        """
+        result = None
        if isinstance(transforms, pa.Field):
            transforms = [transforms]
        if isinstance(transforms, list) and all(
@@ -5598,9 +5815,69 @@ class AsyncTable:
        ):
            transforms = pa.schema(transforms)
        if isinstance(transforms, pa.Schema):
-            return await self._inner.add_columns_with_schema(transforms)
+            result = await self._inner.add_columns_with_schema(transforms)
+        elif transforms is not None:
+            result = await self._inner.add_columns(list(transforms.items()))
+        if computed:
+            # computed binds an expression over a registered function to a
+            # column: {col: fn("input_col")} -- fn("input_col") yields the
+            # expression and carries the inferred type; a tuple key fans a
+            # STRUCT return out to several columns. Declares the binding only;
+            # the server fills the values (server-backed). The legacy
+            # {col: (sql_type, expression)} tuple form is still accepted.
+            for expression, cols in _computed_groups(computed):
+                await self._inner.add_computed_columns(cols, expression)
+        return result
+
+    async def add_computed_column(
+        self,
+        columns,
+        fn,
+        args: Optional[List[str]] = None,
+        types=None,
+    ) -> None:
+        """Declare computed column(s) bound to a UDF (async).
+
+        .. deprecated::
+            Use ``add_columns(computed={"col": fn("input_col")})`` -- a computed
+            column is an expression over a registered function, so bind it that
+            way instead of coupling the UDF to the column here.
+        """
+        import warnings
+
+        warnings.warn(
+            "add_computed_column is deprecated; use add_columns(computed="
+            '{"col": fn("input_col")}).',
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        from .udf import Udf, struct_field_types
+
+        multi = isinstance(columns, (tuple, list))
+        if isinstance(fn, Udf):
+            expr = fn.expression(*(args or []))
+            if types is None:
+                if multi:
+                    if not fn.returns.upper().startswith("STRUCT"):
+                        raise ValueError(
+                            "several columns need a STRUCT-returning function"
+                        )
+                    types = struct_field_types(fn.returns)
+                else:
+                    types = fn.returns
        else:
-            return await self._inner.add_columns(list(transforms.items()))
+            if types is None:
+                raise ValueError("pass types= when fn is a name string")
+            expr = f"{fn}({', '.join(args or [])})"
+        if multi:
+            if len(types) != len(columns):
+                raise ValueError(
+                    f"{len(columns)} columns but {len(types)} output types"
+                )
+            computed = {c: (t, expr) for c, t in zip(columns, types)}
+        else:
+            computed = {columns: (types, expr)}
+        await self.add_columns(computed=computed)

    async def alter_columns(
        self, *alterations: Iterable[dict[str, Any]]
@@ -6349,7 +6626,7 @@ class AsyncBranches:
        if from_ref == "main":
            from_ref = None
        inner = await self._table.branches.create(name, from_ref, from_version)
-        return AsyncTable(inner)
+        return AsyncTable(inner, conn=self._table._conn)

    async def checkout(self, name: str, version: Optional[int] = None) -> "AsyncTable":
        """Check out an existing branch and return a handle scoped to it.
@@ -6363,7 +6640,10 @@ class AsyncBranches:
            handle is a read-only view of that version; when omitted it tracks
            the branch's latest and stays writable.
        """
-        return AsyncTable(await self._table.branches.checkout(name, version))
+        return AsyncTable(
+            await self._table.branches.checkout(name, version),
+            conn=self._table._conn,
+        )

    async def delete(self, name: str) -> None:
        """Delete a branch."""
@@ -0,0 +1,847 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+"""UDF authoring for LanceDB derived compute (server-backed).
+
+`@udf` / `@table_udf` turn a plain Python function into a registrable
+server-side UDF: a cloudpickled (or source) body, a SQL signature inferred
+from type hints, and the runtime options (pip deps, GPUs, batching, ...).
+Register and use them through the existing connection/table API:
+
+    import lancedb
+    from lancedb import udf, table_udf
+
+    db = lancedb.connect("db://my_db", api_key="...", host_override="...")
+
+    @udf(pip=["torch>=2.0"], num_gpus=1)
+    def embed(text: str) -> list[float]:
+        return model.encode(text).tolist()
+
+    db.create_function(embed)                 # CREATE FUNCTION (once)
+    tbl = db.open_table("docs")
+    tbl.add_columns(computed={"vec": embed("text")})  # bind embed(text) -> vec
+    tbl.refresh_column("vec").wait()          # materialize (returns a Job)
+    view = db.create_materialized_view("chunks", tbl, ["id", chunk_fn])
+
+`embed("text")` applies the registered function to the `text` column and yields
+the expression `embed(text)`; the function itself stays decoupled from any
+column, so the same `embed` works on any column or table.
+
+These operations are server-backed (LanceDB Enterprise / Cloud); the
+decorator itself works locally (define + call), only registration needs a
+remote connection.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import dataclasses
+import functools
+import inspect
+import re
+import sys
+import textwrap
+import json
+import time
+import typing
+
+# -- type hints -> SQL type strings -------------------------------------
+
+_SCALARS = {
+    int: "BIGINT",
+    # Pragmatic default for ML workloads: python float maps to FLOAT
+    # (Float32). Use an explicit `returns=` for DOUBLE.
+    float: "FLOAT",
+    str: "VARCHAR",
+    bool: "BOOLEAN",
+    bytes: "BLOB",
+}
+
+
+class TypeInferenceError(TypeError):
+    pass
+
+
+def sql_type(hint) -> str:
+    """SQL type string for a python type hint."""
+    if hint in _SCALARS:
+        return _SCALARS[hint]
+    origin = typing.get_origin(hint)
+    if origin in (list, typing.List):
+        (item,) = typing.get_args(hint) or (None,)
+        if item in _SCALARS:
+            return f"{_SCALARS[item]}[]"
+        raise TypeInferenceError(
+            f"unsupported list item type {item!r}; use an explicit returns="
+        )
+    fields = _struct_fields(hint)
+    if fields is not None:
+        inner = ", ".join(f"{name} {sql_type(h)}" for name, h in fields)
+        return f"STRUCT({inner})"
+    raise TypeInferenceError(
+        f"cannot infer a SQL type for {hint!r}; pass an explicit type string"
+    )
+
+
+def _struct_fields(hint):
+    """(name, hint) pairs for a TypedDict or dataclass, else None."""
+    if dataclasses.is_dataclass(hint):
+        return [(f.name, f.type) for f in dataclasses.fields(hint)]
+    # TypedDict detection: a dict subclass with __annotations__.
+    if (
+        isinstance(hint, type)
+        and issubclass(hint, dict)
+        and typing.get_type_hints(hint)
+    ):
+        return list(typing.get_type_hints(hint).items())
+    return None
+
+
+def return_type(fn, override: "str | None", table: bool) -> str:
+    """SQL return type for a function: explicit override wins, else the
+    return annotation. Table functions render as TABLE(...) and accept
+    struct-shaped hints (TypedDict/dataclass, optionally list-wrapped)."""
+    if override is not None:
+        s = override.strip()
+        if table and not s.upper().startswith("TABLE"):
+            if s.upper().startswith("STRUCT"):
+                return "TABLE" + s[len("STRUCT") :]
+            raise TypeInferenceError(
+                "a table function's returns= must be TABLE(...) or STRUCT(...)"
+            )
+        return s
+
+    hints = typing.get_type_hints(fn)
+    ret = hints.get("return")
+    if ret is None:
+        raise TypeInferenceError(
+            f"function {fn.__name__!r} needs a return annotation or returns="
+        )
+    if table:
+        # Accept list[Row] / Row where Row is a TypedDict or dataclass.
+        if typing.get_origin(ret) in (list, typing.List):
+            (ret,) = typing.get_args(ret)
+        fields = _struct_fields(ret)
+        if fields is None:
+            raise TypeInferenceError(
+                "a table function must return rows shaped as a TypedDict or "
+                "dataclass (optionally list-wrapped); or pass returns=..."
+            )
+        inner = ", ".join(f"{name} {sql_type(h)}" for name, h in fields)
+        return f"TABLE({inner})"
+    return sql_type(ret)
+
+
+def param_types(fn) -> "list[tuple[str, str]]":
+    """(name, sql type) per parameter, from annotations. Each UDF
+    parameter binds to a source column of the same name by default."""
+    hints = typing.get_type_hints(fn)
+    out = []
+    for name, p in inspect.signature(fn).parameters.items():
+        if p.kind in (p.VAR_POSITIONAL, p.VAR_KEYWORD):
+            raise TypeInferenceError("*args/**kwargs are not supported in UDFs")
+        hint = hints.get(name)
+        if hint is None:
+            raise TypeInferenceError(
+                f"parameter {name!r} of {fn.__name__!r} needs a type annotation"
+            )
+        out.append((name, sql_type(hint)))
+    return out
+
+
+# -- column expressions -------------------------------------------------
+
+
+class ColumnExpr(str):
+    """A computed-column expression produced by applying a registered
+    function to column names, e.g. ``embed("data") -> "embed(data)"``.
+
+    It IS the expression string everywhere a string is expected (views, SQL,
+    logging), and additionally carries the function's declared return type so
+    ``add_columns(computed=...)`` can declare the column without a hand-written
+    type. ``field_types`` holds the per-field SQL types of a STRUCT return, for
+    fanning one expression out to several columns.
+    """
+
+    data_type: "str | None"
+    field_types: "list[str] | None"
+
+    def __new__(cls, expr: str, data_type=None, field_types=None):
+        obj = super().__new__(cls, expr)
+        obj.data_type = data_type
+        obj.field_types = field_types
+        return obj
+
+
+def _normalize_computed(computed: dict) -> dict:
+    """Normalize the user-facing ``computed=`` mapping to the canonical
+    ``{name: (sql_type, expression)}`` form.
+
+    Accepts, per entry:
+      - value is a `ColumnExpr` (from ``fn("col")``): the column's SQL type
+        comes from the function's return type -- no hand-written type needed. A
+        tuple key (``("chunk", "idx")``) fans a STRUCT return out to one
+        (type, expression) entry per field, in declared order.
+      - value is a legacy ``(sql_type, expression)`` tuple: passed through (the
+        escape hatch, e.g. bare-name function strings).
+    """
+    out: dict = {}
+    for key, val in computed.items():
+        if isinstance(val, ColumnExpr):
+            expr = str(val)
+            if isinstance(key, (tuple, list)):
+                if not val.field_types:
+                    raise ValueError(
+                        f"columns {tuple(key)} need a STRUCT-returning function; "
+                        f"{expr} returns a single value"
+                    )
+                if len(val.field_types) != len(key):
+                    raise ValueError(
+                        f"{len(key)} columns but {len(val.field_types)} struct fields "
+                        f"in {expr}"
+                    )
+                for name, t in zip(key, val.field_types):
+                    out[name] = (t, expr)
+            else:
+                if val.data_type is None:
+                    raise ValueError(f"cannot infer a type for {expr}; pass types=")
+                out[key] = (val.data_type, expr)
+        else:
+            out[key] = val
+    return out
+
+
+# -- the @udf / @table_udf decorators -----------------------------------
+
+
+class Udf:
+    def __init__(
+        self,
+        fn,
+        *,
+        returns: "str | None" = None,
+        table: bool = False,
+        name: "str | None" = None,
+        pip: "list[str] | None" = None,
+        pip_index_url: "str | None" = None,
+        pip_extra_index_urls: "list[str] | None" = None,
+        find_links: "list[str] | None" = None,
+        requirements: "str | list[str] | None" = None,
+        conda: "list[str] | None" = None,
+        conda_channels: "list[str] | None" = None,
+        env: "dict[str, str] | list[str] | None" = None,
+        num_cpus: "int | None" = None,
+        num_gpus: "int | None" = None,
+        batch_size: "int | None" = None,
+        timeout: "float | None" = None,
+        error_policy: "str | None" = None,
+        max_skip_ratio: "float | None" = None,
+        retries: "int | None" = None,
+        docker_image: "str | None" = None,
+        description: "str | None" = None,
+        prefer_source: bool = False,
+    ):
+        functools.update_wrapper(self, fn)
+        self.fn = fn
+        self.name = name or fn.__name__
+        self.table = table
+        self.params = param_types(fn)
+        self.returns = return_type(fn, returns, table)
+        self.prefer_source = prefer_source
+        self.options: "dict[str, str]" = {}
+        if conda and (pip or requirements):
+            raise ValueError("pass conda or pip/requirements, not both")
+        if conda_channels and not conda:
+            raise ValueError("conda_channels requires conda")
+        if pip:
+            self.options["pip"] = ",".join(pip)
+        if pip_extra_index_urls:
+            self.options["pip_extra_index_urls"] = ",".join(pip_extra_index_urls)
+        if find_links:
+            self.options["find_links"] = ",".join(find_links)
+        if requirements:
+            self.options["requirements"] = _format_requirements(requirements)
+        if conda:
+            self.options["conda"] = ",".join(conda)
+        if conda_channels:
+            self.options["conda_channels"] = ",".join(conda_channels)
+        if env:
+            self.options["env"] = _format_env(env)
+        for key, val in [
+            ("pip_index_url", pip_index_url),
+            ("num_cpus", num_cpus),
+            ("num_gpus", num_gpus),
+            ("batch_size", batch_size),
+            ("timeout", timeout),
+            ("error_policy", error_policy),
+            ("max_skip_ratio", max_skip_ratio),
+            ("retries", retries),
+            ("docker_image", docker_image),
+        ]:
+            if val is not None:
+                self.options[key] = str(val)
+        # Keep the source in the description (when available) so the
+        # catalog stays inspectable even for pickled bodies.
+        if description is not None:
+            self.options["description"] = description
+        else:
+            try:
+                self.options["description"] = textwrap.dedent(inspect.getsource(fn))
+            except (OSError, TypeError):
+                pass
+
+    def __call__(self, *args, **kwargs):
+        """Call with real values to run locally; call with column-name
+        strings to build an expression for backfills and views, e.g.
+        ``embed("data")`` -> the expression ``embed(data)`` (a `ColumnExpr`
+        carrying the function's return type for `add_columns(computed=...)`)."""
+        if args and all(isinstance(a, str) for a in args) and not kwargs:
+            return self.expression(*args)
+        return self.fn(*args, **kwargs)
+
+    def expression(self, *columns: str) -> ColumnExpr:
+        """The expression applying this function to `columns` (default: the
+        function's own parameter names). Returns a `ColumnExpr` -- a string
+        that also carries the declared return type (and struct field types)."""
+        cols = columns or [p for p, _ in self.params]
+        expr = f"{self.name}({', '.join(cols)})"
+        field_types = None
+        if self.returns.upper().startswith("STRUCT"):
+            field_types = struct_field_types(self.returns)
+        return ColumnExpr(expr, data_type=self.returns, field_types=field_types)
+
+    def _body(self) -> "tuple[str, str]":
+        """(body literal, body_format). Source when requested and
+        retrievable; cloudpickle otherwise (handles closures)."""
+        if self.prefer_source:
+            try:
+                src = textwrap.dedent(inspect.getsource(self.fn))
+                # Strip the decorator line(s) so the stored body is a
+                # plain function definition.
+                lines = src.splitlines(keepends=True)
+                while lines and lines[0].lstrip().startswith("@"):
+                    lines.pop(0)
+                return "".join(lines), "source"
+            except (OSError, TypeError):
+                pass
+        import cloudpickle
+
+        raw = cloudpickle.dumps(self.fn)
+        return base64.b64encode(raw).decode("ascii"), "cloudpickle"
+
+    def _body_and_options(self) -> "tuple[str, dict[str, str]]":
+        """The body literal plus the finalized options (body_format /
+        python_version / cloudpickle-pip bookkeeping for a non-source
+        body)."""
+        body, body_format = self._body()
+        options = dict(self.options)
+        if body_format != "source":
+            options["body_format"] = body_format
+            # Pickled code objects only load under the same interpreter
+            # minor version; record ours so the worker can fail with a
+            # clear message instead of a bytecode error.
+            options["python_version"] = self.pickle_environment()
+            # The worker deserializes the body with cloudpickle; make sure
+            # the job's pip environment provides it. Conda bakes inject
+            # cloudpickle server-side, so do not create an invalid pip+conda
+            # declaration here.
+            if "conda" not in options:
+                pip = [d for d in options.get("pip", "").split(",") if d]
+                if not any(d.startswith("cloudpickle") for d in pip):
+                    pip.append("cloudpickle")
+                options["pip"] = ",".join(pip)
+        return body, options
+
+    def create_request(self) -> dict:
+        """Keyword arguments for `connection.create_function`."""
+        body, options = self._body_and_options()
+        return {
+            "name": self.name,
+            "language": "python",
+            "return_type": self.returns,
+            "body": body,
+            "options": options,
+        }
+
+    def create_statement(self) -> str:
+        """The equivalent `CREATE FUNCTION` SQL (for SQL-surface callers)."""
+        params = ", ".join(f"{n} {t}" for n, t in self.params)
+        body, options = self._body_and_options()
+        with_clause = ""
+        if options:
+            rendered = ", ".join(
+                f"{k} = '{_escape(v)}'" for k, v in sorted(options.items())
+            )
+            with_clause = f" WITH ({rendered})"
+        return (
+            f"CREATE FUNCTION {self.name}({params}) RETURNS {self.returns} "
+            f"LANGUAGE python AS '{_escape_body(body)}'{with_clause}"
+        )
+
+    def pickle_environment(self) -> str:
+        """Python version the body pickles under -- workers should match
+        the minor version for cloudpickle compatibility."""
+        return f"{sys.version_info.major}.{sys.version_info.minor}"
+
+
+def _escape(s: str) -> str:
+    return str(s).replace("'", "''")
+
+
+def _format_requirements(requirements: "str | list[str]") -> str:
+    if isinstance(requirements, str):
+        return requirements
+    return "\n".join(str(req) for req in requirements)
+
+
+def _format_env(env: "dict[str, str] | list[str]") -> str:
+    if isinstance(env, dict):
+        return "; ".join(f"{key}={value}" for key, value in env.items())
+    return "; ".join(str(entry) for entry in env)
+
+
+def _escape_body(body: str) -> str:
+    # The server unescapes \n / \t in single-quoted bodies; encode real
+    # newlines accordingly and escape quotes.
+    return (
+        body.replace("\\", "\\\\")
+        .replace("'", "''")
+        .replace("\n", "\\n")
+        .replace("\t", "\\t")
+    )
+
+
+def udf(fn=None, **kwargs):
+    """Decorate a function as a scalar (or struct-returning) UDF.
+
+    @udf
+    def doubled(val: int) -> float: ...
+
+    @udf(pip=["torch>=2"], num_gpus=1)
+    def embed(body: str) -> list[float]: ...
+    """
+    if fn is not None:
+        return Udf(fn, **kwargs)
+    return lambda f: Udf(f, **kwargs)
+
+
+def table_udf(fn=None, **kwargs):
+    """Decorate a table function (UDTF): each input row may emit zero or
+    more output rows. Only usable in materialized views.
+
+        class Chunk(TypedDict):
+            chunk: str
+            chunk_idx: int
+
+        @table_udf
+        def chunker(body: str) -> list[Chunk]: ...
+    """
+    kwargs["table"] = True
+    if fn is not None:
+        return Udf(fn, **kwargs)
+    return lambda f: Udf(f, **kwargs)
+
+
+# -- view / job handles (thin references over a connection) -------------
+
+
+def struct_field_types(returns: str) -> "list[str]":
+    """Field type strings of a STRUCT(...) SQL type, in declared order."""
+    inner = returns.strip()[len("STRUCT(") : -1]
+    fields, depth, start = [], 0, 0
+    for i, c in enumerate(inner):
+        if c in "([":
+            depth += 1
+        elif c in ")]":
+            depth -= 1
+        elif c == "," and depth == 0:
+            fields.append(inner[start:i].strip())
+            start = i + 1
+    fields.append(inner[start:].strip())
+    # Each field is "name TYPE"; drop the name.
+    return [f.split(None, 1)[1] for f in fields]
+
+
+def build_view_query(source, select) -> str:
+    """Assemble a view SELECT from a source (name or table) and select
+    items: a column name, an expression string, a (alias, expression)
+    tuple, or a @udf/@table_udf object."""
+    src = source.name if hasattr(source, "name") else source
+    items = []
+    for item in select:
+        if isinstance(item, Udf):
+            items.append(item.expression())
+        elif isinstance(item, tuple):
+            alias, expr = item
+            expr = expr.expression() if isinstance(expr, Udf) else expr
+            items.append(f"{expr} AS {alias}")
+        else:
+            items.append(item)
+    return f"SELECT {', '.join(items)} FROM {src}"
+
+
+def _job_id_matches(handle_id: str, listed_id: str) -> bool:
+    # The refresh/backfill endpoints return the submission id (a uuid), but
+    # the agent names the manifest job "<table>-<type>-<first 8 of the
+    # submission id>" -- which is what list_jobs and cancel report. Match the
+    # canonical id directly, or by that submission prefix.
+    if listed_id == handle_id:
+        return True
+    prefix = handle_id[:8]
+    return len(prefix) >= 4 and prefix in listed_id
+
+
+class MaterializedView:
+    """A reference to a materialized view (name + connection). Operations are
+    server-backed connection calls bound to the name.
+
+    ``create_materialized_view`` returns one of these; ``job_id`` is the
+    initial-population job (None when the view was created with no data), so
+    ``db.create_materialized_view(...).wait()`` blocks until it is populated.
+    """
+
+    def __init__(self, conn, name: str, job_id: "str | None" = None):
+        self.conn = conn
+        self.name = name
+        #: initial-population job id from create, or None (with_no_data).
+        self.job_id = job_id
+
+    def wait(self, timeout: float = 3600.0, poll: float = 2.0) -> str:
+        """Block until the initial-population job (from create) finishes.
+        A no-op when the view was created with no data."""
+        if self.job_id is None:
+            return "finished"
+        return Job(self.conn, self.job_id, table=self.name).wait(
+            timeout=timeout, poll=poll
+        )
+
+    def refresh(self, full: bool = False) -> "Job":
+        """Refresh the materialized view; returns a `Job` to wait on,
+        poll, or cancel (``view.refresh().wait()``).
+
+        ``full=True`` forces a full rebuild (recompute and replace every row)
+        instead of the default incremental refresh. A full rebuild preserves
+        the view's indexes -- they are reindexed by the distributed indexer.
+        """
+        job_id = self.conn._refresh_materialized_view(self.name, full=full)
+        return Job(self.conn, job_id, table=self.name)
+
+    def explain_refresh(self, full: bool = False):
+        """Plan a refresh without running it (EXPLAIN REFRESH)."""
+        return self.conn.explain_refresh_materialized_view(self.name, full=full)
+
+    def alter(self, auto_refresh: bool) -> None:
+        self.conn.alter_materialized_view(self.name, auto_refresh=auto_refresh)
+
+    def drop(self) -> None:
+        self.conn.drop_materialized_view(self.name)
+
+    # A materialized view is a first-class table: it can be indexed and
+    # searched like any other. These open the materialized dataset by name and
+    # delegate. Indexes declared this way are recorded against the view, so the
+    # engine re-applies them after a full refresh rebuilds the dataset (a full
+    # refresh overwrites the dataset, which would otherwise drop its indices).
+    def _table(self):
+        return self.conn.open_table(self.name)
+
+    def create_index(self, *args, **kwargs):
+        """Build an index on the materialized view (see Table.create_index)."""
+        return self._table().create_index(*args, **kwargs)
+
+    def create_scalar_index(self, *args, **kwargs):
+        """Build a scalar index on the materialized view."""
+        return self._table().create_scalar_index(*args, **kwargs)
+
+    def create_fts_index(self, *args, **kwargs):
+        """Build a full-text-search index on the materialized view."""
+        return self._table().create_fts_index(*args, **kwargs)
+
+    def search(self, *args, **kwargs):
+        """Search the materialized view (vector / FTS / hybrid)."""
+        return self._table().search(*args, **kwargs)
+
+    def lineage(self, column=None, *, direction=None, depth=None):
+        """Lineage of the materialized view (or one of its columns). Delegates
+        to the backing table; the server already includes the view's sources
+        and downstream dependents. Returns a `Lineage`."""
+        return self._table().lineage(column, direction=direction, depth=depth)
+
+
+_PROGRESS = re.compile(r"(\d+)/(\d+)")
+
+
+class JobFailedError(RuntimeError):
+    """Raised by ``Job.wait()`` when the server reports the job ``failed``.
+
+    Carries the server-side error so a doomed backfill (e.g. a multi-column
+    ``REFRESH COLUMN`` of a scalar UDF) surfaces its real cause promptly,
+    instead of the caller blocking until ``wait()``'s timeout.
+    """
+
+    def __init__(self, job_id: str, error: "str | None"):
+        self.job_id = job_id
+        self.error = error
+        super().__init__(f"job {job_id} failed: {error or 'unknown error'}")
+
+
+class Job:
+    """A reference to a server-side job, backed by the platform jobs API.
+
+    Holds the submission (manifest) id and resolves the platform job id
+    lazily; ``status``/``progress``/``wait`` read the registry-backed
+    describe endpoint, so terminal states and errors are first-class.
+    """
+
+    #: How long an unresolved job is treated as still materializing
+    #: (submission -> dispatch -> registry record is async).
+    GRACE_SECONDS = 20.0
+
+    #: Platform lifecycle state -> the user-facing vocabulary.
+    _STATES = {
+        "IN_PROGRESS": "running",
+        "DONE": "finished",
+        "FAILED": "failed",
+        "CANCELLED": "cancelled",
+    }
+
+    def __init__(self, conn, job_id: str, table: "str | None" = None):
+        self.conn = conn
+        #: The submission (manifest) id the launching call handed out.
+        self.id = job_id
+        #: The job's table, when known -- narrows platform-id resolution.
+        self.table = table
+        self._platform_id: "str | None" = None
+        self._created = time.monotonic()
+        self._finished = False
+
+    @classmethod
+    def _completed(cls, conn=None, table: "str | None" = None) -> "Job":
+        """A job for work that completed synchronously within the call that
+        returned it (native tables, scalar/FTS builds). ``status``/``wait``
+        report ``finished`` immediately and ``cancel`` is a no-op."""
+        job = cls(conn, "", table)
+        job._finished = True
+        return job
+
+    def _resolve(self) -> "str | None":
+        if self._platform_id is None:
+            self._platform_id = self.conn.resolve_platform_job_id(self.id, self.table)
+        return self._platform_id
+
+    def _describe(self):
+        platform_id = self._resolve()
+        if platform_id is None:
+            return None
+        return self.conn.describe_platform_job(platform_id)
+
+    @staticmethod
+    def _payload(described) -> dict:
+        # Older records carry the status-store URI string instead of a
+        # payload; anything non-dict means "no structured status".
+        try:
+            payload = json.loads(described.status_json)
+        except (TypeError, ValueError):
+            return {}
+        return payload if isinstance(payload, dict) else {}
+
+    def status(self) -> str:
+        """pending / running / finished / failed / cancelled (or unknown
+        when the job never appeared in the registry)."""
+        if self._finished:
+            return "finished"
+        described = self._describe()
+        if described is not None:
+            return self._STATES.get(described.job_state, described.job_state)
+        if time.monotonic() - self._created < self.GRACE_SECONDS:
+            return "pending"
+        return "unknown"
+
+    def progress(self) -> "tuple[int, int] | None":
+        """(units_done, units_total) once workers have published progress."""
+        if self._finished:
+            return None
+        described = self._describe()
+        if described is None:
+            return None
+        payload = self._payload(described)
+        if payload.get("units_total") is not None:
+            return payload.get("units_done") or 0, payload["units_total"]
+        return None
+
+    def wait(self, timeout: float = 3600.0, poll: float = 2.0) -> str:
+        if self._finished:
+            return "finished"
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            described = self._describe()
+            if described is None:
+                if time.monotonic() - self._created > self.GRACE_SECONDS:
+                    raise JobFailedError(
+                        self.id,
+                        "job did not appear in the job registry within the "
+                        "grace period",
+                    )
+                time.sleep(min(poll, 0.5))
+                continue
+            state = self._STATES.get(described.job_state, described.job_state)
+            if state == "finished":
+                return state
+            if state == "cancelled":
+                return state
+            if state == "failed":
+                raise JobFailedError(self.id, self._payload(described).get("error"))
+            time.sleep(poll)
+        raise TimeoutError(f"job {self.id} still {self.status()} after {timeout}s")
+
+    def cancel(self) -> None:
+        """Request cancellation. Workers drain cooperatively; poll ``status``
+        for the terminal ``cancelled``."""
+        if self._finished:
+            return
+        deadline = time.monotonic() + 5.0
+        while (platform_id := self._resolve()) is None:
+            if time.monotonic() > deadline:
+                raise RuntimeError(
+                    f"job {self.id} has not registered yet; retry cancel shortly"
+                )
+            time.sleep(0.5)
+        self.conn.cancel_platform_job(platform_id)
+
+
+class AsyncMaterializedView:
+    """Async reference to a materialized view (name + async connection)."""
+
+    def __init__(self, conn, name: str, job_id: "str | None" = None):
+        self.conn = conn
+        self.name = name
+        #: initial-population job id from create, or None (with_no_data).
+        self.job_id = job_id
+
+    async def wait(self, timeout: float = 3600.0, poll: float = 2.0) -> str:
+        """Block until the initial-population job (from create) finishes.
+        A no-op when the view was created with no data."""
+        if self.job_id is None:
+            return "finished"
+        return await AsyncJob(self.conn, self.job_id, table=self.name).wait(
+            timeout=timeout, poll=poll
+        )
+
+    async def refresh(self, full: bool = False) -> "AsyncJob":
+        """Refresh the materialized view; returns an `AsyncJob` to wait
+        on, poll, or cancel.
+
+        ``full=True`` forces a full rebuild instead of an incremental refresh
+        (indexes are preserved and reindexed by the distributed indexer).
+        """
+        job_id = await self.conn._refresh_materialized_view(self.name, full=full)
+        return AsyncJob(self.conn, job_id, table=self.name)
+
+    async def explain_refresh(self, full: bool = False):
+        return await self.conn.explain_refresh_materialized_view(self.name, full=full)
+
+    async def alter(self, auto_refresh: bool) -> None:
+        await self.conn.alter_materialized_view(self.name, auto_refresh=auto_refresh)
+
+    async def drop(self) -> None:
+        await self.conn.drop_materialized_view(self.name)
+
+    async def lineage(self, column=None, *, direction=None, depth=None):
+        """Lineage of the materialized view (or column). Returns a `Lineage`."""
+        return await self.conn.lineage(
+            self.name, column, direction=direction, depth=depth
+        )
+
+
+class AsyncJob:
+    """Async reference to a server-side job, backed by the platform jobs API.
+
+    Same contract as `Job` with awaitable methods.
+    """
+
+    GRACE_SECONDS = 20.0
+    _STATES = Job._STATES
+
+    def __init__(self, conn, job_id: str, table: "str | None" = None):
+        self.conn = conn
+        self.id = job_id
+        self.table = table
+        self._platform_id: "str | None" = None
+        self._created = time.monotonic()
+        self._finished = False
+
+    @classmethod
+    def _completed(cls, conn=None, table: "str | None" = None) -> "AsyncJob":
+        """See ``Job._completed``."""
+        job = cls(conn, "", table)
+        job._finished = True
+        return job
+
+    async def _resolve(self) -> "str | None":
+        if self._platform_id is None:
+            self._platform_id = await self.conn.resolve_platform_job_id(
+                self.id, self.table
+            )
+        return self._platform_id
+
+    async def _describe(self):
+        platform_id = await self._resolve()
+        if platform_id is None:
+            return None
+        return await self.conn.describe_platform_job(platform_id)
+
+    async def status(self) -> str:
+        if self._finished:
+            return "finished"
+        described = await self._describe()
+        if described is not None:
+            return self._STATES.get(described.job_state, described.job_state)
+        if time.monotonic() - self._created < self.GRACE_SECONDS:
+            return "pending"
+        return "unknown"
+
+    async def progress(self) -> "tuple[int, int] | None":
+        if self._finished:
+            return None
+        described = await self._describe()
+        if described is None:
+            return None
+        payload = Job._payload(described)
+        if payload.get("units_total") is not None:
+            return payload.get("units_done") or 0, payload["units_total"]
+        return None
+
+    async def wait(self, timeout: float = 3600.0, poll: float = 2.0) -> str:
+        if self._finished:
+            return "finished"
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            described = await self._describe()
+            if described is None:
+                if time.monotonic() - self._created > self.GRACE_SECONDS:
+                    raise JobFailedError(
+                        self.id,
+                        "job did not appear in the job registry within the "
+                        "grace period",
+                    )
+                await asyncio.sleep(min(poll, 0.5))
+                continue
+            state = self._STATES.get(described.job_state, described.job_state)
+            if state in ("finished", "cancelled"):
+                return state
+            if state == "failed":
+                raise JobFailedError(self.id, Job._payload(described).get("error"))
+            await asyncio.sleep(poll)
+        raise TimeoutError(f"job {self.id} still {await self.status()} after {timeout}s")
+
+    async def cancel(self) -> None:
+        if self._finished:
+            return
+        deadline = time.monotonic() + 5.0
+        while (platform_id := await self._resolve()) is None:
+            if time.monotonic() > deadline:
+                raise RuntimeError(
+                    f"job {self.id} has not registered yet; retry cancel shortly"
+                )
+            await asyncio.sleep(0.5)
+        await self.conn.cancel_platform_job(platform_id)
+
@@ -0,0 +1,209 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+"""Job / AsyncJob against the platform jobs API.
+
+The reference resolves its submission (manifest) id to a platform job id,
+then polls describe for registry-backed state: terminal states are
+first-class (DONE / FAILED / CANCELLED), progress comes from the
+owner-written status payload, and a failed job raises ``JobFailedError``
+promptly with the server error.
+"""
+
+import asyncio
+import json
+import time
+
+import pytest
+
+from lancedb.udf import Job, AsyncJob, JobFailedError
+
+
+class FakeDescription:
+    """Mirror of the pyo3 PlatformJobDescription fields the Job reads."""
+
+    def __init__(self, job_state, status=None):
+        self.job_id = "plat-1"
+        self.job_type = "indexer"
+        self.job_subtype = "udf"
+        self.job_state = job_state
+        self.creation_ms = 0
+        self.status_json = json.dumps(status if status is not None else {})
+
+
+class FakeConn:
+    """Scripted timeline: resolve returns None until `resolve_after` calls,
+    then the platform id; describe walks a list of descriptions (holding the
+    last once exhausted)."""
+
+    def __init__(self, descriptions, resolve_after=0):
+        self._descs = list(descriptions)
+        self._resolve_after = resolve_after
+        self.resolve_calls = 0
+        self.describe_calls = 0
+        self.cancelled = []
+
+    def resolve_platform_job_id(self, manifest_job_id, table=None):
+        self.resolve_calls += 1
+        if self.resolve_calls <= self._resolve_after:
+            return None
+        return "plat-1"
+
+    def describe_platform_job(self, platform_job_id):
+        assert platform_job_id == "plat-1"
+        snap = self._descs[min(self.describe_calls, len(self._descs) - 1)]
+        self.describe_calls += 1
+        return snap
+
+    def cancel_platform_job(self, platform_job_id):
+        self.cancelled.append(platform_job_id)
+
+
+class AsyncFakeConn(FakeConn):
+    async def resolve_platform_job_id(self, manifest_job_id, table=None):
+        return FakeConn.resolve_platform_job_id(self, manifest_job_id, table)
+
+    async def describe_platform_job(self, platform_job_id):
+        return FakeConn.describe_platform_job(self, platform_job_id)
+
+    async def cancel_platform_job(self, platform_job_id):
+        return FakeConn.cancel_platform_job(self, platform_job_id)
+
+
+def test_status_maps_platform_states():
+    for wire, want in [
+        ("IN_PROGRESS", "running"),
+        ("DONE", "finished"),
+        ("FAILED", "failed"),
+        ("CANCELLED", "cancelled"),
+    ]:
+        job = Job(FakeConn([FakeDescription(wire)]), "job-1", table="t")
+        assert job.status() == want
+
+
+def test_status_pending_before_resolution():
+    job = Job(FakeConn([], resolve_after=10_000), "job-1", table="t")
+    assert job.status() == "pending"
+
+
+def test_progress_from_status_payload():
+    conn = FakeConn(
+        [
+            FakeDescription(
+                "IN_PROGRESS",
+                status={"units_done": 3, "units_total": 8, "rows_committed": 100},
+            )
+        ]
+    )
+    job = Job(conn, "job-1", table="t")
+    assert job.progress() == (3, 8)
+
+
+def test_progress_none_for_uri_only_status():
+    # Older records carry the status-store URI string, not a payload.
+    desc = FakeDescription("IN_PROGRESS")
+    desc.status_json = json.dumps("s3://bucket/job/job_status")
+    job = Job(FakeConn([desc]), "job-1", table="t")
+    assert job.progress() is None
+
+
+def test_wait_raises_on_failed_promptly():
+    conn = FakeConn(
+        [
+            FakeDescription("IN_PROGRESS"),
+            FakeDescription(
+                "FAILED", status={"error": "multi-column backfill needs a STRUCT"}
+            ),
+        ]
+    )
+    job = Job(conn, "job-1", table="t")
+    t0 = time.monotonic()
+    with pytest.raises(JobFailedError) as exc:
+        job.wait(timeout=30, poll=0.01)
+    assert time.monotonic() - t0 < 5  # prompt, nowhere near the 30s timeout
+    assert "STRUCT" in str(exc.value)
+    assert exc.value.error == "multi-column backfill needs a STRUCT"
+    assert exc.value.job_id == "job-1"
+
+
+def test_wait_returns_finished_on_done():
+    conn = FakeConn([FakeDescription("IN_PROGRESS"), FakeDescription("DONE")])
+    job = Job(conn, "job-1", table="t")
+    assert job.wait(timeout=30, poll=0.01) == "finished"
+
+
+def test_wait_returns_cancelled():
+    conn = FakeConn([FakeDescription("CANCELLED")])
+    job = Job(conn, "job-1", table="t")
+    assert job.wait(timeout=30, poll=0.01) == "cancelled"
+
+
+def test_wait_raises_when_job_never_registers():
+    # An unresolved job past the grace window is a lost submission, not an
+    # eternal "pending" hang.
+    conn = FakeConn([], resolve_after=10_000)
+    job = Job(conn, "job-1", table="t")
+    job.GRACE_SECONDS = 0.05
+    job._created = time.monotonic() - 1.0
+    with pytest.raises(JobFailedError) as exc:
+        job.wait(timeout=5, poll=0.01)
+    assert "registry" in str(exc.value)
+
+
+def test_cancel_resolves_then_cancels():
+    conn = FakeConn([FakeDescription("IN_PROGRESS")], resolve_after=1)
+    job = Job(conn, "job-1", table="t")
+    job.cancel()
+    assert conn.cancelled == ["plat-1"]
+
+
+def test_async_wait_raises_on_failed_promptly():
+    conn = AsyncFakeConn(
+        [FakeDescription("FAILED", status={"error": "boom"})],
+    )
+    job = AsyncJob(conn, "job-1", table="t")
+
+    async def run():
+        t0 = time.monotonic()
+        with pytest.raises(JobFailedError) as exc:
+            await job.wait(timeout=30, poll=0.01)
+        assert time.monotonic() - t0 < 5
+        assert exc.value.error == "boom"
+
+    asyncio.run(run())
+
+
+def test_async_wait_returns_finished():
+    conn = AsyncFakeConn([FakeDescription("IN_PROGRESS"), FakeDescription("DONE")])
+    job = AsyncJob(conn, "job-1", table="t")
+
+    async def run():
+        assert await job.wait(timeout=30, poll=0.01) == "finished"
+
+    asyncio.run(run())
+
+
+def test_completed_job_is_finished_without_conn():
+    job = Job._completed(table="t")
+    assert job.status() == "finished"
+    assert job.wait(timeout=0.01) == "finished"
+    assert job.progress() is None
+    job.cancel()  # no-op, must not touch a connection
+
+
+def test_completed_job_ignores_registry():
+    conn = FakeConn([FakeDescription("IN_PROGRESS")])
+    job = Job._completed(conn, table="t")
+    assert job.wait(timeout=0.01) == "finished"
+    assert conn.resolve_calls == 0
+    assert conn.describe_calls == 0
+
+
+def test_completed_async_job_is_finished():
+    async def run():
+        job = AsyncJob._completed(table="t")
+        assert await job.status() == "finished"
+        assert await job.wait(timeout=0.01) == "finished"
+        assert await job.progress() is None
+        await job.cancel()
+
+    asyncio.run(run())
@@ -128,9 +128,16 @@ def test_split_hash(mem_db):

 def test_split_hash_with_discard(mem_db):
    """Test hash-based splitting with discard weight."""
+    total_rows = 1000
    tbl = mem_db.create_table(
        "test_table",
-        pa.table({"id": range(100), "category": ["A", "B"] * 50, "value": range(100)}),
+        pa.table(
+            {
+                "id": range(total_rows),
+                "category": [f"category-{i}" for i in range(total_rows)],
+                "value": range(total_rows),
+            }
+        ),
    )

    permutation_tbl = (
@@ -142,10 +149,12 @@ def test_split_hash_with_discard(mem_db):
        .execute()
    )

-    # Should have fewer than 100 rows due to discard
+    # Should have fewer rows due to discard, but should not be empty.
    row_count = permutation_tbl.count_rows()
-    assert row_count < 100
-    assert row_count > 0  # But not empty
+    assert 0 < row_count < total_rows
+
+    data = permutation_tbl.search(None).to_arrow().to_pydict()
+    assert set(data["split_id"]) == {0, 1}


 def test_split_sequential(mem_db):
@@ -1273,7 +1273,7 @@ async def test_explain_plan_fts(table_async: AsyncTable):
    query = await table_async.search("dog", query_type="fts", fts_columns="text")
    plan = await query.explain_plan()
    # Should show FTS details (issue #2465 is now fixed)
-    assert "MatchQuery: column=text, query=dog" in plan
+    assert "MatchQuery: column=text, query=[dog]" in plan
    assert "GlobalLimitExec" in plan  # Default limit

    # Test FTS query with limit
@@ -1281,7 +1281,7 @@ async def test_explain_plan_fts(table_async: AsyncTable):
        "dog", query_type="fts", fts_columns="text"
    )
    plan_with_limit = await query_with_limit.limit(1).explain_plan()
-    assert "MatchQuery: column=text, query=dog" in plan_with_limit
+    assert "MatchQuery: column=text, query=[dog]" in plan_with_limit
    assert "GlobalLimitExec: skip=0, fetch=1" in plan_with_limit

    # Test FTS query with offset and limit
@@ -1289,7 +1289,7 @@ async def test_explain_plan_fts(table_async: AsyncTable):
        "dog", query_type="fts", fts_columns="text"
    )
    plan_with_offset = await query_with_offset.offset(1).limit(1).explain_plan()
-    assert "MatchQuery: column=text, query=dog" in plan_with_offset
+    assert "MatchQuery: column=text, query=[dog]" in plan_with_offset
    assert "GlobalLimitExec: skip=1, fetch=1" in plan_with_offset


@@ -1333,7 +1333,7 @@ async def test_explain_plan_with_filters(table_async: AsyncTable):
        "dog", query_type="fts", fts_columns="text"
    )
    plan_fts_filter = await query_fts_filter.where("id = 1").explain_plan()
-    assert "MatchQuery: column=text, query=dog" in plan_fts_filter
+    assert "MatchQuery: column=text, query=[dog]" in plan_fts_filter
    assert "LanceRead" in plan_fts_filter
    assert "full_filter=id = Int64(1)" in plan_fts_filter  # Should show filter details

@@ -18,7 +18,10 @@ use lancedb::{
    connection::Connection as LanceConnection,
    connection::NamespaceClientPushdownOperation,
    database::namespace::LanceNamespaceDatabase,
-    database::{CreateTableMode, Database, ReadConsistency},
+    database::{
+        CreateFunctionRequest, CreateMaterializedViewRequest, CreateTableMode, Database,
+        ReadConsistency, RefreshMaterializedViewRequest, TableLineageRequest,
+    },
 };
 use pyo3::{
    Bound, FromPyObject, Py, PyAny, PyRef, PyResult, Python,
@@ -27,6 +30,107 @@ use pyo3::{
    types::{PyDict, PyDictMethods},
 };

+/// A registered function, as returned by `list_functions`.
+#[pyclass(get_all)]
+#[derive(Clone)]
+pub struct FunctionInfo {
+    pub name: String,
+    pub language: String,
+    pub return_type: String,
+    pub description: String,
+}
+
+/// A registered materialized view definition.
+#[pyclass(get_all)]
+#[derive(Clone)]
+pub struct MaterializedViewInfo {
+    pub name: String,
+    pub source_table: String,
+    pub projection: Vec<String>,
+    pub udf_columns: Vec<String>,
+    pub filter: Option<String>,
+    pub auto_refresh: bool,
+}
+
+/// One inflight server-side job.
+#[pyclass(get_all)]
+#[derive(Clone)]
+pub struct JobInfo {
+    pub table: String,
+    pub job_id: String,
+    pub job_type: String,
+    pub state: String,
+    pub column: Option<String>,
+    pub age_seconds: Option<i64>,
+    pub command: Option<String>,
+    pub units_done: Option<i64>,
+    pub units_total: Option<i64>,
+    pub committed: bool,
+    pub rows_skipped: u64,
+    pub error: Option<String>,
+}
+
+/// A described platform job (POST /v1/jobs/describe).
+#[pyclass(get_all)]
+#[derive(Clone)]
+pub struct PlatformJobDescription {
+    pub job_id: String,
+    pub job_type: String,
+    pub job_subtype: String,
+    /// "IN_PROGRESS" | "CANCELLED" | "FAILED" | "DONE".
+    pub job_state: String,
+    pub creation_ms: i64,
+    /// The owner-written status payload as a JSON string (units_done /
+    /// units_total / rows_committed / error when present).
+    pub status_json: String,
+}
+
+/// One durable, completed/terminal server-side job record (SHOW JOB HISTORY).
+#[pyclass(get_all)]
+#[derive(Clone)]
+pub struct JobHistoryEntry {
+    pub table: String,
+    pub job_id: String,
+    pub job_type: String,
+    pub state: String,
+    pub column: Option<String>,
+    pub created_ms: i64,
+    pub updated_ms: i64,
+    pub completed_ms: Option<i64>,
+    pub rows_processed: Option<i64>,
+    pub rows_skipped: Option<i64>,
+    pub error: Option<String>,
+    pub events: Option<String>,
+}
+
+/// One per-row UDF error recorded by `error_policy=skip` (SHOW ERRORS).
+#[pyclass(get_all)]
+#[derive(Clone)]
+pub struct JobErrorEntry {
+    pub job_id: String,
+    pub table: String,
+    pub column: String,
+    pub error_type: String,
+    pub error_message: String,
+    pub fragment_id: Option<i64>,
+    pub source_row_id: Option<i64>,
+    pub table_version: Option<i64>,
+    pub age_seconds: Option<i64>,
+}
+
+/// The plan a REFRESH MATERIALIZED VIEW would execute (EXPLAIN REFRESH).
+#[pyclass(get_all)]
+#[derive(Clone)]
+pub struct MvRefreshPlan {
+    pub table_name: String,
+    pub has_work: bool,
+    pub source_version: u64,
+    pub last_refreshed_version: Option<u64>,
+    pub full_refresh: bool,
+    pub rebuild: bool,
+    pub units_total: u64,
+}
+
 #[pyclass]
 pub struct Connection {
    inner: Option<LanceConnection>,
@@ -310,6 +414,357 @@ impl Connection {
        })
    }

+    #[pyo3(signature = (name, language, return_type, body, options=None))]
+    pub fn create_function(
+        self_: PyRef<'_, Self>,
+        name: String,
+        language: String,
+        return_type: String,
+        body: String,
+        options: Option<HashMap<String, String>>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            inner
+                .create_function(CreateFunctionRequest {
+                    name,
+                    language,
+                    return_type,
+                    body,
+                    options: options.unwrap_or_default(),
+                })
+                .await
+                .infer_error()
+        })
+    }
+
+    pub fn list_functions(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let functions = inner.list_functions().await.infer_error()?;
+            Ok(functions
+                .into_iter()
+                .map(|f| FunctionInfo {
+                    name: f.name,
+                    language: f.language,
+                    return_type: f.return_type,
+                    description: f.description,
+                })
+                .collect::<Vec<_>>())
+        })
+    }
+
+    pub fn drop_function(self_: PyRef<'_, Self>, name: String) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            inner.drop_function(&name).await.infer_error()
+        })
+    }
+
+    #[pyo3(signature = (name, query, auto_refresh=false, with_no_data=false, partition_by=None))]
+    pub fn create_materialized_view(
+        self_: PyRef<'_, Self>,
+        name: String,
+        query: String,
+        auto_refresh: bool,
+        with_no_data: bool,
+        partition_by: Option<String>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            inner
+                .create_materialized_view(CreateMaterializedViewRequest {
+                    name,
+                    query,
+                    auto_refresh,
+                    with_no_data,
+                    partition_by,
+                })
+                .await
+                .infer_error()
+        })
+    }
+
+    #[pyo3(signature = (name, full=false, src_version=None, num_workers=None, max_workers=None))]
+    pub fn refresh_materialized_view(
+        self_: PyRef<'_, Self>,
+        name: String,
+        full: bool,
+        src_version: Option<u64>,
+        num_workers: Option<u32>,
+        max_workers: Option<u32>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            inner
+                .refresh_materialized_view(RefreshMaterializedViewRequest {
+                    name,
+                    full,
+                    src_version,
+                    num_workers,
+                    max_workers,
+                })
+                .await
+                .infer_error()
+        })
+    }
+
+    /// Derived-compute lineage of a table/view (or column), returned as the
+    /// server's lineage JSON string (the Python layer parses it).
+    pub fn table_lineage(
+        self_: PyRef<'_, Self>,
+        name: String,
+        column: Option<String>,
+        direction: Option<String>,
+        depth: Option<u32>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            inner
+                .table_lineage(TableLineageRequest {
+                    name,
+                    column,
+                    direction,
+                    depth,
+                })
+                .await
+                .infer_error()
+        })
+    }
+
+    #[pyo3(signature = (name, full=false, src_version=None))]
+    pub fn explain_refresh_materialized_view(
+        self_: PyRef<'_, Self>,
+        name: String,
+        full: bool,
+        src_version: Option<u64>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let p = inner
+                .explain_refresh_materialized_view(&name, full, src_version)
+                .await
+                .infer_error()?;
+            Ok(MvRefreshPlan {
+                table_name: p.table_name,
+                has_work: p.has_work,
+                source_version: p.source_version,
+                last_refreshed_version: p.last_refreshed_version,
+                full_refresh: p.full_refresh,
+                rebuild: p.rebuild,
+                units_total: p.units_total,
+            })
+        })
+    }
+
+    pub fn alter_materialized_view(
+        self_: PyRef<'_, Self>,
+        name: String,
+        auto_refresh: bool,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            inner
+                .alter_materialized_view(&name, auto_refresh)
+                .await
+                .infer_error()
+        })
+    }
+
+    pub fn drop_materialized_view(
+        self_: PyRef<'_, Self>,
+        name: String,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            inner.drop_materialized_view(&name).await.infer_error()
+        })
+    }
+
+    pub fn list_materialized_views(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let views = inner.list_materialized_views().await.infer_error()?;
+            Ok(views
+                .into_iter()
+                .map(|v| MaterializedViewInfo {
+                    name: v.name,
+                    source_table: v.source_table,
+                    projection: v.projection,
+                    udf_columns: v.udf_columns,
+                    filter: v.filter,
+                    auto_refresh: v.auto_refresh,
+                })
+                .collect::<Vec<_>>())
+        })
+    }
+
+    pub fn list_jobs(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let jobs = inner.list_jobs().await.infer_error()?;
+            Ok(jobs
+                .into_iter()
+                .map(|j| JobInfo {
+                    table: j.table,
+                    job_id: j.job_id,
+                    job_type: j.job_type,
+                    state: j.state,
+                    column: j.column,
+                    age_seconds: j.age_seconds,
+                    command: j.command,
+                    units_done: j.units_done,
+                    units_total: j.units_total,
+                    committed: j.committed,
+                    rows_skipped: j.rows_skipped,
+                    error: j.error,
+                })
+                .collect::<Vec<_>>())
+        })
+    }
+
+    pub fn cancel_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            inner.cancel_job(&job_id).await.infer_error()
+        })
+    }
+
+    pub fn describe_platform_job(
+        self_: PyRef<'_, Self>,
+        platform_job_id: String,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let described = inner
+                .describe_platform_job(&platform_job_id)
+                .await
+                .infer_error()?;
+            Ok(described.map(|d| PlatformJobDescription {
+                job_id: d.job_id,
+                job_type: d.job_type,
+                job_subtype: d.job_subtype,
+                job_state: d.job_state,
+                creation_ms: d.creation_ms,
+                status_json: d.status.to_string(),
+            }))
+        })
+    }
+
+    #[pyo3(signature = (manifest_job_id, table=None))]
+    pub fn resolve_platform_job_id(
+        self_: PyRef<'_, Self>,
+        manifest_job_id: String,
+        table: Option<String>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            inner
+                .resolve_platform_job_id(&manifest_job_id, table.as_deref())
+                .await
+                .infer_error()
+        })
+    }
+
+    pub fn cancel_platform_job(
+        self_: PyRef<'_, Self>,
+        platform_job_id: String,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            inner
+                .cancel_platform_job(&platform_job_id)
+                .await
+                .infer_error()
+        })
+    }
+
+    #[pyo3(signature = (job_id, table=None))]
+    pub fn get_job(
+        self_: PyRef<'_, Self>,
+        job_id: String,
+        table: Option<String>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let job = inner
+                .get_job(&job_id, table.as_deref())
+                .await
+                .infer_error()?;
+            Ok(job.map(|j| JobInfo {
+                table: j.table,
+                job_id: j.job_id,
+                job_type: j.job_type,
+                state: j.state,
+                column: j.column,
+                age_seconds: j.age_seconds,
+                command: j.command,
+                units_done: j.units_done,
+                units_total: j.units_total,
+                committed: j.committed,
+                rows_skipped: j.rows_skipped,
+                error: j.error,
+            }))
+        })
+    }
+
+    #[pyo3(signature = (job_id=None))]
+    pub fn job_history(
+        self_: PyRef<'_, Self>,
+        job_id: Option<String>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let rows = inner.job_history(job_id.as_deref()).await.infer_error()?;
+            Ok(rows
+                .into_iter()
+                .map(|r| JobHistoryEntry {
+                    table: r.table,
+                    job_id: r.job_id,
+                    job_type: r.job_type,
+                    state: r.state,
+                    column: r.column,
+                    created_ms: r.created_ms,
+                    updated_ms: r.updated_ms,
+                    completed_ms: r.completed_ms,
+                    rows_processed: r.rows_processed,
+                    rows_skipped: r.rows_skipped,
+                    error: r.error,
+                    events: r.events,
+                })
+                .collect::<Vec<_>>())
+        })
+    }
+
+    #[pyo3(signature = (job_id=None, table=None))]
+    pub fn errors(
+        self_: PyRef<'_, Self>,
+        job_id: Option<String>,
+        table: Option<String>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let rows = inner
+                .errors(job_id.as_deref(), table.as_deref())
+                .await
+                .infer_error()?;
+            Ok(rows
+                .into_iter()
+                .map(|e| JobErrorEntry {
+                    job_id: e.job_id,
+                    table: e.table,
+                    column: e.column,
+                    error_type: e.error_type,
+                    error_message: e.error_message,
+                    fragment_id: e.fragment_id,
+                    source_row_id: e.source_row_id,
+                    table_version: e.table_version,
+                    age_seconds: e.age_seconds,
+                })
+                .collect::<Vec<_>>())
+        })
+    }
+
    #[pyo3(signature = (cur_name, new_name, cur_namespace_path=None, new_namespace_path=None))]
    pub fn rename_table(
        self_: PyRef<'_, Self>,
@@ -42,6 +42,12 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
        .write_style("LANCEDB_LOG_STYLE");
    env_logger::init_from_env(env);
    m.add_class::<Connection>()?;
+    m.add_class::<connection::FunctionInfo>()?;
+    m.add_class::<connection::MaterializedViewInfo>()?;
+    m.add_class::<connection::JobInfo>()?;
+    m.add_class::<connection::PlatformJobDescription>()?;
+    m.add_class::<connection::JobHistoryEntry>()?;
+    m.add_class::<connection::JobErrorEntry>()?;
    m.add_class::<Session>()?;
    m.add_class::<Table>()?;
    m.add_class::<PyBlobFile>()?;
@@ -21,7 +21,8 @@ use lancedb::blob::BlobFile;
 use lancedb::index::scalar::FtsIndexBuilder;
 use lancedb::table::{
    AddDataMode, ColumnAlteration, Duration, FieldMetadataUpdate, FtsToken as LanceDbFtsToken,
-    NewColumnTransform, OptimizeAction, OptimizeOptions, Ref, Table as LanceDbTable,
+    LoadColumnsRequest, NewColumnTransform, OptimizeAction, OptimizeOptions, Ref,
+    Table as LanceDbTable,
 };
 use lancedb::tokenize as lancedb_tokenize;
 use pyo3::{
@@ -793,8 +794,8 @@ impl Table {
        }

        future_into_py(self_.py(), async move {
-            op.execute().await.infer_error()?;
-            Ok(())
+            let job_id = op.execute().await.infer_error()?;
+            Ok(job_id)
        })
    }

@@ -1294,6 +1295,83 @@ impl Table {
        })
    }

+    pub fn add_computed_columns(
+        self_: PyRef<'_, Self>,
+        columns: Vec<(String, String)>,
+        expression: String,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            inner
+                .add_computed_columns(&columns, &expression)
+                .await
+                .infer_error()
+        })
+    }
+
+    #[pyo3(signature = (columns, where_clause=None, num_workers=None, max_workers=None, batch_size=None, priority=None))]
+    pub fn refresh_column(
+        self_: PyRef<'_, Self>,
+        columns: Vec<String>,
+        where_clause: Option<String>,
+        num_workers: Option<u32>,
+        max_workers: Option<u32>,
+        batch_size: Option<u32>,
+        priority: Option<String>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        future_into_py(self_.py(), async move {
+            inner
+                .refresh_column(
+                    &columns,
+                    where_clause,
+                    num_workers,
+                    max_workers,
+                    batch_size,
+                    priority,
+                )
+                .await
+                .infer_error()
+        })
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    #[pyo3(signature = (source_uris, source_format, target_key, columns, source_key=None, source_storage_options=None, on_missing=None, num_workers=None, max_workers=None, batch_size=None, commit_granularity=None, priority=None))]
+    pub fn load_columns(
+        self_: PyRef<'_, Self>,
+        source_uris: Vec<String>,
+        source_format: String,
+        target_key: String,
+        columns: Vec<(String, Option<String>)>,
+        source_key: Option<String>,
+        source_storage_options: Option<std::collections::HashMap<String, String>>,
+        on_missing: Option<String>,
+        num_workers: Option<u32>,
+        max_workers: Option<u32>,
+        batch_size: Option<u32>,
+        commit_granularity: Option<u32>,
+        priority: Option<String>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        let request = LoadColumnsRequest {
+            source_uris,
+            source_format,
+            source_storage_options,
+            target_key,
+            source_key,
+            columns,
+            on_missing,
+            num_workers,
+            max_workers,
+            batch_size,
+            commit_granularity,
+            priority,
+        };
+        future_into_py(self_.py(), async move {
+            inner.load_columns(request).await.infer_error()
+        })
+    }
+
    pub fn add_columns(
        self_: PyRef<'_, Self>,
        definitions: Vec<(String, String)>,
@@ -118,8 +118,12 @@ async fn create_empty_table(db: &Connection) -> Result<LanceDbTable> {

 async fn create_index(table: &LanceDbTable) -> Result<()> {
    // --8<-- [start:create_index]
-    table.create_index(&["vector"], Index::Auto).execute().await
+    table
+        .create_index(&["vector"], Index::Auto)
+        .execute()
+        .await?;
    // --8<-- [end:create_index]
+    Ok(())
 }

 async fn search(table: &LanceDbTable) -> Result<Vec<RecordBatch>> {
@@ -23,8 +23,10 @@ use crate::connection::create_table::CreateTableBuilder;
 use crate::data::scannable::Scannable;
 use crate::database::listing::ListingDatabase;
 use crate::database::{
-    CloneTableRequest, Database, DatabaseOptions, OpenTableRequest, ReadConsistency,
-    TableNamesRequest,
+    CloneTableRequest, CreateFunctionRequest, CreateMaterializedViewRequest, Database,
+    DatabaseOptions, FunctionInfo, JobErrorInfo, JobHistoryInfo, JobInfo, MaterializedViewInfo,
+    MvRefreshPlan, OpenTableRequest, PlatformJobDescription, ReadConsistency,
+    RefreshMaterializedViewRequest, TableLineageRequest, TableNamesRequest,
 };
 use crate::embeddings::{EmbeddingRegistry, MemoryRegistry};
 use crate::error::{Error, Result};
@@ -488,6 +490,140 @@ impl Connection {
        )
    }

+    // -- Derived compute: functions, materialized views, jobs -------------
+    // Server-backed features (LanceDB Enterprise / Cloud); local
+    // databases return NotSupported for now.
+
+    /// Register a UDF (CREATE FUNCTION).
+    pub async fn create_function(&self, request: CreateFunctionRequest) -> Result<()> {
+        self.internal.create_function(request).await
+    }
+
+    /// List registered functions (SHOW FUNCTIONS).
+    pub async fn list_functions(&self) -> Result<Vec<FunctionInfo>> {
+        self.internal.list_functions().await
+    }
+
+    /// Drop a registered function (DROP FUNCTION).
+    pub async fn drop_function(&self, name: &str) -> Result<()> {
+        self.internal.drop_function(name).await
+    }
+
+    /// Create a materialized view (CREATE MATERIALIZED VIEW). Returns
+    /// the initial-population job id, absent when `with_no_data`.
+    pub async fn create_materialized_view(
+        &self,
+        request: CreateMaterializedViewRequest,
+    ) -> Result<Option<String>> {
+        self.internal.create_materialized_view(request).await
+    }
+
+    /// Refresh a materialized view; returns the refresh job id.
+    pub async fn refresh_materialized_view(
+        &self,
+        request: RefreshMaterializedViewRequest,
+    ) -> Result<String> {
+        self.internal.refresh_materialized_view(request).await
+    }
+
+    /// Derived-compute lineage of a table/view (or column), as server-defined
+    /// JSON. Read-only.
+    pub async fn table_lineage(&self, request: TableLineageRequest) -> Result<String> {
+        self.internal.table_lineage(request).await
+    }
+
+    /// Plan a materialized-view refresh without submitting work
+    /// (EXPLAIN REFRESH).
+    pub async fn explain_refresh_materialized_view(
+        &self,
+        name: &str,
+        full: bool,
+        src_version: Option<u64>,
+    ) -> Result<MvRefreshPlan> {
+        self.internal
+            .explain_refresh_materialized_view(name, full, src_version)
+            .await
+    }
+
+    /// Update a materialized view's options (ALTER MATERIALIZED VIEW).
+    pub async fn alter_materialized_view(&self, name: &str, auto_refresh: bool) -> Result<()> {
+        self.internal
+            .alter_materialized_view(name, auto_refresh)
+            .await
+    }
+
+    /// Drop a materialized view definition (DROP MATERIALIZED VIEW).
+    pub async fn drop_materialized_view(&self, name: &str) -> Result<()> {
+        self.internal.drop_materialized_view(name).await
+    }
+
+    /// List registered materialized view definitions.
+    pub async fn list_materialized_views(&self) -> Result<Vec<MaterializedViewInfo>> {
+        self.internal.list_materialized_views().await
+    }
+
+    /// List inflight server-side jobs across the database's tables.
+    pub async fn list_jobs(&self) -> Result<Vec<JobInfo>> {
+        self.internal.list_jobs().await
+    }
+
+    /// Cancel an inflight server-side job by id. Returns true if a
+    /// matching inflight job was flagged for cancellation.
+    pub async fn cancel_job(&self, job_id: &str) -> Result<bool> {
+        self.internal.cancel_job(job_id).await
+    }
+
+    /// Describe a platform job (`POST /v1/jobs/describe`): registry-backed
+    /// lifecycle state plus the owner-written status payload. `None` when the
+    /// registry has no such job.
+    pub async fn describe_platform_job(
+        &self,
+        platform_job_id: &str,
+    ) -> Result<Option<PlatformJobDescription>> {
+        self.internal.describe_platform_job(platform_job_id).await
+    }
+
+    /// Resolve a submission (manifest) job id to its platform job id. `None`
+    /// until the job has registered (dispatch is async).
+    pub async fn resolve_platform_job_id(
+        &self,
+        manifest_job_id: &str,
+        table_hint: Option<&str>,
+    ) -> Result<Option<String>> {
+        self.internal
+            .resolve_platform_job_id(manifest_job_id, table_hint)
+            .await
+    }
+
+    /// Cancel a platform job. Idempotent on already-terminal jobs.
+    pub async fn cancel_platform_job(&self, platform_job_id: &str) -> Result<()> {
+        self.internal.cancel_platform_job(platform_job_id).await
+    }
+
+    /// Look up a single server-side job by id -- the `wait()`/status poll path.
+    /// `table_hint` (the job's table) enables an O(1) server-side lookup; `None`
+    /// scans the database's active jobs. A `None` result means unknown / not
+    /// active.
+    pub async fn get_job(&self, job_id: &str, table_hint: Option<&str>) -> Result<Option<JobInfo>> {
+        self.internal.get_job(job_id, table_hint).await
+    }
+
+    /// Durable job history (SHOW JOB HISTORY) across the database's tables.
+    /// Pass `job_id` to narrow to a single job.
+    pub async fn job_history(&self, job_id: Option<&str>) -> Result<Vec<JobHistoryInfo>> {
+        self.internal.job_history(job_id).await
+    }
+
+    /// Per-row UDF errors (SHOW ERRORS) across the database's tables, optionally
+    /// filtered by `job_id` and/or `table`.
+    pub async fn errors(
+        &self,
+        job_id: Option<&str>,
+        table: Option<&str>,
+    ) -> Result<Vec<JobErrorInfo>> {
+        self.internal.errors(job_id, table).await
+    }
+
    /// Rename a table in the database.
    ///
    /// This is only supported in LanceDB Cloud.
@@ -27,7 +27,7 @@ use lance_namespace::models::{
 };

 use crate::data::scannable::Scannable;
-use crate::error::Result;
+use crate::error::{Error, Result};
 use crate::table::{BaseTable, WriteOptions};

 pub mod listing;
@@ -200,6 +200,222 @@ pub enum ReadConsistency {
    Strong,
 }

+/// A request to register a UDF (CREATE FUNCTION).
+///
+/// Functions are first-class database objects, decoupled from any
+/// column; computed columns and materialized views reference them by
+/// name. Server-backed feature (LanceDB Enterprise / Cloud).
+#[derive(Debug, Clone)]
+pub struct CreateFunctionRequest {
+    /// Function name.
+    pub name: String,
+    /// Implementation language (currently "python").
+    pub language: String,
+    /// SQL return type, e.g. `FLOAT`, `FLOAT[1536]`,
+    /// `STRUCT(a FLOAT, b VARCHAR)`, `TABLE(chunk VARCHAR, idx INT)`.
+    pub return_type: String,
+    /// Function body: source text, or base64 cloudpickle bytes when
+    /// `options["body_format"] = "cloudpickle"`.
+    pub body: String,
+    /// Options: input_columns, pip, num_gpus, batch_size, timeout,
+    /// error_policy, docker_image, body_format, ...
+    pub options: HashMap<String, String>,
+}
+
+/// A registered function, as returned by `list_functions`.
+#[derive(Debug, Clone)]
+pub struct FunctionInfo {
+    pub name: String,
+    pub language: String,
+    pub return_type: String,
+    pub description: String,
+}
+
+/// A request to create a materialized view (CREATE MATERIALIZED VIEW).
+#[derive(Debug, Clone)]
+pub struct CreateMaterializedViewRequest {
+    /// View name.
+    pub name: String,
+    /// The view's SELECT statement, e.g.
+    /// `SELECT id, embed(body) AS vec FROM articles WHERE id > 1`.
+    /// Bare columns project through; function-call columns compute via
+    /// registered UDFs (a RETURNS TABLE function makes a row-expanding
+    /// chunker view).
+    pub query: String,
+    /// Refresh automatically when the source table changes.
+    pub auto_refresh: bool,
+    /// Register the definition only; skip the initial population.
+    pub with_no_data: bool,
+    /// Optional source column to partition the view's table function on. If the
+    /// column has an IVF vector index the server partitions by its clusters
+    /// (image-dedup style); otherwise it groups by distinct value.
+    pub partition_by: Option<String>,
+}
+
+impl CreateMaterializedViewRequest {
+    pub fn new(name: impl Into<String>, query: impl Into<String>) -> Self {
+        Self {
+            name: name.into(),
+            query: query.into(),
+            auto_refresh: false,
+            with_no_data: false,
+            partition_by: None,
+        }
+    }
+}
+
+/// A request to refresh a materialized view.
+#[derive(Debug, Clone)]
+pub struct RefreshMaterializedViewRequest {
+    /// View name.
+    pub name: String,
+    /// Force a full rebuild (recompute and replace every row) instead of the
+    /// default incremental refresh.
+    pub full: bool,
+    /// Pin the refresh to a source-table version; latest when absent.
+    pub src_version: Option<u64>,
+    /// Initial worker count.
+    pub num_workers: Option<u32>,
+    /// Elastic worker ceiling.
+    pub max_workers: Option<u32>,
+}
+
+/// A request for the derived-compute lineage of a table/view (or one of its
+/// columns). The response is server-defined lineage JSON, returned opaque so
+/// this client need not model the server's lineage schema.
+#[derive(Debug, Clone, Default)]
+pub struct TableLineageRequest {
+    /// Table or view name.
+    pub name: String,
+    /// Column for column-level lineage; whole table/view when absent.
+    pub column: Option<String>,
+    /// "upstream" | "downstream" | "both" (server default when absent).
+    pub direction: Option<String>,
+    /// Column-hops to walk; transitive when absent.
+    pub depth: Option<u32>,
+}
+
+impl RefreshMaterializedViewRequest {
+    pub fn new(name: impl Into<String>) -> Self {
+        Self {
+            name: name.into(),
+            full: false,
+            src_version: None,
+            num_workers: None,
+            max_workers: None,
+        }
+    }
+}
+
+/// A registered materialized view definition, as returned by
+/// `list_materialized_views`.
+#[derive(Debug, Clone)]
+pub struct MaterializedViewInfo {
+    pub name: String,
+    pub source_table: String,
+    /// Source columns projected through.
+    pub projection: Vec<String>,
+    /// `alias=expression` per UDF-computed column.
+    pub udf_columns: Vec<String>,
+    pub filter: Option<String>,
+    pub auto_refresh: bool,
+}
+
+/// A described platform job (`POST /v1/jobs/describe`): the job registry's
+/// lifecycle state plus the owner-written status payload.
+#[derive(Debug, Clone)]
+pub struct PlatformJobDescription {
+    /// The platform (registry) job id -- what describe/cancel accept.
+    pub job_id: String,
+    pub job_type: String,
+    pub job_subtype: String,
+    /// "IN_PROGRESS" | "CANCELLED" | "FAILED" | "DONE".
+    pub job_state: String,
+    pub creation_ms: i64,
+    /// The owner-written status payload -- `units_done` / `units_total` /
+    /// `rows_committed` / `error` when present. Records whose owner has not
+    /// written a payload yet carry the raw status-store URI string instead.
+    pub status: serde_json::Value,
+}
+
+/// A row from `list_jobs`: one inflight server-side job (index build,
+/// compaction, column refresh, view refresh, ...).
+#[derive(Debug, Clone)]
+pub struct JobInfo {
+    pub table: String,
+    pub job_id: String,
+    pub job_type: String,
+    /// Lifecycle state: "running", "cancelling", or "stale".
+    pub state: String,
+    pub column: Option<String>,
+    pub age_seconds: Option<i64>,
+    pub command: Option<String>,
+    pub units_done: Option<i64>,
+    pub units_total: Option<i64>,
+    /// Whether the job's final commit has completed (output visible).
+    pub committed: bool,
+    pub rows_skipped: u64,
+    pub error: Option<String>,
+}
+
+/// A row from `job_history`: one durable, completed/terminal server-side job
+/// record (SHOW JOB HISTORY), read from a table's `_job_history` store. Unlike
+/// `JobInfo` (live, inflight jobs) this carries created/updated/completed
+/// timestamps and the lifecycle event log.
+#[derive(Debug, Clone)]
+pub struct JobHistoryInfo {
+    pub table: String,
+    pub job_id: String,
+    pub job_type: String,
+    pub state: String,
+    pub column: Option<String>,
+    pub created_ms: i64,
+    pub updated_ms: i64,
+    pub completed_ms: Option<i64>,
+    pub rows_processed: Option<i64>,
+    pub rows_skipped: Option<i64>,
+    pub error: Option<String>,
+    /// Newline-joined lifecycle event log, oldest first.
+    pub events: Option<String>,
+}
+
+/// A row from `errors`: one per-row UDF failure recorded by `error_policy=skip`
+/// (SHOW ERRORS).
+#[derive(Debug, Clone)]
+pub struct JobErrorInfo {
+    pub job_id: String,
+    pub table: String,
+    pub column: String,
+    pub error_type: String,
+    pub error_message: String,
+    pub fragment_id: Option<i64>,
+    pub source_row_id: Option<i64>,
+    pub table_version: Option<i64>,
+    pub age_seconds: Option<i64>,
+}
+
+/// The plan a `REFRESH MATERIALIZED VIEW` would execute, as returned by
+/// `explain_refresh_materialized_view` (EXPLAIN REFRESH). No work is run.
+#[derive(Debug, Clone)]
+pub struct MvRefreshPlan {
+    pub table_name: String,
+    /// Whether a refresh would do anything (rebuild or non-empty units).
+    pub has_work: bool,
+    pub source_version: u64,
+    pub last_refreshed_version: Option<u64>,
+    pub full_refresh: bool,
+    /// Source changed non-append-only since the last refresh -> rebuild.
+    pub rebuild: bool,
+    /// Number of row-range work units the refresh would process.
+    pub units_total: u64,
+}
+
+fn not_supported<T>(what: &str) -> Result<T> {
+    Err(Error::NotSupported {
+        message: format!("{} is not supported by this database", what),
+    })
+}
+
 /// The `Database` trait defines the interface for database implementations.
 ///
 /// A database is responsible for managing tables and their metadata.
@@ -245,6 +461,126 @@ pub trait Database:
    ///
    /// See [`CloneTableRequest`] for detailed documentation and examples.
    async fn clone_table(&self, request: CloneTableRequest) -> Result<Arc<dyn BaseTable>>;
+
+    // -- Derived compute: functions, materialized views, jobs -------------
+    //
+    // Server-backed features (LanceDB Enterprise / Cloud). The defaults
+    // return NotSupported; the remote database overrides them. Local
+    // single-node implementations are planned.
+
+    /// Register a UDF (CREATE FUNCTION).
+    async fn create_function(&self, _request: CreateFunctionRequest) -> Result<()> {
+        not_supported("create_function")
+    }
+    /// List registered functions (SHOW FUNCTIONS).
+    async fn list_functions(&self) -> Result<Vec<FunctionInfo>> {
+        not_supported("list_functions")
+    }
+    /// Drop a registered function (DROP FUNCTION).
+    async fn drop_function(&self, _name: &str) -> Result<()> {
+        not_supported("drop_function")
+    }
+    /// Create a materialized view (CREATE MATERIALIZED VIEW). Returns
+    /// the initial-population job id, absent when `with_no_data`.
+    async fn create_materialized_view(
+        &self,
+        _request: CreateMaterializedViewRequest,
+    ) -> Result<Option<String>> {
+        not_supported("create_materialized_view")
+    }
+    /// Refresh a materialized view; returns the refresh job id.
+    async fn refresh_materialized_view(
+        &self,
+        _request: RefreshMaterializedViewRequest,
+    ) -> Result<String> {
+        not_supported("refresh_materialized_view")
+    }
+    /// Derived-compute lineage of a table/view (or column), as server-defined
+    /// JSON. Read-only.
+    async fn table_lineage(&self, _request: TableLineageRequest) -> Result<String> {
+        not_supported("table_lineage")
+    }
+    /// Plan a materialized-view refresh without submitting work
+    /// (EXPLAIN REFRESH). `full` plans a full rebuild (incremental
+    /// planning requires stable row IDs on the source).
+    async fn explain_refresh_materialized_view(
+        &self,
+        _name: &str,
+        _full: bool,
+        _src_version: Option<u64>,
+    ) -> Result<MvRefreshPlan> {
+        not_supported("explain_refresh_materialized_view")
+    }
+    /// Update a materialized view's options (ALTER MATERIALIZED VIEW).
+    async fn alter_materialized_view(&self, _name: &str, _auto_refresh: bool) -> Result<()> {
+        not_supported("alter_materialized_view")
+    }
+    /// Drop a materialized view definition (DROP MATERIALIZED VIEW).
+    async fn drop_materialized_view(&self, _name: &str) -> Result<()> {
+        not_supported("drop_materialized_view")
+    }
+    /// List registered materialized view definitions.
+    async fn list_materialized_views(&self) -> Result<Vec<MaterializedViewInfo>> {
+        not_supported("list_materialized_views")
+    }
+    /// List inflight server-side jobs across the database's tables.
+    async fn list_jobs(&self) -> Result<Vec<JobInfo>> {
+        not_supported("list_jobs")
+    }
+
+    /// Describe a platform job (`POST /v1/jobs/describe`): registry-backed
+    /// lifecycle state plus the owner-written status payload. `None` when the
+    /// registry has no such job.
+    async fn describe_platform_job(
+        &self,
+        _platform_job_id: &str,
+    ) -> Result<Option<PlatformJobDescription>> {
+        not_supported("describe_platform_job")
+    }
+
+    /// Resolve a submission (manifest) job id to its platform job id via the
+    /// registry's manifest-id filter (`POST /v1/jobs/list`). `None` until the
+    /// job has registered (dispatch is async).
+    async fn resolve_platform_job_id(
+        &self,
+        _manifest_job_id: &str,
+        _table_hint: Option<&str>,
+    ) -> Result<Option<String>> {
+        not_supported("resolve_platform_job_id")
+    }
+
+    /// Cancel a platform job (`POST /v1/jobs/cancel`). Idempotent: cancelling
+    /// an already-terminal job is a no-op success.
+    async fn cancel_platform_job(&self, _platform_job_id: &str) -> Result<()> {
+        not_supported("cancel_platform_job")
+    }
+    /// Cancel an inflight server-side job by id. Returns true if a
+    /// matching inflight job was found and flagged for cancellation,
+    /// false if none was inflight (best-effort, like SQL `CANCEL JOB`).
+    async fn cancel_job(&self, _job_id: &str) -> Result<bool> {
+        not_supported("cancel_job")
+    }
+    /// Point-access for a single job by id -- the `wait()`/status poll path.
+    /// `table_hint` (the job's table, which `wait()` callers know) enables an
+    /// O(1) server-side lookup. `None` if the job is unknown or not active.
+    async fn get_job(&self, _job_id: &str, _table_hint: Option<&str>) -> Result<Option<JobInfo>> {
+        not_supported("get_job")
+    }
+    /// Durable job history (SHOW JOB HISTORY) across the database's tables,
+    /// optionally narrowed to a single `job_id`.
+    async fn job_history(&self, _job_id: Option<&str>) -> Result<Vec<JobHistoryInfo>> {
+        not_supported("job_history")
+    }
+    /// Per-row UDF errors (SHOW ERRORS) recorded by `error_policy=skip` across
+    /// the database's tables, optionally filtered by `job_id` and/or `table`.
+    async fn errors(
+        &self,
+        _job_id: Option<&str>,
+        _table: Option<&str>,
+    ) -> Result<Vec<JobErrorInfo>> {
+        not_supported("errors")
+    }
+
    /// Open a table in the database
    async fn open_table(&self, request: OpenTableRequest) -> Result<Arc<dyn BaseTable>>;
    /// Rename a table in the database
@@ -761,8 +761,8 @@ mod tests {
        verify_splitter(splitter, test_data(), 50, &[11, 8, 9], false).await;
    }

-    #[tokio::test]
-    async fn test_hash_split() {
+    async fn collect_hash_split() -> RecordBatch {
+        let total_rows = 50;
        let data = lance_datagen::gen_batch()
            .with_seed(Seed::from(42))
            .col(
@@ -783,7 +783,7 @@ mod tests {
        );

        let split_batches = splitter
-            .apply(data, 10)
+            .apply(data, total_rows)
            .await
            .unwrap()
            .try_collect::<Vec<_>>()
@@ -791,20 +791,35 @@ mod tests {
            .unwrap();

        let schema = split_batches[0].schema();
-        let split_batch = concat_batches(&schema, &split_batches).unwrap();
+        concat_batches(&schema, &split_batches).unwrap()
+    }

-        // These assertions are all based on fixed seed in data generation but they match
-        // up roughly to what we expect (25% discarded, 25% in split 0, 50% in split 1)
+    #[tokio::test]
+    async fn test_hash_split() {
+        let total_rows = 50;
+        let split_batch = collect_hash_split().await;
+        let split_batch_again = collect_hash_split().await;

-        // 8 rows (16%) are discarded because discard_weight is 1
-        assert_eq!(split_batch.num_rows(), 42);
+        assert_eq!(split_batch.num_rows(), split_batch_again.num_rows());
+        assert_eq!(split_batch.num_columns(), split_batch_again.num_columns());
+        for (left, right) in split_batch
+            .columns()
+            .iter()
+            .zip(split_batch_again.columns().iter())
+        {
+            assert_eq!(left, right);
+        }
+
+        assert!(split_batch.num_rows() > 0);
+        assert!(split_batch.num_rows() < total_rows);
        assert_eq!(split_batch.num_columns(), 2);

        let split_ids = split_batch.column(1).as_primitive::<UInt64Type>().values();
        let num_in_split_0 = split_ids.iter().filter(|v| **v == 0).count();
        let num_in_split_1 = split_ids.iter().filter(|v| **v == 1).count();

-        assert_eq!(num_in_split_0, 12); // 24%
-        assert_eq!(num_in_split_1, 30); // 60%
+        assert_eq!(num_in_split_0 + num_in_split_1, split_batch.num_rows());
+        assert!(num_in_split_0 > 0);
+        assert!(num_in_split_1 > num_in_split_0);
    }
 }
@@ -283,7 +283,10 @@ impl IndexBuilder {
        self
    }

-    pub async fn execute(self) -> Result<()> {
+    /// Returns the server-minted job id when the index build was deferred to
+    /// a background job (remote tables only); `None` when the build completed
+    /// synchronously within this call.
+    pub async fn execute(self) -> Result<Option<String>> {
        self.parent.clone().create_index(self).await
    }
 }
@@ -19,8 +19,10 @@ use lance_namespace::models::{

 use crate::Error;
 use crate::database::{
-    CloneTableRequest, CreateTableMode, CreateTableRequest, Database, DatabaseOptions,
-    OpenTableRequest, ReadConsistency, TableNamesRequest,
+    CloneTableRequest, CreateFunctionRequest, CreateMaterializedViewRequest, CreateTableMode,
+    CreateTableRequest, Database, DatabaseOptions, FunctionInfo, JobErrorInfo, JobHistoryInfo,
+    JobInfo, MaterializedViewInfo, MvRefreshPlan, OpenTableRequest, PlatformJobDescription,
+    ReadConsistency, RefreshMaterializedViewRequest, TableLineageRequest, TableNamesRequest,
 };
 use crate::error::Result;
 use crate::remote::util::stream_as_body;
@@ -33,6 +35,210 @@ use super::client::{
 use super::table::RemoteTable;
 use super::util::parse_server_version;

+// Wire types for the derived-compute routes (functions, materialized
+// views, jobs). Field shapes mirror the server's REST contract.
+#[derive(serde::Serialize)]
+struct RemoteCreateFunctionRequest {
+    language: String,
+    return_type: String,
+    body: String,
+    options: std::collections::HashMap<String, String>,
+}
+
+#[derive(serde::Deserialize)]
+struct RemoteFunctionEntry {
+    name: String,
+    language: String,
+    return_type: String,
+    #[serde(default)]
+    description: String,
+}
+
+#[derive(serde::Deserialize)]
+struct RemoteListFunctionsResponse {
+    functions: Vec<RemoteFunctionEntry>,
+}
+
+#[derive(serde::Serialize)]
+struct RemoteCreateMaterializedViewRequest {
+    query: String,
+    auto_refresh: bool,
+    with_no_data: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    partition_by: Option<String>,
+}
+
+#[derive(serde::Deserialize)]
+struct RemoteCreateMaterializedViewResponse {
+    #[serde(default)]
+    job_id: Option<String>,
+}
+
+#[derive(serde::Serialize)]
+struct RemoteRefreshMaterializedViewRequest {
+    #[serde(skip_serializing_if = "std::ops::Not::not")]
+    full: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    src_version: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    num_workers: Option<u32>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    max_workers: Option<u32>,
+}
+
+#[derive(serde::Deserialize)]
+struct RemoteRefreshMaterializedViewResponse {
+    job_id: String,
+}
+
+#[derive(serde::Serialize)]
+struct RemoteExplainRefreshRequest {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    full: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    src_version: Option<u64>,
+}
+
+#[derive(serde::Deserialize)]
+struct RemoteExplainRefreshResponse {
+    table_name: String,
+    has_work: bool,
+    source_version: u64,
+    last_refreshed_version: Option<u64>,
+    full_refresh: bool,
+    rebuild: bool,
+    units_total: u64,
+}
+
+#[derive(serde::Serialize)]
+struct RemoteAlterMaterializedViewRequest {
+    auto_refresh: bool,
+}
+
+#[derive(serde::Deserialize)]
+struct RemoteMaterializedViewEntry {
+    name: String,
+    source_table: String,
+    #[serde(default)]
+    projection: Vec<String>,
+    #[serde(default)]
+    udf_columns: Vec<String>,
+    #[serde(default)]
+    filter: Option<String>,
+    #[serde(default)]
+    auto_refresh: bool,
+}
+
+#[derive(serde::Deserialize)]
+struct RemoteListMaterializedViewsResponse {
+    views: Vec<RemoteMaterializedViewEntry>,
+}
+
+#[derive(serde::Deserialize)]
+struct RemoteDescribePlatformJobResponse {
+    job_id: String,
+    job_type: String,
+    #[serde(default)]
+    job_subtype: String,
+    job_state: String,
+    #[serde(default)]
+    creation_ms: i64,
+    #[serde(default)]
+    status: serde_json::Value,
+}
+
+#[derive(serde::Deserialize)]
+struct RemoteListPlatformJobsResponse {
+    #[serde(default)]
+    jobs: Vec<RemotePlatformJobRow>,
+}
+
+#[derive(serde::Deserialize)]
+struct RemotePlatformJobRow {
+    job_id: String,
+    #[serde(default)]
+    table: String,
+    #[serde(default)]
+    job_subtype: String,
+    #[serde(default)]
+    state: String,
+    #[serde(default)]
+    created_at_millis: i64,
+    #[serde(default)]
+    status: serde_json::Value,
+}
+
+/// Platform list-row state -> the client's job vocabulary.
+fn platform_state_to_client(state: &str) -> String {
+    match state {
+        "in_progress" => "running",
+        "done" => "finished",
+        other => other,
+    }
+    .to_string()
+}
+
+/// Describe job_state -> the client's job vocabulary.
+fn describe_state_to_client(state: &str) -> String {
+    match state {
+        "IN_PROGRESS" => "running",
+        "DONE" => "finished",
+        "FAILED" => "failed",
+        "CANCELLED" => "cancelled",
+        other => other,
+    }
+    .to_string()
+}
+
+fn payload_i64(status: &serde_json::Value, key: &str) -> Option<i64> {
+    status.get(key).and_then(serde_json::Value::as_i64)
+}
+
+fn payload_error(status: &serde_json::Value) -> Option<String> {
+    status
+        .get("error")
+        .and_then(serde_json::Value::as_str)
+        .map(str::to_string)
+}
+
+#[derive(serde::Deserialize)]
+struct RemoteErrorEntry {
+    job_id: String,
+    table: String,
+    column: String,
+    error_type: String,
+    error_message: String,
+    #[serde(default)]
+    fragment_id: Option<i64>,
+    #[serde(default)]
+    source_row_id: Option<i64>,
+    #[serde(default)]
+    table_version: Option<i64>,
+    #[serde(default)]
+    age_seconds: Option<i64>,
+}
+
+#[derive(serde::Deserialize)]
+struct RemoteErrorsResponse {
+    errors: Vec<RemoteErrorEntry>,
+}
+
+impl From<RemoteErrorEntry> for JobErrorInfo {
+    fn from(e: RemoteErrorEntry) -> Self {
+        JobErrorInfo {
+            job_id: e.job_id,
+            table: e.table,
+            column: e.column,
+            error_type: e.error_type,
+            error_message: e.error_message,
+            fragment_id: e.fragment_id,
+            source_row_id: e.source_row_id,
+            table_version: e.table_version,
+            age_seconds: e.age_seconds,
+        }
+    }
+}
+
 // Request structure for the remote clone table API
 #[derive(serde::Serialize)]
 struct RemoteCloneTableRequest {
@@ -641,6 +847,426 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
        Ok(table)
    }

+    async fn create_function(&self, request: CreateFunctionRequest) -> Result<()> {
+        let body = RemoteCreateFunctionRequest {
+            language: request.language,
+            return_type: request.return_type,
+            body: request.body,
+            options: request.options,
+        };
+        let req = self
+            .client
+            .post(&format!("/v1/function/{}/create", request.name))
+            .json(&body);
+        let (request_id, rsp) = self.client.send(req).await?;
+        self.client.check_response(&request_id, rsp).await?;
+        Ok(())
+    }
+
+    async fn list_functions(&self) -> Result<Vec<FunctionInfo>> {
+        let req = self.client.get("/v1/function/list");
+        let (request_id, rsp) = self.client.send(req).await?;
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body: RemoteListFunctionsResponse = rsp.json().await.err_to_http(request_id)?;
+        Ok(body
+            .functions
+            .into_iter()
+            .map(|f| FunctionInfo {
+                name: f.name,
+                language: f.language,
+                return_type: f.return_type,
+                description: f.description,
+            })
+            .collect())
+    }
+
+    async fn drop_function(&self, name: &str) -> Result<()> {
+        let req = self.client.post(&format!("/v1/function/{}/drop", name));
+        let (request_id, rsp) = self.client.send(req).await?;
+        self.client.check_response(&request_id, rsp).await?;
+        Ok(())
+    }
+
+    async fn create_materialized_view(
+        &self,
+        request: CreateMaterializedViewRequest,
+    ) -> Result<Option<String>> {
+        let body = RemoteCreateMaterializedViewRequest {
+            query: request.query,
+            auto_refresh: request.auto_refresh,
+            with_no_data: request.with_no_data,
+            partition_by: request.partition_by,
+        };
+        let req = self
+            .client
+            .post(&format!("/v1/materialized_view/{}/create", request.name))
+            .json(&body);
+        let (request_id, rsp) = self.client.send(req).await?;
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body: RemoteCreateMaterializedViewResponse =
+            rsp.json().await.err_to_http(request_id)?;
+        Ok(body.job_id)
+    }
+
+    async fn refresh_materialized_view(
+        &self,
+        request: RefreshMaterializedViewRequest,
+    ) -> Result<String> {
+        let body = RemoteRefreshMaterializedViewRequest {
+            full: request.full,
+            src_version: request.src_version,
+            num_workers: request.num_workers,
+            max_workers: request.max_workers,
+        };
+        let req = self
+            .client
+            .post(&format!("/v1/materialized_view/{}/refresh", request.name))
+            .json(&body);
+        let (request_id, rsp) = self.client.send(req).await?;
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body: RemoteRefreshMaterializedViewResponse =
+            rsp.json().await.err_to_http(request_id)?;
+        Ok(body.job_id)
+    }
+
+    async fn table_lineage(&self, request: TableLineageRequest) -> Result<String> {
+        let mut req = self
+            .client
+            .get(&format!("/v1/table/{}/lineage", request.name));
+        if let Some(column) = &request.column {
+            req = req.query(&[("column", column)]);
+        }
+        if let Some(direction) = &request.direction {
+            req = req.query(&[("direction", direction)]);
+        }
+        if let Some(depth) = request.depth {
+            req = req.query(&[("depth", depth.to_string())]);
+        }
+        let (request_id, rsp) = self.client.send(req).await?;
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        // Server-defined lineage JSON, returned opaque (the client does not
+        // model the lineage schema; the Python layer deserializes it).
+        rsp.text().await.err_to_http(request_id)
+    }
+
+    async fn explain_refresh_materialized_view(
+        &self,
+        name: &str,
+        full: bool,
+        src_version: Option<u64>,
+    ) -> Result<MvRefreshPlan> {
+        let body = RemoteExplainRefreshRequest {
+            full: Some(full),
+            src_version,
+        };
+        let req = self
+            .client
+            .post(&format!("/v1/materialized_view/{}/explain_refresh", name))
+            .json(&body);
+        let (request_id, rsp) = self.client.send(req).await?;
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body: RemoteExplainRefreshResponse = rsp.json().await.err_to_http(request_id)?;
+        Ok(MvRefreshPlan {
+            table_name: body.table_name,
+            has_work: body.has_work,
+            source_version: body.source_version,
+            last_refreshed_version: body.last_refreshed_version,
+            full_refresh: body.full_refresh,
+            rebuild: body.rebuild,
+            units_total: body.units_total,
+        })
+    }
+
+    async fn alter_materialized_view(&self, name: &str, auto_refresh: bool) -> Result<()> {
+        let req = self
+            .client
+            .post(&format!("/v1/materialized_view/{}/alter", name))
+            .json(&RemoteAlterMaterializedViewRequest { auto_refresh });
+        let (request_id, rsp) = self.client.send(req).await?;
+        self.client.check_response(&request_id, rsp).await?;
+        Ok(())
+    }
+
+    async fn drop_materialized_view(&self, name: &str) -> Result<()> {
+        let req = self
+            .client
+            .post(&format!("/v1/materialized_view/{}/drop", name));
+        let (request_id, rsp) = self.client.send(req).await?;
+        self.client.check_response(&request_id, rsp).await?;
+        Ok(())
+    }
+
+    async fn list_materialized_views(&self) -> Result<Vec<MaterializedViewInfo>> {
+        let req = self.client.get("/v1/materialized_view/list");
+        let (request_id, rsp) = self.client.send(req).await?;
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body: RemoteListMaterializedViewsResponse = rsp.json().await.err_to_http(request_id)?;
+        Ok(body
+            .views
+            .into_iter()
+            .map(|v| MaterializedViewInfo {
+                name: v.name,
+                source_table: v.source_table,
+                projection: v.projection,
+                udf_columns: v.udf_columns,
+                filter: v.filter,
+                auto_refresh: v.auto_refresh,
+            })
+            .collect())
+    }
+
+    async fn list_jobs(&self) -> Result<Vec<JobInfo>> {
+        let req = self
+            .client
+            .post("/v1/jobs/list")
+            .json(&serde_json::json!({ "include_status": true }));
+        let (request_id, rsp) = self.client.send(req).await?;
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body: RemoteListPlatformJobsResponse = rsp.json().await.err_to_http(request_id)?;
+        let now_ms = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map(|d| d.as_millis() as i64)
+            .unwrap_or(0);
+        Ok(body
+            .jobs
+            .into_iter()
+            .map(|row| JobInfo {
+                table: row.table,
+                job_id: row.job_id,
+                // The platform job_type is always "indexer"; the subtype
+                // (udf / mv_refresh / compaction / ...) is the useful label.
+                job_type: row.job_subtype,
+                state: platform_state_to_client(&row.state),
+                column: None,
+                age_seconds: (row.created_at_millis > 0)
+                    .then(|| (now_ms - row.created_at_millis) / 1000),
+                command: None,
+                units_done: payload_i64(&row.status, "units_done"),
+                units_total: payload_i64(&row.status, "units_total"),
+                committed: row.state == "done",
+                rows_skipped: payload_i64(&row.status, "rows_skipped").unwrap_or(0) as u64,
+                error: payload_error(&row.status),
+            })
+            .collect())
+    }
+
+    async fn get_job(&self, job_id: &str, table: Option<&str>) -> Result<Option<JobInfo>> {
+        // A point snapshot from the platform API: resolve the submission id,
+        // then describe. The snapshot keeps the caller's id.
+        let Some(platform_id) = self.resolve_platform_job_id(job_id, table).await? else {
+            return Ok(None);
+        };
+        let Some(described) = self.describe_platform_job(&platform_id).await? else {
+            return Ok(None);
+        };
+        Ok(Some(JobInfo {
+            table: table.unwrap_or_default().to_string(),
+            job_id: job_id.to_string(),
+            job_type: described.job_subtype,
+            state: describe_state_to_client(&described.job_state),
+            column: None,
+            age_seconds: None,
+            command: None,
+            units_done: payload_i64(&described.status, "units_done"),
+            units_total: payload_i64(&described.status, "units_total"),
+            committed: described.job_state == "DONE",
+            rows_skipped: payload_i64(&described.status, "rows_skipped").unwrap_or(0) as u64,
+            error: payload_error(&described.status),
+        }))
+    }
+
+    async fn describe_platform_job(
+        &self,
+        platform_job_id: &str,
+    ) -> Result<Option<PlatformJobDescription>> {
+        let req = self
+            .client
+            .post("/v1/jobs/describe")
+            .json(&serde_json::json!({ "job_id": platform_job_id }));
+        let (request_id, rsp) = self.client.send(req).await?;
+        if rsp.status().as_u16() == 404 {
+            return Ok(None);
+        }
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body: RemoteDescribePlatformJobResponse = rsp.json().await.err_to_http(request_id)?;
+        Ok(Some(PlatformJobDescription {
+            job_id: body.job_id,
+            job_type: body.job_type,
+            job_subtype: body.job_subtype,
+            job_state: body.job_state,
+            creation_ms: body.creation_ms,
+            status: body.status,
+        }))
+    }
+
+    async fn resolve_platform_job_id(
+        &self,
+        manifest_job_id: &str,
+        table_hint: Option<&str>,
+    ) -> Result<Option<String>> {
+        let req = self.client.post("/v1/jobs/list").json(&serde_json::json!({
+            "manifest_job_id": manifest_job_id,
+            "table_name": table_hint,
+            "job_type": "indexer",
+        }));
+        let (request_id, rsp) = self.client.send(req).await?;
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body: RemoteListPlatformJobsResponse = rsp.json().await.err_to_http(request_id)?;
+        Ok(body.jobs.into_iter().next().map(|row| row.job_id))
+    }
+
+    async fn cancel_platform_job(&self, platform_job_id: &str) -> Result<()> {
+        let req = self
+            .client
+            .post("/v1/jobs/cancel")
+            .json(&serde_json::json!({ "job_id": platform_job_id }));
+        let (request_id, rsp) = self.client.send(req).await?;
+        self.client.check_response(&request_id, rsp).await?;
+        Ok(())
+    }
+
+    async fn cancel_job(&self, job_id: &str) -> Result<bool> {
+        // Resolve the submission id and cancel through the platform API.
+        // False when no matching job has registered (the legacy best-effort
+        // contract).
+        let Some(platform_id) = self.resolve_platform_job_id(job_id, None).await? else {
+            return Ok(false);
+        };
+        self.cancel_platform_job(&platform_id).await?;
+        Ok(true)
+    }
+
+    async fn job_history(&self, job_id: Option<&str>) -> Result<Vec<JobHistoryInfo>> {
+        // One job: describe (identity) plus query_events (timeline). No id:
+        // a registry listing, timeline-free -- pass an id for the event log.
+        let Some(caller_id) = job_id else {
+            let rows = self.list_jobs().await?;
+            let now_ms = std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .map(|d| d.as_millis() as i64)
+                .unwrap_or(0);
+            return Ok(rows
+                .into_iter()
+                .map(|j| JobHistoryInfo {
+                    table: j.table,
+                    job_id: j.job_id,
+                    job_type: j.job_type,
+                    state: j.state,
+                    column: j.column,
+                    created_ms: j.age_seconds.map(|a| now_ms - a * 1000).unwrap_or_default(),
+                    updated_ms: 0,
+                    completed_ms: None,
+                    rows_processed: None,
+                    rows_skipped: j.rows_skipped.try_into().ok(),
+                    error: j.error,
+                    events: None,
+                })
+                .collect());
+        };
+        // Accept either a platform id or a submission id.
+        let platform_id = match self.describe_platform_job(caller_id).await? {
+            Some(_) => caller_id.to_string(),
+            None => match self.resolve_platform_job_id(caller_id, None).await? {
+                Some(id) => id,
+                None => return Ok(Vec::new()),
+            },
+        };
+        let Some(described) = self.describe_platform_job(&platform_id).await? else {
+            return Ok(Vec::new());
+        };
+        let req = self
+            .client
+            .post("/v1/jobs/query_events")
+            .json(&serde_json::json!({ "job_id": platform_id }));
+        let (request_id, rsp) = self.client.send(req).await?;
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body = rsp.bytes().await.err_to_http(request_id.clone())?;
+        let reader = arrow_ipc::reader::StreamReader::try_new(std::io::Cursor::new(body), None)
+            .map_err(|e| Error::Http {
+                source: format!("failed to read job-events IPC stream: {e}").into(),
+                request_id: request_id.clone(),
+                status_code: None,
+            })?;
+
+        let mut created_ms = i64::MAX;
+        let mut updated_ms = 0i64;
+        let mut completed_ms = None;
+        let mut last_error = None;
+        let mut events = Vec::new();
+        for batch in reader {
+            let batch = batch.map_err(|e| Error::Http {
+                source: format!("failed to decode job-events batch: {e}").into(),
+                request_id: request_id.clone(),
+                status_code: None,
+            })?;
+            let states = batch
+                .column_by_name("state")
+                .and_then(|c| c.as_any().downcast_ref::<arrow_array::StringArray>());
+            let times = batch
+                .column_by_name("updated_at_millis")
+                .and_then(|c| c.as_any().downcast_ref::<arrow_array::Int64Array>());
+            let payloads = batch
+                .column_by_name("payload")
+                .and_then(|c| c.as_any().downcast_ref::<arrow_array::StringArray>());
+            let (Some(states), Some(times)) = (states, times) else {
+                continue;
+            };
+            for i in 0..batch.num_rows() {
+                let state = states.value(i);
+                let ts = times.value(i);
+                created_ms = created_ms.min(ts);
+                updated_ms = updated_ms.max(ts);
+                if matches!(state, "succeeded" | "failed" | "timed_out" | "canceled") {
+                    completed_ms = Some(ts);
+                }
+                if let Some(payloads) = payloads {
+                    if !arrow_array::Array::is_null(payloads, i) {
+                        if let Ok(payload) =
+                            serde_json::from_str::<serde_json::Value>(payloads.value(i))
+                        {
+                            if let Some(e) = payload_error(&payload) {
+                                last_error = Some(e);
+                            }
+                        }
+                    }
+                }
+                events.push(format!("{state} {ts}"));
+            }
+        }
+        Ok(vec![JobHistoryInfo {
+            table: String::new(),
+            job_id: caller_id.to_string(),
+            job_type: described.job_subtype,
+            state: describe_state_to_client(&described.job_state),
+            column: None,
+            created_ms: if created_ms == i64::MAX {
+                described.creation_ms
+            } else {
+                created_ms
+            },
+            updated_ms,
+            completed_ms,
+            rows_processed: payload_i64(&described.status, "rows_committed"),
+            rows_skipped: payload_i64(&described.status, "rows_skipped"),
+            error: last_error.or_else(|| payload_error(&described.status)),
+            events: (!events.is_empty()).then(|| events.join("\n")),
+        }])
+    }
+
+    async fn errors(&self, job_id: Option<&str>, table: Option<&str>) -> Result<Vec<JobErrorInfo>> {
+        let mut req = self.client.get("/v1/errors");
+        if let Some(j) = job_id {
+            req = req.query(&[("job", j)]);
+        }
+        if let Some(t) = table {
+            req = req.query(&[("table", t)]);
+        }
+        let (request_id, rsp) = self.client.send(req).await?;
+        let rsp = self.client.check_response(&request_id, rsp).await?;
+        let body: RemoteErrorsResponse = rsp.json().await.err_to_http(request_id)?;
+        Ok(body.errors.into_iter().map(JobErrorInfo::from).collect())
+    }
+
    async fn open_table(&self, request: OpenTableRequest) -> Result<Arc<dyn BaseTable>> {
        let identifier = build_table_identifier(
            &request.name,
@@ -1580,6 +2206,227 @@ mod tests {
        }
    }

+    #[tokio::test]
+    async fn test_derived_compute_routes() {
+        // create_function
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.method(), &reqwest::Method::POST);
+            assert_eq!(request.url().path(), "/v1/function/embed/create");
+            let body: serde_json::Value =
+                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
+            assert_eq!(body["language"], "python");
+            assert_eq!(body["return_type"], "FLOAT[4]");
+            assert_eq!(body["body"], "def embed(x): ...");
+            assert_eq!(body["options"]["pip"], "torch");
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"name":"embed","status":"OK"}"#)
+                .unwrap()
+        });
+        conn.create_function(crate::database::CreateFunctionRequest {
+            name: "embed".into(),
+            language: "python".into(),
+            return_type: "FLOAT[4]".into(),
+            body: "def embed(x): ...".into(),
+            options: [("pip".to_string(), "torch".to_string())].into(),
+        })
+        .await
+        .unwrap();
+
+        // list_functions
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.method(), &reqwest::Method::GET);
+            assert_eq!(request.url().path(), "/v1/function/list");
+            http::Response::builder()
+                .status(200)
+                .body(
+                    r#"{"functions":[{"name":"embed","language":"python","return_type":"Float32","description":""}]}"#,
+                )
+                .unwrap()
+        });
+        let functions = conn.list_functions().await.unwrap();
+        assert_eq!(functions.len(), 1);
+        assert_eq!(functions[0].name, "embed");
+
+        // drop_function
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.method(), &reqwest::Method::POST);
+            assert_eq!(request.url().path(), "/v1/function/embed/drop");
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"name":"embed","status":"OK"}"#)
+                .unwrap()
+        });
+        conn.drop_function("embed").await.unwrap();
+
+        // create_materialized_view
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.method(), &reqwest::Method::POST);
+            assert_eq!(request.url().path(), "/v1/materialized_view/mv1/create");
+            let body: serde_json::Value =
+                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
+            assert_eq!(body["query"], "SELECT id, embed(body) AS vec FROM docs");
+            assert_eq!(body["auto_refresh"], true);
+            assert_eq!(body["with_no_data"], false);
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"name":"mv1","job_id":"j-1"}"#)
+                .unwrap()
+        });
+        let mut request = crate::database::CreateMaterializedViewRequest::new(
+            "mv1",
+            "SELECT id, embed(body) AS vec FROM docs",
+        );
+        request.auto_refresh = true;
+        let job_id = conn.create_materialized_view(request).await.unwrap();
+        assert_eq!(job_id.as_deref(), Some("j-1"));
+
+        // refresh_materialized_view
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.method(), &reqwest::Method::POST);
+            assert_eq!(request.url().path(), "/v1/materialized_view/mv1/refresh");
+            let body: serde_json::Value =
+                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
+            assert_eq!(body["num_workers"], 2);
+            assert!(body.get("src_version").is_none());
+            http::Response::builder()
+                .status(202)
+                .body(r#"{"job_id":"j-2"}"#)
+                .unwrap()
+        });
+        let mut request = crate::database::RefreshMaterializedViewRequest::new("mv1");
+        request.num_workers = Some(2);
+        let job_id = conn.refresh_materialized_view(request).await.unwrap();
+        assert_eq!(job_id, "j-2");
+
+        // alter_materialized_view
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.method(), &reqwest::Method::POST);
+            assert_eq!(request.url().path(), "/v1/materialized_view/mv1/alter");
+            let body: serde_json::Value =
+                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
+            assert_eq!(body["auto_refresh"], false);
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"name":"mv1","status":"OK"}"#)
+                .unwrap()
+        });
+        conn.alter_materialized_view("mv1", false).await.unwrap();
+
+        // drop_materialized_view
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.url().path(), "/v1/materialized_view/mv1/drop");
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"name":"mv1","status":"OK"}"#)
+                .unwrap()
+        });
+        conn.drop_materialized_view("mv1").await.unwrap();
+
+        // list_materialized_views
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.method(), &reqwest::Method::GET);
+            assert_eq!(request.url().path(), "/v1/materialized_view/list");
+            http::Response::builder()
+                .status(200)
+                .body(
+                    r#"{"views":[{"name":"mv1","source_table":"docs","projection":["id"],"udf_columns":["vec=embed(body)"],"filter":null,"auto_refresh":true}]}"#,
+                )
+                .unwrap()
+        });
+        let views = conn.list_materialized_views().await.unwrap();
+        assert_eq!(views.len(), 1);
+        assert_eq!(views[0].source_table, "docs");
+        assert!(views[0].auto_refresh);
+
+        // list_jobs: platform listing with status payloads
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.method(), &reqwest::Method::POST);
+            assert_eq!(request.url().path(), "/v1/jobs/list");
+            http::Response::builder()
+                .status(200)
+                .body(
+                    r#"{"jobs":[{"job_id":"plat-3","table":"docs","job_type":"indexer","job_subtype":"udf","state":"in_progress","created_at_millis":1000,"status":{"units_done":1,"units_total":2}}]}"#,
+                )
+                .unwrap()
+        });
+        let jobs = conn.list_jobs().await.unwrap();
+        assert_eq!(jobs.len(), 1);
+        assert_eq!(jobs[0].state, "running");
+        assert_eq!(jobs[0].job_type, "udf");
+        assert_eq!(jobs[0].units_total, Some(2));
+
+        // cancel_job: resolve via the manifest-id list filter, then cancel
+        let conn = Connection::new_with_handler(|request| match request.url().path() {
+            "/v1/jobs/list" => http::Response::builder()
+                .status(200)
+                .body(r#"{"jobs":[{"job_id":"plat-3","state":"in_progress"}]}"#)
+                .unwrap(),
+            "/v1/jobs/cancel" => {
+                assert_eq!(request.method(), &reqwest::Method::POST);
+                http::Response::builder()
+                    .status(200)
+                    .body(r#"{"job_id":"plat-3"}"#)
+                    .unwrap()
+            }
+            other => panic!("unexpected path {other}"),
+        });
+        assert!(conn.cancel_job("j-3").await.unwrap());
+
+        // cancel_job: never registered -> false, and no cancel request
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.url().path(), "/v1/jobs/list");
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"jobs":[]}"#)
+                .unwrap()
+        });
+        assert!(!conn.cancel_job("gone").await.unwrap());
+
+        // job_history(None): a registry listing, timeline-free
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.url().path(), "/v1/jobs/list");
+            http::Response::builder()
+                .status(200)
+                .body(
+                    r#"{"jobs":[{"job_id":"plat-1","table":"docs","job_type":"indexer","job_subtype":"udf","state":"done","created_at_millis":1000,"status":{"rows_skipped":3}}]}"#,
+                )
+                .unwrap()
+        });
+        let hist = conn.job_history(None).await.unwrap();
+        assert_eq!(hist.len(), 1);
+        assert_eq!(hist[0].state, "finished");
+        assert_eq!(hist[0].rows_skipped, Some(3));
+
+        // job_history(id): unknown everywhere -> empty
+        let conn = Connection::new_with_handler(|request| match request.url().path() {
+            "/v1/jobs/describe" => http::Response::builder().status(404).body("").unwrap(),
+            "/v1/jobs/list" => http::Response::builder()
+                .status(200)
+                .body(r#"{"jobs":[]}"#)
+                .unwrap(),
+            other => panic!("unexpected path {other}"),
+        });
+        assert!(conn.job_history(Some("j-1")).await.unwrap().is_empty());
+
+        // errors: GET /v1/errors with job + table filters
+        let conn = Connection::new_with_handler(|request| {
+            assert_eq!(request.method(), &reqwest::Method::GET);
+            assert_eq!(request.url().path(), "/v1/errors");
+            assert_eq!(request.url().query(), Some("job=j-1&table=docs"));
+            http::Response::builder()
+                .status(200)
+                .body(
+                    r#"{"errors":[{"job_id":"j-1","table":"docs","column":"vec","error_type":"ValueError","error_message":"boom","fragment_id":0,"source_row_id":42,"table_version":7,"age_seconds":5}]}"#,
+                )
+                .unwrap()
+        });
+        let errs = conn.errors(Some("j-1"), Some("docs")).await.unwrap();
+        assert_eq!(errs.len(), 1);
+        assert_eq!(errs[0].error_type, "ValueError");
+        assert_eq!(errs[0].source_row_id, Some(42));
+    }
+
    #[tokio::test]
    async fn test_clone_table() {
        let conn = Connection::new_with_handler(|request| {
@@ -2109,7 +2109,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
        Ok(delete_response)
    }

-    async fn create_index(&self, mut index: IndexBuilder) -> Result<()> {
+    async fn create_index(&self, mut index: IndexBuilder) -> Result<Option<String>> {
        self.check_mutable().await?;
        let request = self
            .client
@@ -2202,14 +2202,28 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {

        let (request_id, response) = self.send(request, true).await?;

-        self.check_table_response(&request_id, response).await?;
+        let response = self.check_table_response(&request_id, response).await?;
+
+        // The server returns a job id only when the build was deferred to a
+        // background job (pending vector index). Older servers return an
+        // empty body; treat anything unparseable as "no job".
+        #[derive(serde::Deserialize)]
+        struct CreateIndexResponse {
+            job_id: Option<String>,
+        }
+        let job_id = response
+            .text()
+            .await
+            .ok()
+            .and_then(|body| serde_json::from_str::<CreateIndexResponse>(&body).ok())
+            .and_then(|r| r.job_id);

        if let Some(wait_timeout) = index.wait_timeout {
            let index_name = index.name.unwrap_or_else(|| format!("{}_idx", column));
            self.wait_for_index(&[&index_name], wait_timeout).await?;
        }

-        Ok(())
+        Ok(job_id)
    }

    /// Poll until the columns are fully indexed. Will return Error::Timeout if the columns
@@ -2422,6 +2436,126 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
            message: "optimize is not supported on LanceDB cloud.".into(),
        })
    }
+    async fn add_computed_columns(
+        &self,
+        columns: &[(String, String)],
+        expression: &str,
+    ) -> Result<()> {
+        let new_columns: Vec<serde_json::Value> = columns
+            .iter()
+            .map(|(name, data_type)| {
+                serde_json::json!({
+                    "name": name,
+                    "computed": { "data_type": data_type, "expression": expression },
+                })
+            })
+            .collect();
+        let request = self
+            .client
+            .post(&format!("/v1/table/{}/add_columns/", self.identifier))
+            .json(&serde_json::json!({ "new_columns": new_columns }));
+        let (request_id, response) = self.send(request, true).await?;
+        self.check_table_response(&request_id, response).await?;
+        Ok(())
+    }
+
+    async fn refresh_column(
+        &self,
+        columns: &[String],
+        where_clause: Option<String>,
+        num_workers: Option<u32>,
+        max_workers: Option<u32>,
+        batch_size: Option<u32>,
+        priority: Option<String>,
+    ) -> Result<String> {
+        let mut body = serde_json::json!({ "columns": columns });
+        if let Some(w) = where_clause {
+            body["where_clause"] = serde_json::Value::String(w);
+        }
+        if let Some(n) = num_workers {
+            body["num_workers"] = n.into();
+        }
+        if let Some(n) = max_workers {
+            body["max_workers"] = n.into();
+        }
+        if let Some(n) = batch_size {
+            body["batch_size"] = n.into();
+        }
+        if let Some(p) = priority {
+            body["priority"] = serde_json::Value::String(p);
+        }
+        let request = self
+            .client
+            .post(&format!("/v1/table/{}/refresh_column", self.identifier))
+            .json(&body);
+        let (request_id, response) = self.send(request, true).await?;
+        let response = self.check_table_response(&request_id, response).await?;
+        #[derive(serde::Deserialize)]
+        struct RefreshColumnResponse {
+            job_id: String,
+        }
+        let body: RefreshColumnResponse = response.json().await.err_to_http(request_id)?;
+        Ok(body.job_id)
+    }
+
+    async fn load_columns(&self, request: crate::table::LoadColumnsRequest) -> Result<String> {
+        let columns: Vec<serde_json::Value> = request
+            .columns
+            .iter()
+            .map(|(target, source)| {
+                serde_json::json!({
+                    "target": target,
+                    "source": source.clone().unwrap_or_else(|| target.clone()),
+                })
+            })
+            .collect();
+        let mut source = serde_json::json!({
+            "uris": request.source_uris,
+            "format": request.source_format,
+        });
+        if let Some(opts) = request.source_storage_options {
+            source["storage_options"] = serde_json::to_value(opts).unwrap_or_default();
+        }
+        let mut body = serde_json::json!({
+            "columns": columns,
+            "source": source,
+            "target_key": request.target_key,
+        });
+        if let Some(k) = request.source_key {
+            body["source_key"] = serde_json::Value::String(k);
+        }
+        if let Some(m) = request.on_missing {
+            body["on_missing"] = serde_json::Value::String(m);
+        }
+        if let Some(n) = request.num_workers {
+            body["num_workers"] = n.into();
+        }
+        if let Some(n) = request.max_workers {
+            body["max_workers"] = n.into();
+        }
+        if let Some(n) = request.batch_size {
+            body["batch_size"] = n.into();
+        }
+        if let Some(n) = request.commit_granularity {
+            body["commit_granularity"] = n.into();
+        }
+        if let Some(p) = request.priority {
+            body["priority"] = serde_json::Value::String(p);
+        }
+        let http_request = self
+            .client
+            .post(&format!("/v1/table/{}/load_columns", self.identifier))
+            .json(&body);
+        let (request_id, response) = self.send(http_request, true).await?;
+        let response = self.check_table_response(&request_id, response).await?;
+        #[derive(serde::Deserialize)]
+        struct LoadColumnsResponse {
+            job_id: String,
+        }
+        let body: LoadColumnsResponse = response.json().await.err_to_http(request_id)?;
+        Ok(body.job_id)
+    }
+
    async fn add_columns(
        &self,
        transforms: NewColumnTransform,
@@ -2916,6 +3050,75 @@ mod tests {
        }
    }

+    #[tokio::test]
+    async fn test_refresh_column() {
+        let table = Table::new_with_handler("my_table", |request| {
+            assert_eq!(request.method(), "POST");
+            assert_eq!(request.url().path(), "/v1/table/my_table/refresh_column");
+            let body: serde_json::Value =
+                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
+            assert_eq!(body["columns"], serde_json::json!(["vec"]));
+            assert_eq!(body["num_workers"], 2);
+            assert!(body.get("where_clause").is_none());
+
+            http::Response::builder()
+                .status(202)
+                .body(r#"{"job_id":"j-9"}"#)
+                .unwrap()
+        });
+
+        let job_id = table
+            .refresh_column(&["vec".to_string()], None, Some(2), None, None, None)
+            .await
+            .unwrap();
+        assert_eq!(job_id, "j-9");
+    }
+
+    #[tokio::test]
+    async fn test_load_columns() {
+        let table = Table::new_with_handler("my_table", |request| {
+            assert_eq!(request.method(), "POST");
+            assert_eq!(request.url().path(), "/v1/table/my_table/load_columns");
+            let body: serde_json::Value =
+                serde_json::from_slice(request.body().unwrap().as_bytes().unwrap()).unwrap();
+            assert_eq!(
+                body["columns"],
+                serde_json::json!([{"target": "embedding", "source": "emb"}])
+            );
+            assert_eq!(body["source"]["format"], "parquet");
+            assert_eq!(
+                body["source"]["uris"],
+                serde_json::json!(["s3://b/x.parquet"])
+            );
+            assert_eq!(body["target_key"], "document_id");
+            assert_eq!(body["source_key"], "doc_id");
+            assert_eq!(body["on_missing"], "null");
+            assert_eq!(body["num_workers"], 4);
+
+            http::Response::builder()
+                .status(202)
+                .body(r#"{"job_id":"lc-7"}"#)
+                .unwrap()
+        });
+
+        let request = crate::table::LoadColumnsRequest {
+            source_uris: vec!["s3://b/x.parquet".to_string()],
+            source_format: "parquet".to_string(),
+            source_storage_options: None,
+            target_key: "document_id".to_string(),
+            source_key: Some("doc_id".to_string()),
+            columns: vec![("embedding".to_string(), Some("emb".to_string()))],
+            on_missing: Some("null".to_string()),
+            num_workers: Some(4),
+            max_workers: None,
+            batch_size: None,
+            commit_granularity: None,
+            priority: None,
+        };
+        let job_id = table.load_columns(request).await.unwrap();
+        assert_eq!(job_id, "lc-7");
+    }
+
    #[tokio::test]
    async fn test_version() {
        let table = Table::new_with_handler("my_table", |request| {
@@ -4450,6 +4653,42 @@ mod tests {
        }
    }

+    #[tokio::test]
+    async fn test_create_index_returns_deferred_job_id() {
+        let table =
+            Table::new_with_handler("my_table", move |request| match request.url().path() {
+                "/v1/table/my_table/describe/" => {
+                    let schema = Schema::new(vec![Field::new(
+                        "vector",
+                        DataType::FixedSizeList(
+                            Arc::new(Field::new("item", DataType::Float32, true)),
+                            128,
+                        ),
+                        false,
+                    )]);
+                    http::Response::builder()
+                        .status(200)
+                        .body(describe_response(&schema))
+                        .unwrap()
+                }
+                "/v1/table/my_table/create_index/" => http::Response::builder()
+                    .status(200)
+                    .body(r#"{"job_id": "0a1b2c3d-4e5f-6789-abcd-ef0123456789"}"#.to_string())
+                    .unwrap(),
+                path => panic!("Unexpected path: {}", path),
+            });
+
+        let job_id = table
+            .create_index(&["vector"], Index::IvfPq(Default::default()))
+            .execute()
+            .await
+            .unwrap();
+        assert_eq!(
+            job_id.as_deref(),
+            Some("0a1b2c3d-4e5f-6789-abcd-ef0123456789")
+        );
+    }
+
    #[tokio::test]
    async fn test_create_index_nested_field_paths() {
        let schema = nested_index_schema();
@@ -502,6 +502,33 @@ pub fn tokenize(query: &str, params: &InvertedIndexParams) -> Result<Vec<FtsToke
        .collect())
 }

+/// Request to fill existing table columns from an external source by
+/// primary-key join (Geneva `Table.load_columns()` parity). Server-backed
+/// feature (LanceDB Enterprise / Cloud).
+#[derive(Debug, Clone)]
+pub struct LoadColumnsRequest {
+    /// External source URIs.
+    pub source_uris: Vec<String>,
+    /// Source format: "parquet" | "lance" | "ipc".
+    pub source_format: String,
+    /// Source-only storage options (e.g. cloud credentials).
+    pub source_storage_options: Option<HashMap<String, String>>,
+    /// Destination primary-key column.
+    pub target_key: String,
+    /// Source primary-key column. Defaults to `target_key` when None.
+    pub source_key: Option<String>,
+    /// Value column mappings as `(target, source)`; a None source defaults to
+    /// the target name.
+    pub columns: Vec<(String, Option<String>)>,
+    /// Missing-row policy: "carry" (default) | "null" | "error".
+    pub on_missing: Option<String>,
+    pub num_workers: Option<u32>,
+    pub max_workers: Option<u32>,
+    pub batch_size: Option<u32>,
+    pub commit_granularity: Option<u32>,
+    pub priority: Option<String>,
+}
+
 /// A trait for anything "table-like".  This is used for both native tables (which target
 /// Lance datasets) and remote tables (which target LanceDB cloud)
 ///
@@ -555,7 +582,10 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
    /// Update rows in the table.
    async fn update(&self, update: UpdateBuilder) -> Result<UpdateResult>;
    /// Create an index on the provided column(s).
-    async fn create_index(&self, index: IndexBuilder) -> Result<()>;
+    ///
+    /// Returns the server-minted job id when the build was deferred to a
+    /// background job (remote tables only); `None` for synchronous builds.
+    async fn create_index(&self, index: IndexBuilder) -> Result<Option<String>>;
    /// List the indices on the table.
    async fn list_indices(&self) -> Result<Vec<IndexConfig>>;
    /// Drop an index from the table.
@@ -661,6 +691,47 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
        transforms: NewColumnTransform,
        read_columns: Option<Vec<String>>,
    ) -> Result<AddColumnsResult>;
+    /// Declare computed columns bound to a registered function: each
+    /// `(name, sql_type)` is added all-null with the expression stored
+    /// as its binding; no compute happens here (the server's lazy
+    /// detector or refresh_column fills them). Several columns map a
+    /// struct-returning function's fields positionally. Server-backed
+    /// feature; the default returns NotSupported.
+    async fn add_computed_columns(
+        &self,
+        _columns: &[(String, String)],
+        _expression: &str,
+    ) -> Result<()> {
+        Err(Error::NotSupported {
+            message: "computed columns are not supported by this table".into(),
+        })
+    }
+    /// Trigger recompute of computed columns. The expression is
+    /// resolved server-side from each column's stored binding; columns
+    /// bound to the same struct-returning function refresh together.
+    /// Returns the refresh job id. Server-backed feature (LanceDB
+    /// Enterprise / Cloud); the default returns NotSupported.
+    async fn refresh_column(
+        &self,
+        _columns: &[String],
+        _where_clause: Option<String>,
+        _num_workers: Option<u32>,
+        _max_workers: Option<u32>,
+        _batch_size: Option<u32>,
+        _priority: Option<String>,
+    ) -> Result<String> {
+        Err(Error::NotSupported {
+            message: "refresh_column is not supported by this table".into(),
+        })
+    }
+    /// Fill existing columns from an external source by primary-key join
+    /// (Geneva `load_columns`). Returns the load job id. Server-backed feature;
+    /// the default returns NotSupported.
+    async fn load_columns(&self, _request: LoadColumnsRequest) -> Result<String> {
+        Err(Error::NotSupported {
+            message: "load_columns is not supported by this table".into(),
+        })
+    }
    /// Alter columns in the table.
    async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result<AlterColumnsResult>;
    /// Drop columns from the table.
@@ -1502,6 +1573,48 @@ impl Table {
        self.inner.add_columns(transforms, read_columns).await
    }

+    /// Declare computed columns bound to a registered function
+    /// (`(name, sql_type)` pairs + a `f(args)` expression). No compute
+    /// happens here. Server-backed feature.
+    pub async fn add_computed_columns(
+        &self,
+        columns: &[(String, String)],
+        expression: &str,
+    ) -> Result<()> {
+        self.inner.add_computed_columns(columns, expression).await
+    }
+
+    /// Trigger recompute of computed columns (REFRESH COLUMN). The
+    /// expression comes from each column's stored binding; columns
+    /// bound to the same struct-returning function refresh together.
+    /// Returns the refresh job id. Server-backed feature.
+    pub async fn refresh_column(
+        &self,
+        columns: &[String],
+        where_clause: Option<String>,
+        num_workers: Option<u32>,
+        max_workers: Option<u32>,
+        batch_size: Option<u32>,
+        priority: Option<String>,
+    ) -> Result<String> {
+        self.inner
+            .refresh_column(
+                columns,
+                where_clause,
+                num_workers,
+                max_workers,
+                batch_size,
+                priority,
+            )
+            .await
+    }
+
+    /// Fill existing columns from an external Parquet/Lance/IPC source by
+    /// primary-key join (Geneva `Table.load_columns()`). Returns the job id.
+    pub async fn load_columns(&self, request: LoadColumnsRequest) -> Result<String> {
+        self.inner.load_columns(request).await
+    }
+
    /// Change a column's name or nullability.
    pub async fn alter_columns(
        &self,
@@ -2921,7 +3034,7 @@ impl BaseTable for NativeTable {
        Ok(AddResult { version })
    }

-    async fn create_index(&self, opts: IndexBuilder) -> Result<()> {
+    async fn create_index(&self, opts: IndexBuilder) -> Result<Option<String>> {
        if opts.columns.len() != 1 {
            return Err(Error::Schema {
                message: "Multi-column (composite) indices are not yet supported".to_string(),
@@ -2944,7 +3057,8 @@ impl BaseTable for NativeTable {
        }
        builder.await?;
        self.dataset.update(dataset);
-        Ok(())
+        // Native builds are synchronous -- there is never a background job.
+        Ok(None)
    }

    async fn drop_index(&self, index_name: &str) -> Result<()> {