fix: bump Lance for custom S3 endpoints

crates/lance-context-core/Cargo.toml

@@ -23,12 +23,13 @@ arrow-array = "58"
 arrow-ipc = "58"
 arrow-json = "58"
 arrow-schema = "58"
+arrow-select = "58"
 chrono = { version = "0.4", default-features = false, features = ["clock"] }
-datafusion = { version = "53", default-features = false, features = ["nested_expressions"] }
-lance = "7.0.0"
+datafusion = { version = "54", default-features = false, features = ["nested_expressions"] }
+lance = "9.0.0"
 lance-context-api = { version = "0.6.5", path = "../lance-context-api" }
-lance-index = "7.0.0"
-lance-namespace = "7.0.0"
+lance-index = "9.0.0"
+lance-namespace = "9.0.0"
 lancedb = "0.30.0"
 lance-graph = "0.5.4"
 # Version-matched with lance-context-server/-master so one process-wide recorder

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

@@ -457,7 +457,7 @@ impl DatagenStore {
         let scanner = match columns {
             Some(columns) => {
                 let refs: Vec<&str> = columns.iter().map(String::as_str).collect();
-                self.lsm_scanner().await?.project(&refs).filter(filter)?
+                self.lsm_scanner().await?.project(&refs)?.filter(filter)?
             }
             None => self.lsm_scanner().await?.filter(filter)?,
         };

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

@@ -288,12 +288,15 @@ impl GenericStore {
         columns: &[String],
     ) -> LanceResult<Vec<Row>> {
         let refs: Vec<&str> = columns.iter().map(String::as_str).collect();
-        let mut scanner = self.base.lsm_scanner().await?.project(&refs);
+        let mut scanner = self.base.lsm_scanner().await?.project(&refs)?;
         if let Some(filter) = filter {
             scanner = scanner.filter(filter)?;
         }
         if limit.is_some() || offset.is_some() {
-            scanner = scanner.limit(limit.unwrap_or(usize::MAX), offset);
+            scanner = scanner.limit(
+                map_i64_bound("limit", limit)?,
+                map_i64_bound("offset", offset)?,
+            )?;
         }
 
         let mut stream = scanner.try_into_stream().await?;
@@ -316,7 +319,7 @@ impl GenericStore {
         let scanner = self
             .base
             .lsm_scanner_for_source(source, snapshots)
-            .project(&refs);
+            .project(&refs)?;
 
         let mut stream = scanner.try_into_stream().await?;
         let mut rows = Vec::new();
@@ -425,6 +428,13 @@ fn escape_sql_literal(value: &str) -> String {
     value.replace('\'', "''")
 }
 
+fn map_i64_bound(name: &str, value: Option<usize>) -> LanceResult<Option<i64>> {
+    value
+        .map(i64::try_from)
+        .transpose()
+        .map_err(|_| LanceError::invalid_input(format!("{name} exceeds i64::MAX")))
+}
+
 /// Row batches, for callers that already have Arrow data.
 impl GenericStore {
     /// Append pre-built [`RecordBatch`]es, bypassing row encoding.
@@ -739,7 +749,7 @@ mod tests {
         let uri = dir.path().to_string_lossy().to_string();
         let rt = tokio::runtime::Runtime::new().unwrap();
         rt.block_on(async {
-            let store = GenericStore::open(&uri, spec(), sealing()).await.unwrap();
+            let mut store = GenericStore::open(&uri, spec(), sealing()).await.unwrap();
             store
                 .add(&[row(json!({"id": "r1", "user_id": "first"}))])
                 .await
@@ -752,6 +762,15 @@ mod tests {
             let rows = store.list(None, None).await.unwrap();
             assert_eq!(rows.len(), 1, "id is the merge key, so rows dedup");
             assert_eq!(rows[0]["user_id"], json!("second"));
+
+            store.cleanup_wal().await.unwrap();
+            let rows = store.list(None, None).await.unwrap();
+            assert_eq!(rows.len(), 1, "cleanup must not duplicate the merge key");
+            assert_eq!(
+                rows[0]["user_id"],
+                json!("second"),
+                "cleanup must retain the newest WAL value"
+            );
         });
     }
 

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

@@ -60,6 +60,11 @@ use arrow_array::{
 };
 use arrow_schema::{ArrowError, DataType, Field, FieldRef, Schema, TimeUnit};
 use datafusion::datasource::MemTable;
+use datafusion::physical_expr::expressions::Column;
+use datafusion::physical_plan::expressions::PhysicalSortExpr;
+use datafusion::physical_plan::limit::GlobalLimitExec;
+use datafusion::physical_plan::sorts::sort::SortExec;
+use datafusion::physical_plan::ExecutionPlan;
 use datafusion::prelude::SessionContext;
 use datafusion::sql::parser::{DFParser, Statement as DFStatement};
 use datafusion::sql::sqlparser::ast::Statement as SqlStatement;
@@ -690,13 +695,16 @@ impl RolloutStore {
     ) -> LanceResult<Vec<RolloutRecord>> {
         let columns = self.non_blob_columns();
         let refs: Vec<&str> = columns.iter().map(String::as_str).collect();
-        let mut scanner = self.lsm_scanner().await?.project(&refs);
+        let mut scanner = self.lsm_scanner().await?.project(&refs)?;
         if let Some(predicate) = filters.and_then(RolloutFilters::expression) {
             scanner = scanner.filter(&predicate)?;
         }
         let post_scan_offset = if limit.is_none() { offset } else { None };
         if let Some(limit) = limit {
-            scanner = scanner.limit(limit, offset);
+            scanner = scanner.limit(
+                map_i64_bound("limit", Some(limit))?,
+                map_i64_bound("offset", offset)?,
+            )?;
         }
 
         let mut stream = scanner.try_into_stream().await?;
@@ -752,9 +760,9 @@ impl RolloutStore {
     /// an unbounded full-table count on every UI request. Each source is read in
     /// one projected, filtered, bounded scan. Fragments use the base [`Dataset`]
     /// scanner directly so Lance can push limit/offset into the scan. WAL-backed
-    /// reads page through a narrow `id`-only LSM scan, then take the selected
-    /// rows directly from their physical datasets so wide text columns never
-    /// participate in the full LSM sort.
+    /// reads page through a narrow, deterministic `id`-only top-K sort, then
+    /// take the selected rows directly from their physical datasets so wide
+    /// text columns never participate in the full LSM sort.
     ///
     /// [`ListSource::Fragments`] skips MemWAL manifest discovery entirely, so its
     /// latency is independent of how far the merge backlog has grown.
@@ -774,14 +782,37 @@ impl RolloutStore {
             let shard_snapshots = self.wal_shard_snapshots().await?;
             let mut scanner = self
                 .lsm_scanner_for_source(source, shard_snapshots.clone())
-                .project(&["id"]);
+                .project(&["id"])?;
             if let Some(filter) = &filter {
                 scanner = scanner.filter(filter)?;
             }
-            scanner = scanner.limit(page_limit, Some(offset));
+
+            // Lance 9's block-list LSM plan is intentionally unordered. Page
+            // boundaries must not inherit that internal union order: otherwise
+            // repeated offset requests can skip or repeat records. Sort only
+            // the narrow key plan and retain at most the requested prefix.
+            let plan = scanner.create_plan().await?;
+            let id_index = plan.schema().index_of("id")?;
+            let fetch = offset.saturating_add(page_limit);
+            let sorted = SortExec::new(
+                [PhysicalSortExpr::new_default(Arc::new(Column::new(
+                    "id", id_index,
+                )))]
+                .into(),
+                plan,
+            )
+            .with_fetch(Some(fetch));
+            let page_plan = Arc::new(GlobalLimitExec::new(
+                Arc::new(sorted),
+                offset,
+                Some(page_limit),
+            ));
+            let ctx = SessionContext::new();
 
             let mut page_ids = Vec::with_capacity(page_limit);
-            let mut stream = scanner.try_into_stream().await?;
+            let mut stream = page_plan
+                .execute(0, ctx.task_ctx())
+                .map_err(|err| LanceError::from(ArrowError::from_external_error(Box::new(err))))?;
             while let Some(batch) = stream.try_next().await? {
                 let ids = column_as::<StringArray>(&batch, "id")?;
                 page_ids.extend((0..ids.len()).map(|row| ids.value(row).to_string()));
@@ -968,7 +999,7 @@ impl RolloutStore {
         let refs: Vec<&str> = columns.iter().map(String::as_str).collect();
         let scanner = self
             .lsm_scanner_for_source(ListSource::All, shard_snapshots)
-            .project(&refs);
+            .project(&refs)?;
         let mut stream = scanner.try_into_stream().await?;
 
         let mut batches: Vec<RecordBatch> = Vec::new();
@@ -1142,7 +1173,7 @@ impl RolloutStore {
         let refs: Vec<&str> = columns.iter().map(String::as_str).collect();
         let scanner = self
             .lsm_scanner_for_source(source, shard_snapshots)
-            .project(&refs)
+            .project(&refs)?
             .filter(&format!("id = '{}'", escaped_id))?;
         let mut stream = scanner.try_into_stream().await?;
         while let Some(batch) = stream.try_next().await? {
@@ -1705,6 +1736,13 @@ fn append_i8_list(builder: &mut ListBuilder<Int8Builder>, values: Option<&[i8]>)
     }
 }
 
+fn map_i64_bound(name: &str, value: Option<usize>) -> LanceResult<Option<i64>> {
+    value
+        .map(i64::try_from)
+        .transpose()
+        .map_err(|_| LanceError::invalid_input(format!("{name} exceeds i64::MAX")))
+}
+
 fn projected_dataset_schema(
     dataset: &Dataset,
     columns: &[String],

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

@@ -1222,7 +1222,7 @@ impl ContextStore {
             let scanner = self
                 .lsm_scanner()
                 .await?
-                .project(&["id", "content_type"])
+                .project(&["id", "content_type"])?
                 .filter(&filter)?;
             let mut stream = scanner.try_into_stream().await?;
             while let Some(batch) = stream.try_next().await? {
@@ -1245,7 +1245,7 @@ impl ContextStore {
             let scanner = self
                 .lsm_scanner()
                 .await?
-                .project(&["external_id", "content_type"])
+                .project(&["external_id", "content_type"])?
                 .filter(&filter)?;
             let mut stream = scanner.try_into_stream().await?;
             while let Some(batch) = stream.try_next().await? {
@@ -1417,6 +1417,12 @@ impl ContextStore {
             results.retain(|record| filters.matches(record));
         }
 
+        results.sort_by(|left, right| {
+            left.created_at
+                .cmp(&right.created_at)
+                .then_with(|| left.id.cmp(&right.id))
+        });
+
         if let Some(offset) = offset {
             results = results.into_iter().skip(offset).collect();
         }
@@ -1700,7 +1706,7 @@ impl ContextStore {
         }
         let columns = self.projected_columns(projection);
         let refs: Vec<&str> = columns.iter().map(String::as_str).collect();
-        Ok(scanner.project(&refs))
+        scanner.project(&refs)
     }
 
     /// Fetch a single record's `binary_payload` on demand, without loading it
@@ -1711,7 +1717,7 @@ impl ContextStore {
         let scanner = self
             .lsm_scanner()
             .await?
-            .project(&["id", "binary_payload"])
+            .project(&["id", "binary_payload"])?
             .filter(&filter)?;
         let mut stream = scanner.try_into_stream().await?;
         while let Some(batch) = stream.try_next().await? {