feat: expose Function publication intervals in refresh options (#4928)

Python refresh APIs currently reject `publication_interval_rows`, so
callers cannot request incremental Function output publication through
the SDK. Add the optional typed parameter through the sync/async Python
APIs, PyO3, and Rust `RefreshColumnOptions`, and serialize it as a
top-level Enterprise refresh option.

An omitted value preserves the existing wire request and whole-fragment
default. Positive values set a publication threshold; zero and negative
Python values fail before a request is sent. Embedded refreshes reject
the Enterprise-only option. Documentation describes a threshold rather
than an exact number of rows per publication.

Validation on the designated development host:
- A matching baseline build reproduced the missing public keyword
argument in both sync and async submission tests.
- The updated PyO3 build passed 54 focused Python tests; Rust payload,
invalid-option, and default-wire tests each passed. Rust formatting and
Python Ruff checks passed.
- The single signed delivery commit differs from that tested
implementation only in a Rustdoc correction describing Embedded
rejection; formatting passed again. No dependency pins or lockfiles
changed.

The matching SDK is prepared for the separate server validation of
failure/cancellation retention and missing-only refill. Those end-to-end
results are still pending and are not implied by these SDK tests.

Refs [ENT-3733](https://linear.app/lancedb/issue/ENT-3733).

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.
 </Note>
 
+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:
+
+<CodeGroup>
+```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?;
+```
+</CodeGroup>
+
+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

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,

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,

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

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},

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"),
     ],

python/src/table.rs

@@ -46,6 +46,7 @@ mod scannable;
 struct RefreshColumnArgs {
     where_clause: Option<String>,
     batch_size: Option<i64>,
+    publication_interval_rows: Option<i64>,
     gpu_class: Option<String>,
     gpu_count: Option<i64>,
     retry_policy: Option<String>,
@@ -73,6 +74,7 @@ fn refresh_column_options(args: RefreshColumnArgs) -> PyResult<RefreshColumnOpti
     let RefreshColumnArgs {
         where_clause,
         batch_size,
+        publication_interval_rows,
         gpu_class,
         gpu_count,
         retry_policy,
@@ -114,6 +116,8 @@ fn refresh_column_options(args: RefreshColumnArgs) -> PyResult<RefreshColumnOpti
         .map(usize::try_from)
         .transpose()
         .map_err(|_| PyValueError::new_err("batch_size must be greater than zero"))?;
+    options.publication_interval_rows =
+        positive_count(publication_interval_rows, "publication_interval_rows")?;
     options.gpu_class = gpu_class;
     options.gpu_count = gpu_count
         .map(|count| {
@@ -2206,13 +2210,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(
         self_: PyRef<'_, Self>,
         column: String,
         r#where: Option<String>,
         batch_size: Option<i64>,
+        publication_interval_rows: Option<i64>,
         gpu_class: Option<String>,
         gpu_count: Option<i64>,
         retry_policy: Option<String>,
@@ -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<String>,
         batch_size: Option<i64>,
+        publication_interval_rows: Option<i64>,
         gpu_class: Option<String>,
         gpu_count: Option<i64>,
         retry_policy: Option<String>,
@@ -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,

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)

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};

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<usize>,
+    /// 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<u64>,
     /// 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<String>) -> 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)?;
         }