Merge branch 'main' of https://github.com/lancedb/lancedb into merge-insert-source-stats

# Conflicts:
#	rust/lancedb/src/remote/table.rs
#	rust/lancedb/src/table/merge.rs
#	rust/lancedb/src/table/merge/lsm.rs

python/Cargo.toml

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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)
@@ -579,7 +588,7 @@ def _cast_to_target_schema(
 def _preserve_blob_v2_fields(
     inferred_schema: pa.Schema, existing_schema: pa.Schema
 ) -> pa.Schema:
-    """Keep existing blob fields when an overwrite infers a replacement schema."""
+    """Keep inferred blob semantics without overriding explicit replacement types."""
     existing_fields = {field.name: field for field in existing_schema}
     fields = [
         _preserve_blob_v2_field(field, existing_fields[field.name])
@@ -591,6 +600,9 @@ def _preserve_blob_v2_fields(
 
 
 def _preserve_blob_v2_field(inferred: pa.Field, existing: pa.Field) -> pa.Field:
+    if is_blob_v2_field(inferred):
+        # An explicitly declared blob layout is part of the replacement schema.
+        return inferred
     if is_blob_v2_field(existing):
         return existing
 
@@ -628,12 +640,10 @@ def _preserve_blob_v2_field(inferred: pa.Field, existing: pa.Field) -> pa.Field:
 def _coerce_blob_write_columns(
     batch: pa.RecordBatch, target_schema: pa.Schema
 ) -> pa.RecordBatch:
-    """Materialize blob storage structs before the stream leaves Python.
+    """Materialize Python bytes and URIs as blob structs before Arrow casting.
 
-    merge_insert requires its source reader to already match the table's
-    physical schema. Unlike add and insert, it does not pass through
-    LanceDB's Rust blob coercion, so preserving binary input here would
-    reach Lance as binary and fail the schema check.
+    Existing blob structs retain their input layout. Rust normalizes them to
+    the stored schema and rejects populated ranges on two-field targets.
     """
     columns = []
     fields = []
@@ -964,6 +974,17 @@ def _align_field(field: pa.Field, target_field: pa.Field) -> pa.Field:
                 "(Utf8, LargeUtf8, Utf8View), or a Struct with a "
                 "'data' or 'uri' child"
             )
+        if pa.types.is_struct(source_type) or (
+            isinstance(source_type, pa.ExtensionType)
+            and (
+                not isinstance(target_field.type, pa.ExtensionType)
+                or type(source_type) is type(target_field.type)
+            )
+        ):
+            # Keep the source layout until Rust can fill missing range fields or
+            # validate every batch before dropping all-null range fields. Arrow
+            # casting here would reject differing layouts before that validation.
+            return field.with_metadata(target_field.metadata)
     # LanceDB exposes stored JSON columns as lance.json (JSONB-backed LargeBinary), but
     # casting the input to that storage type here merely relabels the raw JSON bytes as
     # JSONB. Lance must see arrow.json so it can perform the JSONB encoding.
@@ -1823,8 +1844,8 @@ class Table(ABC):
             - pyarrow.Table or pyarrow.RecordBatch
         mode: str
             The mode to use when writing the data. Valid values are
-            "append" and "overwrite". Overwrite infers a replacement schema
-            but preserves existing blob v2 fields present in the new data.
+            "append" and "overwrite". Overwrite adopts explicitly declared blob
+            layouts, while inferred bytes retain the existing blob v2 fields.
         on_bad_vectors: str, default "error"
             What to do if any of the vectors are not the same size or contains NaNs.
             One of "error", "drop", "fill".
@@ -2190,8 +2211,9 @@ class Table(ABC):
         One missing or deleted row ID rejects the entire batch; no partial
         results are returned.
 
-        Row IDs can be obtained from a query with ``with_row_id(True)``. This
-        API is currently supported only by local tables.
+        Row IDs can be obtained from a query with ``with_row_id(True)``.
+        Remote tables read the requested ranges through version-pinned blob
+        handles using HTTP byte-range requests.
         """
 
     @abstractmethod
@@ -4101,8 +4123,8 @@ class LanceTable(Table):
             The data to insert into the table.
         mode: str
             The mode to use when writing the data. Valid values are
-            "append" and "overwrite". Overwrite infers a replacement schema
-            but preserves existing blob v2 fields present in the new data.
+            "append" and "overwrite". Overwrite adopts explicitly declared blob
+            layouts, while inferred bytes retain the existing blob v2 fields.
         on_bad_vectors: str, default "error"
             What to do if any of the vectors are not the same size or contains NaNs.
             One of "error", "drop", "fill", "null".
@@ -6135,8 +6157,8 @@ class AsyncTable:
             - pyarrow.Table or pyarrow.RecordBatch
         mode: str
             The mode to use when writing the data. Valid values are
-            "append" and "overwrite". Overwrite infers a replacement schema
-            but preserves existing blob v2 fields present in the new data.
+            "append" and "overwrite". Overwrite adopts explicitly declared blob
+            layouts, while inferred bytes retain the existing blob v2 fields.
         on_bad_vectors: str, default "error"
             What to do if any of the vectors are not the same size or contains NaNs.
             One of "error", "drop", "fill", "null".

python/python/tests/test_blob.py

@@ -63,6 +63,25 @@ def _blob_array(name, values):
     return pa.ExtensionArray.from_storage(blob_type, storage)
 
 
+def _blob_layout_data(rows, with_ranges, *, as_extension=True):
+    field = lancedb.blob("image")
+    storage_type = field.type.storage_type
+    if not with_ranges or not as_extension:
+        storage_type = pa.struct(list(storage_type)[: (4 if with_ranges else 2)])
+        field = pa.field(
+            "image",
+            storage_type,
+            metadata={"ARROW:extension:name": "lance.blob.v2"},
+        )
+    images = pa.array([row["image"] for row in rows], type=storage_type)
+    if isinstance(field.type, pa.ExtensionType):
+        images = pa.ExtensionArray.from_storage(field.type, images)
+    return pa.Table.from_arrays(
+        [pa.array([row["id"] for row in rows], type=pa.int64()), images],
+        schema=pa.schema([pa.field("id", pa.int64()), field]),
+    )
+
+
 def _row_ids_by_id(table):
     hits = table.search().with_row_id(True).limit(1000).to_arrow()
     assert "_rowid" in hits.column_names
@@ -998,6 +1017,133 @@ def test_merge_insert_writes_python_bytes():
     assert blobs.to_pylist() == [b"updated", b"inserted"]
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+@pytest.mark.parametrize("input_type", ["extension", "struct", "reader"])
+@pytest.mark.parametrize("target_with_ranges", [False, True])
+async def test_merge_insert_normalizes_blob_layouts(
+    use_async, input_type, target_with_ranges
+):
+    data = _blob_layout_data(
+        [{"id": 1, "image": {"data": b"before"}}], target_with_ranges
+    )
+    table = lancedb.connect("memory:///").create_table("merge_layouts", data=data)
+    original_schema = table.schema
+    updates = _blob_layout_data(
+        [
+            {"id": 1, "image": {"data": b"updated"}},
+            {"id": 2, "image": {"data": b"inserted"}},
+            {"id": 3, "image": None},
+        ],
+        not target_with_ranges,
+        as_extension=input_type != "struct",
+    )
+    if input_type == "reader":
+        updates = updates.to_reader(max_chunksize=1)
+    writer = table._table if use_async else table
+    result = (
+        writer.merge_insert("id")
+        .when_matched_update_all()
+        .when_not_matched_insert_all()
+        .execute(updates)
+    )
+    if use_async:
+        result = await result
+
+    assert result.num_updated_rows == 1
+    assert result.num_inserted_rows == 2
+    assert table.schema == original_schema
+    by_id = _row_ids_by_id(table)
+    assert table.fetch_blobs("image", [by_id[i] for i in (1, 2, 3)]).to_pylist() == [
+        b"updated",
+        b"inserted",
+        None,
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+@pytest.mark.parametrize(
+    "byte_range", [{"position": 0}, {"size": 6}, {"position": 4, "size": 6}]
+)
+async def test_merge_insert_legacy_blob_rejects_ranges_atomically(
+    use_async, byte_range
+):
+    data = _blob_layout_data([{"id": 1, "image": {"data": b"before"}}], False)
+    table = lancedb.connect("memory:///").create_table("reject_ranges", data=data)
+    original_schema, original_version = table.schema, table.version
+    updates = _blob_layout_data(
+        [
+            {"id": 1, "image": {"data": b"updated"}},
+            {"id": 2, "image": {"uri": "file:///payload", **byte_range}},
+        ],
+        True,
+    ).to_reader(max_chunksize=1)
+    writer = table._table if use_async else table
+    with pytest.raises(RuntimeError, match="cannot store byte ranges"):
+        result = (
+            writer.merge_insert("id")
+            .when_matched_update_all()
+            .when_not_matched_insert_all()
+            .execute(updates)
+        )
+        if use_async:
+            await result
+
+    assert table.version == original_version
+    assert table.schema == original_schema
+    assert table.count_rows() == 1
+    assert table.fetch_blobs("image", [_row_ids_by_id(table)[1]]).to_pylist() == [
+        b"before"
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+@pytest.mark.parametrize("input_type", ["extension", "struct", "reader"])
+@pytest.mark.parametrize("source_with_ranges", [False, True])
+async def test_add_overwrite_adopts_explicit_blob_layout(
+    tmp_path, use_async, input_type, source_with_ranges
+):
+    data = _blob_layout_data(
+        [{"id": 1, "image": {"data": b"before"}}], not source_with_ranges
+    )
+    table = lancedb.connect("memory:///").create_table("overwrite_layouts", data=data)
+    payload = b"0123456789abcdef"
+    path = tmp_path / "payload.bin"
+    path.write_bytes(payload)
+    value = (
+        {"uri": path.as_uri(), "position": 4, "size": 6}
+        if source_with_ranges
+        else {"data": b"updated"}
+    )
+    updates = _blob_layout_data(
+        [{"id": 2, "image": value}, {"id": 3, "image": None}],
+        source_with_ranges,
+        as_extension=input_type != "struct",
+    )
+    if input_type == "reader":
+        updates = updates.to_reader(max_chunksize=1)
+    writer = table._table if use_async else table
+    result = writer.add(
+        updates, mode="overwrite", allow_external_blob_outside_bases=True
+    )
+    if use_async:
+        await result
+
+    blob_type = table.schema.field("image").type
+    if isinstance(blob_type, pa.ExtensionType):
+        blob_type = blob_type.storage_type
+    assert blob_type.num_fields == (4 if source_with_ranges else 2)
+    assert table.blob_columns() == ["image"]
+    assert table.count_rows() == 2
+    by_id = _row_ids_by_id(table)
+    assert table.fetch_blobs("image", [by_id[2], by_id[3]]).to_pylist() == [
+        payload[4:10] if source_with_ranges else b"updated",
+        None,
+    ]
+
+
 @pytest.mark.parametrize("write", ["merge_insert", "create_table"])
 def test_uri_string_blob_write_reaches_external_base_validation(tmp_path, write):
     blob_path = tmp_path / "payload.bin"
@@ -2305,6 +2451,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

@@ -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

@@ -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

@@ -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/python/tests/test_remote_db.py

@@ -1623,7 +1623,6 @@ def remote_unsupported_operations():
     return [
         ("uses_v2_manifest_paths", (), {}, "uses_v2_manifest_paths"),
         ("migrate_v2_manifest_paths", (), {}, "migrate_manifest_paths_v2"),
-        ("fetch_blob_ranges", ("image", [(1, 0, 1)]), {}, "fetch_blob_ranges"),
         ("optimize", (), {}, "optimize"),
         ("create_index", ("vector",), {"config": HnswPq()}, "IVF_HNSW_PQ"),
     ]
@@ -3255,6 +3254,89 @@ BLOB_DESCRIBE_RESPONSE = {
 }
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+@pytest.mark.parametrize("mode", ["merge", "overwrite"])
+@pytest.mark.parametrize("target_with_ranges", [False, True])
+async def test_remote_writes_normalize_blob_layouts(
+    use_async, mode, target_with_ranges
+):
+    description = json.loads(json.dumps(BLOB_DESCRIBE_RESPONSE))
+    if target_with_ranges:
+        description["schema"]["fields"][1]["type"]["fields"].extend(
+            {"name": name, "type": {"type": "uint64"}, "nullable": True}
+            for name in ("position", "size")
+        )
+    received = []
+
+    def handler(request):
+        if request.path == "/v1/table/test/describe/":
+            response = description
+        else:
+            expected = "merge_insert" if mode == "merge" else "insert"
+            assert urlparse(request.path).path == f"/v1/table/test/{expected}/"
+            if request.headers.get("Transfer-Encoding") == "chunked":
+                body = bytearray()
+                while size := int(request.rfile.readline().split(b";")[0], 16):
+                    body.extend(request.rfile.read(size))
+                    assert request.rfile.read(2) == b"\r\n"
+                assert request.rfile.read(2) == b"\r\n"
+            else:
+                body = request.rfile.read(int(request.headers["Content-Length"]))
+            received.append(pa.ipc.open_stream(body).read_all())
+            response = {
+                "version": 2,
+                "num_updated_rows": 0,
+                "num_inserted_rows": 2,
+                "num_deleted_rows": 0,
+            }
+        request.send_response(200)
+        request.send_header("Content-Type", "application/json")
+        request.send_header("phalanx-version", "0.5.0")
+        request.end_headers()
+        request.wfile.write(json.dumps(response).encode())
+
+    field = lancedb.blob("image")
+    storage_type = field.type.storage_type
+    if target_with_ranges:
+        storage_type = pa.struct(list(storage_type)[:2])
+        field = pa.field(
+            "image",
+            storage_type,
+            metadata={"ARROW:extension:name": "lance.blob.v2"},
+        )
+    images = pa.array([{"data": b"payload"}, None], type=storage_type)
+    if isinstance(field.type, pa.ExtensionType):
+        images = pa.ExtensionArray.from_storage(field.type, images)
+    data = pa.Table.from_arrays(
+        [pa.array([1, 2], type=pa.int64()), images],
+        schema=pa.schema([pa.field("id", pa.int64()), field]),
+    )
+
+    with mock_lancedb_connection(handler) as db:
+        remote_table = db.open_table("test")
+        table = remote_table._table if use_async else remote_table
+        if mode == "merge":
+            result = (
+                table.merge_insert("id").when_not_matched_insert_all().execute(data)
+            )
+        else:
+            result = table.add(data, mode="overwrite")
+        if use_async:
+            await result
+
+    assert len(received) == 1
+    images = received[0]["image"].combine_chunks()
+    if isinstance(images.type, pa.ExtensionType):
+        images = images.storage
+    with_ranges = target_with_ranges if mode == "merge" else not target_with_ranges
+    assert images.type.num_fields == (4 if with_ranges else 2)
+    expected = {"data": b"payload", "uri": None}
+    if with_ranges:
+        expected.update(position=None, size=None)
+    assert images.to_pylist() == [expected, None]
+
+
 def blob_query_response_table():
     image_field = pa.field(
         "image",
@@ -3362,13 +3444,20 @@ def blob_remote_table(*, server_version=Version("0.5.0")):
             else:
                 assert body["with_row_id"] is True
                 if columns == ["image"]:
-                    # fetch_blob_files sizes its handles from a descriptor take.
+                    # Blob handles are sized from a descriptor take.
                     assert body["filter"].startswith("_rowid IN")
                     response_table = blob_descriptor_take_table()
+                    row_ids = [
+                        int(value) for value in re.findall(r"\d+", body["filter"])
+                    ]
+                    response_table = response_table.filter(
+                        pa.compute.is_in(response_table["_rowid"], pa.array(row_ids))
+                    )
                 else:
                     response_table = blob_query_response_table()
             request.send_response(200)
             request.send_header("Content-Type", "application/vnd.apache.arrow.file")
+            request.send_header("x-lancedb-version", "1")
             request.end_headers()
             with pa.ipc.new_file(request.wfile, response_table.schema) as writer:
                 writer.write_table(response_table)
@@ -3400,6 +3489,42 @@ def test_remote_blob_columns_and_fetch():
         assert blobs.to_pylist() == [b"alpha", None, b"gamma"]
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("use_async", [False, True])
+async def test_remote_fetch_blob_ranges(use_async):
+    with blob_remote_table() as remote_table:
+        table = remote_table._table if use_async else remote_table
+        requests = [
+            (30, 2, 3),
+            (10, 1, 3),
+            (20, 0, 1),
+            (10, 0, 2),
+            (40, 0, 0),
+            (10, 5, 0),
+            (30, 2, 3),
+        ]
+        result = table.fetch_blob_ranges("image", requests)
+        if use_async:
+            result = await result
+        assert isinstance(result, pa.LargeBinaryArray)
+        assert result.to_pylist() == [b"mma", b"lph", None, b"al", b"", b"", b"mma"]
+
+        for invalid, message in [
+            ([(10, 0, 1), (99, 0, 1)], "only 1 exist"),
+            ([(10, 4, 2)], "exceeds blob size"),
+            ([(20, 2**64 - 1, 1)], "overflows u64"),
+        ]:
+            with pytest.raises(ValueError, match=message):
+                result = table.fetch_blob_ranges("image", invalid)
+                if use_async:
+                    await result
+
+        result = table.fetch_blob_ranges("image", [])
+        if use_async:
+            result = await result
+        assert result.to_pylist() == []
+
+
 @pytest.mark.asyncio
 async def test_async_remote_to_lance_rejects_storage_access(monkeypatch):
     with blob_remote_table() as remote_table: