rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-10-22 19:52:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge main into gatekeeper/fix-4260-1

Browse Source 
Resolve Lance pins to beta.21 and refresh the vendored Arrow crate while preserving the sliced-child validity fix and regression tests.
This commit is contained in:
Gatefixer
2026-10-19 22:10:18 +00:00
parent b9da9a0c22 6507615a6f
commit eb4535398a
129 changed files with 1818 additions and 498 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/Cargo.toml
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lancedb-python"
version = "0.41.0-beta.7"
version = "0.41.0-beta.9"
publish = false
edition.workspace = true
description = "Python bindings for LanceDB"
python/README.md
+2
View File
@@ -8,6 +8,8 @@ A Python library for [LanceDB](https://github.com/lancedb/lancedb).
pip install lancedb
```
If no wheel is published for your platform, pip builds `lancedb` from the source distribution. That build needs a [Rust toolchain](https://rustup.rs/) and [protoc](https://grpc.io/docs/protoc-installation/) (the Protocol Buffers compiler) on your `PATH`, and can take a while.
### Pre-Haswell x86_64 hosts: `lancedb-compat`
The default `lancedb` wheel targets `x86-64-haswell` (AVX2 + FMA + F16C) for full performance on modern hardware. Pre-Haswell hosts — Intel Sandy Bridge / Ivy Bridge / Westmere; AMD Bulldozer / Piledriver / Steamroller — don't have AVX2 and crash with `Illegal instruction` at `import lancedb`.
python/python/lancedb/embeddings/__init__.py
+7 -1
View File
@@ -3,7 +3,12 @@
# ruff: noqa: F401
from .base import EmbeddingFunction, EmbeddingFunctionConfig, TextEmbeddingFunction
from .base import (
EmbeddingFunction,
EmbeddingFunctionConfig,
TextEmbeddingFunction,
UntrustedEmbeddingConfigError,
)
from .bedrock import BedRockText
from .cohere import CohereEmbeddingFunction
from .gemini_text import GeminiText
@@ -29,6 +34,7 @@ __all__ = [
"EmbeddingFunction",
"EmbeddingFunctionConfig",
"TextEmbeddingFunction",
"UntrustedEmbeddingConfigError",
"EmbeddingFunctionRegistry",
"get_registry",
"register",
python/python/lancedb/embeddings/base.py
+96 -2
View File
@@ -3,7 +3,7 @@
from abc import ABC, abstractmethod
import copy
from typing import List, Union
from typing import List, Optional, Tuple, Union
from lancedb.util import add_note
import numpy as np
@@ -13,6 +13,18 @@ from pydantic import BaseModel, Field, PrivateAttr
from .utils import TEXT, retry_with_exponential_backoff
class UntrustedEmbeddingConfigError(PermissionError):
"""
Raised when using an embedding function restored from table metadata whose
settings this process has not approved.
See
[allow_remote_code][lancedb.embeddings.EmbeddingFunctionRegistry.allow_remote_code]
and
[allow_endpoint][lancedb.embeddings.EmbeddingFunctionRegistry.allow_endpoint].
"""
class EmbeddingFunction(BaseModel, ABC):
"""
An ABC for embedding functions.
@@ -37,11 +49,29 @@ class EmbeddingFunction(BaseModel, ABC):
"""
Create an instance of the embedding function
"""
from .registry import EmbeddingFunctionRegistry
resolved_kwargs = cls.__resolveVariables(kwargs)
instance = cls(**resolved_kwargs)
instance._original_args = kwargs
instance._original_args = dict(kwargs)
if (
instance.remote_code_model() is not None
and "trust_remote_code" in cls.model_fields
):
# Record the flag even when left at its default, so a missing flag
# in table metadata only ever means a table written before this.
instance._original_args.setdefault(
"trust_remote_code", instance.trust_remote_code
)
# The caller wrote this config, so this process trusts it. Without this,
# a table created with it could not be reopened in the same process.
EmbeddingFunctionRegistry.get_instance()._trust(instance)
return instance
@classmethod
def _resolve_variables(cls, args: dict) -> dict:
return cls.__resolveVariables(args)
@classmethod
def __resolveVariables(cls, args: dict) -> dict:
"""
@@ -92,6 +122,38 @@ class EmbeddingFunction(BaseModel, ABC):
"""
return []
def remote_code_model(self) -> Optional[Tuple[str, Optional[str]]]:
"""
Return the model whose repository code this function runs when its
`trust_remote_code` field is enabled, as a `(name, revision)` pair.
Return None if the function never runs code from a model repository.
When the function is restored from table metadata, `trust_remote_code`
is only enabled for models allowed with
[allow_remote_code][lancedb.embeddings.EmbeddingFunctionRegistry.allow_remote_code].
"""
return None
@classmethod
def _legacy_trust_remote_code_default(cls) -> bool:
"""
The `trust_remote_code` to use for table metadata that omits it. Only
tables written before the flag was always recorded omit it, so this is
the default from that time.
"""
field = cls.model_fields.get("trust_remote_code")
return bool(field.default) if field is not None else False
def endpoints(self) -> List[str]:
"""
Return the non-default URLs this function sends data or credentials to.
When the function is restored from table metadata, each of these must be
allowed with
[allow_endpoint][lancedb.embeddings.EmbeddingFunctionRegistry.allow_endpoint].
"""
return []
@abstractmethod
def compute_query_embeddings(self, *args, **kwargs) -> list[Union[np.array, None]]:
"""
@@ -233,3 +295,35 @@ class TextEmbeddingFunction(EmbeddingFunction):
) -> list[Union[np.array, None]]:
"""Generate the embeddings for the given texts"""
pass
class UntrustedEmbeddingFunction(EmbeddingFunction):
"""
Stands in for an embedding function restored from table metadata whose
settings this process has not approved.
The table stays usable for anything that doesn't compute embeddings, while
computing embeddings raises
[UntrustedEmbeddingConfigError][lancedb.embeddings.UntrustedEmbeddingConfigError].
The real function is never constructed, since some load their model in
`__init__`.
"""
function_name: str
reasons: Tuple[str, ...]
def _error(self) -> UntrustedEmbeddingConfigError:
details = "\n".join(f" - {reason}" for reason in self.reasons)
return UntrustedEmbeddingConfigError(
f"Embedding function '{self.function_name}' was restored from table "
"metadata with settings this process has not approved:\n" + details
)
def compute_query_embeddings(self, *args, **kwargs):
raise self._error()
def compute_source_embeddings(self, *args, **kwargs):
raise self._error()
def ndims(self) -> int:
raise self._error()
python/python/lancedb/embeddings/ollama.py
+18 -1
View File
@@ -14,6 +14,9 @@ if TYPE_CHECKING:
import ollama
DEFAULT_OLLAMA_HOST = "http://localhost:11434"
@register("ollama")
class OllamaEmbeddings(TextEmbeddingFunction):
"""
@@ -24,11 +27,25 @@ class OllamaEmbeddings(TextEmbeddingFunction):
"""
name: str = "nomic-embed-text"
host: str = "http://localhost:11434"
host: str = DEFAULT_OLLAMA_HOST
options: Optional[dict] = None # type = ollama.Options
keep_alive: Optional[Union[float, str]] = None
ollama_client_kwargs: Optional[dict] = {}
def endpoints(self) -> List[str]:
urls = []
if self.host.rstrip("/") != DEFAULT_OLLAMA_HOST:
urls.append(self.host)
# These kwargs go to the underlying httpx client, where a proxy would
# see every request.
for key in ("proxy", "proxies"):
proxy = (self.ollama_client_kwargs or {}).get(key)
if isinstance(proxy, str):
urls.append(proxy)
elif isinstance(proxy, dict):
urls.extend(v for v in proxy.values() if isinstance(v, str))
return urls
def ndims(self) -> int:
return len(self.generate_embeddings(["foo"])[0])
python/python/lancedb/embeddings/openai.py
+3
View File
@@ -58,6 +58,9 @@ class OpenAIEmbeddings(TextEmbeddingFunction):
def sensitive_keys():
return ["api_key"]
def endpoints(self) -> List[str]:
return [self.base_url] if self.base_url else []
@staticmethod
def model_names():
return [
python/python/lancedb/embeddings/registry.py
+131 -6
View File
@@ -2,9 +2,15 @@
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
import json
from typing import Dict, Optional, Type
from typing import Dict, Optional, Set, Tuple, Type
from .base import EmbeddingFunction, EmbeddingFunctionConfig
from pydantic import TypeAdapter
from .base import (
EmbeddingFunction,
EmbeddingFunctionConfig,
UntrustedEmbeddingFunction,
)
class EmbeddingFunctionRegistry:
@@ -42,6 +48,9 @@ class EmbeddingFunctionRegistry:
def __init__(self):
self._functions = {}
self._variables = {}
# Revision is a str, None for the default branch, or _ANY_REVISION.
self._allowed_remote_code: Set[Tuple[str, object]] = set()
self._allowed_endpoints: Set[str] = set()
def register(self, alias: Optional[str] = None):
"""
@@ -122,20 +131,129 @@ class EmbeddingFunctionRegistry:
obj["vector_column"]: EmbeddingFunctionConfig(
vector_column=obj["vector_column"],
source_column=obj["source_column"],
function=self.get(obj["name"]).create(**obj["model"]),
function=self._restore(obj["name"], obj["model"]),
)
for obj in raw_list
}
def _restore(self, name: str, args: dict) -> EmbeddingFunction:
"""
Create an embedding function from table metadata. Anyone who can write
the table controls `args`, so settings that run model code or send data
elsewhere must already be trusted by this process.
"""
cls = self.get(name)
resolved = cls._resolve_variables(args)
# Inspect the config without running __init__, which may load a model.
probe = cls.model_construct(**resolved)
reasons = []
model = probe.remote_code_model()
if model is not None:
allowed = self._remote_code_allowed(*model)
explicit = "trust_remote_code" in resolved
if explicit:
requested = TypeAdapter(bool).validate_python(
resolved["trust_remote_code"]
)
else:
requested = cls._legacy_trust_remote_code_default()
# Always set the flag, since a subclass may default it to True.
resolved["trust_remote_code"] = requested and allowed
# A legacy table that omits the flag is downgraded rather than
# blocked: most sentence-transformers models don't need remote code,
# and those tables relied on the old True default without asking.
if requested and not allowed and explicit:
model_name, revision = model
reasons.append(
f"runs code from model '{model_name}' at revision "
f"{revision!r}. If you trust it, call "
f"get_registry().allow_remote_code({model_name!r}, "
f"revision={revision!r})"
)
reasons.extend(
f"sends data to endpoint '{url}'. If you trust it, call "
f"get_registry().allow_endpoint({url!r})"
for url in probe.endpoints()
if _normalize_url(url) not in self._allowed_endpoints
)
if reasons:
func = UntrustedEmbeddingFunction(
function_name=name, reasons=tuple(reasons)
)
else:
func = cls(**resolved)
func._original_args = args
return func
def allow_remote_code(self, model: str, revision: Optional[str] = None) -> None:
"""
Trust the code in a model repository when an embedding function that
loads it is restored from table metadata.
Table metadata cannot enable `trust_remote_code` by itself, since anyone
who can write to the table could then run code in this process. Embedding
functions created in this process with `trust_remote_code=True` are
allowed automatically.
Parameters
----------
model : str
The model name, such as `"nomic-ai/nomic-embed-text-v1.5"`.
revision : Optional[str]
The revision the table's embedding function must pin. Pinning a
commit hash is recommended, since the code at a branch or tag can
change after you review it. If None, any revision is trusted,
including pull request refs that anyone can push to the repository.
"""
self._allowed_remote_code.add(
(model, _ANY_REVISION if revision is None else revision)
)
def allow_endpoint(self, url: str) -> None:
"""
Trust a custom endpoint, such as an OpenAI `base_url` or Ollama `host`,
when an embedding function that uses it is restored from table metadata.
Table metadata cannot pick an endpoint by itself, since it could send
your data and API keys to a server you don't control. Embedding functions
created in this process are allowed to use their endpoints automatically.
Parameters
----------
url : str
The endpoint URL. Trailing slashes are ignored.
"""
self._allowed_endpoints.add(_normalize_url(url))
def _remote_code_allowed(self, model: str, revision: Optional[str]) -> bool:
return (model, _ANY_REVISION) in self._allowed_remote_code or (
model,
revision,
) in self._allowed_remote_code
def _trust(self, func: EmbeddingFunction) -> None:
model = func.remote_code_model()
if model is not None and getattr(func, "trust_remote_code", False):
# Only the revision this code selected, where None means the default
# branch. Other revisions, such as PR refs, may hold other code.
self._allowed_remote_code.add(model)
for url in func.endpoints():
self.allow_endpoint(url)
def function_to_metadata(self, conf: "EmbeddingFunctionConfig"):
"""
Convert the given embedding function and source / vector column configs
into a config dictionary that can be serialized into arrow metadata
"""
func = conf.function
name = getattr(
func, "__embedding_function_registry_alias__", func.__class__.__name__
)
if isinstance(func, UntrustedEmbeddingFunction):
name = func.function_name
else:
name = getattr(
func, "__embedding_function_registry_alias__", func.__class__.__name__
)
json_data = func.safe_model_dump()
return {
"name": name,
@@ -180,6 +298,13 @@ class EmbeddingFunctionRegistry:
return self._variables[name]
_ANY_REVISION = object()
def _normalize_url(url: str) -> str:
return url.rstrip("/")
# Global instance
__REGISTRY__ = EmbeddingFunctionRegistry()
python/python/lancedb/embeddings/sentence_transformers.py
+21 -5
View File
@@ -2,7 +2,7 @@
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
from typing import List, Union
from typing import List, Optional, Tuple, Union
import numpy as np
@@ -27,19 +27,32 @@ class SentenceTransformerEmbeddings(TextEmbeddingFunction):
The device to use for the model
normalize : bool, default True
Whether to normalize the embeddings
trust_remote_code : bool, default True
Whether to trust the remote code
trust_remote_code : bool, default False
Whether to run code from the model's repository. Only enable this for
repositories you trust. When restored from table metadata, this is only
enabled for models allowed with
[allow_remote_code][lancedb.embeddings.EmbeddingFunctionRegistry.allow_remote_code].
revision : Optional[str], default None
The model revision to load, such as a commit hash.
"""
name: str = "all-MiniLM-L6-v2"
device: str = "cpu"
normalize: bool = True
trust_remote_code: bool = True
trust_remote_code: bool = False
revision: Optional[str] = None
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._ndims = None
def remote_code_model(self) -> Optional[Tuple[str, Optional[str]]]:
return (self.name, self.revision)
@classmethod
def _legacy_trust_remote_code_default(cls) -> bool:
return True
@property
def embedding_model(self):
"""
@@ -84,5 +97,8 @@ class SentenceTransformerEmbeddings(TextEmbeddingFunction):
"sentence_transformers", "sentence-transformers"
)
return sentence_transformers.SentenceTransformer(
self.name, device=self.device, trust_remote_code=self.trust_remote_code
self.name,
device=self.device,
trust_remote_code=self.trust_remote_code,
revision=self.revision,
)
python/python/lancedb/embeddings/transformers.py
+14 -4
View File
@@ -3,7 +3,7 @@
from functools import cached_property
from typing import List, Any
from typing import List, Any, Optional, Tuple
import numpy as np
@@ -34,7 +34,11 @@ class TransformersEmbeddingFunction(EmbeddingFunction):
Whether or not to allow for custom models defined on the HuggingFace
Hub in their own modeling files. This option should only be set to True
for repositories you trust and in which you have read the code, as it
will execute code present on the Hub on your local machine.
will execute code present on the Hub on your local machine. When
restored from table metadata, this is only enabled for models allowed with
[allow_remote_code][lancedb.embeddings.EmbeddingFunctionRegistry.allow_remote_code].
revision : Optional[str]
The model revision to load, such as a commit hash. Default is None.
to download package, run :
`pip install transformers`
@@ -45,6 +49,7 @@ class TransformersEmbeddingFunction(EmbeddingFunction):
name: str = "colbert-ir/colbertv2.0"
device: str = "cpu"
trust_remote_code: bool = False
revision: Optional[str] = None
_tokenizer: Any = PrivateAttr()
_model: Any = PrivateAttr()
@@ -52,12 +57,17 @@ class TransformersEmbeddingFunction(EmbeddingFunction):
super().__init__(*args, **kwargs)
self._ndims = None
transformers = attempt_import_or_raise("transformers")
self._tokenizer = transformers.AutoTokenizer.from_pretrained(self.name)
self._tokenizer = transformers.AutoTokenizer.from_pretrained(
self.name, revision=self.revision
)
self._model = transformers.AutoModel.from_pretrained(
self.name, trust_remote_code=self.trust_remote_code
self.name, trust_remote_code=self.trust_remote_code, revision=self.revision
)
self._model.to(self.device)
def remote_code_model(self) -> Optional[Tuple[str, Optional[str]]]:
return (self.name, self.revision)
model_config = ConfigDict(ignored_types=(cached_property,))
def ndims(self):
python/python/lancedb/embeddings/utils.py
+5
View File
@@ -304,6 +304,11 @@ def _is_non_retryable_error(error: Exception) -> bool:
Returns:
True if the error should not be retried, False otherwise
"""
from .base import UntrustedEmbeddingConfigError
if isinstance(error, UntrustedEmbeddingConfigError):
return True
# Check for OpenAI authentication errors
error_type = type(error).__name__
if error_type == "AuthenticationError":
python/python/lancedb/embeddings/watsonx.py
+3
View File
@@ -87,6 +87,9 @@ class WatsonxEmbeddings(TextEmbeddingFunction):
url: Optional[str] = None
params: Optional[Dict] = None
def endpoints(self) -> List[str]:
return [self.url] if self.url else []
@staticmethod
def sensitive_keys():
return ["api_key"]
python/python/lancedb/functions.py
+2 -5
View File
@@ -697,6 +697,7 @@ _GRAMMAR_PRIMITIVES = (
(pa.string(), "utf8"),
(pa.large_string(), "large_utf8"),
(pa.binary(), "binary"),
(pa.large_binary(), "large_binary"),
(pa.date32(), "date32"),
(pa.date64(), "date64"),
)
@@ -846,11 +847,7 @@ def _exact_blob_storage_type(field: pa.Field) -> dict[str, Any]:
{
"name": child.name,
"nullable": child.nullable,
"type": (
{"type": "large_binary"}
if pa.types.is_large_binary(child.type)
else _exact_arrow_type(child.type)
),
"type": _exact_arrow_type(child.type),
}
for child in storage
],
python/python/lancedb/streaming.py
+19 -5
View File
@@ -261,10 +261,15 @@ class StreamingDataset(IterableDataset):
storage) at the cost of higher memory usage per split buffer. Defaults
to ``DEFAULT_READ_BATCH_SIZE`` (64).
io_queue_depth:
Number of I/O batches to keep in flight per split. Higher values
overlap storage latency with transform and training compute at the cost
of more memory and threads. Must be greater than zero. Defaults to
``DEFAULT_PREFETCH_BATCHES`` (4).
Number of fetched batches to keep ahead of the transform stage per
split: this bounds the combined count of in-flight fetches and
already-fetched batches still waiting for a transform slot. Higher
values overlap storage latency with transform and training compute
at the cost of more memory and threads, but I/O stops running ahead
once this many batches are buffered, applying backpressure when the
transform stage (or ``transform_queue_depth``) is the bottleneck.
Must be greater than zero. Defaults to ``DEFAULT_PREFETCH_BATCHES``
(4).
columns:
Optional list of column names to read. When set, only those columns
are fetched from storage; all others are omitted. ``None`` (the
@@ -836,8 +841,17 @@ class StreamingDataset(IterableDataset):
io_pending[i].append((abs_start, io_pool.submit(_io_call, perm_i, indices)))
def _fill_io(i: int) -> None:
# Budget covers both in-flight fetches (io_pending) and batches
# that have already landed but are still waiting for a transform
# slot (raw_batches). Without raw_batches in the count, a slow
# transform stage never applies backpressure to Stage 1: each
# completed fetch is drained into raw_batches and immediately
# replaced by a new one, so I/O keeps racing arbitrarily far
# ahead of the transform stage instead of merely overlapping
# storage latency with it.
while (
len(io_pending[i]) < io_queue_depth and fetch_head[i] < split_sizes[i]
len(io_pending[i]) + len(raw_batches[i]) < io_queue_depth
and fetch_head[i] < split_sizes[i]
):
_submit_io(i)
python/python/lancedb/table.py
+13 -4
View File
@@ -341,10 +341,19 @@ def _blob_input_to_arrow(data: Any, schema: Optional[pa.Schema]) -> Optional[pa.
if isinstance(field.type, pa.ExtensionType)
else field.type
)
storage = pa.array(
[_blob_value_to_storage(value) for value in values[field.name]],
type=storage_type,
)
storage_values = [_blob_value_to_storage(value) for value in values[field.name]]
missing_range_fields = {"position", "size"} - set(storage_type.names)
# Arrow silently ignores dictionary keys absent from the storage type.
if missing_range_fields and any(
value is not None
and any(value.get(name) is not None for name in missing_range_fields)
for value in storage_values
):
raise ValueError(
f"Blob column '{field.name}' storage schema cannot store byte ranges. "
"Create the column with lancedb.blob() to use position and size."
)
storage = pa.array(storage_values, type=storage_type)
column = (
pa.ExtensionArray.from_storage(field.type, storage)
if isinstance(field.type, pa.ExtensionType)
python/python/tests/test_blob.py
+84
View File
@@ -2305,6 +2305,90 @@ def test_add_external_uri_string_round_trips_with_flag(tmp_path):
assert blobs[0].as_py() == payload
@pytest.mark.parametrize("as_pandas", [False, True])
@pytest.mark.parametrize("as_blob", [False, True])
@pytest.mark.parametrize("minimal_schema", [False, True])
@pytest.mark.parametrize("position", [0, 4])
def test_add_external_blob_range_matches_storage_schema(
tmp_path, as_pandas, as_blob, minimal_schema, position
):
payload = b"0123456789abcdef"
blob_path = tmp_path / "payload.bin"
blob_path.write_bytes(payload)
field = (
pa.field("image", pa.large_binary(), metadata={"lance-encoding:blob": "true"})
if minimal_schema
else lancedb.blob("image")
)
db = lancedb.connect(tmp_path / "db")
table = db.create_table(
"ranges", schema=pa.schema([pa.field("id", pa.int64()), field])
)
value = (
Blob.from_uri(blob_path.as_uri(), position=position, size=6)
if as_blob
else {"uri": blob_path.as_uri(), "position": position, "size": 6}
)
rows = [{"id": 1, "image": b"inline"}, {"id": 2, "image": value}]
data = pd.DataFrame(rows) if as_pandas else rows
if minimal_schema:
with pytest.raises(ValueError, match="Blob column 'image'.*storage schema"):
table.add(data, allow_external_blob_outside_bases=True)
assert table.count_rows() == 0
else:
table.add(data, allow_external_blob_outside_bases=True)
by_id = _row_ids_by_id(table)
assert table.fetch_blobs("image", [by_id[1], by_id[2]]).to_pylist() == [
b"inline",
payload[position : position + 6],
]
@pytest.mark.parametrize("as_pandas", [False, True])
def test_add_minimal_blob_schema_preserves_whole_file_and_inline_values(
tmp_path, as_pandas
):
payload = b"whole-file"
blob_path = tmp_path / "payload.bin"
blob_path.write_bytes(payload)
uri = blob_path.as_uri()
schema = pa.schema(
[
pa.field("id", pa.int64()),
pa.field(
"image", pa.large_binary(), metadata={"lance-encoding:blob": "true"}
),
]
)
table = lancedb.connect(tmp_path / "db").create_table("minimal", schema=schema)
values = [
None,
b"inline",
Blob.from_bytes(b"from-blob"),
uri,
Blob.from_uri(uri),
{"uri": uri, "position": None, "size": None},
]
rows = [{"id": i, "image": value} for i, value in enumerate(values)]
table.add(
pd.DataFrame(rows) if as_pandas else rows,
allow_external_blob_outside_bases=True,
)
by_id = _row_ids_by_id(table)
assert table.fetch_blobs(
"image", [by_id[i] for i in range(len(values))]
).to_pylist() == [
None,
b"inline",
b"from-blob",
payload,
payload,
payload,
]
@pytest.mark.parametrize("as_pandas", [False, True])
def test_add_bytes_and_uri_in_one_batch_preserves_external_uri(tmp_path, as_pandas):
payload = b"external-payload"
python/python/tests/test_elastic_dataloader.py
+86
View File
@@ -2036,6 +2036,92 @@ def test_transform_queue_depth_bounds_cooked_rows(lance_table):
)
def test_raw_queue_depth_bounded_when_transform_is_the_bottleneck(tmp_path):
"""io_queue_depth bounds raw_batches even when transform is the bottleneck.
Regression test for a bug where Stage 1 (I/O) applied no backpressure at
all: _fill_io only capped io_pending (in-flight fetches) at
io_queue_depth, but every completed fetch was drained into raw_batches
unconditionally (_drain_io) and immediately replaced by a new fetch
(_fill_io again), so I/O kept racing arbitrarily far ahead of a slow
transform stage -- raw_batches had no cap of its own.
transform_queue_depth did not help either: its admission check in
_try_submit_tx only counts cooked rows and in-flight transform futures
(`len(cooked[i]) + len(tx_pending[i]) * batch_size`); it never looks at
len(raw_batches[i]), so it cannot throttle Stage 1.
The fix makes _fill_io count both io_pending and raw_batches against the
io_queue_depth budget, so raw_queue_depth should never exceed
io_queue_depth * read_batch_size per split, regardless of how slow the
transform stage is.
I/O is forced to run synchronously (same technique as
test_transform_queue_depth_does_not_admit_at_capacity_minus_one) so the
assertion is deterministic regardless of machine load, rather than
relying on a real-time race between I/O and transform threads. Only the
transform runs on a real background thread (with a short sleep) so the
consumer genuinely blocks waiting for it, letting Stage 1 attempt to run
ahead between blocking waits.
"""
import concurrent.futures as cf
from concurrent.futures import ThreadPoolExecutor
from unittest.mock import patch
import pyarrow as pa
import time
n_rows = 200
read_batch_size = 10
io_queue_depth = 2
transform_queue_depth = 2
max_allowed = io_queue_depth * read_batch_size
db = lancedb.connect(tmp_path)
table = db.create_table("t", pa.table({"id": list(range(n_rows))}))
def slow_transform(batch: pa.RecordBatch) -> list:
time.sleep(0.005)
return batch.column("id").to_pylist()
original_submit = ThreadPoolExecutor.submit
def synchronous_io_submit(self, fn, *args, **kwargs):
if getattr(fn, "__name__", "") == "_io_call":
fut = cf.Future()
try:
fut.set_result(fn(*args, **kwargs))
except Exception as exc:
fut.set_exception(exc)
return fut
return original_submit(self, fn, *args, **kwargs)
with patch.object(ThreadPoolExecutor, "submit", synchronous_io_submit):
ds = StreamingDataset(
table,
num_splits=1,
shuffle_seed=SHUFFLE_SEED,
read_batch_size=read_batch_size,
io_queue_depth=io_queue_depth,
transform_queue_depth=transform_queue_depth,
transform_parallelism=1,
transform=slow_transform,
)
peak_raw = 0
ids = []
for item in ds:
ids.append(item) # slow_transform yields bare ids, not dicts
if ds.raw_queue_depth > peak_raw:
peak_raw = ds.raw_queue_depth
assert peak_raw <= max_allowed, (
f"raw_queue_depth peaked at {peak_raw}, expected <= {max_allowed} "
"(io_queue_depth * read_batch_size)"
)
# The backpressure fix must not come at the cost of dropping or
# duplicating rows.
assert sorted(ids) == list(range(n_rows))
def test_transform_queue_depth_does_not_admit_at_capacity_minus_one(tmp_path):
"""Admission requires a full read_batch_size of free space, not just one slot.
python/python/tests/test_embeddings_trust.py
+291
View File
@@ -0,0 +1,291 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors
"""
Restoring an embedding function from table metadata must not let the table grant
itself trust: remote model code and custom endpoints need approval from the
process that opens the table.
"""
import json
from unittest.mock import patch
import lancedb
import pyarrow as pa
import pytest
from lancedb.embeddings import (
EmbeddingFunctionConfig,
OllamaEmbeddings,
OpenAIEmbeddings,
SentenceTransformerEmbeddings,
UntrustedEmbeddingConfigError,
get_registry,
)
from lancedb.embeddings.base import UntrustedEmbeddingFunction
@pytest.fixture(autouse=True)
def fresh_trust(monkeypatch):
registry = get_registry()
monkeypatch.setattr(registry, "_allowed_remote_code", set())
monkeypatch.setattr(registry, "_allowed_endpoints", set())
def metadata_for(name: str, model: dict) -> dict:
entry = {
"name": name,
"model": model,
"source_column": "text",
"vector_column": "vector",
}
return {b"embedding_functions": json.dumps([entry]).encode()}
def restore(name: str, model: dict):
return get_registry().parse_functions(metadata_for(name, model))["vector"].function
def test_sentence_transformers_does_not_trust_remote_code_by_default():
assert SentenceTransformerEmbeddings.create().trust_remote_code is False
@pytest.mark.parametrize("name", ["sentence-transformers", "huggingface", "colbert"])
def test_restored_remote_code_requires_allowlist(name):
model = {"name": "attacker/model", "trust_remote_code": True}
func = restore(name, model)
with pytest.raises(UntrustedEmbeddingConfigError, match="attacker/model"):
func.compute_query_embeddings("hello")
with pytest.raises(UntrustedEmbeddingConfigError):
func.ndims()
def test_blocked_function_does_not_construct_model():
# huggingface loads its model in __init__, so the check has to happen first.
with patch(
"lancedb.embeddings.transformers.attempt_import_or_raise"
) as attempt_import:
restore("huggingface", {"name": "attacker/model", "trust_remote_code": True})
attempt_import.assert_not_called()
def test_allowlisted_model_restores_with_remote_code():
get_registry().allow_remote_code("org/model")
func = restore(
"sentence-transformers", {"name": "org/model", "trust_remote_code": True}
)
assert isinstance(func, SentenceTransformerEmbeddings)
assert func.trust_remote_code is True
def test_allowlist_does_not_cover_other_models():
get_registry().allow_remote_code("org/model")
func = restore(
"sentence-transformers", {"name": "org/other", "trust_remote_code": True}
)
with pytest.raises(UntrustedEmbeddingConfigError, match="org/other"):
func.compute_query_embeddings("hello")
def test_pinned_revision_must_match():
get_registry().allow_remote_code("org/model", revision="abc123")
pinned = restore(
"sentence-transformers",
{"name": "org/model", "revision": "abc123", "trust_remote_code": True},
)
assert pinned.trust_remote_code is True
for model in [
{"name": "org/model", "revision": "def456", "trust_remote_code": True},
{"name": "org/model", "trust_remote_code": True},
]:
func = restore("sentence-transformers", model)
with pytest.raises(UntrustedEmbeddingConfigError, match="org/model"):
func.compute_query_embeddings("hello")
def test_restored_trust_flag_follows_allowlist_when_absent():
# Tables written before the default changed don't record trust_remote_code.
assert restore(
"sentence-transformers", {"name": "org/model"}
).trust_remote_code is (False)
get_registry().allow_remote_code("org/model")
assert restore("sentence-transformers", {"name": "org/model"}).trust_remote_code
def test_created_function_records_trust_flag():
# Otherwise a new table using the default would look like a pre-fix table,
# where a missing flag meant True.
conf = EmbeddingFunctionConfig(
source_column="text",
vector_column="vector",
function=SentenceTransformerEmbeddings.create(name="org/model"),
)
assert get_registry().function_to_metadata(conf)["model"] == {
"name": "org/model",
"trust_remote_code": False,
}
get_registry().allow_remote_code("org/model")
metadata = get_registry().get_table_metadata([conf])
restored = get_registry().parse_functions(metadata)["vector"].function
assert restored.trust_remote_code is False
def test_huggingface_missing_flag_keeps_false_default():
# huggingface always defaulted to False, so approval must not flip it.
get_registry().allow_remote_code("org/model")
with patch("lancedb.embeddings.transformers.attempt_import_or_raise"):
func = restore("huggingface", {"name": "org/model"})
assert func.trust_remote_code is False
def test_restored_explicit_false_stays_false():
get_registry().allow_remote_code("org/model")
func = restore(
"sentence-transformers", {"name": "org/model", "trust_remote_code": False}
)
assert func.trust_remote_code is False
def test_create_trusts_its_own_config(tmp_path):
func = SentenceTransformerEmbeddings.create(
name="org/model", trust_remote_code=True
)
db = lancedb.connect(tmp_path)
schema = pa.schema(
[pa.field("text", pa.string()), pa.field("vector", pa.list_(pa.float32(), 4))]
)
db.create_table(
"t",
schema=schema,
embedding_functions=[
EmbeddingFunctionConfig(
source_column="text", vector_column="vector", function=func
)
],
)
restored = db.open_table("t").embedding_functions["vector"].function
assert isinstance(restored, SentenceTransformerEmbeddings)
assert restored.trust_remote_code is True
def test_create_trusts_only_the_revision_it_selected():
# Anyone can push a PR ref to a Hugging Face repo, so trusting a model's
# default branch must not extend to its other revisions.
SentenceTransformerEmbeddings.create(name="org/model", trust_remote_code=True)
func = restore(
"sentence-transformers",
{"name": "org/model", "revision": "refs/pr/123", "trust_remote_code": True},
)
assert isinstance(func, UntrustedEmbeddingFunction)
same = restore(
"sentence-transformers", {"name": "org/model", "trust_remote_code": True}
)
assert same.trust_remote_code is True
def test_explicit_unpinned_approval_covers_any_revision():
get_registry().allow_remote_code("org/model")
func = restore(
"sentence-transformers",
{"name": "org/model", "revision": "abc123", "trust_remote_code": True},
)
assert func.trust_remote_code is True
def test_missing_flag_does_not_inherit_unapproved_trust_default():
class LegacySentenceTransformer(SentenceTransformerEmbeddings):
trust_remote_code: bool = True
registry = get_registry()
registry.register("legacy-trust-default")(LegacySentenceTransformer)
try:
func = restore("legacy-trust-default", {"name": "attacker/model"})
assert not func.trust_remote_code
finally:
del registry._functions["legacy-trust-default"]
def test_trusting_one_model_does_not_trust_a_swapped_name():
SentenceTransformerEmbeddings.create(name="org/model", trust_remote_code=True)
func = restore(
"sentence-transformers", {"name": "attacker/model", "trust_remote_code": True}
)
with pytest.raises(UntrustedEmbeddingConfigError):
func.compute_query_embeddings("hello")
ATTACKER_URL = "https://attacker.example/v1"
@pytest.mark.parametrize(
"name,model",
[
("openai", {"base_url": ATTACKER_URL}),
("ollama", {"host": ATTACKER_URL}),
("ollama", {"ollama_client_kwargs": {"proxy": ATTACKER_URL}}),
("watsonx", {"url": ATTACKER_URL}),
],
)
def test_restored_custom_endpoint_requires_allowlist(name, model):
func = restore(name, model)
with pytest.raises(UntrustedEmbeddingConfigError, match="attacker.example"):
func.compute_query_embeddings("hello")
get_registry().allow_endpoint(ATTACKER_URL)
assert isinstance(restore(name, model), get_registry().get(name))
def test_default_endpoints_need_no_allowlist():
assert isinstance(restore("openai", {}), OpenAIEmbeddings)
assert isinstance(
restore("ollama", {"host": "http://localhost:11434"}), OllamaEmbeddings
)
def test_endpoint_allowlist_ignores_trailing_slash():
get_registry().allow_endpoint("http://my-host:11434/")
assert isinstance(
restore("ollama", {"host": "http://my-host:11434"}), OllamaEmbeddings
)
def test_create_trusts_its_own_endpoint():
OpenAIEmbeddings.create(base_url="http://my-proxy/v1")
assert isinstance(
restore("openai", {"base_url": "http://my-proxy/v1"}), OpenAIEmbeddings
)
def test_blocked_function_round_trips_metadata():
model = {"name": "attacker/model", "trust_remote_code": True}
func = restore("sentence-transformers", model)
conf = EmbeddingFunctionConfig(
source_column="text", vector_column="vector", function=func
)
entry = get_registry().function_to_metadata(conf)
assert entry["name"] == "sentence-transformers"
assert entry["model"] == model
def test_table_with_blocked_function_still_opens(tmp_path):
db = lancedb.connect(tmp_path)
schema = pa.schema(
[pa.field("text", pa.string()), pa.field("vector", pa.list_(pa.float32(), 2))],
metadata=metadata_for(
"sentence-transformers",
{"name": "attacker/model", "trust_remote_code": True},
),
)
db.create_table("t", schema=schema)
tbl = db.open_table("t")
tbl.add([{"text": "hi", "vector": [1.0, 2.0]}])
assert tbl.search([1.0, 2.0]).limit(1).to_list()[0]["text"] == "hi"
with pytest.raises(UntrustedEmbeddingConfigError, match="allow_remote_code"):
tbl.search("hi").limit(1).to_list()
python/python/tests/test_first_class_function_slice2.py
+22 -1
View File
@@ -622,7 +622,6 @@ def test_canonical_arrow_type_prefers_the_compact_grammar():
for outside in [
pa.timestamp("us"),
pa.decimal128(10, 2),
pa.large_binary(),
pa.binary(4),
pa.duration("s"),
pa.list_(pa.float32(), 0),
@@ -1280,6 +1279,28 @@ def test_explicit_large_utf8_schemas_use_the_canonical_function_name():
}
def test_large_binary_schemas_use_the_canonical_function_name():
payload = pa.field("data", pa.large_binary(), nullable=True)
@udf(
input_schema=pa.schema([payload]),
output_schema=pa.field("result", pa.large_binary(), nullable=False),
)
def preserve(data):
return data
signature = preserve.registration_request.signature
assert signature.inputs[0].arrow_type == "large_binary"
assert signature.output.arrow_type == "large_binary"
cells = pa.large_list(pa.field("item", pa.struct([payload]), nullable=True))
assert json.loads(_canonical_arrow_type(cells))["fields"][0]["type"] == {
"type": "struct",
"fields": [
{"name": "data", "nullable": True, "type": {"type": "large_binary"}}
],
}
def test_nested_struct_output_uses_canonical_exact_json():
token = pa.struct(
[
python/vendor/lance-arrow/Cargo.toml
Vendored
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "lance-arrow"
version = "14.0.0-beta.19"
version = "14.0.0-beta.21"
edition = "2024"
authors = ["Lance Devs <dev@lance.org>"]
license = "Apache-2.0"
python/vendor/lance-arrow/README.lancedb.md
Vendored
+2 -2
View File
@@ -1,5 +1,5 @@
This directory contains `rust/lance-arrow` from Lance tag `v14.0.0-beta.19`
(commit `42e593fb4ad0e1ccc22c1fa5e3c30bfe92a5384d`), licensed under Apache-2.0.
This directory contains `rust/lance-arrow` from Lance tag `v14.0.0-beta.21`
(commit `8c233c2e6aa6948524f352e5e4cd52bd1364f6ff`), licensed under Apache-2.0.
LanceDB carries a local fix in `src/lib.rs` for sliced child validity bitmaps
until a Lance release includes the fix. The patch is selected in the root
python/vendor/lance-arrow/src/lib.rs
Vendored
+174 -122
View File
@@ -912,145 +912,94 @@ fn project(struct_array: &StructArray, fields: &Fields) -> Result<StructArray> {
StructArray::try_new(fields.clone(), columns, struct_array.nulls().cloned())
}
fn lists_have_same_offsets_helper<T: OffsetSizeTrait>(left: &dyn Array, right: &dyn Array) -> bool {
let left_list: &GenericListArray<T> = left.as_list();
let right_list: &GenericListArray<T> = right.as_list();
left_list.offsets().inner() == right_list.offsets().inner()
/// Whether `data_type` has a struct somewhere below it, through lists.
fn holds_struct(data_type: &DataType) -> bool {
match data_type {
DataType::Struct(_) => true,
DataType::List(item) | DataType::LargeList(item) => holds_struct(item.data_type()),
_ => false,
}
}
fn merge_list_structs_helper<T: OffsetSizeTrait>(
/// Merge the values of two lists: structs field by field, nested lists
/// recursively, anything else keeps the left side.
fn merge_list_values(left: &ArrayRef, right: &ArrayRef) -> ArrayRef {
match (left.data_type(), right.data_type()) {
(DataType::Struct(_), DataType::Struct(_)) => {
Arc::new(merge(left.as_struct(), right.as_struct()))
}
(DataType::List(_), DataType::List(_))
| (DataType::LargeList(_), DataType::LargeList(_))
if left.data_type() != right.data_type()
&& holds_struct(left.data_type())
&& holds_struct(right.data_type()) =>
{
merge_list_struct(left, right)
}
_ => left.clone(),
}
}
fn merge_list_struct_helper<T: OffsetSizeTrait>(
left: &dyn Array,
right: &dyn Array,
items_field_name: impl Into<String>,
items_nullable: bool,
) -> Arc<dyn Array> {
let left_list: &GenericListArray<T> = left.as_list();
let right_list: &GenericListArray<T> = right.as_list();
let left_struct = left_list.values();
let right_struct = right_list.values();
let left_struct_arr = left_struct.as_struct();
let right_struct_arr = right_struct.as_struct();
let merged_items = Arc::new(merge(left_struct_arr, right_struct_arr));
let item_name = match left.data_type() {
DataType::List(item) | DataType::LargeList(item) => item.name().clone(),
_ => unreachable!(),
};
// One side all-null (a placeholder for columns this fragment lacks) has
// no values, so stand in nulls shaped like the other side's.
let null_left = left_list.null_count() == left_list.len();
let null_right = !null_left && right_list.null_count() == right_list.len();
let (merged_items, list, items_nullable) = if null_left || null_right {
let not_null = if null_left { right_list } else { left_list };
let values_len = not_null.values().len();
let values = |list: &GenericListArray<T>, is_null: bool| {
if is_null {
new_null_array(list.values().data_type(), values_len)
} else {
list.values().clone()
}
};
let merged = merge_list_values(
&values(left_list, null_left),
&values(right_list, null_right),
);
(merged, not_null, true)
} else {
if left_list.offsets().inner() != right_list.offsets().inner() {
panic!("Attempt to merge list struct arrays which do not have same offsets");
}
let merged = merge_list_values(left_list.values(), right_list.values());
let nullable = match left.data_type() {
DataType::List(item) | DataType::LargeList(item) => item.is_nullable(),
_ => unreachable!(),
};
(merged, left_list, nullable)
};
let items_field = Arc::new(Field::new(
items_field_name,
item_name,
merged_items.data_type().clone(),
items_nullable,
));
Arc::new(GenericListArray::<T>::new(
items_field,
left_list.offsets().clone(),
list.offsets().clone(),
merged_items,
left_list.nulls().cloned(),
list.nulls().cloned(),
))
}
fn merge_list_struct_null_helper<T: OffsetSizeTrait>(
left: &dyn Array,
right: &dyn Array,
not_null: &dyn Array,
items_field_name: impl Into<String>,
) -> Arc<dyn Array> {
let left_list: &GenericListArray<T> = left.as_list::<T>();
let not_null_list = not_null.as_list::<T>();
let right_list = right.as_list::<T>();
let left_struct = left_list.values().as_struct();
let not_null_struct: &StructArray = not_null_list.values().as_struct();
let right_struct = right_list.values().as_struct();
let values_len = not_null_list.values().len();
let mut merged_fields =
Vec::with_capacity(not_null_struct.num_columns() + right_struct.num_columns());
let mut merged_columns =
Vec::with_capacity(not_null_struct.num_columns() + right_struct.num_columns());
for (_, field) in left_struct.columns().iter().zip(left_struct.fields()) {
merged_fields.push(field.clone());
if let Some(val) = not_null_struct.column_by_name(field.name()) {
merged_columns.push(val.clone());
} else {
merged_columns.push(new_null_array(field.data_type(), values_len))
}
}
for (_, field) in right_struct
.columns()
.iter()
.zip(right_struct.fields())
.filter(|(_, field)| left_struct.column_by_name(field.name()).is_none())
{
merged_fields.push(field.clone());
if let Some(val) = not_null_struct.column_by_name(field.name()) {
merged_columns.push(val.clone());
} else {
merged_columns.push(new_null_array(field.data_type(), values_len));
}
}
let merged_struct = Arc::new(StructArray::new(
Fields::from(merged_fields),
merged_columns,
not_null_struct.nulls().cloned(),
));
let items_field = Arc::new(Field::new(
items_field_name,
merged_struct.data_type().clone(),
true,
));
Arc::new(GenericListArray::<T>::new(
items_field,
not_null_list.offsets().clone(),
merged_struct,
not_null_list.nulls().cloned(),
))
}
fn merge_list_struct_null(
left: &dyn Array,
right: &dyn Array,
not_null: &dyn Array,
) -> Arc<dyn Array> {
match left.data_type() {
DataType::List(left_field) => {
merge_list_struct_null_helper::<i32>(left, right, not_null, left_field.name())
}
DataType::LargeList(left_field) => {
merge_list_struct_null_helper::<i64>(left, right, not_null, left_field.name())
}
_ => unreachable!(),
}
}
/// Merge two lists whose items hold structs. Only succeeds if both lists have
/// the same offsets or one of them is all-null; panics otherwise.
fn merge_list_struct(left: &dyn Array, right: &dyn Array) -> Arc<dyn Array> {
// Merging fields into a list<struct<...>> is tricky and can only succeed
// in two ways. First, if both lists have the same offsets. Second, if
// one of the lists is all-null
if left.null_count() == left.len() {
return merge_list_struct_null(left, right, right);
} else if right.null_count() == right.len() {
return merge_list_struct_null(left, right, left);
}
match (left.data_type(), right.data_type()) {
(DataType::List(left_field), DataType::List(_)) => {
if !lists_have_same_offsets_helper::<i32>(left, right) {
panic!("Attempt to merge list struct arrays which do not have same offsets");
}
merge_list_structs_helper::<i32>(
left,
right,
left_field.name(),
left_field.is_nullable(),
)
}
(DataType::LargeList(left_field), DataType::LargeList(_)) => {
if !lists_have_same_offsets_helper::<i64>(left, right) {
panic!("Attempt to merge list struct arrays which do not have same offsets");
}
merge_list_structs_helper::<i64>(
left,
right,
left_field.name(),
left_field.is_nullable(),
)
(DataType::List(_), DataType::List(_)) => merge_list_struct_helper::<i32>(left, right),
(DataType::LargeList(_), DataType::LargeList(_)) => {
merge_list_struct_helper::<i64>(left, right)
}
_ => unreachable!(),
}
@@ -1279,8 +1228,9 @@ fn merge(left_struct_array: &StructArray, right_struct_array: &StructArray) -> S
columns.push(Arc::new(merged_sub_array) as ArrayRef);
}
(DataType::List(left_list), DataType::List(right_list))
if left_list.data_type().is_struct()
&& right_list.data_type().is_struct() =>
| (DataType::LargeList(left_list), DataType::LargeList(right_list))
if holds_struct(left_list.data_type())
&& holds_struct(right_list.data_type()) =>
{
// Identical inner types: nothing to merge, keep the left column.
if left_list.data_type() == right_list.data_type() {
@@ -1932,6 +1882,108 @@ mod tests {
assert_eq!(merged, expected);
}
#[test]
fn test_merge_all_null_list_adds_nested_child() {
// An all-null placeholder (what a fragment reader merges in for children
// missing from its files) adds `b` to a struct below one or two lists.
let strukt = |cols: Vec<(&str, ArrayRef)>| -> ArrayRef {
let fields = cols
.iter()
.map(|(name, col)| Field::new(*name, col.data_type().clone(), true))
.collect::<Vec<_>>();
Arc::new(StructArray::new(
fields.into(),
cols.into_iter().map(|(_, col)| col).collect(),
None,
))
};
// Two rows: `[values]` and null.
let list = |values: ArrayRef, large: bool| -> ArrayRef {
let field = Arc::new(Field::new("item", values.data_type().clone(), true));
let lengths = [values.len(), 0];
let nulls = Some(arrow_buffer::NullBuffer::from(vec![true, false]));
if large {
Arc::new(LargeListArray::new(
field,
OffsetBuffer::from_lengths(lengths),
values,
nulls,
))
} else {
Arc::new(ListArray::new(
field,
OffsetBuffer::from_lengths(lengths),
values,
nulls,
))
}
};
let a = || Arc::new(Int32Array::from(vec![7])) as ArrayRef;
let b = || new_null_array(&DataType::Utf8, 1);
for large in [false, true] {
// list<struct<t, in: struct<..>>>, built with or without `t`.
let under_list = |inner: ArrayRef, with_t: bool| {
let mut cols = vec![("in", inner)];
if with_t {
cols.insert(0, ("t", Arc::new(Int32Array::from(vec![1])) as ArrayRef));
}
list(strukt(cols), large)
};
// list<list<struct<..>>>
let under_two_lists = |inner: ArrayRef, _: bool| {
let inner_list: ArrayRef = if large {
let field = Arc::new(Field::new("item", inner.data_type().clone(), true));
Arc::new(LargeListArray::new(
field,
OffsetBuffer::from_lengths([1]),
inner,
None,
))
} else {
let field = Arc::new(Field::new("item", inner.data_type().clone(), true));
Arc::new(ListArray::new(
field,
OffsetBuffer::from_lengths([1]),
inner,
None,
))
};
list(inner_list, large)
};
let shapes: [&dyn Fn(ArrayRef, bool) -> ArrayRef; 2] = [&under_list, &under_two_lists];
for shape in shapes {
let data = shape(strukt(vec![("a", a())]), true);
let expected = shape(strukt(vec![("a", a()), ("b", b())]), true);
let placeholder =
new_null_array(shape(strukt(vec![("b", b())]), false).data_type(), 2);
let batch = |col: ArrayRef| {
RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new(
"c",
col.data_type().clone(),
true,
)])),
vec![col],
)
.unwrap()
};
let expected = batch(expected);
// Either side may be the placeholder; field order follows the left.
for merged in [
batch(data.clone())
.merge(&batch(placeholder.clone()))
.unwrap(),
batch(placeholder.clone())
.merge(&batch(data.clone()))
.unwrap(),
] {
let merged = merged.project_by_schema(expected.schema_ref()).unwrap();
assert_eq!(merged, expected, "large={large}");
}
}
}
}
#[test]
fn test_merge_list_struct_identical_schema() {
// Merging two batches whose `List<Struct>` columns have identical types