feat: add materialized view refresh resource budgets (#4902)

Expose aggregate CPU, memory, and GPU budgets plus GPU-class placement
on materialized-view refreshes in Python and Rust. Forward options
through SQL and reject invalid or unsupported Embedded requests instead
of silently dropping constraints.

```python
job = view.refresh_async(
    cpu_millis=8000,
    memory_bytes=16 * 1024**3,
    gpu_count=2,
    gpu_class="h100",
)
job.wait()
```

```sql
REFRESH MATERIALIZED VIEW curated_training_set
WITH (cpu_millis = 8000, memory_bytes = 17179869184,
      gpu_count = 2, gpu_class = 'h100');
```

The same options apply to blocking refreshes, async Python connections,
and the Rust refresh builder. CPU/memory budgets require all three
resource fields; GPU-only caps and class-only placement are also
supported. These are execution-wide caps and require a supporting
Enterprise server with an isolated execution backend. A cap does not
increase refresh parallelism; current MV refreshes use at most one GPU.

Updates API documentation and SQL reference, with Python sync/async and
Rust remote-submission regression coverage. Formatting and Python lint
pass; local builds and tests are intentionally skipped.

The Linux CI build and example steps explicitly select the LanceDB
package, matching the feature-test step. This avoids rebuilding and
retaining a second dependency graph after a workspace-wide build, which
exhausted runner disk space on both this branch and main.

Related: [ENT-3386](https://linear.app/lancedb/issue/ENT-3386)
This commit is contained in:
Jack Ye
2026-11-09 15:02:13 -08:00
committed by GitHub
parent 1eeb200705
commit 794fa023cd
11 changed files with 456 additions and 28 deletions
+16 -2
View File
@@ -654,10 +654,24 @@ class Table:
limit: Optional[int] = None,
) -> FunctionErrors: ...
async def refresh_materialized_view(
self, source_version: Optional[int] = None, *, full: bool = False
self,
source_version: Optional[int] = None,
*,
full: bool = False,
cpu_millis: Optional[int] = None,
memory_bytes: Optional[int] = None,
gpu_count: Optional[int] = None,
gpu_class: Optional[str] = None,
) -> RefreshMaterializedViewResult: ...
async def refresh_materialized_view_async(
self, source_version: Optional[int] = None, *, full: bool = False
self,
source_version: Optional[int] = None,
*,
full: bool = False,
cpu_millis: Optional[int] = None,
memory_bytes: Optional[int] = None,
gpu_count: Optional[int] = None,
gpu_class: Optional[str] = None,
) -> Job: ...
async def materialized_view_definition(self) -> str: ...
async def add_columns_with_schema(self, schema: pa.Schema) -> AddColumnsResult: ...
+95 -8
View File
@@ -17,6 +17,7 @@ from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Tuple, Union
import json
from dataclasses import dataclass
from .functions import _validate_gpu_class_name
from .background_loop import LOOP
from .job import AsyncJob, Job, _typed_job
@@ -156,6 +157,34 @@ def normalize_select(select: SelectArg) -> Optional[List[Tuple[str, str]]]:
return normalized
def _refresh_execution_options(cpu_millis, memory_bytes, gpu_count, gpu_class):
values = {
"cpu_millis": cpu_millis,
"memory_bytes": memory_bytes,
"gpu_count": gpu_count,
}
for name, value in values.items():
maximum = 2**32 - 1 if name == "gpu_count" else 2**64 - 1
minimum = 0 if name == "gpu_count" else 1
if value is not None and (
type(value) is not int or not minimum <= value <= maximum
):
raise ValueError(
f"{name} must be an integer between {minimum} and {maximum}"
)
if (cpu_millis is not None or memory_bytes is not None) and any(
value is None for value in values.values()
):
raise ValueError(
"cpu_millis, memory_bytes, and gpu_count must be specified together"
)
if gpu_class is not None:
values["gpu_class"] = _validate_gpu_class_name(gpu_class)
if gpu_count == 0:
raise ValueError("gpu_class requires a positive GPU budget")
return {name: value for name, value in values.items() if value is not None}
class AsyncMaterializedView:
"""A handle on a materialized view: its table plus its definition.
@@ -185,7 +214,14 @@ class AsyncMaterializedView:
return _definition_from_json(raw)
async def refresh(
self, *, full: bool = False, source_version: Optional[int] = None
self,
*,
full: bool = False,
source_version: Optional[int] = None,
cpu_millis: Optional[int] = None,
memory_bytes: Optional[int] = None,
gpu_count: Optional[int] = None,
gpu_class: Optional[str] = None,
) -> "RefreshMaterializedViewResult":
"""Recompute the view from its source.
@@ -194,16 +230,35 @@ class AsyncMaterializedView:
and ``source_version`` selects a source version instead of the latest.
LanceDB Enterprise SQL refreshes do not support source-version pinning.
``cpu_millis`` (1000 = one core), ``memory_bytes``, and ``gpu_count``
cap aggregate resources for this execution, including its Function
processes. CPU or memory requires all three budget fields; GPU count
may be zero for CPU-only views. ``gpu_class`` selects an operator-defined
GPU class for this execution only. Omitted options preserve defaults.
These options require an Enterprise backend that enforces isolation;
Embedded refreshes reject them. Other servers may reject the new options.
LanceDB Enterprise results leave ``mode``, ``rows_written``, and
``source_version`` as ``None`` because the SQL job adapter does not expose
these metrics.
"""
return await self._table._inner.refresh_materialized_view(
source_version=source_version, full=full
source_version=source_version,
full=full,
**_refresh_execution_options(
cpu_millis, memory_bytes, gpu_count, gpu_class
),
)
async def refresh_async(
self, *, full: bool = False, source_version: Optional[int] = None
self,
*,
full: bool = False,
source_version: Optional[int] = None,
cpu_millis: Optional[int] = None,
memory_bytes: Optional[int] = None,
gpu_count: Optional[int] = None,
gpu_class: Optional[str] = None,
) -> "AsyncJob[RefreshMaterializedViewResult]":
"""Submit a refresh and return its job without waiting.
@@ -221,7 +276,11 @@ class AsyncMaterializedView:
return _typed_job(
await self._table._inner.refresh_materialized_view_async(
source_version=source_version, full=full
source_version=source_version,
full=full,
**_refresh_execution_options(
cpu_millis, memory_bytes, gpu_count, gpu_class
),
),
RefreshMaterializedViewResult.from_json,
)
@@ -253,14 +312,36 @@ class MaterializedView:
return LOOP.run(self._async.definition())
def refresh(
self, *, full: bool = False, source_version: Optional[int] = None
self,
*,
full: bool = False,
source_version: Optional[int] = None,
cpu_millis: Optional[int] = None,
memory_bytes: Optional[int] = None,
gpu_count: Optional[int] = None,
gpu_class: Optional[str] = None,
) -> "RefreshMaterializedViewResult":
"""Recompute the view from its source. See
[AsyncMaterializedView.refresh][lancedb.materialized_view.AsyncMaterializedView.refresh]."""
return LOOP.run(self._async.refresh(full=full, source_version=source_version))
return LOOP.run(
self._async.refresh(
full=full,
source_version=source_version,
**_refresh_execution_options(
cpu_millis, memory_bytes, gpu_count, gpu_class
),
)
)
def refresh_async(
self, *, full: bool = False, source_version: Optional[int] = None
self,
*,
full: bool = False,
source_version: Optional[int] = None,
cpu_millis: Optional[int] = None,
memory_bytes: Optional[int] = None,
gpu_count: Optional[int] = None,
gpu_class: Optional[str] = None,
) -> "Job[RefreshMaterializedViewResult]":
"""Submit a refresh and return its job without waiting.
@@ -269,6 +350,12 @@ class MaterializedView:
"""
return Job(
LOOP.run(
self._async.refresh_async(full=full, source_version=source_version)
self._async.refresh_async(
full=full,
source_version=source_version,
**_refresh_execution_options(
cpu_millis, memory_bytes, gpu_count, gpu_class
),
)
)
)
+34 -3
View File
@@ -63,7 +63,17 @@ def mock_remote_materialized_view_refresh():
@pytest.mark.parametrize("sync", [False, True])
@pytest.mark.parametrize("background", [False, True])
@pytest.mark.parametrize("full", [False, True])
async def test_remote_refresh_reports_unknown_metrics(sync, background, full):
@pytest.mark.parametrize(
"resources",
[
{},
{"gpu_count": 2, "gpu_class": "h100"},
{"cpu_millis": 8000, "memory_bytes": 16 * 1024**3, "gpu_count": 0},
],
)
async def test_remote_refresh_reports_unknown_metrics(
sync, background, full, resources
):
flight = pytest.importorskip("pyarrow.flight")
class RefreshServer(flight.FlightServerBase):
@@ -90,7 +100,7 @@ async def test_remote_refresh_reports_unknown_metrics(sync, background, full):
mock_remote_materialized_view_refresh() as (host, requests),
RefreshServer() as server,
):
options = {"full": True} if full else {}
options = {"full": full, **resources}
connect = lancedb.connect if sync else lancedb.connect_async
db = connect(
"db://dev",
@@ -140,7 +150,28 @@ async def test_remote_refresh_reports_unknown_metrics(sync, background, full):
)
(refresh,) = refreshes
assert b'REFRESH MATERIALIZED VIEW "view"' in refresh
assert (b"WITH (full = true)" in refresh) is full
assert (b"full = true" in refresh) is full
for key, value in resources.items():
expected = (
f"{key} = '{value}'" if key == "gpu_class" else f"{key} = {value}"
)
assert expected.encode() in refresh
previous = len(server.commands)
for invalid in [
{"cpu_millis": 0},
{"memory_bytes": 1},
{"gpu_count": -1},
{"gpu_count": True},
{"gpu_count": 2**32},
{"gpu_class": "H200:8"},
{"gpu_count": 0, "gpu_class": "h100"},
]:
with pytest.raises(ValueError):
if sync:
view.refresh(**invalid)
else:
await view.refresh(**invalid)
assert len(server.commands) == previous
assert not any("count_rows" in path for path in requests)
+36 -2
View File
@@ -2309,11 +2309,16 @@ impl Table {
})
}
#[pyo3(signature = (source_version=None, *, full=false))]
#[pyo3(signature = (source_version=None, *, full=false, cpu_millis=None, memory_bytes=None, gpu_count=None, gpu_class=None))]
#[allow(clippy::too_many_arguments)]
pub fn refresh_materialized_view(
self_: PyRef<'_, Self>,
source_version: Option<u64>,
full: bool,
cpu_millis: Option<u64>,
memory_bytes: Option<u64>,
gpu_count: Option<u32>,
gpu_class: Option<String>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
@@ -2321,6 +2326,18 @@ impl Table {
.await
.infer_error()?;
let mut builder = view.refresh().full(full);
if let Some(value) = cpu_millis {
builder = builder.cpu_millis(value);
}
if let Some(value) = memory_bytes {
builder = builder.memory_bytes(value);
}
if let Some(value) = gpu_count {
builder = builder.gpu_count(value);
}
if let Some(value) = gpu_class {
builder = builder.gpu_class(value);
}
if let Some(version) = source_version {
builder = builder.source_version(version);
}
@@ -2329,11 +2346,16 @@ impl Table {
})
}
#[pyo3(signature = (source_version=None, *, full=false))]
#[pyo3(signature = (source_version=None, *, full=false, cpu_millis=None, memory_bytes=None, gpu_count=None, gpu_class=None))]
#[allow(clippy::too_many_arguments)]
pub fn refresh_materialized_view_async(
self_: PyRef<'_, Self>,
source_version: Option<u64>,
full: bool,
cpu_millis: Option<u64>,
memory_bytes: Option<u64>,
gpu_count: Option<u32>,
gpu_class: Option<String>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
@@ -2341,6 +2363,18 @@ impl Table {
.await
.infer_error()?;
let mut builder = view.refresh().full(full);
if let Some(value) = cpu_millis {
builder = builder.cpu_millis(value);
}
if let Some(value) = memory_bytes {
builder = builder.memory_bytes(value);
}
if let Some(value) = gpu_count {
builder = builder.gpu_count(value);
}
if let Some(value) = gpu_class {
builder = builder.gpu_class(value);
}
if let Some(version) = source_version {
builder = builder.source_version(version);
}