Merge origin/main into gatekeeper/fix-2325-1

# Conflicts:
#	python/python/lancedb/__init__.py

python/Cargo.toml

@@ -15,10 +15,10 @@ name = "_lancedb"
 crate-type = ["cdylib"]
 
 [dependencies]
-arrow = { version = "58.0.0", features = ["pyarrow"] }
-async-trait = "0.1"
-bytes = "1"
-lancedb = { path = "../rust/lancedb", default-features = false }
+arrow = { workspace = true, features = ["pyarrow"] }
+async-trait.workspace = true
+bytes.workspace = true
+lancedb.workspace = true
 datafusion-common.workspace = true
 lance-core.workspace = true
 lance-namespace.workspace = true
@@ -27,17 +27,17 @@ lance-io.workspace = true
 env_logger.workspace = true
 log.workspace = true
 pyo3 = { version = "0.28", features = ["extension-module", "abi3-py310", "chrono"] }
-chrono = { version = "0.4", default-features = false, features = ["clock"] }
+chrono.workspace = true
 pyo3-async-runtimes = { version = "0.28", features = [
     "attributes",
     "tokio-runtime",
 ] }
-pin-project = "1.1.5"
+pin-project.workspace = true
 futures.workspace = true
-serde = "1"
-serde_json = "1"
+serde.workspace = true
+serde_json.workspace = true
 snafu.workspace = true
-tokio = { version = "1.40", features = ["sync", "rt-multi-thread"] }
+tokio.workspace = true
 libc = "0.2"
 
 [build-dependencies]

python/python/lancedb/__init__.py

@@ -22,6 +22,12 @@ from .remote.db import RemoteDBConnection
 from .expr import Expr, col, lit, func
 from .schema import blob, vector, BlobType
 from .job import AsyncJob, Job
+from .functions import (
+    FunctionApplication as FunctionApplication,
+    FunctionBinding as FunctionBinding,
+    FunctionVersion as FunctionVersion,
+    PythonRuntimeSpec as PythonRuntimeSpec,
+)
 from .table import AsyncTable, CompactionOptions, Table
 from .types import BaseTokenizerType
 from ._lancedb import Session

python/python/lancedb/functions.py

@@ -0,0 +1,379 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""Canonical values exchanged with LanceDB Enterprise Function services.
+
+These immutable models contain client/wire state only. Catalog persistence,
+environment bake, secret resolution, and execution are owned by Sophon.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping
+from typing import Any, Optional
+
+import pydantic
+from pydantic import BaseModel, Field, conint
+
+_PYDANTIC_V2 = int(pydantic.VERSION.split(".", 1)[0]) >= 2
+if _PYDANTIC_V2:
+    from pydantic import field_validator, model_validator
+else:
+    from pydantic import root_validator, validator
+
+_Int32 = conint(strict=True, ge=-(2**31), le=2**31 - 1)
+_UInt32 = conint(strict=True, ge=0, le=2**32 - 1)
+_UInt64 = conint(strict=True, ge=0, le=2**64 - 1)
+
+
+class _FrozenDict(dict):
+    def _immutable(self, *args, **kwargs):
+        raise TypeError("remote canonical values are immutable")
+
+    __setitem__ = _immutable
+    __delitem__ = _immutable
+    clear = _immutable
+    pop = _immutable
+    popitem = _immutable
+    setdefault = _immutable
+    update = _immutable
+
+    def __ior__(self, other):
+        self._immutable()
+
+
+def _freeze_value(value):
+    if isinstance(value, Mapping):
+        return _FrozenDict({key: _freeze_value(child) for key, child in value.items()})
+    if isinstance(value, (list, tuple)):
+        return tuple(_freeze_value(child) for child in value)
+    return value
+
+
+def _validate_literal(value):
+    if isinstance(value, float):
+        raise ValueError(
+            "floating-point Function literals are not part of the Slice 1 "
+            "canonical wire contract"
+        )
+    if isinstance(value, int) and not isinstance(value, bool):
+        if not -(2**63) <= value <= 2**64 - 1:
+            raise ValueError(
+                "Function integer literal is outside the canonical JSON range"
+            )
+    elif isinstance(value, Mapping):
+        for child in value.values():
+            _validate_literal(child)
+    elif isinstance(value, (list, tuple)):
+        for child in value:
+            _validate_literal(child)
+    return value
+
+
+def _known_wire_value(value):
+    if isinstance(value, _RemoteValue):
+        return value._known_dict()
+    if isinstance(value, Mapping):
+        return {key: _known_wire_value(child) for key, child in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_known_wire_value(child) for child in value]
+    return value
+
+
+class _RemoteValue(BaseModel):
+    if _PYDANTIC_V2:
+        model_config = {"extra": "ignore", "frozen": True}
+    else:
+
+        class Config:
+            allow_mutation = False
+            extra = "ignore"
+
+    if _PYDANTIC_V2:
+
+        @model_validator(mode="after")
+        def _freeze_mappings(self):
+            for name, value in self.__dict__.items():
+                object.__setattr__(self, name, _freeze_value(value))
+            return self
+
+    else:
+
+        @root_validator
+        def _freeze_mappings(cls, values):
+            return {name: _freeze_value(value) for name, value in values.items()}
+
+    @classmethod
+    def from_json(cls, payload: str):
+        if _PYDANTIC_V2:
+            return cls.model_validate_json(payload)
+        return cls.parse_raw(payload)
+
+    def _known_dict(self) -> dict[str, Any]:
+        fields = self.__class__.model_fields if _PYDANTIC_V2 else self.__fields__
+        known = {}
+        for name, field in fields.items():
+            value = getattr(self, name)
+            if value is None:
+                continue
+            required = field.is_required() if _PYDANTIC_V2 else field.required
+            if not required:
+                default_factory = field.default_factory
+                if default_factory is not None and value == default_factory():
+                    continue
+                if default_factory is None and value == field.default:
+                    continue
+            known[name] = _known_wire_value(value)
+        return known
+
+    def _copy(self, *, update: Mapping[str, Any]):
+        update = {name: _freeze_value(value) for name, value in update.items()}
+        if _PYDANTIC_V2:
+            return self.model_copy(update=update)
+        return self.copy(update=update)
+
+    def to_canonical_json(self) -> str:
+        return json.dumps(
+            self._known_dict(),
+            ensure_ascii=False,
+            allow_nan=False,
+            sort_keys=True,
+            separators=(",", ":"),
+        )
+
+
+class FunctionArtifact(_RemoteValue):
+    """Content-addressed Python artifact identity."""
+
+    kind: str
+    digest: str
+    entrypoint: str
+
+
+class FunctionParameter(_RemoteValue):
+    name: str
+    arrow_type: str
+    nullable: bool
+
+
+class FunctionResultField(_RemoteValue):
+    name: str
+    arrow_type: str
+    nullable: bool
+
+
+class FunctionOutput(_RemoteValue):
+    """Scalar or ordered named-struct output; unknown kinds remain decodable."""
+
+    kind: str
+    arrow_type: Optional[str] = None
+    nullable: Optional[bool] = None
+    fields: tuple[FunctionResultField, ...] = ()
+
+
+class FunctionSignature(_RemoteValue):
+    inputs: tuple[FunctionParameter, ...]
+    output: FunctionOutput
+
+
+class PythonEnvironmentSpec(_RemoteValue):
+    """One Sophon-managed Python environment source."""
+
+    kind: str
+    packages: tuple[str, ...] = ()
+    path: Optional[str] = None
+    modules: tuple[str, ...] = ()
+    image: Optional[str] = None
+
+
+class PythonRuntimeSpec(_RemoteValue):
+    """Remote runtime definition with non-secret environment values.
+
+    V1 supports ``kind="python"``. Newer runtime kinds remain readable, while
+    their unknown payload fields are intentionally not retained by the client.
+    """
+
+    kind: str
+    python_version: Optional[str] = None
+    environment: Optional[PythonEnvironmentSpec] = None
+    env: Optional[Mapping[str, str]] = None
+
+    if _PYDANTIC_V2:
+
+        @model_validator(mode="after")
+        def _validate_runtime_kind(self):
+            if self.kind == "python":
+                if self.python_version is None:
+                    raise ValueError("python runtime requires python_version")
+                if self.environment is None:
+                    raise ValueError("python runtime requires environment")
+            else:
+                object.__setattr__(self, "python_version", None)
+                object.__setattr__(self, "environment", None)
+                object.__setattr__(self, "env", None)
+            return self
+
+    else:
+
+        @root_validator
+        def _validate_runtime_kind(cls, values):
+            if values.get("kind") == "python":
+                if values.get("python_version") is None:
+                    raise ValueError("python runtime requires python_version")
+                if values.get("environment") is None:
+                    raise ValueError("python runtime requires environment")
+            else:
+                values["python_version"] = None
+                values["environment"] = None
+                values["env"] = None
+            return values
+
+
+class FunctionVersion(_RemoteValue):
+    """An exact immutable Function version returned by Enterprise.
+
+    Scheduling resources, priority, concurrency, and retry policy belong to
+    the submitting Job and are not part of this identity.
+    """
+
+    name: str
+    version: str
+    artifact: FunctionArtifact
+    signature: FunctionSignature
+    runtime: PythonRuntimeSpec
+    runtime_digest: str
+    environment_digest: str
+    required_secrets: tuple[str, ...] = ()
+    created_at: str
+
+
+class FunctionVersionRef(_RemoteValue):
+    name: str
+    version: str
+
+
+class ApplicationInput(_RemoteValue):
+    """One parameter value.
+
+    Slice 1 freezes integers, strings, booleans, nulls, arrays, and objects.
+    Floating-point literal encoding is deferred until Python authoring is
+    introduced with a language-neutral numeric representation.
+    """
+
+    parameter: str
+    kind: str
+    value: Any
+
+    if _PYDANTIC_V2:
+
+        @field_validator("value")
+        @classmethod
+        def _validate_value(cls, value):
+            return _validate_literal(value)
+
+    else:
+
+        @validator("value")
+        def _validate_value(cls, value):
+            return _validate_literal(value)
+
+
+class FunctionApplication(_RemoteValue):
+    """Immutable pre-declaration application of an exact Function version."""
+
+    function: FunctionVersionRef
+    inputs: tuple[ApplicationInput, ...]
+    output: FunctionOutput
+    group_id: str
+    columns: Mapping[str, str] = Field(default_factory=dict)
+
+    def rename(self, *, columns: Mapping[str, str]) -> FunctionApplication:
+        """Return a copy with result-field to table-column aliases."""
+        if self.output.kind != "named_struct":
+            raise ValueError("rename(columns=...) requires a named-struct application")
+        result_fields = {field.name for field in self.output.fields}
+        unknown = set(columns) - result_fields
+        if unknown:
+            raise ValueError(f"unknown Function result fields: {sorted(unknown)!r}")
+        merged = dict(self.columns)
+        merged.update(columns)
+        destinations = tuple(
+            merged.get(field.name, field.name) for field in self.output.fields
+        )
+        if len(set(destinations)) != len(destinations):
+            raise ValueError("FunctionApplication rename destinations must be unique")
+        return self._copy(update={"columns": merged})
+
+
+class InputBinding(_RemoteValue):
+    parameter: str
+    field_id: _Int32
+    field_path: str
+    arrow_type: str
+    nullable: bool
+
+
+class OutputMapping(_RemoteValue):
+    """One stable result-field mapping.
+
+    Assignment state is outside the Slice 1 client contract. During the NULL
+    transition Lance exposes no public cell-flag identifier to persist here.
+    """
+
+    result_field: str
+    output_name: str
+    output_field_id: _Int32
+    output_ordinal: _UInt32
+    arrow_type: str
+    nullable: bool
+
+
+class FunctionBinding(_RemoteValue):
+    """Immutable grouped binding persisted by the Enterprise table service."""
+
+    binding_id: str
+    revision: _UInt64
+    function: FunctionVersionRef
+    group_id: str
+    inputs: tuple[InputBinding, ...]
+    outputs: tuple[OutputMapping, ...]
+
+
+class RefreshColumnResult(_RemoteValue):
+    """Terminal result of a remote Function-column refresh Job."""
+
+    rows_assigned: _UInt64
+    rows_failed: _UInt64
+    rows_remaining: _UInt64
+    source_version: _UInt64
+    published_version: Optional[_UInt64] = None
+
+    @property
+    def rows_filled(self) -> int:
+        """Deprecated compatibility alias for :attr:`rows_assigned`."""
+        return self.rows_assigned
+
+    @property
+    def version(self) -> Optional[int]:
+        """Deprecated compatibility alias for :attr:`published_version`."""
+        return self.published_version
+
+
+__all__ = [
+    "ApplicationInput",
+    "FunctionApplication",
+    "FunctionArtifact",
+    "FunctionBinding",
+    "FunctionOutput",
+    "FunctionParameter",
+    "FunctionResultField",
+    "FunctionSignature",
+    "FunctionVersion",
+    "FunctionVersionRef",
+    "InputBinding",
+    "OutputMapping",
+    "PythonEnvironmentSpec",
+    "PythonRuntimeSpec",
+    "RefreshColumnResult",
+]

python/python/lancedb/table.py

@@ -516,6 +516,20 @@ def _cast_to_target_schema(
     return pa.RecordBatchReader.from_batches(reordered_schema, gen())
 
 
+def _field_extension_name(field: pa.Field) -> Optional[str]:
+    extension_name = getattr(field.type, "extension_name", None)
+    if extension_name is not None:
+        return extension_name
+
+    metadata = field.metadata or {}
+    extension_name = metadata.get(b"ARROW:extension:name") or metadata.get(
+        "ARROW:extension:name"
+    )
+    if isinstance(extension_name, bytes):
+        return extension_name.decode()
+    return extension_name
+
+
 def _align_field_types(
     fields: List[pa.Field],
     target_fields: List[pa.Field],
@@ -528,6 +542,16 @@ def _align_field_types(
         target_field = next((f for f in target_fields if f.name == field.name), None)
         if target_field is None:
             raise ValueError(f"Field '{field.name}' not found in target schema")
+        # Preserve arrow.json input until it reaches Lance. 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.
+        if (
+            _field_extension_name(field) == "arrow.json"
+            and _field_extension_name(target_field) == "lance.json"
+        ):
+            new_fields.append(field)
+            continue
         if pa.types.is_struct(target_field.type):
             if pa.types.is_struct(field.type):
                 new_type = pa.struct(

python/python/tests/test_first_class_function_slice1.py

@@ -0,0 +1,225 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+import json
+from pathlib import Path
+
+import pytest
+
+import lancedb.functions as functions
+from lancedb.functions import (
+    FunctionApplication,
+    FunctionBinding,
+    FunctionVersion,
+    PythonRuntimeSpec,
+    RefreshColumnResult,
+)
+
+
+FIXTURES = (
+    Path(__file__).parents[3]
+    / "rust"
+    / "lancedb"
+    / "tests"
+    / "fixtures"
+    / "first_class_functions"
+    / "v1"
+)
+
+
+def fixture(name: str) -> str:
+    return (FIXTURES / name).read_text()
+
+
+def job_result(name: str) -> dict:
+    return json.loads(fixture(name))["result"]
+
+
+def assert_no_secret_values(value):
+    if isinstance(value, dict):
+        for key, child in value.items():
+            assert key not in {
+                "secret_value",
+                "secret_values",
+                "resolved_secret",
+                "resolved_secrets",
+            }
+            assert_no_secret_values(child)
+    elif isinstance(value, list):
+        for child in value:
+            assert_no_secret_values(child)
+
+
+def test_public_function_values_are_in_api_reference():
+    docs = Path(__file__).parents[3] / "docs" / "src" / "python" / "python.md"
+    rendered = docs.read_text()
+    for name in functions.__all__:
+        assert f"::: lancedb.functions.{name}" in rendered
+
+
+@pytest.mark.parametrize(
+    ("fixture_name", "canonical_name", "model", "nested_result"),
+    [
+        (
+            "remote_function_job.json",
+            "remote_function_version.canonical.json",
+            FunctionVersion,
+            True,
+        ),
+        (
+            "remote_function_application.json",
+            "remote_function_application.canonical.json",
+            FunctionApplication,
+            False,
+        ),
+        (
+            "remote_function_binding.json",
+            "remote_function_binding.canonical.json",
+            FunctionBinding,
+            False,
+        ),
+        (
+            "remote_refresh_job.json",
+            "remote_refresh_result.canonical.json",
+            RefreshColumnResult,
+            True,
+        ),
+        (
+            "remote_refresh_result_without_published_version.json",
+            "remote_refresh_result_without_published_version.canonical.json",
+            RefreshColumnResult,
+            False,
+        ),
+    ],
+)
+def test_python_and_rust_share_remote_canonical_goldens(
+    fixture_name, canonical_name, model, nested_result
+):
+    value = json.loads(fixture(fixture_name))
+    if nested_result:
+        value = value["result"]
+    decoded = model.from_json(json.dumps(value))
+    assert decoded.to_canonical_json() == fixture(canonical_name).strip()
+
+
+def test_function_version_identity_is_immutable_and_exact():
+    value = job_result("remote_function_job.json")
+    version = FunctionVersion.from_json(json.dumps(value))
+    assert version.name == "embed"
+    assert version.version == "fv_01K3EXACT"
+    assert version.required_secrets == ("HF_TOKEN",)
+
+    with pytest.raises((TypeError, ValueError)):
+        version.version = "fv_changed"
+    with pytest.raises(TypeError, match="immutable"):
+        version.runtime.env["TOKENIZERS_PARALLELISM"] = "true"
+
+    changed = dict(value)
+    changed["version"] = "fv_changed"
+    assert FunctionVersion(**changed) != version
+
+
+def test_unknown_fields_and_discriminators_are_forward_decodable():
+    value = job_result("remote_function_job.json")
+    value["future_version_metadata"] = {"retention_class": "catalog"}
+    value["runtime"] = {"kind": "wasm", "module_digest": "sha256:wasm"}
+    value["signature"]["output"]["kind"] = "future_output_shape"
+
+    version = FunctionVersion.from_json(json.dumps(value))
+    assert version.runtime.kind == "wasm"
+    assert version.runtime.python_version is None
+    assert version.runtime.environment is None
+    assert json.loads(version.to_canonical_json())["runtime"] == {"kind": "wasm"}
+    assert version.signature.output.kind == "future_output_shape"
+
+
+def test_function_application_uses_rename_columns_only():
+    application = FunctionApplication.from_json(
+        fixture("remote_function_application.json")
+    )
+    renamed = application.rename(columns={"normalized_text": "body_normalized"})
+
+    assert application.columns["normalized_text"] == "search_text"
+    assert renamed.columns["normalized_text"] == "body_normalized"
+    assert renamed.function == application.function
+    assert renamed.group_id == application.group_id
+    assert not hasattr(application, "rename_outputs")
+    with pytest.raises(TypeError, match="immutable"):
+        renamed.columns["normalized_text"] = "changed"
+    with pytest.raises(TypeError, match="immutable"):
+        application.inputs[0].value["path"] = "changed"
+
+    with pytest.raises(ValueError, match="unknown Function result fields"):
+        application.rename(columns={"missing": "search_text"})
+    with pytest.raises(ValueError, match="destinations must be unique"):
+        application.rename(columns={"normalized_text": "same", "token_count": "same"})
+
+    bare_value = json.loads(fixture("remote_function_application.json"))
+    bare_value.pop("columns")
+    bare = FunctionApplication(**bare_value)
+    with pytest.raises(ValueError, match="destinations must be unique"):
+        bare.rename(columns={"normalized_text": "token_count"})
+
+
+def test_binding_and_refresh_result_keep_stable_remote_fields():
+    binding = FunctionBinding.from_json(fixture("remote_function_binding.json"))
+    assert binding.revision == 3
+    assert binding.function.version == "fv_01K3TEXT"
+    assert [output.output_ordinal for output in binding.outputs] == [0, 1]
+
+    result = RefreshColumnResult.from_json(
+        json.dumps(job_result("remote_refresh_job.json"))
+    )
+    assert result.rows_filled == result.rows_assigned
+    assert result.version == result.published_version
+
+    result = RefreshColumnResult.from_json(
+        fixture("remote_refresh_result_without_published_version.json")
+    )
+    assert result.published_version is None
+    assert RefreshColumnResult.from_json(result.to_canonical_json()) == result
+
+
+def test_function_literal_numeric_domain_matches_rust():
+    with pytest.raises(ValueError, match="floating-point Function literals"):
+        FunctionApplication.from_json(fixture("remote_function_application_float.json"))
+
+    value = json.loads(fixture("remote_function_application_float.json"))
+    value["inputs"][0]["value"] = 2**64
+    with pytest.raises(ValueError, match="outside the canonical JSON range"):
+        FunctionApplication.from_json(json.dumps(value))
+
+
+def test_empty_default_maps_have_stable_canonical_bytes():
+    runtime = PythonRuntimeSpec(
+        kind="python", python_version="3.12", environment={"kind": "pip"}
+    )
+    assert runtime.to_canonical_json() == (
+        '{"environment":{"kind":"pip"},"kind":"python","python_version":"3.12"}'
+    )
+
+    value = json.loads(fixture("remote_function_application.json"))
+    value.pop("columns")
+    application = FunctionApplication.from_json(json.dumps(value))
+    assert "columns" not in json.loads(application.to_canonical_json())
+
+
+@pytest.mark.parametrize("field", ["rows_assigned", "source_version"])
+def test_refresh_result_rejects_non_u64_values(field):
+    value = job_result("remote_refresh_job.json")
+    value[field] = -1
+    with pytest.raises(ValueError):
+        RefreshColumnResult.from_json(json.dumps(value))
+
+    value[field] = "1"
+    with pytest.raises(ValueError):
+        RefreshColumnResult.from_json(json.dumps(value))
+
+
+def test_canonical_client_values_contain_secret_names_only():
+    version = FunctionVersion.from_json(
+        json.dumps(job_result("remote_function_job.json"))
+    )
+    canonical = json.loads(version.to_canonical_json())
+    assert canonical["required_secrets"] == ["HF_TOKEN"]
+    assert_no_secret_values(canonical)

python/python/tests/test_table.py

@@ -2772,6 +2772,56 @@ async def test_merge_insert_async(mem_db_async: AsyncConnection):
     assert (await table.to_arrow()).sort_by("a") == expected
 
 
+@pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
+@pytest.mark.asyncio
+async def test_merge_insert_encodes_json(mem_db_async: AsyncConnection):
+    json_type = pa.json_()
+    schema = pa.schema([pa.field("id", pa.string()), pa.field("j", json_type)])
+
+    def json_table(rows):
+        json_values = pa.ExtensionArray.from_storage(
+            json_type,
+            pa.array([value for _, value in rows], type=json_type.storage_type),
+        )
+        return pa.Table.from_arrays(
+            [pa.array([row_id for row_id, _ in rows]), json_values], schema=schema
+        )
+
+    table = await mem_db_async.create_table("json_merge", schema=schema)
+    await table.add(json_table([("a", '{"k": 1}'), ("b", '{"k": 9}')]))
+
+    await (
+        table.merge_insert("id")
+        .when_matched_update_all()
+        .execute(json_table([("a", '{"k": 2}')]))
+    )
+
+    rows = sorted(await table.query().to_list(), key=lambda row: row["id"])
+    assert rows == [
+        {"id": "a", "j": '{"k":2}'},
+        {"id": "b", "j": '{"k":9}'},
+    ]
+    filtered = await table.query().where("json_extract(j, '$.k') = '2'").to_list()
+    assert filtered == [{"id": "a", "j": '{"k":2}'}]
+
+
+@pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
+@pytest.mark.asyncio
+async def test_add_sanitization_encodes_json(mem_db_async: AsyncConnection):
+    json_type = pa.json_()
+    schema = pa.schema([pa.field("id", pa.string()), pa.field("j", json_type)])
+    json_values = pa.ExtensionArray.from_storage(
+        json_type, pa.array(['{"k": 3}'], type=json_type.storage_type)
+    )
+    data = pa.Table.from_arrays([pa.array(["c"]), json_values], schema=schema)
+
+    table = await mem_db_async.create_table("json_add", schema=schema)
+    await table.add(data, on_bad_vectors="fill")
+
+    rows = await table.query().where("json_extract(j, '$.k') = '3'").to_list()
+    assert rows == [{"id": "c", "j": '{"k":3}'}]
+
+
 def test_create_with_embedding_function(mem_db: DBConnection):
     class MyTable(LanceModel):
         text: str