chore: upgrade Lance dependencies to 13.0.0

crates/lance-context-core/Cargo.toml

@@ -19,9 +19,10 @@ metrics = ["dep:metrics"]
 
 [dependencies]
 async-trait = "0.1.92"
-lance-table = "9.0.0"
-lance-io = "9.0.0"
-object_store = "0.13.2"
+lance-table = "13.0.0"
+lance-io = "13.0.0"
+lance-file = "13.0.0"
+object_store = "0.14.2"
 base64 = "0.22"
 bytes = "1"
 arrow-array = "58"
@@ -31,10 +32,10 @@ arrow-schema = "58"
 arrow-select = "58"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
 datafusion = { version = "54", default-features = false, features = ["nested_expressions"] }
-lance = "9.0.0"
+lance = "13.0.0"
 lance-context-api = { version = "0.6.7", path = "../lance-context-api" }
-lance-index = "9.0.0"
-lance-namespace = "9.0.0"
+lance-index = "13.0.0"
+lance-namespace = "13.0.0"
 # Version-matched with lance-context-server/-master so one process-wide recorder
 # serves every crate.
 metrics = { version = "0.24", optional = true }

crates/lance-context-core/examples/generic_batch_get.rs

@@ -1,5 +1,6 @@
 //! Run with `cargo run -p lance-context-core --example generic_batch_get`.
 //! Supply `--uri` to benchmark an existing store without changing its data.
+#![recursion_limit = "256"]
 
 use std::time::Instant;
 

crates/lance-context-core/src/datagen_store.rs

@@ -335,7 +335,7 @@ impl DatagenStore {
         let mut generations: Vec<(u64, String)> = snapshots
             .iter()
             .flat_map(|snapshot| {
-                snapshot.flushed_generations.iter().map(|generation| {
+                snapshot.sstables.iter().map(|generation| {
                     (
                         generation.generation,
                         self.base

crates/lance-context-core/src/generic_store.rs

@@ -528,7 +528,7 @@ impl GenericStore {
         // pending WAL generations before the key/work budgets can reject it.
         collect_scan_keys(dataset, &mut ids, &mut id_bytes, &mut scanned).await?;
         for snapshot in snapshots {
-            for generation in &snapshot.flushed_generations {
+            for generation in &snapshot.sstables {
                 let path = self
                     .base
                     .flushed_generation_uri(snapshot.shard_id, &generation.path);

crates/lance-context-core/src/merge_write_scope.rs

@@ -318,6 +318,15 @@ struct Layered {
 }
 
 impl lance_io::object_store::WrappingObjectStore for Layered {
+    fn wrap_paginated(
+        &self,
+        name: &str,
+        original: Arc<dyn object_store::list::PaginatedListStore>,
+    ) -> Option<Arc<dyn object_store::list::PaginatedListStore>> {
+        self.inner
+            .wrap_paginated(name, original)
+            .and_then(|inner| self.outer.wrap_paginated(name, inner))
+    }
     fn wrap(
         &self,
         name: &str,
@@ -439,7 +448,7 @@ pub(crate) async fn drain_generations(
     let resource = format!("shard:{}", store.shard_id());
     for _ in 0..10 {
         let mut next = store
-            .read_latest()
+            .refresh_latest()
             .await?
             .ok_or_else(|| Error::io("Shard manifest not found"))?;
         if next.writer_epoch != epoch {
@@ -449,17 +458,17 @@ pub(crate) async fn drain_generations(
             .version
             .checked_add(1)
             .ok_or_else(|| Error::io("manifest version overflow"))?;
-        next.flushed_generations
+        next.sstables
             .retain(|g| !generations.contains(&g.generation));
         authorize(&resource, next.version).await?;
         let writer = store.clone();
         let scope = CURRENT.try_with(Arc::clone).ok();
         let (next, written) = shield_observed(
             async move {
-                let written = writer.write(&next).await;
+                let written = commit_exact_shard(&writer, &next).await;
                 if written
                     .as_ref()
-                    .is_err_and(|e| !e.to_string().contains("already exists"))
+                    .is_err_and(|e| !matches!(e, Error::RetryableCommitConflict { .. }))
                 {
                     if let Some(scope) = scope {
                         scope.mark_uncertain();
@@ -472,13 +481,38 @@ pub(crate) async fn drain_generations(
         .await?;
         match written {
             Ok(_) => return Ok(next),
-            Err(error) if error.to_string().contains("already exists") => continue,
+            Err(Error::RetryableCommitConflict { .. }) => continue,
             Err(error) => return Err(error),
         }
     }
     Err(Error::io("shard drain exceeded conditional-write retries"))
 }
 
+/// Adapt Lance's retrying update API to one previously authorized version.
+/// Returning the SAME manifest on a retry makes Lance's successor check reject
+/// a moved predecessor before any new PUT; the caller must authorize again.
+pub(crate) async fn commit_exact_shard(
+    store: &lance::dataset::mem_wal::ShardManifestStore,
+    next: &lance_index::mem_wal::ShardManifest,
+) -> Result<lance_index::mem_wal::ShardManifest> {
+    let moved = std::sync::atomic::AtomicBool::new(false);
+    let result = store
+        .commit_update(next.writer_epoch, |current| {
+            if current.version.checked_add(1) != Some(next.version) {
+                moved.store(true, Ordering::Relaxed);
+            }
+            next.clone()
+        })
+        .await;
+    if result.is_err() && moved.load(Ordering::Relaxed) {
+        return Err(Error::retryable_commit_conflict_source(
+            next.version,
+            "shard manifest predecessor changed; reauthorize".into(),
+        ));
+    }
+    result
+}
+
 /// The owned protocol is supported only with immutable conditional manifest
 /// creation, not Lance's fallback UnsafeCommitHandler or external catalogues.
 pub fn supports_version_fencing(uri: &str) -> bool {
@@ -569,7 +603,7 @@ pub async fn fence_manifest_versions(
         let mut fenced = false;
         for _ in 0..64 {
             let mut next = store
-                .read_latest()
+                .refresh_latest()
                 .await?
                 .ok_or_else(|| Error::io("Shard manifest missing during recovery"))?;
             if next.version > *high {
@@ -580,9 +614,9 @@ pub async fn fence_manifest_versions(
                 .version
                 .checked_add(1)
                 .ok_or_else(|| Error::io("manifest version overflow"))?;
-            match store.write(&next).await {
+            match commit_exact_shard(&store, &next).await {
                 Ok(_) => {}
-                Err(error) if error.to_string().contains("already exists") => {}
+                Err(Error::RetryableCommitConflict { .. }) => {}
                 Err(error) => return Err(error),
             }
         }

crates/lance-context-core/src/preparation_io.rs

@@ -13,6 +13,13 @@ use std::sync::{
 pub(crate) struct ProgressWrapper(pub Arc<AtomicU64>);
 
 impl WrappingObjectStore for ProgressWrapper {
+    fn wrap_paginated(
+        &self,
+        _: &str,
+        original: Arc<dyn object_store::list::PaginatedListStore>,
+    ) -> Option<Arc<dyn object_store::list::PaginatedListStore>> {
+        Some(original)
+    }
     fn wrap(&self, _: &str, original: Arc<dyn ObjectStore>) -> Arc<dyn ObjectStore> {
         Arc::new(ProgressStore {
             inner: original,
@@ -81,6 +88,7 @@ impl ObjectStore for ProgressStore {
         let meta = result.meta.clone();
         let range = result.range.clone();
         let attributes = result.attributes.clone();
+        let extensions = result.extensions.clone();
         let steps = self.steps.clone();
         // Count body chunks after they arrive, not successful response headers.
         // into_stream also observes actual local file reads without buffering.
@@ -99,6 +107,7 @@ impl ObjectStore for ProgressStore {
             meta,
             range,
             attributes,
+            extensions,
         })
     }
 
@@ -257,6 +266,13 @@ pub mod test_support {
     pub struct ThrottleWrapper(pub Throttle);
 
     impl WrappingObjectStore for ThrottleWrapper {
+        fn wrap_paginated(
+            &self,
+            _: &str,
+            original: Arc<dyn object_store::list::PaginatedListStore>,
+        ) -> Option<Arc<dyn object_store::list::PaginatedListStore>> {
+            Some(original)
+        }
         fn wrap(&self, _: &str, original: Arc<dyn ObjectStore>) -> Arc<dyn ObjectStore> {
             Arc::new(ThrottledStore {
                 inner: original,