mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-12 14:52:26 +00:00
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>
This commit is contained in:
lancedb-gatefixer[bot]
@@ -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"):
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user