diff --git a/docs/web/feature-engineering/refresh-column.mdx b/docs/web/feature-engineering/refresh-column.mdx index ac4ff93fe..0dfe3b33d 100644 --- a/docs/web/feature-engineering/refresh-column.mdx +++ b/docs/web/feature-engineering/refresh-column.mdx @@ -261,6 +261,38 @@ release. TypeScript does not expose GPU class discovery; pass `gpuClass` and `gpuCount` in the `refreshColumnAsync` options. +Completed Function outputs publish with their fragment by default. To opt into +a row-count publication threshold, set `publication_interval_rows` in Python or +`publication_interval_rows()` in Rust: + + +```python Python icon="python" +job = table.refresh_column_async( + "embedding", + publication_interval_rows=60, +) +job.wait() +``` + +```rust Rust icon="rust" +use lancedb::table::RefreshColumnOptions; + +let _job = table + .refresh_column_async_with_options( + "embedding", + RefreshColumnOptions::default().publication_interval_rows(60), + ) + .await?; +``` + + +The threshold is opt-in and must be greater than zero. The number of rows in an +individual publication can differ from this value depending on completed work +and Function batching; this is not an exact-N contract. Omit it to keep +whole-fragment publication. + +The TypeScript SDK does not expose the publication threshold option yet. + The job APIs also take the row filter and failure handling of the SQL `REFRESH COLUMN ... WHERE ... WITH (...)` form, so a refresh can be narrowed to new rows (see [what a filter recomputes](#partial)) and told to skip rows diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 2ac4dde58..f2d5285fa 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -619,6 +619,7 @@ class Table: *, where: Optional[str] = None, batch_size: Optional[int] = None, + publication_interval_rows: Optional[int] = None, gpu_class: Optional[str] = None, gpu_count: Optional[int] = None, retry_policy: Optional[str] = None, @@ -636,6 +637,7 @@ class Table: *, where: Optional[str] = None, batch_size: Optional[int] = None, + publication_interval_rows: Optional[int] = None, gpu_class: Optional[str] = None, gpu_count: Optional[int] = None, retry_policy: Optional[str] = None, diff --git a/python/python/lancedb/remote/table.py b/python/python/lancedb/remote/table.py index 8f1f741fb..a963ab55e 100644 --- a/python/python/lancedb/remote/table.py +++ b/python/python/lancedb/remote/table.py @@ -1143,6 +1143,7 @@ class RemoteTable(Table): *, where: Optional[str] = None, batch_size: Optional[int] = None, + publication_interval_rows: Optional[int] = None, gpu_class: Optional[str] = None, gpu_count: Optional[int] = None, retry_policy: Optional[FunctionRetryPolicy] = None, @@ -1159,6 +1160,7 @@ class RemoteTable(Table): column, where=where, batch_size=batch_size, + publication_interval_rows=publication_interval_rows, gpu_class=gpu_class, gpu_count=gpu_count, retry_policy=retry_policy, @@ -1178,6 +1180,7 @@ class RemoteTable(Table): *, where: Optional[str] = None, batch_size: Optional[int] = None, + publication_interval_rows: Optional[int] = None, gpu_class: Optional[str] = None, gpu_count: Optional[int] = None, retry_policy: Optional[FunctionRetryPolicy] = None, @@ -1195,6 +1198,7 @@ class RemoteTable(Table): column, where=where, batch_size=batch_size, + publication_interval_rows=publication_interval_rows, gpu_class=gpu_class, gpu_count=gpu_count, retry_policy=retry_policy, diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 75975a2c3..06bace5b8 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -3178,6 +3178,7 @@ class Table(ABC): *, where: Optional[str] = None, batch_size: Optional[int] = None, + publication_interval_rows: Optional[int] = None, gpu_class: Optional[str] = None, gpu_count: Optional[int] = None, retry_policy: Optional[FunctionRetryPolicy] = None, @@ -3216,6 +3217,12 @@ class Table(ABC): Maximum rows per Function call on the remote server. Must be greater than zero. None uses the server default. Has no effect on local expression-backed columns. + publication_interval_rows: int, optional + Opt-in publication threshold for completed Function outputs. Must + be greater than zero. None keeps whole-fragment publication. The + actual publication boundary can be larger than this threshold + because it also depends on completed work and Function batching. + LanceDB Enterprise only; local refreshes reject this option. gpu_class: str, optional GPU class for Function workers. Must be a lowercase name of 1-63 letters, digits, or hyphens, without a leading or trailing hyphen. @@ -3263,6 +3270,7 @@ class Table(ABC): *, where: Optional[str] = None, batch_size: Optional[int] = None, + publication_interval_rows: Optional[int] = None, gpu_class: Optional[str] = None, gpu_count: Optional[int] = None, retry_policy: Optional[FunctionRetryPolicy] = None, @@ -3300,6 +3308,12 @@ class Table(ABC): greater than zero. None uses the server default. Lower this for slow Functions to stay within the executor's per-call deadline. Has no effect on local expression-backed columns. + publication_interval_rows: int, optional + Opt-in publication threshold for completed Function outputs. Must + be greater than zero. None keeps whole-fragment publication. The + actual publication boundary can be larger than this threshold + because it also depends on completed work and Function batching. + LanceDB Enterprise only; local refreshes reject this option. gpu_class: str, optional GPU class for Function workers. Must be a lowercase name of 1-63 letters, digits, or hyphens, without a leading or trailing hyphen. @@ -5442,6 +5456,7 @@ class LanceTable(Table): *, where: Optional[str] = None, batch_size: Optional[int] = None, + publication_interval_rows: Optional[int] = None, gpu_class: Optional[str] = None, gpu_count: Optional[int] = None, retry_policy: Optional[FunctionRetryPolicy] = None, @@ -5461,6 +5476,7 @@ class LanceTable(Table): column, where=where, batch_size=batch_size, + publication_interval_rows=publication_interval_rows, gpu_class=gpu_class, gpu_count=gpu_count, retry_policy=retry_policy, @@ -5480,6 +5496,7 @@ class LanceTable(Table): *, where: Optional[str] = None, batch_size: Optional[int] = None, + publication_interval_rows: Optional[int] = None, gpu_class: Optional[str] = None, gpu_count: Optional[int] = None, retry_policy: Optional[FunctionRetryPolicy] = None, @@ -5501,6 +5518,7 @@ class LanceTable(Table): column, where=where, batch_size=batch_size, + publication_interval_rows=publication_interval_rows, gpu_class=gpu_class, gpu_count=gpu_count, retry_policy=retry_policy, @@ -7865,6 +7883,7 @@ class AsyncTable: *, where: Optional[str] = None, batch_size: Optional[int] = None, + publication_interval_rows: Optional[int] = None, gpu_class: Optional[str] = None, gpu_count: Optional[int] = None, retry_policy: Optional[FunctionRetryPolicy] = None, @@ -7903,6 +7922,12 @@ class AsyncTable: Maximum rows per Function call on the remote server. Must be greater than zero. None uses the server default. Has no effect on local expression-backed columns. + publication_interval_rows: int, optional + Opt-in publication threshold for completed Function outputs. Must + be greater than zero. None keeps whole-fragment publication. The + actual publication boundary can be larger than this threshold + because it also depends on completed work and Function batching. + LanceDB Enterprise only; local refreshes reject this option. gpu_class: str, optional GPU class for Function workers. Must be a lowercase name of 1-63 letters, digits, or hyphens, without a leading or trailing hyphen. @@ -7945,6 +7970,7 @@ class AsyncTable: column, where=where, batch_size=batch_size, + publication_interval_rows=publication_interval_rows, gpu_class=gpu_class, gpu_count=gpu_count, retry_policy=None @@ -8054,6 +8080,7 @@ class AsyncTable: *, where: Optional[str] = None, batch_size: Optional[int] = None, + publication_interval_rows: Optional[int] = None, gpu_class: Optional[str] = None, gpu_count: Optional[int] = None, retry_policy: Optional[FunctionRetryPolicy] = None, @@ -8091,6 +8118,12 @@ class AsyncTable: greater than zero. None uses the server default. Lower this for slow Functions to stay within the executor's per-call deadline. Has no effect on local expression-backed columns. + publication_interval_rows: int, optional + Opt-in publication threshold for completed Function outputs. Must + be greater than zero. None keeps whole-fragment publication. The + actual publication boundary can be larger than this threshold + because it also depends on completed work and Function batching. + LanceDB Enterprise only; local refreshes reject this option. gpu_class: str, optional GPU class for Function workers. Must be a lowercase name of 1-63 letters, digits, or hyphens, without a leading or trailing hyphen. @@ -8150,6 +8183,7 @@ class AsyncTable: column, where=where, batch_size=batch_size, + publication_interval_rows=publication_interval_rows, gpu_class=gpu_class, gpu_count=gpu_count, retry_policy=None diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py index 685bc809d..671d0ff23 100644 --- a/python/python/tests/test_remote_db.py +++ b/python/python/tests/test_remote_db.py @@ -1863,6 +1863,7 @@ def test_remote_create_index_async_returns_job(): {"batch_size": None}, {"batch_size": 1}, {"batch_size": 8}, + {"publication_interval_rows": 60}, {"gpu_class": "a10g", "gpu_count": 2}, { "retry_policy": {"transient": True, "max_attempts": 3}, @@ -1911,6 +1912,7 @@ async def test_remote_refresh_async_returns_typed_terminal_result( expected["error_policy"] = refresh_kwargs["error_policy"] for key in ( "where", + "publication_interval_rows", "max_failed_fraction", "max_failed_rows", "row_timeout_seconds", @@ -2088,6 +2090,8 @@ def test_remote_refresh_async_forwards_progress_from_submitted_handle(): ({"max_failed_rows": -1}, "max_failed_rows"), ({"row_timeout_seconds": 0}, "row_timeout_seconds"), ({"init_timeout_seconds": -5}, "init_timeout_seconds"), + ({"publication_interval_rows": 0}, "publication_interval_rows"), + ({"publication_interval_rows": -1}, "publication_interval_rows"), ({"num_workers": 0}, "num_workers"), ( {"max_executor_processes_per_worker": -2}, diff --git a/python/python/tests/test_table.py b/python/python/tests/test_table.py index 2a502fe62..157870bb5 100644 --- a/python/python/tests/test_table.py +++ b/python/python/tests/test_table.py @@ -5995,6 +5995,7 @@ def test_refresh_column_rejects_invalid_batch_size(tmp_path, method, batch_size) ({"max_failed_rows": 10}, "max_failed_rows"), ({"row_timeout_seconds": 30}, "row_timeout_seconds"), ({"init_timeout_seconds": 60}, "init_timeout_seconds"), + ({"publication_interval_rows": 60}, "publication_interval_rows"), ({"num_workers": 2}, "num_workers"), ({"max_executor_processes_per_worker": 2}, "max_executor_processes_per_worker"), ], diff --git a/python/src/table.rs b/python/src/table.rs index 92a6ef835..23c5b69d7 100644 --- a/python/src/table.rs +++ b/python/src/table.rs @@ -46,6 +46,7 @@ mod scannable; struct RefreshColumnArgs { where_clause: Option, batch_size: Option, + publication_interval_rows: Option, gpu_class: Option, gpu_count: Option, retry_policy: Option, @@ -73,6 +74,7 @@ fn refresh_column_options(args: RefreshColumnArgs) -> PyResult PyResult, column: String, r#where: Option, batch_size: Option, + publication_interval_rows: Option, gpu_class: Option, gpu_count: Option, retry_policy: Option, @@ -2227,6 +2232,7 @@ impl Table { let options = refresh_column_options(RefreshColumnArgs { where_clause: r#where, batch_size, + publication_interval_rows, gpu_class, gpu_count, retry_policy, @@ -2248,13 +2254,14 @@ impl Table { }) } - #[pyo3(signature = (column, *, r#where=None, batch_size=None, gpu_class=None, gpu_count=None, retry_policy=None, error_policy=None, max_failed_fraction=None, max_failed_rows=None, row_timeout_seconds=None, init_timeout_seconds=None, num_workers=None, max_executor_processes_per_worker=None))] + #[pyo3(signature = (column, *, r#where=None, batch_size=None, publication_interval_rows=None, gpu_class=None, gpu_count=None, retry_policy=None, error_policy=None, max_failed_fraction=None, max_failed_rows=None, row_timeout_seconds=None, init_timeout_seconds=None, num_workers=None, max_executor_processes_per_worker=None))] #[allow(clippy::too_many_arguments)] pub fn refresh_column_async( self_: PyRef<'_, Self>, column: String, r#where: Option, batch_size: Option, + publication_interval_rows: Option, gpu_class: Option, gpu_count: Option, retry_policy: Option, @@ -2269,6 +2276,7 @@ impl Table { let options = refresh_column_options(RefreshColumnArgs { where_clause: r#where, batch_size, + publication_interval_rows, gpu_class, gpu_count, retry_policy, diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs index 6956905c4..a69611306 100644 --- a/rust/lancedb/src/remote/table.rs +++ b/rust/lancedb/src/remote/table.rs @@ -11072,6 +11072,7 @@ mod tests { serde_json::json!({ "column": "quality", "where": "id % 2 = 0", + "publication_interval_rows": 60, "error_policy": "skip", "max_failed_fraction": 0.01, "max_failed_rows": 50, @@ -11088,6 +11089,7 @@ mod tests { }); let options = RefreshColumnOptions::default() .where_clause("id % 2 = 0") + .publication_interval_rows(60) .error_policy(FunctionErrorPolicy::Skip) .max_failed_fraction(0.01) .max_failed_rows(50) @@ -11135,6 +11137,10 @@ mod tests { RefreshColumnOptions::default().max_executor_processes_per_worker(0), "max_executor_processes_per_worker", ), + ( + RefreshColumnOptions::default().publication_interval_rows(0), + "publication_interval_rows", + ), ] { let error = table .refresh_column_async_with_options("quality", options) diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index a58567b2e..80b6bd7ad 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -2141,10 +2141,14 @@ impl Table { /// /// [`RefreshColumnOptions::batch_size()`] caps rows per Function call in /// LanceDB Enterprise and must be greater than zero when set. + /// [`RefreshColumnOptions::publication_interval_rows()`] sets an opt-in + /// row-count publication threshold for completed Function outputs; unset + /// keeps whole-fragment publication. /// [`RefreshColumnOptions::gpu_class()`] selects a GPU pool and /// [`RefreshColumnOptions::gpu_count()`] sets the aggregate GPU budget. - /// These options have no effect on expression-backed columns in LanceDB - /// Embedded. Remote refreshes require + /// Batch-size and GPU options have no effect on expression-backed columns + /// in LanceDB Embedded; a publication threshold is rejected there. + /// Remote refreshes require /// [`Table::refresh_column_async_with_options`]. /// /// ``` @@ -2201,12 +2205,16 @@ impl Table { /// Set [`RefreshColumnOptions::batch_size()`] to cap the rows each remote /// Function call receives, for example to keep slow Functions within the /// executor's per-call deadline. Set + /// [`RefreshColumnOptions::publication_interval_rows()`] to publish + /// completed Function outputs at a row-count threshold; unset keeps + /// whole-fragment publication. Set /// [`RefreshColumnOptions::gpu_class()`] and /// [`RefreshColumnOptions::gpu_count()`] to override the registered GPU /// class or set an aggregate GPU budget for Function workers. Unset values /// use the Function and server defaults; zero counts and malformed class /// names are rejected before submitting a job. Embedded expression - /// refreshes ignore all three options. + /// refreshes ignore the batch-size and GPU options, and reject publication + /// thresholds and other Enterprise-only options. /// /// ``` /// # use lancedb::{Table, table::RefreshColumnOptions}; diff --git a/rust/lancedb/src/table/refresh.rs b/rust/lancedb/src/table/refresh.rs index 3c8ded1cf..a8be34b17 100644 --- a/rust/lancedb/src/table/refresh.rs +++ b/rust/lancedb/src/table/refresh.rs @@ -100,6 +100,14 @@ pub struct RefreshColumnOptions { /// effect on expression-backed columns in LanceDB Embedded. #[serde(skip_serializing_if = "Option::is_none")] pub batch_size: Option, + /// Opt-in row-count threshold for publishing completed Function outputs. + /// + /// Must be greater than zero. `None` keeps whole-fragment publication. + /// Actual publication boundaries can be larger than this threshold because + /// they also depend on completed work and Function batching. LanceDB Embedded + /// rejects this option. + #[serde(skip_serializing_if = "Option::is_none")] + pub publication_interval_rows: Option, /// GPU class for Function workers in LanceDB Enterprise. /// /// `None` keeps the Function's registered class, or the deployment default @@ -174,6 +182,13 @@ impl RefreshColumnOptions { self } + /// Use `rows` as the opt-in publication threshold for completed Function + /// outputs. The actual publication boundary can be larger than this value. + pub fn publication_interval_rows(mut self, rows: u64) -> Self { + self.publication_interval_rows = Some(rows); + self + } + /// Request Function workers from `gpu_class`. pub fn gpu_class(mut self, gpu_class: impl Into) -> Self { self.gpu_class = Some(gpu_class.into()); @@ -201,6 +216,10 @@ impl RefreshColumnOptions { "max_executor_processes_per_worker", self.max_executor_processes_per_worker.is_some(), ), + ( + "publication_interval_rows", + self.publication_interval_rows.is_some(), + ), ] .into_iter() .find_map(|(name, set)| set.then_some(name)) @@ -253,6 +272,11 @@ impl RefreshColumnOptions { message: "batch_size must be greater than zero".into(), }); } + if self.publication_interval_rows == Some(0) { + return Err(Error::InvalidInput { + message: "publication_interval_rows must be greater than zero".into(), + }); + } if let Some(gpu_class) = &self.gpu_class { validate_gpu_class_name(gpu_class)?; }