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();