mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-27 14:06:14 +00:00
docs: document Blob V2 in the multimodal guide (#4623)
This commit is contained in:
@@ -0,0 +1,55 @@
|
||||
# Blob v2 documentation tests
|
||||
|
||||
`docs/web-tests/ts` pins `@lancedb/lancedb` 0.31.0 and `docs/web-tests/rs` pins `lancedb` 0.31.0. Those releases do not include `blob()`. Run the Blob v2 examples here.
|
||||
|
||||
## Python
|
||||
|
||||
`lancedb==0.41.0b10` from the LanceDB package index, without pylance. PyPI `lancedb` 0.40.0 does not provide `lancedb.blob()`.
|
||||
|
||||
```shell
|
||||
python3.13 -m venv .venv
|
||||
.venv/bin/pip install -r docs/web-tests/blob-v2/py/requirements.txt
|
||||
.venv/bin/python -m pytest docs/web-tests/blob-v2/py/test_multimodal.py
|
||||
```
|
||||
|
||||
`test_blob_pylance.py` checks custom thresholds and `to_pandas()` in `lazy` and `bytes` mode. Those calls import pylance. Use a second virtualenv. `test_multimodal.py` fails if `lance` can be imported. Snippets from this file are written to `docs/web/snippets/blob_pylance.mdx`.
|
||||
|
||||
This install is for the optional tests only. The public guide does not tell users to bypass the dependency resolver. PyPI `pylance` 13.0.0 requires `lance-namespace<0.12`, and `lancedb==0.41.0b10` requires `lance-namespace>=0.13`, so `lancedb[pylance]` does not resolve. The three tests passed with pylance installed after the base requirements, without its own dependency pin.
|
||||
|
||||
```shell
|
||||
python3.13 -m venv .venv-pylance
|
||||
.venv-pylance/bin/pip install -r docs/web-tests/blob-v2/py/requirements.txt
|
||||
.venv-pylance/bin/pip install --no-deps pylance==13.0.0
|
||||
.venv-pylance/bin/python -m pytest docs/web-tests/blob-v2/py/test_blob_pylance.py
|
||||
```
|
||||
|
||||
## TypeScript
|
||||
|
||||
`@lancedb/lancedb@0.41.0-beta.8`. Node.js 22.6 or newer. `npm test` passes `--experimental-strip-types`, which Node added in 22.6.0.
|
||||
|
||||
```shell
|
||||
cd docs/web-tests/blob-v2/ts
|
||||
npm ci
|
||||
npm test
|
||||
```
|
||||
|
||||
## Rust
|
||||
|
||||
crates.io `lancedb` 0.40.0 does not include `blob()`. This example builds the workspace crate, `lancedb` 0.41.0-beta.10. Pass `--features remote` so this checkout compiles. The example uses a local database. The Rust workflow runs this example. It is not a default feature, so a workspace build without `remote` does not turn that feature on for the rest of the workspace.
|
||||
|
||||
```shell
|
||||
cargo run -p lancedb-blob-v2-docs --features remote --example multimodal
|
||||
```
|
||||
|
||||
## CI
|
||||
|
||||
Python and TypeScript above install preview packages. Pre-release builds are not guaranteed to stay available for more than 6 months, so those runs are not required PR checks. Run them when the guide's pins change. The Rust example is the in-repo check.
|
||||
|
||||
Regenerate snippets from the repository root:
|
||||
|
||||
```shell
|
||||
python3 docs/web-tests/mdx_snippets_gen.py \
|
||||
-s docs/web-tests/py -s docs/web-tests/ts -s docs/web-tests/rs \
|
||||
-s docs/web-tests/blob-v2/py -s docs/web-tests/blob-v2/ts -s docs/web-tests/blob-v2/rs \
|
||||
-o docs/web/snippets
|
||||
```
|
||||
@@ -0,0 +1,3 @@
|
||||
[pytest]
|
||||
asyncio_mode = auto
|
||||
testpaths = test_multimodal.py
|
||||
@@ -0,0 +1,8 @@
|
||||
--extra-index-url https://pypi.fury.io/lancedb/
|
||||
|
||||
lancedb==0.41.0b10
|
||||
numpy==2.2.6
|
||||
pandas==2.3.3
|
||||
pillow==12.1.1
|
||||
pytest==8.4.2
|
||||
pytest-asyncio==1.2.0
|
||||
@@ -0,0 +1,154 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
"""Blob reads that import pylance.
|
||||
|
||||
Install pylance==13.0.0 with --no-deps after lancedb==0.41.0b10.
|
||||
See docs/web-tests/blob-v2/README.md.
|
||||
"""
|
||||
|
||||
import lance
|
||||
import lancedb
|
||||
import numpy as np
|
||||
import pyarrow as pa
|
||||
import pytest
|
||||
|
||||
|
||||
def test_blob_api_to_pandas(tmp_path):
|
||||
db = lancedb.connect(tmp_path)
|
||||
schema = pa.schema(
|
||||
[
|
||||
pa.field("id", pa.int64()),
|
||||
lancedb.blob("video"),
|
||||
]
|
||||
)
|
||||
tbl = db.create_table(
|
||||
"videos",
|
||||
data=[
|
||||
{"id": 1, "video": b"fake_video_bytes_1"},
|
||||
{"id": 2, "video": b"fake_video_bytes_2"},
|
||||
],
|
||||
schema=schema,
|
||||
mode="overwrite",
|
||||
)
|
||||
|
||||
# --8<-- [start:blob_api_to_pandas]
|
||||
df_lazy = tbl.to_pandas()
|
||||
|
||||
df_bytes = tbl.to_pandas(blob_mode="bytes")
|
||||
|
||||
df_desc = tbl.to_pandas(blob_mode="descriptions")
|
||||
|
||||
arrow_bytes = tbl.to_arrow(blob_mode="bytes")
|
||||
|
||||
df_typed = tbl.to_pandas(split_blocks=True, self_destruct=True)
|
||||
# --8<-- [end:blob_api_to_pandas]
|
||||
|
||||
assert len(df_lazy) == 2
|
||||
assert all(isinstance(value, lance.BlobFile) for value in df_lazy["video"])
|
||||
assert isinstance(df_bytes["video"].iloc[0], bytes)
|
||||
assert df_bytes["video"].tolist() == [
|
||||
b"fake_video_bytes_1",
|
||||
b"fake_video_bytes_2",
|
||||
]
|
||||
assert arrow_bytes["video"].to_pylist() == [
|
||||
b"fake_video_bytes_1",
|
||||
b"fake_video_bytes_2",
|
||||
]
|
||||
assert "_rowid" not in df_bytes.columns
|
||||
assert len(df_desc) == 2
|
||||
assert set(df_desc["video"].iloc[0]) == {
|
||||
"kind",
|
||||
"position",
|
||||
"size",
|
||||
"blob_id",
|
||||
"blob_uri",
|
||||
}
|
||||
assert len(df_typed) == 2
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_query_to_pandas_kwargs(tmp_path):
|
||||
schema = pa.schema(
|
||||
[
|
||||
pa.field("id", pa.int64()),
|
||||
pa.field("vector", pa.list_(pa.float32(), 128)),
|
||||
lancedb.blob("video"),
|
||||
]
|
||||
)
|
||||
data = [
|
||||
{
|
||||
"id": i,
|
||||
"vector": np.random.rand(128).astype(np.float32),
|
||||
"video": f"fake_video_bytes_{i}".encode(),
|
||||
}
|
||||
for i in range(10)
|
||||
]
|
||||
|
||||
db = lancedb.connect(tmp_path / "sync")
|
||||
tbl = db.create_table("search_demo", data=data, schema=schema, mode="overwrite")
|
||||
|
||||
async_db = await lancedb.connect_async(str(tmp_path / "async"))
|
||||
tbl_async = await async_db.create_table(
|
||||
"search_demo", data=data, schema=schema, mode="overwrite"
|
||||
)
|
||||
|
||||
query_vector = np.random.rand(128).astype(np.float32)
|
||||
|
||||
# --8<-- [start:query_to_pandas_kwargs]
|
||||
df_lazy = (
|
||||
tbl.search().where("id = 1").select(["id", "video"]).to_pandas(blob_mode="lazy")
|
||||
)
|
||||
|
||||
df_bytes = await (
|
||||
tbl_async.query()
|
||||
.where("id = 1")
|
||||
.select(["id", "video"])
|
||||
.to_pandas(blob_mode="bytes")
|
||||
)
|
||||
|
||||
df_vec = (
|
||||
tbl.search(query_vector)
|
||||
.limit(2)
|
||||
.select(["id", "video"])
|
||||
.to_pandas(blob_mode="bytes", split_blocks=True, self_destruct=True)
|
||||
)
|
||||
# --8<-- [end:query_to_pandas_kwargs]
|
||||
|
||||
assert len(df_lazy) == 1
|
||||
assert "_rowid" not in df_bytes.columns
|
||||
assert len(df_bytes) == 1
|
||||
assert df_bytes["video"].iloc[0] == b"fake_video_bytes_1"
|
||||
assert len(df_vec) == 2
|
||||
assert isinstance(df_vec["video"].iloc[0], bytes)
|
||||
|
||||
|
||||
def test_append_keeps_table_blob_thresholds(tmp_path):
|
||||
db = lancedb.connect(tmp_path)
|
||||
schema = pa.schema(
|
||||
[
|
||||
pa.field("id", pa.int64()),
|
||||
lancedb.blob("video", inline_size_threshold=64 * 1024),
|
||||
]
|
||||
)
|
||||
tbl = db.create_table("videos", schema=schema)
|
||||
other = pa.schema(
|
||||
[
|
||||
pa.field("id", pa.int64()),
|
||||
lancedb.blob("video", inline_size_threshold=8 * 1024),
|
||||
]
|
||||
)
|
||||
payload = b"x" * (20 * 1024)
|
||||
tbl.add(
|
||||
pa.table(
|
||||
[pa.array([1], type=pa.int64()), lance.blob_array([payload])],
|
||||
schema=other,
|
||||
)
|
||||
)
|
||||
|
||||
stored = tbl.schema.field("video").metadata[
|
||||
b"lance-encoding:blob-inline-size-threshold"
|
||||
]
|
||||
assert stored == b"65536"
|
||||
descriptor = tbl.search().select(["video"]).to_arrow(blob_mode="descriptions")
|
||||
assert descriptor.column("video")[0].as_py()["kind"] == 0
|
||||
@@ -0,0 +1,204 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
import importlib.util
|
||||
|
||||
import lancedb
|
||||
import pyarrow as pa
|
||||
|
||||
if importlib.util.find_spec("lance") is not None:
|
||||
raise RuntimeError(
|
||||
"Blob v2 documentation tests must run without pylance. "
|
||||
"Install docs/web-tests/blob-v2/py/requirements.txt into a fresh virtualenv."
|
||||
)
|
||||
|
||||
INLINE_MAX = 64 * 1024
|
||||
DEDICATED_MIN = 4 * 1024 * 1024
|
||||
|
||||
|
||||
def test_blob_api_definition(tmp_path, monkeypatch):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
# --8<-- [start:blob_api_setup]
|
||||
import lancedb
|
||||
|
||||
db = lancedb.connect("./blob_db")
|
||||
# --8<-- [end:blob_api_setup]
|
||||
assert db.list_tables().tables == []
|
||||
|
||||
# --8<-- [start:blob_api_schema]
|
||||
import lancedb
|
||||
import pyarrow as pa
|
||||
|
||||
schema = pa.schema(
|
||||
[
|
||||
pa.field("id", pa.int64()),
|
||||
pa.field("label", pa.string()),
|
||||
lancedb.blob("image"),
|
||||
]
|
||||
)
|
||||
# --8<-- [end:blob_api_schema]
|
||||
image = schema.field("image").type
|
||||
assert image.extension_name == "lance.blob.v2"
|
||||
assert type(image).__module__ == "lancedb.schema"
|
||||
|
||||
# --8<-- [start:blob_api_ingest]
|
||||
import io
|
||||
from PIL import Image
|
||||
|
||||
def jpeg_with_make(color, make):
|
||||
image = Image.new("RGB", (8, 8), color=color)
|
||||
exif = Image.Exif()
|
||||
exif[0x010F] = make
|
||||
buf = io.BytesIO()
|
||||
image.save(buf, format="JPEG", exif=exif)
|
||||
return buf.getvalue()
|
||||
|
||||
tbl = db.create_table(
|
||||
"photos",
|
||||
data=[
|
||||
{"id": 1, "label": "red", "image": jpeg_with_make("red", "TestCam")},
|
||||
{"id": 2, "label": "blue", "image": jpeg_with_make("blue", "OtherCam")},
|
||||
],
|
||||
schema=schema,
|
||||
mode="overwrite",
|
||||
)
|
||||
# --8<-- [end:blob_api_ingest]
|
||||
assert len(tbl) == 2
|
||||
|
||||
# --8<-- [start:blob_api_bytes]
|
||||
rows = (
|
||||
tbl.search().where("id = 1").select(["id", "image"]).to_list(blob_mode="bytes")
|
||||
)
|
||||
# --8<-- [end:blob_api_bytes]
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["id"] == 1
|
||||
assert "_rowid" not in rows[0]
|
||||
assert rows[0]["image"].startswith(b"\xff\xd8")
|
||||
|
||||
# --8<-- [start:blob_api_files]
|
||||
from PIL import ExifTags
|
||||
|
||||
rows = tbl.search().where("id = 1").select(["label"]).with_row_id(True).to_list()
|
||||
with tbl.fetch_blob_files("image", [rows[0]["_rowid"]])[0] as image_file:
|
||||
make = Image.open(image_file).getexif().get(ExifTags.Base.Make)
|
||||
print(rows[0]["label"], make)
|
||||
# --8<-- [end:blob_api_files]
|
||||
assert rows[0]["label"] == "red"
|
||||
assert make == "TestCam"
|
||||
|
||||
# --8<-- [start:blob_api_fetch]
|
||||
rows = tbl.search().select(["id"]).with_row_id(True).to_list()
|
||||
row_ids = [row["_rowid"] for row in rows]
|
||||
payloads = tbl.fetch_blobs("image", row_ids)
|
||||
# --8<-- [end:blob_api_fetch]
|
||||
assert all(payload.startswith(b"\xff\xd8") for payload in payloads.to_pylist())
|
||||
|
||||
# --8<-- [start:blob_api_ranges]
|
||||
slices = tbl.fetch_blob_ranges("image", [(row_ids[0], 0, 2)])
|
||||
|
||||
with tbl.fetch_blob_files("image", [row_ids[0]])[0] as image_file:
|
||||
parts = image_file.read_ranges([(0, 2)])
|
||||
# --8<-- [end:blob_api_ranges]
|
||||
assert slices.to_pylist() == [b"\xff\xd8"]
|
||||
assert parts == [b"\xff\xd8"]
|
||||
|
||||
|
||||
def test_eager_query_returns_blob_bytes_without_row_id(tmp_path):
|
||||
db = lancedb.connect(tmp_path)
|
||||
schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
|
||||
payload = b"inline-bytes"
|
||||
tbl = db.create_table(
|
||||
"photos",
|
||||
data=[{"id": 1, "image": payload}],
|
||||
schema=schema,
|
||||
)
|
||||
|
||||
rows = (
|
||||
tbl.search().where("id = 1").select(["id", "image"]).to_list(blob_mode="bytes")
|
||||
)
|
||||
arrow_bytes = (
|
||||
tbl.search().where("id = 1").select(["id", "image"]).to_arrow(blob_mode="bytes")
|
||||
)
|
||||
described = tbl.to_pandas(blob_mode="descriptions")
|
||||
|
||||
assert rows == [{"id": 1, "image": payload}]
|
||||
assert "_rowid" not in rows[0]
|
||||
assert arrow_bytes.column("image").to_pylist() == [payload]
|
||||
assert "_rowid" not in arrow_bytes.column_names
|
||||
assert described["image"].iloc[0]["kind"] == 0
|
||||
|
||||
|
||||
def test_deferred_fetch_reads_blob_by_row_id(tmp_path):
|
||||
db = lancedb.connect(tmp_path)
|
||||
schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
|
||||
payload = b"deferred-bytes"
|
||||
tbl = db.create_table(
|
||||
"photos",
|
||||
data=[{"id": 1, "image": payload}],
|
||||
schema=schema,
|
||||
)
|
||||
|
||||
described = (
|
||||
tbl.search()
|
||||
.where("id = 1")
|
||||
.select(["id", "image"])
|
||||
.to_list(blob_mode="descriptions")
|
||||
)
|
||||
assert described[0]["image"]["kind"] == 0
|
||||
assert "_rowid" not in described[0]
|
||||
|
||||
found = tbl.search().where("id = 1").select(["id"]).with_row_id(True).to_list()
|
||||
row_id = found[0]["_rowid"]
|
||||
fetched = tbl.fetch_blobs("image", [row_id]).to_pylist()
|
||||
with tbl.fetch_blob_files("image", [row_id])[0] as handle:
|
||||
streamed = handle.read()
|
||||
|
||||
assert fetched == [payload]
|
||||
assert streamed == payload
|
||||
|
||||
|
||||
def _marked(size, marker):
|
||||
payload = bytearray(size)
|
||||
payload[0] = marker
|
||||
payload[-1] = marker
|
||||
return bytes(payload)
|
||||
|
||||
|
||||
def test_storage_tiers_round_trip(tmp_path):
|
||||
cases = [
|
||||
(1, _marked(INLINE_MAX, 11), 0),
|
||||
(2, _marked(INLINE_MAX + 1, 22), 1),
|
||||
(3, _marked(DEDICATED_MIN, 33), 1),
|
||||
(4, _marked(DEDICATED_MIN + 1, 44), 2),
|
||||
]
|
||||
db = lancedb.connect(tmp_path)
|
||||
schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
|
||||
tbl = db.create_table(
|
||||
"tiers",
|
||||
data=[{"id": row_id, "image": payload} for row_id, payload, _kind in cases],
|
||||
schema=schema,
|
||||
)
|
||||
|
||||
described = {
|
||||
row["id"]: row["image"]
|
||||
for row in tbl.search()
|
||||
.select(["id", "image"])
|
||||
.to_list(blob_mode="descriptions")
|
||||
}
|
||||
located = {
|
||||
row["id"]: row["_rowid"]
|
||||
for row in tbl.search().select(["id"]).with_row_id(True).to_list()
|
||||
}
|
||||
fetched = {
|
||||
row_id: tbl.fetch_blobs("image", [located[row_id]]).to_pylist()[0]
|
||||
for row_id, _payload, _kind in cases
|
||||
}
|
||||
|
||||
for row_id, payload, kind in cases:
|
||||
descriptor = described[row_id]
|
||||
assert descriptor["kind"] == kind
|
||||
assert descriptor["size"] == len(payload)
|
||||
assert fetched[row_id] == payload
|
||||
with tbl.fetch_blob_files("image", [located[row_id]])[0] as handle:
|
||||
assert handle.read_range(0, 1) == payload[:1]
|
||||
assert handle.read_range(len(payload) - 1, 1) == payload[-1:]
|
||||
@@ -0,0 +1,21 @@
|
||||
[package]
|
||||
name = "lancedb-blob-v2-docs"
|
||||
version = "0.0.0"
|
||||
edition.workspace = true
|
||||
publish = false
|
||||
license.workspace = true
|
||||
|
||||
[features]
|
||||
remote = ["lancedb/remote"]
|
||||
|
||||
[dependencies]
|
||||
arrow-array = { workspace = true }
|
||||
arrow-schema = { workspace = true }
|
||||
futures-util = "0.3"
|
||||
lancedb = { workspace = true }
|
||||
tempfile = { workspace = true }
|
||||
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
|
||||
|
||||
[[example]]
|
||||
name = "multimodal"
|
||||
path = "multimodal.rs"
|
||||
@@ -0,0 +1,344 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
// --8<-- [start:blob_api_imports]
|
||||
use std::sync::Arc;
|
||||
|
||||
use arrow_array::{
|
||||
Array, Int64Array, LargeBinaryArray, RecordBatch, RecordBatchIterator, StringArray, UInt64Array,
|
||||
};
|
||||
use arrow_schema::{DataType, Field, Schema};
|
||||
use futures_util::TryStreamExt;
|
||||
use lancedb::blob::{BlobRangeRequest, blob};
|
||||
use lancedb::connect;
|
||||
use lancedb::database::CreateTableMode;
|
||||
use lancedb::query::{ExecutableQuery, QueryBase};
|
||||
// --8<-- [end:blob_api_imports]
|
||||
|
||||
use arrow_array::{StructArray, UInt8Array};
|
||||
|
||||
const INLINE_MAX: usize = 64 * 1024;
|
||||
const DEDICATED_MIN: usize = 4 * 1024 * 1024;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
tutorial().await;
|
||||
deferred_fetch().await;
|
||||
storage_tiers().await;
|
||||
}
|
||||
|
||||
async fn tutorial() {
|
||||
let temp_dir = tempfile::tempdir().unwrap();
|
||||
let original_dir = std::env::current_dir().unwrap();
|
||||
std::env::set_current_dir(temp_dir.path()).unwrap();
|
||||
|
||||
// --8<-- [start:blob_api_setup]
|
||||
let db = connect("./blob_db").execute().await.unwrap();
|
||||
// --8<-- [end:blob_api_setup]
|
||||
|
||||
// --8<-- [start:blob_api_schema]
|
||||
let blob_schema = Arc::new(Schema::new(vec![
|
||||
Field::new("id", DataType::Int64, false),
|
||||
Field::new("label", DataType::Utf8, true),
|
||||
blob("image", true),
|
||||
]));
|
||||
// --8<-- [end:blob_api_schema]
|
||||
|
||||
// --8<-- [start:blob_api_ingest]
|
||||
let blob_table = db
|
||||
.create_empty_table("photos", blob_schema)
|
||||
.mode(CreateTableMode::Overwrite)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let input_schema = Arc::new(Schema::new(vec![
|
||||
Field::new("id", DataType::Int64, false),
|
||||
Field::new("label", DataType::Utf8, true),
|
||||
Field::new("image", DataType::LargeBinary, true),
|
||||
]));
|
||||
let input_batch = RecordBatch::try_new(
|
||||
input_schema.clone(),
|
||||
vec![
|
||||
Arc::new(Int64Array::from(vec![1_i64, 2_i64])),
|
||||
Arc::new(StringArray::from(vec![Some("door"), Some("window")])),
|
||||
Arc::new(LargeBinaryArray::from_iter_values([
|
||||
b"inline_image_bytes".as_slice(),
|
||||
b"second_image_bytes".as_slice(),
|
||||
])),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
let input_reader: Box<dyn arrow_array::RecordBatchReader + Send> = Box::new(
|
||||
RecordBatchIterator::new(vec![Ok(input_batch)].into_iter(), input_schema),
|
||||
);
|
||||
blob_table.add(input_reader).execute().await.unwrap();
|
||||
// --8<-- [end:blob_api_ingest]
|
||||
assert_eq!(blob_table.count_rows(None).await.unwrap(), 2);
|
||||
|
||||
// --8<-- [start:blob_api_files]
|
||||
let batches = blob_table
|
||||
.query()
|
||||
.with_row_id()
|
||||
.execute()
|
||||
.await
|
||||
.unwrap()
|
||||
.try_collect::<Vec<_>>()
|
||||
.await
|
||||
.unwrap();
|
||||
let row_ids = batches[0]
|
||||
.column_by_name("_rowid")
|
||||
.unwrap()
|
||||
.as_any()
|
||||
.downcast_ref::<UInt64Array>()
|
||||
.unwrap()
|
||||
.values()
|
||||
.to_vec();
|
||||
|
||||
let handles = blob_table
|
||||
.fetch_blob_files("image", &row_ids[..1])
|
||||
.await
|
||||
.unwrap();
|
||||
let header = handles[0].as_ref().unwrap().read_range(0..6).await.unwrap();
|
||||
handles[0].as_ref().unwrap().close().await.unwrap();
|
||||
// --8<-- [end:blob_api_files]
|
||||
assert_eq!(header.as_ref(), b"inline");
|
||||
|
||||
// --8<-- [start:blob_api_fetch]
|
||||
let payloads = blob_table.fetch_blobs("image", &row_ids).await.unwrap();
|
||||
// --8<-- [end:blob_api_fetch]
|
||||
assert_eq!(payloads.len(), 2);
|
||||
|
||||
// --8<-- [start:blob_api_ranges]
|
||||
let slices = blob_table
|
||||
.fetch_blob_ranges(
|
||||
"image",
|
||||
[
|
||||
BlobRangeRequest::new(row_ids[0], 0, 6),
|
||||
BlobRangeRequest::new(row_ids[1], 0, 6),
|
||||
BlobRangeRequest::new(row_ids[1], 7, 5),
|
||||
],
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let second_handles = blob_table
|
||||
.fetch_blob_files("image", &row_ids[1..2])
|
||||
.await
|
||||
.unwrap();
|
||||
let second_image = second_handles[0].as_ref().unwrap();
|
||||
let parts = second_image.read_ranges(&[0..6, 7..12]).await.unwrap();
|
||||
second_image.close().await.unwrap();
|
||||
// --8<-- [end:blob_api_ranges]
|
||||
assert_eq!(slices.value(0), b"inline");
|
||||
assert_eq!(slices.value(1), b"second");
|
||||
assert_eq!(slices.value(2), b"image");
|
||||
assert_eq!(parts[0].as_ref(), b"second");
|
||||
assert_eq!(parts[1].as_ref(), b"image");
|
||||
|
||||
std::env::set_current_dir(original_dir).unwrap();
|
||||
}
|
||||
|
||||
async fn deferred_fetch() {
|
||||
let temp_dir = tempfile::tempdir().unwrap();
|
||||
let db = connect(temp_dir.path().to_str().unwrap())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let schema = Arc::new(Schema::new(vec![
|
||||
Field::new("id", DataType::Int64, false),
|
||||
blob("image", true),
|
||||
]));
|
||||
let table = db
|
||||
.create_empty_table("photos", schema)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let input_schema = Arc::new(Schema::new(vec![
|
||||
Field::new("id", DataType::Int64, false),
|
||||
Field::new("image", DataType::LargeBinary, true),
|
||||
]));
|
||||
let batch = RecordBatch::try_new(
|
||||
input_schema.clone(),
|
||||
vec![
|
||||
Arc::new(Int64Array::from(vec![1_i64])),
|
||||
Arc::new(LargeBinaryArray::from_iter_values([
|
||||
b"deferred-bytes".as_slice()
|
||||
])),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
let reader: Box<dyn arrow_array::RecordBatchReader + Send> = Box::new(
|
||||
RecordBatchIterator::new(vec![Ok(batch)].into_iter(), input_schema),
|
||||
);
|
||||
table.add(reader).execute().await.unwrap();
|
||||
|
||||
let described = table
|
||||
.query()
|
||||
.execute()
|
||||
.await
|
||||
.unwrap()
|
||||
.try_collect::<Vec<_>>()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(descriptor_kind(&described[0], 0), 0);
|
||||
assert!(described[0].column_by_name("_rowid").is_none());
|
||||
|
||||
let located = table
|
||||
.query()
|
||||
.with_row_id()
|
||||
.execute()
|
||||
.await
|
||||
.unwrap()
|
||||
.try_collect::<Vec<_>>()
|
||||
.await
|
||||
.unwrap();
|
||||
let row_id = located[0]
|
||||
.column_by_name("_rowid")
|
||||
.unwrap()
|
||||
.as_any()
|
||||
.downcast_ref::<UInt64Array>()
|
||||
.unwrap()
|
||||
.value(0);
|
||||
let payloads = table.fetch_blobs("image", &[row_id]).await.unwrap();
|
||||
let handles = table.fetch_blob_files("image", &[row_id]).await.unwrap();
|
||||
let streamed = handles[0]
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.read_range(0..payloads.value(0).len() as u64)
|
||||
.await
|
||||
.unwrap();
|
||||
handles[0].as_ref().unwrap().close().await.unwrap();
|
||||
|
||||
assert_eq!(payloads.value(0), b"deferred-bytes");
|
||||
assert_eq!(streamed.as_ref(), b"deferred-bytes");
|
||||
}
|
||||
|
||||
async fn storage_tiers() {
|
||||
let cases = [
|
||||
(1_i64, marked(INLINE_MAX, 11), 0_u8),
|
||||
(2_i64, marked(INLINE_MAX + 1, 22), 1_u8),
|
||||
(3_i64, marked(DEDICATED_MIN, 33), 1_u8),
|
||||
(4_i64, marked(DEDICATED_MIN + 1, 44), 2_u8),
|
||||
];
|
||||
let temp_dir = tempfile::tempdir().unwrap();
|
||||
let db = connect(temp_dir.path().to_str().unwrap())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let schema = Arc::new(Schema::new(vec![
|
||||
Field::new("id", DataType::Int64, false),
|
||||
blob("image", true),
|
||||
]));
|
||||
let table = db
|
||||
.create_empty_table("tiers", schema)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let input_schema = Arc::new(Schema::new(vec![
|
||||
Field::new("id", DataType::Int64, false),
|
||||
Field::new("image", DataType::LargeBinary, true),
|
||||
]));
|
||||
let batch = RecordBatch::try_new(
|
||||
input_schema.clone(),
|
||||
vec![
|
||||
Arc::new(Int64Array::from_iter_values(cases.iter().map(|row| row.0))),
|
||||
Arc::new(LargeBinaryArray::from_iter_values(
|
||||
cases.iter().map(|row| row.1.as_slice()),
|
||||
)),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
let reader: Box<dyn arrow_array::RecordBatchReader + Send> = Box::new(
|
||||
RecordBatchIterator::new(vec![Ok(batch)].into_iter(), input_schema),
|
||||
);
|
||||
table.add(reader).execute().await.unwrap();
|
||||
|
||||
let described = table
|
||||
.query()
|
||||
.execute()
|
||||
.await
|
||||
.unwrap()
|
||||
.try_collect::<Vec<_>>()
|
||||
.await
|
||||
.unwrap();
|
||||
let located = table
|
||||
.query()
|
||||
.with_row_id()
|
||||
.execute()
|
||||
.await
|
||||
.unwrap()
|
||||
.try_collect::<Vec<_>>()
|
||||
.await
|
||||
.unwrap();
|
||||
let mut kind_by_id = std::collections::HashMap::new();
|
||||
for batch in &described {
|
||||
let ids = batch
|
||||
.column_by_name("id")
|
||||
.unwrap()
|
||||
.as_any()
|
||||
.downcast_ref::<Int64Array>()
|
||||
.unwrap();
|
||||
for row in 0..batch.num_rows() {
|
||||
kind_by_id.insert(ids.value(row), descriptor_kind(batch, row));
|
||||
}
|
||||
}
|
||||
let mut row_id_by_id = std::collections::HashMap::new();
|
||||
for batch in &located {
|
||||
let ids = batch
|
||||
.column_by_name("id")
|
||||
.unwrap()
|
||||
.as_any()
|
||||
.downcast_ref::<Int64Array>()
|
||||
.unwrap();
|
||||
let row_ids = batch
|
||||
.column_by_name("_rowid")
|
||||
.unwrap()
|
||||
.as_any()
|
||||
.downcast_ref::<UInt64Array>()
|
||||
.unwrap();
|
||||
for row in 0..batch.num_rows() {
|
||||
row_id_by_id.insert(ids.value(row), row_ids.value(row));
|
||||
}
|
||||
}
|
||||
|
||||
for (id, payload, kind) in &cases {
|
||||
assert_eq!(kind_by_id[id], *kind);
|
||||
let row_id = row_id_by_id[id];
|
||||
let fetched = table.fetch_blobs("image", &[row_id]).await.unwrap();
|
||||
assert_eq!(fetched.value(0), payload.as_slice());
|
||||
let handles = table.fetch_blob_files("image", &[row_id]).await.unwrap();
|
||||
let handle = handles[0].as_ref().unwrap();
|
||||
let head = handle.read_range(0..1).await.unwrap();
|
||||
let tail = handle
|
||||
.read_range((payload.len() - 1) as u64..payload.len() as u64)
|
||||
.await
|
||||
.unwrap();
|
||||
handle.close().await.unwrap();
|
||||
assert_eq!(head.as_ref(), &payload[..1]);
|
||||
assert_eq!(tail.as_ref(), &payload[payload.len() - 1..]);
|
||||
}
|
||||
}
|
||||
|
||||
fn marked(size: usize, marker: u8) -> Vec<u8> {
|
||||
let mut payload = vec![0_u8; size];
|
||||
payload[0] = marker;
|
||||
payload[size - 1] = marker;
|
||||
payload
|
||||
}
|
||||
|
||||
fn descriptor_kind(batch: &RecordBatch, row: usize) -> u8 {
|
||||
let image = batch
|
||||
.column_by_name("image")
|
||||
.unwrap()
|
||||
.as_any()
|
||||
.downcast_ref::<StructArray>()
|
||||
.unwrap_or_else(|| panic!("image column was {}", batch.schema()));
|
||||
image
|
||||
.column_by_name("kind")
|
||||
.unwrap()
|
||||
.as_any()
|
||||
.downcast_ref::<UInt8Array>()
|
||||
.unwrap()
|
||||
.value(row)
|
||||
}
|
||||
@@ -0,0 +1,204 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
import assert from "node:assert/strict";
|
||||
import { mkdtempSync, rmSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { chdir, cwd } from "node:process";
|
||||
import { test } from "node:test";
|
||||
import * as arrow from "apache-arrow";
|
||||
import { Buffer } from "node:buffer";
|
||||
import * as lancedb from "@lancedb/lancedb";
|
||||
|
||||
const INLINE_MAX = 64 * 1024;
|
||||
const DEDICATED_MIN = 4 * 1024 * 1024;
|
||||
|
||||
function marked(size: number, marker: number): Buffer {
|
||||
const payload = Buffer.alloc(size);
|
||||
payload[0] = marker;
|
||||
payload[payload.length - 1] = marker;
|
||||
return payload;
|
||||
}
|
||||
|
||||
function kindOf(value: unknown): number {
|
||||
assert.equal(typeof value, "object");
|
||||
assert.ok(value !== null);
|
||||
const kind = (value as { kind?: unknown }).kind;
|
||||
assert.equal(typeof kind, "number");
|
||||
return kind;
|
||||
}
|
||||
|
||||
test("blob column tutorial", async () => {
|
||||
const directory = mkdtempSync(join(tmpdir(), "blob-v2-"));
|
||||
const previous = cwd();
|
||||
chdir(directory);
|
||||
try {
|
||||
// --8<-- [start:blob_api_setup]
|
||||
const db = await lancedb.connect("./blob_db");
|
||||
// --8<-- [end:blob_api_setup]
|
||||
|
||||
// --8<-- [start:blob_api_schema]
|
||||
const blobSchema = new arrow.Schema([
|
||||
new arrow.Field("id", new arrow.Int64()),
|
||||
new arrow.Field("label", new arrow.Utf8()),
|
||||
lancedb.blob("image"),
|
||||
]);
|
||||
// --8<-- [end:blob_api_schema]
|
||||
|
||||
// --8<-- [start:blob_api_ingest]
|
||||
const blobTable = await db.createEmptyTable("photos", blobSchema, {
|
||||
mode: "overwrite",
|
||||
});
|
||||
|
||||
await blobTable.add([
|
||||
{ id: 1n, label: "door", image: Buffer.from("inline_image_bytes") },
|
||||
{ id: 2n, label: "window", image: Buffer.from("second_image_bytes") },
|
||||
]);
|
||||
// --8<-- [end:blob_api_ingest]
|
||||
assert.equal(await blobTable.countRows(), 2);
|
||||
|
||||
// --8<-- [start:blob_api_files]
|
||||
const rows = await blobTable
|
||||
.query()
|
||||
.select(["id", "label"])
|
||||
.withRowId()
|
||||
.toArray();
|
||||
const rowIds = rows.map((row) => row._rowid as bigint);
|
||||
|
||||
const [image] = await blobTable.fetchBlobFiles("image", [rowIds[0]!]);
|
||||
const header = await image!.readRange(0n, 6n);
|
||||
await image!.close();
|
||||
// --8<-- [end:blob_api_files]
|
||||
assert.equal(header.toString("utf8"), "inline");
|
||||
|
||||
// --8<-- [start:blob_api_fetch]
|
||||
const payloads = await blobTable.fetchBlobs("image", rowIds);
|
||||
// --8<-- [end:blob_api_fetch]
|
||||
assert.equal(payloads.length, 2);
|
||||
|
||||
// --8<-- [start:blob_api_ranges]
|
||||
const [secondImage] = await blobTable.fetchBlobFiles("image", [rowIds[1]!]);
|
||||
const parts = await secondImage!.readRanges([
|
||||
{ start: 0n, end: 6n },
|
||||
{ start: 7n, end: 12n },
|
||||
]);
|
||||
await secondImage!.close();
|
||||
// --8<-- [end:blob_api_ranges]
|
||||
assert.deepEqual(
|
||||
parts.map((part) => part.toString("utf8")),
|
||||
["second", "image"],
|
||||
);
|
||||
|
||||
// --8<-- [start:blob_api_bytes_mode]
|
||||
const withBytes = await blobTable
|
||||
.query()
|
||||
.select(["id", "image"])
|
||||
.toArray({ blobMode: "bytes" });
|
||||
const firstImage: Uint8Array | null = withBytes[0].image;
|
||||
// --8<-- [end:blob_api_bytes_mode]
|
||||
assert.equal(Buffer.from(firstImage!).toString("utf8"), "inline_image_bytes");
|
||||
assert.equal(Object.hasOwn(withBytes[0], "_rowid"), false);
|
||||
} finally {
|
||||
chdir(previous);
|
||||
rmSync(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("eager query returns blob bytes without a row id", async () => {
|
||||
const directory = mkdtempSync(join(tmpdir(), "blob-v2-eager-"));
|
||||
try {
|
||||
const db = await lancedb.connect(directory);
|
||||
const schema = new arrow.Schema([
|
||||
new arrow.Field("id", new arrow.Int64()),
|
||||
lancedb.blob("image"),
|
||||
]);
|
||||
const table = await db.createEmptyTable("photos", schema);
|
||||
const payload = Buffer.from("inline-bytes");
|
||||
await table.add([{ id: 1n, image: payload }]);
|
||||
|
||||
const rows = await table
|
||||
.query()
|
||||
.where("id = 1")
|
||||
.select(["id", "image"])
|
||||
.toArray({ blobMode: "bytes" });
|
||||
|
||||
assert.equal(rows.length, 1);
|
||||
assert.equal(Object.hasOwn(rows[0], "_rowid"), false);
|
||||
assert.equal(Buffer.from(rows[0].image).toString("utf8"), "inline-bytes");
|
||||
} finally {
|
||||
rmSync(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("deferred fetch reads a blob by row id", async () => {
|
||||
const directory = mkdtempSync(join(tmpdir(), "blob-v2-deferred-"));
|
||||
try {
|
||||
const db = await lancedb.connect(directory);
|
||||
const schema = new arrow.Schema([
|
||||
new arrow.Field("id", new arrow.Int64()),
|
||||
lancedb.blob("image"),
|
||||
]);
|
||||
const table = await db.createEmptyTable("photos", schema);
|
||||
await table.add([{ id: 1n, image: Buffer.from("deferred-bytes") }]);
|
||||
|
||||
const described = await table
|
||||
.query()
|
||||
.where("id = 1")
|
||||
.select(["id", "image"])
|
||||
.toArray();
|
||||
assert.equal(kindOf(described[0].image), 0);
|
||||
assert.equal(Object.hasOwn(described[0], "_rowid"), false);
|
||||
|
||||
const found = await table.query().where("id = 1").select(["id"]).withRowId().toArray();
|
||||
const rowId = found[0]._rowid as bigint;
|
||||
const payloads = await table.fetchBlobs("image", [rowId]);
|
||||
const [handle] = await table.fetchBlobFiles("image", [rowId]);
|
||||
const streamed = await handle!.readRange(0n, handle!.size());
|
||||
await handle!.close();
|
||||
|
||||
assert.equal(Buffer.from(payloads[0]).toString("utf8"), "deferred-bytes");
|
||||
assert.equal(Buffer.from(streamed).toString("utf8"), "deferred-bytes");
|
||||
} finally {
|
||||
rmSync(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("storage tiers round-trip", async () => {
|
||||
const cases = [
|
||||
{ id: 1n, image: marked(INLINE_MAX, 11), kind: 0 },
|
||||
{ id: 2n, image: marked(INLINE_MAX + 1, 22), kind: 1 },
|
||||
{ id: 3n, image: marked(DEDICATED_MIN, 33), kind: 1 },
|
||||
{ id: 4n, image: marked(DEDICATED_MIN + 1, 44), kind: 2 },
|
||||
];
|
||||
const directory = mkdtempSync(join(tmpdir(), "blob-v2-tiers-"));
|
||||
try {
|
||||
const db = await lancedb.connect(directory);
|
||||
const schema = new arrow.Schema([
|
||||
new arrow.Field("id", new arrow.Int64()),
|
||||
lancedb.blob("image"),
|
||||
]);
|
||||
const table = await db.createEmptyTable("tiers", schema);
|
||||
await table.add(cases.map(({ id, image }) => ({ id, image })));
|
||||
|
||||
const described = await table.query().select(["id", "image"]).toArray();
|
||||
const located = await table.query().select(["id"]).withRowId().toArray();
|
||||
const rowIdByValue = new Map(located.map((row) => [row.id as bigint, row._rowid as bigint]));
|
||||
|
||||
for (const item of cases) {
|
||||
const descriptor = described.find((row) => row.id === item.id);
|
||||
assert.ok(descriptor);
|
||||
assert.equal(kindOf(descriptor.image), item.kind);
|
||||
const rowId = rowIdByValue.get(item.id);
|
||||
assert.ok(rowId !== undefined);
|
||||
const [payload] = await table.fetchBlobs("image", [rowId]);
|
||||
assert.deepEqual(Buffer.from(payload), item.image);
|
||||
const [handle] = await table.fetchBlobFiles("image", [rowId]);
|
||||
const streamed = await handle!.read();
|
||||
await handle!.close();
|
||||
assert.deepEqual(streamed, item.image);
|
||||
}
|
||||
} finally {
|
||||
rmSync(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
+1565
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,16 @@
|
||||
{
|
||||
"name": "lancedb-blob-v2-docs",
|
||||
"private": true,
|
||||
"version": "0.0.0",
|
||||
"type": "module",
|
||||
"scripts": {
|
||||
"test": "node --experimental-strip-types --test multimodal.test.ts"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=22.6.0"
|
||||
},
|
||||
"dependencies": {
|
||||
"@lancedb/lancedb": "0.41.0-beta.8",
|
||||
"apache-arrow": "21.2.0"
|
||||
}
|
||||
}
|
||||
@@ -1,137 +1,134 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
import io
|
||||
import pytest
|
||||
import lancedb
|
||||
import numpy as np
|
||||
import pyarrow as pa
|
||||
|
||||
try:
|
||||
# --8<-- [start:multimodal_imports]
|
||||
import io
|
||||
import lancedb
|
||||
import numpy as np
|
||||
import pyarrow as pa
|
||||
import io
|
||||
from PIL import Image
|
||||
# --8<-- [end:multimodal_imports]
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
# --8<-- [start:multimodal_imports]
|
||||
import lancedb
|
||||
import pyarrow as pa
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
import io
|
||||
from PIL import Image
|
||||
# --8<-- [end:multimodal_imports]
|
||||
|
||||
def test_multimodal_ingestion(db_path_factory):
|
||||
# Ensure dependencies are available
|
||||
pytest.importorskip("PIL")
|
||||
pytest.importorskip("lancedb")
|
||||
pytest.importorskip("numpy")
|
||||
|
||||
# --8<-- [start:create_dummy_data]
|
||||
# Create some dummy images
|
||||
def create_dummy_image(color):
|
||||
img = Image.new('RGB', (100, 100), color=color)
|
||||
img = Image.new("RGB", (100, 100), color=color)
|
||||
buf = io.BytesIO()
|
||||
img.save(buf, format='PNG')
|
||||
img.save(buf, format="PNG")
|
||||
return buf.getvalue()
|
||||
|
||||
# Create dataset with metadata, vectors, and image blobs
|
||||
data = [
|
||||
{
|
||||
"id": 1,
|
||||
"filename": "red_square.png",
|
||||
"vector": np.random.rand(128).astype(np.float32),
|
||||
"image_blob": create_dummy_image('red'),
|
||||
"label": "red"
|
||||
"image_blob": create_dummy_image("red"),
|
||||
"label": "red",
|
||||
},
|
||||
{
|
||||
"id": 2,
|
||||
"filename": "blue_square.png",
|
||||
"vector": np.random.rand(128).astype(np.float32),
|
||||
"image_blob": create_dummy_image('blue'),
|
||||
"label": "blue"
|
||||
}
|
||||
"image_blob": create_dummy_image("blue"),
|
||||
"label": "blue",
|
||||
},
|
||||
]
|
||||
# --8<-- [end:create_dummy_data]
|
||||
|
||||
# --8<-- [start:define_schema]
|
||||
# Define schema explicitly to ensure image_blob is treated as binary
|
||||
schema = pa.schema([
|
||||
pa.field("id", pa.int32()),
|
||||
pa.field("filename", pa.string()),
|
||||
pa.field("vector", pa.list_(pa.float32(), 128)),
|
||||
pa.field("image_blob", pa.binary()), # Important: Use pa.binary() for blobs
|
||||
pa.field("label", pa.string())
|
||||
])
|
||||
schema = pa.schema(
|
||||
[
|
||||
pa.field("id", pa.int32()),
|
||||
pa.field("filename", pa.string()),
|
||||
pa.field("vector", pa.list_(pa.float32(), 128)),
|
||||
pa.field("image_blob", pa.binary()),
|
||||
pa.field("label", pa.string()),
|
||||
]
|
||||
)
|
||||
# --8<-- [end:define_schema]
|
||||
|
||||
db_uri = db_path_factory("multimodal_db")
|
||||
db = lancedb.connect(db_uri)
|
||||
db = lancedb.connect(db_path_factory("multimodal_db"))
|
||||
|
||||
# --8<-- [start:ingest_data]
|
||||
tbl = db.create_table("images", data=data, schema=schema, mode="overwrite")
|
||||
# --8<-- [end:ingest_data]
|
||||
|
||||
|
||||
assert len(tbl) == 2
|
||||
|
||||
# --8<-- [start:search_data]
|
||||
# Search for similar images
|
||||
query_vector = np.random.rand(128).astype(np.float32)
|
||||
results = tbl.search(query_vector).limit(1).to_pandas()
|
||||
# --8<-- [end:search_data]
|
||||
|
||||
# --8<-- [start:process_results]
|
||||
# Convert back to PIL Image
|
||||
for _, row in results.iterrows():
|
||||
image_bytes = row['image_blob']
|
||||
image_bytes = row["image_blob"]
|
||||
image = Image.open(io.BytesIO(image_bytes))
|
||||
print(f"Retrieved image: {row['filename']}, Size: {image.size}")
|
||||
# You can now use 'image' with other libraries or display it
|
||||
# --8<-- [end:process_results]
|
||||
|
||||
|
||||
assert len(results) == 1
|
||||
|
||||
def test_blob_api_definition(db_path_factory):
|
||||
# --8<-- [start:blob_api_schema]
|
||||
|
||||
def test_legacy_blob_api_definition(tmp_path, monkeypatch):
|
||||
monkeypatch.chdir(tmp_path)
|
||||
# --8<-- [start:legacy_blob_api_schema]
|
||||
import pyarrow as pa
|
||||
|
||||
# Define schema with Blob API metadata for lazy loading
|
||||
schema = pa.schema([
|
||||
pa.field("id", pa.int64()),
|
||||
pa.field(
|
||||
"video",
|
||||
pa.large_binary(),
|
||||
metadata={"lance-encoding:blob": "true"} # Enable Blob API
|
||||
),
|
||||
])
|
||||
# --8<-- [end:blob_api_schema]
|
||||
schema = pa.schema(
|
||||
[
|
||||
pa.field("id", pa.int64()),
|
||||
pa.field(
|
||||
"video",
|
||||
pa.large_binary(),
|
||||
metadata={"lance-encoding:blob": "true"}, # Enable Blob API
|
||||
),
|
||||
]
|
||||
)
|
||||
# --8<-- [end:legacy_blob_api_schema]
|
||||
|
||||
# --8<-- [start:blob_api_ingest]
|
||||
# --8<-- [start:legacy_blob_api_ingest]
|
||||
import lancedb
|
||||
|
||||
db = lancedb.connect(db_path_factory("blob_db"))
|
||||
|
||||
db = lancedb.connect("./blob_db")
|
||||
|
||||
# Create sample data
|
||||
data = [
|
||||
{"id": 1, "video": b"fake_video_bytes_1"},
|
||||
{"id": 2, "video": b"fake_video_bytes_2"}
|
||||
{"id": 2, "video": b"fake_video_bytes_2"},
|
||||
]
|
||||
|
||||
|
||||
# Create the table
|
||||
tbl = db.create_table("videos", data=data, schema=schema)
|
||||
# --8<-- [end:blob_api_ingest]
|
||||
# --8<-- [end:legacy_blob_api_ingest]
|
||||
assert len(tbl) == 2
|
||||
|
||||
|
||||
def test_blob_api_to_pandas(db_path_factory):
|
||||
db = lancedb.connect(db_path_factory("blob_to_pandas_db"))
|
||||
schema = pa.schema([
|
||||
pa.field("id", pa.int64()),
|
||||
pa.field(
|
||||
"video",
|
||||
pa.large_binary(),
|
||||
metadata={"lance-encoding:blob": "true"},
|
||||
),
|
||||
])
|
||||
def test_legacy_blob_api_to_pandas(db_path_factory):
|
||||
db = lancedb.connect(db_path_factory("legacy_blob_to_pandas_db"))
|
||||
schema = pa.schema(
|
||||
[
|
||||
pa.field("id", pa.int64()),
|
||||
pa.field(
|
||||
"video",
|
||||
pa.large_binary(),
|
||||
metadata={"lance-encoding:blob": "true"},
|
||||
),
|
||||
]
|
||||
)
|
||||
tbl = db.create_table(
|
||||
"videos",
|
||||
data=[
|
||||
@@ -142,7 +139,7 @@ def test_blob_api_to_pandas(db_path_factory):
|
||||
mode="overwrite",
|
||||
)
|
||||
|
||||
# --8<-- [start:blob_api_to_pandas]
|
||||
# --8<-- [start:legacy_blob_api_to_pandas]
|
||||
# Default: blob columns come back lazily
|
||||
df_lazy = tbl.to_pandas()
|
||||
|
||||
@@ -154,7 +151,7 @@ def test_blob_api_to_pandas(db_path_factory):
|
||||
|
||||
# Forward extra kwargs to PyArrow's to_pandas
|
||||
df_typed = tbl.to_pandas(split_blocks=True, self_destruct=True)
|
||||
# --8<-- [end:blob_api_to_pandas]
|
||||
# --8<-- [end:legacy_blob_api_to_pandas]
|
||||
|
||||
assert len(df_lazy) == 2
|
||||
assert isinstance(df_bytes["video"].iloc[0], bytes)
|
||||
@@ -167,16 +164,18 @@ def test_blob_api_to_pandas(db_path_factory):
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_query_to_pandas_kwargs(db_path_factory):
|
||||
schema = pa.schema([
|
||||
pa.field("id", pa.int64()),
|
||||
pa.field("vector", pa.list_(pa.float32(), 128)),
|
||||
pa.field(
|
||||
"video",
|
||||
pa.large_binary(),
|
||||
metadata={"lance-encoding:blob": "true"},
|
||||
),
|
||||
])
|
||||
async def test_legacy_query_to_pandas_kwargs(db_path_factory):
|
||||
schema = pa.schema(
|
||||
[
|
||||
pa.field("id", pa.int64()),
|
||||
pa.field("vector", pa.list_(pa.float32(), 128)),
|
||||
pa.field(
|
||||
"video",
|
||||
pa.large_binary(),
|
||||
metadata={"lance-encoding:blob": "true"},
|
||||
),
|
||||
]
|
||||
)
|
||||
data = [
|
||||
{
|
||||
"id": i,
|
||||
@@ -186,11 +185,11 @@ async def test_query_to_pandas_kwargs(db_path_factory):
|
||||
for i in range(10)
|
||||
]
|
||||
|
||||
db = lancedb.connect(db_path_factory("query_to_pandas_db"))
|
||||
db = lancedb.connect(db_path_factory("legacy_query_to_pandas_db"))
|
||||
tbl = db.create_table("search_demo", data=data, schema=schema, mode="overwrite")
|
||||
|
||||
async_db = await lancedb.connect_async(
|
||||
str(db_path_factory("query_to_pandas_async_db"))
|
||||
str(db_path_factory("legacy_query_to_pandas_async_db"))
|
||||
)
|
||||
tbl_async = await async_db.create_table(
|
||||
"search_demo", data=data, schema=schema, mode="overwrite"
|
||||
@@ -198,7 +197,8 @@ async def test_query_to_pandas_kwargs(db_path_factory):
|
||||
|
||||
query_vector = np.random.rand(128).astype(np.float32)
|
||||
|
||||
# --8<-- [start:query_to_pandas_kwargs]
|
||||
# fmt: off
|
||||
# --8<-- [start:legacy_query_to_pandas_kwargs]
|
||||
# Plain scan query: blob_mode is supported end to end
|
||||
df_lazy = (
|
||||
tbl.search()
|
||||
@@ -223,7 +223,8 @@ async def test_query_to_pandas_kwargs(db_path_factory):
|
||||
.select(["id", "vector"])
|
||||
.to_pandas(split_blocks=True, self_destruct=True)
|
||||
)
|
||||
# --8<-- [end:query_to_pandas_kwargs]
|
||||
# --8<-- [end:legacy_query_to_pandas_kwargs]
|
||||
# fmt: on
|
||||
|
||||
assert len(df_lazy) == 1
|
||||
assert len(df_bytes) == 1
|
||||
|
||||
@@ -7,7 +7,7 @@ use std::sync::Arc;
|
||||
|
||||
use arrow_array::types::Float32Type;
|
||||
use arrow_array::{
|
||||
BinaryArray, FixedSizeListArray, Int32Array, Int64Array, LargeBinaryArray, RecordBatch,
|
||||
Array, BinaryArray, FixedSizeListArray, Int32Array, Int64Array, LargeBinaryArray, RecordBatch,
|
||||
RecordBatchIterator, StringArray,
|
||||
};
|
||||
use arrow_schema::{DataType, Field, Schema};
|
||||
@@ -83,8 +83,9 @@ async fn main() {
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
let image_reader: Box<dyn arrow_array::RecordBatchReader + Send> =
|
||||
Box::new(RecordBatchIterator::new(vec![Ok(image_batch)].into_iter(), schema.clone()));
|
||||
let image_reader: Box<dyn arrow_array::RecordBatchReader + Send> = Box::new(
|
||||
RecordBatchIterator::new(vec![Ok(image_batch)].into_iter(), schema.clone()),
|
||||
);
|
||||
let table = db
|
||||
.create_table("images", image_reader)
|
||||
.mode(CreateTableMode::Overwrite)
|
||||
@@ -137,42 +138,45 @@ async fn main() {
|
||||
let search_rows: usize = results.iter().map(|batch| batch.num_rows()).sum();
|
||||
assert_eq!(search_rows, 1);
|
||||
|
||||
// --8<-- [start:blob_api_schema]
|
||||
let blob_metadata = HashMap::from([(
|
||||
// --8<-- [start:legacy_blob_api_schema]
|
||||
let legacy_blob_metadata = HashMap::from([(
|
||||
"lance-encoding:blob".to_string(),
|
||||
"true".to_string(),
|
||||
)]);
|
||||
let blob_schema = Schema::new(vec![
|
||||
let legacy_blob_schema = Schema::new(vec![
|
||||
Field::new("id", DataType::Int64, false),
|
||||
Field::new("video", DataType::LargeBinary, true).with_metadata(blob_metadata),
|
||||
Field::new("video", DataType::LargeBinary, true).with_metadata(legacy_blob_metadata),
|
||||
]);
|
||||
// --8<-- [end:blob_api_schema]
|
||||
// --8<-- [end:legacy_blob_api_schema]
|
||||
|
||||
// --8<-- [start:blob_api_ingest]
|
||||
let blob_rows = vec![
|
||||
// --8<-- [start:legacy_blob_api_ingest]
|
||||
let legacy_blob_rows = vec![
|
||||
(1_i64, b"fake_video_bytes_1".to_vec()),
|
||||
(2_i64, b"fake_video_bytes_2".to_vec()),
|
||||
];
|
||||
|
||||
let blob_schema = Arc::new(blob_schema);
|
||||
let blob_batch = RecordBatch::try_new(
|
||||
blob_schema.clone(),
|
||||
let legacy_blob_schema = Arc::new(legacy_blob_schema);
|
||||
let legacy_blob_batch = RecordBatch::try_new(
|
||||
legacy_blob_schema.clone(),
|
||||
vec![
|
||||
Arc::new(Int64Array::from_iter_values(blob_rows.iter().map(|row| row.0))),
|
||||
Arc::new(Int64Array::from_iter_values(
|
||||
legacy_blob_rows.iter().map(|row| row.0),
|
||||
)),
|
||||
Arc::new(LargeBinaryArray::from_iter_values(
|
||||
blob_rows.iter().map(|row| row.1.as_slice()),
|
||||
legacy_blob_rows.iter().map(|row| row.1.as_slice()),
|
||||
)),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
let blob_reader: Box<dyn arrow_array::RecordBatchReader + Send> =
|
||||
Box::new(RecordBatchIterator::new(vec![Ok(blob_batch)].into_iter(), blob_schema));
|
||||
let blob_table = db
|
||||
.create_table("videos", blob_reader)
|
||||
let legacy_blob_reader: Box<dyn arrow_array::RecordBatchReader + Send> = Box::new(
|
||||
RecordBatchIterator::new(vec![Ok(legacy_blob_batch)].into_iter(), legacy_blob_schema),
|
||||
);
|
||||
let legacy_blob_table = db
|
||||
.create_table("legacy_videos", legacy_blob_reader)
|
||||
.mode(CreateTableMode::Overwrite)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
// --8<-- [end:blob_api_ingest]
|
||||
assert_eq!(blob_table.count_rows(None).await.unwrap(), 2);
|
||||
// --8<-- [end:legacy_blob_api_ingest]
|
||||
assert_eq!(legacy_blob_table.count_rows(None).await.unwrap(), 2);
|
||||
}
|
||||
|
||||
@@ -75,8 +75,8 @@ test("multimodal snippets (async)", async () => {
|
||||
// --8<-- [end:process_results]
|
||||
expect(results).toHaveLength(1);
|
||||
|
||||
// --8<-- [start:blob_api_schema]
|
||||
const blobSchema = new arrow.Schema([
|
||||
// --8<-- [start:legacy_blob_api_schema]
|
||||
const legacyBlobSchema = new arrow.Schema([
|
||||
new arrow.Field("id", new arrow.Int64()),
|
||||
new arrow.Field(
|
||||
"video",
|
||||
@@ -85,20 +85,24 @@ test("multimodal snippets (async)", async () => {
|
||||
new Map([["lance-encoding:blob", "true"]]),
|
||||
),
|
||||
]);
|
||||
// --8<-- [end:blob_api_schema]
|
||||
// --8<-- [end:legacy_blob_api_schema]
|
||||
|
||||
// --8<-- [start:blob_api_ingest]
|
||||
const blobData = lancedb.makeArrowTable(
|
||||
// --8<-- [start:legacy_blob_api_ingest]
|
||||
const legacyBlobData = lancedb.makeArrowTable(
|
||||
[
|
||||
{ id: 1, video: Buffer.from("fake_video_bytes_1") },
|
||||
{ id: 2, video: Buffer.from("fake_video_bytes_2") },
|
||||
],
|
||||
{ schema: blobSchema },
|
||||
{ schema: legacyBlobSchema },
|
||||
);
|
||||
const blobTable = await db.createTable("videos", blobData, {
|
||||
mode: "overwrite",
|
||||
});
|
||||
// --8<-- [end:blob_api_ingest]
|
||||
expect(await blobTable.countRows()).toBe(2);
|
||||
const legacyBlobTable = await db.createTable(
|
||||
"legacy_videos",
|
||||
legacyBlobData,
|
||||
{
|
||||
mode: "overwrite",
|
||||
},
|
||||
);
|
||||
// --8<-- [end:legacy_blob_api_ingest]
|
||||
expect(await legacyBlobTable.countRows()).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user