From 51ea02ea8c23c7addd7f0586aad5bedd6d08a15b Mon Sep 17 00:00:00 2001
From: "lancedb-gatefixer[bot]"
 <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>
Date: Tue, 29 Sep 2026 08:35:19 -0500
Subject: [PATCH] fix: pin remote BlobFile reads to the opened version (#4295)

## Cause

On latest-tracking remote tables, `fetch_blob_files` probed a concrete
table version but retained `None` on each handle. Later range reads
resolved the latest version again, so an update or delete could make an
opened handle fail.

## Fix

For nonempty blobs, capture `x-lancedb-version` from the 206 size probe
and include it on every later range request. Reject a nonempty probe
without a valid version. Accept an empty-blob 416 probe with or without
that header, because an empty handle makes no later data range request.
The Python remote mock now covers both response shapes.

## Limitation

A table dropped and recreated with the same name can reuse version
numbers and row addresses. Version pinning does not identify the table
itself, so an old handle can still return data from the recreated table.
That identity case needs a separate fix. Callers that replace a table
should discard its old blob handles.

## Validation

- `cargo test --quiet --features remote -p lancedb --lib
remote::table::blobs::tests` (24 passed)
- Rebuilt the Python extension; `uv run --extra tests pytest
python/tests/test_remote_db.py -q -k remote_blob` (6 passed)
- `cargo clippy --quiet --features remote -p lancedb --lib --tests
--examples`
- `cargo fmt --all -- --check`, `ruff check .`, and `ruff format .`

Fixes #4276

<!-- lance-gatekeeper-fix:v1 agent=e309bdf75b1a87f6492a6dc3ee1bfd81
generation=1 -->

---------

Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>
---
 python/python/tests/test_remote_db.py  |  20 +++-
 rust/lancedb/src/remote/table.rs       |   1 +
 rust/lancedb/src/remote/table/blobs.rs | 147 +++++++++++++++++++++++--
 3 files changed, 152 insertions(+), 16 deletions(-)

diff --git a/python/python/tests/test_remote_db.py b/python/python/tests/test_remote_db.py
index e32dbe8ca..cd4b78a1b 100644
--- a/python/python/tests/test_remote_db.py
+++ b/python/python/tests/test_remote_db.py
@@ -2403,11 +2403,16 @@ def blob_remote_table(*, server_version=Version("0.5.0")):
         elif request.path.startswith("/v1/table/test/blob/image/"):
             path = request.path.partition("?")[0]
             row_id = int(path.split("/")[-2])
-            payload = {10: b"alpha", 20: None, 30: b"gamma"}[row_id]
+            payload = {10: b"alpha", 20: None, 30: b"gamma", 40: b""}[row_id]
             if payload is None:
                 request.send_response(204)
                 request.end_headers()
                 return
+            if not payload:
+                request.send_response(416)
+                request.send_header("Content-Range", "bytes */0")
+                request.end_headers()
+                return
             byte_range = request.headers["Range"].removeprefix("bytes=")
             start_text, end_text = byte_range.split("-", maxsplit=1)
             start = int(start_text)
@@ -2416,6 +2421,9 @@ def blob_remote_table(*, server_version=Version("0.5.0")):
             request.send_response(206)
             request.send_header("Content-Range", f"bytes {start}-{end}/{len(payload)}")
             request.send_header("Content-Length", str(len(chunk)))
+            request.send_header(
+                "x-lancedb-version", str(BLOB_DESCRIBE_RESPONSE["version"])
+            )
             request.end_headers()
             request.wfile.write(chunk)
         elif request.path == "/v1/table/test/query/":
@@ -2459,17 +2467,21 @@ def test_remote_blob_columns_and_fetch():
 
 def test_remote_blob_files_are_lazy_seekable_handles():
     with blob_remote_table() as table:
-        files = table.fetch_blob_files("image", [10, 20, 30])
+        files = table.fetch_blob_files("image", [10, 20, 30, 40])
 
-        assert len(files) == 3
-        alpha, null_row, gamma = files
+        assert len(files) == 4
+        alpha, null_row, gamma, empty = files
         assert null_row is None
         assert alpha is not None
         assert gamma is not None
+        assert empty is not None
         assert alpha.size() == 5
         assert alpha.read_range(1, 3) == b"lph"
         gamma.seek(2)
         assert gamma.read() == b"mma"
+        assert empty.size() == 0
+        assert empty.read() == b""
+        assert empty.read_range(0, 0) == b""
         alpha.close()
         assert alpha.closed
         with pytest.raises(RuntimeError, match="already closed"):
diff --git a/rust/lancedb/src/remote/table.rs b/rust/lancedb/src/remote/table.rs
index 195c18ea5..238187931 100644
--- a/rust/lancedb/src/remote/table.rs
+++ b/rust/lancedb/src/remote/table.rs
@@ -5326,6 +5326,7 @@ mod tests {
                     "bytes=0-0" => http::Response::builder()
                         .status(206)
                         .header(reqwest::header::CONTENT_RANGE, "bytes 0-0/10")
+                        .header(VERSION_HEADER, "42")
                         .body(b"0".to_vec())
                         .unwrap(),
                     "bytes=0-" => http::Response::builder()
diff --git a/rust/lancedb/src/remote/table/blobs.rs b/rust/lancedb/src/remote/table/blobs.rs
index 29023eeed..07aeeae63 100644
--- a/rust/lancedb/src/remote/table/blobs.rs
+++ b/rust/lancedb/src/remote/table/blobs.rs
@@ -22,7 +22,8 @@ use crate::remote::client::{HttpSend, RequestResultExt, RestfulLanceDbClient};
 use crate::table::BaseTable;
 
 use super::{
-    FreshnessHeaders, FreshnessState, ReadSnapshot, RemoteTable, freshness_headers_snapshot,
+    FreshnessHeaders, FreshnessState, ReadSnapshot, RemoteTable, VERSION_HEADER,
+    freshness_headers_snapshot,
 };
 
 // The Cloud route rejects larger row-id lists. Its separate 64 MiB byte limit
@@ -50,6 +51,7 @@ trait BlobRangeRequester: Send + Sync + std::fmt::Debug {
         &self,
         range_header: &str,
         mode: RangeRequestMode,
+        version: Option<u64>,
     ) -> Result<(String, Response)>;
 }
 
@@ -71,13 +73,14 @@ impl<S: HttpSend> BlobRangeRequester for TableBlobRangeRequester<S> {
         &self,
         range_header: &str,
         mode: RangeRequestMode,
+        version: Option<u64>,
     ) -> Result<(String, Response)> {
         let freshness_request =
             freshness_headers_snapshot(&self.freshness, self.read_consistency_interval);
         let mut request = freshness_request
             .apply(self.client.get(&self.path))
             .header(header::RANGE, range_header);
-        if let Some(version) = self.version {
+        if let Some(version) = version.or(self.version) {
             request = request.query(&[("version", version)]);
         }
         if let Some(branch) = &self.branch {
@@ -112,21 +115,25 @@ struct RemoteBlobState {
 }
 
 /// Seekable Cloud blob handle over HTTP Range.
+///
+/// Nonempty handles read the table version returned by their size probe.
 #[derive(Debug)]
 pub struct RemoteBlobFile {
     requester: Arc<dyn BlobRangeRequester>,
     state: Mutex<RemoteBlobState>,
     closed: AtomicBool,
     size: u64,
+    version: Option<u64>,
 }
 
 impl RemoteBlobFile {
-    fn new(requester: Arc<dyn BlobRangeRequester>, size: u64) -> Self {
+    fn new(requester: Arc<dyn BlobRangeRequester>, size: u64, version: Option<u64>) -> Self {
         Self {
             requester,
             state: Mutex::new(RemoteBlobState::default()),
             closed: AtomicBool::new(false),
             size,
+            version,
         }
     }
 
@@ -175,7 +182,7 @@ impl RemoteBlobFile {
         let range_header = format!("bytes={}-{}", range.start, range.end - 1);
         let (request_id, response) = self
             .requester
-            .request_range(&range_header, RangeRequestMode::DataRead)
+            .request_range(&range_header, RangeRequestMode::DataRead, self.version)
             .await
             .map_err(remote_blob_error)?;
         self.ensure_open()?;
@@ -251,7 +258,7 @@ impl RemoteBlobFile {
                 let range_header = format!("bytes={cursor}-");
                 let (request_id, response) = self
                     .requester
-                    .request_range(&range_header, RangeRequestMode::DataRead)
+                    .request_range(&range_header, RangeRequestMode::DataRead, self.version)
                     .await
                     .map_err(remote_blob_error)?;
                 self.ensure_open()?;
@@ -579,12 +586,25 @@ async fn probe_blob_files(
 
 const BLOB_REQUEST_CONCURRENCY: usize = 8;
 
+fn probe_blob_version(response: &Response, request_id: &str) -> Result<u64> {
+    response
+        .headers()
+        .get(&VERSION_HEADER)
+        .and_then(|value| value.to_str().ok())
+        .and_then(|value| value.parse().ok())
+        .ok_or_else(|| Error::Http {
+            source: "blob size probe returned a missing or invalid x-lancedb-version header".into(),
+            request_id: request_id.to_string(),
+            status_code: Some(response.status()),
+        })
+}
+
 /// Probe one blob's size.
 ///
 /// `204` represents null. `416` with `bytes */0` represents an empty blob.
 async fn probe_blob_file(requester: Arc<dyn BlobRangeRequester>) -> Result<Option<BlobFile>> {
     let (request_id, response) = requester
-        .request_range("bytes=0-0", RangeRequestMode::SizeProbe)
+        .request_range("bytes=0-0", RangeRequestMode::SizeProbe, None)
         .await?;
     match response.status() {
         StatusCode::NO_CONTENT => {
@@ -617,7 +637,8 @@ async fn probe_blob_file(requester: Arc<dyn BlobRangeRequester>) -> Result<Optio
                     });
                 }
             }
-            Ok(Some(RemoteBlobFile::new(requester, 0).into()))
+            // An empty handle never makes a data range request, so no version is needed.
+            Ok(Some(RemoteBlobFile::new(requester, 0, None).into()))
         }
         StatusCode::PARTIAL_CONTENT => {
             let size = response
@@ -633,6 +654,7 @@ async fn probe_blob_file(requester: Arc<dyn BlobRangeRequester>) -> Result<Optio
                     request_id: request_id.clone(),
                     status_code: Some(StatusCode::PARTIAL_CONTENT),
                 })?;
+            let version = probe_blob_version(&response, &request_id)?;
             let probe_body = response.bytes().await.err_to_http(request_id.clone())?;
             if probe_body.len() != 1 {
                 return Err(Error::Http {
@@ -645,7 +667,9 @@ async fn probe_blob_file(requester: Arc<dyn BlobRangeRequester>) -> Result<Optio
                     status_code: Some(StatusCode::PARTIAL_CONTENT),
                 });
             }
-            Ok(Some(RemoteBlobFile::new(requester, size).into()))
+            Ok(Some(
+                RemoteBlobFile::new(requester, size, Some(version)).into(),
+            ))
         }
         status => Err(Error::Http {
             source: format!("blob size probe expected HTTP 206 Partial Content, got {status}")
@@ -658,7 +682,7 @@ async fn probe_blob_file(requester: Arc<dyn BlobRangeRequester>) -> Result<Optio
 
 #[cfg(test)]
 mod tests {
-    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
     use std::sync::{Arc, Mutex as StdMutex};
     use std::time::Duration;
 
@@ -708,6 +732,7 @@ mod tests {
                 header::CONTENT_RANGE,
                 format!("bytes {start}-{end}/{}", payload.len()),
             )
+            .header(VERSION_HEADER, "5")
             .body(payload[start..=end].to_vec())
             .unwrap()
     }
@@ -778,6 +803,68 @@ mod tests {
         assert!(requests.lock().unwrap().contains(&"bytes=5-11".to_string()));
     }
 
+    #[tokio::test]
+    async fn remote_blob_file_reads_the_probed_version_after_table_changes() {
+        let changed = Arc::new(AtomicBool::new(false));
+        let requests = Arc::new(StdMutex::new(Vec::new()));
+        let table = RemoteTable::new_mock(
+            "my_table".to_string(),
+            {
+                let changed = changed.clone();
+                let requests = requests.clone();
+                move |request| {
+                    let range = request
+                        .headers()
+                        .get(header::RANGE)
+                        .unwrap()
+                        .to_str()
+                        .unwrap();
+                    let version = request
+                        .url()
+                        .query_pairs()
+                        .find(|(name, _)| name == "version")
+                        .map(|(_, value)| value.into_owned());
+                    requests
+                        .lock()
+                        .unwrap()
+                        .push((range.to_string(), version.clone()));
+                    if changed.load(Ordering::SeqCst) && version.as_deref() != Some("5") {
+                        return http::Response::builder()
+                            .status(StatusCode::BAD_REQUEST)
+                            .body(b"row id is absent from the latest version".to_vec())
+                            .unwrap();
+                    }
+                    range_response(&request, PAYLOAD)
+                }
+            },
+            Some(Version::new(0, 5, 0)),
+        );
+
+        let file = table
+            .fetch_blob_files_impl("image", &[10])
+            .await
+            .unwrap()
+            .pop()
+            .flatten()
+            .unwrap();
+        changed.store(true, Ordering::SeqCst);
+
+        assert_eq!(file.read_range(5..9).await.unwrap(), &PAYLOAD[5..9]);
+        assert_eq!(file.read_up_to(4).await.unwrap(), &PAYLOAD[..4]);
+        file.seek(10).await.unwrap();
+        assert_eq!(file.read().await.unwrap(), &PAYLOAD[10..]);
+
+        assert_eq!(
+            requests.lock().unwrap().as_slice(),
+            [
+                ("bytes=0-0".to_string(), None),
+                ("bytes=5-8".to_string(), Some("5".to_string())),
+                ("bytes=0-".to_string(), Some("5".to_string())),
+                ("bytes=10-".to_string(), Some("5".to_string())),
+            ]
+        );
+    }
+
     #[tokio::test]
     async fn remote_blob_file_keeps_the_open_timeline_after_parent_checkout() {
         let range_requests = Arc::new(StdMutex::new(Vec::new()));
@@ -842,7 +929,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn remote_blob_files_return_a_handle_for_an_empty_blob() {
+    async fn remote_blob_files_allow_an_empty_blob_without_a_version_header() {
         let requests = Arc::new(StdMutex::new(Vec::new()));
         let table = mock_remote_blob_table(requests.clone());
 
@@ -855,8 +942,10 @@ mod tests {
         assert!(files[1].is_none());
         let empty = files[2].take().unwrap();
         assert_eq!(empty.size(), 0);
+        let probe_requests = requests.lock().unwrap().len();
         assert!(empty.read_range(0..0).await.unwrap().is_empty());
         assert!(empty.read().await.unwrap().is_empty());
+        assert_eq!(requests.lock().unwrap().len(), probe_requests);
         let nonempty = files[0].take().unwrap();
         assert_eq!(nonempty.size(), PAYLOAD.len() as u64);
     }
@@ -914,6 +1003,30 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn remote_blob_files_reject_a_probe_without_a_version() {
+        let table = RemoteTable::new_mock(
+            "my_table".to_string(),
+            |request| {
+                let mut response = range_response(&request, PAYLOAD);
+                response.headers_mut().remove(VERSION_HEADER);
+                response
+            },
+            Some(Version::new(0, 5, 0)),
+        );
+
+        let error = table
+            .fetch_blob_files_impl("image", &[10])
+            .await
+            .unwrap_err();
+        assert!(
+            error
+                .to_string()
+                .contains("missing or invalid x-lancedb-version"),
+            "got: {error}"
+        );
+    }
+
     #[tokio::test]
     async fn remote_blob_files_reject_unsupported_server_version() {
         let table = RemoteTable::new_mock(
@@ -1105,6 +1218,7 @@ mod tests {
             &self,
             range_header: &str,
             _mode: RangeRequestMode,
+            _version: Option<u64>,
         ) -> Result<(String, Response)> {
             assert_eq!(range_header, "bytes=0-0");
             let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1;
@@ -1119,6 +1233,7 @@ mod tests {
                     header::CONTENT_RANGE,
                     format!("bytes 0-0/{}", 100 + self.index),
                 )
+                .header(VERSION_HEADER, "5")
                 .body(vec![0u8])
                 .unwrap();
             Ok((format!("probe-{}", self.index), Response::from(response)))
@@ -1237,6 +1352,7 @@ mod tests {
                             header::CONTENT_RANGE,
                             format!("bytes 0-0/{}", PAYLOAD.len()),
                         )
+                        .header(VERSION_HEADER, "5")
                         .body(vec![PAYLOAD[0]])
                         .unwrap();
                 }
@@ -1283,6 +1399,7 @@ mod tests {
             &self,
             range_header: &str,
             _mode: RangeRequestMode,
+            _version: Option<u64>,
         ) -> Result<(String, Response)> {
             let now = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1;
             self.max_in_flight.fetch_max(now, Ordering::SeqCst);
@@ -1315,7 +1432,7 @@ mod tests {
             in_flight,
             max_in_flight: max_in_flight.clone(),
         });
-        let file = RemoteBlobFile::new(requester, PAYLOAD.len() as u64);
+        let file = RemoteBlobFile::new(requester, PAYLOAD.len() as u64, Some(5));
 
         let ranges: Vec<_> = (0..16u64).map(|start| start..start + 2).collect();
         let output = file.read_ranges(&ranges).await.unwrap();
@@ -1372,6 +1489,7 @@ mod tests {
                             header::CONTENT_RANGE,
                             format!("bytes 0-0/{}", PAYLOAD.len()),
                         )
+                        .header(VERSION_HEADER, "5")
                         .body(vec![PAYLOAD[0]])
                         .unwrap();
                 }
@@ -1402,6 +1520,7 @@ mod tests {
             &self,
             range_header: &str,
             _mode: RangeRequestMode,
+            _version: Option<u64>,
         ) -> Result<(String, Response)> {
             if range_header.ends_with('-') {
                 self.started.notify_one();
@@ -1427,7 +1546,11 @@ mod tests {
             release: release.clone(),
             started: started.clone(),
         });
-        let file = Arc::new(RemoteBlobFile::new(requester, PAYLOAD.len() as u64));
+        let file = Arc::new(RemoteBlobFile::new(
+            requester,
+            PAYLOAD.len() as u64,
+            Some(5),
+        ));
 
         let reader = {
             let file = file.clone();