From 0abf6417333d303dcbf63b8b5e6193e0f53920ae Mon Sep 17 00:00:00 2001 From: Brendan Clement Date: Wed, 17 Jun 2026 13:30:53 -0400 Subject: [PATCH] feat: send read-freshness signal on the lance-namespace path (#3551) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ### Description `db://`-style connections that use the lance-namespace path (`LanceNamespaceDatabase` → `NativeTable` + the lance-namespace REST client) never sent a read-freshness signal. Against a server configured to serve cached table metadata up to some staleness window, this allows stale-read-after-write across handles and processes. The remote table path already solved this (#3439). This brings the namespace path to parity. The namespace REST client doesn't let callers attach headers directly, but it forwards a `DynamicContextProvider`'s `headers.*` context entries as HTTP headers per request. So: - A shared per-table baseline map is created before the namespace client. I built and installed on the `ConnectBuilder` via a context provider. - On read operations the provider emits ·x-lancedb-min-timestamp = max(baseline, now − read_consistency_interval)` (RFC3339), keyed by the operation's `object_id`. - Each table handle bumps its baseline (monotonically) on `checkout_latest()`, `restore()`, and every data/schema write. `checkout_latest()` is the primary hook: consumers refresh a handle there after writing elsewhere, then poll. Read operations that carry the floor: `describe_table`, `list_table_versions`, `query_table`, `list_tables`. `list_table_versions` is what resolves "latest" for managed-versioning tables (`get_latest_version`), so it's the op that makes `checkout_latest()` actually observe a prior write. `describe_table_version` is excluded (pinned to an immutable version). This mirrors #3439 (timestamp baseline, `max(baseline, now − interval)`, monotonic); no `min_version` and no body channel, since the namespace path has no version-returning write responses. ### Testing - Unit tests for `compute_min_timestamp` / `next_freshness_baseline` and the provider (header at/after a bumped baseline; nothing for an empty baseline + no interval; interval floor applies; non-read ops emit nothing; `list_tables` uses only the interval floor). - Verified end-to-end against a local server that honors the header: reads carry `x-lancedb-min-timestamp`, writes don't, and read-your-write holds. --- rust/lancedb/src/database.rs | 1 + rust/lancedb/src/database/namespace.rs | 54 +++- rust/lancedb/src/database/read_freshness.rs | 312 ++++++++++++++++++++ rust/lancedb/src/table.rs | 54 +++- 4 files changed, 410 insertions(+), 11 deletions(-) create mode 100644 rust/lancedb/src/database/read_freshness.rs diff --git a/rust/lancedb/src/database.rs b/rust/lancedb/src/database.rs index 787221909..6ce5bdbc9 100644 --- a/rust/lancedb/src/database.rs +++ b/rust/lancedb/src/database.rs @@ -32,6 +32,7 @@ use crate::table::{BaseTable, WriteOptions}; pub mod listing; pub mod namespace; +pub(crate) mod read_freshness; pub trait DatabaseOptions { fn serialize_into_map(&self, map: &mut HashMap); diff --git a/rust/lancedb/src/database/namespace.rs b/rust/lancedb/src/database/namespace.rs index 5ce6a0f55..149339cd4 100644 --- a/rust/lancedb/src/database/namespace.rs +++ b/rust/lancedb/src/database/namespace.rs @@ -4,7 +4,7 @@ //! Namespace-based database implementation that delegates table management to lance-namespace use std::collections::{HashMap, HashSet}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use async_trait::async_trait; use lance::io::commit::namespace_manifest::LanceNamespaceExternalManifestStore; @@ -29,6 +29,9 @@ use crate::database::listing::{ NewTableConfig, OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS, OPT_NEW_TABLE_STORAGE_VERSION, OPT_NEW_TABLE_V2_MANIFEST_PATHS, }; +use crate::database::read_freshness::{ + FreshnessBaselines, ReadFreshnessContextProvider, TableFreshness, +}; use crate::error::{Error, Result}; use crate::table::{NativeTable, map_namespace_lance_error}; use lance::dataset::WriteMode; @@ -51,6 +54,10 @@ fn is_table_already_exists_namespace_error(err: &lance::Error) -> bool { false } +/// Object-id delimiter default (matches `RestNamespaceBuilder`'s); overridable +/// via the `delimiter` property. +const DEFAULT_NAMESPACE_DELIMITER: &str = "$"; + /// A database implementation that uses lance-namespace for table management pub struct LanceNamespaceDatabase { namespace: Arc, @@ -70,6 +77,17 @@ pub struct LanceNamespaceDatabase { ns_properties: HashMap, // Options for tables created by this connection new_table_config: NewTableConfig, + // Per-table read-freshness baselines, shared with the context provider. + freshness_baselines: FreshnessBaselines, + // Delimiter for building freshness keys; see `table_freshness`. + delimiter: String, +} + +fn resolve_delimiter(ns_properties: &HashMap) -> String { + ns_properties + .get("delimiter") + .cloned() + .unwrap_or_else(|| DEFAULT_NAMESPACE_DELIMITER.to_string()) } impl LanceNamespaceDatabase { @@ -82,6 +100,9 @@ impl LanceNamespaceDatabase { session: Option>, namespace_client_pushdown_operations: HashSet, ) -> Self { + // Client is pre-built, so we can't install the freshness provider here; + // baselines are still tracked for a uniform bump path. + let delimiter = resolve_delimiter(&namespace_client_properties); Self { namespace: namespace_client, storage_options, @@ -92,6 +113,8 @@ impl LanceNamespaceDatabase { ns_impl: namespace_client_impl, ns_properties: namespace_client_properties, new_table_config: NewTableConfig::default(), + freshness_baselines: Arc::new(Mutex::new(HashMap::new())), + delimiter, } } @@ -136,10 +159,19 @@ impl LanceNamespaceDatabase { if let Some(ref sess) = session { builder = builder.session(sess.clone()); } + + // Install the read-freshness provider before building the client. + let freshness_baselines: FreshnessBaselines = Arc::new(Mutex::new(HashMap::new())); + builder = builder.context_provider(Arc::new(ReadFreshnessContextProvider::new( + freshness_baselines.clone(), + read_consistency_interval, + ))); + let namespace = builder.connect().await.map_err(|e| Error::InvalidInput { message: format!("Failed to connect to namespace: {:?}", e), })?; + let delimiter = resolve_delimiter(&ns_properties); Ok(Self { namespace, storage_options, @@ -150,9 +182,20 @@ impl LanceNamespaceDatabase { ns_impl: ns_impl.to_string(), ns_properties, new_table_config, + freshness_baselines, + delimiter, }) } + /// Build a table's freshness handle, keyed to match the `object_id` the + /// namespace client sends on reads (table-id parts joined by the delimiter). + fn table_freshness(&self, namespace_path: &[String], name: &str) -> TableFreshness { + let mut parts = namespace_path.to_vec(); + parts.push(name.to_string()); + let key = parts.join(&self.delimiter); + TableFreshness::new(self.freshness_baselines.clone(), key) + } + fn extract_storage_overrides( &self, request: &DbCreateTableRequest, @@ -331,7 +374,8 @@ impl Database for LanceNamespaceDatabase { self.pushdown_operations.clone(), self.session.clone(), ) - .await?; + .await? + .with_freshness(self.table_freshness(&request.namespace_path, &request.name)); return Ok(Arc::new(native_table)); } @@ -462,7 +506,8 @@ impl Database for LanceNamespaceDatabase { self.pushdown_operations.clone(), self.session.clone(), ) - .await?; + .await? + .with_freshness(self.table_freshness(&request.namespace_path, &request.name)); Ok(Arc::new(native_table)) } @@ -478,7 +523,8 @@ impl Database for LanceNamespaceDatabase { self.pushdown_operations.clone(), self.session.clone(), ) - .await?; + .await? + .with_freshness(self.table_freshness(&request.namespace_path, &request.name)); Ok(Arc::new(native_table)) } diff --git a/rust/lancedb/src/database/read_freshness.rs b/rust/lancedb/src/database/read_freshness.rs new file mode 100644 index 000000000..ac36802a4 --- /dev/null +++ b/rust/lancedb/src/database/read_freshness.rs @@ -0,0 +1,312 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Read-freshness signaling for the lance-namespace path. +//! +//! Against a server that serves cached table metadata up to some staleness +//! window, a handle that just wrote (or asked for the latest version via +//! `checkout_latest`) can still read a stale snapshot. To prevent that, reads +//! routed through the namespace client carry an `x-lancedb-min-timestamp` +//! header naming the oldest snapshot the caller will accept. +//! +//! This mirrors `remote::table`: a per-table baseline is bumped to "now" on +//! every write and on `checkout_latest()`, and reads send +//! `max(baseline, now - read_consistency_interval)`. Since the namespace client +//! takes no headers directly, a [`DynamicContextProvider`] injects it per request. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, SystemTime}; + +use lance_namespace_impls::{DynamicContextProvider, OperationInfo}; + +/// Provider context keys prefixed with `headers.` become HTTP headers (prefix +/// stripped), so this emits the `x-lancedb-min-timestamp` header. +const MIN_TIMESTAMP_CONTEXT_KEY: &str = "headers.x-lancedb-min-timestamp"; + +/// Per-table freshness baselines (keyed by namespace object id), shared between +/// the provider that reads them and the table handles that bump them. +pub type FreshnessBaselines = Arc>>; + +/// `max(baseline, now - interval)`, or `None` when neither constraint applies. +fn compute_min_timestamp( + baseline: Option, + interval: Option, + now: SystemTime, +) -> Option { + let interval_based = match interval { + None => None, + Some(d) if d.is_zero() => Some(now), + Some(d) => Some(now.checked_sub(d).unwrap_or(now)), + }; + match (interval_based, baseline) { + (None, None) => None, + (Some(t), None) | (None, Some(t)) => Some(t), + (Some(a), Some(b)) => Some(a.max(b)), + } +} + +/// Advance the baseline to `now`, never backwards, so a concurrent handle's +/// write can't lower a floor another handle already set. +fn next_freshness_baseline(prev: Option, now: SystemTime) -> SystemTime { + match prev { + Some(p) => p.max(now), + None => now, + } +} + +/// A handle's view of the shared baseline map for a single table. +#[derive(Clone, Debug)] +pub struct TableFreshness { + baselines: FreshnessBaselines, + /// Namespace object id for this table (matches the read's `object_id`). + key: String, +} + +impl TableFreshness { + pub fn new(baselines: FreshnessBaselines, key: String) -> Self { + Self { baselines, key } + } + + pub fn bump(&self) { + let now = SystemTime::now(); + let mut baselines = self.baselines.lock().unwrap(); + let prev = baselines.get(&self.key).copied(); + baselines.insert(self.key.clone(), next_freshness_baseline(prev, now)); + } +} + +/// Read ops that can be served stale and so carry the freshness floor. +/// `list_table_versions` resolves "latest" for managed-versioning tables, so it +/// is what makes `checkout_latest()` observe a prior write. +fn is_read_operation(operation: &str) -> bool { + matches!( + operation, + "describe_table" | "list_table_versions" | "query_table" | "list_tables" + ) +} + +/// Injects `x-lancedb-min-timestamp` on namespace reads, per addressed table. +#[derive(Debug)] +pub struct ReadFreshnessContextProvider { + baselines: FreshnessBaselines, + read_consistency_interval: Option, +} + +impl ReadFreshnessContextProvider { + pub fn new(baselines: FreshnessBaselines, read_consistency_interval: Option) -> Self { + Self { + baselines, + read_consistency_interval, + } + } +} + +impl DynamicContextProvider for ReadFreshnessContextProvider { + fn provide_context(&self, info: &OperationInfo) -> HashMap { + if !is_read_operation(&info.operation) { + return HashMap::new(); + } + + let baseline = self.baselines.lock().unwrap().get(&info.object_id).copied(); + match compute_min_timestamp(baseline, self.read_consistency_interval, SystemTime::now()) { + Some(ts) => { + let dt: chrono::DateTime = ts.into(); + HashMap::from([(MIN_TIMESTAMP_CONTEXT_KEY.to_string(), dt.to_rfc3339())]) + } + None => HashMap::new(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Allowed slop when comparing a header timestamp against a locally + /// captured wall-clock bound. Tests run fast enough that 1s is plenty. + const TOLERANCE: Duration = Duration::from_secs(1); + + fn parse_header_ts(headers: &HashMap) -> SystemTime { + let value = headers + .get(MIN_TIMESTAMP_CONTEXT_KEY) + .expect("expected min-timestamp context key"); + chrono::DateTime::parse_from_rfc3339(value) + .unwrap() + .with_timezone(&chrono::Utc) + .into() + } + + #[test] + fn test_compute_min_timestamp_combines_baseline_and_interval() { + let now = SystemTime::now(); + let baseline = now - Duration::from_secs(60); + + // No interval, no baseline -> no header. + assert_eq!(compute_min_timestamp(None, None, now), None); + + // Baseline only -> baseline. + assert_eq!( + compute_min_timestamp(Some(baseline), None, now), + Some(baseline) + ); + + // ZERO interval, no baseline -> now (strong consistency). + assert_eq!( + compute_min_timestamp(None, Some(Duration::ZERO), now), + Some(now) + ); + + // Positive interval, no baseline -> now - interval. + assert_eq!( + compute_min_timestamp(None, Some(Duration::from_secs(10)), now), + Some(now - Duration::from_secs(10)) + ); + + // Both: pick the more-recent (tighter) constraint. + // baseline = now-60, now-interval = now-10. now-10 is newer. + assert_eq!( + compute_min_timestamp(Some(baseline), Some(Duration::from_secs(10)), now), + Some(now - Duration::from_secs(10)) + ); + + // Both, baseline newer: pick baseline. + let recent_baseline = now - Duration::from_secs(5); + assert_eq!( + compute_min_timestamp(Some(recent_baseline), Some(Duration::from_secs(60)), now), + Some(recent_baseline) + ); + } + + #[test] + fn test_next_freshness_baseline_is_monotonic() { + let now = SystemTime::now(); + let earlier = now - Duration::from_secs(30); + let later = now + Duration::from_secs(30); + + // No prior baseline -> now. + assert_eq!(next_freshness_baseline(None, now), now); + // Prior baseline older than now -> now. + assert_eq!(next_freshness_baseline(Some(earlier), now), now); + // Prior baseline newer than now -> keep the newer baseline. + assert_eq!(next_freshness_baseline(Some(later), now), later); + } + + fn provider_with( + entries: &[(&str, SystemTime)], + interval: Option, + ) -> ReadFreshnessContextProvider { + let map: HashMap = + entries.iter().map(|(k, v)| (k.to_string(), *v)).collect(); + ReadFreshnessContextProvider::new(Arc::new(Mutex::new(map)), interval) + } + + #[test] + fn test_provider_emits_header_at_or_after_bumped_baseline() { + // A baseline set "now" with no interval: every read op must carry a + // floor at or after that baseline. `list_table_versions` is the hook + // that makes managed-versioning `checkout_latest()` observe a write. + let baseline = SystemTime::now(); + let provider = provider_with(&[("ns$tbl", baseline)], None); + + // These ops are keyed by the table id, so they pick up the per-table + // baseline. (`list_tables` is keyed by the namespace, so it is covered + // separately by the interval-floor test.) + for op in ["describe_table", "list_table_versions", "query_table"] { + let ctx = provider.provide_context(&OperationInfo::new(op, "ns$tbl")); + let sent = parse_header_ts(&ctx); + assert!( + sent >= baseline - TOLERANCE && sent <= baseline + TOLERANCE, + "operation {op} should carry a floor at the bumped baseline" + ); + } + } + + #[test] + fn test_provider_list_tables_uses_interval_floor_not_table_baseline() { + // `list_tables` is addressed by the namespace id, which never matches a + // per-table baseline key, so a bumped table baseline must not leak onto + // it. With no interval it sends nothing; with one it sends now-interval. + let provider = provider_with(&[("ns$tbl", SystemTime::now())], None); + let ctx = provider.provide_context(&OperationInfo::new("list_tables", "ns")); + assert!( + ctx.is_empty(), + "list_tables must not inherit a per-table baseline" + ); + + let interval = Duration::from_secs(30); + let provider = provider_with(&[("ns$tbl", SystemTime::now())], Some(interval)); + let before = SystemTime::now(); + let ctx = provider.provide_context(&OperationInfo::new("list_tables", "ns")); + let after = SystemTime::now(); + let sent = parse_header_ts(&ctx); + assert!( + sent >= before - interval - TOLERANCE && sent <= after - interval + TOLERANCE, + "list_tables should carry the interval floor" + ); + } + + #[test] + fn test_provider_no_header_for_empty_baseline_and_no_interval() { + // Manual consistency (no interval) on a table that was never bumped: + // no floor, so the server may serve from cache. + let provider = provider_with(&[], None); + let ctx = provider.provide_context(&OperationInfo::new("describe_table", "ns$tbl")); + assert!(ctx.is_empty()); + } + + #[test] + fn test_provider_interval_floor_applies_without_baseline() { + // With a consistency interval and no baseline, the floor is now-interval. + let interval = Duration::from_secs(30); + let provider = provider_with(&[], Some(interval)); + + let before = SystemTime::now(); + let ctx = provider.provide_context(&OperationInfo::new("query_table", "ns$tbl")); + let after = SystemTime::now(); + + let sent = parse_header_ts(&ctx); + assert!( + sent >= before - interval - TOLERANCE && sent <= after - interval + TOLERANCE, + "expected floor at roughly now - interval" + ); + } + + #[test] + fn test_provider_non_read_ops_emit_nothing() { + // Even with a fresh baseline and a zero interval, a non-read operation + // (which establishes rather than consumes a baseline) sends no header. + let provider = provider_with(&[("ns$tbl", SystemTime::now())], Some(Duration::ZERO)); + for op in [ + "create_table", + "register_table", + "drop_table", + "rename_table", + // Pinned to an immutable version, so it cannot be served stale. + "describe_table_version", + ] { + let ctx = provider.provide_context(&OperationInfo::new(op, "ns$tbl")); + assert!( + ctx.is_empty(), + "operation {op} must not send a freshness header" + ); + } + } + + #[test] + fn test_provider_uses_per_table_baseline() { + // The floor is looked up by object id, so an unrelated table's baseline + // does not leak onto another table's read. + let baseline = SystemTime::now(); + let provider = provider_with(&[("ns$has_baseline", baseline)], None); + + // The bumped table gets a header. + let hit = + provider.provide_context(&OperationInfo::new("describe_table", "ns$has_baseline")); + assert!(!hit.is_empty()); + + // A different table with no baseline (and no interval) gets nothing. + let miss = provider.provide_context(&OperationInfo::new("describe_table", "ns$other")); + assert!(miss.is_empty()); + } +} diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 93fdecec1..91f7080bf 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -43,6 +43,7 @@ use crate::connection::NamespaceClientPushdownOperation; use crate::data::scannable::{PeekedScannable, Scannable, estimate_write_partitions}; use crate::database::Database; +use crate::database::read_freshness::TableFreshness; use crate::embeddings::{EmbeddingDefinition, EmbeddingRegistry, MemoryRegistry}; use crate::error::{Error, Result}; use crate::index::IndexStatistics; @@ -1763,6 +1764,8 @@ pub struct NativeTable { // Operations to push down to the namespace server. // pub(crate) so query.rs can access the field for server-side query execution. pub(crate) pushdown_operations: HashSet, + // Read-freshness baseline; `Some` only for namespace-backed tables. + freshness: Option, } impl std::fmt::Debug for NativeTable { @@ -1923,6 +1926,7 @@ impl NativeTable { read_consistency_interval, namespace_client, pushdown_operations, + freshness: None, }) } @@ -1934,6 +1938,12 @@ impl NativeTable { self } + /// Attach the read-freshness baseline handle (namespace connections only). + pub(crate) fn with_freshness(mut self, freshness: TableFreshness) -> Self { + self.freshness = Some(freshness); + self + } + /// Build a sibling `NativeTable` with the same identity but a different /// (independent) dataset wrapper — used to hand out branch-scoped handles. fn with_dataset(&self, dataset: dataset::DatasetConsistencyWrapper) -> Self { @@ -1946,6 +1956,14 @@ impl NativeTable { read_consistency_interval: self.read_consistency_interval, namespace_client: self.namespace_client.clone(), pushdown_operations: self.pushdown_operations.clone(), + freshness: self.freshness.clone(), + } + } + + /// Bump the read-freshness baseline; no-op for non-namespace tables. + fn bump_freshness(&self) { + if let Some(freshness) = &self.freshness { + freshness.bump(); } } @@ -2045,6 +2063,7 @@ impl NativeTable { read_consistency_interval, namespace_client: stored_namespace_client, pushdown_operations, + freshness: None, }) } @@ -2134,6 +2153,7 @@ impl NativeTable { read_consistency_interval, namespace_client, pushdown_operations, + freshness: None, }) } @@ -2265,6 +2285,7 @@ impl NativeTable { read_consistency_interval, namespace_client: stored_namespace_client, pushdown_operations, + freshness: None, }) } @@ -2424,6 +2445,8 @@ impl BaseTable for NativeTable { } async fn checkout_latest(&self) -> Result<()> { + // Bump before resolving "latest" so that request carries the floor. + self.bump_freshness(); self.dataset.as_latest().await?; self.dataset.reload().await } @@ -2511,6 +2534,8 @@ impl BaseTable for NativeTable { debug_assert_eq!(dataset.version().version, version); dataset.restore().await?; } + // Restore moves "latest", so bump before resolving it (as RemoteTable does). + self.bump_freshness(); self.dataset.as_latest().await?; Ok(()) } @@ -2624,6 +2649,7 @@ impl BaseTable for NativeTable { } let version = ds_wrapper.get().await?.manifest().version; + self.bump_freshness(); Ok(AddResult { version }) } @@ -2674,7 +2700,9 @@ impl BaseTable for NativeTable { async fn update(&self, update: UpdateBuilder) -> Result { // Delegate to the submodule implementation - update::execute_update(self, update).await + let result = update::execute_update(self, update).await?; + self.bump_freshness(); + Ok(result) } async fn create_plan( @@ -2706,7 +2734,9 @@ impl BaseTable for NativeTable { params: MergeInsertBuilder, new_data: Box, ) -> Result { - merge::execute_merge_insert(self, params, new_data).await + let result = merge::execute_merge_insert(self, params, new_data).await?; + self.bump_freshness(); + Ok(result) } async fn set_unenforced_primary_key(&self, columns: &[&str]) -> Result<()> { @@ -2727,7 +2757,9 @@ impl BaseTable for NativeTable { /// Delete rows from the table async fn delete(&self, predicate: Predicate<'_>) -> Result { - delete::execute_delete(self, predicate).await + let result = delete::execute_delete(self, predicate).await?; + self.bump_freshness(); + Ok(result) } async fn tags(&self) -> Result> { @@ -2746,22 +2778,30 @@ impl BaseTable for NativeTable { transforms: NewColumnTransform, read_columns: Option>, ) -> Result { - schema_evolution::execute_add_columns(self, transforms, read_columns).await + let result = schema_evolution::execute_add_columns(self, transforms, read_columns).await?; + self.bump_freshness(); + Ok(result) } async fn alter_columns(&self, alterations: &[ColumnAlteration]) -> Result { - schema_evolution::execute_alter_columns(self, alterations).await + let result = schema_evolution::execute_alter_columns(self, alterations).await?; + self.bump_freshness(); + Ok(result) } async fn update_field_metadata( &self, updates: &[FieldMetadataUpdate], ) -> Result { - schema_evolution::execute_update_field_metadata(self, updates).await + let result = schema_evolution::execute_update_field_metadata(self, updates).await?; + self.bump_freshness(); + Ok(result) } async fn drop_columns(&self, columns: &[&str]) -> Result { - schema_evolution::execute_drop_columns(self, columns).await + let result = schema_evolution::execute_drop_columns(self, columns).await?; + self.bump_freshness(); + Ok(result) } async fn list_indices(&self) -> Result> {