Compare commits

..

1 Commits

Author SHA1 Message Date
lancedb automation 667cf32e78 chore: update lance dependency to v10.1.0-beta.2 2026-08-02 00:12:47 +00:00
21 changed files with 433 additions and 2324 deletions
Generated
+298 -260
View File
File diff suppressed because it is too large Load Diff
+14 -14
View File
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
rust-version = "1.91.0"
[workspace.dependencies]
lance = { "version" = "=10.1.0-beta.1", default-features = false, "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance-core = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance-datagen = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance-file = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance-io = { "version" = "=10.1.0-beta.1", default-features = false, "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance-index = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance-linalg = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace-impls = { "version" = "=10.1.0-beta.1", default-features = false, "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance-table = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance-testing = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance-datafusion = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance-encoding = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance-arrow = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
lance = { "version" = "=10.1.0-beta.2", default-features = false, "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-core = { "version" = "=10.1.0-beta.2", "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-datagen = { "version" = "=10.1.0-beta.2", "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-file = { "version" = "=10.1.0-beta.2", "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-io = { "version" = "=10.1.0-beta.2", default-features = false, "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-index = { "version" = "=10.1.0-beta.2", "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-linalg = { "version" = "=10.1.0-beta.2", "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace = { "version" = "=10.1.0-beta.2", "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-namespace-impls = { "version" = "=10.1.0-beta.2", default-features = false, "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-table = { "version" = "=10.1.0-beta.2", "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-testing = { "version" = "=10.1.0-beta.2", "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-datafusion = { "version" = "=10.1.0-beta.2", "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-encoding = { "version" = "=10.1.0-beta.2", "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
lance-arrow = { "version" = "=10.1.0-beta.2", "tag" = "v10.1.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
ahash = "0.8"
# Note that this one does not include pyarrow
arrow = { version = "58.0.0", optional = false }
+1 -1
View File
@@ -28,7 +28,7 @@
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<arrow.version>15.0.0</arrow.version>
<lance-core.version>10.1.0-beta.1</lance-core.version>
<lance-core.version>10.1.0-beta.2</lance-core.version>
<spotless.skip>false</spotless.skip>
<spotless.version>2.30.0</spotless.version>
<spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>
+2 -6
View File
@@ -339,9 +339,7 @@ impl Table {
let transforms = NewColumnTransform::SqlExpressions(transforms);
let res = self
.inner_ref()?
.add_columns()
.transform(transforms)
.execute()
.add_columns(transforms, None)
.await
.default_error()?;
Ok(res.into())
@@ -358,9 +356,7 @@ impl Table {
let transforms = NewColumnTransform::AllNulls(schema);
let res = self
.inner_ref()?
.add_columns()
.transform(transforms)
.execute()
.add_columns(transforms, None)
.await
.default_error()?;
Ok(res.into())
+1 -3
View File
@@ -92,10 +92,8 @@ class LanceMergeInsertBuilder(object):
self._when_not_matched_by_source_delete = True
if isinstance(condition, Expr):
self._when_not_matched_by_source_condition_expr = condition._inner
self._when_not_matched_by_source_condition = None
else:
elif condition is not None:
self._when_not_matched_by_source_condition = condition
self._when_not_matched_by_source_condition_expr = None
return self
def use_index(self, use_index: bool) -> LanceMergeInsertBuilder:
+1 -1
View File
@@ -2697,7 +2697,7 @@ class LanceHybridQueryBuilder(LanceQueryBuilder):
self._fts_query.phrase_query(True)
if self._distance_type:
self._vector_query.metric(self._distance_type)
if self._minimum_nprobes is not None:
if self._minimum_nprobes:
self._vector_query.minimum_nprobes(self._minimum_nprobes)
if self._maximum_nprobes is not None:
self._vector_query.maximum_nprobes(self._maximum_nprobes)
+2 -2
View File
@@ -1595,8 +1595,8 @@ class Table(ABC):
Prefer this over :meth:`fetch_blobs` for large payloads. ``row_ids`` is
a ``list[int]`` or a query ``pyarrow.Table`` carrying row identity via
``_rowid`` or a ``_lance_row_id`` field on the blob descriptor. Null
rows are ``None``. Remote tables require LanceDB Cloud server 0.5.0 or
newer.
rows are ``None``. Unsupported on LanceDB Cloud, where
:meth:`fetch_blobs` returns full bytes instead.
"""
@abstractmethod
-13
View File
@@ -123,19 +123,6 @@ async def test_async_hybrid_query_default_limit(table: AsyncTable):
assert texts.count("a") == 1
def test_hybrid_query_minimum_nprobes_zero_raises(sync_table: Table):
# minimum_nprobes(0) must raise the same validation error a plain vector
# query raises, not silently no-op because 0 is falsy.
with pytest.raises(ValueError, match="minimum_nprobes must be greater than 0"):
(
sync_table.search(query_type="hybrid")
.vector([0.0, 0.4])
.text("dog")
.minimum_nprobes(0)
.to_arrow()
)
def test_hybrid_query_distance_range(sync_table: Table):
reranker = RRFReranker(return_score="all")
result = (
+2 -33
View File
@@ -2055,24 +2055,6 @@ def blob_remote_table(*, server_version=Version("0.5.0")):
request.send_header("phalanx-version", str(server_version))
request.end_headers()
request.wfile.write(json.dumps(BLOB_DESCRIBE_RESPONSE).encode())
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]
if payload is None:
request.send_response(204)
request.end_headers()
return
byte_range = request.headers["Range"].removeprefix("bytes=")
start_text, end_text = byte_range.split("-", maxsplit=1)
start = int(start_text)
end = int(end_text) if end_text else len(payload) - 1
chunk = payload[start : end + 1]
request.send_response(206)
request.send_header("Content-Range", f"bytes {start}-{end}/{len(payload)}")
request.send_header("Content-Length", str(len(chunk)))
request.end_headers()
request.wfile.write(chunk)
elif request.path == "/v1/table/test/query/":
content_len = int(request.headers.get("Content-Length", 0))
body = json.loads(request.rfile.read(content_len))
@@ -2110,21 +2092,8 @@ def test_remote_blob_columns_and_fetch():
assert table.blob_columns() == ["image"]
blobs = table.fetch_blobs("image", [10, 20, 30])
assert blobs.to_pylist() == [b"alpha", None, b"gamma"]
def test_remote_blob_files_are_lazy_seekable_handles():
with blob_remote_table() as table:
files = table.fetch_blob_files("image", [10, 20, 30])
assert len(files) == 3
alpha, null_row, gamma = files
assert null_row is None
assert alpha is not None
assert gamma is not None
assert alpha.size() == 5
assert alpha.read_range(1, 3) == b"lph"
gamma.seek(2)
assert gamma.read() == b"mma"
with pytest.raises(NotImplementedError, match="Use fetch_blobs for full bytes"):
table.fetch_blob_files("image", [10, 20, 30])
def test_remote_blob_fetch_accepts_query_table():
-62
View File
@@ -2341,45 +2341,6 @@ def test_merge_insert(mem_db: DBConnection):
)
def test_merge_insert_with_null_on_column(mem_db: DBConnection):
table = mem_db.create_table(
"users",
data=pa.table(
{
"id": [0, 1],
"name": ["Alice", "Bob"],
"record_type": [None, "personal"],
}
),
)
new_data = pa.table(
{
"id": [1, 2],
"name": ["Bobby", "Charlie"],
"record_type": ["personal", None],
}
)
result = (
table.merge_insert(["id", "record_type"])
.when_matched_update_all()
.when_not_matched_insert_all()
.execute(new_data)
)
assert result.num_inserted_rows == 1
assert result.num_updated_rows == 1
assert result.num_deleted_rows == 0
expected = pa.table(
{
"id": [0, 1, 2],
"name": ["Alice", "Bobby", "Charlie"],
"record_type": [None, "personal", None],
}
)
assert table.to_arrow().sort_by("id") == expected
def test_merge_insert_by_source_delete_expr(mem_db: DBConnection):
table = mem_db.create_table(
"my_table",
@@ -2403,29 +2364,6 @@ def test_merge_insert_by_source_delete_expr(mem_db: DBConnection):
assert table.to_arrow().sort_by("a") == expected
def test_merge_insert_by_source_delete_reconfigure(mem_db: DBConnection):
# Calling when_not_matched_by_source_delete() again with no condition must
# widen the delete to unconditional, not keep the earlier condition around.
table = mem_db.create_table(
"my_table",
data=pa.table({"a": [1, 2, 3], "b": ["a", "b", "c"]}),
)
new_data = pa.table({"a": [2, 4], "b": ["x", "z"]})
merge_insert_res = (
table.merge_insert("a")
.when_matched_update_all()
.when_not_matched_insert_all()
.when_not_matched_by_source_delete("a > 2")
.when_not_matched_by_source_delete()
.execute(new_data)
)
assert merge_insert_res.num_deleted_rows == 2
expected = pa.table({"a": [2, 4], "b": ["x", "z"]})
assert table.to_arrow().sort_by("a") == expected
@pytest.mark.asyncio
async def test_merge_insert_by_source_delete_expr_async(
mem_db_async: AsyncConnection,
+13 -37
View File
@@ -426,11 +426,9 @@ pub struct PyBlobFile {
impl PyBlobFile {
fn read_bytes(self_: PyRef<'_, Self>) -> PyResult<Py<PyBytes>> {
let inner = self_.inner.clone();
let py = self_.py();
let bytes = py
.detach(move || block_on(async move { inner.read().await }))
let bytes = block_on(async move { inner.read().await })
.map_err(|e| PyRuntimeError::new_err(format!("blob read failed: {e}")))?;
Ok(PyBytes::new(py, bytes.as_ref()).unbind())
Ok(PyBytes::new(self_.py(), bytes.as_ref()).unbind())
}
pub fn read(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
@@ -446,32 +444,24 @@ impl PyBlobFile {
fn close(self_: PyRef<'_, Self>) -> PyResult<()> {
let inner = self_.inner.clone();
self_
.py()
.detach(move || block_on(async move { inner.close().await }))
block_on(async move { inner.close().await })
.map_err(|e| PyRuntimeError::new_err(format!("blob close failed: {e}")))
}
fn is_closed(self_: PyRef<'_, Self>) -> bool {
let inner = self_.inner.clone();
self_
.py()
.detach(move || block_on(async move { inner.is_closed().await }))
block_on(async move { inner.is_closed().await })
}
fn seek(self_: PyRef<'_, Self>, position: u64) -> PyResult<()> {
let inner = self_.inner.clone();
self_
.py()
.detach(move || block_on(async move { inner.seek(position).await }))
block_on(async move { inner.seek(position).await })
.map_err(|e| PyRuntimeError::new_err(format!("blob seek failed: {e}")))
}
fn tell(self_: PyRef<'_, Self>) -> PyResult<u64> {
let inner = self_.inner.clone();
self_
.py()
.detach(move || block_on(async move { inner.tell().await }))
block_on(async move { inner.tell().await })
.map_err(|e| PyRuntimeError::new_err(format!("blob tell failed: {e}")))
}
@@ -485,20 +475,16 @@ impl PyBlobFile {
.checked_add(length as u64)
.ok_or_else(|| PyValueError::new_err("offset + length overflowed"))?;
let inner = self_.inner.clone();
let py = self_.py();
let bytes = py
.detach(move || block_on(async move { inner.read_range(offset..end).await }))
let bytes = block_on(async move { inner.read_range(offset..end).await })
.map_err(|e| PyRuntimeError::new_err(format!("blob read_range failed: {e}")))?;
Ok(PyBytes::new(py, bytes.as_ref()).unbind())
Ok(PyBytes::new(self_.py(), bytes.as_ref()).unbind())
}
fn read_up_to(self_: PyRef<'_, Self>, length: usize) -> PyResult<Py<PyBytes>> {
let inner = self_.inner.clone();
let py = self_.py();
let bytes = py
.detach(move || block_on(async move { inner.read_up_to(length).await }))
.map_err(|e| PyRuntimeError::new_err(format!("blob read_up_to failed: {e}")))?;
Ok(PyBytes::new(py, bytes.as_ref()).unbind())
let bytes = block_on(async move { inner.read_up_to(length).await })
.map_err(|e| PyRuntimeError::new_err(format!("blob read failed: {e}")))?;
Ok(PyBytes::new(self_.py(), bytes.as_ref()).unbind())
}
}
@@ -1375,12 +1361,7 @@ impl Table {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let result = inner
.add_columns()
.transform(definitions)
.execute()
.await
.infer_error()?;
let result = inner.add_columns(definitions, None).await.infer_error()?;
Ok(AddColumnsResult::from(result))
})
}
@@ -1394,12 +1375,7 @@ impl Table {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let result = inner
.add_columns()
.transform(transform)
.execute()
.await
.infer_error()?;
let result = inner.add_columns(transform, None).await.infer_error()?;
Ok(AddColumnsResult::from(result))
})
}
+2 -2
View File
@@ -49,8 +49,8 @@ lance-namespace = { workspace = true }
lance-namespace-impls = { workspace = true }
metrics = { workspace = true, optional = true }
metrics-util = { workspace = true, optional = true }
# Pin the transitive GooseFS SDK until the 0.1.6 compile break is fixed upstream.
goosefs-sdk = { version = "=0.1.5", optional = true }
# Keep the direct dependency aligned with the version required by OpenDAL.
goosefs-sdk = { version = "=0.1.8", optional = true }
moka = { workspace = true }
pin-project = { workspace = true }
tokio = { version = "1.23", features = ["rt-multi-thread", "sync"] }
+2 -198
View File
@@ -9,7 +9,6 @@
//!
//! Blob tables require Lance file format >= 2.2 and stable row ids at create.
use std::ops::Range;
use std::sync::Arc;
use arrow_array::LargeBinaryArray;
@@ -18,202 +17,10 @@ use arrow_schema::{DataType, Field, Schema};
use lance::dataset::{BlobRangeRequest as LanceBlobRangeRequest, Dataset, WriteParams};
use lance_arrow::FieldExt;
use lance_encoding::version::LanceFileVersion;
use lance_io::object_store::ObjectStore;
use object_store::path::Path;
use crate::error::{Error, Result};
/// Seekable handle for one blob value, backed by local storage or a remote
/// HTTP byte-range endpoint.
#[derive(Debug)]
pub struct BlobFile {
inner: BlobFileInner,
}
#[derive(Debug)]
enum BlobFileInner {
Native(lance::dataset::BlobFile),
#[cfg(feature = "remote")]
Remote(Box<crate::remote::table::blobs::RemoteBlobFile>),
}
impl From<lance::dataset::BlobFile> for BlobFile {
fn from(value: lance::dataset::BlobFile) -> Self {
Self {
inner: BlobFileInner::Native(value),
}
}
}
#[cfg(feature = "remote")]
impl From<crate::remote::table::blobs::RemoteBlobFile> for BlobFile {
fn from(value: crate::remote::table::blobs::RemoteBlobFile) -> Self {
Self {
inner: BlobFileInner::Remote(Box::new(value)),
}
}
}
impl BlobFile {
/// Inline reader over a data-file slice.
pub fn new_inline(
object_store: Arc<ObjectStore>,
path: Path,
position: u64,
size: u64,
) -> Self {
lance::dataset::BlobFile::new_inline(object_store, path, position, size).into()
}
/// Dedicated sidecar-file reader.
pub fn new_dedicated(object_store: Arc<ObjectStore>, path: Path, size: u64) -> Self {
lance::dataset::BlobFile::new_dedicated(object_store, path, size).into()
}
/// Packed reader for a slice in a shared sidecar.
pub fn new_packed(
object_store: Arc<ObjectStore>,
path: Path,
position: u64,
size: u64,
) -> Self {
lance::dataset::BlobFile::new_packed(object_store, path, position, size).into()
}
/// External reader at a resolved object location.
pub fn new_external(
object_store: Arc<ObjectStore>,
path: Path,
uri: String,
position: u64,
size: u64,
) -> Self {
lance::dataset::BlobFile::new_external(object_store, path, uri, position, size).into()
}
/// Close the handle.
pub async fn close(&self) -> lance_core::Result<()> {
match &self.inner {
BlobFileInner::Native(file) => file.close().await,
#[cfg(feature = "remote")]
BlobFileInner::Remote(file) => file.close().await,
}
}
/// Whether the handle is closed.
pub async fn is_closed(&self) -> bool {
match &self.inner {
BlobFileInner::Native(file) => file.is_closed().await,
#[cfg(feature = "remote")]
BlobFileInner::Remote(file) => file.is_closed(),
}
}
/// Read a range without moving the cursor.
pub async fn read_range(&self, range: Range<u64>) -> lance_core::Result<bytes::Bytes> {
match &self.inner {
BlobFileInner::Native(file) => file.read_range(range).await,
#[cfg(feature = "remote")]
BlobFileInner::Remote(file) => file.read_range(range).await,
}
}
/// Read ranges without moving the cursor.
pub async fn read_ranges(
&self,
ranges: &[Range<u64>],
) -> lance_core::Result<Vec<bytes::Bytes>> {
match &self.inner {
BlobFileInner::Native(file) => file.read_ranges(ranges).await,
#[cfg(feature = "remote")]
BlobFileInner::Remote(file) => file.read_ranges(ranges).await,
}
}
/// Read from the cursor to the end.
pub async fn read(&self) -> lance_core::Result<bytes::Bytes> {
match &self.inner {
BlobFileInner::Native(file) => file.read().await,
#[cfg(feature = "remote")]
BlobFileInner::Remote(file) => file.read().await,
}
}
/// Read up to `len` bytes and advance the cursor.
pub async fn read_up_to(&self, len: usize) -> lance_core::Result<bytes::Bytes> {
match &self.inner {
BlobFileInner::Native(file) => file.read_up_to(len).await,
#[cfg(feature = "remote")]
BlobFileInner::Remote(file) => file.read_up_to(len).await,
}
}
/// Move the cursor to `new_cursor`.
pub async fn seek(&self, new_cursor: u64) -> lance_core::Result<()> {
match &self.inner {
BlobFileInner::Native(file) => file.seek(new_cursor).await,
#[cfg(feature = "remote")]
BlobFileInner::Remote(file) => file.seek(new_cursor).await,
}
}
/// Current cursor position.
pub async fn tell(&self) -> lance_core::Result<u64> {
match &self.inner {
BlobFileInner::Native(file) => file.tell().await,
#[cfg(feature = "remote")]
BlobFileInner::Remote(file) => file.tell().await,
}
}
/// Blob length in bytes.
pub fn size(&self) -> u64 {
match &self.inner {
BlobFileInner::Native(file) => file.size(),
#[cfg(feature = "remote")]
BlobFileInner::Remote(file) => file.size(),
}
}
/// Physical byte offset in the data file. `None` on remote handles. The
/// Cloud byte-range route does not expose storage layout.
pub fn position(&self) -> Option<u64> {
match &self.inner {
BlobFileInner::Native(file) => Some(file.position()),
#[cfg(feature = "remote")]
BlobFileInner::Remote(_) => None,
}
}
/// Path of the data file holding the blob. `None` on remote handles. The
/// Cloud byte-range route does not expose storage layout.
pub fn data_path(&self) -> Option<&Path> {
match &self.inner {
BlobFileInner::Native(file) => Some(file.data_path()),
#[cfg(feature = "remote")]
BlobFileInner::Remote(_) => None,
}
}
/// Native storage layout. `None` on remote handles. The Cloud byte-range
/// route does not expose layout.
pub fn kind(&self) -> Option<lance_core::datatypes::BlobKind> {
match &self.inner {
BlobFileInner::Native(file) => Some(file.kind()),
#[cfg(feature = "remote")]
BlobFileInner::Remote(_) => None,
}
}
/// External URI for native handles. Remote handles do not expose storage URIs.
pub fn uri(&self) -> Option<&str> {
match &self.inner {
BlobFileInner::Native(file) => file.uri(),
#[cfg(feature = "remote")]
BlobFileInner::Remote(_) => None,
}
}
}
pub use lance::dataset::BlobFile;
/// One row-specific blob range read request.
///
@@ -457,10 +264,7 @@ pub(crate) async fn take_blob_files_aligned(
let handles = dataset.take_blobs(row_ids, column).await?;
ensure_all_row_ids_resolved(column, row_ids.len(), handles.len())?;
Ok(handles
.into_iter()
.map(|handle| handle.map(Into::into))
.collect())
Ok(handles)
}
#[cfg(test)]
+45 -27
View File
@@ -1,7 +1,7 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
pub mod blobs;
mod blobs;
pub mod insert;
use self::insert::{RemoteWriteExec, WriteOp};
@@ -2939,7 +2939,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
}
#[derive(Serialize, Clone, Debug)]
pub(crate) struct MergeInsertRequest {
pub struct MergeInsertRequest {
on: String,
when_matched_update_all: bool,
when_matched_update_all_filt: Option<String>,
@@ -3089,12 +3089,10 @@ mod tests {
Box::pin(table.delete("false").map_ok(|_| ())),
Box::pin(
table
.add_columns()
.transform(NewColumnTransform::SqlExpressions(vec![(
"x".into(),
"y".into(),
)]))
.execute()
.add_columns(
NewColumnTransform::SqlExpressions(vec![("x".into(), "y".into())]),
None,
)
.map_ok(|_| ()),
),
Box::pin(async {
@@ -4302,9 +4300,32 @@ mod tests {
"fetch_blobs",
);
let message = table
.fetch_blob_files("image", &[1])
.await
.unwrap_err()
.to_string();
assert!(
message.contains("fetch_blob_files is not supported on LanceDB Cloud"),
"got: {message}"
);
assert!(
!message.contains("Use fetch_blobs"),
"old server must not be told to use fetch_blobs, got: {message}"
);
}
#[tokio::test]
async fn test_blob_files_point_at_fetch_blobs_on_a_blob_capable_server() {
let table = Table::new_with_handler_version(
"my_table",
semver::Version::new(0, 5, 0),
|_| -> http::Response<String> { panic!("fetch_blob_files must not reach the server") },
);
assert_not_supported_error(
table.fetch_blob_files("image", &[1]).await.unwrap_err(),
"requires LanceDB Cloud server 0.5.0 or newer",
"Use fetch_blobs for full bytes",
);
}
@@ -6390,12 +6411,13 @@ mod tests {
});
let result = table
.add_columns()
.transform(NewColumnTransform::SqlExpressions(vec![
("b".into(), "a + 1".into()),
("x".into(), "cast(NULL as int32)".into()),
]))
.execute()
.add_columns(
NewColumnTransform::SqlExpressions(vec![
("b".into(), "a + 1".into()),
("x".into(), "cast(NULL as int32)".into()),
]),
None,
)
.await
.unwrap();
@@ -7120,12 +7142,10 @@ mod tests {
}
"add_columns" => {
let _ = table
.add_columns()
.transform(NewColumnTransform::SqlExpressions(vec![(
"c".into(),
"a + 1".into(),
)]))
.execute()
.add_columns(
NewColumnTransform::SqlExpressions(vec![("c".into(), "a + 1".into())]),
None,
)
.await;
}
"drop_columns" => {
@@ -9883,12 +9903,10 @@ mod tests {
.await
.unwrap();
branch
.add_columns()
.transform(NewColumnTransform::SqlExpressions(vec![(
"b".into(),
"a + 1".into(),
)]))
.execute()
.add_columns(
NewColumnTransform::SqlExpressions(vec![("b".into(), "a + 1".into())]),
None,
)
.await
.unwrap();
branch
File diff suppressed because it is too large Load Diff
+2 -2
View File
@@ -33,7 +33,7 @@ use crate::table::{AddResult, MergeResult};
/// same Arrow-IPC streaming body and error side-channel; only the target
/// endpoint, query parameters, and parsed result type differ.
#[derive(Debug, Clone)]
pub(crate) enum WriteOp {
pub enum WriteOp {
/// `add`: stream to `/v1/table/{id}/insert/`, optionally overwriting.
Insert { overwrite: bool },
/// `merge_insert`: stream to `/v1/table/{id}/merge_insert/` with the merge
@@ -49,7 +49,7 @@ pub(crate) enum WriteOp {
/// The parsed server response for a completed write, discriminated by the
/// operation that produced it.
#[derive(Debug, Clone)]
pub(crate) enum WriteResult {
pub enum WriteResult {
Add(AddResult),
Merge(MergeResult),
}
+7 -5
View File
@@ -3,7 +3,6 @@
//! LanceDB Table APIs
use crate::blob::BlobFile;
use arrow_array::{LargeBinaryArray, RecordBatch, RecordBatchReader};
use arrow_schema::{Schema, SchemaRef};
use async_trait::async_trait;
@@ -13,6 +12,7 @@ use datafusion_physical_plan::ExecutionPlan;
use datafusion_physical_plan::display::DisplayableExecutionPlan;
use futures::StreamExt;
use futures::stream::FuturesUnordered;
use lance::dataset::BlobFile;
pub use lance::dataset::ColumnAlteration;
pub use lance::dataset::NewColumnTransform;
pub use lance::dataset::ReadParams;
@@ -65,7 +65,6 @@ use crate::utils::{PatchReadParam, PatchWriteParam, resolve_arrow_field_path};
use self::dataset::DatasetConsistencyWrapper;
use self::merge::MergeInsertBuilder;
pub mod add_columns;
mod add_data;
pub mod branch_merge;
mod create_index;
@@ -80,7 +79,6 @@ pub mod schema_evolution;
pub mod update;
pub mod write_progress;
use crate::index::waiter::wait_for_index;
pub use add_columns::AddColumnsBuilder;
#[cfg(feature = "remote")]
pub(crate) use add_data::PreprocessingOutput;
pub use add_data::{AddDataBuilder, AddDataMode, AddResult, NaNVectorBehavior};
@@ -1622,8 +1620,12 @@ impl Table {
}
/// Add new columns to the table, providing values to fill in.
pub fn add_columns(&self) -> AddColumnsBuilder {
AddColumnsBuilder::new(self.inner.clone())
pub async fn add_columns(
&self,
transforms: NewColumnTransform,
read_columns: Option<Vec<String>>,
) -> Result<AddColumnsResult> {
self.inner.add_columns(transforms, read_columns).await
}
/// Change a column's name or nullability.
-161
View File
@@ -1,161 +0,0 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
//! Builder for adding columns to a table.
use std::sync::Arc;
use lance::dataset::NewColumnTransform;
use super::BaseTable;
use super::schema_evolution::AddColumnsResult;
use crate::{Error, Result};
/// Adds columns to a table. See [`Table::add_columns`](super::Table::add_columns).
pub struct AddColumnsBuilder {
parent: Arc<dyn BaseTable>,
transform: Option<NewColumnTransform>,
read_columns: Option<Vec<String>>,
}
impl std::fmt::Debug for AddColumnsBuilder {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AddColumnsBuilder")
.field("parent", &self.parent)
.field("has_transform", &self.transform.is_some())
.field("read_columns", &self.read_columns)
.finish()
}
}
impl AddColumnsBuilder {
pub(crate) fn new(parent: Arc<dyn BaseTable>) -> Self {
Self {
parent,
transform: None,
read_columns: None,
}
}
/// Set how the new columns' values are produced. Required.
pub fn transform(mut self, transform: NewColumnTransform) -> Self {
self.transform = Some(transform);
self
}
/// Limit which existing columns a [`NewColumnTransform::BatchUDF`] mapper
/// receives. Every other transform determines what it reads, so setting
/// this alongside one is an error rather than a silent no-op.
pub fn read_columns(mut self, columns: impl IntoIterator<Item = impl Into<String>>) -> Self {
self.read_columns = Some(columns.into_iter().map(Into::into).collect());
self
}
/// Add the columns.
pub async fn execute(self) -> Result<AddColumnsResult> {
let Self {
parent,
transform,
read_columns,
} = self;
let Some(transform) = transform else {
return Err(Error::InvalidInput {
message: "add_columns requires a transform".into(),
});
};
if read_columns.is_some() && !matches!(transform, NewColumnTransform::BatchUDF(_)) {
return Err(Error::InvalidInput {
message: "read_columns applies only to a BatchUDF transform; \
every other transform determines what it reads"
.into(),
});
}
parent.add_columns(transform, read_columns).await
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use arrow_array::{Int32Array, RecordBatch, record_batch};
use arrow_schema::{DataType, Field, Schema};
use lance::dataset::{BatchUDF, NewColumnTransform};
use crate::Table;
use crate::connect;
async fn table_with_two_columns(name: &str) -> Table {
let conn = connect("memory://").execute().await.unwrap();
let batch = record_batch!(("x", Int32, [1, 2, 3]), ("y", Int32, [10, 20, 30])).unwrap();
conn.create_table(name, batch).execute().await.unwrap()
}
#[tokio::test]
async fn test_requires_a_transform() {
let table = table_with_two_columns("no_transform").await;
let err = table.add_columns().execute().await.unwrap_err();
assert!(
err.to_string().contains("requires a transform"),
"got: {err}"
);
}
#[tokio::test]
async fn test_read_columns_with_sql_expressions_is_rejected() {
let table = table_with_two_columns("read_cols_sql").await;
let err = table
.add_columns()
.transform(NewColumnTransform::SqlExpressions(vec![(
"doubled".into(),
"x * 2".into(),
)]))
.read_columns(["x"])
.execute()
.await
.unwrap_err();
assert!(err.to_string().contains("BatchUDF"), "got: {err}");
let schema = table.schema().await.unwrap();
assert!(
schema.field_with_name("doubled").is_err(),
"a rejected call must not commit"
);
}
#[tokio::test]
async fn test_read_columns_limits_what_a_batch_udf_sees() {
let table = table_with_two_columns("read_cols_udf").await;
let output_schema = Arc::new(Schema::new(vec![Field::new("sum", DataType::Int32, true)]));
let mapper_schema = output_schema.clone();
let udf = BatchUDF {
mapper: Box::new(move |batch: &RecordBatch| {
assert!(batch.column_by_name("x").is_some());
assert!(batch.column_by_name("y").is_none(), "y was not requested");
let x = batch["x"].as_any().downcast_ref::<Int32Array>().unwrap();
let doubled: Int32Array = x.iter().map(|v| v.map(|v| v * 2)).collect();
Ok(RecordBatch::try_new(
mapper_schema.clone(),
vec![Arc::new(doubled)],
)?)
}),
output_schema,
result_checkpoint: None,
};
table
.add_columns()
.transform(NewColumnTransform::BatchUDF(udf))
.read_columns(["x"])
.execute()
.await
.unwrap();
let schema = table.schema().await.unwrap();
assert!(schema.field_with_name("sum").is_ok());
}
}
+5 -9
View File
@@ -576,12 +576,10 @@ mod tests {
// Add a new physical column AFTER the embedding column.
table
.add_columns()
.transform(NewColumnTransform::SqlExpressions(vec![(
"score".into(),
"42.0".into(),
)]))
.execute()
.add_columns(
NewColumnTransform::SqlExpressions(vec![("score".into(), "42.0".into())]),
None,
)
.await
.unwrap();
@@ -685,9 +683,7 @@ mod tests {
true,
)]));
table
.add_columns()
.transform(NewColumnTransform::AllNulls(nested_schema))
.execute()
.add_columns(NewColumnTransform::AllNulls(nested_schema), None)
.await
.unwrap();
+19 -24
View File
@@ -193,12 +193,10 @@ mod tests {
// Add a computed column
let result = table
.add_columns()
.transform(NewColumnTransform::SqlExpressions(vec![(
"doubled".into(),
"id * 2".into(),
)]))
.execute()
.add_columns(
NewColumnTransform::SqlExpressions(vec![("doubled".into(), "id * 2".into())]),
None,
)
.await
.unwrap();
@@ -253,12 +251,13 @@ mod tests {
// Add multiple columns at once
table
.add_columns()
.transform(NewColumnTransform::SqlExpressions(vec![
("y".into(), "x + 1".into()),
("z".into(), "x * x".into()),
]))
.execute()
.add_columns(
NewColumnTransform::SqlExpressions(vec![
("y".into(), "x + 1".into()),
("z".into(), "x * x".into()),
]),
None,
)
.await
.unwrap();
@@ -284,12 +283,10 @@ mod tests {
// Add a column with a constant value
table
.add_columns()
.transform(NewColumnTransform::SqlExpressions(vec![(
"constant".into(),
"42".into(),
)]))
.execute()
.add_columns(
NewColumnTransform::SqlExpressions(vec![("constant".into(), "42".into())]),
None,
)
.await
.unwrap();
@@ -662,12 +659,10 @@ mod tests {
// Add column increments version
let add_result = table
.add_columns()
.transform(NewColumnTransform::SqlExpressions(vec![(
"c".into(),
"a + b".into(),
)]))
.execute()
.add_columns(
NewColumnTransform::SqlExpressions(vec![("c".into(), "a + b".into())]),
None,
)
.await
.unwrap();
assert!(add_result.version > v1);
+2 -254
View File
@@ -9,17 +9,14 @@ use arrow_array::{
};
use arrow_schema::{DataType, Field, Fields, Schema};
use futures::TryStreamExt;
use lance::Dataset;
use lance_encoding::version::LanceFileVersion;
use lancedb::{
Connection, Error, Result, Table,
blob::{BlobRangeRequest, blob},
connect, connect_namespace,
database::listing::{
ListingDatabaseOptions, NewTableConfig, OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS,
},
database::listing::OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS,
query::{ExecutableQuery, QueryBase},
table::{AddDataMode, CompactionOptions, OptimizeAction, OptimizeStats},
table::{AddDataMode, CompactionOptions, OptimizeAction},
};
use tempfile::tempdir;
@@ -1078,252 +1075,3 @@ async fn fetch_blob_files_aligns_across_fragments_with_nulls_and_dups() -> Resul
}
Ok(())
}
/// Rows exercising the null/empty interleavings from
/// <https://github.com/lancedb/lancedb/issues/3744>: a payload, a null, a valid
/// empty value, then payloads whose descriptors a fragment rewrite used to zero.
fn null_empty_input_batch() -> RecordBatch {
let owned = [
Some(dedicated_blob_bytes(1)),
None,
Some(Vec::new()),
Some(dedicated_blob_bytes(4)),
Some(dedicated_blob_bytes(5)),
Some(dedicated_blob_bytes(6)),
];
let payloads: Vec<Option<&[u8]>> = owned.iter().map(|payload| payload.as_deref()).collect();
binary_input_batch(&[1, 2, 3, 4, 5, 6], &payloads)
}
/// One `(id, Some((payload length, first byte)))` per live row, or `(id, None)`
/// for a null blob. Comparing lengths and first bytes keeps failure output
/// readable where comparing whole payloads would not.
type BlobSummary = Vec<(i64, Option<(usize, Option<u8>)>)>;
/// The rows [`null_empty_input_batch`] leaves behind after `id IN (1, 4)` is
/// deleted: a null, a valid empty value, and the two payloads that follow them.
fn expected_null_empty_survivors() -> BlobSummary {
vec![
(2, None),
(3, Some((0, None))),
(5, Some((DEDICATED_BLOB_LEN, Some(5)))),
(6, Some((DEDICATED_BLOB_LEN, Some(6)))),
]
}
/// `optimize()` only rewrites a fragment when lance's compaction planner selects
/// it — here because the delete pushes the fragment past
/// `materialize_deletions_threshold` (0.1 by default; these tests delete 2 of 6
/// rows). Without this check, a planner or threshold change upstream would leave
/// both regression tests green while no rewrite happened at all.
fn assert_compacted(stats: &OptimizeStats) {
let metrics = stats
.compaction
.as_ref()
.expect("OptimizeAction::All runs compaction");
assert!(
metrics.fragments_removed >= 1,
"optimize() rewrote no fragment, so this test proves nothing: {metrics:?}"
);
}
fn summarize(rows: &[(i64, Option<Vec<u8>>)]) -> BlobSummary {
rows.iter()
.map(|(id, payload)| {
(
*id,
payload
.as_ref()
.map(|bytes| (bytes.len(), bytes.first().copied())),
)
})
.collect()
}
async fn sorted_id_rowid(table: &Table) -> Result<Vec<(i64, u64)>> {
let mut pairs = collect_id_rowid(table).await?;
pairs.sort_by_key(|(id, _)| *id);
Ok(pairs)
}
/// `{position, size}` descriptors of a legacy v1 blob column, keyed by `id`.
async fn v1_blob_descriptors(table: &Table) -> Result<Vec<(i64, Option<(u64, u64)>)>> {
let batches = table
.query()
.execute()
.await?
.try_collect::<Vec<_>>()
.await?;
let batch = arrow_select::concat::concat_batches(&batches[0].schema(), &batches).unwrap();
let ids = batch
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
let descriptors = batch
.column_by_name("image")
.unwrap()
.as_any()
.downcast_ref::<StructArray>()
.expect("v1 blob column reads back as a descriptor struct");
let position = descriptors
.column_by_name("position")
.unwrap()
.as_any()
.downcast_ref::<UInt64Array>()
.unwrap();
let size = descriptors
.column_by_name("size")
.unwrap()
.as_any()
.downcast_ref::<UInt64Array>()
.unwrap();
let mut rows: Vec<(i64, Option<(u64, u64)>)> = (0..batch.num_rows())
.map(|row| {
let descriptor =
(!descriptors.is_null(row)).then(|| (position.value(row), size.value(row)));
(ids.value(row), descriptor)
})
.collect();
rows.sort_by_key(|(id, _)| *id);
Ok(rows)
}
/// Payload bytes of every live row of a legacy v1 blob column, keyed by `id`.
/// [`Table::fetch_blobs`] rejects v1 columns, so read them through lance.
async fn v1_blob_payloads(dataset_uri: &str, table: &Table) -> Result<Vec<(i64, Option<Vec<u8>>)>> {
let pairs = sorted_id_rowid(table).await?;
let row_ids: Vec<u64> = pairs.iter().map(|(_, row_id)| *row_id).collect();
let dataset = Arc::new(Dataset::open(dataset_uri).await?);
let files = dataset.take_blobs(&row_ids, "image").await?;
assert_eq!(
files.len(),
pairs.len(),
"take_blobs returned {} handles for {} live rows",
files.len(),
pairs.len()
);
let mut rows = Vec::with_capacity(pairs.len());
for ((id, _), file) in pairs.iter().zip(files) {
let payload = match file {
Some(file) => Some(file.read().await?.to_vec()),
None => None,
};
rows.push((*id, payload));
}
Ok(rows)
}
/// Length and first byte of every live blob v2 value, keyed by `id`.
async fn blob_v2_values(table: &Table) -> Result<BlobSummary> {
let pairs = sorted_id_rowid(table).await?;
let row_ids: Vec<u64> = pairs.iter().map(|(_, row_id)| *row_id).collect();
let bytes = table.fetch_blobs("image", &row_ids).await?;
Ok(pairs
.iter()
.enumerate()
.map(|(slot, (id, _))| {
let value = (!bytes.is_null(slot))
.then(|| (bytes.value(slot).len(), bytes.value(slot).first().copied()));
(*id, value)
})
.collect())
}
/// Regression test for [#3744]: on storage 2.0 (legacy v1 descriptors),
/// compaction rewrote every payload following a null or empty value in the same
/// fragment as `{position: 0, size: 0}`, so the payload bytes read back as `b""`
/// and the new fragment no longer referenced them at all.
///
/// [#3744]: https://github.com/lancedb/lancedb/issues/3744
#[tokio::test]
async fn optimize_preserves_v1_blob_payloads_with_null_and_empty() -> Result<()> {
let tmp = tempdir().unwrap();
let db_uri = tmp.path().to_str().unwrap().to_string();
let db = connect(&db_uri)
.database_options(&ListingDatabaseOptions {
new_table_config: NewTableConfig {
data_storage_version: Some(LanceFileVersion::V2_0),
..Default::default()
},
..Default::default()
})
.execute()
.await?;
let legacy = Field::new("image", DataType::LargeBinary, true).with_metadata(
std::collections::HashMap::from([("lance-encoding:blob".to_string(), "true".to_string())]),
);
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
legacy,
]));
let table = db.create_empty_table("t", schema).execute().await?;
table.add(null_empty_input_batch()).execute().await?;
assert_eq!(
storage_format_version(&table).await,
LanceFileVersion::V2_0.resolve(),
"v1 blob descriptors only exist below storage 2.2"
);
let dataset_uri = table.uri().await?;
// Any rewrite triggers it; deleting rows is the shape from the issue.
table.delete("id IN (1, 4)").await?;
let descriptors_before = v1_blob_descriptors(&table).await?;
let before = v1_blob_payloads(&dataset_uri, &table).await?;
assert_eq!(
summarize(&before),
expected_null_empty_survivors(),
"test setup no longer produces the null/empty/payload mix"
);
let stats = table.optimize(OptimizeAction::All).await?;
assert_compacted(&stats);
let descriptors_after = v1_blob_descriptors(&table).await?;
let after = v1_blob_payloads(&dataset_uri, &table).await?;
assert_eq!(
summarize(&after),
summarize(&before),
"optimize() lost blob payloads; descriptors before={descriptors_before:?} after={descriptors_after:?}"
);
assert!(after == before, "optimize() changed blob payload bytes");
Ok(())
}
/// Regression test for the blob v2 half of [#3744]: compaction rewrote a valid
/// empty value as null, destroying the null-vs-empty distinction.
///
/// [#3744]: https://github.com/lancedb/lancedb/issues/3744
#[tokio::test]
async fn optimize_preserves_blob_v2_null_and_empty_distinction() -> Result<()> {
let tmp = tempdir().unwrap();
let db = connect(tmp.path().to_str().unwrap()).execute().await?;
let table = db
.create_empty_table("t", blob_table_schema())
.execute()
.await?;
table.add(null_empty_input_batch()).execute().await?;
assert!(
storage_format_version(&table).await >= LanceFileVersion::V2_2,
"blob v2 columns require storage >= 2.2"
);
table.delete("id IN (1, 4)").await?;
let before = blob_v2_values(&table).await?;
assert_eq!(
before,
expected_null_empty_survivors(),
"test setup no longer produces the null/empty/payload mix"
);
let stats = table.optimize(OptimizeAction::All).await?;
assert_compacted(&stats);
assert_eq!(
blob_v2_values(&table).await?,
before,
"optimize() changed blob v2 values"
);
Ok(())
}