diff --git a/rust/error.rs b/rust/error.rs index 70e41de3..09b6107f 100644 --- a/rust/error.rs +++ b/rust/error.rs @@ -67,6 +67,7 @@ pub enum ErrorCode { Exec = 52, DatasetMerge = 53, NamespaceQueryTable = 54, + DatasetCheckoutLatest = 55, } struct LastError { diff --git a/rust/ffi/dataset.rs b/rust/ffi/dataset.rs index 3e5dec8e..32f73a67 100644 --- a/rust/ffi/dataset.rs +++ b/rust/ffi/dataset.rs @@ -216,6 +216,122 @@ fn open_dataset_with_storage_options_inner( Ok(DatasetHandle::new(dataset)) } +/// Refresh a dataset handle to the latest committed version if it is stale. +/// +/// Compares the handle's checked-out manifest identity against the latest +/// committed manifest on storage (a metadata-only lookup). The identity is +/// version + naming scheme + manifest e-tag, mirroring Lance's own +/// `already_checked_out` semantics: a bare version-id comparison would treat +/// a dataset that was dropped and re-created at the same URI (whose history +/// restarts at the same version id) as fresh. A missing e-tag on either side +/// is conservatively treated as stale. +/// +/// If the cached manifest is stale, a *new* dataset handle checked out at the +/// latest version is written to `out_new_dataset`; the input handle is left +/// untouched so that concurrent readers of the old handle stay valid. If the +/// handle is already at the latest version, `out_new_dataset` is set to null. +/// +/// Returns `0` on success and `-1` on error. +#[no_mangle] +pub unsafe extern "C" fn lance_dataset_checkout_latest_if_stale( + dataset: *mut c_void, + out_new_dataset: *mut *mut c_void, +) -> i32 { + match checkout_latest_if_stale_inner(dataset, out_new_dataset) { + Ok(()) => { + clear_last_error(); + 0 + } + Err(err) => { + set_last_error(err.code, err.message); + -1 + } + } +} + +/// Refresh `dataset` to the latest committed version if its checked-out +/// manifest is no longer the current one. Shared by the plain and the +/// namespace-aware revalidation FFI entry points. +/// +/// Returns `Ok(None)` when the dataset is already at the latest version and +/// `Ok(Some(refreshed))` when a newer (or re-created) manifest was found. +pub(crate) async fn refresh_dataset_if_stale( + dataset: &Arc, +) -> Result, String> { + let (_, latest_location) = dataset + .latest_manifest() + .await + .map_err(|err| format!("latest manifest lookup: {err}"))?; + let cached_location = dataset.manifest_location(); + // Compare the full manifest identity, not just the version id: a + // dataset dropped and re-created at the same URI restarts its history + // at the same version id, so only the e-tag distinguishes it from the + // cached manifest. Like Lance's own `already_checked_out`, a missing + // e-tag on either side means the identity cannot be confirmed and the + // entry is treated as stale. + let is_current = latest_location.version == cached_location.version + && latest_location.naming_scheme == cached_location.naming_scheme + && latest_location.e_tag.as_ref().is_some_and(|latest_e_tag| { + cached_location + .e_tag + .as_ref() + .is_some_and(|cached_e_tag| latest_e_tag == cached_e_tag) + }); + if is_current { + return Ok(None); + } + // A different manifest was committed, possibly by an external writer. + // Check out the latest version on a clone so the original handle + // remains usable by existing holders. `checkout_latest` re-reads the + // manifest keyed by (version, e-tag), so re-created datasets are + // loaded fresh rather than served from the metadata cache. + let mut refreshed_dataset = (**dataset).clone(); + refreshed_dataset + .checkout_latest() + .await + .map_err(|err| format!("checkout latest: {err}"))?; + Ok(Some(refreshed_dataset)) +} + +fn checkout_latest_if_stale_inner( + dataset: *mut c_void, + out_new_dataset: *mut *mut c_void, +) -> FfiResult<()> { + if out_new_dataset.is_null() { + return Err(FfiError::new( + ErrorCode::InvalidArgument, + "out_new_dataset is null", + )); + } + + let handle = unsafe { super::util::dataset_handle(dataset)? }; + + let refreshed = match runtime::block_on(refresh_dataset_if_stale(&handle.dataset)) { + Ok(Ok(v)) => v, + Ok(Err(message)) => { + return Err(FfiError::new( + ErrorCode::DatasetCheckoutLatest, + format!("dataset checkout latest if stale: {message}"), + )) + } + Err(err) => return Err(FfiError::new(ErrorCode::Runtime, format!("runtime: {err}"))), + }; + + let new_handle = match refreshed { + Some(refreshed_dataset) => { + Box::into_raw(Box::new(DatasetHandle::new(Arc::new(refreshed_dataset)))) as *mut c_void + } + None => ptr::null_mut(), + }; + + // SAFETY: `out_new_dataset` was null-checked above and is provided by the + // caller as a valid output location. + unsafe { + std::ptr::write_unaligned(out_new_dataset, new_handle); + } + Ok(()) +} + #[no_mangle] pub unsafe extern "C" fn lance_close_dataset(dataset: *mut c_void) { if !dataset.is_null() { @@ -883,3 +999,190 @@ fn dataset_delete_inner( } Ok(()) } + +#[cfg(test)] +mod tests { + use std::ffi::CString; + use std::fs; + + use arrow_array::{Int32Array, RecordBatchIterator}; + use arrow_schema::{DataType, Field, Schema as ArrowSchema}; + use lance::dataset::{WriteMode, WriteParams}; + use lance::Dataset; + + use super::*; + + /// Commit a batch to `uri` through the Lance Rust API, bypassing any FFI + /// dataset handle (i.e. acting as an external writer). + fn external_write(uri: &str, ids: Vec, mode: WriteMode) { + let schema = Arc::new(ArrowSchema::new(vec![Field::new( + "id", + DataType::Int32, + false, + )])); + let batch = arrow_array::RecordBatch::try_new( + schema.clone(), + vec![Arc::new(Int32Array::from(ids))], + ) + .unwrap(); + let reader = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema); + let params = WriteParams { + mode, + ..Default::default() + }; + runtime::block_on(Dataset::write(reader, uri, Some(params))) + .unwrap() + .unwrap(); + } + + #[test] + fn test_checkout_latest_if_stale_sees_external_commits() { + // Opening datasets mutates the process-global debug counters, which + // other tests assert on; serialize with them. + let _counter_guard = crate::ffi::session::debug_counter_test_lock(); + let dataset_dir = + std::env::temp_dir().join(format!("ffi-dataset-revalidate-{}", rand::random::())); + let uri = dataset_dir.to_string_lossy().to_string(); + + external_write(&uri, vec![1, 2, 3], WriteMode::Create); + + unsafe { + let uri_c = CString::new(uri.clone()).unwrap(); + let handle = lance_open_dataset(uri_c.as_ptr()); + assert!(!handle.is_null()); + assert_eq!(lance_dataset_count_rows(handle), 3); + + // Already at the latest version: no refreshed handle is produced. + let mut refreshed: *mut c_void = ptr::null_mut(); + assert_eq!( + lance_dataset_checkout_latest_if_stale(handle, &mut refreshed), + 0 + ); + assert!(refreshed.is_null()); + + // External append committed behind the open handle's back. + external_write(&uri, vec![4, 5], WriteMode::Append); + + // The stale handle still serves the old version... + assert_eq!(lance_dataset_count_rows(handle), 3); + + // ...but revalidation yields a refreshed handle at the latest + // version while leaving the original handle untouched. + assert_eq!( + lance_dataset_checkout_latest_if_stale(handle, &mut refreshed), + 0 + ); + assert!(!refreshed.is_null()); + assert_eq!(lance_dataset_count_rows(refreshed), 5); + assert_eq!(lance_dataset_count_rows(handle), 3); + + // External delete on top of the refreshed handle. + { + let mut ds = runtime::block_on(Dataset::open(&uri)).unwrap().unwrap(); + runtime::block_on(ds.delete("id = 1")).unwrap().unwrap(); + } + let mut refreshed_again: *mut c_void = ptr::null_mut(); + assert_eq!( + lance_dataset_checkout_latest_if_stale(refreshed, &mut refreshed_again), + 0 + ); + assert!(!refreshed_again.is_null()); + assert_eq!(lance_dataset_count_rows(refreshed_again), 4); + + lance_close_dataset(refreshed_again); + lance_close_dataset(refreshed); + lance_close_dataset(handle); + } + + let _ = fs::remove_dir_all(dataset_dir); + } + + #[test] + fn test_checkout_latest_if_stale_sees_recreated_dataset_with_same_version_id() { + // Opening datasets mutates the process-global debug counters, which + // other tests assert on; serialize with them. + let _counter_guard = crate::ffi::session::debug_counter_test_lock(); + let dataset_dir = std::env::temp_dir().join(format!( + "ffi-dataset-revalidate-recreate-{}", + rand::random::() + )); + let uri = dataset_dir.to_string_lossy().to_string(); + + external_write(&uri, vec![1, 2, 3], WriteMode::Create); + + unsafe { + let uri_c = CString::new(uri.clone()).unwrap(); + let handle = lance_open_dataset(uri_c.as_ptr()); + assert!(!handle.is_null()); + assert_eq!(lance_dataset_count_rows(handle), 3); + + // External drop + re-create at the same URI: the new dataset's + // history restarts at the same version id as the cached handle, + // so only the manifest e-tag distinguishes the two tables. + fs::remove_dir_all(&dataset_dir).unwrap(); + external_write(&uri, vec![10, 20], WriteMode::Create); + + let mut refreshed: *mut c_void = ptr::null_mut(); + assert_eq!( + lance_dataset_checkout_latest_if_stale(handle, &mut refreshed), + 0 + ); + assert!(!refreshed.is_null()); + assert_eq!(lance_dataset_count_rows(refreshed), 2); + + lance_close_dataset(refreshed); + lance_close_dataset(handle); + } + + let _ = fs::remove_dir_all(dataset_dir); + } + + #[test] + fn test_checkout_latest_if_stale_rejects_invalid_arguments() { + unsafe { + // Null output pointer. + let mut refreshed: *mut c_void = ptr::null_mut(); + assert_eq!( + lance_dataset_checkout_latest_if_stale(ptr::null_mut(), &mut refreshed), + -1 + ); + // Null dataset handle. + assert_eq!( + lance_dataset_checkout_latest_if_stale(ptr::null_mut(), ptr::null_mut()), + -1 + ); + } + } + + #[test] + fn test_checkout_latest_if_stale_errors_when_dataset_removed() { + // Opening datasets mutates the process-global debug counters, which + // other tests assert on; serialize with them. + let _counter_guard = crate::ffi::session::debug_counter_test_lock(); + let dataset_dir = std::env::temp_dir().join(format!( + "ffi-dataset-revalidate-removed-{}", + rand::random::() + )); + let uri = dataset_dir.to_string_lossy().to_string(); + + external_write(&uri, vec![1, 2, 3], WriteMode::Create); + + unsafe { + let uri_c = CString::new(uri.clone()).unwrap(); + let handle = lance_open_dataset(uri_c.as_ptr()); + assert!(!handle.is_null()); + + // The dataset vanishing from storage must surface as an error, + // not as a silent stale read. + fs::remove_dir_all(&dataset_dir).unwrap(); + let mut refreshed: *mut c_void = ptr::null_mut(); + assert_eq!( + lance_dataset_checkout_latest_if_stale(handle, &mut refreshed), + -1 + ); + assert!(refreshed.is_null()); + + lance_close_dataset(handle); + } + } +} diff --git a/rust/ffi/namespace.rs b/rust/ffi/namespace.rs index ff0b2325..4999de1e 100644 --- a/rust/ffi/namespace.rs +++ b/rust/ffi/namespace.rs @@ -202,7 +202,12 @@ fn describe_table_info_inner( // FIX: a qualified table id (e.g. "catalog.schema.table") must be sent as // its multi-segment namespace path, not a single segment. Split on the // delimiter so the server sees the full 3-level id instead of "got: 1". - req.id = Some(table_id.split(delimiter.as_str()).map(|s| s.to_string()).collect()); + req.id = Some( + table_id + .split(delimiter.as_str()) + .map(|s| s.to_string()) + .collect(), + ); req.with_table_uri = Some(true); let resp = namespace.describe_table(req).await.map_err(|err| { FfiError::new( @@ -306,8 +311,10 @@ fn create_empty_table_inner( .delimiter(delimiter.clone()) .build(); - let table_id_segments: Vec = - table_id.split(delimiter.as_str()).map(|s| s.to_string()).collect(); + let table_id_segments: Vec = table_id + .split(delimiter.as_str()) + .map(|s| s.to_string()) + .collect(); let (location, storage_options_tsv) = runtime::block_on(async move { let mut req = DeclareTableRequest::new(); req.id = Some(table_id_segments); @@ -413,8 +420,10 @@ fn drop_table_inner( .delimiter(delimiter.clone()) .build(); - let table_id_segments: Vec = - table_id.split(delimiter.as_str()).map(|s| s.to_string()).collect(); + let table_id_segments: Vec = table_id + .split(delimiter.as_str()) + .map(|s| s.to_string()) + .collect(); runtime::block_on(async move { let mut req = DropTableRequest::new(); req.id = Some(table_id_segments); @@ -488,7 +497,12 @@ fn describe_table_with_schema_inner( let schema_json = runtime::block_on(async move { let mut req = DescribeTableRequest::new(); // FIX: split the qualified id into its namespace segments (see describe_table_info_inner). - req.id = Some(table_id.split(delimiter.as_str()).map(|s| s.to_string()).collect()); + req.id = Some( + table_id + .split(delimiter.as_str()) + .map(|s| s.to_string()) + .collect(), + ); req.with_table_uri = Some(true); req.load_detailed_metadata = Some(true); let resp = namespace.describe_table(req).await.map_err(|err| { @@ -587,20 +601,26 @@ fn open_dataset_in_namespace_inner( let session = unsafe { optional_session_handle(session)? }; // FIX: split the qualified id into namespace segments so the crate's internal // describe (DatasetBuilder::from_namespace) gets the full 3-level id, not 1. - let table_id_segments: Vec = - table_id.split(delimiter.as_str()).map(|s| s.to_string()).collect(); + let table_id_segments: Vec = table_id + .split(delimiter.as_str()) + .map(|s| s.to_string()) + .collect(); let (dataset, table_uri) = runtime::block_on(async move { + // `DatasetBuilder::from_namespace` performs the describe internally + // and, when the namespace vends storage options, installs Lance's + // dynamic storage-options provider so later credential rotation is + // refreshed from the namespace automatically — one describe per open, + // no snapshot to keep in sync. record_namespace_describe(); - let mut builder = - DatasetBuilder::from_namespace(Arc::new(namespace), table_id_segments) - .await - .map_err(|err| { - FfiError::new( - ErrorCode::NamespaceDescribeTable, - format!("namespace describe_table: {err}"), - ) - })?; + let mut builder = DatasetBuilder::from_namespace(Arc::new(namespace), table_id_segments) + .await + .map_err(|err| { + FfiError::new( + ErrorCode::NamespaceDescribeTable, + format!("namespace describe_table: {err}"), + ) + })?; if let Some(session) = session { builder = builder.with_session(session); } @@ -704,6 +724,220 @@ pub unsafe extern "C" fn lance_open_dataset_in_namespace_with_session( } } +/// Refresh a namespace-backed dataset handle if the table moved or is stale. +/// +/// Namespace tables are cached by endpoint/table id rather than by physical +/// URI, so revalidating only the already-resolved handle would miss an +/// external drop/re-create that re-points the table to a new location. This +/// entry point first re-describes the table through the namespace (one +/// namespace round trip — the same order of cost as the namespace open path, +/// which itself starts with a describe) and reopens through the namespace when +/// the resolved location changed. When the location is unchanged, it falls +/// back to the manifest-identity revalidation used for plain datasets, which +/// catches both ordinary new commits and same-location re-creates (e-tag). +/// +/// Namespace-vended credential rotation needs no handling here: the handle +/// was opened via `DatasetBuilder::from_namespace`, which installs Lance's +/// dynamic storage-options provider, so fresh credentials are fetched from +/// the namespace by the object store itself. Revalidation only needs to +/// detect location moves and new commits. +/// +/// On success writes the refreshed handle (or null when the cached handle is +/// current) to `out_new_dataset`. `out_table_uri` receives the newly resolved +/// table URI only when the table moved; the caller frees it with +/// `lance_free_string`. Returns `0` on success and `-1` on error. +#[no_mangle] +#[allow(clippy::too_many_arguments)] +pub unsafe extern "C" fn lance_dataset_namespace_checkout_latest_if_stale( + dataset: *mut c_void, + endpoint: *const c_char, + table_id: *const c_char, + bearer_token: *const c_char, + api_key: *const c_char, + delimiter: *const c_char, + headers_tsv: *const c_char, + session: *mut c_void, + out_new_dataset: *mut *mut c_void, + out_table_uri: *mut *const c_char, +) -> i32 { + if !out_table_uri.is_null() { + unsafe { + std::ptr::write_unaligned(out_table_uri, ptr::null()); + } + } + match namespace_checkout_latest_if_stale_inner( + dataset, + endpoint, + table_id, + bearer_token, + api_key, + delimiter, + headers_tsv, + session, + out_new_dataset, + out_table_uri, + ) { + Ok(()) => { + clear_last_error(); + 0 + } + Err(err) => { + set_last_error(err.code, err.message); + -1 + } + } +} + +#[allow(clippy::too_many_arguments)] +fn namespace_checkout_latest_if_stale_inner( + dataset: *mut c_void, + endpoint: *const c_char, + table_id: *const c_char, + bearer_token: *const c_char, + api_key: *const c_char, + delimiter: *const c_char, + headers_tsv: *const c_char, + session: *mut c_void, + out_new_dataset: *mut *mut c_void, + out_table_uri: *mut *const c_char, +) -> FfiResult<()> { + if out_new_dataset.is_null() { + return Err(FfiError::new( + ErrorCode::InvalidArgument, + "out_new_dataset is null", + )); + } + + let handle = unsafe { super::util::dataset_handle(dataset)? }; + let endpoint = unsafe { cstr_to_str(endpoint, "endpoint")? }; + let table_id = unsafe { cstr_to_str(table_id, "table_id")? }; + let delimiter = unsafe { optional_cstr_to_string(delimiter, "delimiter")? }; + let bearer_token = unsafe { optional_cstr_to_string(bearer_token, "bearer_token")? }; + let api_key = unsafe { optional_cstr_to_string(api_key, "api_key")? }; + let headers_tsv = unsafe { optional_cstr_to_string(headers_tsv, "headers_tsv")? }; + let session = unsafe { optional_session_handle(session)? }; + + let delimiter = delimiter.unwrap_or_else(|| "$".to_string()); + let namespace = build_config( + endpoint, + bearer_token.as_deref(), + api_key.as_deref(), + headers_tsv.as_deref(), + ) + .delimiter(delimiter.clone()) + .build(); + // FIX: split the qualified id into its namespace segments once (see + // describe_table_info_inner) and reuse the vector for both the describe + // below and the namespace reopen, so a multi-level table resolves under + // the correct namespace during revalidation too. + let table_id_segments: Vec = table_id + .split(delimiter.as_str()) + .map(|s| s.to_string()) + .collect(); + + let (refreshed, moved_uri) = runtime::block_on(async move { + // Re-resolve the table location through the namespace before trusting + // the cached handle: an external drop/re-create can re-point the table + // to a different physical URI. + record_namespace_describe(); + let request = DescribeTableRequest { + id: Some(table_id_segments.clone()), + // Mirror the other describe paths: request the complete table URI + // so namespaces that only report `table_uri` (the response model + // allows omitting `location`) can still be revalidated. + with_table_uri: Some(true), + ..Default::default() + }; + let response = namespace.describe_table(request).await.map_err(|err| { + FfiError::new( + ErrorCode::NamespaceDescribeTable, + format!("namespace describe_table: {err}"), + ) + })?; + let location = response.location; + let table_uri = response.table_uri; + if location.is_none() && table_uri.is_none() { + return Err(FfiError::new( + ErrorCode::NamespaceDescribeTable, + "table location not found in namespace response", + )); + } + + // Treat the cached handle as current when either reported form + // matches its URI: `DatasetBuilder::from_namespace` derives + // `dataset.uri()` from `location`, so requiring the preferred + // `table_uri` form to match would flag a perpetual (false) move on + // servers that report both fields in different spellings. + let cached_uri = handle.dataset.uri(); + let location_matches = + location.as_deref() == Some(cached_uri) || table_uri.as_deref() == Some(cached_uri); + + if !location_matches { + // The table was re-pointed to a new location: reopen through the + // namespace path so managed versioning and namespace-provided + // storage options are re-applied (this re-describes internally; + // the extra round trip only happens on this path). + let mut builder = + DatasetBuilder::from_namespace(Arc::new(namespace), table_id_segments) + .await + .map_err(|err| { + FfiError::new( + ErrorCode::NamespaceDescribeTable, + format!("namespace describe_table: {err}"), + ) + })?; + if let Some(session) = session { + builder = builder.with_session(session); + } + let reopened = builder.load().await.map_err(|err| { + FfiError::new( + ErrorCode::DatasetOpen, + format!("namespace dataset open: {err}"), + ) + })?; + record_dataset_open(); + let uri = reopened.uri().to_string(); + return Ok::<_, FfiError>((Some(reopened), Some(uri))); + } + + // Same location: fall back to the manifest-identity revalidation. + // Rotated namespace-vended credentials do not require a reopen — the + // handle's object store refreshes them through the dynamic + // storage-options provider installed at open time. + let refreshed = super::dataset::refresh_dataset_if_stale(&handle.dataset) + .await + .map_err(|message| { + FfiError::new( + ErrorCode::DatasetCheckoutLatest, + format!("dataset checkout latest if stale: {message}"), + ) + })?; + Ok((refreshed, None)) + }) + .map_err(|err| FfiError::new(ErrorCode::Runtime, format!("runtime: {err}")))??; + + let new_handle = match refreshed { + Some(refreshed_dataset) => { + Box::into_raw(Box::new(DatasetHandle::new(Arc::new(refreshed_dataset)))) as *mut c_void + } + None => ptr::null_mut(), + }; + + // SAFETY: `out_new_dataset` was null-checked above and is provided by the + // caller as a valid output location. + unsafe { + std::ptr::write_unaligned(out_new_dataset, new_handle); + } + if let (Some(uri), false) = (moved_uri, out_table_uri.is_null()) { + let uri_c = CString::new(uri).unwrap_or_else(|_| to_c_string("invalid uri")); + // SAFETY: `out_table_uri` was null-checked in the tuple condition. + unsafe { + std::ptr::write_unaligned(out_table_uri, uri_c.into_raw() as *const c_char); + } + } + Ok(()) +} + /// Convert a JSON Arrow schema string to Arrow C Data Interface ArrowSchema. #[no_mangle] pub unsafe extern "C" fn lance_json_arrow_schema_to_c( @@ -743,3 +977,667 @@ pub unsafe extern "C" fn lance_json_arrow_schema_to_c( } } } + +#[cfg(test)] +mod tests { + use std::ffi::CString; + use std::io::{Read, Write}; + use std::net::TcpListener; + use std::sync::{Arc, Mutex}; + + use arrow_array::{Int32Array, RecordBatch, RecordBatchIterator}; + use arrow_schema::{DataType, Field, Schema as ArrowSchema}; + use lance::dataset::{WriteMode, WriteParams}; + use lance::Dataset; + + use super::super::dataset::{lance_close_dataset, lance_dataset_count_rows}; + use super::*; + use crate::runtime; + + /// Commit a batch to `uri` through the Lance Rust API, bypassing any FFI + /// dataset handle (i.e. acting as an external writer). + fn external_write(uri: &str, ids: Vec, mode: WriteMode) { + let schema = Arc::new(ArrowSchema::new(vec![Field::new( + "id", + DataType::Int32, + false, + )])); + let batch = + RecordBatch::try_new(schema.clone(), vec![Arc::new(Int32Array::from(ids))]).unwrap(); + let reader = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema); + let params = WriteParams { + mode, + ..Default::default() + }; + runtime::block_on(Dataset::write(reader, uri, Some(params))) + .unwrap() + .unwrap(); + } + + /// (endpoint, stop flag, log of raw requests) handed out by + /// `spawn_describe_server`. + type MockServer = (String, Arc>, Arc>>); + + /// Minimal REST namespace mock: answers every request with the + /// `describe_table`-shaped JSON body currently stored in `body`, so tests + /// can switch between location-only, table_uri-only, and re-pointed + /// responses (emulating an external drop/re-create that moves the table). + /// + /// Every raw request (head + body) is appended to the returned log so + /// tests can assert on what actually went over the wire. When + /// `expected_id` is set, requests whose JSON body `id` array differs are + /// rejected with a 404, emulating a server that resolves multi-level + /// identifiers: the REST client joins the segments with the delimiter in + /// the URL path (identical for one segment or many), so the body `id` + /// array is where a qualified id sent as a single segment shows up. + fn spawn_describe_server( + body: Arc>, + expected_id: Option>, + ) -> MockServer { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + let stop = Arc::new(Mutex::new(false)); + let stop_flag = stop.clone(); + let requests = Arc::new(Mutex::new(Vec::new())); + let requests_log = requests.clone(); + std::thread::spawn(move || { + for stream in listener.incoming() { + if *stop_flag.lock().unwrap() { + break; + } + let Ok(mut stream) = stream else { + continue; + }; + // Read the request head and the content-length body so the + // client sees a complete exchange. + let mut buf = Vec::new(); + let mut chunk = [0u8; 1024]; + let header_end = loop { + let Ok(n) = stream.read(&mut chunk) else { + break None; + }; + if n == 0 { + break None; + } + buf.extend_from_slice(&chunk[..n]); + if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") { + break Some(pos + 4); + } + }; + let Some(header_end) = header_end else { + continue; + }; + let head = String::from_utf8_lossy(&buf[..header_end]).to_string(); + let content_length = head + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + if name.eq_ignore_ascii_case("content-length") { + value.trim().parse::().ok() + } else { + None + } + }) + .unwrap_or(0); + while buf.len() < header_end + content_length { + let Ok(n) = stream.read(&mut chunk) else { + break; + }; + if n == 0 { + break; + } + buf.extend_from_slice(&chunk[..n]); + } + + let request_body = String::from_utf8_lossy(&buf[header_end..]).to_string(); + requests_log + .lock() + .unwrap() + .push(format!("{head}{request_body}")); + + let id_matches = expected_id.as_ref().is_none_or(|expected| { + serde_json::from_str::(&request_body) + .ok() + .and_then(|request| request.get("id").and_then(|id| id.as_array()).cloned()) + .is_some_and(|segments| { + segments + .iter() + .map(|segment| segment.as_str().unwrap_or_default()) + .eq(expected.iter().copied()) + }) + }); + let response = if id_matches { + let body = body.lock().unwrap().clone(); + format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\ + Content-Length: {}\r\nConnection: close\r\n\r\n{}", + body.len(), + body + ) + } else { + // Spec-shaped error body: the numeric `code` field drives + // client-side error classification. + let error = "{\"code\": 404, \"error\": \"table not found\"}"; + format!( + "HTTP/1.1 404 Not Found\r\nContent-Type: application/json\r\n\ + Content-Length: {}\r\nConnection: close\r\n\r\n{}", + error.len(), + error + ) + }; + let _ = stream.write_all(response.as_bytes()); + } + }); + (endpoint, stop, requests) + } + + #[test] + fn test_namespace_checkout_latest_if_stale_rejects_invalid_arguments() { + unsafe { + let endpoint = CString::new("http://127.0.0.1:1").unwrap(); + let table_id = CString::new("t").unwrap(); + let mut refreshed: *mut c_void = ptr::null_mut(); + // Null output pointer. + assert_eq!( + lance_dataset_namespace_checkout_latest_if_stale( + ptr::null_mut(), + endpoint.as_ptr(), + table_id.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null_mut(), + ptr::null_mut(), + ptr::null_mut(), + ), + -1 + ); + // Null dataset handle. + assert_eq!( + lance_dataset_namespace_checkout_latest_if_stale( + ptr::null_mut(), + endpoint.as_ptr(), + table_id.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null_mut(), + &mut refreshed, + ptr::null_mut(), + ), + -1 + ); + } + } + + #[test] + fn test_namespace_checkout_latest_if_stale_ignores_storage_option_rotation() { + // Opening datasets mutates the process-global debug counters, which + // other tests assert on; serialize with them. + let _counter_guard = crate::ffi::session::debug_counter_test_lock(); + + let base = std::env::temp_dir().join(format!("ffi-ns-opts-{}", rand::random::())); + let table = base.join("t.lance"); + let uri = table.to_string_lossy().to_string(); + external_write(&uri, vec![1, 2, 3], WriteMode::Create); + + let body = Arc::new(Mutex::new(format!("{{\"location\": \"{uri}\"}}"))); + let (endpoint, stop, _requests) = spawn_describe_server(body.clone(), None); + + unsafe { + let endpoint_c = CString::new(endpoint).unwrap(); + let table_id_c = CString::new("t").unwrap(); + + let mut opened_uri: *const c_char = ptr::null(); + let handle = lance_open_dataset_in_namespace( + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + &mut opened_uri, + ); + assert!(!handle.is_null()); + if !opened_uri.is_null() { + crate::error::lance_free_string(opened_uri); + } + assert_eq!(lance_dataset_count_rows(handle), 3); + + // Same location, no external commit: current. + let mut refreshed: *mut c_void = ptr::null_mut(); + let mut moved_uri: *const c_char = ptr::null(); + assert_eq!( + lance_dataset_namespace_checkout_latest_if_stale( + handle, + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null_mut(), + &mut refreshed, + &mut moved_uri, + ), + 0 + ); + assert!(refreshed.is_null()); + assert!(moved_uri.is_null()); + + // The namespace now vends different storage options for the same + // location (e.g. rotated credentials). That must NOT force a + // reopen: the handle's object store refreshes credentials through + // the dynamic storage-options provider installed at open time. + *body.lock().unwrap() = format!( + "{{\"location\": \"{uri}\", \"storage_options\": {{\"test_option\": \"v1\"}}}}" + ); + assert_eq!( + lance_dataset_namespace_checkout_latest_if_stale( + handle, + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null_mut(), + &mut refreshed, + &mut moved_uri, + ), + 0 + ); + assert!(refreshed.is_null()); + assert!(moved_uri.is_null()); + + // Rotated options must not mask a real external commit: the + // manifest-identity fallback still refreshes the handle. + external_write(&uri, vec![4, 5], WriteMode::Append); + assert_eq!( + lance_dataset_namespace_checkout_latest_if_stale( + handle, + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null_mut(), + &mut refreshed, + &mut moved_uri, + ), + 0 + ); + assert!(!refreshed.is_null()); + assert!(moved_uri.is_null()); + assert_eq!(lance_dataset_count_rows(refreshed), 5); + + lance_close_dataset(refreshed); + lance_close_dataset(handle); + } + + *stop.lock().unwrap() = true; + let _ = std::fs::remove_dir_all(base); + } + + #[test] + fn test_namespace_checkout_latest_if_stale_follows_moved_table() { + // Opening datasets mutates the process-global debug counters, which + // other tests assert on; serialize with them. + let _counter_guard = crate::ffi::session::debug_counter_test_lock(); + + let base = + std::env::temp_dir().join(format!("ffi-ns-revalidate-{}", rand::random::())); + let table_a = base.join("a.lance"); + let table_b = base.join("b.lance"); + let uri_a = table_a.to_string_lossy().to_string(); + let uri_b = table_b.to_string_lossy().to_string(); + external_write(&uri_a, vec![1, 2, 3], WriteMode::Create); + external_write(&uri_b, vec![10, 20], WriteMode::Create); + + let body = Arc::new(Mutex::new(format!("{{\"location\": \"{uri_a}\"}}"))); + let (endpoint, stop, _requests) = spawn_describe_server(body.clone(), None); + + unsafe { + let endpoint_c = CString::new(endpoint).unwrap(); + let table_id_c = CString::new("t").unwrap(); + + let mut opened_uri: *const c_char = ptr::null(); + let handle = lance_open_dataset_in_namespace( + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + &mut opened_uri, + ); + assert!(!handle.is_null()); + if !opened_uri.is_null() { + crate::error::lance_free_string(opened_uri); + } + assert_eq!(lance_dataset_count_rows(handle), 3); + + // Same location, no external commit: the cached handle is current. + let mut refreshed: *mut c_void = ptr::null_mut(); + let mut moved_uri: *const c_char = ptr::null(); + assert_eq!( + lance_dataset_namespace_checkout_latest_if_stale( + handle, + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null_mut(), + &mut refreshed, + &mut moved_uri, + ), + 0 + ); + assert!(refreshed.is_null()); + assert!(moved_uri.is_null()); + + // A namespace that reports only `table_uri` (no `location`) must + // also revalidate cleanly instead of failing with "table location + // not found". + *body.lock().unwrap() = format!("{{\"table_uri\": \"{uri_a}\"}}"); + assert_eq!( + lance_dataset_namespace_checkout_latest_if_stale( + handle, + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null_mut(), + &mut refreshed, + &mut moved_uri, + ), + 0 + ); + assert!(refreshed.is_null()); + assert!(moved_uri.is_null()); + *body.lock().unwrap() = format!("{{\"location\": \"{uri_a}\"}}"); + + // Same location, external append: the manifest-identity fallback + // must produce a refreshed handle. + external_write(&uri_a, vec![4, 5], WriteMode::Append); + assert_eq!( + lance_dataset_namespace_checkout_latest_if_stale( + handle, + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null_mut(), + &mut refreshed, + &mut moved_uri, + ), + 0 + ); + assert!(!refreshed.is_null()); + assert!(moved_uri.is_null()); + assert_eq!(lance_dataset_count_rows(refreshed), 5); + let refreshed_same_location = refreshed; + + // The namespace re-points the table to a new location (external + // drop/re-create): revalidation must reopen through the namespace + // and observe the new table, not checkout the old URI. + *body.lock().unwrap() = format!("{{\"location\": \"{uri_b}\"}}"); + let mut moved: *mut c_void = ptr::null_mut(); + assert_eq!( + lance_dataset_namespace_checkout_latest_if_stale( + refreshed_same_location, + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null_mut(), + &mut moved, + &mut moved_uri, + ), + 0 + ); + assert!(!moved.is_null()); + assert_eq!(lance_dataset_count_rows(moved), 2); + assert!(!moved_uri.is_null()); + let moved_uri_str = std::ffi::CStr::from_ptr(moved_uri) + .to_string_lossy() + .to_string(); + assert!(moved_uri_str.contains("b.lance"), "uri: {moved_uri_str}"); + crate::error::lance_free_string(moved_uri); + + lance_close_dataset(moved); + lance_close_dataset(refreshed_same_location); + lance_close_dataset(handle); + } + + *stop.lock().unwrap() = true; + let _ = std::fs::remove_dir_all(base); + } + + #[test] + fn test_namespace_open_and_revalidate_multi_level_table_id() { + // Opening datasets mutates the process-global debug counters, which + // other tests assert on; serialize with them. + let _counter_guard = crate::ffi::session::debug_counter_test_lock(); + + let base = std::env::temp_dir().join(format!("ffi-ns-multi-{}", rand::random::())); + let table_a = base.join("a.lance"); + let table_b = base.join("b.lance"); + let uri_a = table_a.to_string_lossy().to_string(); + let uri_b = table_b.to_string_lossy().to_string(); + external_write(&uri_a, vec![1, 2, 3], WriteMode::Create); + external_write(&uri_b, vec![10, 20], WriteMode::Create); + + // The server only answers for the parsed 3-segment id: a qualified id + // sent as a single segment gets a 404, so every green assertion below + // proves the multi-level identifier went over the wire. + let body = Arc::new(Mutex::new(format!("{{\"location\": \"{uri_a}\"}}"))); + let (endpoint, stop, requests) = + spawn_describe_server(body.clone(), Some(vec!["parent", "child", "tbl"])); + + unsafe { + let endpoint_c = CString::new(endpoint).unwrap(); + let table_id_c = CString::new("parent$child$tbl").unwrap(); + + let mut opened_uri: *const c_char = ptr::null(); + let handle = lance_open_dataset_in_namespace( + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + &mut opened_uri, + ); + assert!(!handle.is_null()); + if !opened_uri.is_null() { + crate::error::lance_free_string(opened_uri); + } + assert_eq!(lance_dataset_count_rows(handle), 3); + + // The open path must issue exactly one describe (the one inside + // `DatasetBuilder::from_namespace`), with the segments joined and + // percent-encoded in the URL path ("$" form-encodes to "%24"). + { + let log = requests.lock().unwrap(); + assert_eq!(log.len(), 1, "open must describe exactly once"); + assert!( + log[0].contains("/v1/table/parent%24child%24tbl/describe"), + "request: {}", + log[0] + ); + } + + // Fresh cache hit: revalidation describes under the multi-level + // id and reports the handle as current. + let mut refreshed: *mut c_void = ptr::null_mut(); + let mut moved_uri: *const c_char = ptr::null(); + assert_eq!( + lance_dataset_namespace_checkout_latest_if_stale( + handle, + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null_mut(), + &mut refreshed, + &mut moved_uri, + ), + 0 + ); + assert!(refreshed.is_null()); + assert!(moved_uri.is_null()); + + // New external commit at the same location: the manifest-identity + // fallback refreshes the handle. + external_write(&uri_a, vec![4, 5], WriteMode::Append); + assert_eq!( + lance_dataset_namespace_checkout_latest_if_stale( + handle, + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null_mut(), + &mut refreshed, + &mut moved_uri, + ), + 0 + ); + assert!(!refreshed.is_null()); + assert!(moved_uri.is_null()); + assert_eq!(lance_dataset_count_rows(refreshed), 5); + let refreshed_same_location = refreshed; + + // Re-point the multi-level table to a new location: revalidation + // reopens through the namespace (again under the 3-segment id) + // and reports the moved URI. + *body.lock().unwrap() = format!("{{\"location\": \"{uri_b}\"}}"); + let mut moved: *mut c_void = ptr::null_mut(); + assert_eq!( + lance_dataset_namespace_checkout_latest_if_stale( + refreshed_same_location, + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null_mut(), + &mut moved, + &mut moved_uri, + ), + 0 + ); + assert!(!moved.is_null()); + assert_eq!(lance_dataset_count_rows(moved), 2); + assert!(!moved_uri.is_null()); + let moved_uri_str = std::ffi::CStr::from_ptr(moved_uri) + .to_string_lossy() + .to_string(); + assert!(moved_uri_str.contains("b.lance"), "uri: {moved_uri_str}"); + crate::error::lance_free_string(moved_uri); + + lance_close_dataset(moved); + lance_close_dataset(refreshed_same_location); + lance_close_dataset(handle); + } + + *stop.lock().unwrap() = true; + let _ = std::fs::remove_dir_all(base); + } + + #[test] + fn test_namespace_revalidate_multi_level_custom_delimiter() { + // Opening datasets mutates the process-global debug counters, which + // other tests assert on; serialize with them. + let _counter_guard = crate::ffi::session::debug_counter_test_lock(); + + let base = std::env::temp_dir().join(format!("ffi-ns-delim-{}", rand::random::())); + let table = base.join("t.lance"); + let uri = table.to_string_lossy().to_string(); + external_write(&uri, vec![1, 2, 3], WriteMode::Create); + + let body = Arc::new(Mutex::new(format!("{{\"location\": \"{uri}\"}}"))); + let (endpoint, stop, _requests) = + spawn_describe_server(body.clone(), Some(vec!["parent", "child", "tbl"])); + + unsafe { + let endpoint_c = CString::new(endpoint).unwrap(); + let table_id_c = CString::new("parent.child.tbl").unwrap(); + let delimiter_c = CString::new(".").unwrap(); + + // Without the matching delimiter the qualified id stays a single + // segment, which does not name the same table: the open must fail. + let mut opened_uri: *const c_char = ptr::null(); + let unsplit = lance_open_dataset_in_namespace( + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), // default "$" delimiter leaves "parent.child.tbl" whole + ptr::null(), + &mut opened_uri, + ); + assert!(unsplit.is_null()); + + // With the configured delimiter the id splits into three segments + // and resolves. + let handle = lance_open_dataset_in_namespace( + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + delimiter_c.as_ptr(), + ptr::null(), + &mut opened_uri, + ); + assert!(!handle.is_null()); + if !opened_uri.is_null() { + crate::error::lance_free_string(opened_uri); + } + assert_eq!(lance_dataset_count_rows(handle), 3); + + // Cache-hit revalidation must parse with the same delimiter. + let mut refreshed: *mut c_void = ptr::null_mut(); + let mut moved_uri: *const c_char = ptr::null(); + assert_eq!( + lance_dataset_namespace_checkout_latest_if_stale( + handle, + endpoint_c.as_ptr(), + table_id_c.as_ptr(), + ptr::null(), + ptr::null(), + delimiter_c.as_ptr(), + ptr::null(), + ptr::null_mut(), + &mut refreshed, + &mut moved_uri, + ), + 0 + ); + assert!(refreshed.is_null()); + assert!(moved_uri.is_null()); + + lance_close_dataset(handle); + } + + *stop.lock().unwrap() = true; + let _ = std::fs::remove_dir_all(base); + } +} diff --git a/rust/ffi/session.rs b/rust/ffi/session.rs index 5f61dcea..20715ea2 100644 --- a/rust/ffi/session.rs +++ b/rust/ffi/session.rs @@ -14,6 +14,18 @@ static DATASET_OPEN_COUNT: AtomicU64 = AtomicU64::new(0); static NAMESPACE_DESCRIBE_COUNT: AtomicU64 = AtomicU64::new(0); static COMMIT_COUNT: AtomicU64 = AtomicU64::new(0); +/// The debug counters above are process-global, so unit tests that either +/// assert on them or mutate them (e.g. by opening datasets through the FFI) +/// must be serialized against each other to stay deterministic under the +/// parallel test runner. +#[cfg(test)] +pub(crate) fn debug_counter_test_lock() -> std::sync::MutexGuard<'static, ()> { + static LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + // A poisoned lock only means another test failed while holding it; the + // guarded state is just the counters, so continuing is safe. + LOCK.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) +} + #[repr(C)] #[derive(Clone, Copy, Debug, Default)] pub struct LanceSessionStats { @@ -180,6 +192,7 @@ mod tests { #[test] fn test_open_dataset_with_session_records_debug_counters() { + let _counter_guard = debug_counter_test_lock(); let dataset_dir = std::env::temp_dir().join(format!("ffi-session-{}", rand::random::())); let uri = dataset_dir.to_string_lossy().to_string(); diff --git a/src/include/lance_ffi.hpp b/src/include/lance_ffi.hpp index 76c8e4ca..1ccc5f29 100644 --- a/src/include/lance_ffi.hpp +++ b/src/include/lance_ffi.hpp @@ -79,6 +79,13 @@ void *lance_open_dataset_in_namespace_with_session( const char *endpoint, const char *table_id, const char *bearer_token, const char *api_key, const char *delimiter, const char *headers_tsv, void *session, const char **out_table_uri); +int32_t lance_dataset_checkout_latest_if_stale(void *dataset, + void **out_new_dataset); +int32_t lance_dataset_namespace_checkout_latest_if_stale( + void *dataset, const char *endpoint, const char *table_id, + const char *bearer_token, const char *api_key, const char *delimiter, + const char *headers_tsv, void *session, void **out_new_dataset, + const char **out_table_uri); void lance_close_dataset(void *dataset); void *lance_get_schema(void *dataset); diff --git a/src/include/lance_scan_bind_data.hpp b/src/include/lance_scan_bind_data.hpp index 44ba942d..fa2b2a2f 100644 --- a/src/include/lance_scan_bind_data.hpp +++ b/src/include/lance_scan_bind_data.hpp @@ -38,6 +38,13 @@ struct LanceScanBindData : public TableFunctionData { optional_idx pushed_limit = optional_idx::Invalid(); idx_t pushed_offset = 0; + // The bind data pins a dataset handle checked out at bind time. Cached + // prepared statements would keep scanning that version forever, bypassing + // the cache revalidation that runs at bind; force a rebind per execution so + // prepared statements observe external commits like ad-hoc statements + // (same approach as DuckDB's multi-file scans). + bool SupportStatementCache() const override { return false; } + ~LanceScanBindData() override; }; diff --git a/src/include/lance_table_entry.hpp b/src/include/lance_table_entry.hpp index f9db4ced..d9ca4a0e 100644 --- a/src/include/lance_table_entry.hpp +++ b/src/include/lance_table_entry.hpp @@ -1,8 +1,11 @@ #pragma once #include "duckdb/catalog/catalog_entry/table_catalog_entry.hpp" +#include "duckdb/parser/parsed_data/alter_table_info.hpp" #include "duckdb/storage/table_storage_info.hpp" +struct ArrowSchema; + namespace duckdb { struct AlterInfo; @@ -31,6 +34,35 @@ struct LanceNamespaceTableConfig { bool IsRest() const { return kind == LanceNamespaceKind::Rest; } }; +// C++-only ALTER marker (never produced by SQL) that asks a LanceTableEntry +// to rebuild itself from the current dataset state; used to replace a stale +// entry through DuckDB's transactional catalog version chain (see +// LanceTableEntry::AlterEntry and LanceSchemaEntry::ReplaceStaleTableEntry). +// It must derive from a stock AlterTableInfo subclass because +// CatalogSet::AlterEntry serializes the info into the undo buffer and the +// commit path deserializes it through the stock registry; the payload must +// also report an empty GetColumnName(), since the commit path casts the old +// entry to DuckTableEntry only for column-affecting alters (see +// CommitState::CommitEntryDrop). SET_PARTITIONED_BY with no keys satisfies +// both: the inherited Serialize() emits a valid stock encoding and the base +// GetColumnName() stays empty; the deserialized stock object is discarded at +// commit. Detected via dynamic_cast in LanceTableEntry::AlterEntry. +// (SET_COMMENT is unsuitable: CatalogSet::AlterEntry special-cases it with +// entry->Copy instead of calling the entry's AlterEntry.) +struct LanceRefreshTableAlterInfo final : SetPartitionedByInfo { + explicit LanceRefreshTableAlterInfo(AlterEntryData data) + : SetPartitionedByInfo(std::move(data), + vector>()) {} + + // Preserve the marker type across Copy() so a copied info still triggers + // the refresh path instead of a stock (rejected) SET_PARTITIONED_BY. + unique_ptr Copy() const override { + auto result = make_uniq(GetAlterEntryData()); + result->allow_internal = allow_internal; + return std::move(result); + } +}; + // LanceTableEntry represents a Lance dataset as a DuckDB base table entry. // It supports scanning via a Lance-backed table scan function and appending via // DuckDB's INSERT planning path (implemented at the catalog level). @@ -51,6 +83,23 @@ class LanceTableEntry final : public TableCatalogEntry { TableFunction GetScanFunction(ClientContext &context, unique_ptr &bind_data) override; + // Validate that this entry's declared schema state (columns, coerced + // columns, NOT NULL constraints) still matches the dataset on storage. + // Statements that bind a scan get this implicitly via GetScanFunction; + // write-only statements (e.g. plain INSERT) must call it before planning. + // On mismatch the stale entry is replaced through the catalog version + // chain (this object stays alive until undo-buffer cleanup, so existing + // references remain valid) and a "changed externally" error is thrown; the + // next access serves the entry rebuilt from the current schema. + void VerifySchemaFreshness(ClientContext &context); + + // Non-throwing-on-mismatch staleness probe for catalog resolution + // (DESCRIBE / information_schema expose entry metadata without binding a + // scan): reports whether the declared schema state diverged from the + // dataset on storage without replacing anything. Infrastructure errors + // (e.g. dataset unreachable) still propagate as exceptions. + bool IsSchemaStale(ClientContext &context); + unique_ptr GetStatistics(ClientContext &, column_t) override { return nullptr; } @@ -79,10 +128,47 @@ class LanceTableEntry final : public TableCatalogEntry { coerced_column_names = std::move(names); } +private: + // Pure comparison of this entry's declared schema state against a live + // dataset schema produced by the shared population pipeline. + bool + MatchesLiveSchemaState(const vector &live_names, + const vector &live_types, + const std::vector &live_coerced_columns, + const ArrowSchema &live_schema_root) const; + + // Fetch the (revalidated) dataset handle and compare; shared by + // IsSchemaStale and VerifySchemaFreshness. + bool FetchLiveSchemaMatches(ClientContext &context, string &out_display_uri); + + // Shared freshness comparator for the scan bind path and + // VerifySchemaFreshness; replaces this entry through the catalog version + // chain and throws on mismatch. + void ValidateLiveSchemaOrReplace( + ClientContext &context, const vector &live_names, + const vector &live_types, + const std::vector &live_coerced_columns, + const ArrowSchema &live_schema_root, const string &display_uri); + private: string dataset_uri; unique_ptr namespace_config; vector coerced_column_names; }; +// Replace a catalog table entry whose declared schema state no longer +// matches the dataset on storage (external schema evolution) with an entry +// rebuilt from the current dataset state. The replacement goes through +// DuckDB's transactional catalog version chain (CatalogSet::AlterEntry with +// the caller's real transaction), so the old generation - `table` itself - +// stays alive, and raw references held by active scans, DML operators, or +// prepared plans stay valid, until normal undo-buffer cleanup reclaims it +// once no active transaction can reference it. Returns false (serving the +// existing entry unchanged) when the entry is not part of a Lance-attached +// schema, was already replaced or dropped, the caller has no real +// transaction to own the undo entry, or the replace lost a write-write +// conflict; the replacement is best-effort and always fail-open. +bool LanceTryReplaceStaleTableEntry(ClientContext &context, + LanceTableEntry &table); + } // namespace duckdb diff --git a/src/lance_dataset_cache.cpp b/src/lance_dataset_cache.cpp index 316acf41..651044f3 100644 --- a/src/lance_dataset_cache.cpp +++ b/src/lance_dataset_cache.cpp @@ -6,6 +6,7 @@ #include "lance_table_entry.hpp" #include "duckdb/common/types/hash.hpp" +#include "duckdb/common/unordered_set.hpp" #include "duckdb/main/client_context_state.hpp" #include @@ -15,16 +16,21 @@ namespace duckdb { static constexpr const char *LANCE_DATASET_CACHE_STATE_KEY = "lance_dataset_cache_state"; +// All methods take the state lock per-call; compound sequences (Get + +// revalidate + Replace, or the memo check followed by MarkRevalidated) run +// unlocked in between. Statement binds are single-threaded, so those +// check-then-act windows are benign; the per-call lock keeps the individual +// container accesses safe against execution-time users of the same +// connection-local state (e.g. maintenance table functions invalidating +// entries from executor threads). class LanceDatasetCacheState final : public ClientContextState { public: shared_ptr Get(const string &key) { lock_guard guard(lock); auto entry = entries.find(key); if (entry == entries.end()) { - query_misses++; return nullptr; } - query_hits++; return entry->second; } @@ -40,28 +46,74 @@ class LanceDatasetCacheState final : public ClientContextState { return entry; } + void Replace(const string &key, shared_ptr entry) { + lock_guard guard(lock); + entries[key] = std::move(entry); + } + + void RecordHit() { + lock_guard guard(lock); + query_hits++; + } + + void RecordMiss() { + lock_guard guard(lock); + query_misses++; + } + + void RecordRevalidation() { + lock_guard guard(lock); + query_revalidations++; + } + + // Statement-scoped revalidation memo. Invariant: within one statement, + // every reference to the same cache key shares one immutable dataset + // generation - the first reference revalidates (or freshly opens) the + // entry and pins it for the statement; later references reuse the pinned + // generation. External commits landing mid-statement are observed by the + // NEXT statement, whose QueryBegin clears the memo. + bool WasRevalidatedThisStatement(const string &key) { + lock_guard guard(lock); + return revalidated_keys.count(key) > 0; + } + + void MarkRevalidatedThisStatement(const string &key) { + lock_guard guard(lock); + revalidated_keys.insert(key); + } + void Invalidate(const string &key) { lock_guard guard(lock); entries.erase(key); + // A key without a cached entry has no pinned generation: the next access + // in this statement takes the miss path and pins whatever it opens. + revalidated_keys.erase(key); } void QueryBegin(ClientContext &) override { lock_guard guard(lock); query_hits = 0; query_misses = 0; + query_revalidations = 0; + revalidated_keys.clear(); } void WriteProfilingInformation(std::ostream &ss) override { lock_guard guard(lock); ss << "Lance Dataset Cache: entries=" << entries.size() - << " hits=" << query_hits << " misses=" << query_misses << "\n"; + << " hits=" << query_hits << " misses=" << query_misses + << " revalidations=" << query_revalidations << "\n"; } private: mutex lock; unordered_map> entries; + // Keys whose entry has already been revalidated (or freshly opened) during + // the current statement; cleared in QueryBegin. + unordered_set revalidated_keys; idx_t query_hits = 0; idx_t query_misses = 0; + idx_t query_revalidations = 0; }; LanceDatasetCacheEntry::LanceDatasetCacheEntry(void *dataset_p, @@ -227,19 +279,113 @@ static void *OpenDirNamespaceDataset(ClientContext &context, const string &root, return dataset; } +// Revalidation callback for cache hits: given the cached dataset handle, +// writes a refreshed handle (or nullptr when the cached handle is current) to +// `out_new_dataset` and optionally a new display URI when the dataset moved. +// Returns 0 on success and -1 on error (details via lance_last_error_*). +using LanceDatasetRevalidateFn = std::function; + +// Default revalidation: compare the cached handle's manifest identity with the +// latest committed manifest at the dataset's resolved URI. +static int32_t RevalidateDatasetHandle(void *dataset, void **out_new_dataset, + string &out_new_display_uri) { + out_new_display_uri.clear(); + return lance_dataset_checkout_latest_if_stale(dataset, out_new_dataset); +} + +// Revalidate a cached dataset against the latest committed version. Writes +// through this connection invalidate the touched entries eagerly, but commits +// made by external writers (another process or another connection) would +// otherwise never be observed and the cached entry would serve stale data +// forever. The check is metadata-only (it resolves the latest manifest +// version) and therefore cheap relative to a scan, so it is performed on the +// first cache hit per key per statement: correctness of latest-intent reads +// takes precedence over the micro-cost of one version lookup per query, +// while the statement-scoped memo (see GetOrOpenDatasetCacheEntry) keeps all +// references within one statement on a single dataset generation. +// Returns the (possibly refreshed) entry, or throws if the latest version +// cannot be determined. +static shared_ptr RevalidateDatasetCacheEntry( + LanceDatasetCacheState &state, const string &cache_key, + shared_ptr entry, + const LanceDatasetRevalidateFn &revalidate, bool &out_refreshed) { + // Counted even when the check fails below: the version lookup did run. + // The counter makes the statement-scoped memoization observable in tests + // (a self-join must report one revalidation, not one per reference). + state.RecordRevalidation(); + void *refreshed_dataset = nullptr; + string refreshed_display_uri; + if (revalidate(entry->Handle(), &refreshed_dataset, refreshed_display_uri) != + 0) { + // The entry can no longer be trusted (e.g. the dataset was dropped by an + // external writer); drop it so the next access retries a fresh open. + state.Invalidate(cache_key); + throw IOException("Failed to check latest version of Lance dataset: " + + entry->DisplayUri() + LanceFormatErrorSuffix()); + } + if (!refreshed_dataset) { + // Already at the latest committed version. + out_refreshed = false; + return entry; + } + // A newer version was committed externally: replace the cached entry with + // the refreshed handle. The old entry stays alive for existing holders via + // shared ownership. The display URI only changes when the revalidation + // resolved a new physical location (namespace-backed tables). + auto refreshed_entry = make_shared_ptr( + refreshed_dataset, refreshed_display_uri.empty() + ? entry->DisplayUri() + : std::move(refreshed_display_uri)); + state.Replace(cache_key, refreshed_entry); + out_refreshed = true; + return refreshed_entry; +} + static shared_ptr GetOrOpenDatasetCacheEntry( ClientContext &context, const string &cache_key, const std::function()> &open_dataset, - bool *out_cache_hit) { + bool *out_cache_hit, + const LanceDatasetRevalidateFn &revalidate = RevalidateDatasetHandle) { auto state = GetOrCreateLanceDatasetCacheState(context); auto entry = state->Get(cache_key); if (entry) { + // Statement-scoped memoization: revalidating independently on every hit + // would let two references to the same table in one statement (e.g. a + // self-join's two binds) pin different dataset versions if an external + // commit lands between their binds. The first reference in a statement + // revalidates and pins the entry for the key; every later reference in + // the same statement shares that one immutable dataset generation. + // External commits landing mid-statement are observed by the NEXT + // statement, which starts with a cleared memo. + if (state->WasRevalidatedThisStatement(cache_key)) { + state->RecordHit(); + if (out_cache_hit) { + *out_cache_hit = true; + } + return entry; + } + bool refreshed = false; + entry = RevalidateDatasetCacheEntry(*state, cache_key, std::move(entry), + revalidate, refreshed); + // Mark only after a successful revalidation: the failure path above + // drops the entry and throws, and the next access must retry a fresh + // open instead of trusting a memo for an entry that no longer exists. + state->MarkRevalidatedThisStatement(cache_key); + // A stale hit had to be refreshed from storage, so report it as a miss + // both in the profiling counters and in the per-scan cache-hit flag. + if (refreshed) { + state->RecordMiss(); + } else { + state->RecordHit(); + } if (out_cache_hit) { - *out_cache_hit = true; + *out_cache_hit = !refreshed; } return entry; } + state->RecordMiss(); auto opened = open_dataset(); if (!opened) { return nullptr; @@ -247,7 +393,13 @@ static shared_ptr GetOrOpenDatasetCacheEntry( if (out_cache_hit) { *out_cache_hit = false; } - return state->PutOrGetExisting(cache_key, opened); + auto result = state->PutOrGetExisting(cache_key, opened); + // A dataset opened by this statement is already the statement's pinned + // generation for its key: later references in the same statement must + // reuse it rather than revalidate it against storage (which could observe + // a newer external commit and split the statement across two versions). + state->MarkRevalidatedThisStatement(cache_key); + return result; } shared_ptr @@ -280,6 +432,32 @@ shared_ptr LanceGetOrOpenDatasetEntryInNamespace( const string &headers_tsv, string &out_display_uri, bool *out_cache_hit) { auto cache_key = LanceBuildNamespaceDatasetCacheKey( endpoint, table_id, bearer_token, api_key, delimiter, headers_tsv); + // Namespace tables are cached by endpoint/table id rather than by physical + // URI, so cache hits must be revalidated through the namespace: an external + // drop/re-create can re-point the table to a new location, which a checkout + // against the already-resolved handle would never observe. The re-describe + // costs one namespace round trip per hit — the same order as the namespace + // open path itself, which also starts with a describe. + auto namespace_revalidate = [&](void *dataset, void **out_new_dataset, + string &out_new_display_uri) -> int32_t { + out_new_display_uri.clear(); + const char *bearer_ptr = + bearer_token.empty() ? nullptr : bearer_token.c_str(); + const char *api_key_ptr = api_key.empty() ? nullptr : api_key.c_str(); + const char *delimiter_ptr = delimiter.empty() ? nullptr : delimiter.c_str(); + const char *headers_ptr = + headers_tsv.empty() ? nullptr : headers_tsv.c_str(); + const char *uri_ptr = nullptr; + auto rc = lance_dataset_namespace_checkout_latest_if_stale( + dataset, endpoint.c_str(), table_id.c_str(), bearer_ptr, api_key_ptr, + delimiter_ptr, headers_ptr, LanceGetSessionHandle(context), + out_new_dataset, &uri_ptr); + if (uri_ptr) { + out_new_display_uri = uri_ptr; + lance_free_string(uri_ptr); + } + return rc; + }; auto entry = GetOrOpenDatasetCacheEntry( context, cache_key, [&]() { @@ -296,7 +474,7 @@ shared_ptr LanceGetOrOpenDatasetEntryInNamespace( return make_shared_ptr(dataset, std::move(display_uri)); }, - out_cache_hit); + out_cache_hit, namespace_revalidate); if (entry) { out_display_uri = entry->DisplayUri(); } else { diff --git a/src/lance_insert.cpp b/src/lance_insert.cpp index 6f0e0c08..953b95e9 100644 --- a/src/lance_insert.cpp +++ b/src/lance_insert.cpp @@ -229,6 +229,15 @@ PhysicalOperator &PlanLanceInsertAppend(ClientContext &context, if (!lance_table) { throw InternalException("PlanLanceInsertAppend called for non-Lance table"); } + // Plain INSERT does not bind a scan of the target, so the scan-bind + // freshness check never runs for it; validate here so a stale entry (e.g. + // a coerced-column list outdated by an external type evolution) fails + // closed instead of gating the write on stale state. Throws and replaces + // the entry on mismatch. This is defense-in-depth rather than the only + // guard for prepared INSERT: LanceDuckCatalog::GetCatalogVersion opts the + // catalog out of prepared-plan reuse, so every EXECUTE rebinds and reaches + // this check with a freshly resolved entry instead of a cached plan. + lance_table->VerifySchemaFreshness(context); if (lance_table->HasCoercedColumns()) { throw NotImplementedException( "INSERT into Lance table '" + lance_table->name + diff --git a/src/lance_scan.cpp b/src/lance_scan.cpp index 3d9c3e2f..50202194 100644 --- a/src/lance_scan.cpp +++ b/src/lance_scan.cpp @@ -514,6 +514,11 @@ struct LanceExecBindData : public TableFunctionData { ArrowTableSchema arrow_table; vector names; vector types; + + // Pins a dataset handle checked out at bind time; force a rebind per + // execution so prepared statements observe external commits (see + // LanceScanBindData::SupportStatementCache). + bool SupportStatementCache() const override { return false; } }; static bool LanceSupportsPushdownType(const FunctionData &bind_data, @@ -3421,6 +3426,17 @@ LanceTableEntry::AlterEntry(CatalogTransaction transaction, AlterInfo &info) { throw InternalException( "LanceTableEntry::AlterEntry missing client context"); } + // Internal stale-entry refresh marker (see ReplaceStaleTableEntry in + // lance_storage.cpp): rebuild this entry from the current dataset state + // without touching the dataset itself. CatalogSet::AlterEntry chains the + // rebuilt entry over this one in the version chain, so this (old) + // generation stays alive until undo-buffer cleanup and holders of raw + // references to it remain safe. Bypasses the autocommit gate below on + // purpose: the refresh is a catalog-only, fully transactional operation + // (rollback restores this entry), unlike Lance dataset DDL. + if (dynamic_cast(&info)) { + return BuildUpdatedLanceTableEntry(*transaction.context, *this, internal); + } return AlterEntry(*transaction.context, info); } @@ -3651,6 +3667,162 @@ unique_ptr LanceTableEntry::Copy(ClientContext &context) const { return unique_ptr_cast(std::move(copy)); } +// Collect the logical column indexes carrying NOT NULL constraints. Both the +// entry side (constraints built from Arrow flags during discovery/rebuild) +// and the live side (Arrow flags of the coerced schema) are produced by the +// same population pipeline, so the resulting index sets are comparable. +static vector NotNullIndexesFromConstraints( + const vector> &constraints) { + vector indexes; + for (auto &constraint : constraints) { + if (constraint->type == ConstraintType::NOT_NULL) { + indexes.push_back(constraint->Cast().index.index); + } + } + std::sort(indexes.begin(), indexes.end()); + return indexes; +} + +static vector +NotNullIndexesFromArrowSchema(const ArrowSchema &schema_root) { + vector indexes; + for (int64_t child_idx = 0; child_idx < schema_root.n_children; child_idx++) { + auto *child = schema_root.children[child_idx]; + if (child && (child->flags & ARROW_FLAG_NULLABLE) == 0) { + indexes.push_back(NumericCast(child_idx)); + } + } + return indexes; +} + +// External writers can evolve the dataset schema behind this catalog entry +// (same-connection ALTERs rebuild the entry via AlterEntry, but external +// commits do not). DuckDB binds column ids against this entry's columns +// while scans and writers resolve state through the live dataset schema; +// serving a mismatched pair would mislabel columns, read the wrong fields, +// or apply stale write gates. Fail closed instead: replace the stale entry +// through the catalog version chain (the same mechanism ALTER TABLE uses) +// so the next access serves the entry rebuilt from the current schema, and +// surface an explicit error for this statement. +// +// The comparison covers, beyond top-level names/types: +// - the coerced-column state: Arrow types that the reader-boundary layer +// maps to the same DuckDB type (e.g. float16 and float32 both surface as +// FLOAT) are indistinguishable by name/type, yet the entry's +// coerced-column list gates writes (INSERT/UPDATE/MERGE reject coerced +// columns); +// - the nullability state: external ALTER SET/DROP NOT NULL changes only +// the Arrow flags, which the entry mirrors as NOT NULL constraints. +// All compared states come from the same population pipeline over the +// (entry-build vs live) schema, so the comparison is exact. +bool LanceTableEntry::MatchesLiveSchemaState( + const vector &live_names, const vector &live_types, + const std::vector &live_coerced_columns, + const ArrowSchema &live_schema_root) const { + bool schema_matches = columns.LogicalColumnCount() == live_names.size() && + CoercedColumnNames() == live_coerced_columns && + NotNullIndexesFromConstraints(constraints) == + NotNullIndexesFromArrowSchema(live_schema_root); + if (schema_matches) { + idx_t col_idx = 0; + for (auto &col : columns.Logical()) { + if (col.Name() != live_names[col_idx] || + col.Type() != live_types[col_idx]) { + schema_matches = false; + break; + } + col_idx++; + } + } + return schema_matches; +} + +void LanceTableEntry::ValidateLiveSchemaOrReplace( + ClientContext &context, const vector &live_names, + const vector &live_types, + const std::vector &live_coerced_columns, + const ArrowSchema &live_schema_root, const string &display_uri) { + if (MatchesLiveSchemaState(live_names, live_types, live_coerced_columns, + live_schema_root)) { + return; + } + // The replace supersedes this entry in the catalog version chain, but this + // (old) generation stays alive until undo-buffer cleanup, so reading + // members of `this` for the error message below remains safe. The heal is + // best-effort (fail-open); the error is thrown regardless so this + // statement never runs against a mismatched binding. + (void)LanceTryReplaceStaleTableEntry(context, *this); + throw CatalogException( + "Lance table \"%s\" was changed externally: the dataset schema at " + "'%s' no longer matches the catalog entry. The entry has been " + "refreshed from the current schema - please re-run the query.", + name, display_uri); +} + +// Fetch the (revalidated) dataset handle and compare its live schema state +// against this entry. Returns true when the entry is current. Throws on +// infrastructure errors (dataset unreachable etc.). +bool LanceTableEntry::FetchLiveSchemaMatches(ClientContext &context, + string &out_display_uri) { + auto entry = + LanceGetOrOpenDatasetEntryForTable(context, *this, out_display_uri); + auto *dataset = entry ? entry->Handle() : nullptr; + if (!dataset) { + throw IOException("Failed to open Lance dataset: " + out_display_uri + + LanceFormatErrorSuffix()); + } + + auto *schema_handle = lance_get_schema(dataset); + if (!schema_handle) { + throw IOException("Failed to get schema from Lance dataset: " + + out_display_uri + LanceFormatErrorSuffix()); + } + ArrowSchemaWrapper schema_root; + memset(&schema_root.arrow_schema, 0, sizeof(schema_root.arrow_schema)); + if (lance_schema_to_arrow(schema_handle, &schema_root.arrow_schema) != 0) { + lance_free_schema(schema_handle); + throw IOException( + "Failed to export Lance schema to Arrow C Data Interface" + + LanceFormatErrorSuffix()); + } + lance_free_schema(schema_handle); + auto live_coerced_columns = + LanceCoerceArrowSchemaForDuckDB(&schema_root.arrow_schema); + ArrowTableSchema arrow_table; + ArrowTableFunction::PopulateArrowTableSchema(context, arrow_table, + schema_root.arrow_schema); + return MatchesLiveSchemaState(arrow_table.GetNames(), arrow_table.GetTypes(), + live_coerced_columns, schema_root.arrow_schema); +} + +// Non-throwing-on-mismatch staleness probe for catalog resolution: reports +// whether this entry's declared schema state diverged from the dataset on +// storage without replacing anything. Infrastructure errors still propagate. +bool LanceTableEntry::IsSchemaStale(ClientContext &context) { + string display_uri; + return !FetchLiveSchemaMatches(context, display_uri); +} + +// Freshness validation entry point for statements that do not bind a scan of +// the table (e.g. plain INSERT): fetches the (revalidated) dataset handle and +// compares its live schema state against this entry, replacing the entry and +// failing closed on mismatch just like the scan bind path. +void LanceTableEntry::VerifySchemaFreshness(ClientContext &context) { + string display_uri; + if (FetchLiveSchemaMatches(context, display_uri)) { + return; + } + // As in ValidateLiveSchemaOrReplace: the superseded generation (`this`) + // stays alive in the version chain until undo-buffer cleanup, so using its + // members for the error message remains safe after the best-effort heal. + (void)LanceTryReplaceStaleTableEntry(context, *this); + throw CatalogException( + "Lance table \"%s\" was changed externally: the dataset schema at " + "'%s' no longer matches the catalog entry. The entry has been " + "refreshed from the current schema - please re-run the query.", + name, display_uri); +} + TableFunction LanceTableEntry::GetScanFunction(ClientContext &context, unique_ptr &bind_data) { @@ -3685,12 +3857,17 @@ LanceTableEntry::GetScanFunction(ClientContext &context, LanceFormatErrorSuffix()); } lance_free_schema(schema_handle); - LanceCoerceArrowSchemaForDuckDB(&result->schema_root.arrow_schema); + auto live_coerced_columns = + LanceCoerceArrowSchemaForDuckDB(&result->schema_root.arrow_schema); ArrowTableFunction::PopulateArrowTableSchema( context, result->arrow_table, result->schema_root.arrow_schema); result->names = result->arrow_table.GetNames(); result->types = result->arrow_table.GetTypes(); + ValidateLiveSchemaOrReplace( + context, result->names, result->types, live_coerced_columns, + result->schema_root.arrow_schema, result->file_path); + auto *scan_schema_handle = lance_get_schema_for_scan(result->dataset); if (!scan_schema_handle) { throw IOException("Failed to get scan schema from Lance dataset: " + diff --git a/src/lance_search.cpp b/src/lance_search.cpp index 1b52227b..d394232c 100644 --- a/src/lance_search.cpp +++ b/src/lance_search.cpp @@ -335,6 +335,11 @@ struct LanceKnnBindData : public TableFunctionData { vector types; vector lance_pushed_filter_ir_parts; + + // Pins a dataset handle checked out at bind time; force a rebind per + // execution so prepared statements observe external commits (see + // LanceScanBindData::SupportStatementCache). + bool SupportStatementCache() const override { return false; } }; struct LanceKnnGlobalState : public GlobalTableFunctionState { @@ -1029,6 +1034,11 @@ struct LanceSearchBindData : public TableFunctionData { ArrowTableSchema arrow_table; vector names; vector types; + + // Pins a dataset handle checked out at bind time; force a rebind per + // execution so prepared statements observe external commits (see + // LanceScanBindData::SupportStatementCache). + bool SupportStatementCache() const override { return false; } }; struct LanceSearchGlobalState : public GlobalTableFunctionState { diff --git a/src/lance_storage.cpp b/src/lance_storage.cpp index 1edb73f6..34443201 100644 --- a/src/lance_storage.cpp +++ b/src/lance_storage.cpp @@ -9,6 +9,8 @@ #include "duckdb/catalog/default/default_schemas.hpp" #include "duckdb/catalog/duck_catalog.hpp" #include "duckdb/common/arrow/arrow_converter.hpp" +#include "duckdb/common/exception/catalog_exception.hpp" +#include "duckdb/common/exception/transaction_exception.hpp" #include "duckdb/common/exception_format_value.hpp" #include "duckdb/common/file_system.hpp" #include "duckdb/common/string_util.hpp" @@ -25,6 +27,7 @@ #include "duckdb/parser/parsed_data/copy_info.hpp" #include "duckdb/parser/parsed_data/create_schema_info.hpp" #include "duckdb/parser/parsed_data/create_table_info.hpp" +#include "duckdb/parser/constraints/not_null_constraint.hpp" #include "duckdb/parser/parsed_data/create_view_info.hpp" #include "duckdb/parser/parsed_data/drop_info.hpp" #include "duckdb/planner/operator/logical_create_table.hpp" @@ -147,9 +150,9 @@ static string GetLanceNamespaceHeaders(const AttachInfo &info) { return headers_tsv; } -static void PopulateColumnsFromArrowSchema(ClientContext &context, - ArrowSchema &arrow_schema, - ColumnList &out_columns) { +static void PopulateColumnsFromArrowSchema( + ClientContext &context, ArrowSchema &arrow_schema, ColumnList &out_columns, + vector> *out_constraints = nullptr) { ArrowTableSchema arrow_table; ArrowTableFunction::PopulateArrowTableSchema(context, arrow_table, arrow_schema); @@ -161,12 +164,22 @@ static void PopulateColumnsFromArrowSchema(ClientContext &context, } for (idx_t i = 0; i < names.size(); i++) { out_columns.AddColumn(ColumnDefinition(names[i], types[i])); + + // Reflect not-null constraints from the Arrow flags so discovered entries + // carry the same nullability state as entries rebuilt after ALTER (see + // PopulateLanceTableSchemaFromDataset); the schema freshness check + // compares this state against the live dataset schema. + auto *child = arrow_schema.children[i]; + if (out_constraints && child && (child->flags & ARROW_FLAG_NULLABLE) == 0) { + out_constraints->push_back(make_uniq(LogicalIndex(i))); + } } } static void PopulateLanceTableColumnsFromDataset( ClientContext &context, void *dataset, ColumnList &out_columns, - vector *out_coerced_columns = nullptr) { + vector *out_coerced_columns = nullptr, + vector> *out_constraints = nullptr) { auto *schema_handle = lance_get_schema(dataset); if (!schema_handle) { throw IOException("Failed to get schema from Lance dataset" + @@ -186,8 +199,8 @@ static void PopulateLanceTableColumnsFromDataset( if (out_coerced_columns) { *out_coerced_columns = std::move(coerced); } - PopulateColumnsFromArrowSchema(context, schema_root.arrow_schema, - out_columns); + PopulateColumnsFromArrowSchema(context, schema_root.arrow_schema, out_columns, + out_constraints); } static string JoinNamespacePath(const string &root, const string &child) { @@ -313,7 +326,7 @@ class LanceDirectoryDefaultGenerator : public DefaultGenerator { vector coerced; try { PopulateLanceTableColumnsFromDataset(context, dataset, info.columns, - &coerced); + &coerced, &info.constraints); } catch (...) { lance_close_dataset(dataset); return nullptr; @@ -376,7 +389,8 @@ class LanceDirectoryDefaultGenerator : public DefaultGenerator { static void PopulateLanceTableColumnsFromJsonSchema( ClientContext &context, const string &schema_json, ColumnList &out_columns, - vector *out_coerced_columns = nullptr) { + vector *out_coerced_columns = nullptr, + vector> *out_constraints = nullptr) { ArrowSchemaWrapper schema_root; memset(&schema_root.arrow_schema, 0, sizeof(schema_root.arrow_schema)); if (lance_json_arrow_schema_to_c(schema_json.c_str(), @@ -389,8 +403,8 @@ static void PopulateLanceTableColumnsFromJsonSchema( if (out_coerced_columns) { *out_coerced_columns = std::move(coerced); } - PopulateColumnsFromArrowSchema(context, schema_root.arrow_schema, - out_columns); + PopulateColumnsFromArrowSchema(context, schema_root.arrow_schema, out_columns, + out_constraints); } class LanceRestNamespaceDefaultGenerator : public DefaultGenerator { @@ -471,8 +485,8 @@ class LanceRestNamespaceDefaultGenerator : public DefaultGenerator { info.on_conflict = OnCreateConflict::IGNORE_ON_CONFLICT; vector coerced; try { - PopulateLanceTableColumnsFromJsonSchema(context, schema_json, - info.columns, &coerced); + PopulateLanceTableColumnsFromJsonSchema( + context, schema_json, info.columns, &coerced, &info.constraints); } catch (...) { continue; // Schema conversion failed, try next candidate. } @@ -504,7 +518,7 @@ class LanceRestNamespaceDefaultGenerator : public DefaultGenerator { vector coerced; try { PopulateLanceTableColumnsFromDataset(context, dataset, info.columns, - &coerced); + &coerced, &info.constraints); } catch (...) { lance_close_dataset(dataset); continue; @@ -663,6 +677,24 @@ class LanceSchemaEntry final : public DuckSchemaEntry { table_default_generator = generator; } + // Non-transactional CatalogTransaction for direct catalog surgery on this + // ephemeral in-memory catalog (the user ALTER/DROP paths, where DuckDB's + // transactional TABLE_ENTRY machinery would cast entries to DuckTableEntry + // at commit and therefore cannot be used). GetSystemTransaction() has + // start_time 1 and thus reports a write-write conflict against any entry + // committed by a real transaction - commit ids exceed 1 - which is exactly + // what stale-entry heals produce (ReplaceStaleTableEntry commits through + // the caller's transaction). Use a start time that treats every committed + // entry as current; genuinely uncommitted heads still conflict and are + // rejected up-front by the pending-heal guards in Alter and DropEntry. + // Entries written under this transaction carry timestamp 1 (committed, + // visible to everyone immediately), matching the catalog's pre-existing + // system-surgery semantics. + CatalogTransaction GetSurgeryTransaction() { + return CatalogTransaction(catalog.GetDatabase(), /*transaction_id_p=*/1, + /*start_time_p=*/TRANSACTION_ID_START - 1); + } + void Alter(CatalogTransaction transaction, AlterInfo &info) override { auto &set = GetCatalogSet(info.GetCatalogType()); auto entry = set.GetEntry(transaction, info.name); @@ -680,6 +712,22 @@ class LanceSchemaEntry final : public DuckSchemaEntry { "Lance DDL does not support explicit transactions yet"); } + // The catalog surgery below runs under the non-transactional surgery + // transaction, which cannot modify a version chain whose head is an + // uncommitted entry - exactly the state the stale-entry heal + // (ReplaceStaleTableEntry) leaves behind when this statement's own bind + // refreshed the entry after an external change. Fail closed before any + // dataset side effects; any successful statement on the table commits + // the refreshed entry, after which the ALTER goes through. + if (set.GetEntry(GetSurgeryTransaction(), info.name).get() != entry.get()) { + throw TransactionException( + "Cannot alter Lance table \"%s\": its catalog entry was refreshed " + "in the current transaction after an external change and the " + "refresh has not committed yet. Run a query on the table first and " + "retry the ALTER.", + info.name); + } + // Allow altering internal entries for attached Lance catalogs. info.allow_internal = true; @@ -710,23 +758,95 @@ class LanceSchemaEntry final : public DuckSchemaEntry { LanceInvalidateDatasetCacheForTable(context, *lance_entry); } - auto system_tx = - CatalogTransaction::GetSystemTransaction(catalog.GetDatabase()); - system_tx.context = &context; + auto surgery_tx = GetSurgeryTransaction(); + surgery_tx.context = &context; if (info.type == AlterType::CHANGE_OWNERSHIP) { - if (!set.AlterOwnership(system_tx, info.Cast())) { + if (!set.AlterOwnership(surgery_tx, info.Cast())) { throw CatalogException("Couldn't change ownership!"); } return; } - if (!set.AlterEntry(system_tx, info.name, info)) { + if (!set.AlterEntry(surgery_tx, info.name, info)) { throw CatalogException::MissingEntry(info.GetCatalogType(), info.name, string()); } } + // Catalog-only reads (DESCRIBE, information_schema/duckdb_columns) expose + // the entry's column and NOT NULL metadata without ever binding a scan or + // planning DML, so the freshness check must already run when the entry is + // resolved. When the entry is stale it is replaced through the catalog + // version chain and the lookup retried once so the caller transparently + // receives the freshly rebuilt entry instead of an error. The staleness + // probe costs one dataset-cache access per resolution - the same order as + // the per-hit revalidation that scans already perform at bind. + optional_ptr + RefreshStaleTableEntry(CatalogTransaction transaction, + const EntryLookupInfo &lookup_info, + optional_ptr entry) { + if (!entry || lookup_info.GetCatalogType() != CatalogType::TABLE_ENTRY || + !transaction.context) { + return entry; + } + auto *lance_entry = dynamic_cast(entry.get()); + if (!lance_entry) { + return entry; + } + bool replaced = false; + try { + if (!lance_entry->IsSchemaStale(*transaction.context)) { + return entry; + } + replaced = ReplaceStaleTableEntry(transaction, *lance_entry); + } catch (...) { + // Fail open for catalog resolution: an unreachable dataset must not + // block metadata access or DROP TABLE cleanup; scans surface the real + // error with proper context. (Covers both the staleness probe and the + // entry rebuild inside the replace, which also does storage I/O.) + return entry; + } + if (!replaced) { + return entry; + } + // `entry`/`lance_entry` now point at the superseded generation, which + // stays alive in the version chain until undo-buffer cleanup. Re-lookup + // to serve the rebuilt entry; no retry loop - if the rebuilt entry is + // somehow still stale, the scan-bind check fails closed. + return DuckSchemaEntry::LookupEntry(transaction, lookup_info); + } + + optional_ptr + LookupEntry(CatalogTransaction transaction, + const EntryLookupInfo &lookup_info) override { + auto entry = DuckSchemaEntry::LookupEntry(transaction, lookup_info); + return RefreshStaleTableEntry(transaction, lookup_info, entry); + } + + void Scan(ClientContext &context, CatalogType type, + const std::function &callback) override { + if (type == CatalogType::TABLE_ENTRY) { + // Enumerations (information_schema, duckdb_columns, SHOW) also expose + // entry metadata. Collect the names first: the staleness probe does + // storage I/O and the stale-entry replace mutates the catalog set, + // neither of which may run under the set's scan lock. + vector table_names; + DuckSchemaEntry::Scan(context, type, [&](CatalogEntry &scanned) { + if (dynamic_cast(&scanned)) { + table_names.push_back(scanned.name); + } + }); + auto transaction = GetCatalogTransaction(context); + for (auto &table_name : table_names) { + EntryLookupInfo lookup_info(CatalogType::TABLE_ENTRY, table_name); + auto entry = DuckSchemaEntry::LookupEntry(transaction, lookup_info); + (void)RefreshStaleTableEntry(transaction, lookup_info, entry); + } + } + DuckSchemaEntry::Scan(context, type, callback); + } + void DropEntry(ClientContext &context, DropInfo &info) override { if (info.type != CatalogType::TABLE_ENTRY) { DuckSchemaEntry::DropEntry(context, info); @@ -751,6 +871,25 @@ class LanceSchemaEntry final : public DuckSchemaEntry { auto cache_key = LanceBuildDatasetCacheKeyForTable(context, *lance_entry); auto existing_type = existing_entry->type; + // The catalog part of DROP below runs under the non-transactional + // surgery transaction (see the comment there), which cannot modify a + // version chain whose head is an uncommitted entry - exactly the state + // the stale-entry heal (ReplaceStaleTableEntry) leaves behind when this + // DROP's own bind refreshed the entry after an external change. Fail + // closed BEFORE the dataset is deleted: proceeding would delete the data + // and then abort on the catalog conflict, wedging the entry. Any + // successful statement on the table commits the refreshed entry, after + // which DROP goes through. + if (set.GetEntry(GetSurgeryTransaction(), info.name).get() != + existing_entry.get()) { + throw TransactionException( + "Cannot drop Lance table \"%s\": its catalog entry was refreshed " + "in the current transaction after an external change and the " + "refresh has not committed yet. Run a query on the table first and " + "retry the DROP.", + info.name); + } + if (rest_ns) { unordered_map overrides; if (!rest_ns->bearer_token_override.empty()) { @@ -841,27 +980,26 @@ class LanceSchemaEntry final : public DuckSchemaEntry { // Note: TABLE_ENTRY and VIEW_ENTRY share the same underlying catalog set. // DuckDB's transactional DROP path assumes TABLE_ENTRY is always a // DuckTableEntry (and will fail for extension-backed TableCatalogEntry - // implementations). To avoid that, perform the catalog drop using a system - // (non-transactional) CatalogTransaction. + // implementations). To avoid that, perform the catalog drop using the + // non-transactional surgery CatalogTransaction. if (existing_type != CatalogType::TABLE_ENTRY && existing_type != CatalogType::VIEW_ENTRY) { throw InternalException( "Unexpected catalog entry type for DROP TABLE: %s", CatalogTypeToString(existing_type)); } - auto system_transaction = - CatalogTransaction::GetSystemTransaction(catalog.GetDatabase()); - if (!set.DropEntry(system_transaction, info.name, info.cascade, true)) { + auto surgery_transaction = GetSurgeryTransaction(); + if (!set.DropEntry(surgery_transaction, info.name, info.cascade, true)) { throw InternalException( "Could not drop element because of an internal error"); } - // DropEntry with a system (committed) CatalogTransaction leaves a committed - // tombstone behind. This blocks subsequent lazy discovery of a recreated - // dataset with the same name, because CatalogSet::GetEntryDetailed will - // find the tombstone and never consult the default generator. Since ATTACH - // TYPE LANCE catalogs are ephemeral, we can eagerly clean up the entry - // chain (old entry + tombstone). + // DropEntry with a non-transactional (committed) CatalogTransaction leaves + // a committed tombstone behind. This blocks subsequent lazy discovery of a + // recreated dataset with the same name, because + // CatalogSet::GetEntryDetailed will find the tombstone and never consult + // the default generator. Since ATTACH TYPE LANCE catalogs are ephemeral, + // we can eagerly clean up the entry chain (old entry + tombstone). set.CleanupEntry(*existing_entry); LanceInvalidateDatasetCache(context, cache_key); @@ -1089,6 +1227,89 @@ class LanceSchemaEntry final : public DuckSchemaEntry { return nullptr; } + // Replace a table entry whose declared columns no longer match the dataset + // schema on storage with an entry rebuilt from the current dataset state. + // A drop-and-cleanup (the DROP TABLE mechanism) would destroy the stale + // object immediately, but active scans, DML operators and prepared plans + // may still hold raw LanceTableEntry references. Instead the entry is + // replaced through DuckDB's transactional catalog version chain: + // CatalogSet::AlterEntry hands the refresh marker to + // LanceTableEntry::AlterEntry (which rebuilds the entry from the live + // dataset), chains the rebuilt entry over the stale one, and pushes the old + // generation into the caller's undo buffer, so undo-buffer cleanup reclaims + // it only once no active transaction can reference it. The marker + // serializes as a stock ALTER TABLE payload with an empty column name, + // which the commit path treats as metadata-only (no DuckTableEntry cast; + // see CommitState::CommitEntryDrop), and this attached catalog is + // in-memory, so no WAL is involved. Rollback restores the old generation + // generically. + // + // NOTE: until the caller's transaction commits, the refreshed entry is an + // uncommitted head on the version chain, which the system-transaction + // catalog surgery of user-issued ALTER/DROP cannot run on top of; those + // paths detect the pending refresh and fail closed before any dataset + // side effects (see the guards in Alter and DropEntry above). + bool ReplaceStaleTableEntry(CatalogTransaction transaction, + LanceTableEntry &table) { + if (!transaction.transaction) { + // Without a real transaction there is no undo buffer: + // CatalogSet::AlterEntry would destroy the old generation immediately + // (the TakeChild path), reintroducing the use-after-free this replace + // exists to avoid. Serve the current entry unchanged; a later + // transactional access heals it. + return false; + } + auto &set = GetCatalogSet(CatalogType::TABLE_ENTRY); + auto existing_entry = set.GetEntry(transaction, table.name); + if (!existing_entry || existing_entry.get() != &table) { + // Already replaced or dropped concurrently; nothing to refresh. + return false; + } + // The replace pushes the superseded generation into this transaction's + // undo buffer, which DuckDB only allows on transactions marked + // read-write (DuckTransactionManager::PushCatalogEntry fails closed + // otherwise). The heal typically runs while BINDING a read statement, + // before anything has marked this catalog's transaction as writing, so + // upgrade it here. Deliberately NOT MetaTransaction::ModifyDatabase: the + // heal is an internal repair of this in-memory catalog mirror, not a + // user write - it must neither claim the meta transaction's single + // writable-database slot (that would break statements that read a stale + // Lance table while writing to another attached database) nor be + // rejected in read-only meta transactions. Commit and rollback of the + // undo entry are WAL-free for this in-memory catalog. + if (transaction.transaction->IsReadOnly()) { + transaction.transaction->SetReadWrite(); + } + LanceRefreshTableAlterInfo refresh_info(AlterEntryData( + catalog.GetName(), name, table.name, OnEntryNotFound::THROW_EXCEPTION)); + // Defensive: match the user-ALTER flow above, which also allows internal + // entries. (TableCatalogEntry does not propagate CreateTableInfo.internal + // to the entry, so lance entries are not actually flagged internal, but + // the refresh must never bounce off that check if this ever changes.) + refresh_info.allow_internal = true; + try { + if (!set.AlterEntry(transaction, table.name, refresh_info)) { + return false; + } + } catch (CatalogException &) { + // Entry vanished or changed under us (e.g. a concurrent DROP); fail + // open and keep serving the current entry, like the staleness probe. + return false; + } catch (TransactionException &) { + // Catalog write-write conflict: another transaction already created a + // newer version of this entry. Fail open; the next statement observes + // the winner's entry. + return false; + } catch (IOException &) { + // The rebuild inside AlterEntry re-opens the dataset and can hit infra + // errors (dataset unreachable). The heal is best-effort: keep serving + // the current entry; the caller's fail-closed error or the next bind + // surfaces the real problem with proper context. + return false; + } + return true; + } + private: void InvalidateTableDefaults() { if (!table_default_generator) { @@ -1102,6 +1323,19 @@ class LanceSchemaEntry final : public DuckSchemaEntry { DefaultGenerator *table_default_generator = nullptr; }; +bool LanceTryReplaceStaleTableEntry(ClientContext &context, + LanceTableEntry &table) { + auto *lance_schema = dynamic_cast(&table.schema); + if (!lance_schema) { + return false; + } + // Use the caller's real catalog transaction so the superseded entry lands + // in its undo buffer; ReplaceStaleTableEntry skips the refresh (fail open) + // for non-transactional callers. + auto transaction = table.ParentCatalog().GetCatalogTransaction(context); + return lance_schema->ReplaceStaleTableEntry(transaction, table); +} + class LanceDuckCatalog final : public DuckCatalog { public: using DuckCatalog::PlanDelete; @@ -1115,6 +1349,23 @@ class LanceDuckCatalog final : public DuckCatalog { using DuckCatalog::PlanUpdate; + // Lance table entries mirror external datasets whose commits never bump any + // DuckDB catalog version, so a "current" version number would vouch for + // state this catalog cannot actually track. Returning an invalid version + // opts the catalog out of catalog-version identity checks - the documented + // escape hatch in CheckCatalogIdentity (prepared_statement_data.cpp) for + // catalogs that don't support catalog versions - which forces every EXECUTE + // of a prepared statement that reads or writes this catalog to rebind and + // observe the same latest state as ad-hoc statements. Scans already force + // this through LanceScanBindData::SupportStatementCache() returning false, + // but target-only DML (INSERT/UPDATE/DELETE/MERGE without a scan of the + // target, e.g. INSERT ... VALUES) binds no scan: its cached physical plan + // pins a LanceTableEntry reference that would outlive a catalog-entry + // replacement and bypass all bind-time freshness checks. + optional_idx GetCatalogVersion(ClientContext &) override { + return optional_idx(); + } + ErrorData SupportsCreateTable(BoundCreateTableInfo &info) override { auto &base = info.Base().Cast(); if (!base.partition_keys.empty()) { diff --git a/test/data/float16_dml_fixture.lance/_transactions/0-b59216a0-fc78-4d4f-a876-69559ffebb33.txn b/test/data/float16_dml_fixture.lance/_transactions/0-b59216a0-fc78-4d4f-a876-69559ffebb33.txn new file mode 100644 index 00000000..6a1235f7 Binary files /dev/null and b/test/data/float16_dml_fixture.lance/_transactions/0-b59216a0-fc78-4d4f-a876-69559ffebb33.txn differ diff --git a/test/data/float16_dml_fixture.lance/_versions/18446744073709551614.manifest b/test/data/float16_dml_fixture.lance/_versions/18446744073709551614.manifest new file mode 100644 index 00000000..42f2e144 Binary files /dev/null and b/test/data/float16_dml_fixture.lance/_versions/18446744073709551614.manifest differ diff --git a/test/data/float16_dml_fixture.lance/_versions/latest_version_hint.json b/test/data/float16_dml_fixture.lance/_versions/latest_version_hint.json new file mode 100644 index 00000000..491d7344 --- /dev/null +++ b/test/data/float16_dml_fixture.lance/_versions/latest_version_hint.json @@ -0,0 +1 @@ +{"version":1} \ No newline at end of file diff --git a/test/data/float16_dml_fixture.lance/data/11110100110010100100011030db544065b1f38c588619e6d5.lance b/test/data/float16_dml_fixture.lance/data/11110100110010100100011030db544065b1f38c588619e6d5.lance new file mode 100644 index 00000000..6a5b63fc Binary files /dev/null and b/test/data/float16_dml_fixture.lance/data/11110100110010100100011030db544065b1f38c588619e6d5.lance differ diff --git a/test/data/float16_evolution_fixture.lance/_transactions/0-872dd43f-2396-41e7-981e-c9d07348f4a4.txn b/test/data/float16_evolution_fixture.lance/_transactions/0-872dd43f-2396-41e7-981e-c9d07348f4a4.txn new file mode 100644 index 00000000..926a8503 Binary files /dev/null and b/test/data/float16_evolution_fixture.lance/_transactions/0-872dd43f-2396-41e7-981e-c9d07348f4a4.txn differ diff --git a/test/data/float16_evolution_fixture.lance/_versions/18446744073709551614.manifest b/test/data/float16_evolution_fixture.lance/_versions/18446744073709551614.manifest new file mode 100644 index 00000000..adb58dff Binary files /dev/null and b/test/data/float16_evolution_fixture.lance/_versions/18446744073709551614.manifest differ diff --git a/test/data/float16_evolution_fixture.lance/_versions/latest_version_hint.json b/test/data/float16_evolution_fixture.lance/_versions/latest_version_hint.json new file mode 100644 index 00000000..491d7344 --- /dev/null +++ b/test/data/float16_evolution_fixture.lance/_versions/latest_version_hint.json @@ -0,0 +1 @@ +{"version":1} \ No newline at end of file diff --git a/test/data/float16_evolution_fixture.lance/data/0110111010000001100110102d6c274a3d93638df2c5d22e67.lance b/test/data/float16_evolution_fixture.lance/data/0110111010000001100110102d6c274a3d93638df2c5d22e67.lance new file mode 100644 index 00000000..6a5b63fc Binary files /dev/null and b/test/data/float16_evolution_fixture.lance/data/0110111010000001100110102d6c274a3d93638df2c5d22e67.lance differ diff --git a/test/sql/dataset_cache_catalog_only_freshness.test b/test/sql/dataset_cache_catalog_only_freshness.test new file mode 100644 index 00000000..82d1729a --- /dev/null +++ b/test/sql/dataset_cache_catalog_only_freshness.test @@ -0,0 +1,66 @@ +# name: test/sql/dataset_cache_catalog_only_freshness.test +# description: Catalog-only reads (DESCRIBE / information_schema) observe external schema changes +# group: [sql] + +require lance + +# DESCRIBE and information_schema expose the catalog entry's metadata without +# ever binding a scan or planning DML, so entry resolution itself must run +# the freshness check: when the first access after an external schema or +# nullability change is catalog-only, the stale entry is replaced through +# the catalog version chain and the fresh one returned transparently. + +statement ok con1 +COPY ( + SELECT 1::BIGINT AS id, 'a'::VARCHAR AS s + UNION ALL + SELECT 2::BIGINT AS id, 'b'::VARCHAR AS s +) TO 'test/.tmp/cache_catalog_only.lance' (FORMAT lance, mode 'overwrite'); + +statement ok con1 +ATTACH 'test/.tmp' AS cat_reader_ns (TYPE LANCE); + +statement ok con2 +ATTACH 'test/.tmp' AS cat_writer_ns (TYPE LANCE); + +# The reader discovers the table with s nullable. +query TTTTTT con1 +DESCRIBE cat_reader_ns.main.cache_catalog_only; +---- +id BIGINT YES NULL NULL NULL +s VARCHAR YES NULL NULL NULL + +# External SET NOT NULL; the first reader access is catalog-only and must +# already reflect the new nullability. +statement ok con2 +ALTER TABLE cat_writer_ns.main.cache_catalog_only ALTER COLUMN s SET NOT NULL; + +query TTTTTT con1 +DESCRIBE cat_reader_ns.main.cache_catalog_only; +---- +id BIGINT YES NULL NULL NULL +s VARCHAR NO NULL NULL NULL + +# External DROP COLUMN; the first reader access goes through +# information_schema-style enumeration and must not list the dropped column. +statement ok con2 +ALTER TABLE cat_writer_ns.main.cache_catalog_only DROP COLUMN s; + +query T con1 +SELECT column_name FROM duckdb_columns() +WHERE database_name = 'cat_reader_ns' AND table_name = 'cache_catalog_only' +ORDER BY column_name; +---- +id + +# Scans agree with the refreshed metadata without any intermediate error. +query I con1 +SELECT count(*) FROM cat_reader_ns.main.cache_catalog_only; +---- +2 + +statement ok con1 +DETACH cat_reader_ns; + +statement ok con2 +DETACH cat_writer_ns; diff --git a/test/sql/dataset_cache_external_dml_freshness.test b/test/sql/dataset_cache_external_dml_freshness.test new file mode 100644 index 00000000..104112b9 --- /dev/null +++ b/test/sql/dataset_cache_external_dml_freshness.test @@ -0,0 +1,54 @@ +# name: test/sql/dataset_cache_external_dml_freshness.test +# description: Write-only DML validates entry freshness without requiring a prior SELECT +# group: [sql] + +require lance + +# Plain INSERT does not bind a scan of the target, so it never passes through +# the scan-bind freshness check. Without validating freshness in the INSERT +# planning path, a stale entry keeps gating writes on outdated state: after an +# external float16 -> float32 evolution, INSERT would keep rejecting the +# column as coerced until the user happens to run a SELECT first. +# +# Fixture: test/data/float16_dml_fixture.lance stores column "h" as float16. +# NOTE: like dml_alter_table_vector_schema_evolution.test, this test evolves +# a checked-in fixture in place; re-running on a dirty tree requires +# restoring test/data first (fresh CI checkouts are unaffected). + +statement ok con1 +ATTACH 'test/data' AS dml_reader (TYPE LANCE); + +statement ok con2 +ATTACH 'test/data' AS dml_writer (TYPE LANCE); + +# The reader discovers the table with h coerced; writes are gated. +query I con1 +SELECT count(*) FROM dml_reader.main.float16_dml_fixture; +---- +3 + +statement error con1 +INSERT INTO dml_reader.main.float16_dml_fixture VALUES (4, 3.5); +---- +:.*coerced.* + +# External evolution through the writer catalog: h becomes real float32. +statement ok con2 +ALTER TABLE dml_writer.main.float16_dml_fixture ALTER COLUMN h TYPE FLOAT; + +# A direct INSERT (no SELECT in between) must not keep rejecting the column +# as coerced: entry resolution heals the stale entry transparently and the +# write goes through against the refreshed entry. +statement ok con1 +INSERT INTO dml_reader.main.float16_dml_fixture VALUES (4, 3.5); + +query I con1 +SELECT count(*) FROM dml_reader.main.float16_dml_fixture; +---- +4 + +statement ok con1 +DETACH dml_reader; + +statement ok con2 +DETACH dml_writer; diff --git a/test/sql/dataset_cache_external_nullability.test b/test/sql/dataset_cache_external_nullability.test new file mode 100644 index 00000000..97202a98 --- /dev/null +++ b/test/sql/dataset_cache_external_nullability.test @@ -0,0 +1,69 @@ +# name: test/sql/dataset_cache_external_nullability.test +# description: External SET/DROP NOT NULL replaces stale catalog entries +# group: [sql] + +require lance + +# Nullability changes leave the DuckDB names/types and the coerced-column +# list untouched, so the freshness check must also compare the NOT NULL +# state mirrored from the Arrow flags; otherwise the reader keeps exposing +# and enforcing the old nullability until an unrelated schema change occurs. + +statement ok con1 +COPY ( + SELECT 1::BIGINT AS id, 'a'::VARCHAR AS s + UNION ALL + SELECT 2::BIGINT AS id, 'b'::VARCHAR AS s +) TO 'test/.tmp/cache_nullability.lance' (FORMAT lance, mode 'overwrite'); + +statement ok con1 +ATTACH 'test/.tmp' AS null_reader_ns (TYPE LANCE); + +statement ok con2 +ATTACH 'test/.tmp' AS null_writer_ns (TYPE LANCE); + +# The reader discovers the table with s nullable. +query I con1 +SELECT count(*) FROM null_reader_ns.main.cache_nullability; +---- +2 + +query TTTTTT con1 +DESCRIBE null_reader_ns.main.cache_nullability; +---- +id BIGINT YES NULL NULL NULL +s VARCHAR YES NULL NULL NULL + +# External SET NOT NULL committed through the writer catalog. +statement ok con2 +ALTER TABLE null_writer_ns.main.cache_nullability ALTER COLUMN s SET NOT NULL; + +# The reader's stale entry must not silently keep the old nullability: entry +# resolution heals it transparently and the very next access reflects the +# current state. +query I con1 +SELECT count(*) FROM null_reader_ns.main.cache_nullability; +---- +2 + +query TTTTTT con1 +DESCRIBE null_reader_ns.main.cache_nullability; +---- +id BIGINT YES NULL NULL NULL +s VARCHAR NO NULL NULL NULL + +# External DROP NOT NULL: same transparent self-heal cycle. +statement ok con2 +ALTER TABLE null_writer_ns.main.cache_nullability ALTER COLUMN s DROP NOT NULL; + +query TTTTTT con1 +DESCRIBE null_reader_ns.main.cache_nullability; +---- +id BIGINT YES NULL NULL NULL +s VARCHAR YES NULL NULL NULL + +statement ok con1 +DETACH null_reader_ns; + +statement ok con2 +DETACH null_writer_ns; diff --git a/test/sql/dataset_cache_external_schema_coercion.test b/test/sql/dataset_cache_external_schema_coercion.test new file mode 100644 index 00000000..11f02dee --- /dev/null +++ b/test/sql/dataset_cache_external_schema_coercion.test @@ -0,0 +1,60 @@ +# name: test/sql/dataset_cache_external_schema_coercion.test +# description: External type evolution between coercion-equivalent Arrow types replaces stale entries +# group: [sql] + +require lance + +# Fixture: test/data/float16_evolution_fixture.lance stores column "h" as +# float16, which the reader boundary surfaces as FLOAT and whose coerced +# state gates writes. An external ALTER of "h" to real float32 keeps the +# post-coercion name/type pair identical, so only the coerced-column state +# distinguishes the stale catalog entry from the evolved dataset. +# +# NOTE: like dml_alter_table_vector_schema_evolution.test, this test evolves +# a checked-in fixture in place; re-running on a dirty tree requires +# restoring test/data first (fresh CI checkouts are unaffected). + +statement ok con1 +ATTACH 'test/data' AS coerce_reader (TYPE LANCE); + +statement ok con2 +ATTACH 'test/data' AS coerce_writer (TYPE LANCE); + +# The reader discovers the table with the float16 column surfaced as FLOAT... +query TIR con1 +SELECT typeof(h), id, h FROM coerce_reader.main.float16_evolution_fixture ORDER BY id LIMIT 1; +---- +FLOAT 1 1.5 + +# ...and the coerced column rejects writes. +statement error con1 +INSERT INTO coerce_reader.main.float16_evolution_fixture VALUES (4, 3.5); +---- +:.*coerced.* + +# External evolution: the writer catalog converts h to real float32. The +# post-coercion schema (name FLOAT) is unchanged; only the coerced state is. +statement ok con2 +ALTER TABLE coerce_writer.main.float16_evolution_fixture ALTER COLUMN h TYPE FLOAT; + +# The reader's stale entry still marks h as coerced; entry resolution heals +# it transparently and reads see the current (uncoerced) schema immediately. +query IR con1 +SELECT id, h FROM coerce_reader.main.float16_evolution_fixture ORDER BY id LIMIT 1; +---- +1 1.5 + +# Writes now succeed: the refreshed entry no longer lists coerced columns. +statement ok con1 +INSERT INTO coerce_reader.main.float16_evolution_fixture VALUES (4, 3.5); + +query I con1 +SELECT count(*) FROM coerce_reader.main.float16_evolution_fixture; +---- +4 + +statement ok con1 +DETACH coerce_reader; + +statement ok con2 +DETACH coerce_writer; diff --git a/test/sql/dataset_cache_external_schema_evolution.test b/test/sql/dataset_cache_external_schema_evolution.test new file mode 100644 index 00000000..48799a12 --- /dev/null +++ b/test/sql/dataset_cache_external_schema_evolution.test @@ -0,0 +1,108 @@ +# name: test/sql/dataset_cache_external_schema_evolution.test +# description: External schema evolution behind a cached catalog entry fails closed and self-heals +# group: [sql] + +require lance + +# Each ATTACH creates its own catalog with its own lazily discovered table +# entries. A schema change committed through the writer catalog acts as an +# external schema evolution from the reader catalog's perspective: the +# reader's entry columns were captured at discovery time and no longer match +# the dataset schema. Serving that stale binding would mislabel columns, so +# the reader must fail with an explicit error and re-discover the table with +# its current schema on the next access. + +statement ok con1 +COPY ( + SELECT 1::BIGINT AS id, 'a'::VARCHAR AS s, 10::BIGINT AS extra + UNION ALL + SELECT 2::BIGINT AS id, 'b'::VARCHAR AS s, 20::BIGINT AS extra +) TO 'test/.tmp/cache_schema_evo.lance' (FORMAT lance, mode 'overwrite'); + +statement ok con1 +ATTACH 'test/.tmp' AS evo_reader (TYPE LANCE); + +statement ok con2 +ATTACH 'test/.tmp' AS evo_writer (TYPE LANCE); + +# con1 discovers the table with the original three-column schema. +query ITI con1 +SELECT id, s, extra FROM evo_reader.main.cache_schema_evo ORDER BY id; +---- +1 a 10 +2 b 20 + +# External drop-column committed through the writer catalog. +statement ok con2 +ALTER TABLE evo_writer.main.cache_schema_evo DROP COLUMN s; + +# The reader's stale entry must not serve mislabeled results (without the +# schema check, "s" would silently return the values of "extra"). Entry +# resolution heals the stale entry transparently, so the query binds against +# the current schema and fails with a plain missing-column error. +statement error con1 +SELECT s FROM evo_reader.main.cache_schema_evo ORDER BY id; +---- +:.*(not found|does not exist).* + +# The healed entry serves the current schema immediately. +query II con1 +SELECT id, extra FROM evo_reader.main.cache_schema_evo ORDER BY id; +---- +1 10 +2 20 + +query II con1 +SELECT * FROM evo_reader.main.cache_schema_evo ORDER BY id; +---- +1 10 +2 20 + +# External rename: same fail-closed + self-heal cycle. +statement ok con2 +ALTER TABLE evo_writer.main.cache_schema_evo RENAME extra TO score; + +statement error con1 +SELECT extra FROM evo_reader.main.cache_schema_evo ORDER BY id; +---- +:.*(not found|does not exist).* + +query I con1 +SELECT score FROM evo_reader.main.cache_schema_evo ORDER BY id; +---- +10 +20 + +# DROP TABLE as the very first statement after an external change: the DROP's +# own bind refreshes the stale entry inside the DROP's transaction, and the +# non-transactional catalog part of DROP cannot run on top of that +# uncommitted refresh. It must fail closed BEFORE deleting the dataset +# (proceeding would delete the data and then abort on the catalog conflict). +statement ok con2 +ALTER TABLE evo_writer.main.cache_schema_evo ADD COLUMN flag BIGINT; + +statement error con1 +DROP TABLE evo_reader.main.cache_schema_evo; +---- +:.*refreshed in the current transaction.* + +# The dataset is intact; a successful statement commits the refreshed entry. +query I con1 +SELECT count(*) FROM evo_reader.main.cache_schema_evo; +---- +2 + +# With the refresh committed, the DROP goes through. +statement ok con1 +DROP TABLE evo_reader.main.cache_schema_evo; + +statement error con1 +SELECT count(*) FROM evo_reader.main.cache_schema_evo; +---- +:.*(does not exist|not found).* + +statement ok con1 +DETACH evo_reader; + +statement ok con2 +DETACH evo_writer; diff --git a/test/sql/dataset_cache_external_writer_revalidation.test b/test/sql/dataset_cache_external_writer_revalidation.test new file mode 100644 index 00000000..39123376 --- /dev/null +++ b/test/sql/dataset_cache_external_writer_revalidation.test @@ -0,0 +1,142 @@ +# name: test/sql/dataset_cache_external_writer_revalidation.test +# description: Cache hits are revalidated against the latest committed version so external commits become visible +# group: [sql] + +require lance + +# The dataset cache is connection-local, so a commit made through con2 acts as +# an "external writer" from con1's perspective: the write invalidates only +# con2's own cache entry. Without revalidation on cache hit, con1 would keep +# serving the pre-commit version forever. + +statement ok con1 +COPY ( + SELECT 1::BIGINT AS id, 'a'::VARCHAR AS s + UNION ALL + SELECT 2::BIGINT AS id, 'b'::VARCHAR AS s + UNION ALL + SELECT 3::BIGINT AS id, 'c'::VARCHAR AS s +) TO 'test/.tmp/cache_external_writer.lance' (FORMAT lance, mode 'overwrite'); + +# con1 caches the dataset at the initial version. +query I con1 +SELECT count(*) FROM 'test/.tmp/cache_external_writer.lance'; +---- +3 + +query II con1 +EXPLAIN (FORMAT JSON) SELECT * FROM 'test/.tmp/cache_external_writer.lance'; +---- +physical_plan :[\s\S]*"Lance Dataset Cache Hit": "true"[\s\S]* + +# External delete: committed through con2, bypassing con1's connection. +statement ok con2 +ATTACH 'test/.tmp' AS ext_writer_ns (TYPE LANCE); + +query I con2 +DELETE FROM ext_writer_ns.main.cache_external_writer WHERE id = 3; +---- +1 + +# con1 must observe the externally committed delete on its next read. +query I con1 +SELECT count(*) FROM 'test/.tmp/cache_external_writer.lance'; +---- +2 + +query I con1 +SELECT id FROM 'test/.tmp/cache_external_writer.lance' ORDER BY id; +---- +1 +2 + +# External append: con2 commits a new row. +statement ok con2 +COPY (SELECT 4::BIGINT AS id, 'd'::VARCHAR AS s) +TO 'test/.tmp/cache_external_writer.lance' (FORMAT lance, mode 'append'); + +# The first con1 access after the external commit refreshes the stale entry +# and reports it as a cache miss... +query II con1 +EXPLAIN (FORMAT JSON) SELECT * FROM 'test/.tmp/cache_external_writer.lance'; +---- +physical_plan :[\s\S]*"Lance Dataset Cache Hit": "false"[\s\S]* + +# ...and sees the appended row. +query I con1 +SELECT count(*) FROM 'test/.tmp/cache_external_writer.lance'; +---- +3 + +# Once refreshed, subsequent hits stay hits until the next external commit. +query II con1 +EXPLAIN (FORMAT JSON) SELECT * FROM 'test/.tmp/cache_external_writer.lance'; +---- +physical_plan :[\s\S]*"Lance Dataset Cache Hit": "true"[\s\S]* + +# Catalog-attached reads go through the same cache and must also observe +# external commits. +statement ok con1 +ATTACH 'test/.tmp' AS ext_reader_ns (TYPE LANCE); + +query I con1 +SELECT count(*) FROM ext_reader_ns.main.cache_external_writer; +---- +3 + +query I con2 +DELETE FROM ext_writer_ns.main.cache_external_writer WHERE id = 1; +---- +1 + +query I con1 +SELECT count(*) FROM ext_reader_ns.main.cache_external_writer; +---- +2 + +# External drop + re-create at the same URI: the new table's history restarts +# at the same version id as the cached entry (both are at version 1), so only +# the manifest identity (e-tag) distinguishes the two tables. The cached +# reader must observe the re-created table, not the dropped one. +statement ok con1 +COPY ( + SELECT 1::BIGINT AS id, 'a'::VARCHAR AS s + UNION ALL + SELECT 2::BIGINT AS id, 'b'::VARCHAR AS s + UNION ALL + SELECT 3::BIGINT AS id, 'c'::VARCHAR AS s +) TO 'test/.tmp/cache_recreate.lance' (FORMAT lance, mode 'overwrite'); + +# con1 caches the dataset at version 1 of the original table. +query I con1 +SELECT count(*) FROM 'test/.tmp/cache_recreate.lance'; +---- +3 + +statement ok con2 +DROP TABLE ext_writer_ns.main.cache_recreate; + +statement ok con2 +COPY ( + SELECT 10::BIGINT AS id, 'x'::VARCHAR AS s + UNION ALL + SELECT 20::BIGINT AS id, 'y'::VARCHAR AS s +) TO 'test/.tmp/cache_recreate.lance' (FORMAT lance, mode 'overwrite'); + +# The re-created table is also at version 1; con1 must still see it. +query I con1 +SELECT count(*) FROM 'test/.tmp/cache_recreate.lance'; +---- +2 + +query I con1 +SELECT id FROM 'test/.tmp/cache_recreate.lance' ORDER BY id; +---- +10 +20 + +statement ok con1 +DETACH ext_reader_ns; + +statement ok con2 +DETACH ext_writer_ns; diff --git a/test/sql/dataset_cache_prepared_dml.test b/test/sql/dataset_cache_prepared_dml.test new file mode 100644 index 00000000..b79084f5 --- /dev/null +++ b/test/sql/dataset_cache_prepared_dml.test @@ -0,0 +1,200 @@ +# name: test/sql/dataset_cache_prepared_dml.test +# description: Prepared DML observes external commits instead of reusing stale plans +# group: [sql] + +require lance + +# Target-only DML (INSERT ... VALUES and friends) binds no scan of the +# target, so the scan bind data never forces a rebind for it; a reused +# prepared plan would pin a stale LanceTableEntry and a freshness check that +# only ran at plan time. LanceDuckCatalog::GetCatalogVersion opts the catalog +# out of prepared-plan reuse, so every EXECUTE rebinds against the latest +# state like an ad-hoc statement. UPDATE/DELETE/MERGE bind a scan of the +# target and were already forced to rebind through the scan bind data; they +# are covered below as regression guards for the same contract. + +statement ok con1 +COPY (SELECT 1::BIGINT AS id, 10::BIGINT AS v) +TO 'test/.tmp/prep_dml_ins.lance' (FORMAT lance, mode 'overwrite'); + +statement ok con1 +COPY ( + SELECT 1::BIGINT AS id, 10::BIGINT AS v + UNION ALL + SELECT 2::BIGINT AS id, 20::BIGINT AS v +) TO 'test/.tmp/prep_dml_upd.lance' (FORMAT lance, mode 'overwrite'); + +statement ok con1 +COPY ( + SELECT 1::BIGINT AS id, 10::BIGINT AS v + UNION ALL + SELECT 2::BIGINT AS id, 20::BIGINT AS v +) TO 'test/.tmp/prep_dml_del.lance' (FORMAT lance, mode 'overwrite'); + +statement ok con1 +COPY (SELECT 1::BIGINT AS id, 10::BIGINT AS v) +TO 'test/.tmp/prep_dml_mrg.lance' (FORMAT lance, mode 'overwrite'); + +statement ok con1 +ATTACH 'test/.tmp' AS prep_dml_reader (TYPE LANCE); + +statement ok con2 +ATTACH 'test/.tmp' AS prep_dml_writer (TYPE LANCE); + +# --------------------------------------------------------------------------- +# INSERT: target-only, external schema evolution +# --------------------------------------------------------------------------- + +statement ok con1 +PREPARE prep_dml_insert AS INSERT INTO prep_dml_reader.main.prep_dml_ins VALUES (2, 20); + +statement ok con1 +EXECUTE prep_dml_insert; + +query I con1 +SELECT count(*) FROM prep_dml_reader.main.prep_dml_ins; +---- +2 + +# External schema evolution committed through the writer catalog. +statement ok con2 +ALTER TABLE prep_dml_writer.main.prep_dml_ins ADD COLUMN extra BIGINT; + +# An ad-hoc read on con1 heals the reader's stale entry. The prepared plan +# still references the superseded entry generation: before the version-chain +# replace, healing destroyed that catalog object outright and the next +# EXECUTE dereferenced freed memory; now the old generation stays alive until +# undo-buffer cleanup, and the EXECUTE below never touches it anyway because +# it is forced to rebind. +query I con1 +SELECT count(*) FROM prep_dml_reader.main.prep_dml_ins; +---- +2 + +# The forced rebind re-resolves the target: the two-value VALUES list no +# longer matches the evolved three-column schema, so EXECUTE fails against +# the current schema instead of appending through the stale plan. +statement error con1 +EXECUTE prep_dml_insert; +---- +:.*(3 columns but 2 values|changed externally).* + +# A statement prepared against the evolved schema goes through. +statement ok con1 +PREPARE prep_dml_insert_v2 AS INSERT INTO prep_dml_reader.main.prep_dml_ins VALUES (3, 30, 300); + +statement ok con1 +EXECUTE prep_dml_insert_v2; + +query I con1 +SELECT count(*) FROM prep_dml_reader.main.prep_dml_ins; +---- +3 + +# --------------------------------------------------------------------------- +# UPDATE: external data commit between EXECUTEs +# --------------------------------------------------------------------------- + +statement ok con1 +PREPARE prep_dml_update AS UPDATE prep_dml_reader.main.prep_dml_upd SET v = v + 1 WHERE id >= 1; + +query I con1 +EXECUTE prep_dml_update; +---- +2 + +statement ok con2 +INSERT INTO prep_dml_writer.main.prep_dml_upd VALUES (3, 30); + +# The re-executed statement must observe the externally inserted row: three +# rows match the predicate now, not the two known at PREPARE time. +query I con1 +EXECUTE prep_dml_update; +---- +3 + +query I con1 +SELECT sum(v) FROM prep_dml_reader.main.prep_dml_upd; +---- +65 + +# --------------------------------------------------------------------------- +# DELETE: external data commit between EXECUTEs +# --------------------------------------------------------------------------- + +statement ok con1 +PREPARE prep_dml_delete AS DELETE FROM prep_dml_reader.main.prep_dml_del WHERE id >= 1; + +query I con1 +EXECUTE prep_dml_delete; +---- +2 + +statement ok con2 +INSERT INTO prep_dml_writer.main.prep_dml_del VALUES (3, 30), (4, 40); + +# Only the externally inserted rows remain; deleting them proves the +# re-executed statement observed the external commit. +query I con1 +EXECUTE prep_dml_delete; +---- +2 + +query I con1 +SELECT count(*) FROM prep_dml_reader.main.prep_dml_del; +---- +0 + +# --------------------------------------------------------------------------- +# MERGE: matched-vs-not-matched flips on external data commit +# --------------------------------------------------------------------------- +# DuckDB's SQL-level PREPARE does not accept MERGE (the parser's +# PreparableStmt rule covers SELECT/INSERT/UPDATE/DELETE/COPY only), so MERGE +# cannot pin a stale plan through PREPARE/EXECUTE. API-level prepared MERGE +# statements go through the same RequireRebind machinery exercised above +# (invalid catalog version) plus the target-scan bind data that already +# forces rebinds; assert the resulting contract - re-running the identical +# MERGE observes external commits - with repeated ad-hoc executions. + +# id 5 is absent: NOT MATCHED inserts (5, 50). +query I con1 +MERGE INTO prep_dml_reader.main.prep_dml_mrg AS t +USING (SELECT 5::BIGINT AS id, 50::BIGINT AS v) AS src +ON t.id = src.id +WHEN MATCHED THEN UPDATE SET v = 999 +WHEN NOT MATCHED THEN INSERT (id, v) VALUES (src.id, src.v); +---- +1 + +query II con1 +SELECT id, v FROM prep_dml_reader.main.prep_dml_mrg WHERE id = 5; +---- +5 50 + +# External commit touching the row the merge matches on. +statement ok con2 +UPDATE prep_dml_writer.main.prep_dml_mrg SET v = 1 WHERE id = 5; + +# The re-executed merge must match the externally updated row: a binding +# pinned before the first execution would consider id 5 absent and insert a +# duplicate row instead of updating. +query I con1 +MERGE INTO prep_dml_reader.main.prep_dml_mrg AS t +USING (SELECT 5::BIGINT AS id, 50::BIGINT AS v) AS src +ON t.id = src.id +WHEN MATCHED THEN UPDATE SET v = 999 +WHEN NOT MATCHED THEN INSERT (id, v) VALUES (src.id, src.v); +---- +1 + +query II con1 +SELECT id, v FROM prep_dml_reader.main.prep_dml_mrg ORDER BY id; +---- +1 10 +5 999 + +statement ok con1 +DETACH prep_dml_reader; + +statement ok con2 +DETACH prep_dml_writer; diff --git a/test/sql/dataset_cache_prepared_statements.test b/test/sql/dataset_cache_prepared_statements.test new file mode 100644 index 00000000..9c9644db --- /dev/null +++ b/test/sql/dataset_cache_prepared_statements.test @@ -0,0 +1,97 @@ +# name: test/sql/dataset_cache_prepared_statements.test +# description: Prepared statements observe external commits like ad-hoc statements +# group: [sql] + +require lance + +# The scan bind data pins a dataset handle checked out at bind time. Without +# forcing a rebind per execution, a prepared statement would keep scanning the +# version captured at PREPARE time and never pass through the cache +# revalidation that ad-hoc statements get at bind. + +statement ok con1 +COPY ( + SELECT 1::BIGINT AS id, 'a'::VARCHAR AS s + UNION ALL + SELECT 2::BIGINT AS id, 'b'::VARCHAR AS s + UNION ALL + SELECT 3::BIGINT AS id, 'c'::VARCHAR AS s +) TO 'test/.tmp/cache_prepared.lance' (FORMAT lance, mode 'overwrite'); + +statement ok con1 +PREPARE prepared_path_count AS SELECT count(*) FROM 'test/.tmp/cache_prepared.lance'; + +query I con1 +EXECUTE prepared_path_count; +---- +3 + +# External append committed through con2. +statement ok con2 +COPY (SELECT 4::BIGINT AS id, 'd'::VARCHAR AS s) +TO 'test/.tmp/cache_prepared.lance' (FORMAT lance, mode 'append'); + +# The prepared statement must observe the external commit on re-execution. +query I con1 +EXECUTE prepared_path_count; +---- +4 + +# Same for catalog-attached tables. +statement ok con1 +ATTACH 'test/.tmp' AS prep_reader_ns (TYPE LANCE); + +statement ok con1 +PREPARE prepared_table_count AS SELECT count(*) FROM prep_reader_ns.main.cache_prepared; + +query I con1 +EXECUTE prepared_table_count; +---- +4 + +statement ok con2 +COPY (SELECT 5::BIGINT AS id, 'e'::VARCHAR AS s) +TO 'test/.tmp/cache_prepared.lance' (FORMAT lance, mode 'append'); + +query I con1 +EXECUTE prepared_table_count; +---- +5 + +# External schema evolution behind a prepared statement: the forced rebind +# runs the freshness check and fails closed instead of serving the stale +# binding; re-execution after the entry is replaced succeeds with the new +# schema. +statement ok con2 +ATTACH 'test/.tmp' AS prep_writer_ns (TYPE LANCE); + +statement ok con1 +PREPARE prepared_table_s AS SELECT s FROM prep_reader_ns.main.cache_prepared ORDER BY id; + +query T con1 +EXECUTE prepared_table_s; +---- +a +b +c +d +e + +statement ok con2 +ALTER TABLE prep_writer_ns.main.cache_prepared DROP COLUMN s; + +statement error con1 +EXECUTE prepared_table_s; +---- +:.*(changed externally|does not exist|not found).* + +query I con1 +EXECUTE prepared_table_count; +---- +5 + +statement ok con1 +DETACH prep_reader_ns; + +statement ok con2 +DETACH prep_writer_ns; diff --git a/test/sql/dataset_cache_statement_memoization.test b/test/sql/dataset_cache_statement_memoization.test new file mode 100644 index 00000000..d1d7a21e --- /dev/null +++ b/test/sql/dataset_cache_statement_memoization.test @@ -0,0 +1,64 @@ +# name: test/sql/dataset_cache_statement_memoization.test +# description: All references to one table within a statement share one revalidated dataset generation +# group: [sql] + +require lance + +# Cache-hit revalidation is memoized per cache key for the duration of one +# statement. Without the memo, a self-join would revalidate once per table +# reference, and an external commit landing between the two binds could pin +# them to different dataset versions within the same statement. The +# "revalidations" profiling counter makes the memo observable: it counts +# actual version lookups, not memoized reuses. + +statement ok con1 +COPY ( + SELECT 1::BIGINT AS id, 'a'::VARCHAR AS s + UNION ALL + SELECT 2::BIGINT AS id, 'b'::VARCHAR AS s +) TO 'test/.tmp/cache_stmt_memo.lance' (FORMAT lance, mode 'overwrite'); + +# Warm the cache: a miss opens the dataset fresh, and the freshly opened +# dataset is already this statement's pinned generation - no revalidation. +query II con1 +EXPLAIN ANALYZE SELECT count(*) FROM 'test/.tmp/cache_stmt_memo.lance'; +---- +analyzed_plan :[\s\S]*Lance Dataset Cache: entries=1 hits=0 misses=1 revalidations=0[\s\S]* + +# A single reference on a warmed cache revalidates exactly once. +query II con1 +EXPLAIN ANALYZE SELECT count(*) FROM 'test/.tmp/cache_stmt_memo.lance'; +---- +analyzed_plan :[\s\S]*Lance Dataset Cache: entries=1 hits=1 misses=0 revalidations=1[\s\S]* + +# A self-join binds two references to the same cache key but still +# revalidates exactly once: the second reference reuses the generation the +# first reference pinned for this statement (pre-memoization this reported +# revalidations=2, and an external commit between the two binds could have +# split the join across two dataset versions). +query II con1 +EXPLAIN ANALYZE SELECT count(*) +FROM 'test/.tmp/cache_stmt_memo.lance' a +JOIN 'test/.tmp/cache_stmt_memo.lance' b ON a.id = b.id; +---- +analyzed_plan :[\s\S]*Lance Dataset Cache: entries=1 hits=2 misses=0 revalidations=1[\s\S]* + +# The memo is statement-scoped, not connection-scoped: the next statement +# revalidates again and thereby observes external commits. +statement ok con2 +COPY (SELECT 3::BIGINT AS id, 'c'::VARCHAR AS s) +TO 'test/.tmp/cache_stmt_memo.lance' (FORMAT lance, mode 'append'); + +# The refreshed hit is reported as a miss (see the revalidation counters) but +# still required exactly one revalidation for both join sides. +query II con1 +EXPLAIN ANALYZE SELECT count(*) +FROM 'test/.tmp/cache_stmt_memo.lance' a +JOIN 'test/.tmp/cache_stmt_memo.lance' b ON a.id = b.id; +---- +analyzed_plan :[\s\S]*Lance Dataset Cache: entries=1 hits=1 misses=1 revalidations=1[\s\S]* + +query I con1 +SELECT count(*) FROM 'test/.tmp/cache_stmt_memo.lance'; +---- +3 diff --git a/test/sql/index_ddl.test b/test/sql/index_ddl.test index 3e8593e0..57c56c9c 100644 --- a/test/sql/index_ddl.test +++ b/test/sql/index_ddl.test @@ -262,11 +262,14 @@ path_paren_idx :.* escaped.`a)b` 8 :.*|NULL statement ok ATTACH 'test/.tmp' AS nested_ns (TYPE LANCE); +# Catalog entry resolution validates schema freshness against the (cached) +# dataset before the scan binds, so the first catalog access already warms +# the dataset cache and the scan bind reports a hit. query II EXPLAIN (FORMAT JSON) SELECT count(*) FROM nested_ns.main.index_ddl_nested WHERE right_struct.value = 4; ---- -physical_plan :[\s\S]*"Lance Dataset Cache Hit": "false"[\s\S]* +physical_plan :[\s\S]*"Lance Dataset Cache Hit": "true"[\s\S]* statement ok CREATE INDEX right_value_idx ON nested_ns.main.index_ddl_nested (index_ddl_nested.right_struct.value) @@ -282,11 +285,14 @@ path_comma_idx :.* escaped.`a,b` 8 :.*|NULL path_paren_idx :.* escaped.`a)b` 8 :.*|NULL right_value_idx :.* right_struct.value 8 :.*|NULL +# CREATE INDEX invalidated the cached dataset; the catalog freshness probe +# reopens it during entry resolution (taking the miss), so the scan bind +# itself reports a hit again. query II EXPLAIN (ANALYZE, FORMAT JSON) SELECT count(*) FROM nested_ns.main.index_ddl_nested WHERE right_struct.value = 4; ---- -analyzed_plan :[\s\S]*"Lance Dataset Cache Hit": "false"[\s\S]*"Lance Scan Mode": "dataset"[\s\S]* +analyzed_plan :[\s\S]*"Lance Dataset Cache Hit": "true"[\s\S]*"Lance Scan Mode": "dataset"[\s\S]* statement error CREATE INDEX ambiguous_leaf_idx ON 'test/.tmp/index_ddl_nested.lance' (value)