mirror of
https://github.com/lancedb/lancedb.git
synced 2026-11-15 07:06:52 +00:00
fix: preserve function freshness across input-preserving writes (#4948)
Updating an unrelated column or merging rows with unchanged Function inputs rewrites fragment files, causing already computed outputs to be treated as stale. Preserve their freshness from the successful write transaction when the inputs and outputs are provably unchanged; keep changed or uncertain inputs eligible for recomputation. Mixed update/insert writes keep a separate pending-fill proof so retained outputs can be reused while inserted NULL outputs still need computation. Partial refreshes under a new Function definition discard the previous definition’s completed and pending proofs. Preservation uses a version-fenced metadata commit, and an optional preservation failure does not turn a successful data write into an error. Validation: 67 freshness tests pass, including the new partial-definition regression (confirmed failing before its fix), unrelated updates, identical/mixed merges, NULL/nested/blob inputs, dirty-state preservation, concurrency fences and sidecar retention. `cargo fmt --all` and `cargo clippy --quiet --locked --features remote --tests --examples` pass. Compatibility: `StalenessPlan` adds a public `pending_fill` field, so direct Rust struct literals must be updated. Refresh consumers must handle pending-fill admission and publication before adopting this behavior; a dependency bump alone does not complete consumer integration. End-to-end consumer validation and deployment remain separate. No latency claim is made for this SDK-only change.
This commit is contained in:
@@ -211,6 +211,11 @@ pub const FUNCTION_BINDINGS_VERSION: u32 = 1;
|
||||
/// declared before signatures existed.
|
||||
pub const SOURCE_SIGNATURE_META_KEY: &str = "computed_refresh.source_signature";
|
||||
|
||||
/// Signatures of input-preserving rewrites whose retained Function outputs
|
||||
/// are current but whose inserted rows still need their first computation.
|
||||
/// A sidecar reference keeps this separate from completed freshness stamps.
|
||||
pub const PENDING_FILL_SIGNATURE_META_KEY: &str = "computed_refresh.pending_fill_signature";
|
||||
|
||||
/// Field metadata key holding the definition digest a column was last
|
||||
/// computed under. A change to it makes every row stale.
|
||||
pub const DEFINITION_VERSION_META_KEY: &str = "computed_refresh.definition_version";
|
||||
|
||||
+1066
-11
File diff suppressed because it is too large
Load Diff
@@ -385,6 +385,8 @@ pub(crate) async fn execute_merge_insert(
|
||||
}
|
||||
|
||||
let dataset = table.dataset.get().await?;
|
||||
let source_schema = source.schema();
|
||||
let on = params.on.clone();
|
||||
let mut builder = LanceMergeInsertBuilder::try_new(dataset.clone(), params.on)?;
|
||||
match (
|
||||
params.when_matched_update_all,
|
||||
@@ -432,8 +434,21 @@ pub(crate) async fn execute_merge_insert(
|
||||
Either::Right(job.execute(source).map_err(|e| e.into()))
|
||||
};
|
||||
let (new_dataset, stats) = future.await?;
|
||||
let mut new_dataset = new_dataset.as_ref().clone();
|
||||
if let Err(error) = super::freshness::preserve_function_freshness(
|
||||
&mut new_dataset,
|
||||
super::freshness::FunctionWrite::Merge {
|
||||
source_schema: &source_schema,
|
||||
on: &on,
|
||||
num_inserted_rows: stats.num_inserted_rows,
|
||||
},
|
||||
)
|
||||
.await
|
||||
{
|
||||
log::warn!("merge committed but Function freshness could not be preserved: {error}");
|
||||
}
|
||||
let version = new_dataset.manifest().version;
|
||||
table.dataset.update(new_dataset.as_ref().clone());
|
||||
table.dataset.update(new_dataset);
|
||||
Ok(MergeResult {
|
||||
version,
|
||||
num_updated_rows: stats.num_updated_rows,
|
||||
|
||||
@@ -113,6 +113,11 @@ pub(crate) async fn execute_update(
|
||||
}
|
||||
|
||||
// 4. Apply the columns (SET clause)
|
||||
let written_columns = update
|
||||
.columns
|
||||
.iter()
|
||||
.map(|(column, _)| column.clone())
|
||||
.collect::<Vec<_>>();
|
||||
for (column, value) in update.columns {
|
||||
builder = builder.set(column, &value)?;
|
||||
}
|
||||
@@ -120,13 +125,25 @@ pub(crate) async fn execute_update(
|
||||
// 5. Execute the operation (Write new files)
|
||||
let operation = builder.build()?;
|
||||
let res = operation.execute().await?;
|
||||
let mut new_dataset = res.new_dataset.as_ref().clone();
|
||||
if let Err(error) = super::freshness::preserve_function_freshness(
|
||||
&mut new_dataset,
|
||||
super::freshness::FunctionWrite::Update {
|
||||
columns: &written_columns,
|
||||
},
|
||||
)
|
||||
.await
|
||||
{
|
||||
log::warn!("update committed but Function freshness could not be preserved: {error}");
|
||||
}
|
||||
|
||||
// 6. Update the table's view of the latest version
|
||||
table.dataset.update(res.new_dataset.as_ref().clone());
|
||||
let version = new_dataset.version().version;
|
||||
table.dataset.update(new_dataset);
|
||||
|
||||
Ok(UpdateResult {
|
||||
rows_updated: res.rows_updated,
|
||||
version: res.new_dataset.version().version,
|
||||
version,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user