mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-27 14:06:14 +00:00
fix: allow headerless empty remote blob probes
This commit is contained in:
@@ -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"):
|
||||
|
||||
@@ -100,18 +100,20 @@ struct RemoteBlobState {
|
||||
sequential_response: Option<SequentialResponse>,
|
||||
}
|
||||
|
||||
/// Seekable Cloud blob handle over HTTP Range, pinned to the version opened by its size probe.
|
||||
/// 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: u64,
|
||||
version: Option<u64>,
|
||||
}
|
||||
|
||||
impl RemoteBlobFile {
|
||||
fn new(requester: Arc<dyn BlobRangeRequester>, size: u64, version: u64) -> Self {
|
||||
fn new(requester: Arc<dyn BlobRangeRequester>, size: u64, version: Option<u64>) -> Self {
|
||||
Self {
|
||||
requester,
|
||||
state: Mutex::new(RemoteBlobState::default()),
|
||||
@@ -166,11 +168,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,
|
||||
Some(self.version),
|
||||
)
|
||||
.request_range(&range_header, RangeRequestMode::DataRead, self.version)
|
||||
.await
|
||||
.map_err(remote_blob_error)?;
|
||||
self.ensure_open()?;
|
||||
@@ -246,11 +244,7 @@ impl RemoteBlobFile {
|
||||
let range_header = format!("bytes={cursor}-");
|
||||
let (request_id, response) = self
|
||||
.requester
|
||||
.request_range(
|
||||
&range_header,
|
||||
RangeRequestMode::DataRead,
|
||||
Some(self.version),
|
||||
)
|
||||
.request_range(&range_header, RangeRequestMode::DataRead, self.version)
|
||||
.await
|
||||
.map_err(remote_blob_error)?;
|
||||
self.ensure_open()?;
|
||||
@@ -564,8 +558,8 @@ async fn probe_blob_file(requester: Arc<dyn BlobRangeRequester>) -> Result<Optio
|
||||
});
|
||||
}
|
||||
}
|
||||
let version = probe_blob_version(&response, &request_id)?;
|
||||
Ok(Some(RemoteBlobFile::new(requester, 0, version).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
|
||||
@@ -594,7 +588,9 @@ async fn probe_blob_file(requester: Arc<dyn BlobRangeRequester>) -> Result<Optio
|
||||
status_code: Some(StatusCode::PARTIAL_CONTENT),
|
||||
});
|
||||
}
|
||||
Ok(Some(RemoteBlobFile::new(requester, size, version).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}")
|
||||
@@ -629,7 +625,6 @@ mod tests {
|
||||
http::Response::builder()
|
||||
.status(StatusCode::RANGE_NOT_SATISFIABLE)
|
||||
.header(header::CONTENT_RANGE, "bytes */0")
|
||||
.header(VERSION_HEADER, "5")
|
||||
.body(Vec::new())
|
||||
.unwrap()
|
||||
}
|
||||
@@ -855,7 +850,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());
|
||||
|
||||
@@ -868,8 +863,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);
|
||||
}
|
||||
@@ -1356,7 +1353,7 @@ mod tests {
|
||||
in_flight,
|
||||
max_in_flight: max_in_flight.clone(),
|
||||
});
|
||||
let file = RemoteBlobFile::new(requester, PAYLOAD.len() as u64, 5);
|
||||
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();
|
||||
@@ -1470,7 +1467,11 @@ mod tests {
|
||||
release: release.clone(),
|
||||
started: started.clone(),
|
||||
});
|
||||
let file = Arc::new(RemoteBlobFile::new(requester, PAYLOAD.len() as u64, 5));
|
||||
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