mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-25 21:22:29 +00:00
fix(python): support custom blob thresholds without pylance (#4634)
This commit is contained in:
@@ -4,6 +4,7 @@
|
||||
|
||||
"""Schema helpers for Lance blob columns."""
|
||||
|
||||
import ctypes
|
||||
import importlib
|
||||
from typing import TYPE_CHECKING, Optional
|
||||
|
||||
@@ -16,6 +17,7 @@ if TYPE_CHECKING:
|
||||
_BLOB_EXTENSION_NAME = "lance.blob.v2"
|
||||
_BLOB_V1_KEY = "lance-encoding:blob"
|
||||
_ARROW_EXT_NAME_KEY = "ARROW:extension:name"
|
||||
_MAX_RUST_USIZE = ctypes.c_size_t(-1).value
|
||||
_BLOB_V2_STORAGE_TYPE = pa.struct(
|
||||
[
|
||||
pa.field("data", pa.large_binary(), nullable=True),
|
||||
@@ -209,32 +211,44 @@ def blob(
|
||||
Maximum size in bytes of a single packed blob sidecar file before
|
||||
a new one is started. Must be positive.
|
||||
|
||||
The thresholds are forwarded to ``lance.blob_field``, which validates
|
||||
them and stores them as field metadata, so pylance is required whenever
|
||||
any threshold is set. When all thresholds are omitted the behavior is
|
||||
unchanged.
|
||||
Thresholds are validated and stored as Arrow field metadata, with the
|
||||
same format as ``lance.blob_field``. They do not require pylance.
|
||||
"""
|
||||
if (
|
||||
inline_size_threshold is None
|
||||
and dedicated_size_threshold is None
|
||||
and pack_file_size_threshold is None
|
||||
metadata = {}
|
||||
for threshold, value, allow_zero, key in (
|
||||
(
|
||||
"inline_size_threshold",
|
||||
inline_size_threshold,
|
||||
True,
|
||||
"lance-encoding:blob-inline-size-threshold",
|
||||
),
|
||||
(
|
||||
"dedicated_size_threshold",
|
||||
dedicated_size_threshold,
|
||||
False,
|
||||
"lance-encoding:blob-dedicated-size-threshold",
|
||||
),
|
||||
(
|
||||
"pack_file_size_threshold",
|
||||
pack_file_size_threshold,
|
||||
False,
|
||||
"lance-encoding:blob-pack-file-size-threshold",
|
||||
),
|
||||
):
|
||||
blob_type = _resolve_blob_type()
|
||||
return pa.field(name, blob_type(), nullable=nullable)
|
||||
try:
|
||||
lance = importlib.import_module("lance")
|
||||
except ModuleNotFoundError as err:
|
||||
raise ValueError(
|
||||
"Setting blob tier thresholds requires pylance to be installed"
|
||||
) from err
|
||||
_resolve_blob_type()
|
||||
return lance.blob_field(
|
||||
name,
|
||||
nullable=nullable,
|
||||
inline_size_threshold=inline_size_threshold,
|
||||
dedicated_size_threshold=dedicated_size_threshold,
|
||||
pack_file_size_threshold=pack_file_size_threshold,
|
||||
)
|
||||
if value is None:
|
||||
continue
|
||||
if isinstance(value, bool) or not isinstance(value, int):
|
||||
raise TypeError(f"{threshold} must be an int, got {type(value).__name__}")
|
||||
if value < 0 or (value == 0 and not allow_zero):
|
||||
bound = "non-negative" if allow_zero else "positive"
|
||||
raise ValueError(f"{threshold} must be {bound}")
|
||||
if value > _MAX_RUST_USIZE:
|
||||
raise OverflowError(f"{threshold} must fit in a Rust usize")
|
||||
metadata[key] = str(value)
|
||||
|
||||
blob_type = _resolve_blob_type()
|
||||
field = pa.field(name, blob_type(), nullable=nullable)
|
||||
return field.with_metadata(metadata) if metadata else field
|
||||
|
||||
|
||||
def vector(dimension: int, value_type: pa.DataType = pa.float32()) -> pa.DataType:
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
import ctypes
|
||||
import io
|
||||
import subprocess
|
||||
import sys
|
||||
@@ -14,6 +15,7 @@ import pytest
|
||||
from lance.blob import Blob, BlobType as LanceBlobType
|
||||
|
||||
import lancedb
|
||||
from lancedb import schema as schema_module
|
||||
from lancedb._blob import (
|
||||
_set_blob_column,
|
||||
blob_v2_projection_sources,
|
||||
@@ -139,6 +141,41 @@ def test_blob_without_thresholds_has_no_threshold_metadata():
|
||||
assert not any(b"threshold" in key for key in metadata)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("nullable", [False, True])
|
||||
@pytest.mark.parametrize(
|
||||
"thresholds",
|
||||
[
|
||||
{},
|
||||
{"inline_size_threshold": 0},
|
||||
{"inline_size_threshold": 8192},
|
||||
{"dedicated_size_threshold": 1 << 20},
|
||||
{"pack_file_size_threshold": 1 << 26},
|
||||
{
|
||||
"inline_size_threshold": 8192,
|
||||
"dedicated_size_threshold": 1 << 20,
|
||||
"pack_file_size_threshold": 1 << 26,
|
||||
},
|
||||
{
|
||||
"inline_size_threshold": ctypes.c_size_t(-1).value,
|
||||
"dedicated_size_threshold": ctypes.c_size_t(-1).value,
|
||||
"pack_file_size_threshold": ctypes.c_size_t(-1).value,
|
||||
},
|
||||
],
|
||||
)
|
||||
def test_blob_thresholds_match_pylance(monkeypatch, nullable, thresholds):
|
||||
expected = lance.blob_field("payload", nullable=nullable, **thresholds)
|
||||
actual = lancedb.blob("payload", nullable=nullable, **thresholds)
|
||||
assert type(actual.type) is LanceBlobType
|
||||
assert actual.equals(expected, check_metadata=True)
|
||||
|
||||
# Both extension classes must encode the same Arrow field on the wire.
|
||||
monkeypatch.setattr(
|
||||
schema_module, "_resolve_blob_type", lambda: schema_module._FallbackBlobType
|
||||
)
|
||||
fallback = lancedb.blob("payload", nullable=nullable, **thresholds)
|
||||
assert pa.schema([fallback]).serialize() == pa.schema([expected]).serialize()
|
||||
|
||||
|
||||
def test_blob_thresholds_are_validated():
|
||||
with pytest.raises(ValueError, match="inline_size_threshold must be non-negative"):
|
||||
lancedb.blob("payload", inline_size_threshold=-1)
|
||||
|
||||
@@ -59,6 +59,124 @@ def _blob_v2_table(db: DBConnection, name: str):
|
||||
return table
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"threshold",
|
||||
[
|
||||
"inline_size_threshold",
|
||||
"dedicated_size_threshold",
|
||||
"pack_file_size_threshold",
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"value,error",
|
||||
[
|
||||
(-1, ValueError),
|
||||
(True, TypeError),
|
||||
(False, TypeError),
|
||||
(1.5, TypeError),
|
||||
("8192", TypeError),
|
||||
(ctypes.c_size_t(-1).value + 1, OverflowError),
|
||||
],
|
||||
)
|
||||
def test_blob_thresholds_reject_invalid_values(threshold, value, error):
|
||||
# Also runs in CI with pylance actually uninstalled.
|
||||
with pytest.raises(error, match=f"{threshold} must"):
|
||||
lancedb.blob("blob", **{threshold: value})
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"threshold", ["dedicated_size_threshold", "pack_file_size_threshold"]
|
||||
)
|
||||
def test_blob_thresholds_reject_zero_for_positive_thresholds(threshold):
|
||||
with pytest.raises(ValueError, match=f"{threshold} must be positive"):
|
||||
lancedb.blob("blob", **{threshold: 0})
|
||||
|
||||
|
||||
def test_blob_thresholds_accept_integer_boundaries():
|
||||
maximum = ctypes.c_size_t(-1).value
|
||||
field = lancedb.blob(
|
||||
"blob",
|
||||
nullable=False,
|
||||
inline_size_threshold=0,
|
||||
dedicated_size_threshold=maximum,
|
||||
pack_file_size_threshold=maximum,
|
||||
)
|
||||
assert not field.nullable
|
||||
assert field.metadata == {
|
||||
b"lance-encoding:blob-inline-size-threshold": b"0",
|
||||
b"lance-encoding:blob-dedicated-size-threshold": str(maximum).encode(),
|
||||
b"lance-encoding:blob-pack-file-size-threshold": str(maximum).encode(),
|
||||
}
|
||||
|
||||
|
||||
def test_blob_thresholds_control_writes_after_reopen(tmp_path):
|
||||
# Keep this in test_table.py: CI runs it without pylance or pandas installed.
|
||||
field = lancedb.blob(
|
||||
"blob",
|
||||
inline_size_threshold=4,
|
||||
dedicated_size_threshold=12,
|
||||
pack_file_size_threshold=8,
|
||||
)
|
||||
metadata = {
|
||||
b"lance-encoding:blob-inline-size-threshold": b"4",
|
||||
b"lance-encoding:blob-dedicated-size-threshold": b"12",
|
||||
b"lance-encoding:blob-pack-file-size-threshold": b"8",
|
||||
}
|
||||
assert field.metadata == metadata
|
||||
schema = pa.schema([pa.field("id", pa.int64()), field])
|
||||
db = lancedb.connect(tmp_path)
|
||||
table = db.create_table("blobs", schema=schema)
|
||||
# Every payload fits in the default inline tier; custom thresholds must
|
||||
# instead send the larger values to packed and dedicated storage.
|
||||
rows = [
|
||||
{"id": 0, "blob": b"a" * 2},
|
||||
{"id": 1, "blob": b"b" * 8},
|
||||
{"id": 2, "blob": b"c" * 16},
|
||||
{"id": 3, "blob": b"d" * 8},
|
||||
]
|
||||
|
||||
def check_writes(table, expected):
|
||||
assert table.schema.field("blob").metadata.items() >= metadata.items()
|
||||
hits = table.search().with_row_id(True).limit(10).to_arrow()
|
||||
actual = {row["id"]: row for row in hits.to_pylist()}
|
||||
assert actual.keys() == {row["id"] for row in expected}
|
||||
for row in expected:
|
||||
size = len(row["blob"])
|
||||
descriptor = actual[row["id"]]["blob"]
|
||||
assert descriptor["size"] == size
|
||||
assert descriptor["kind"] == (2 if size > 12 else 1 if size > 4 else 0)
|
||||
row_ids = [actual[row["id"]]["_rowid"] for row in expected]
|
||||
assert table.fetch_blobs("blob", row_ids).to_pylist() == [
|
||||
row["blob"] for row in expected
|
||||
]
|
||||
# Two packed values reach the configured sidecar limit independently.
|
||||
packed = [
|
||||
actual[row["id"]]["blob"] for row in expected if len(row["blob"]) == 8
|
||||
]
|
||||
assert len({desc["blob_id"] for desc in packed}) == len(packed)
|
||||
|
||||
table.add(rows)
|
||||
check_writes(table, rows)
|
||||
table = lancedb.connect(tmp_path).open_table("blobs")
|
||||
check_writes(table, rows)
|
||||
updates = [
|
||||
{"id": 0, "blob": b"e" * 16},
|
||||
{"id": 1, "blob": b"f" * 2},
|
||||
{"id": 2, "blob": b"g" * 8},
|
||||
{"id": 3, "blob": b"h" * 8},
|
||||
{"id": 4, "blob": b"i" * 16},
|
||||
]
|
||||
result = (
|
||||
table.merge_insert("id")
|
||||
.when_matched_update_all()
|
||||
.when_not_matched_insert_all()
|
||||
.execute(updates)
|
||||
)
|
||||
assert result.num_updated_rows == 4
|
||||
assert result.num_inserted_rows == 1
|
||||
check_writes(lancedb.connect(tmp_path).open_table("blobs"), updates)
|
||||
|
||||
|
||||
async def _blob_v2_table_async(db: AsyncConnection, name: str):
|
||||
schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("blob")])
|
||||
table = await db.create_table(name, schema=schema)
|
||||
|
||||
Reference in New Issue
Block a user