From 90b3715975abbb5c9698c6173628f59e84e13789 Mon Sep 17 00:00:00 2001 From: Wyatt Alt Date: Sat, 18 Jul 2026 21:02:29 -0700 Subject: [PATCH] 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 --- python/python/lancedb/db.py | 6 +- python/python/lancedb/table.py | 96 ++++++++++++++++++++------ python/python/lancedb/udf.py | 34 +++++++++ python/python/tests/test_job_handle.py | 27 ++++++++ 4 files changed, 139 insertions(+), 24 deletions(-) diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index 1b6883fa1..804552df9 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -1927,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, @@ -2000,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": @@ -2057,7 +2057,7 @@ 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 diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index a37f11454..498830a7c 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -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, @@ -954,7 +955,7 @@ class Table(ABC): wait_timeout: Optional[timedelta] = ..., name: Optional[str] = ..., train: bool = ..., - ) -> None: ... + ) -> "Job": ... # Legacy API overload (deprecated) @overload @@ -978,7 +979,7 @@ class Table(ABC): name: Optional[str] = ..., train: bool = ..., target_partition_size: Optional[int] = ..., - ) -> None: ... + ) -> "Job": ... def create_index( self, @@ -1029,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): @@ -2196,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, @@ -2599,7 +2608,7 @@ class LanceTable(Table): wait_timeout: Optional[timedelta] = ..., name: Optional[str] = ..., train: bool = ..., - ) -> None: ... + ) -> "Job": ... # Legacy API overload (deprecated) @overload @@ -2625,7 +2634,7 @@ class LanceTable(Table): name: Optional[str] = ..., train: bool = ..., target_partition_size: Optional[int] = ..., - ) -> None: ... + ) -> "Job": ... def create_index( self, @@ -2684,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): @@ -2759,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 @@ -2796,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, @@ -3059,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( @@ -3143,7 +3175,7 @@ class LanceTable(Table): ) try: - LOOP.run( + ajob = LOOP.run( self._table.create_index( field_names, replace=replace, @@ -3158,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: @@ -4563,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, @@ -4576,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() @@ -4883,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( @@ -4908,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, @@ -4925,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. @@ -6575,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. @@ -6589,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.""" diff --git a/python/python/lancedb/udf.py b/python/python/lancedb/udf.py index cb86f8654..a90c4cafc 100644 --- a/python/python/lancedb/udf.py +++ b/python/python/lancedb/udf.py @@ -612,6 +612,16 @@ class Job: 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: @@ -637,6 +647,8 @@ class Job: 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) @@ -646,6 +658,8 @@ class Job: 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 @@ -655,6 +669,8 @@ class Job: 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() @@ -680,6 +696,8 @@ class Job: 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: @@ -749,6 +767,14 @@ class AsyncJob: 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: @@ -764,6 +790,8 @@ class AsyncJob: 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) @@ -772,6 +800,8 @@ class AsyncJob: 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 @@ -781,6 +811,8 @@ class AsyncJob: 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() @@ -802,6 +834,8 @@ class AsyncJob: 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: diff --git a/python/python/tests/test_job_handle.py b/python/python/tests/test_job_handle.py index 10ed12079..2398f0548 100644 --- a/python/python/tests/test_job_handle.py +++ b/python/python/tests/test_job_handle.py @@ -180,3 +180,30 @@ def test_async_wait_returns_finished(): 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())