# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright The LanceDB Authors

from __future__ import annotations

import asyncio
import inspect
import deprecation
import warnings
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime, timedelta
from functools import cached_property
from typing import (
    TYPE_CHECKING,
    Any,
    Callable,
    Dict,
    Iterable,
    List,
    Literal,
    Optional,
    Sequence,
    Tuple,
    Union,
    overload,
)
from urllib.parse import urlparse

from lancedb.scannable import _register_optional_converters, to_scannable

from . import __version__
from ._blob import (
    BlobFile,
    _normalize_blob_row_ids,
    _wrap_blob_files,
    strip_auto_row_ids,
    validate_blob_mode,
)
from .types import BlobMode
from lancedb.arrow import peek_reader
from lancedb.background_loop import LOOP, embedding_executor
from lancedb.job import AsyncJob, Job, _typed_job
from .dependencies import (
    _check_for_hugging_face,
    _check_for_lance,
    _check_for_pandas,
    lance,
    pandas as pd,
    polars as pl,
)
import pyarrow as pa
import pyarrow.dataset
import pyarrow.compute as pc
import pyarrow.fs as pa_fs
import numpy as np

from .common import DATA, VEC, VECTOR_COLUMN_NAME
from .embeddings import EmbeddingFunctionConfig, EmbeddingFunctionRegistry
from .index import (
    BTree,
    IvfFlat,
    IvfPq,
    IvfSq,
    Bitmap,
    IvfRq,
    LabelList,
    Fm,
    HnswPq,
    HnswSq,
    HnswFlat,
    FTS,
)
from .expr import Expr
from .functions import (
    FunctionApplication,
    RefreshColumnResult as RefreshColumnJobResult,
)
from .merge import LanceMergeInsertBuilder
from .pydantic import LanceModel, model_to_dict
from .query import (
    AnalyzePlanDistributedMetrics,
    AsyncFTSQuery,
    AsyncHybridQuery,
    AsyncQuery,
    AsyncTakeQuery,
    AsyncVectorQuery,
    DocumentGranularity,
    FullTextQuery,
    LanceEmptyQueryBuilder,
    LanceFtsQueryBuilder,
    LanceHybridQueryBuilder,
    LanceQueryBuilder,
    LanceVectorQueryBuilder,
    LanceTakeQueryBuilder,
    Query,
)
from .util import (
    add_note,
    fs_from_uri,
    get_uri_scheme,
    infer_vector_column_name,
    join_uri,
    value_to_sql,
)
from .index import lang_mapping
from .schema import (
    blob_v2_column_paths,
    is_blob_v2_field,
    row_addressable_blob_v2_paths,
    schema_has_blob_field,
)


def _should_push_down_query_table(
    namespace_client: Optional[Any], pushdown_operations: set
) -> bool:
    return namespace_client is not None and "QueryTable" in pushdown_operations


def _polars_predicate_pushdown_barrier(frame: Any) -> Any:
    """Return a Polars frame unchanged while blocking predicate pushdown."""
    return frame


_MODEL_BACKED_TOKENIZER_PREFIXES = ("jieba", "lindera")
_MODEL_BACKED_TOKENIZER_ERRORS = (
    "unknown base tokenizer",
    "Invalid directory path:",
    "Failed to load Jieba",
    "Failed to load tokenizer config",
    "Failed to initialize default tokenizer",
)


def _add_unique_note(exception: BaseException, note: str) -> None:
    existing_notes = getattr(exception, "__notes__", ()) or ()
    message = (
        exception.args[0]
        if exception.args and isinstance(exception.args[0], str)
        else ""
    )
    if note not in existing_notes and note not in message:
        add_note(exception, note)


def _is_model_backed_tokenizer(base_tokenizer: str) -> bool:
    return any(
        base_tokenizer == prefix or base_tokenizer.startswith(f"{prefix}/")
        for prefix in _MODEL_BACKED_TOKENIZER_PREFIXES
    )


def _maybe_add_fts_error_note(
    exception: BaseException, *, base_tokenizer: str, language: Optional[str] = None
) -> None:
    message = str(exception)
    if language is not None and "not support the requested language" in message:
        supported_langs = ", ".join(lang_mapping.values())
        _add_unique_note(exception, f"Supported languages: {supported_langs}")
        return

    if not _is_model_backed_tokenizer(base_tokenizer):
        return

    if not any(marker in message for marker in _MODEL_BACKED_TOKENIZER_ERRORS):
        return

    _add_unique_note(
        exception,
        "Model-backed tokenizers such as 'jieba/default' and 'lindera/ipadic' "
        "require tokenizer models in Lance's language model home. Set "
        "LANCE_LANGUAGE_MODEL_HOME to override the default platform data "
        "directory under 'lance/language_models'. Expected layouts include "
        "'<model-home>/jieba/default/...' and "
        "'<model-home>/lindera/ipadic/...'.",
    )


if TYPE_CHECKING:
    from .db import LanceDBConnection
    from ._lancedb import (
        Table as LanceDBTable,
        OptimizeStats,
        CleanupStats,
        CompactionStats,
        Tag,
        AddColumnsResult,
        RefreshColumnResult,
        AddResult,
        AlterColumnsResult,
        UpdateFieldMetadataResult,
        DeleteResult,
        DropColumnsResult,
        FtsToken,
        LsmWriteSpec,
        MergeResult,
        UpdateResult,
    )
    from .index import IndexConfig
    import pandas
    import PIL
    from .types import (
        QueryType,
        OnBadVectorsType,
        AddMode,
        CreateMode,
        VectorIndexType,
        ScalarIndexType,
        BaseTokenizerType,
        DistanceType,
    )

# Type alias for index configuration objects
IndexConfigType = Union[
    IvfFlat,
    IvfPq,
    IvfSq,
    IvfRq,
    HnswFlat,
    HnswPq,
    HnswSq,
    BTree,
    Bitmap,
    LabelList,
    Fm,
    FTS,
]

# Known distance metrics for legacy API detection
KNOWN_METRICS = {"l2", "cosine", "dot", "hamming"}


def _into_pyarrow_reader(
    data, schema: Optional[pa.Schema] = None
) -> pa.RecordBatchReader:
    from lancedb.dependencies import datasets

    if _check_for_hugging_face(data):
        if isinstance(data, datasets.Dataset):
            schema = data.features.arrow_schema
            return pa.RecordBatchReader.from_batches(schema, data.data.to_batches())
        elif isinstance(data, datasets.dataset_dict.DatasetDict):
            schema = _schema_from_hf(data, None)
            if "split" not in schema.names:
                schema = schema.append(pa.field("split", pa.string()))
            return pa.RecordBatchReader.from_batches(
                schema, _to_batches_with_split(data)
            )
    if isinstance(data, LanceModel):
        raise ValueError("Cannot add a single LanceModel to a table. Use a list.")

    if isinstance(data, dict):
        raise ValueError(
            "Cannot create or add rows from a single dictionary. "
            "Use a list of dictionaries instead."
        )

    if isinstance(data, list):
        # Handle empty list case
        if not data:
            if schema is None:
                raise ValueError("Cannot create table from empty list without a schema")
            return pa.Table.from_pylist(data, schema=schema).to_reader()

        # convert to list of dict if data is a bunch of LanceModels
        if isinstance(data[0], LanceModel):
            schema = data[0].__class__.to_arrow_schema()
            data = [model_to_dict(d) for d in data]
            return pa.Table.from_pylist(data, schema=schema).to_reader()
        elif isinstance(data[0], pa.RecordBatch):
            return pa.Table.from_batches(data).to_reader()
        else:
            return pa.Table.from_pylist(data).to_reader()
    elif _check_for_pandas(data) and isinstance(data, pd.DataFrame):
        table = pa.Table.from_pandas(data, preserve_index=False)
        # Do not serialize Pandas metadata
        meta = table.schema.metadata if table.schema.metadata is not None else {}
        meta = {k: v for k, v in meta.items() if k != b"pandas"}
        return table.replace_schema_metadata(meta).to_reader()
    elif isinstance(data, pa.Table):
        return data.to_reader()
    elif isinstance(data, pa.RecordBatch):
        return pa.RecordBatchReader.from_batches(data.schema, [data])
    elif _check_for_lance(data) and isinstance(data, lance.LanceDataset):
        return data.scanner().to_reader()
    elif isinstance(data, pa.dataset.Dataset):
        return data.scanner().to_reader()
    elif isinstance(data, pa.dataset.Scanner):
        return data.to_reader()
    elif isinstance(data, pa.RecordBatchReader):
        return data
    elif (
        type(data).__module__.startswith("polars")
        and data.__class__.__name__ == "DataFrame"
    ):
        return data.to_arrow().to_reader()
    elif (
        type(data).__module__.startswith("polars")
        and data.__class__.__name__ == "LazyFrame"
    ):
        return data.collect().to_arrow().to_reader()
    elif isinstance(data, Iterable):
        return _iterator_to_reader(data)
    else:
        raise TypeError(
            f"Unknown data type {type(data)}. "
            "Supported types: list of dicts, pandas DataFrame, polars DataFrame, "
            "pyarrow Table/RecordBatch, or Pydantic models. "
            "See https://docs.lancedb.com/tables/ for examples."
        )


def _iterator_to_reader(data: Iterable) -> pa.RecordBatchReader:
    # Each batch is treated as it's own reader, mainly so we can
    # re-use the _into_pyarrow_reader logic.
    first = _into_pyarrow_reader(next(data))
    schema = first.schema

    def gen():
        yield from first
        for batch in data:
            table: pa.Table = _into_pyarrow_reader(batch).read_all()
            if table.schema != schema:
                try:
                    table = table.cast(schema)
                except pa.lib.ArrowInvalid:
                    raise ValueError(
                        f"Input iterator yielded a batch with schema that "
                        f"does not match the schema of other batches.\n"
                        f"Expected:\n{schema}\nGot:\n{batch.schema}"
                    )
            yield from table.to_batches()

    return pa.RecordBatchReader.from_batches(schema, gen())


def _sanitize_data(
    data: "DATA",
    target_schema: Optional[pa.Schema] = None,
    metadata: Optional[dict] = None,  # embedding metadata
    on_bad_vectors: OnBadVectorsType = "error",
    fill_value: float = 0.0,
    *,
    allow_subschema: bool = False,
) -> pa.RecordBatchReader:
    """
    Handle input data, applying all standard transformations.

    This includes:

     * Converting the data to a PyArrow Table
     * Adding vector columns defined in the metadata
     * Adding embedding metadata into the schema
     * Casting the table to the target schema
     * Handling bad vectors

    Parameters
    ----------
    target_schema : Optional[pa.Schema], default None
        The schema to cast the table to. This is typically the schema of the table
        if it already exists. Otherwise it might be a user-requested schema.
    allow_subschema : bool, default False
        If True, the input table is allowed to omit columns from the target schema.
        The target schema will be filtered to only include columns that are present
        in the input table before casting.
    metadata : Optional[dict], default None
        The embedding metadata to add to the schema.
    on_bad_vectors : Literal["error", "drop", "fill", "null"], default "error"
        What to do if any of the vectors are not the same size or contains NaNs.
    fill_value : float, default 0.0
        The value to use when filling vectors. Only used if on_bad_vectors="fill".
        All entries in the vector will be set to this value.
    """
    # At this point, the table might not match the schema we are targeting:
    # 1. There might be embedding columns missing that will be added
    #    in the add_embeddings step.
    # 2. If `allow_subschemas` is True, there might be columns missing.
    reader = _into_pyarrow_reader(data, target_schema)

    reader = _append_vector_columns(reader, target_schema, metadata=metadata)

    # This happens before the cast so we can fix vector columns with
    # incorrect lengths before they are cast to FSL.
    reader = _handle_bad_vectors(
        reader,
        on_bad_vectors=on_bad_vectors,
        fill_value=fill_value,
        target_schema=target_schema,
        metadata=metadata,
    )

    if target_schema is None:
        target_schema, reader = _infer_target_schema(reader)

    if metadata:
        target_schema = target_schema.with_metadata(
            _merge_metadata(target_schema.metadata, metadata)
        )

    _validate_schema(target_schema)
    reader = _cast_to_target_schema(reader, target_schema, allow_subschema)

    return reader


def _cast_to_target_schema(
    reader: pa.RecordBatchReader,
    target_schema: pa.Schema,
    allow_subschema: bool = False,
) -> pa.RecordBatchReader:
    # pa.Table.cast expects field order not to be changed.
    # Lance doesn't care about field order, so we don't need to rearrange fields
    # to match the target schema. We just need to correctly cast the fields.
    if reader.schema.equals(target_schema, check_metadata=True):
        # Fast path when the schemas are already the same
        return reader

    fields = _align_field_types(list(iter(reader.schema)), list(iter(target_schema)))
    reordered_schema = pa.schema(fields, metadata=target_schema.metadata)
    if not allow_subschema and len(reordered_schema) != len(target_schema):
        raise ValueError(
            "Input table has different number of columns than target schema"
        )

    if allow_subschema and len(reordered_schema) != len(target_schema):
        fields = _infer_subschema(
            list(iter(reader.schema)), list(iter(reordered_schema))
        )
        reordered_schema = pa.schema(fields, metadata=target_schema.metadata)

    def gen():
        for batch in reader:
            batch = _coerce_blob_write_columns(batch, reordered_schema)
            # Table but not RecordBatch has cast.
            cast_batches = (
                pa.Table.from_batches([batch]).cast(reordered_schema).to_batches()
            )
            if cast_batches:
                yield pa.RecordBatch.from_arrays(
                    cast_batches[0].columns, schema=reordered_schema
                )

    return pa.RecordBatchReader.from_batches(reordered_schema, gen())


def _coerce_blob_write_columns(
    batch: pa.RecordBatch, target_schema: pa.Schema
) -> pa.RecordBatch:
    """Materialize blob storage structs before the stream leaves Python.

    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.
    """
    columns = []
    fields = []
    changed = False
    for field, column in zip(batch.schema, batch.columns):
        target_field = target_schema.field(field.name)
        coerced = _coerce_blob_value(column, target_field)
        if coerced is not column:
            column = coerced
            field = pa.field(
                field.name,
                coerced.type,
                field.nullable,
                target_field.metadata,
            )
            changed = True
        columns.append(column)
        fields.append(field)
    if not changed:
        return batch
    return pa.RecordBatch.from_arrays(
        columns, schema=pa.schema(fields, metadata=batch.schema.metadata)
    )


def _coerce_blob_value(column: pa.Array, target_field: pa.Field) -> pa.Array:
    if is_blob_v2_field(target_field) and _can_coerce_to_blob(column.type):
        return _coerce_value_to_blob(column, target_field)

    target_type = target_field.type
    if pa.types.is_struct(target_type) and pa.types.is_struct(column.type):
        children = []
        fields = []
        changed = False
        for source_field in column.type:
            source_column = column.field(source_field.name)
            nested_target = next(
                (field for field in target_type if field.name == source_field.name),
                None,
            )
            if nested_target is None:
                children.append(source_column)
                fields.append(source_field)
                continue
            coerced = _coerce_blob_value(source_column, nested_target)
            if coerced is not source_column:
                changed = True
            child_array, child_type = _physical_array_and_type(coerced)
            children.append(child_array)
            fields.append(
                pa.field(
                    source_field.name,
                    child_type,
                    source_field.nullable,
                    nested_target.metadata,
                )
            )
        if not changed:
            return column
        return pa.StructArray.from_arrays(
            children,
            fields=fields,
            mask=column.is_null() if column.null_count else None,
        )

    if _is_list_like(target_type) and _is_list_like(column.type):
        return _coerce_blob_list_values(column, target_type.value_field)

    return column


def _coerce_blob_list_values(
    column: pa.Array, target_value_field: pa.Field
) -> pa.Array:
    """Coerce blob values inside a list column, preserving offsets and nulls.

    Works on the raw child values window instead of ``pc.list_flatten`` because
    flatten drops values spanned by null slots, which would misalign offsets.
    """
    mask = column.is_null() if column.null_count else None
    if pa.types.is_fixed_size_list(column.type):
        list_size = column.type.list_size
        values = column.values.slice(column.offset * list_size, len(column) * list_size)
        coerced = _coerce_blob_value(values, target_value_field)
        if coerced is values:
            return column
        physical_values, _ = _physical_array_and_type(coerced)
        return pa.FixedSizeListArray.from_arrays(physical_values, list_size, mask=mask)
    offsets = column.offsets
    first_offset = offsets[0].as_py()
    values = column.values.slice(
        first_offset,
        offsets[-1].as_py() - first_offset,
    )
    coerced = _coerce_blob_value(values, target_value_field)
    if coerced is values:
        return column
    physical_values, _ = _physical_array_and_type(coerced)
    if first_offset:
        offsets = pc.subtract(offsets, pa.scalar(first_offset, offsets.type))
    if pa.types.is_large_list(column.type):
        return pa.LargeListArray.from_arrays(offsets, physical_values, mask=mask)
    return pa.ListArray.from_arrays(offsets, physical_values, mask=mask)


def _coerce_value_to_blob(values: pa.Array, target_field: pa.Field) -> pa.Array:
    if pa.types.is_null(values.type):
        data = pa.nulls(len(values), type=pa.large_binary())
    elif pa.types.is_large_binary(values.type):
        data = values
    else:
        data = values.cast(pa.large_binary())
    length = len(values)
    storage_type = target_field.type
    if isinstance(storage_type, pa.ExtensionType):
        storage_type = storage_type.storage_type
    storage_fields = list(storage_type)
    children = []
    for storage_field in storage_fields:
        if storage_field.name == "data":
            children.append(data)
        else:
            children.append(pa.nulls(length, type=storage_field.type))
    storage = pa.StructArray.from_arrays(
        children,
        fields=storage_fields,
        mask=values.is_null() if values.null_count else None,
    )
    if isinstance(target_field.type, pa.ExtensionType):
        return pa.ExtensionArray.from_storage(target_field.type, storage)
    return storage


def _physical_array_and_type(array: pa.Array) -> tuple[pa.Array, pa.DataType]:
    if isinstance(array.type, pa.ExtensionType):
        return array.storage, array.type.storage_type
    return array, array.type


def _can_coerce_to_blob(data_type: pa.DataType) -> bool:
    return _is_binary_like(data_type) or pa.types.is_null(data_type)


def _is_binary_like(data_type: pa.DataType) -> bool:
    return (
        pa.types.is_binary(data_type)
        or pa.types.is_large_binary(data_type)
        or pa.types.is_binary_view(data_type)
    )


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],
) -> List[pa.Field]:
    """
    Apply the data types from the target_fields to the fields.
    """
    new_fields = []
    for field in fields:
        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")
        new_fields.append(_align_field(field, target_field))
    return new_fields


def _align_list_value_field(
    value_field: pa.Field, target_value_field: pa.Field
) -> pa.Field:
    # A list has exactly one child, so the inferred child name ("item") aligns
    # positionally and adopts the table's child name; pa.Table.cast renames it.
    return _align_field(value_field, target_value_field).with_name(
        target_value_field.name
    )


def _align_field(field: pa.Field, target_field: pa.Field) -> pa.Field:
    # 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"
    ):
        return field
    if pa.types.is_struct(target_field.type):
        if pa.types.is_struct(field.type):
            new_type = pa.struct(
                _align_field_types(
                    field.type.fields,
                    target_field.type.fields,
                )
            )
        else:
            new_type = target_field.type
    elif pa.types.is_list(target_field.type):
        if _is_list_like(field.type):
            new_type = pa.list_(
                _align_list_value_field(
                    field.type.value_field, target_field.type.value_field
                )
            )
        else:
            new_type = target_field.type
    elif pa.types.is_large_list(target_field.type):
        if _is_list_like(field.type):
            new_type = pa.large_list(
                _align_list_value_field(
                    field.type.value_field, target_field.type.value_field
                )
            )
        else:
            new_type = target_field.type
    elif pa.types.is_fixed_size_list(target_field.type):
        if _is_list_like(field.type):
            new_type = pa.list_(
                _align_list_value_field(
                    field.type.value_field, target_field.type.value_field
                ),
                target_field.type.list_size,
            )
        else:
            new_type = target_field.type
    else:
        new_type = target_field.type
    return pa.field(field.name, new_type, field.nullable, target_field.metadata)


def _infer_subschema(
    schema: List[pa.Field],
    reference_fields: List[pa.Field],
) -> List[pa.Field]:
    """
    Transform the list of fields so the types match the reference_fields.

    The order of the fields is preserved.

    ``schema`` may have fewer fields than `reference_fields`, but it may not have
    more fields.

    """
    fields = []
    lookup = {f.name: f for f in reference_fields}
    for field in schema:
        reference = lookup.get(field.name)
        if reference is None:
            raise ValueError("Unexpected field in schema: {}".format(field))

        if pa.types.is_struct(reference.type):
            new_type = pa.struct(
                _infer_subschema(
                    field.type.fields,
                    reference.type.fields,
                )
            )
            new_field = pa.field(
                field.name,
                new_type,
                reference.nullable,
            )
        else:
            new_field = reference

        fields.append(new_field)

    return fields


def sanitize_create_table(
    data,
    schema: Union[pa.Schema, LanceModel],
    metadata=None,
    on_bad_vectors: OnBadVectorsType = "error",
    fill_value: float = 0.0,
):
    if inspect.isclass(schema) and issubclass(schema, LanceModel):
        # convert LanceModel to pyarrow schema
        # note that it's possible this contains
        # embedding function metadata already
        schema: pa.Schema = schema.to_arrow_schema()

    if data is not None:
        if metadata is None and schema is not None:
            metadata = schema.metadata
        data = _sanitize_data(
            data,
            schema,
            metadata=metadata,
            on_bad_vectors=on_bad_vectors,
            fill_value=fill_value,
        )
        schema = data.schema
    else:
        if schema is not None:
            data = pa.Table.from_batches([], schema=schema)
    if schema is None:
        if data is None:
            raise ValueError("Either data or schema must be provided")
        elif hasattr(data, "schema"):
            schema = data.schema

    if metadata:
        metadata = _merge_metadata(schema.metadata, metadata)
        schema = schema.with_metadata(metadata)
        # Need to apply metadata to the data as well
        if isinstance(data, pa.Table):
            data = data.replace_schema_metadata(metadata)
        elif isinstance(data, pa.RecordBatchReader):
            data = pa.RecordBatchReader.from_batches(schema, data)

    return data, schema


def _schema_from_hf(data, schema) -> pa.Schema:
    """
    Extract pyarrow schema from HuggingFace DatasetDict
    and validate that they're all the same schema between
    splits
    """
    for dataset in data.values():
        if schema is None:
            schema = dataset.features.arrow_schema
        elif schema != dataset.features.arrow_schema:
            msg = "All datasets in a HuggingFace DatasetDict must have the same schema"
            raise TypeError(msg)
    return schema


def _to_batches_with_split(data):
    """
    Return a generator of RecordBatches from a HuggingFace DatasetDict
    with an extra `split` column
    """
    for key, dataset in data.items():
        for batch in dataset.data.to_batches():
            table = pa.Table.from_batches([batch])
            if "split" not in table.column_names:
                table = table.append_column(
                    "split", pa.array([key] * batch.num_rows, pa.string())
                )
            for b in table.to_batches():
                yield b


def _append_vector_columns(
    reader: pa.RecordBatchReader,
    schema: Optional[pa.Schema] = None,
    *,
    metadata: Optional[dict] = None,
) -> pa.RecordBatchReader:
    """
    Use the embedding function to automatically embed the source columns and add the
    vector columns to the table.
    """
    if schema is None:
        metadata = _merge_metadata(metadata)
    else:
        metadata = _merge_metadata(schema.metadata, metadata)
    functions = EmbeddingFunctionRegistry.get_instance().parse_functions(metadata)

    if not functions:
        return reader

    fields = list(reader.schema)
    for vector_column, conf in functions.items():
        if vector_column not in reader.schema.names:
            if schema is not None:
                field = schema.field(vector_column)
            else:
                dtype = pa.list_(pa.float32(), conf.function.ndims())
                field = pa.field(vector_column, type=dtype, nullable=True)
            fields.append(field)
    schema = pa.schema(fields, metadata=reader.schema.metadata)

    def gen():
        for batch in reader:
            for vector_column, conf in functions.items():
                func = conf.function
                no_vector_column = vector_column not in batch.column_names
                if no_vector_column or pc.all(pc.is_null(batch[vector_column])).as_py():
                    col_data = func.compute_source_embeddings_with_retry(
                        batch[conf.source_column]
                    )
                    # Replace vectors with wrong length (including empty lists
                    # returned for inputs like empty strings) with None so that
                    # _handle_bad_vectors can process them according to the
                    # on_bad_vectors policy instead of crashing when PyArrow
                    # tries to cast them into a fixed-size list array.
                    expected_ndims = conf.function.ndims()
                    col_data = [
                        v if v is not None and len(v) == expected_ndims else None
                        for v in col_data
                    ]
                    if no_vector_column:
                        batch = batch.append_column(
                            schema.field(vector_column),
                            pa.array(col_data, type=schema.field(vector_column).type),
                        )
                    else:
                        batch = batch.set_column(
                            batch.column_names.index(vector_column),
                            schema.field(vector_column),
                            pa.array(col_data, type=schema.field(vector_column).type),
                        )
            yield batch

    return pa.RecordBatchReader.from_batches(schema, gen())


def _table_path(base: str, table_name: str) -> str:
    """
    Get a table path that can be used in PyArrow FS.

    Removes any weird schemes (such as "s3+ddb") and drops any query params.
    """
    uri = _table_uri(base, table_name)
    # Parse as URL
    parsed = urlparse(uri)
    # If scheme is s3+ddb, convert to s3
    if parsed.scheme == "s3+ddb":
        parsed = parsed._replace(scheme="s3")
    # Remove query parameters
    return parsed._replace(query=None).geturl()


def _table_uri(base: str, table_name: str) -> str:
    return join_uri(base, f"{table_name}.lance")


def _normalize_progress(progress):
    """Normalize a ``progress`` parameter for :meth:`Table.add`.

    Returns ``(progress_obj, owns)`` where *owns* is True when we created a
    tqdm bar that the caller must close.
    """
    if progress is True:
        from tqdm.auto import tqdm

        return tqdm(unit=" rows"), True
    if progress is False or progress is None:
        return None, False
    return progress, False


class Table(ABC):
    """
    A Table is a collection of Records in a LanceDB Database.

    Examples
    --------

    Create using [DBConnection.create_table][lancedb.DBConnection.create_table]
    (more examples in that method's documentation).

    >>> import lancedb
    >>> db = lancedb.connect("./.lancedb")
    >>> table = db.create_table("my_table", data=[{"vector": [1.1, 1.2], "b": 2}])
    >>> table.head()
    pyarrow.Table
    vector: fixed_size_list<item: float>[2]
      child 0, item: float
    b: int64
    ----
    vector: [[[1.1,1.2]]]
    b: [[2]]

    Can append new data with [Table.add()][lancedb.table.Table.add].

    >>> table.add([{"vector": [0.5, 1.3], "b": 4}])
    AddResult(version=2)

    Can query the table with [Table.search][lancedb.table.Table.search].

    >>> table.search([0.4, 0.4]).select(["b", "vector"]).to_pandas()
       b      vector  _distance
    0  4  [0.5, 1.3]       0.82
    1  2  [1.1, 1.2]       1.13

    Search queries are much faster when an index is created. See
    [Table.create_index][lancedb.table.Table.create_index].
    """

    @property
    @abstractmethod
    def name(self) -> str:
        """The name of this Table"""
        raise NotImplementedError

    @property
    @abstractmethod
    def version(self) -> int:
        """The version of this Table"""
        raise NotImplementedError

    @property
    @abstractmethod
    def schema(self) -> pa.Schema:
        """The [Arrow Schema](https://arrow.apache.org/docs/python/api/datatypes.html#)
        of this Table

        """
        raise NotImplementedError

    @property
    @abstractmethod
    def tags(self) -> Tags:
        """Tag management for the table.

        Similar to Git, tags are a way to add metadata to a specific version of the
        table.

        .. warning::

            Tagged versions are exempted from the :py:meth:`cleanup_old_versions()`
            process.

            To remove a version that has been tagged, you must first
            :py:meth:`~Tags.delete` the associated tag.

        Examples
        --------

        .. code-block:: python

            table = db.open_table("my_table")
            table.tags.create("v2-prod-20250203", 10)

            tags = table.tags.list()

        """
        raise NotImplementedError

    @property
    def branches(self) -> "Branches":
        """Branch management for the table.

        Branches are isolated, writable lines of history forked from another
        branch (or version). Writes on a branch do not affect ``main``.
        """
        raise NotImplementedError

    def current_branch(self) -> Optional[str]:
        """The branch this table handle is scoped to, or ``None`` for ``main``."""
        raise NotImplementedError

    def __len__(self) -> int:
        """The number of rows in this Table"""
        return self.count_rows(None)

    @property
    @abstractmethod
    def embedding_functions(self) -> Dict[str, EmbeddingFunctionConfig]:
        """
        Get a mapping from vector column name to it's configured embedding function.
        """

    @abstractmethod
    def count_rows(self, filter: Optional[str] = None) -> int:
        """
        Count the number of rows in the table.

        Parameters
        ----------
        filter: str, optional
            A SQL where clause to filter the rows to count.
        """
        raise NotImplementedError

    def to_pandas(self, blob_mode: BlobMode = "lazy", **kwargs) -> "pandas.DataFrame":
        """Return the table as a pandas DataFrame.

        Parameters
        ----------
        blob_mode: str, default "lazy"
            Controls how blob columns are returned for backends that support
            Lance blob-aware pandas conversion.
        **kwargs
            Forwarded to PyArrow / Lance pandas conversion.

        Returns
        -------
        pd.DataFrame
        """
        return self.to_arrow().to_pandas(**kwargs)

    @abstractmethod
    def to_arrow(self) -> pa.Table:
        """Return the table as a pyarrow Table.

        Returns
        -------
        pa.Table
        """
        raise NotImplementedError

    def to_lance(self, **kwargs) -> lance.LanceDataset:
        """Return the table as a lance.LanceDataset.

        Returns
        -------
        lance.LanceDataset
        """
        raise NotImplementedError

    def to_polars(self, **kwargs) -> "pl.LazyFrame":
        """Return the table as a Polars LazyFrame.

        Note
        ----
        The Polars streaming engine is not supported because it does not currently
        implement Python PyArrow dataset scans. Use the default engine when collecting
        this LazyFrame.

        Returns
        -------
        polars.LazyFrame
        """
        raise NotImplementedError

    # New unified API overload
    @overload
    def create_index(
        self,
        column: str,
        /,
        *,
        config: IndexConfigType,
        replace: bool = ...,
        wait_timeout: Optional[timedelta] = ...,
        name: Optional[str] = ...,
        train: bool = ...,
    ) -> None: ...

    # Legacy API overload (deprecated)
    @overload
    def create_index(
        self,
        metric: Literal["l2", "cosine", "dot", "hamming"] = ...,
        num_partitions: Optional[int] = ...,
        num_sub_vectors: Optional[int] = ...,
        vector_column_name: str = ...,
        replace: bool = ...,
        accelerator: Optional[str] = ...,
        index_cache_size: Optional[int] = ...,
        *,
        index_type: VectorIndexType = ...,
        wait_timeout: Optional[timedelta] = ...,
        num_bits: int = ...,
        max_iterations: int = ...,
        sample_rate: int = ...,
        m: int = ...,
        ef_construction: int = ...,
        name: Optional[str] = ...,
        train: bool = ...,
        target_partition_size: Optional[int] = ...,
    ) -> None: ...

    def create_index(
        self,
        metric: DistanceType = "l2",
        num_partitions: Optional[int] = None,
        num_sub_vectors: Optional[int] = None,
        vector_column_name: str = VECTOR_COLUMN_NAME,
        replace: bool = True,
        accelerator: Optional[str] = None,
        index_cache_size: Optional[int] = None,
        *,
        index_type: VectorIndexType = "IVF_PQ",
        wait_timeout: Optional[timedelta] = None,
        num_bits: int = 8,
        max_iterations: int = 50,
        sample_rate: int = 256,
        m: int = 20,
        ef_construction: int = 300,
        config: Optional[IndexConfigType] = None,
        name: Optional[str] = None,
        train: bool = True,
        target_partition_size: Optional[int] = None,
    ):
        """Create an index on a column.

        This method supports both the new unified API and the legacy API
        for backwards compatibility. The new API takes the column name as the
        first positional argument and an index configuration object via
        ``config``; the legacy API takes the distance metric as the first
        argument plus separate ``vector_column_name`` / ``num_partitions`` /
        etc. parameters, and emits a ``DeprecationWarning``.

        Parameters
        ----------
        metric : str
            For new API: the column name to index.
            For legacy API: the distance metric ("l2", "cosine", "dot", "hamming").
        config : IndexConfigType, optional
            The index configuration object. If provided, uses the new unified API.
            Can be one of: IvfFlat, IvfPq, IvfSq, IvfRq, HnswPq, HnswSq,
            BTree, Bitmap, LabelList, Fm, FTS.
        replace : bool, default True
            Whether to replace an existing index on this column.
        wait_timeout : timedelta, optional
            Timeout to wait for async indexing to complete.
        name : str, optional
            Custom name for the index.
        train : bool, default True
            Whether to train the index with existing data.

        Examples
        --------
        New API (recommended):

        >>> table.create_index(  # doctest: +SKIP
        ...     "vector", config=IvfPq(distance_type="l2")
        ... )
        >>> table.create_index("category", config=BTree())  # doctest: +SKIP
        >>> table.create_index("content", config=FTS())  # doctest: +SKIP

        Legacy API (deprecated):

        >>> table.create_index(  # doctest: +SKIP
        ...     "l2", vector_column_name="vector"
        ... )
        """
        raise NotImplementedError

    def create_index_async(
        self,
        column: str,
        *,
        config: IndexConfigType,
        replace: Optional[bool] = None,
        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
        train: bool = True,
    ) -> Job:
        """Create an index, returning a handle to the indexing job.

        Takes the same arguments as :meth:`create_index`. The job may already
        be complete when returned; callers must not assume the index exists
        until :meth:`Job.wait` returns.
        """
        raise NotImplementedError

    def drop_index(self, name: str) -> None:
        """
        Drop an index from the table.

        Parameters
        ----------
        name: str
            The name of the index to drop.

        Notes
        -----
        This does not delete the index from disk, it just removes it from the table.
        To delete the index, run [optimize][lancedb.table.Table.optimize]
        after dropping the index.

        Use [list_indices][lancedb.table.Table.list_indices] to find the names of
        the indices.
        """
        raise NotImplementedError

    def wait_for_index(
        self, index_names: Iterable[str], timeout: timedelta = timedelta(seconds=300)
    ) -> None:
        """
        Wait for indexing to complete for the given index names.
        This will poll the table until all the indices are fully indexed,
        or raise a timeout exception if the timeout is reached.

        Parameters
        ----------
        index_names: str
            The name of the indices to poll
        timeout: timedelta
            Timeout to wait for asynchronous indexing. The default is 5 minutes.
        """
        raise NotImplementedError

    @abstractmethod
    def stats(self) -> TableStatistics:
        """
        Retrieve table and fragment statistics.
        """
        raise NotImplementedError

    @abstractmethod
    def create_scalar_index(
        self,
        column: str,
        *,
        replace: bool = True,
        index_type: ScalarIndexType = "BTREE",
        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
    ):
        """Create a scalar index on a column.

        Parameters
        ----------
        column : str
            The column to be indexed.  Must be a boolean, integer, float,
            or string column.
        replace : bool, default True
            Replace the existing index if it exists.
        index_type: Literal["BTREE", "BITMAP", "LABEL_LIST"], default "BTREE"
            The type of index to create.
        wait_timeout: timedelta, optional
            The timeout to wait if indexing is asynchronous.
        name: str, optional
            The name of the index. If not provided, a default name will be generated.
        Examples
        --------

        Scalar indices, like vector indices, can be used to speed up scans.  A scalar
        index can speed up scans that contain filter expressions on the indexed column.
        For example, the following scan will be faster if the column ``my_col`` has
        a scalar index:

        >>> import lancedb # doctest: +SKIP
        >>> db = lancedb.connect("/data/lance") # doctest: +SKIP
        >>> img_table = db.open_table("images") # doctest: +SKIP
        >>> my_df = img_table.search().where("my_col = 7", # doctest: +SKIP
        ...                                  prefilter=True).to_pandas()

        Scalar indices can also speed up scans containing a vector search and a
        prefilter:

        >>> import lancedb # doctest: +SKIP
        >>> db = lancedb.connect("/data/lance") # doctest: +SKIP
        >>> img_table = db.open_table("images") # doctest: +SKIP
        >>> img_table.search([1, 2, 3, 4], vector_column_name="vector") # doctest: +SKIP
        ...     .where("my_col != 7", prefilter=True)
        ...     .to_pandas()

        Scalar indices can only speed up scans for basic filters using
        equality, comparison, range (e.g. ``my_col BETWEEN 0 AND 100``), and set
        membership (e.g. `my_col IN (0, 1, 2)`)

        Scalar indices can be used if the filter contains multiple indexed columns and
        the filter criteria are AND'd or OR'd together
        (e.g. ``my_col < 0 AND other_col> 100``)

        Scalar indices may be used if the filter contains non-indexed columns but,
        depending on the structure of the filter, they may not be usable.  For example,
        if the column ``not_indexed`` does not have a scalar index then the filter
        ``my_col = 0 OR not_indexed = 1`` will not be able to use any scalar index on
        ``my_col``.
        """
        raise NotImplementedError

    def create_fts_index(
        self,
        field_names: Union[str, List[str]],
        *,
        ordering_field_names: Optional[Union[str, List[str]]] = None,
        replace: bool = False,
        writer_heap_size: Optional[int] = 1024 * 1024 * 1024,
        use_tantivy: bool = False,
        tokenizer_name: Optional[str] = None,
        with_position: bool = False,
        # tokenizer configs:
        base_tokenizer: BaseTokenizerType = "simple",
        language: str = "English",
        max_token_length: Optional[int] = 40,
        lower_case: bool = True,
        stem: bool = True,
        remove_stop_words: bool = True,
        custom_stop_words: Optional[List[str]] = None,
        ascii_folding: bool = True,
        ngram_min_length: int = 3,
        ngram_max_length: int = 3,
        prefix_only: bool = False,
        block_size: int = 128,
        document_granularity: DocumentGranularity = DocumentGranularity.ROW,
        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
    ):
        """Create a full-text search index on the table.

        Warning - this API is highly experimental and is highly likely to change
        in the future.

        Parameters
        ----------
        field_names: str or list of str
            The name of the field to index. Native FTS indexes can only be
            created on a single field at a time. To search over multiple text
            fields, create a separate FTS index for each field.
        replace: bool, default False
            If True, replace the existing index if it exists. Note that this is
            not yet an atomic operation; the index will be temporarily
            unavailable while the new index is being created.
        writer_heap_size: int, default 1GB
            Deprecated legacy Tantivy parameter. Any value other than the
            default raises an error.
        ordering_field_names:
            Deprecated legacy Tantivy parameter. Setting this raises an error.
        tokenizer_name: str, default "default"
            A compatibility alias for native tokenizer configs. Can be "raw",
            "default" or the 2 letter language code followed by "_stem". So
            for english it would be "en_stem". For new native FTS indexes, use
            ``base_tokenizer`` directly; ``tokenizer_name`` is a legacy
            compatibility alias and does not expose model-backed tokenizer names
            such as ``jieba/default`` or ``lindera/ipadic``.
        use_tantivy: bool, default False
            Deprecated legacy Tantivy parameter. Setting this to True raises an
            error.
        with_position: bool, default False
            If False, do not store the positions of the terms in the text.
            This can reduce the size of the index and improve indexing speed.
            But it will raise an exception for phrase queries.
        base_tokenizer : str, default "simple"
            The base tokenizer to use for tokenization. Options are:
            - "simple": Splits text by whitespace and punctuation.
            - "whitespace": Split text by whitespace, but not punctuation.
            - "raw": No tokenization. The entire text is treated as a single token.
            - "ngram": N-Gram tokenizer.
            - "icu": ICU dictionary-based word segmentation.
            - "icu/split": ICU segmentation with simple-style delimiter splitting.
            - "jieba/*": Jieba tokenizer loaded from Lance's language model home.
            - "lindera/*": Lindera tokenizer loaded from Lance's language model home.
        language : str, default "English"
            The language to use for stemming and stop-word removal. This is not
            the primary way to enable CJK tokenization.
        max_token_length : int, default 40
            The maximum token length to index. Tokens longer than this length will be
            ignored.
        lower_case : bool, default True
            Whether to convert the token to lower case. This makes queries
            case-insensitive.
        stem : bool, default True
            Whether to stem the token. Stemming reduces words to their root form.
            For example, in English "running" and "runs" would both be reduced to "run".
        remove_stop_words : bool, default True
            Whether to remove stop words. Stop words are common words that are often
            removed from text before indexing. For example, in English "the" and "and".
        custom_stop_words : list of str, optional
            Custom words that replace the built-in language stop words. ``None``
            uses the built-in list; an empty list explicitly uses no stop words.
        ascii_folding : bool, default True
            Whether to fold ASCII characters. This converts accented characters to
            their ASCII equivalent. For example, "café" would be converted to "cafe".
        ngram_min_length: int, default 3
            The minimum length of an n-gram.
        ngram_max_length: int, default 3
            The maximum length of an n-gram.
        prefix_only: bool, default False
            Whether to only index the prefix of the token for ngram tokenizer.
        block_size: int, default 128
            The number of documents per compressed posting block. Must be 128
            or 256. A value of 256 uses the experimental FTS V3 format and
            may introduce breaking changes.
        document_granularity: DocumentGranularity, default ROW
            ``ROW`` treats the selected text in one table row as one document.
            ``LIST_ELEMENT`` treats each element of the deepest list on the field
            path as one document and returns its physical coordinates in
            ``_doc_index`` for matching queries.
        wait_timeout: timedelta, optional
            The timeout to wait if indexing is asynchronous.
        name: str, optional
            The name of the index. If not provided, a default name will be generated.

        Notes
        -----
        Model-backed tokenizers such as ``jieba/default`` and ``lindera/ipadic``
        require tokenizer models in Lance's language model home. Set
        ``LANCE_LANGUAGE_MODEL_HOME`` to override the default platform data
        directory under ``lance/language_models``.
        """
        raise NotImplementedError

    @abstractmethod
    def add(
        self,
        data: DATA,
        mode: AddMode = "append",
        on_bad_vectors: OnBadVectorsType = "error",
        fill_value: float = 0.0,
        progress: Optional[Union[bool, Callable, Any]] = None,
        write_parallelism: Optional[int] = None,
        allow_external_blob_outside_bases: bool = False,
    ) -> AddResult:
        """Add more data to the [Table][lancedb.table.Table].

        Parameters
        ----------
        data: DATA
            The data to insert into the table. Acceptable types are:

            - list-of-dict

            - pandas.DataFrame

            - pyarrow.Table or pyarrow.RecordBatch
        mode: str
            The mode to use when writing the data. Valid values are
            "append" and "overwrite".
        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".
        fill_value: float, default 0.
            The value to use when filling vectors. Only used if on_bad_vectors="fill".
        progress: bool, callable, or tqdm-like, optional
            Progress reporting during the add operation. Can be:

            - ``True`` to automatically create and display a tqdm progress
              bar (requires ``tqdm`` to be installed)::

                table.add(data, progress=True)

            - A **callable** that receives a dict with keys ``output_rows``,
              ``output_bytes``, ``total_rows``, ``elapsed_seconds``,
              ``active_tasks``, ``total_tasks``, and ``done``::

                def on_progress(p):
                    print(f"{p['output_rows']}/{p['total_rows']} rows, "
                          f"{p['active_tasks']}/{p['total_tasks']} workers")
                table.add(data, progress=on_progress)

            - A **tqdm-compatible** progress bar whose ``total`` and
              ``update()`` will be called automatically. The postfix shows
              write throughput (MB/s) and active worker count::

                with tqdm() as pbar:
                    table.add(data, progress=pbar)

        write_parallelism: int, optional
            Number of partitions to write in parallel. Higher values increase
            throughput but also peak memory use, since each partition buffers
            data in flight. Defaults to an estimate based on the data size,
            capped at the number of CPU cores. Lower this if bulk ingestion is
            using too much memory.
        allow_external_blob_outside_bases: bool, default False
            Store blob URIs that sit outside registered blob bases. The row
            keeps a reference, so the object has to stay readable. Local
            tables only.

        Returns
        -------
        AddResult
            An object containing the new version number of the table after adding data.
        """
        raise NotImplementedError

    def merge_insert(self, on: Union[str, Iterable[str]]) -> LanceMergeInsertBuilder:
        """
        Returns a [`LanceMergeInsertBuilder`][lancedb.merge.LanceMergeInsertBuilder]
        that can be used to create a "merge insert" operation

        This operation can add rows, update rows, and remove rows all in a single
        transaction. It is a very generic tool that can be used to create
        behaviors like "insert if not exists", "update or insert (i.e. upsert)",
        or even replace a portion of existing data with new data (e.g. replace
        all data where month="january")

        The merge insert operation works by combining new data from a
        **source table** with existing data in a **target table** by using a
        join.  There are three categories of records.

        "Matched" records are records that exist in both the source table and
        the target table. "Not matched" records exist only in the source table
        (e.g. these are new data) "Not matched by source" records exist only
        in the target table (this is old data)

        The builder returned by this method can be used to customize what
        should happen for each category of data.

        Please note that the data may appear to be reordered as part of this
        operation.  This is because updated rows will be deleted from the
        dataset and then reinserted at the end with the new values.

        Parameters
        ----------

        on: Union[str, Iterable[str]]
            A column (or columns) to join on.  This is how records from the
            source table and target table are matched.  Typically this is some
            kind of key or id column.  Passing several columns matches on the
            composite key: a source row updates a target row only when it
            agrees on every one of them.

        Examples
        --------
        >>> import lancedb
        >>> data = pa.table({"a": [2, 1, 3], "b": ["a", "b", "c"]})
        >>> db = lancedb.connect("./.lancedb")
        >>> table = db.create_table("my_table", data)
        >>> new_data = pa.table({"a": [2, 3, 4], "b": ["x", "y", "z"]})
        >>> # Perform a "upsert" operation
        >>> res = table.merge_insert("a")     \\
        ...      .when_matched_update_all()     \\
        ...      .when_not_matched_insert_all() \\
        ...      .execute(new_data)
        >>> res
        MergeResult(version=2, num_updated_rows=2, num_inserted_rows=1, num_deleted_rows=0, num_attempts=1, num_rows=3)
        >>> # The order of new rows is non-deterministic since we use
        >>> # a hash-join as part of this operation and so we sort here
        >>> table.to_arrow().sort_by("a").to_pandas()
           a  b
        0  1  b
        1  2  x
        2  3  y
        3  4  z
        """  # noqa: E501
        on = [on] if isinstance(on, str) else list(iter(on))

        return LanceMergeInsertBuilder(self, on)

    @abstractmethod
    def search(
        self,
        query: Optional[
            Union[VEC, str, "PIL.Image.Image", Tuple, FullTextQuery]
        ] = None,
        vector_column_name: Optional[str] = None,
        query_type: QueryType = "auto",
        ordering_field_name: Optional[str] = None,
        fts_columns: Optional[Union[str, List[str]]] = None,
    ) -> LanceQueryBuilder:
        """Create a search query to find the nearest neighbors
        of the given query vector. We currently support [vector search](https://lancedb.com/docs/search/vector-search/)
        and [full-text search](https://lancedb.com/docs/search/full-text-search/).

        All query options are defined in
        [LanceQueryBuilder][lancedb.query.LanceQueryBuilder].

        Examples
        --------
        >>> import lancedb
        >>> db = lancedb.connect("./.lancedb")
        >>> data = [
        ...    {"original_width": 100, "caption": "bar", "vector": [0.1, 2.3, 4.5]},
        ...    {"original_width": 2000, "caption": "foo",  "vector": [0.5, 3.4, 1.3]},
        ...    {"original_width": 3000, "caption": "test", "vector": [0.3, 6.2, 2.6]}
        ... ]
        >>> table = db.create_table("my_table", data)
        >>> query = [0.4, 1.4, 2.4]
        >>> (table.search(query)
        ...     .where("original_width > 1000", prefilter=True)
        ...     .select(["caption", "original_width", "vector"])
        ...     .limit(2)
        ...     .to_pandas())
          caption  original_width           vector  _distance
        0     foo            2000  [0.5, 3.4, 1.3]   5.220000
        1    test            3000  [0.3, 6.2, 2.6]  23.089996

        Parameters
        ----------
        query: list/np.ndarray/str/PIL.Image.Image, default None
            The targetted vector to search for.

            - *default None*.
            Acceptable types are: list, np.ndarray, PIL.Image.Image

            - If None then the select/where/limit clauses are applied to filter
            the table
        vector_column_name: str, optional
            The name of the vector column to search.

            The vector column needs to be a pyarrow fixed size list type

            - If not specified then the vector column is inferred from
            the table schema

            - If the table has multiple vector columns then the *vector_column_name*
            needs to be specified. Otherwise, an error is raised.
        query_type: str
            *default "auto"*.
            Acceptable types are: "vector", "fts", "hybrid", or "auto"

            - If "auto" then the query type is inferred from the query;

                - If `query` is a list/np.ndarray then the query type is
                "vector";

                - If `query` is a PIL.Image.Image then either do vector search,
                or raise an error if no corresponding embedding function is found.

            - If `query` is a string, then the query type is "vector" if the
            table has embedding functions else the query type is "fts"

        Returns
        -------
        LanceQueryBuilder
            A query builder object representing the query.
            Once executed, the query returns

            - selected columns

            - the vector

            - and also the "_distance" column which is the distance between the query
            vector and the returned vector.
        """
        raise NotImplementedError

    @abstractmethod
    def take_offsets(
        self, offsets: list[int], *, with_row_id: bool = False
    ) -> LanceTakeQueryBuilder:
        """
        Take a list of offsets from the table.

        Offsets are 0-indexed and relative to the current version of the table.  Offsets
        are not stable.  A row with an offset of N may have a different offset in a
        different version of the table (e.g. if an earlier row is deleted).

        Offsets are mostly useful for sampling as the set of all valid offsets is easily
        known in advance to be [0, len(table)).

        No guarantees are made regarding the order in which results are returned.
        Repeated offsets produce repeated rows, which makes this method suitable for
        sampling with replacement.

        Parameters
        ----------
        offsets: list[int]
            The offsets to take.

        Returns
        -------
        pa.RecordBatch
            A record batch containing the rows at the given offsets.
        """

    def __getitems__(self, offsets: list[int]) -> pa.RecordBatch:
        """
        Take a list of offsets from the table and return as a record batch.

        This method uses the `take_offsets` method to take the rows.  However, it
        aligns the offsets to the passed in offsets.  This means the return type
        is a record batch (and so users should take care not to pass in too many
        offsets)

        Note: this method is primarily intended to fulfill the Dataset contract
        for pytorch.

        Parameters
        ----------
        offsets: list[int]
            The offsets to take.

        Returns
        -------
        pa.RecordBatch
            A record batch containing the rows at the given offsets.
        """
        # We don't know the order of the results at all.  So we calculate a permutation
        # for ordering the given offsets.  Then we load the data with the _rowoffset
        # column.  Then we sort by _rowoffset and apply the inverse of the permutation
        # that we calculated.
        #
        # Note: this is potentially a lot of memory copy if we're operating on large
        # batches :(
        num_offsets = len(offsets)
        indices = list(range(num_offsets))
        permutation = sorted(indices, key=lambda idx: offsets[idx])
        permutation_inv = [0] * num_offsets
        for i in range(num_offsets):
            permutation_inv[permutation[i]] = i

        columns = self.schema.names
        columns.append("_rowoffset")
        tbl = (
            self.take_offsets(offsets)
            .select(columns)
            .to_arrow()
            .sort_by("_rowoffset")
            .take(permutation_inv)
            .combine_chunks()
            .drop_columns(["_rowoffset"])
        )

        return tbl

    @abstractmethod
    def take_row_ids(
        self, row_ids: list[int], *, with_row_id: bool = False
    ) -> LanceTakeQueryBuilder:
        """
        Take a list of row ids from the table.

        Row ids are not stable and are relative to the current version of the table.
        They can change due to compaction and updates.

        No guarantees are made regarding the order in which results are returned.  If
        you desire an output order that matches the order of the given ids, you will
        need to add the row id column to the output and align it yourself.

        Unlike offsets, row ids are not 0-indexed and no assumptions should be made
        about the possible range of row ids.  In order to use this method you must
        first obtain the row ids by scanning or searching the table.

        Even so, row ids are more stable than offsets and can be useful in some
        situations.

        There is an ongoing effort to make row ids stable which is tracked at
        https://github.com/lancedb/lancedb/issues/1120

        Parameters
        ----------
        row_ids: list[int]
            The row ids to take.

        Returns
        -------
        AsyncTakeQuery
            A query object that can be executed to get the rows.
        """

    @abstractmethod
    def blob_columns(self) -> list[str]:
        """Names of the blob v2 columns declared on this table."""

    @abstractmethod
    def fetch_blobs(
        self, column: str, row_ids: Union[list[int], pa.Table]
    ) -> pa.LargeBinaryArray:
        """Materialize full blob bytes for ``column`` at the given rows.

        The result has the same length and order as ``row_ids``. Null blobs
        produce null slots; valid empty blobs produce ``b""``.

        ``_rowid`` values stay valid after compaction when the table has stable
        row ids.

        Convenience for small payloads. For large values use
        :meth:`fetch_blob_files`.
        """

    @abstractmethod
    def fetch_blob_ranges(
        self,
        column: str,
        requests: Sequence[Tuple[int, int, int]],
    ) -> pa.LargeBinaryArray:
        """Materialize row-specific byte ranges from a blob v2 column.

        Each request is a ``(row_id, offset, length)`` tuple. Requests may be
        repeated or reordered, including multiple ranges for the same blob.
        The result has the same length and order as ``requests``; null blobs
        produce null slots and empty ranges on non-null blobs produce ``b""``.

        ``_rowid`` values stay valid after compaction when the table has stable
        row ids.

        Row IDs can be obtained from a query with ``with_row_id(True)``. This
        API is currently supported only by local tables.
        """

    @abstractmethod
    def fetch_blob_files(
        self, column: str, row_ids: Union[list[int], pa.Table]
    ) -> "list[Optional[BlobFile]]":
        """Open lazy, seekable :class:`~lancedb._blob.BlobFile` handles.

        Prefer this over :meth:`fetch_blobs` for large payloads. ``row_ids`` is
        a ``list[int]`` or a query ``pyarrow.Table`` carrying row identity via
        ``_rowid`` or a ``_lance_row_id`` field on the blob descriptor. Null
        rows are ``None``. Remote tables require LanceDB Cloud server 0.5.0 or
        newer.

        ``_rowid`` values stay valid after compaction when the table has stable
        row ids.
        """

    @abstractmethod
    def _execute_query(
        self,
        query: Query,
        *,
        batch_size: Optional[int] = None,
        timeout: Optional[timedelta] = None,
    ) -> pa.RecordBatchReader: ...

    @abstractmethod
    def _explain_plan(self, query: Query, verbose: Optional[bool] = False) -> str: ...

    @abstractmethod
    def _analyze_plan(
        self,
        query: Query,
        *,
        distributed_metrics: AnalyzePlanDistributedMetrics = "aggregate",
    ) -> str: ...

    @abstractmethod
    def _output_schema(self, query: Query) -> pa.Schema: ...

    @abstractmethod
    def _do_merge(
        self,
        merge: LanceMergeInsertBuilder,
        new_data: DATA,
        on_bad_vectors: OnBadVectorsType,
        fill_value: float,
    ) -> MergeResult: ...

    @abstractmethod
    def delete(self, where: Union[str, Expr]) -> DeleteResult:
        """Delete rows from the table.

        This can be used to delete a single row, many rows, all rows, or
        sometimes no rows (if your predicate matches nothing).

        Parameters
        ----------
        where: str or :class:`~lancedb.expr.Expr`
            The filter condition. Can be a SQL string or a type-safe
            :class:`~lancedb.expr.Expr` built with :func:`~lancedb.expr.col`
            and :func:`~lancedb.expr.lit`.

            The filter must not be empty, or it will error.

        Returns
        -------
        DeleteResult
            An object containing the new version number of the table after deletion.

        Examples
        --------
        >>> import lancedb
        >>> data = [
        ...    {"x": 1, "vector": [1.0, 2]},
        ...    {"x": 2, "vector": [3.0, 4]},
        ...    {"x": 3, "vector": [5.0, 6]}
        ... ]
        >>> db = lancedb.connect("./.lancedb")
        >>> table = db.create_table("my_table", data)
        >>> table.to_pandas()
           x      vector
        0  1  [1.0, 2.0]
        1  2  [3.0, 4.0]
        2  3  [5.0, 6.0]
        >>> table.delete("x = 2")
        DeleteResult(num_deleted_rows=1, version=2)
        >>> table.to_pandas()
           x      vector
        0  1  [1.0, 2.0]
        1  3  [5.0, 6.0]

        If you have a list of values to delete, you can combine them into a
        stringified list and use the `IN` operator:

        >>> to_remove = [1, 5]
        >>> to_remove = ", ".join([str(v) for v in to_remove])
        >>> to_remove
        '1, 5'
        >>> table.delete(f"x IN ({to_remove})")
        DeleteResult(num_deleted_rows=1, version=3)
        >>> table.to_pandas()
           x      vector
        0  3  [5.0, 6.0]
        """
        raise NotImplementedError

    @abstractmethod
    def update(
        self,
        where: Optional[Union[str, Expr]] = None,
        values: Optional[dict] = None,
        *,
        values_sql: Optional[Dict[str, str]] = None,
    ) -> UpdateResult:
        """
        This can be used to update zero to all rows depending on how many
        rows match the where clause. If no where clause is provided, then
        all rows will be updated.

        Either `values` or `values_sql` must be provided. You cannot provide
        both.

        Parameters
        ----------
        where: str or [Expr][lancedb.expr.Expr], optional
            The filter condition. Can be a SQL string or a type-safe
            [Expr][lancedb.expr.Expr] built with [col][lancedb.expr.col] and
            [lit][lancedb.expr.lit]. The filter must not be empty, or it will
            error.
        values: dict, optional
            The values to update. The keys are the column names and the values
            are the values to set.
        values_sql: dict, optional
            The values to update, expressed as SQL expression strings. These can
            reference existing columns. For example, {"x": "x + 1"} will increment
            the x column by 1.

        Returns
        -------
        UpdateResult
            - rows_updated: The number of rows that were updated
            - version: The new version number of the table after the update

        Examples
        --------
        >>> import lancedb
        >>> from lancedb.expr import col
        >>> import pandas as pd
        >>> data = pd.DataFrame({"x": [1, 2, 3], "vector": [[1.0, 2], [3, 4], [5, 6]]})
        >>> db = lancedb.connect("./.lancedb")
        >>> table = db.create_table("my_table", data)
        >>> table.to_pandas()
           x      vector
        0  1  [1.0, 2.0]
        1  2  [3.0, 4.0]
        2  3  [5.0, 6.0]
        >>> table.update(where=col("x") == 2, values={"vector": [10.0, 10]})
        UpdateResult(rows_updated=1, version=2)
        >>> table.to_pandas()
           x        vector
        0  1    [1.0, 2.0]
        1  3    [5.0, 6.0]
        2  2  [10.0, 10.0]
        >>> table.update(values_sql={"x": "x + 1"})
        UpdateResult(rows_updated=3, version=3)
        >>> table.to_pandas()
           x        vector
        0  2    [1.0, 2.0]
        1  4    [5.0, 6.0]
        2  3  [10.0, 10.0]
        """
        raise NotImplementedError

    @abstractmethod
    def cleanup_old_versions(
        self,
        older_than: Optional[timedelta] = None,
        *,
        delete_unverified: bool = False,
    ) -> "CleanupStats":
        """
        Clean up old versions of the table, freeing disk space.

        Parameters
        ----------
        older_than: timedelta, default None
            The minimum age of the version to delete. If None, then this defaults
            to two weeks.
        delete_unverified: bool, default False
            Because they may be part of an in-progress transaction, files newer
            than 7 days old are not deleted by default. If you are sure that
            there are no in-progress transactions, then you can set this to True
            to delete all files older than `older_than`.

        Returns
        -------
        CleanupStats
            The stats of the cleanup operation, including how many bytes were
            freed.

        See Also
        --------
        [Table.optimize][lancedb.table.Table.optimize]: A more comprehensive
            optimization operation that includes cleanup as well as other operations.

        Notes
        -----
        This function is not available in LanceDb Cloud (since LanceDB
        Cloud manages cleanup for you automatically)
        """

    @abstractmethod
    def compact_files(self, *args, **kwargs):
        """
        Run the compaction process on the table.
        This can be run after making several small appends to optimize the table
        for faster reads.

        Arguments are passed onto Lance's
        `lance.dataset.DatasetOptimizer.compact_files`.
        For most cases, the default should be fine.

        See Also
        --------
        [Table.optimize][lancedb.table.Table.optimize]: A more comprehensive
            optimization operation that includes cleanup as well as other operations.

        Notes
        -----
        This function is not available in LanceDB Cloud (since LanceDB
        Cloud manages compaction for you automatically)
        """

    @abstractmethod
    def optimize(
        self,
        *,
        cleanup_older_than: Optional[timedelta] = None,
        delete_unverified: bool = False,
        retrain: bool = False,
    ):
        """
        Optimize the on-disk data and indices for better performance.

        Modeled after ``VACUUM`` in PostgreSQL.

        Optimization covers three operations:

         * Compaction: Merges small files into larger ones
         * Prune: Removes old versions of the dataset
         * Index: Optimizes the indices, adding new data to existing indices

        Parameters
        ----------
        cleanup_older_than: timedelta, optional default 7 days
            All files belonging to versions older than this will be removed.  Set
            to 0 days to remove all versions except the latest.  The latest version
            is never removed.
        delete_unverified: bool, default False
            Files leftover from a failed transaction may appear to be part of an
            in-progress operation (e.g. appending new data) and these files will not
            be deleted unless they are at least 7 days old. If delete_unverified is True
            then these files will be deleted regardless of their age.

            .. warning::

                This should only be set to True if you can guarantee that no other
                process is currently working on this dataset. Otherwise the dataset
                could be put into a corrupted state.

        retrain: bool, default False
            This parameter is no longer used and is deprecated.

        Notes
        -----
        The frequency an application should call optimize is based on the frequency of
        data modifications.  If data is frequently added, deleted, or updated then
        optimize should be run frequently.  A good rule of thumb is to run optimize if
        you have added or modified 100,000 or more records or run more than 20 data
        modification operations.
        """

    @abstractmethod
    def list_indices(self) -> Iterable[IndexConfig]:
        """
        List all indices that have been created with
        [Table.create_index][lancedb.table.Table.create_index]
        """

    @abstractmethod
    def tokenize(
        self,
        query: str,
        *,
        column: Optional[str] = None,
        index_name: Optional[str] = None,
    ) -> Iterable[FtsToken]:
        """
        Tokenize a query using the tokenizer configured on an FTS index.

        Specify exactly one of ``column`` or ``index_name``.

        Model-backed tokenizers such as ``jieba/*`` and ``lindera/*`` are
        rebuilt in the client process from index metadata. For remote tables,
        this means the same tokenizer model files must also exist locally.
        """

    @abstractmethod
    def index_stats(self, index_name: str) -> Optional[IndexStatistics]:
        """
        Retrieve statistics about an index

        Parameters
        ----------
        index_name: str
            The name of the index to retrieve statistics for

        Returns
        -------
        IndexStatistics or None
            The statistics about the index. Returns None if the index does not exist.
        """

    @abstractmethod
    def add_columns(
        self,
        transforms: Dict[str, str | FunctionApplication]
        | FunctionApplication
        | pa.Field
        | List[pa.Field]
        | pa.Schema
        | None = None,
        *,
        computed: Dict[str, str] | None = None,
    ):
        """
        Add new columns with defined values.

        Parameters
        ----------
        transforms: Dict[str, str | FunctionApplication], FunctionApplication,
            pa.Field, List[pa.Field], pa.Schema
            A map of column name to a SQL expression to use to calculate the
            value of the new column. These expressions will be evaluated for
            each row in the table, and can reference existing columns.
            Alternatively, a pyarrow Field or Schema can be provided to add
            new columns with the specified data types. The new columns will
            be initialized with null values.

            A mapping with one ``FunctionApplication`` value keeps its scalar
            or named-struct result in the named table column. A bare
            named-struct application expands its ordered result fields as one
            atomic binding; aliases come from ``rename(columns=...)``.
            Function columns are supported only on LanceDB Cloud and
            Enterprise.
        computed: Dict[str, str], optional
            A mapping from output column names to SQL expressions derives each
            output field from its expression. A direct projection of a Blob v2
            field inherits Blob v2 semantics; other expressions derive their
            ordinary Arrow type. Mapping order is declaration and dependency
            order.

            Unlike ``transforms``, the expression is stored rather than
            evaluated now: the column is committed with no values, and rows get
            them from [`refresh_column`][lancedb.table.Table.refresh_column].
            Declaring one therefore costs the same on a large table as on an
            empty one.

            A refresh does not revisit rows it has already filled, so mutating
            an input leaves the value computed at fill time; recomputing means
            dropping the column and declaring it again. While a declaration
            reads a column, that column cannot be renamed, retyped or dropped.

            On LanceDB Cloud and Enterprise the expression is planned by the
            server, and the refresh runs as a server job -- see
            [`refresh_column_async`][lancedb.table.Table.refresh_column_async].
            Cannot be combined with ``transforms``.

        Returns
        -------
        AddColumnsResult
            version: the new version number of the table after adding columns.

        Examples
        --------
        >>> import lancedb
        >>> db = lancedb.connect("./.lancedb")
        >>> table = db.create_table("computed_demo", [{"x": 1}, {"x": 2}])
        >>> table.add_columns(computed={"doubled": "x * 2"})
        AddColumnsResult(version=2)
        >>> table.refresh_column("doubled")
        RefreshColumnResult(rows_filled=2, version=3)
        >>> table.to_arrow().sort_by("x").to_pandas()
           x  doubled
        0  1        2
        1  2        4
        """

    @abstractmethod
    def refresh_column(self, column: str) -> "RefreshColumnResult":
        """
        Fill the rows of a computed column that hold no value yet.

        Declared with ``add_columns(computed=...)``, a column starts empty and
        gets its values here. Rows appended since the last refresh are filled
        by the next one; rows already filled are left as they are, so the call
        is idempotent and does not observe a mutated input.

        Local tables only: a remote refresh runs as a server job, through
        [`refresh_column_async`][lancedb.table.Table.refresh_column_async].

        Parameters
        ----------
        column: str
            The name of the computed column to fill.

        Returns
        -------
        RefreshColumnResult
            rows_filled: the number of rows given a value.
            version: the new version number of the table.
        """

    @abstractmethod
    def refresh_column_async(self, column: str) -> Job[RefreshColumnJobResult]:
        """
        Like :meth:`refresh_column`, but returns a handle to the refresh job
        instead of blocking until it completes.

        The job may already be complete when returned; callers must not assume
        the column is filled until :meth:`Job.wait` returns. Invalid input --
        an unknown column, or one that is not computed -- raises here rather
        than failing the job. On local tables the job runs in-process; on
        LanceDB Cloud and Enterprise it is the server's backfill job.

        Returns
        -------
        Job[RefreshColumnResult]
            A job whose successful ``wait`` returns row counts plus the source
            and published table versions.

        Examples
        --------
        >>> import lancedb
        >>> db = lancedb.connect("./.lancedb")
        >>> table = db.create_table("computed_job_demo", [{"x": 1}, {"x": 2}])
        >>> table.add_columns(computed={"doubled": "x * 2"})
        AddColumnsResult(version=2)
        >>> job = table.refresh_column_async("doubled")
        >>> result = job.wait()
        >>> result.rows_assigned
        2
        >>> job.status()
        'finished'
        """

    @abstractmethod
    def alter_columns(self, *alterations: Iterable[Dict[str, str]]):
        """
        Alter column names and nullability.

        Parameters
        ----------
        alterations : Iterable[Dict[str, Any]]
            A sequence of dictionaries, each with the following keys:
            - "path": str
                The column path to alter. For a top-level column, this is the name.
                For a nested column, this is the dot-separated path, e.g. "a.b.c".
            - "rename": str, optional
                The new name of the column. If not specified, the column name is
                not changed.
            - "data_type": pyarrow.DataType, optional
               The new data type of the column. Existing values will be casted
               to this type. If not specified, the column data type is not changed.
            - "nullable": bool, optional
                Whether the column should be nullable. If not specified, the column
                nullability is not changed. Only non-nullable columns can be changed
                to nullable. Currently, you cannot change a nullable column to
                non-nullable.

        Returns
        -------
        AlterColumnsResult
            version: the new version number of the table after the alteration.
        """

    @abstractmethod
    def update_field_metadata(
        self, *updates: dict[str, Any]
    ) -> UpdateFieldMetadataResult:
        """
        Update per-field (column) metadata.

        Parameters
        ----------
        updates : dict
            One or more dicts, each with:

            - "path": str — dot-path to the field (e.g. "embedding" or "a.b.c").
            - "metadata": dict[str, str | None] — keys to set; a value of ``None``
              deletes that key.
            - "replace": bool, optional — replace the field's whole metadata map
              instead of merging (default False).

            The following keys are treated specially, by convention, and should
            be used when appropriate:

            - "lancedb:description": for a human-readable description of a field.
            - ``"lancedb:tag:<name>"`` for a user-defined key-value tag, where the
                suffix names the tag category; e.g. "lancedb:tag:model": "clip".
            - "lancedb:logical-column" for a column grouping; e.g. "feature_v1"
                and "feature_v2" might be in the same logical column.
            - "lancedb:status" for status options ("production", "candidate",
                "deprecated", "archived") to designate the current life cycle
                state of this column.

        Returns
        -------
        UpdateFieldMetadataResult
            version: the new table version after the update.
        """

    @abstractmethod
    def drop_columns(self, columns: Iterable[str]) -> DropColumnsResult:
        """
        Drop columns from the table.

        Parameters
        ----------
        columns : Iterable[str]
            The names of the columns to drop.

        Returns
        -------
        DropColumnsResult
            version: the new version number of the table dropping the columns.
        """

    @abstractmethod
    def checkout(self, version: Union[int, str]):
        """
        Checks out a specific version of the Table

        Any read operation on the table will now access the data at the checked out
        version. As a consequence, calling this method will disable any read consistency
        interval that was previously set.

        This is a read-only operation that turns the table into a sort of "view"
        or "detached head".  Other table instances will not be affected.  To make the
        change permanent you can use the `[Self::restore]` method.

        Any operation that modifies the table will fail while the table is in a checked
        out state. To return the table to a normal state use
        `[Self::checkout_latest]`.

        Parameters
        ----------
        version: int | str,
            The version to check out. A version number (`int`) or a tag
            (`str`) can be provided.
        """

    @abstractmethod
    def checkout_latest(self):
        """
        Ensures the table is pointing at the latest version

        This can be used to manually update a table when the read_consistency_interval
        is None
        It can also be used to undo a `[Self::checkout]` operation
        """

    @abstractmethod
    def restore(self, version: Optional[Union[int, str]] = None):
        """Restore a version of the table. This is an in-place operation.

        This creates a new version where the data is equivalent to the
        specified previous version. Data is not copied (as of python-v0.2.1).

        Parameters
        ----------
        version : int or str, default None
            The version number or version tag to restore.
            If unspecified then restores the currently checked out version.
            If the currently checked out version is the
            latest version then this is a no-op.
        """

    @abstractmethod
    def list_versions(self) -> List[Dict[str, Any]]:
        """List all versions of the table"""

    @cached_property
    def _dataset_uri(self) -> str:
        return _table_uri(self._conn.uri, self.name)

    def _get_fts_index_path(self) -> Tuple[str, pa_fs.FileSystem, bool]:
        from .remote.table import RemoteTable

        if isinstance(self, RemoteTable) or get_uri_scheme(self._dataset_uri) != "file":
            return ("", None, False)
        path = join_uri(self._dataset_uri, "_indices", "fts")
        fs, path = fs_from_uri(path)
        index_exists = fs.get_file_info(path).type != pa_fs.FileType.NotFound
        return (path, fs, index_exists)

    def _ensure_no_legacy_fts_index(self):
        path, _, exists = self._get_fts_index_path()
        if exists:
            raise ValueError(
                "Legacy Tantivy FTS index detected at "
                f"{path}. Tantivy-based FTS has been removed. "
                "Delete the legacy index and recreate it with "
                "table.create_fts_index(...)."
            )

    @abstractmethod
    def uses_v2_manifest_paths(self) -> bool:
        """
        Check if the table is using the new v2 manifest paths.

        Returns
        -------
        bool
            True if the table is using the new v2 manifest paths, False otherwise.
        """

    @abstractmethod
    def migrate_v2_manifest_paths(self):
        """
        Migrate the manifest paths to the new format.

        This will update the manifest to use the new v2 format for paths.

        This function is idempotent, and can be run multiple times without
        changing the state of the object store.

        !!! danger

            This should not be run while other concurrent operations are happening.
            And it should also run until completion before resuming other operations.

        You can use
        [Table.uses_v2_manifest_paths][lancedb.table.Table.uses_v2_manifest_paths]
        to check if the table is already using the new path style.
        """


class LanceTable(Table):
    """
    A table in a LanceDB database.

    This can be opened in two modes: standard and time-travel.

    Standard mode is the default. In this mode, the table is mutable and tracks
    the latest version of the table. The level of read consistency is controlled
    by the `read_consistency_interval` parameter on the connection.

    Time-travel mode is activated by specifying a version number. In this mode,
    the table is immutable and fixed to a specific version. This is useful for
    querying historical versions of the table.
    """

    def __init__(
        self,
        connection: "LanceDBConnection",
        name: str,
        *,
        namespace_path: Optional[List[str]] = None,
        storage_options: Optional[Dict[str, str]] = None,
        index_cache_size: Optional[int] = None,
        location: Optional[str] = None,
        namespace_client: Optional[Any] = None,
        managed_versioning: Optional[bool] = None,
        pushdown_operations: Optional[set] = None,
        route_pushdown_to_rust: bool = False,
        _async: AsyncTable = None,
    ):
        if namespace_path is None:
            namespace_path = []
        self._conn = connection
        self._namespace_path = namespace_path
        self._location = location  # Store location for use in _dataset_path
        self._namespace_client = namespace_client
        self._pushdown_operations = pushdown_operations or set()
        # When the connection built the namespace client natively (e.g. an
        # enterprise "rest" connection), the underlying Rust table already
        # executes QueryTable pushdown itself -- and, unlike this Python urllib3
        # path, it routes through the read-freshness context provider that emits
        # the ``x-lancedb-min-timestamp`` header. So we must defer pushdown to
        # Rust instead of calling the Python ``namespace_client.query_table``
        # directly, or reads silently bypass read-freshness (stale results).
        self._route_pushdown_to_rust = route_pushdown_to_rust
        if _async is not None:
            self._table = _async
        else:
            self._table = LOOP.run(
                connection._conn.open_table(
                    name,
                    namespace_path=namespace_path,
                    storage_options=storage_options,
                    index_cache_size=index_cache_size,
                    location=location,
                    namespace_client=namespace_client,
                    managed_versioning=managed_versioning,
                )
            )

    @property
    def name(self) -> str:
        return self._table.name

    @property
    def namespace(self) -> List[str]:
        """Return the namespace path of the table."""
        return self._namespace_path

    @property
    def id(self) -> str:
        """Return the full identifier of the table (namespace$name)."""
        if self._namespace_path:
            return "$".join(self._namespace_path + [self.name])
        return self.name

    @classmethod
    async def from_inner(cls, tbl: LanceDBTable):
        from .db import AsyncConnection, LanceDBConnection

        async_tbl = AsyncTable(tbl)
        inner_conn = tbl.database()
        read_consistency_interval = await AsyncConnection(
            inner_conn
        ).get_read_consistency_interval()
        conn = LanceDBConnection.from_inner(inner_conn, read_consistency_interval)
        return cls(
            conn,
            async_tbl.name,
            _async=async_tbl,
        )

    @classmethod
    def open(
        cls,
        db,
        name,
        *,
        namespace_path: Optional[List[str]] = None,
        storage_options: Optional[Dict[str, str]] = None,
        index_cache_size: Optional[int] = None,
        location: Optional[str] = None,
        namespace_client: Optional[Any] = None,
        managed_versioning: Optional[bool] = None,
        pushdown_operations: Optional[set] = None,
    ):
        if namespace_path is None:
            namespace_path = []
        tbl = cls(
            db,
            name,
            namespace_path=namespace_path,
            storage_options=storage_options,
            index_cache_size=index_cache_size,
            location=location,
            namespace_client=namespace_client,
            managed_versioning=managed_versioning,
            pushdown_operations=pushdown_operations,
        )

        # check the dataset exists
        try:
            tbl.version
        except ValueError as e:
            if "Not found:" in str(e):
                raise FileNotFoundError(f"Table {name} does not exist")
            raise e

        return tbl

    @cached_property
    def _dataset_path(self) -> str:
        # Cacheable since it's deterministic
        # If table was opened with explicit location (e.g., from namespace),
        # use that location directly instead of constructing from base URI
        if self._location is not None:
            return self._location
        return _table_path(self._conn.uri, self.name)

    def to_lance(self, **kwargs) -> lance.LanceDataset:
        """Return the LanceDataset backing this table."""
        try:
            import lance
        except ImportError:
            raise ImportError(
                "The lance library is required to use this function. "
                "Please install with `pip install pylance`."
            )

        branch = self.current_branch()
        version = None if branch is not None else self.version
        namespace_client = self._namespace_client
        if namespace_client is None:
            conn_uri = getattr(self._conn, "uri", "")
            if get_uri_scheme(conn_uri) == "namespace" or self._namespace_path:
                namespace_client = self._conn.namespace_client()
                self._namespace_client = namespace_client

        if namespace_client is not None:
            table_id = self._namespace_path + [self.name]
            ds = lance.dataset(
                version=version,
                storage_options=self._conn.storage_options,
                namespace_client=namespace_client,
                table_id=table_id,
                **kwargs,
            )
        else:
            ds = lance.dataset(
                self._dataset_path,
                version=version,
                storage_options=self._conn.storage_options,
                **kwargs,
            )
        if branch is not None:
            ds = ds.checkout_version((branch, self.version))
        return ds

    @property
    def schema(self) -> pa.Schema:
        """Return the schema of the table.

        Returns
        -------
        pa.Schema
            A PyArrow schema object."""
        return LOOP.run(self._table.schema())

    def list_versions(self) -> List[Dict[str, Any]]:
        """List all versions of the table"""
        return LOOP.run(self._table.list_versions())

    @property
    def version(self) -> int:
        """Get the current version of the table"""
        return LOOP.run(self._table.version())

    def take_offsets(self, offsets: list[int]) -> LanceTakeQueryBuilder:
        return LanceTakeQueryBuilder(self._table.take_offsets(offsets))

    def take_row_ids(self, row_ids: list[int]) -> LanceTakeQueryBuilder:
        return LanceTakeQueryBuilder(self._table.take_row_ids(row_ids))

    def blob_columns(self) -> list[str]:
        return LOOP.run(self._table.blob_columns())

    def fetch_blobs(
        self, column: str, row_ids: Union[list[int], pa.Table]
    ) -> pa.LargeBinaryArray:
        return LOOP.run(self._table.fetch_blobs(column, row_ids))

    def fetch_blob_ranges(
        self,
        column: str,
        requests: Sequence[Tuple[int, int, int]],
    ) -> pa.LargeBinaryArray:
        return LOOP.run(self._table.fetch_blob_ranges(column, list(requests)))

    def fetch_blob_files(
        self, column: str, row_ids: Union[list[int], pa.Table]
    ) -> "list[Optional[BlobFile]]":
        return LOOP.run(self._table.fetch_blob_files(column, row_ids))

    @property
    def tags(self) -> Tags:
        """Tag management for the table.

        Similar to Git, tags are a way to add metadata to a specific version of the
        table.

        .. warning::

            Tagged versions are exempted from the :py:meth:`cleanup_old_versions()`
            process.

            To remove a version that has been tagged, you must first
            :py:meth:`~Tags.delete` the associated tag.

        Returns
        -------
        Tags
            The tag manager for managing tags for the table.

        Examples
        --------
        >>> import lancedb
        >>> db = lancedb.connect("./.lancedb")
        >>> table = db.create_table("my_table",
        ...    [{"vector": [1.1, 0.9], "type": "vector"}])
        >>> table.tags.create("v1", table.version)
        >>> table.add([{"vector": [0.5, 0.2], "type": "vector"}])
        AddResult(version=2)
        >>> tags = table.tags.list()
        >>> print(tags["v1"]["version"])
        1
        >>> table.checkout("v1")
        >>> table.to_pandas()
               vector    type
        0  [1.1, 0.9]  vector
        """
        return Tags(self._table)

    @property
    def branches(self) -> "Branches":
        """Branch management for the table.

        ``create``/``checkout`` return a new table handle scoped to the branch;
        writes on it do not affect ``main``.
        """
        return Branches(self)

    def current_branch(self) -> Optional[str]:
        """The branch this table handle is scoped to, or ``None`` for ``main``."""
        return self._table.current_branch()

    def _wrap_branch_handle(
        self, async_table: "AsyncTable", version: Optional[int] = None
    ) -> "LanceTable":
        # version is unused locally: the pin already lives on async_table and a
        # local handle is not reopened via a serialized connection.
        return LanceTable(
            self._conn,
            async_table.name,
            namespace_path=self._namespace_path,
            namespace_client=self._namespace_client,
            pushdown_operations=self._pushdown_operations,
            route_pushdown_to_rust=self._route_pushdown_to_rust,
            location=self._location,
            _async=async_table,
        )

    def checkout(self, version: Union[int, str]):
        """Checkout a version of the table. This is an in-place operation.

        This allows viewing previous versions of the table. If you wish to
        keep writing to the dataset starting from an old version, then use
        the `restore` function.

        Calling this method will set the table into time-travel mode. If you
        wish to return to standard mode, call `checkout_latest`.

        Parameters
        ----------
        version: int | str,
            The version to check out. A version number (`int`) or a tag
            (`str`) can be provided.

        Examples
        --------
        >>> import lancedb
        >>> db = lancedb.connect("./.lancedb")
        >>> table = db.create_table("my_table",
        ...    [{"vector": [1.1, 0.9], "type": "vector"}])
        >>> table.version
        1
        >>> table.to_pandas()
               vector    type
        0  [1.1, 0.9]  vector
        >>> table.add([{"vector": [0.5, 0.2], "type": "vector"}])
        AddResult(version=2)
        >>> table.version
        2
        >>> table.checkout(1)
        >>> table.to_pandas()
               vector    type
        0  [1.1, 0.9]  vector
        """
        LOOP.run(self._table.checkout(version))

    def checkout_latest(self):
        """Checkout the latest version of the table. This is an in-place operation.

        The table will be set back into standard mode, and will track the latest
        version of the table.
        """
        LOOP.run(self._table.checkout_latest())

    def restore(self, version: Optional[Union[int, str]] = None):
        """Restore a version of the table. This is an in-place operation.

        This creates a new version where the data is equivalent to the
        specified previous version. Data is not copied (as of python-v0.2.1).

        Parameters
        ----------
        version : int or str, default None
            The version number or version tag to restore.
            If unspecified then restores the currently checked out version.
            If the currently checked out version is the
            latest version then this is a no-op.

        Examples
        --------
        >>> import lancedb
        >>> db = lancedb.connect("./.lancedb")
        >>> table = db.create_table("my_table", [
        ...     {"vector": [1.1, 0.9], "type": "vector"}])
        >>> table.version
        1
        >>> table.to_pandas()
               vector    type
        0  [1.1, 0.9]  vector
        >>> table.add([{"vector": [0.5, 0.2], "type": "vector"}])
        AddResult(version=2)
        >>> table.version
        2
        >>> table.tags.create("v2", 2)
        >>> table.restore(1)
        >>> table.to_pandas()
               vector    type
        0  [1.1, 0.9]  vector
        >>> len(table.list_versions())
        3
        >>> table.restore("v2")
        >>> table.to_pandas()
               vector    type
        0  [1.1, 0.9]  vector
        1  [0.5, 0.2]  vector
        >>> len(table.list_versions())
        4
        """
        if version is not None:
            LOOP.run(self._table.checkout(version))
        LOOP.run(self._table.restore())

    def count_rows(self, filter: Optional[str] = None) -> int:
        return LOOP.run(self._table.count_rows(filter))

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}(name={self.name!r}, _conn={self._conn!r})"

    def __str__(self) -> str:
        return self.__repr__()

    def head(self, n=5) -> pa.Table:
        """Return the first n rows of the table."""
        return LOOP.run(self._table.head(n))

    def to_pandas(self, blob_mode: BlobMode = "lazy", **kwargs) -> "pd.DataFrame":
        """Return the table as a pandas DataFrame.

        Parameters
        ----------
        blob_mode: str, default "lazy"
            Controls how Lance blob columns are returned.
        **kwargs
            Forwarded to Lance pandas conversion.

        Returns
        -------
        pd.DataFrame
        """
        validate_blob_mode(blob_mode)
        if blob_mode == "descriptions" or not schema_has_blob_field(self.schema):
            arrow_tbl = self.to_arrow()
            if blob_mode == "descriptions":
                arrow_tbl = strip_auto_row_ids(
                    arrow_tbl, row_addressable_blob_v2_paths(self.schema)
                )
            return arrow_tbl.to_pandas(**kwargs)

        if (
            blob_mode == "lazy"
            and self._namespace_client is None
            and get_uri_scheme(self._dataset_path) == "memory"
        ):
            return self.to_arrow().to_pandas(**kwargs)

        if blob_mode == "bytes" and blob_v2_column_paths(self.schema):
            return self.search().to_pandas(blob_mode=blob_mode, **kwargs)

        return self.to_lance().to_pandas(blob_mode=blob_mode, **kwargs)

    def to_arrow(self) -> pa.Table:
        """Return the table as a pyarrow Table.

        Returns
        -------
        pa.Table"""
        if (
            _should_push_down_query_table(
                self._namespace_client, self._pushdown_operations
            )
            and not self._route_pushdown_to_rust
        ):
            return self._execute_query(Query()).read_all()

        return LOOP.run(self._table.to_arrow())

    def to_polars(self, batch_size=None) -> "pl.LazyFrame":
        """Return the table as a polars LazyFrame.

        Parameters
        ----------
        batch_size: int, optional
            Passed to polars. This is the maximum row count for
            scanned pyarrow record batches

        Note
        ----
        1. This requires polars to be installed separately
        2. Currently we've disabled push-down of the filters from polars
           because polars pushdown into pyarrow uses pyarrow compute
           expressions rather than SQl strings (which LanceDB supports)
        3. The Polars streaming engine is not supported because it does not
           currently implement Python PyArrow dataset scans. Use the default
           engine when collecting this LazyFrame.

        Returns
        -------
        pl.LazyFrame
        """
        from lancedb.integrations.pyarrow import PyarrowDatasetAdapter

        dataset = PyarrowDatasetAdapter(self)
        # Polars 1.32's non-PyArrow callback path passes batch_size twice.  Keep
        # the compatible PyArrow path, but block predicates because this adapter
        # cannot translate PyArrow expressions into LanceDB filters.
        return pl.scan_pyarrow_dataset(dataset, batch_size=batch_size).map_batches(
            _polars_predicate_pushdown_barrier,
            predicate_pushdown=False,
        )

    # New unified API overload
    @overload
    def create_index(
        self,
        column: str,
        /,
        *,
        config: IndexConfigType,
        replace: bool = ...,
        wait_timeout: Optional[timedelta] = ...,
        name: Optional[str] = ...,
        train: bool = ...,
    ) -> None: ...

    # Legacy API overload (deprecated)
    @overload
    def create_index(
        self,
        metric: Literal["l2", "cosine", "dot", "hamming"] = ...,
        num_partitions: Optional[int] = ...,
        num_sub_vectors: Optional[int] = ...,
        vector_column_name: str = ...,
        replace: bool = ...,
        accelerator: Optional[str] = ...,
        index_cache_size: Optional[int] = ...,
        num_bits: int = ...,
        index_type: Literal[
            "IVF_FLAT", "IVF_SQ", "IVF_PQ", "IVF_RQ", "IVF_HNSW_SQ", "IVF_HNSW_PQ"
        ] = ...,
        max_iterations: int = ...,
        sample_rate: int = ...,
        m: int = ...,
        ef_construction: int = ...,
        *,
        wait_timeout: Optional[timedelta] = ...,
        name: Optional[str] = ...,
        train: bool = ...,
        target_partition_size: Optional[int] = ...,
    ) -> None: ...

    def create_index(
        self,
        metric: str = "l2",
        num_partitions: Optional[int] = None,
        num_sub_vectors: Optional[int] = None,
        vector_column_name: str = VECTOR_COLUMN_NAME,
        replace: bool = True,
        accelerator: Optional[str] = None,
        index_cache_size: Optional[int] = None,
        num_bits: int = 8,
        index_type: Literal[
            "IVF_FLAT",
            "IVF_SQ",
            "IVF_PQ",
            "IVF_RQ",
            "IVF_HNSW_SQ",
            "IVF_HNSW_PQ",
            "IVF_HNSW_FLAT",
        ] = "IVF_PQ",
        max_iterations: int = 50,
        sample_rate: int = 256,
        m: int = 20,
        ef_construction: int = 300,
        *,
        config: Optional[IndexConfigType] = None,
        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
        train: bool = True,
        target_partition_size: Optional[int] = None,
    ):
        """Create an index on a column.

        This method supports both the new unified API and the legacy API
        for backwards compatibility. The new API takes the column name as the
        first positional argument and an index configuration object via
        ``config``; the legacy API takes the distance metric as the first
        argument plus separate ``vector_column_name`` / ``num_partitions`` /
        etc. parameters, and emits a ``DeprecationWarning``.

        Parameters
        ----------
        metric : str
            For new API: the column name to index.
            For legacy API: the distance metric ("l2", "cosine", "dot", "hamming").
        config : IndexConfigType, optional
            The index configuration object. If provided, uses the new unified API.
            Can be one of: IvfFlat, IvfPq, IvfSq, IvfRq, HnswPq, HnswSq,
            BTree, Bitmap, LabelList, Fm, FTS.
        replace : bool, default True
            Whether to replace an existing index on this column.
        wait_timeout : timedelta, optional
            Timeout to wait for async indexing to complete.
        name : str, optional
            Custom name for the index.
        train : bool, default True
            Whether to train the index with existing data.

        Examples
        --------
        New API (recommended):

        >>> table.create_index(  # doctest: +SKIP
        ...     "vector", config=IvfPq(distance_type="l2")
        ... )
        >>> table.create_index("category", config=BTree())  # doctest: +SKIP
        >>> table.create_index("content", config=FTS())  # doctest: +SKIP

        Legacy API (deprecated):

        >>> table.create_index(  # doctest: +SKIP
        ...     "l2", vector_column_name="vector"
        ... )
        """
        # Detect whether this is a legacy API call
        is_legacy = self._is_legacy_create_index_call(
            metric,
            config,
            num_partitions,
            num_sub_vectors,
            vector_column_name,
            accelerator,
            index_cache_size,
        )

        if is_legacy:
            warnings.warn(
                "The create_index() API with metric/num_partitions parameters is "
                "deprecated and will be removed in a future version. "
                "Please migrate to the new unified API:\n"
                "  # Old (deprecated):\n"
                "  table.create_index('l2', vector_column_name='my_vector')\n"
                "  # New (recommended):\n"
                "  table.create_index('my_vector', config=IvfPq(distance_type='l2'))",
                DeprecationWarning,
                stacklevel=2,
            )

            # Legacy API: first arg is the distance metric
            column = vector_column_name

            # Build config from legacy parameters
            config = self._build_vector_config_from_legacy_params(
                metric=metric,
                index_type=index_type,
                num_partitions=num_partitions,
                num_sub_vectors=num_sub_vectors,
                num_bits=num_bits,
                max_iterations=max_iterations,
                sample_rate=sample_rate,
                m=m,
                ef_construction=ef_construction,
                target_partition_size=target_partition_size,
                accelerator=accelerator,
            )

            # Handle accelerator through pylance
            if accelerator is not None:
                self.to_lance().create_index(
                    column=column,
                    index_type=index_type,
                    metric=metric,
                    num_partitions=num_partitions,
                    num_sub_vectors=num_sub_vectors,
                    replace=replace,
                    accelerator=accelerator,
                    index_cache_size=index_cache_size,
                    num_bits=num_bits,
                    m=m,
                    ef_construction=ef_construction,
                    target_partition_size=target_partition_size,
                )
                self.checkout_latest()
                return
        else:
            # New API: metric is the column name
            column = metric

            # Check if config has accelerator set and dispatch to pylance
            if config is not None and hasattr(config, "accelerator"):
                acc = getattr(config, "accelerator", None)
                if acc is not None:
                    # Dispatch to pylance for GPU acceleration
                    index_type_map = {
                        "IvfFlat": "IVF_FLAT",
                        "IvfSq": "IVF_SQ",
                        "IvfPq": "IVF_PQ",
                        "IvfRq": "IVF_RQ",
                        "HnswPq": "IVF_HNSW_PQ",
                        "HnswSq": "IVF_HNSW_SQ",
                    }
                    cfg_type = type(config).__name__
                    lance_index_type = index_type_map.get(cfg_type, "IVF_PQ")

                    self.to_lance().create_index(
                        column=column,
                        index_type=lance_index_type,
                        metric=getattr(config, "distance_type", "l2"),
                        num_partitions=getattr(config, "num_partitions", None),
                        num_sub_vectors=getattr(config, "num_sub_vectors", None),
                        replace=replace,
                        accelerator=acc,
                        num_bits=getattr(config, "num_bits", 8),
                        m=getattr(config, "m", 20),
                        ef_construction=getattr(config, "ef_construction", 300),
                        target_partition_size=getattr(
                            config, "target_partition_size", None
                        ),
                    )
                    self.checkout_latest()
                    return

        return LOOP.run(
            self._table.create_index(
                column,
                replace=replace,
                config=config,
                wait_timeout=wait_timeout,
                name=name,
                train=train,
            )
        )

    def create_index_async(
        self,
        column: str,
        *,
        config: IndexConfigType,
        replace: Optional[bool] = None,
        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
        train: bool = True,
    ) -> Job:
        """Create an index, returning a handle to the indexing job.

        The job may already be complete when returned; callers must not assume
        the index exists until :meth:`Job.wait` returns.
        """
        return Job(
            LOOP.run(
                self._table.create_index_async(
                    column,
                    replace=replace,
                    config=config,
                    wait_timeout=wait_timeout,
                    name=name,
                    train=train,
                )
            )
        )

    def _is_legacy_create_index_call(
        self,
        first_arg: str,
        config: Optional[IndexConfigType],
        num_partitions: Optional[int],
        num_sub_vectors: Optional[int],
        vector_column_name: str,
        accelerator: Optional[str],
        index_cache_size: Optional[int],
    ) -> bool:
        """Detect if this is a legacy create_index call."""
        # If config is provided, it's definitely the new API
        if config is not None:
            return False

        # If old-style parameters were explicitly set, it's legacy
        if any(
            x is not None
            for x in (num_partitions, num_sub_vectors, accelerator, index_cache_size)
        ):
            return True

        # If vector_column_name differs from default, it's legacy
        if vector_column_name != VECTOR_COLUMN_NAME:
            return True

        # If first arg is a known metric, assume legacy
        if first_arg.lower() in KNOWN_METRICS:
            return True

        # Otherwise assume new API
        return False

    def _build_vector_config_from_legacy_params(
        self,
        metric: str,
        index_type: str,
        num_partitions: Optional[int],
        num_sub_vectors: Optional[int],
        num_bits: int,
        max_iterations: int,
        sample_rate: int,
        m: int,
        ef_construction: int,
        target_partition_size: Optional[int],
        accelerator: Optional[str],
    ) -> IndexConfigType:
        """Build an index config object from legacy parameters."""
        if index_type == "IVF_FLAT":
            return IvfFlat(
                distance_type=metric,
                num_partitions=num_partitions,
                max_iterations=max_iterations,
                sample_rate=sample_rate,
                target_partition_size=target_partition_size,
                accelerator=accelerator,
            )
        elif index_type == "IVF_SQ":
            return IvfSq(
                distance_type=metric,
                num_partitions=num_partitions,
                max_iterations=max_iterations,
                sample_rate=sample_rate,
                target_partition_size=target_partition_size,
                accelerator=accelerator,
            )
        elif index_type == "IVF_PQ":
            return IvfPq(
                distance_type=metric,
                num_partitions=num_partitions,
                num_sub_vectors=num_sub_vectors,
                num_bits=num_bits,
                max_iterations=max_iterations,
                sample_rate=sample_rate,
                target_partition_size=target_partition_size,
                accelerator=accelerator,
            )
        elif index_type == "IVF_RQ":
            return IvfRq(
                distance_type=metric,
                num_partitions=num_partitions,
                num_bits=num_bits,
                max_iterations=max_iterations,
                sample_rate=sample_rate,
                target_partition_size=target_partition_size,
                accelerator=accelerator,
            )
        elif index_type == "IVF_HNSW_PQ":
            return HnswPq(
                distance_type=metric,
                num_partitions=num_partitions,
                num_sub_vectors=num_sub_vectors,
                num_bits=num_bits,
                max_iterations=max_iterations,
                sample_rate=sample_rate,
                m=m,
                ef_construction=ef_construction,
                target_partition_size=target_partition_size,
                accelerator=accelerator,
            )
        elif index_type == "IVF_HNSW_SQ":
            return HnswSq(
                distance_type=metric,
                num_partitions=num_partitions,
                max_iterations=max_iterations,
                sample_rate=sample_rate,
                m=m,
                ef_construction=ef_construction,
                target_partition_size=target_partition_size,
                accelerator=accelerator,
            )
        elif index_type == "IVF_HNSW_FLAT":
            return HnswFlat(
                distance_type=metric,
                num_partitions=num_partitions,
                max_iterations=max_iterations,
                sample_rate=sample_rate,
                m=m,
                ef_construction=ef_construction,
                target_partition_size=target_partition_size,
            )
        else:
            raise ValueError(f"Unknown index type {index_type}")

    def drop_index(self, name: str) -> None:
        """
        Drops an index from the table

        Parameters
        ----------
        name: str
            The name of the index to drop
        """
        return LOOP.run(self._table.drop_index(name))

    def prewarm_index(self, name: str) -> None:
        """
        Prewarm an index in the table.

        This is a hint to the database that the index will be accessed in the
        future and should be loaded into memory if possible.  This can reduce
        cold-start latency for subsequent queries.

        This call initiates prewarming and returns once the request is accepted.
        It is idempotent and safe to call from multiple clients concurrently.

        It is generally wasteful to call this if the index does not fit into the
        available cache.  Not all index types support prewarming; unsupported
        indices will silently ignore the request.

        Parameters
        ----------
        name: str
            The name of the index to prewarm
        """
        return LOOP.run(self._table.prewarm_index(name))

    def prewarm_data(self, columns: Optional[List[str]] = None) -> None:
        """
        Prewarm data for the table.

        This is a hint to the database that the given columns will be accessed
        in the future and the database should prefetch the data if possible.
        Currently only supported on remote tables.

        This call initiates prewarming and returns once the request is accepted.
        It is idempotent and safe to call from multiple clients concurrently.

        This operation has a large upfront cost but can speed up future queries
        that need to fetch the given columns.  Large columns such as embeddings
        or binary data may not be practical to prewarm.  This feature is intended
        for workloads that issue many queries against the same columns.

        Parameters
        ----------
        columns: list of str, optional
            The columns to prewarm. If None, all columns are prewarmed.
        """
        return LOOP.run(self._table.prewarm_data(columns))

    def wait_for_index(
        self, index_names: Iterable[str], timeout: timedelta = timedelta(seconds=300)
    ) -> None:
        return LOOP.run(self._table.wait_for_index(index_names, timeout))

    def stats(self) -> TableStatistics:
        return LOOP.run(self._table.stats())

    @property
    def uri(self) -> str:
        return LOOP.run(self._table.uri())

    def initial_storage_options(self) -> Optional[Dict[str, str]]:
        """Get the initial storage options that were passed in when opening this table.

        For dynamically refreshed options (e.g., credential vending), use
        :meth:`latest_storage_options`.

        Warning: This is an internal API and the return value is subject to change.

        Returns
        -------
        Optional[Dict[str, str]]
            The storage options, or None if no storage options were configured.
        """
        return LOOP.run(self._table.initial_storage_options())

    def latest_storage_options(self) -> Optional[Dict[str, str]]:
        """Get the latest storage options, refreshing from provider if configured.

        This method is useful for credential vending scenarios where storage options
        may be refreshed dynamically. If no dynamic provider is configured, this
        returns the initial static options.

        Warning: This is an internal API and the return value is subject to change.

        Returns
        -------
        Optional[Dict[str, str]]
            The storage options, or None if no storage options were configured.
        """
        return LOOP.run(self._table.latest_storage_options())

    @deprecation.deprecated(
        deprecated_in="0.25.0",
        current_version=__version__,
        details="Use create_index() with config=BTree()/Bitmap()/LabelList() instead.",
    )
    def create_scalar_index(
        self,
        column: str,
        *,
        replace: bool = True,
        index_type: ScalarIndexType = "BTREE",
        name: Optional[str] = None,
    ):
        """Create a scalar index on a column.

        .. deprecated:: 0.25.0
            Use :meth:`create_index` with a BTree, Bitmap, or LabelList config instead.
            Example: ``table.create_index("column", config=BTree())``
        """
        if index_type == "BTREE":
            config = BTree()
        elif index_type == "BITMAP":
            config = Bitmap()
        elif index_type == "LABEL_LIST":
            config = LabelList()
        else:
            raise ValueError(f"Unknown index type {index_type}")
        return LOOP.run(
            self._table.create_index(column, replace=replace, config=config, name=name)
        )

    @deprecation.deprecated(
        deprecated_in="0.25.0",
        current_version=__version__,
        details="Use create_index() with config=FTS() instead.",
    )
    def create_fts_index(
        self,
        field_names: Union[str, List[str]],
        *,
        ordering_field_names: Optional[Union[str, List[str]]] = None,
        replace: bool = False,
        writer_heap_size: Optional[int] = 1024 * 1024 * 1024,
        use_tantivy: bool = False,
        tokenizer_name: Optional[str] = None,
        with_position: bool = False,
        # tokenizer configs:
        base_tokenizer: BaseTokenizerType = "simple",
        language: str = "English",
        max_token_length: Optional[int] = 40,
        lower_case: bool = True,
        stem: bool = True,
        remove_stop_words: bool = True,
        custom_stop_words: Optional[List[str]] = None,
        ascii_folding: bool = True,
        ngram_min_length: int = 3,
        ngram_max_length: int = 3,
        prefix_only: bool = False,
        block_size: int = 128,
        document_granularity: DocumentGranularity = DocumentGranularity.ROW,
        name: Optional[str] = None,
    ):
        """Create a full-text search index on a column.

        .. deprecated:: 0.25.0
            Use :meth:`create_index` with an FTS config instead.
            Example: ``table.create_index("text_column", config=FTS())``
        """
        self._ensure_no_legacy_fts_index()

        if use_tantivy:
            raise ValueError(
                "Tantivy-based FTS has been removed. "
                "Remove use_tantivy and recreate the index with native FTS."
            )
        if ordering_field_names is not None:
            raise ValueError(
                "ordering_field_names was only supported by the removed "
                "Tantivy-based FTS implementation."
            )
        if writer_heap_size != 1024 * 1024 * 1024:
            raise ValueError(
                "writer_heap_size was only supported by the removed "
                "Tantivy-based FTS implementation."
            )
        if not isinstance(field_names, str):
            raise ValueError(
                "Native FTS indexes can only be created on a single field "
                "at a time. To search over multiple text fields, create a "
                "separate FTS index for each field."
            )

        if tokenizer_name is None:
            tokenizer_configs = {
                "base_tokenizer": base_tokenizer,
                "language": language,
                "with_position": with_position,
                "max_token_length": max_token_length,
                "lower_case": lower_case,
                "stem": stem,
                "remove_stop_words": remove_stop_words,
                "custom_stop_words": custom_stop_words,
                "ascii_folding": ascii_folding,
                "ngram_min_length": ngram_min_length,
                "ngram_max_length": ngram_max_length,
                "prefix_only": prefix_only,
            }
        else:
            tokenizer_configs = self.infer_tokenizer_configs(tokenizer_name)
            tokenizer_configs["custom_stop_words"] = custom_stop_words

        config = FTS(
            block_size=block_size,
            document_granularity=document_granularity,
            **tokenizer_configs,
        )

        try:
            LOOP.run(
                self._table.create_index(
                    field_names,
                    replace=replace,
                    config=config,
                    name=name,
                )
            )
        except (ValueError, RuntimeError) as e:
            _maybe_add_fts_error_note(
                e,
                base_tokenizer=config.base_tokenizer,
                language=config.language,
            )
            raise e

    @staticmethod
    def infer_tokenizer_configs(tokenizer_name: str) -> dict:
        if tokenizer_name == "default":
            return {
                "base_tokenizer": "simple",
                "language": "English",
                "max_token_length": 40,
                "lower_case": True,
                "stem": False,
                "remove_stop_words": False,
                "ascii_folding": False,
                "ngram_min_length": 3,
                "ngram_max_length": 3,
                "prefix_only": False,
            }
        elif tokenizer_name == "raw":
            return {
                "base_tokenizer": "raw",
                "language": "English",
                "max_token_length": None,
                "lower_case": False,
                "stem": False,
                "remove_stop_words": False,
                "ascii_folding": False,
                "ngram_min_length": 3,
                "ngram_max_length": 3,
                "prefix_only": False,
            }
        elif tokenizer_name == "whitespace":
            return {
                "base_tokenizer": "whitespace",
                "language": "English",
                "max_token_length": None,
                "lower_case": False,
                "stem": False,
                "remove_stop_words": False,
                "ascii_folding": False,
                "ngram_min_length": 3,
                "ngram_max_length": 3,
                "prefix_only": False,
            }

        # or it's with language stemming with pattern like "en_stem"
        if len(tokenizer_name) != 7:
            raise ValueError(f"Invalid tokenizer name {tokenizer_name}")
        lang = tokenizer_name[:2]
        if tokenizer_name[-5:] != "_stem":
            raise ValueError(f"Invalid tokenizer name {tokenizer_name}")
        if lang not in lang_mapping:
            raise ValueError(f"Invalid language code {lang}")
        return {
            "base_tokenizer": "simple",
            "language": lang_mapping[lang],
            "max_token_length": 40,
            "lower_case": True,
            "stem": True,
            "remove_stop_words": False,
            "ascii_folding": False,
            "ngram_min_length": 3,
            "ngram_max_length": 3,
            "prefix_only": False,
        }

    def add(
        self,
        data: DATA,
        mode: AddMode = "append",
        on_bad_vectors: OnBadVectorsType = "error",
        fill_value: float = 0.0,
        progress: Optional[Union[bool, Callable, Any]] = None,
        write_parallelism: Optional[int] = None,
        allow_external_blob_outside_bases: bool = False,
    ) -> AddResult:
        """Add data to the table.
        If vector columns are missing and the table
        has embedding functions, then the vector columns
        are automatically computed and added.

        Parameters
        ----------
        data: list-of-dict, pd.DataFrame
            The data to insert into the table.
        mode: str
            The mode to use when writing the data. Valid values are
            "append" and "overwrite".
        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".
        fill_value: float, default 0.
            The value to use when filling vectors. Only used if on_bad_vectors="fill".
        progress: bool, callable, or tqdm-like, optional
            A callback or tqdm-compatible progress bar. See
            :meth:`Table.add` for details.
        write_parallelism: int, optional
            Number of partitions to write in parallel. Higher values increase
            throughput but also peak memory use, since each partition buffers
            data in flight. Defaults to an estimate based on the data size,
            capped at the number of CPU cores. Lower this if bulk ingestion is
            using too much memory.
        allow_external_blob_outside_bases: bool, default False
            Allow blob URIs outside registered bases. See :meth:`Table.add`.
            Local tables only.

        Returns
        -------
        int
            The number of vectors in the table.
        """
        progress, owns = _normalize_progress(progress)
        try:
            return LOOP.run(
                self._table.add(
                    data,
                    mode=mode,
                    on_bad_vectors=on_bad_vectors,
                    fill_value=fill_value,
                    progress=progress,
                    write_parallelism=write_parallelism,
                    allow_external_blob_outside_bases=allow_external_blob_outside_bases,
                )
            )
        finally:
            if owns:
                progress.close()

    def merge(
        self,
        other_table: Union[LanceTable, DATA],
        left_on: str,
        right_on: Optional[str] = None,
        schema: Optional[Union[pa.Schema, LanceModel]] = None,
    ):
        """Merge another table into this table.

        Performs a left join, where the dataset is the left side and other_table
        is the right side. Rows existing in the dataset but not on the left will
        be filled with null values, unless Lance doesn't support null values for
        some types, in which case an error will be raised. The only overlapping
        column allowed is the join column. If other overlapping columns exist,
        an error will be raised.

        Parameters
        ----------
        other_table: LanceTable or Reader-like
            The data to be merged. Acceptable types are:
            - Pandas DataFrame, Pyarrow Table, Dataset, Scanner,
            Iterator[RecordBatch], or RecordBatchReader
            - LanceTable
        left_on: str
            The name of the column in the dataset to join on.
        right_on: str or None
            The name of the column in other_table to join on. If None, defaults to
            left_on.
        schema: pa.Schema or LanceModel, optional
            The schema of the other_table.
            If not provided, the schema is inferred from the data.

        Examples
        --------
        >>> import lancedb
        >>> import pyarrow as pa
        >>> df = pa.table({'x': [1, 2, 3], 'y': ['a', 'b', 'c']})
        >>> db = lancedb.connect("./.lancedb")
        >>> table = db.create_table("dataset", df)
        >>> table.to_pandas()
           x  y
        0  1  a
        1  2  b
        2  3  c
        >>> new_df = pa.table({'x': [1, 2, 3], 'z': ['d', 'e', 'f']})
        >>> table.merge(new_df, 'x')
        >>> table.to_pandas()
           x  y  z
        0  1  a  d
        1  2  b  e
        2  3  c  f
        """
        if isinstance(other_table, LanceTable):
            other_table = other_table.to_lance()
        else:
            other_table = _sanitize_data(
                other_table,
                schema,
            )
        self.to_lance().merge(
            other_table, left_on=left_on, right_on=right_on, schema=schema
        )
        self.checkout_latest()

    @cached_property
    def embedding_functions(self) -> Dict[str, EmbeddingFunctionConfig]:
        """
        Get the embedding functions for the table

        Returns
        -------
        funcs: Dict[str, EmbeddingFunctionConfig]
            A mapping of the vector column to the embedding function
            or empty dict if not configured.
        """
        return EmbeddingFunctionRegistry.get_instance().parse_functions(
            self.schema.metadata
        )

    @overload
    def search(  # type: ignore
        self,
        query: Optional[Union[VEC, str, "PIL.Image.Image", Tuple]] = None,
        vector_column_name: Optional[str] = None,
        query_type: Literal["vector"] = "vector",
        ordering_field_name: Optional[str] = None,
        fts_columns: Optional[Union[str, List[str]]] = None,
    ) -> LanceVectorQueryBuilder: ...

    @overload
    def search(
        self,
        query: Optional[Union[VEC, str, "PIL.Image.Image", Tuple]] = None,
        vector_column_name: Optional[str] = None,
        query_type: Literal["fts"] = "fts",
        ordering_field_name: Optional[str] = None,
        fts_columns: Optional[Union[str, List[str]]] = None,
    ) -> LanceFtsQueryBuilder: ...

    @overload
    def search(
        self,
        query: Optional[
            Union[VEC, str, "PIL.Image.Image", Tuple, FullTextQuery]
        ] = None,
        vector_column_name: Optional[str] = None,
        query_type: Literal["hybrid"] = "hybrid",
        ordering_field_name: Optional[str] = None,
        fts_columns: Optional[Union[str, List[str]]] = None,
    ) -> LanceHybridQueryBuilder: ...

    @overload
    def search(
        self,
        query: None = None,
        vector_column_name: Optional[str] = None,
        query_type: QueryType = "auto",
        ordering_field_name: Optional[str] = None,
        fts_columns: Optional[Union[str, List[str]]] = None,
    ) -> LanceEmptyQueryBuilder: ...

    def search(
        self,
        query: Optional[
            Union[VEC, str, "PIL.Image.Image", Tuple, FullTextQuery]
        ] = None,
        vector_column_name: Optional[str] = None,
        query_type: QueryType = "auto",
        ordering_field_name: Optional[str] = None,
        fts_columns: Optional[Union[str, List[str]]] = None,
    ) -> LanceQueryBuilder:
        """Create a search query to find the nearest neighbors
        of the given query vector. We currently support [vector search](https://lancedb.com/docs/search/vector-search/)
        and [full-text search](https://lancedb.com/docs/search/full-text-search/).

        Examples
        --------
        >>> import lancedb
        >>> db = lancedb.connect("./.lancedb")
        >>> data = [
        ...    {"original_width": 100, "caption": "bar", "vector": [0.1, 2.3, 4.5]},
        ...    {"original_width": 2000, "caption": "foo",  "vector": [0.5, 3.4, 1.3]},
        ...    {"original_width": 3000, "caption": "test", "vector": [0.3, 6.2, 2.6]}
        ... ]
        >>> table = db.create_table("my_table", data)
        >>> query = [0.4, 1.4, 2.4]
        >>> (table.search(query)
        ...     .where("original_width > 1000", prefilter=True)
        ...     .select(["caption", "original_width", "vector"])
        ...     .limit(2)
        ...     .to_pandas())
          caption  original_width           vector  _distance
        0     foo            2000  [0.5, 3.4, 1.3]   5.220000
        1    test            3000  [0.3, 6.2, 2.6]  23.089996

        Parameters
        ----------
        query: list/np.ndarray/str/PIL.Image.Image, default None
            The targetted vector to search for.

            - *default None*.
            Acceptable types are: list, np.ndarray, PIL.Image.Image

            - If None then the
            select/[where][lancedb.query.LanceQueryBuilder.where]/limit clauses
            are applied to filter the table
        vector_column_name: str, optional
            The name of the vector column to search.

            The vector column needs to be a pyarrow fixed size list type
            *default "vector"*

            - If not specified then the vector column is inferred from
            the table schema

            - If the table has multiple vector columns then the *vector_column_name*
            needs to be specified. Otherwise, an error is raised.
        query_type: str, default "auto"
            "vector", "fts", or "auto"
            If "auto" then the query type is inferred from the query;
            If `query` is a list/np.ndarray then the query type is "vector";
            If `query` is a PIL.Image.Image then either do vector search
            or raise an error if no corresponding embedding function is found.
            If the `query` is a string, then the query type is "vector" if the
            table has embedding functions, else the query type is "fts"
        fts_columns: str or list of str, default None
            The column(s) to search in for full-text search.
            If None then the search is performed on all indexed columns.
            For now, only one column can be searched at a time.

        Returns
        -------
        LanceQueryBuilder
            A query builder object representing the query.
            Once executed, the query returns selected columns, the vector,
            and also the "_distance" column which is the distance between the query
            vector and the returned vector.
        """
        if isinstance(query, FullTextQuery):
            query_type = "fts"
        vector_column_name = infer_vector_column_name(
            schema=self.schema,
            query_type=query_type,
            query=query,
            vector_column_name=vector_column_name,
        )

        return LanceQueryBuilder.create(
            self,
            query,
            query_type,
            vector_column_name=vector_column_name,
            ordering_field_name=ordering_field_name,
            fts_columns=fts_columns or [],
        )

    @classmethod
    def create(
        cls,
        db: LanceDBConnection,
        name: str,
        data: Optional[DATA] = None,
        schema: Optional[pa.Schema] = None,
        mode: CreateMode = "create",
        exist_ok: bool = False,
        on_bad_vectors: OnBadVectorsType = "error",
        fill_value: float = 0.0,
        embedding_functions: Optional[List[EmbeddingFunctionConfig]] = None,
        *,
        namespace_path: Optional[List[str]] = None,
        storage_options: Optional[Dict[str, str | bool]] = None,
        data_storage_version: Optional[str] = None,
        enable_v2_manifest_paths: Optional[bool] = None,
        location: Optional[str] = None,
        namespace_client: Optional[Any] = None,
        pushdown_operations: Optional[set] = None,
        route_pushdown_to_rust: bool = False,
    ):
        """
        Create a new table.

        Examples
        --------
        >>> import lancedb
        >>> data = [
        ...    {"x": 1, "vector": [1.0, 2]},
        ...    {"x": 2, "vector": [3.0, 4]},
        ...    {"x": 3, "vector": [5.0, 6]}
        ... ]
        >>> db = lancedb.connect("./.lancedb")
        >>> table = db.create_table("my_table", data)
        >>> table.to_pandas()
           x      vector
        0  1  [1.0, 2.0]
        1  2  [3.0, 4.0]
        2  3  [5.0, 6.0]

        Parameters
        ----------
        db: LanceDB
            The LanceDB instance to create the table in.
        name: str
            The name of the table to create.
        data: list-of-dict, dict, pd.DataFrame, default None
            The data to insert into the table.
            At least one of `data` or `schema` must be provided.
        schema: pa.Schema or LanceModel, optional
            The schema of the table. If not provided,
            the schema is inferred from the data.
            At least one of `data` or `schema` must be provided.
        mode: str, default "create"
            The mode to use when writing the data. Valid values are
            "create", "overwrite", and "append".
        exist_ok: bool, default False
            If the table already exists then raise an error if False,
            otherwise just open the table, it will not add the provided
            data but will validate against any schema that's specified.
        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".
        fill_value: float, default 0.
            The value to use when filling vectors. Only used if on_bad_vectors="fill".
        embedding_functions: list of EmbeddingFunctionModel, default None
            The embedding functions to use when creating the table.
        data_storage_version: optional, str, default "stable"
            Deprecated.  Set `storage_options` when connecting to the database and set
            `new_table_data_storage_version` in the options.
        enable_v2_manifest_paths: optional, bool, default False
            Deprecated.  Set `storage_options` when connecting to the database and set
            `new_table_enable_v2_manifest_paths` in the options.
        """
        if namespace_path is None:
            namespace_path = []
        self = cls.__new__(cls)
        self._conn = db
        self._namespace_path = namespace_path
        self._location = location
        self._namespace_client = namespace_client
        self._pushdown_operations = pushdown_operations or set()
        self._route_pushdown_to_rust = route_pushdown_to_rust

        if data_storage_version is not None:
            warnings.warn(
                "setting data_storage_version directly on create_table is deprecated. "
                "Use database_options instead.",
                DeprecationWarning,
                stacklevel=2,
            )
            if storage_options is None:
                storage_options = {}
            storage_options["new_table_data_storage_version"] = data_storage_version
        if enable_v2_manifest_paths is not None:
            warnings.warn(
                "setting enable_v2_manifest_paths directly on create_table is "
                "deprecated. Use database_options instead.",
                DeprecationWarning,
                stacklevel=2,
            )
            if storage_options is None:
                storage_options = {}
            storage_options["new_table_enable_v2_manifest_paths"] = (
                enable_v2_manifest_paths
            )

        self._table = LOOP.run(
            self._conn._conn.create_table(
                name,
                data,
                schema=schema,
                mode=mode,
                exist_ok=exist_ok,
                on_bad_vectors=on_bad_vectors,
                fill_value=fill_value,
                embedding_functions=embedding_functions,
                namespace_path=namespace_path,
                storage_options=storage_options,
                location=location,
                namespace_client=namespace_client,
            )
        )
        return self

    def delete(self, where: Union[str, Expr]) -> DeleteResult:
        predicate = where._inner if isinstance(where, Expr) else where
        return LOOP.run(self._table.delete(predicate))

    def update(
        self,
        where: Optional[Union[str, Expr]] = None,
        values: Optional[dict] = None,
        *,
        values_sql: Optional[Dict[str, str]] = None,
    ) -> UpdateResult:
        """
        This can be used to update zero to all rows depending on how many
        rows match the where clause.

        Parameters
        ----------
        where: str or [Expr][lancedb.expr.Expr], optional
            The filter condition. Can be a SQL string or a type-safe
            [Expr][lancedb.expr.Expr] built with [col][lancedb.expr.col] and
            [lit][lancedb.expr.lit]. The filter must not be empty, or it will
            error.
        values: dict, optional
            The values to update. The keys are the column names and the values
            are the values to set.
        values_sql: dict, optional
            The values to update, expressed as SQL expression strings. These can
            reference existing columns. For example, {"x": "x + 1"} will increment
            the x column by 1.

        Returns
        -------
        UpdateResult
            - rows_updated: The number of rows that were updated
            - version: The new version number of the table after the update

        Examples
        --------
        >>> import lancedb
        >>> from lancedb.expr import col
        >>> import pandas as pd
        >>> data = pd.DataFrame({"x": [1, 2, 3], "vector": [[1.0, 2], [3, 4], [5, 6]]})
        >>> db = lancedb.connect("./.lancedb")
        >>> table = db.create_table("my_table", data)
        >>> table.to_pandas()
           x      vector
        0  1  [1.0, 2.0]
        1  2  [3.0, 4.0]
        2  3  [5.0, 6.0]
        >>> table.update(where=col("x") == 2, values={"vector": [10.0, 10]})
        UpdateResult(rows_updated=1, version=2)
        >>> table.to_pandas()
           x        vector
        0  1    [1.0, 2.0]
        1  3    [5.0, 6.0]
        2  2  [10.0, 10.0]

        """
        return LOOP.run(self._table.update(values, where=where, updates_sql=values_sql))

    def _execute_query(
        self,
        query: Query,
        *,
        batch_size: Optional[int] = None,
        timeout: Optional[timedelta] = None,
    ) -> pa.RecordBatchReader:
        # Branch queries run locally: the server-side query protocol can't
        # carry a branch yet.
        # TODO: push down server-side once it can (with remote table support).
        if (
            _should_push_down_query_table(
                self._namespace_client, self._pushdown_operations
            )
            and not self._route_pushdown_to_rust
            and self.current_branch() is None
            and query.take_offsets is None
        ):
            from lancedb.namespace import _execute_server_side_query

            table_id = self._namespace_path + [self.name]
            return _execute_server_side_query(self._namespace_client, table_id, query)

        async_iter = LOOP.run(
            self._table._execute_query(query, batch_size=batch_size, timeout=timeout)
        )

        def iter_sync():
            try:
                while True:
                    yield LOOP.run(async_iter.__anext__())
            except StopAsyncIteration:
                return

        return pa.RecordBatchReader.from_batches(async_iter.schema, iter_sync())

    def _explain_plan(self, query: Query, verbose: Optional[bool] = False) -> str:
        return LOOP.run(self._table._explain_plan(query, verbose))

    def _analyze_plan(
        self,
        query: Query,
        *,
        distributed_metrics: AnalyzePlanDistributedMetrics = "aggregate",
    ) -> str:
        return LOOP.run(
            self._table._analyze_plan(query, distributed_metrics=distributed_metrics)
        )

    def _output_schema(self, query: Query) -> pa.Schema:
        return LOOP.run(self._table._output_schema(query))

    def _do_merge(
        self,
        merge: LanceMergeInsertBuilder,
        new_data: DATA,
        on_bad_vectors: OnBadVectorsType,
        fill_value: float,
    ) -> MergeResult:
        return LOOP.run(
            self._table._do_merge(merge, new_data, on_bad_vectors, fill_value)
        )

    @property
    def _inner(self) -> LanceDBTable:
        return self._table._inner

    @deprecation.deprecated(
        deprecated_in="0.21.0",
        current_version=__version__,
        details="Use `Table.optimize` instead.",
    )
    def cleanup_old_versions(
        self,
        older_than: Optional[timedelta] = None,
        *,
        delete_unverified: bool = False,
    ) -> "CleanupStats":
        """
        Clean up old versions of the table, freeing disk space.

        Parameters
        ----------
        older_than: timedelta, default None
            The minimum age of the version to delete. If None, then this defaults
            to two weeks.
        delete_unverified: bool, default False
            Because they may be part of an in-progress transaction, files newer
            than 7 days old are not deleted by default. If you are sure that
            there are no in-progress transactions, then you can set this to True
            to delete all files older than `older_than`.

        Returns
        -------
        CleanupStats
            The stats of the cleanup operation, including how many bytes were
            freed.
        """
        return self.to_lance().cleanup_old_versions(
            older_than, delete_unverified=delete_unverified
        )

    @deprecation.deprecated(
        deprecated_in="0.21.0",
        current_version=__version__,
        details="Use `Table.optimize` instead.",
    )
    def compact_files(self, *args, **kwargs) -> CompactionStats:
        """
        Run the compaction process on the table.

        This can be run after making several small appends to optimize the table
        for faster reads.

        Arguments are passed onto `lance.dataset.DatasetOptimizer.compact_files`.
         (see Lance documentation for more details) For most cases, the default
        should be fine.
        """
        stats = self.to_lance().optimize.compact_files(*args, **kwargs)
        self.checkout_latest()
        return stats

    def optimize(
        self,
        *,
        cleanup_older_than: Optional[timedelta] = None,
        delete_unverified: bool = False,
        retrain: bool = False,
    ):
        """
        Optimize the on-disk data and indices for better performance.

        Modeled after ``VACUUM`` in PostgreSQL.

        Optimization covers three operations:

         * Compaction: Merges small files into larger ones
         * Prune: Removes old versions of the dataset
         * Index: Optimizes the indices, adding new data to existing indices

        Parameters
        ----------
        cleanup_older_than: timedelta, optional default 7 days
            All files belonging to versions older than this will be removed.  Set
            to 0 days to remove all versions except the latest.  The latest version
            is never removed.
        delete_unverified: bool, default False
            Files leftover from a failed transaction may appear to be part of an
            in-progress operation (e.g. appending new data) and these files will not
            be deleted unless they are at least 7 days old. If delete_unverified is True
            then these files will be deleted regardless of their age.

            .. warning::

                This should only be set to True if you can guarantee that no other
                process is currently working on this dataset. Otherwise the dataset
                could be put into a corrupted state.

        retrain: bool, default False
            This parameter is no longer used and is deprecated.

        Notes
        -----
        The frequency an application should call optimize is based on the frequency of
        data modifications.  If data is frequently added, deleted, or updated then
        optimize should be run frequently.  A good rule of thumb is to run optimize if
        you have added or modified 100,000 or more records or run more than 20 data
        modification operations.
        """
        LOOP.run(
            self._table.optimize(
                cleanup_older_than=cleanup_older_than,
                delete_unverified=delete_unverified,
                retrain=retrain,
            )
        )

    def list_indices(self) -> Iterable[IndexConfig]:
        """
        List all indices that have been created with Self::create_index
        """
        return LOOP.run(self._table.list_indices())

    def tokenize(
        self,
        query: str,
        *,
        column: Optional[str] = None,
        index_name: Optional[str] = None,
    ) -> Iterable[FtsToken]:
        """
        Tokenize a query using the tokenizer configured on an FTS index.

        Specify exactly one of ``column`` or ``index_name``.

        Model-backed tokenizers such as ``jieba/*`` and ``lindera/*`` are
        rebuilt in the client process from index metadata. For remote tables,
        this means the same tokenizer model files must also exist locally.
        """
        return LOOP.run(
            self._table.tokenize(query, column=column, index_name=index_name)
        )

    def index_stats(self, index_name: str) -> Optional[IndexStatistics]:
        """
        Retrieve statistics about an index

        Parameters
        ----------
        index_name: str
            The name of the index to retrieve statistics for

        Returns
        -------
        IndexStatistics or None
            The statistics about the index. Returns None if the index does not exist.
        """
        return LOOP.run(self._table.index_stats(index_name))

    def add_columns(
        self,
        transforms: Dict[str, str | FunctionApplication]
        | FunctionApplication
        | pa.Field
        | List[pa.Field]
        | pa.Schema
        | None = None,
        *,
        computed: Dict[str, str] | None = None,
    ) -> AddColumnsResult:
        return LOOP.run(self._table.add_columns(transforms, computed=computed))

    def refresh_column(self, column: str) -> "RefreshColumnResult":
        """Fill a computed column's unfilled rows. See
        [`AsyncTable.refresh_column`][lancedb.AsyncTable.refresh_column]."""
        return LOOP.run(self._table.refresh_column(column))

    def refresh_column_async(self, column: str) -> Job[RefreshColumnJobResult]:
        """Fill a computed column's unfilled rows, returning a handle to the
        refresh job. See
        [`Table.refresh_column_async`][lancedb.table.Table.refresh_column_async].
        """
        return Job(LOOP.run(self._table.refresh_column_async(column)))

    def alter_columns(
        self, *alterations: Iterable[Dict[str, str]]
    ) -> AlterColumnsResult:
        return LOOP.run(self._table.alter_columns(*alterations))

    def update_field_metadata(
        self, *updates: dict[str, Any]
    ) -> UpdateFieldMetadataResult:
        return LOOP.run(self._table.update_field_metadata(*updates))

    def drop_columns(self, columns: Iterable[str]) -> DropColumnsResult:
        return LOOP.run(self._table.drop_columns(columns))

    def set_unenforced_primary_key(self, columns: Union[str, Iterable[str]]) -> None:
        """Set the unenforced primary key. See
        [`AsyncTable.set_unenforced_primary_key`][lancedb.AsyncTable.set_unenforced_primary_key]."""
        return LOOP.run(self._table.set_unenforced_primary_key(columns))

    def set_lsm_write_spec(self, spec: "LsmWriteSpec") -> None:
        """Install an LsmWriteSpec. See
        [`AsyncTable.set_lsm_write_spec`][lancedb.AsyncTable.set_lsm_write_spec]."""
        return LOOP.run(self._table.set_lsm_write_spec(spec))

    def unset_lsm_write_spec(self) -> None:
        """Remove the LsmWriteSpec. See
        [`AsyncTable.unset_lsm_write_spec`][lancedb.AsyncTable.unset_lsm_write_spec]."""
        return LOOP.run(self._table.unset_lsm_write_spec())

    def get_lsm_write_spec(self) -> Optional["LsmWriteSpec"]:
        """Read the installed LsmWriteSpec, or ``None``. See
        [`AsyncTable.get_lsm_write_spec`][lancedb.AsyncTable.get_lsm_write_spec]."""
        return LOOP.run(self._table.get_lsm_write_spec())

    def checkpoint_lsm(self) -> None:
        """Synchronous version of
        [`AsyncTable.checkpoint_lsm`][lancedb.AsyncTable.checkpoint_lsm]."""
        return LOOP.run(self._table.checkpoint_lsm())

    def flush_lsm(self) -> None:
        """Synchronous version of
        [`AsyncTable.flush_lsm`][lancedb.AsyncTable.flush_lsm]."""
        return LOOP.run(self._table.flush_lsm())

    def compact_lsm(self) -> None:
        """Synchronous version of
        [`AsyncTable.compact_lsm`][lancedb.AsyncTable.compact_lsm]."""
        return LOOP.run(self._table.compact_lsm())

    def get_lsm_stats(self, *, include_generation_rows: bool = False) -> Optional[dict]:
        """Synchronous version of
        [`AsyncTable.get_lsm_stats`][lancedb.AsyncTable.get_lsm_stats]."""
        return LOOP.run(
            self._table.get_lsm_stats(include_generation_rows=include_generation_rows)
        )

    def close_lsm_writers(self) -> None:
        """Close cached MemWAL shard writers. See
        [`AsyncTable.close_lsm_writers`][lancedb.AsyncTable.close_lsm_writers]."""
        return LOOP.run(self._table.close_lsm_writers())

    def uses_v2_manifest_paths(self) -> bool:
        """
        Check if the table is using the new v2 manifest paths.

        Returns
        -------
        bool
            True if the table is using the new v2 manifest paths, False otherwise.
        """
        return LOOP.run(self._table.uses_v2_manifest_paths())

    def migrate_v2_manifest_paths(self):
        """
        Migrate the manifest paths to the new format.

        This will update the manifest to use the new v2 format for paths.

        This function is idempotent, and can be run multiple times without
        changing the state of the object store.

        !!! danger

            This should not be run while other concurrent operations are happening.
            And it should also run until completion before resuming other operations.

        You can use
        [LanceTable.uses_v2_manifest_paths][lancedb.table.LanceTable.uses_v2_manifest_paths]
        to check if the table is already using the new path style.
        """
        LOOP.run(self._table.migrate_v2_manifest_paths())

    @deprecation.deprecated(
        deprecated_in="0.33.1",
        current_version=__version__,
        details="Use update_field_metadata() instead.",
    )
    def replace_field_metadata(self, field_name: str, new_metadata: Dict[str, str]):
        """
        Replace the metadata of a field in the schema

        .. deprecated:: 0.33.1
            Use :func:`update_field_metadata` instead.

        Parameters
        ----------
        field_name: str
            The name of the field to replace the metadata for
        new_metadata: dict
            The new metadata to set
        """
        LOOP.run(self._table.replace_field_metadata(field_name, new_metadata))


def _handle_bad_vectors(
    reader: pa.RecordBatchReader,
    on_bad_vectors: Literal["error", "drop", "fill", "null"] = "error",
    fill_value: float = 0.0,
    target_schema: Optional[pa.Schema] = None,
    metadata: Optional[dict] = None,
) -> pa.RecordBatchReader:
    vector_columns = _find_vector_columns(reader.schema, target_schema, metadata)
    if not vector_columns:
        return reader

    output_schema = _vector_output_schema(reader.schema, vector_columns)

    def gen():
        for batch in reader:
            pending_dims = []
            for vector_column in vector_columns:
                dim = vector_column["expected_dim"]
                if target_schema is not None and dim is None:
                    dim = _infer_vector_dim(batch[vector_column["name"]])
                    pending_dims.append(vector_column)
                batch = _handle_bad_vector_column(
                    batch,
                    vector_column_name=vector_column["name"],
                    on_bad_vectors=on_bad_vectors,
                    fill_value=fill_value,
                    expected_dim=dim,
                    expected_value_type=vector_column["expected_value_type"],
                )
            for vector_column in pending_dims:
                if vector_column["expected_dim"] is None:
                    vector_column["expected_dim"] = _infer_vector_dim(
                        batch[vector_column["name"]]
                    )
            if batch.schema.equals(output_schema, check_metadata=True):
                yield batch
                continue

            cast_batches = (
                pa.Table.from_batches([batch]).cast(output_schema).to_batches()
            )
            if cast_batches:
                yield pa.RecordBatch.from_arrays(
                    cast_batches[0].columns,
                    schema=output_schema,
                )

    return pa.RecordBatchReader.from_batches(output_schema, gen())


def _find_vector_columns(
    reader_schema: pa.Schema,
    target_schema: Optional[pa.Schema],
    metadata: Optional[dict],
) -> List[dict]:
    if target_schema is None:
        vector_columns = []
        for field in reader_schema:
            named_vector_col = (
                _is_list_like(field.type)
                and pa.types.is_floating(field.type.value_type)
                and field.name == VECTOR_COLUMN_NAME
            )
            likely_vector_col = (
                pa.types.is_fixed_size_list(field.type)
                and pa.types.is_floating(field.type.value_type)
                and (field.type.list_size >= 10)
            )
            if named_vector_col or likely_vector_col:
                vector_columns.append(
                    {
                        "name": field.name,
                        "expected_dim": None,
                        "expected_value_type": None,
                    }
                )
        return vector_columns

    reader_column_names = set(reader_schema.names)
    active_metadata = _merge_metadata(target_schema.metadata, metadata)
    embedding_function_columns = set(
        EmbeddingFunctionRegistry.get_instance().parse_functions(active_metadata).keys()
    )
    vector_columns = []
    for field in target_schema:
        if field.name not in reader_column_names:
            continue
        if not _is_list_like(field.type) or not pa.types.is_floating(
            field.type.value_type
        ):
            continue

        reader_field = reader_schema.field(field.name)
        named_vector_col = (
            field.name in embedding_function_columns
            or field.name == VECTOR_COLUMN_NAME
            or (field.name == "embedding" and pa.types.is_fixed_size_list(field.type))
        )
        typed_fixed_vector_col = (
            pa.types.is_fixed_size_list(reader_field.type)
            and pa.types.is_floating(reader_field.type.value_type)
            and reader_field.type.list_size >= 10
        )

        if named_vector_col or typed_fixed_vector_col:
            vector_columns.append(
                {
                    "name": field.name,
                    "expected_dim": (
                        field.type.list_size
                        if pa.types.is_fixed_size_list(field.type)
                        else None
                    ),
                    "expected_value_type": field.type.value_type,
                }
            )

    return vector_columns


def _vector_output_schema(
    reader_schema: pa.Schema,
    vector_columns: List[dict],
) -> pa.Schema:
    columns_by_name = {column["name"]: column for column in vector_columns}
    fields = []
    for field in reader_schema:
        column = columns_by_name.get(field.name)
        if column is None:
            output_type = field.type
        else:
            output_type = _vector_output_type(field, column)
        fields.append(pa.field(field.name, output_type, field.nullable, field.metadata))
    return pa.schema(fields, metadata=reader_schema.metadata)


def _vector_output_type(field: pa.Field, vector_column: dict) -> pa.DataType:
    if not _is_list_like(field.type):
        return field.type

    if vector_column["expected_value_type"] is not None and (
        pa.types.is_null(field.type.value_type)
        or pa.types.is_integer(field.type.value_type)
        or pa.types.is_unsigned_integer(field.type.value_type)
    ):
        return pa.list_(vector_column["expected_value_type"])

    if (
        vector_column["expected_dim"] is not None
        and pa.types.is_fixed_size_list(field.type)
        and field.type.list_size != vector_column["expected_dim"]
    ):
        return pa.list_(field.type.value_type)

    return field.type


def _handle_bad_vector_column(
    data: pa.RecordBatch,
    vector_column_name: str,
    on_bad_vectors: str = "error",
    fill_value: float = 0.0,
    expected_dim: Optional[int] = None,
    expected_value_type: Optional[pa.DataType] = None,
) -> pa.RecordBatch:
    """
    Ensure that the vector column exists and has type fixed_size_list(float)

    Parameters
    ----------
    data: pa.Table
        The table to sanitize.
    vector_column_name: str
        The name of the vector column.
    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".
    fill_value: float, default 0.0
        The value to use when filling vectors. Only used if on_bad_vectors="fill".
    """
    position = data.column_names.index(vector_column_name)
    vec_arr = data[vector_column_name]
    if not _is_list_like(vec_arr.type):
        return data

    if (
        expected_dim is not None
        and pa.types.is_fixed_size_list(vec_arr.type)
        and vec_arr.type.list_size != expected_dim
    ):
        vec_arr = pa.array(vec_arr.to_pylist(), type=pa.list_(vec_arr.type.value_type))
        data = data.set_column(position, vector_column_name, vec_arr)

    if expected_value_type is not None and (
        pa.types.is_integer(vec_arr.type.value_type)
        or pa.types.is_unsigned_integer(vec_arr.type.value_type)
    ):
        vec_arr = pa.array(vec_arr.to_pylist(), type=pa.list_(expected_value_type))
        data = data.set_column(position, vector_column_name, vec_arr)

    if pa.types.is_floating(vec_arr.type.value_type):
        has_nan = has_nan_values(vec_arr)
    else:
        has_nan = pa.array([False] * len(vec_arr))

    if expected_dim is not None:
        dim = expected_dim
    elif pa.types.is_fixed_size_list(vec_arr.type):
        dim = vec_arr.type.list_size
    else:
        dim = _infer_vector_dim(vec_arr)
        if dim is None:
            return data

    is_null = pc.is_null(vec_arr)
    # pc.list_value_length returns null for null list entries, so
    # pc.not_equal(null, dim) also returns null. Use or_kleene so that
    # True OR null = True (Kleene three-valued logic), ensuring null vectors
    # are counted as wrong-dim.
    has_wrong_dim = pc.or_kleene(
        is_null,
        pc.not_equal(pc.list_value_length(vec_arr), dim),
    )

    has_bad_vectors = pc.any(has_nan).as_py() or pc.any(has_wrong_dim).as_py()

    if has_bad_vectors:
        is_bad = pc.or_(has_nan, has_wrong_dim)
        if on_bad_vectors == "error":
            if pc.any(has_wrong_dim).as_py():
                raise ValueError(
                    f"Vector column '{vector_column_name}' has variable length "
                    "vectors. Set on_bad_vectors='drop' to remove them, "
                    "set on_bad_vectors='fill' and fill_value=<value> to replace them, "
                    "or set on_bad_vectors='null' to replace them with null."
                )
            else:
                raise ValueError(
                    f"Vector column '{vector_column_name}' has NaNs. "
                    "Set on_bad_vectors='drop' to remove them, "
                    "set on_bad_vectors='fill' and fill_value=<value> to replace them, "
                    "or set on_bad_vectors='null' to replace them with null."
                )
        elif on_bad_vectors == "null":
            vec_arr = pc.if_else(
                is_bad,
                pa.scalar(None),
                vec_arr,
            )
        elif on_bad_vectors == "drop":
            data = data.filter(pc.invert(is_bad))
            vec_arr = data[vector_column_name]
        elif on_bad_vectors == "fill":
            if fill_value is None:
                raise ValueError(
                    "`fill_value` must not be None if `on_bad_vectors` is 'fill'"
                )
            vec_arr = _fill_bad_vector_values(vec_arr, dim, fill_value)
        else:
            raise ValueError(f"Invalid value for on_bad_vectors: {on_bad_vectors}")

    return data.set_column(position, vector_column_name, vec_arr)


def _fill_bad_vector_values(
    arr: Union[pa.Array, pa.ChunkedArray],
    dim: int,
    fill_value: float,
) -> pa.Array:
    if not isinstance(arr, pa.ChunkedArray):
        arr = pa.chunked_array([arr])
    arr = arr.combine_chunks()

    # A fixed-size slice truncates long vectors and pads short vectors with nulls.
    # Slice an array marking the original child nulls in parallel so padding nulls
    # can be distinguished from null values that were already present.
    sliced = pc.list_slice(arr, 0, dim, return_fixed_size_list=True)
    child_nulls = pc.is_null(arr.values)
    parent_nulls = pc.is_null(arr)
    if pa.types.is_list(arr.type):
        original_child_nulls = pa.ListArray.from_arrays(
            arr.offsets, child_nulls, mask=parent_nulls
        )
    elif pa.types.is_large_list(arr.type):
        original_child_nulls = pa.LargeListArray.from_arrays(
            arr.offsets, child_nulls, mask=parent_nulls
        )
    else:
        original_child_nulls = pa.FixedSizeListArray.from_arrays(
            child_nulls, arr.type.list_size, mask=parent_nulls
        )
    sliced_child_nulls = pc.list_slice(
        original_child_nulls, 0, dim, return_fixed_size_list=True
    )
    needs_fill = pc.is_null(sliced_child_nulls.values)

    values = sliced.values
    if pa.types.is_floating(values.type):
        values_for_nan_check = (
            values.cast(pa.float32()) if pa.types.is_float16(values.type) else values
        )
        needs_fill = pc.or_kleene(needs_fill, pc.is_nan(values_for_nan_check))

    fill_scalar = pa.scalar(fill_value).cast(values.type)
    filled_values = pc.if_else(needs_fill, fill_scalar, values)
    filled = pa.FixedSizeListArray.from_arrays(filled_values, dim)
    return filled.cast(arr.type)


def has_nan_values(arr: Union[pa.ListArray, pa.ChunkedArray]) -> pa.BooleanArray:
    if isinstance(arr, pa.ChunkedArray):
        values = pa.chunked_array([chunk.flatten() for chunk in arr.chunks])
    else:
        values = arr.flatten()
    if pa.types.is_float16(values.type):
        # is_nan isn't yet implemented for f16, so we cast to f32
        # https://github.com/apache/arrow/issues/45083
        values_has_nan = pc.is_nan(values.cast(pa.float32()))
    else:
        values_has_nan = pc.is_nan(values)
    values_indices = pc.list_parent_indices(arr)
    has_nan_indices = pc.unique(pc.filter(values_indices, values_has_nan))
    indices = pa.array(range(len(arr)), type=pa.uint32())
    return pc.is_in(indices, has_nan_indices)


def _is_list_like(data_type: pa.DataType) -> bool:
    return (
        pa.types.is_list(data_type)
        or pa.types.is_large_list(data_type)
        or pa.types.is_fixed_size_list(data_type)
    )


def _merge_metadata(*metadata_dicts: Optional[dict]) -> dict:
    merged = {}
    for metadata in metadata_dicts:
        if metadata is None:
            continue
        for key, value in metadata.items():
            if isinstance(key, str):
                key = key.encode("utf-8")
            if isinstance(value, str):
                value = value.encode("utf-8")
            merged[key] = value
    return merged


def _name_suggests_vector_column(field_name: str) -> bool:
    """Check if a field name indicates a vector column."""
    name_lower = field_name.lower()
    return "vector" in name_lower or "embedding" in name_lower


def _infer_target_schema(
    reader: pa.RecordBatchReader,
) -> Tuple[pa.Schema, pa.RecordBatchReader]:
    schema = reader.schema
    peeked = None

    for i, field in enumerate(schema):
        is_list_type = pa.types.is_list(field.type) or pa.types.is_large_list(
            field.type
        )

        if _name_suggests_vector_column(field.name) and is_list_type:
            if peeked is None:
                peeked, reader = peek_reader(reader)
            # Use the most common length of the list as the dimensions
            dim = _modal_list_size(peeked.column(i))

            # Determine target type based on value type
            if pa.types.is_floating(field.type.value_type):
                target_type = pa.list_(pa.float32(), dim)
            elif pa.types.is_integer(field.type.value_type):
                values = peeked.column(i)

                if isinstance(values, pa.ChunkedArray):
                    values = values.combine_chunks()

                flattened = values.flatten()
                valid_count = pc.count(flattened, mode="only_valid").as_py()

                if valid_count == 0:
                    target_type = pa.list_(pa.uint8(), dim)
                else:
                    min_max = pc.min_max(flattened)
                    min_value = min_max["min"].as_py()
                    max_value = min_max["max"].as_py()

                    if (min_value is not None and min_value < 0) or (
                        max_value is not None and max_value > 255
                    ):
                        target_type = pa.list_(pa.float32(), dim)
                    else:
                        target_type = pa.list_(pa.uint8(), dim)
            else:
                continue  # Skip non-numeric types

            new_field = pa.field(
                field.name,  # preserve original field name
                target_type,
                nullable=field.nullable,
            )

            schema = schema.set(i, new_field)

    return schema, reader


def _modal_list_size(arr: Union[pa.ListArray, pa.ChunkedArray]) -> int:
    # Use the most common length of the list as the dimensions
    return pc.mode(pc.list_value_length(arr))[0].as_py()["mode"]


def _infer_vector_dim(arr: Union[pa.Array, pa.ChunkedArray]) -> Optional[int]:
    if not _is_list_like(arr.type):
        return None
    lengths = pc.list_value_length(arr)
    lengths = pc.filter(lengths, pc.greater(lengths, 0))
    if len(lengths) == 0:
        return None
    return pc.mode(lengths)[0].as_py()["mode"]


def _validate_schema(schema: pa.Schema):
    """
    Make sure the metadata is valid utf8
    """
    if schema.metadata is not None:
        _validate_metadata(schema.metadata)


def _validate_metadata(metadata: dict):
    """
    Make sure the metadata values are valid utf8 (can be nested)

    Raises ValueError if not valid utf8
    """
    for k, v in metadata.items():
        if isinstance(v, bytes):
            try:
                v.decode("utf8")
            except UnicodeDecodeError:
                raise ValueError(
                    f"Metadata key {k} is not valid utf8. "
                    "Consider base64 encode for generic binary metadata."
                )
        elif isinstance(v, dict):
            _validate_metadata(v)


class AsyncTable:
    """
    An AsyncTable is a collection of Records in a LanceDB Database.

    An AsyncTable can be obtained from the
    [AsyncConnection.create_table][lancedb.AsyncConnection.create_table] and
    [AsyncConnection.open_table][lancedb.AsyncConnection.open_table] methods.

    An AsyncTable object is expected to be long lived and reused for multiple
    operations. AsyncTable objects will cache a certain amount of index data in memory.
    This cache will be freed when the Table is garbage collected.  To eagerly free the
    cache you can call the [close][lancedb.AsyncTable.close] method.  Once the
    AsyncTable is closed, it cannot be used for any further operations.

    An AsyncTable can also be used as a context manager, and will automatically close
    when the context is exited.  Closing a table is optional.  If you do not close the
    table, it will be closed when the AsyncTable object is garbage collected.

    Examples
    --------

    Create using [AsyncConnection.create_table][lancedb.AsyncConnection.create_table]
    (more examples in that method's documentation).

    >>> import lancedb
    >>> async def create_a_table():
    ...     db = await lancedb.connect_async("./.lancedb")
    ...     data = [{"vector": [1.1, 1.2], "b": 2}]
    ...     table = await db.create_table("my_table", data=data)
    ...     print(await table.query().limit(5).to_arrow())
    >>> import asyncio
    >>> asyncio.run(create_a_table())
    pyarrow.Table
    vector: fixed_size_list<item: float>[2]
      child 0, item: float
    b: int64
    ----
    vector: [[[1.1,1.2]]]
    b: [[2]]

    Can append new data with [AsyncTable.add()][lancedb.table.AsyncTable.add].

    >>> async def add_to_table():
    ...     db = await lancedb.connect_async("./.lancedb")
    ...     table = await db.open_table("my_table")
    ...     await table.add([{"vector": [0.5, 1.3], "b": 4}])
    >>> asyncio.run(add_to_table())

    Can query the table with
    [AsyncTable.vector_search][lancedb.table.AsyncTable.vector_search].

    >>> async def search_table_for_vector():
    ...     db = await lancedb.connect_async("./.lancedb")
    ...     table = await db.open_table("my_table")
    ...     results = (
    ...       await table.vector_search([0.4, 0.4]).select(["b", "vector"]).to_pandas()
    ...     )
    ...     print(results)
    >>> asyncio.run(search_table_for_vector())
       b      vector  _distance
    0  4  [0.5, 1.3]       0.82
    1  2  [1.1, 1.2]       1.13

    Search queries are much faster when an index is created. See
    [AsyncTable.create_index][lancedb.table.AsyncTable.create_index].
    """

    def __init__(
        self,
        table: LanceDBTable,
        *,
        namespace_path: Optional[List[str]] = None,
        namespace_client: Optional[Any] = None,
        pushdown_operations: Optional[set] = None,
        route_pushdown_to_rust: bool = False,
    ):
        """Create a new AsyncTable object.

        You should not create AsyncTable objects directly.

        Use [AsyncConnection.create_table][lancedb.AsyncConnection.create_table] and
        [AsyncConnection.open_table][lancedb.AsyncConnection.open_table] to obtain
        Table objects."""
        self._inner = table
        self._namespace_path = namespace_path or []
        self._namespace_client = namespace_client
        self._pushdown_operations = pushdown_operations or set()
        # See LanceTable.__init__: defer QueryTable pushdown to Rust (which emits
        # the read-freshness header) for natively-built namespace clients.
        self._route_pushdown_to_rust = route_pushdown_to_rust

    def _set_namespace_context(
        self,
        *,
        namespace_path: Optional[List[str]] = None,
        namespace_client: Optional[Any] = None,
        pushdown_operations: Optional[set] = None,
        route_pushdown_to_rust: bool = False,
    ) -> "AsyncTable":
        self._namespace_path = namespace_path or []
        self._namespace_client = namespace_client
        self._pushdown_operations = pushdown_operations or set()
        self._route_pushdown_to_rust = route_pushdown_to_rust
        return self

    def __repr__(self):
        return self._inner.__repr__()

    def __enter__(self):
        return self

    def __exit__(self, *_):
        self.close()

    def is_open(self) -> bool:
        """Return True if the table is open."""
        return self._inner.is_open()

    def close(self):
        """Close the table and free any resources associated with it.

        It is safe to call this method multiple times.

        Any attempt to use the table after it has been closed will raise an error."""
        return self._inner.close()

    async def set_unenforced_primary_key(
        self, columns: Union[str, Iterable[str]]
    ) -> None:
        """Set the unenforced primary key for this table to the given
        ordered list of columns.

        "Unenforced" means LanceDB does not check uniqueness on writes; the
        columns are recorded in the schema as the primary key so that
        features such as `merge_insert` can use them. Calling this again
        replaces any previously-set primary key.

        Parameters
        ----------
        columns : str or Iterable[str]
            Either a single column name (single-column key) or an ordered
            iterable of column names (composite key). Each column dtype
            must be one of: int32, int64, utf8, large_utf8, binary,
            large_binary, fixed_size_binary.
        """
        if isinstance(columns, str):
            columns = [columns]
        else:
            columns = list(columns)
        await self._inner.set_unenforced_primary_key(columns)

    async def set_lsm_write_spec(self, spec: "LsmWriteSpec") -> None:
        """Install an LsmWriteSpec on this table.

        The spec selects Lance's MemWAL LSM-style write path for future
        `merge_insert` calls. ``LsmWriteSpec`` chooses one of three sharding
        strategies:

        - ``LsmWriteSpec.bucket(column, num_buckets)`` — hash-bucket writes by
          the single-column unenforced primary key.
        - ``LsmWriteSpec.identity(column)`` — shard by the raw value of a
          scalar column.
        - ``LsmWriteSpec.unsharded()`` — route every write to a single shard.

        All variants require the table to have an unenforced primary key set
        via [`set_unenforced_primary_key`]; bucket sharding additionally
        requires it to be the single column being bucketed.

        By default the MemWAL maintains every index on the table, resolved
        here — a snapshot, so an index created afterwards needs the spec unset
        and set again. This fails if one cannot be maintained; name the set
        with ``with_maintained_indexes`` to install anyway. That pins an exact
        set (a still-building index is rejected, not omitted); ``[]`` maintains
        none.

        Parameters
        ----------
        spec : LsmWriteSpec
            The sharding spec to install.

        Examples
        --------
        >>> from lancedb import LsmWriteSpec
        >>> # table.set_unenforced_primary_key("id")
        >>> # table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 16))
        """
        await self._inner.set_lsm_write_spec(spec)

    async def unset_lsm_write_spec(self) -> None:
        """Remove the LsmWriteSpec from this table.

        Reverts to the standard `merge_insert` write path. Errors if no spec
        is currently set.
        """
        await self._inner.unset_lsm_write_spec()

    async def get_lsm_write_spec(self) -> Optional["LsmWriteSpec"]:
        """Read the LsmWriteSpec currently installed on this table.

        Returns ``None`` when the MemWAL LSM write path is not enabled (no
        spec has been set, or it was removed with `unset_lsm_write_spec`).
        The returned spec mirrors what was passed to `set_lsm_write_spec`,
        except that ``maintained_indexes`` always reports the concrete list
        resolved when the spec was set — ``None`` never round-trips.
        """
        return await self._inner.get_lsm_write_spec()

    async def checkpoint_lsm(self) -> None:
        """Converge this table's LSM write path into its base table.

        One flush, sealing every memtable into L0, then compaction triggers
        until every generation that existed at that moment has reached base.
        The loop runs client-side, reading progress from ``get_lsm_stats``.

        Best-effort: generations created *while* it runs are deliberately not
        waited on, which is what lets it terminate on a table taking writes.
        Idempotent and safe on a cadence.

        There is no deadline, and the caller owns that. It returns when the
        target generations are gone, raises on a terminal server fault, and
        otherwise waits however long the server takes. A slow table and a
        stuck one are the same picture from the client: the compactor pool is
        shared across every table on the node, so a checkpoint queued behind
        unrelated work looks exactly like one that is merging. Wrap this in
        ``asyncio.wait_for`` for a wall-clock bound; abandoning it partway
        costs nothing.
        """
        await self._inner.checkpoint_lsm()

    async def flush_lsm(self) -> None:
        """Seal every bucket's active memtable into L0.

        Does not touch the base table — moving L0 into base is
        `compact_lsm`. On a node that has not claimed this table, this claims
        it and replays its WAL log first.
        """
        await self._inner.flush_lsm()

    async def compact_lsm(self) -> None:
        """Trigger a background L0 to base compaction pass per bucket.

        Returns once the passes are dispatched, not once they finish: watch
        ``get_lsm_stats`` for progress, or use ``checkpoint_lsm`` to loop
        until the current L0 has reached base.
        """
        await self._inner.compact_lsm()

    async def get_lsm_stats(
        self, *, include_generation_rows: bool = False
    ) -> Optional[dict]:
        """Read live per-bucket LSM state.

        Answers "how far behind is my fresh tier", "which bucket is hot", and
        "why is my fresh-tier vector search brute-force". Mutates no table
        state, though on a node that has not claimed this table it claims it,
        exactly as a read would.

        Returns ``None`` only when the LSM write path is not enabled.

        Parameters
        ----------
        include_generation_rows
            Report a row count per L0 generation. Off by default: each count
            opens an uncached Lance dataset, and ``checkpoint_lsm`` polls this
            needing only generation numbers.
        """
        return await self._inner.get_lsm_stats(include_generation_rows)

    async def close_lsm_writers(self) -> None:
        """Drain and close any cached MemWAL shard writers for this table.

        When an LSM write spec is installed, `merge_insert` opens MemWAL shard
        writers and caches them for reuse across calls. This closes them,
        flushing pending data; writers reopen lazily on the next
        `merge_insert`. It is a no-op when no writers are cached.
        """
        await self._inner.close_lsm_writers()

    @property
    def name(self) -> str:
        """The name of the table."""
        return self._inner.name()

    async def schema(self) -> pa.Schema:
        """The [Arrow Schema](https://arrow.apache.org/docs/python/api/datatypes.html#)
        of this Table

        """
        return await self._inner.schema()

    async def embedding_functions(self) -> Dict[str, EmbeddingFunctionConfig]:
        """
        Get the embedding functions for the table

        Returns
        -------
        funcs: Dict[str, EmbeddingFunctionConfig]
            A mapping of the vector column to the embedding function
            or empty dict if not configured.
        """
        schema = await self.schema()
        return EmbeddingFunctionRegistry.get_instance().parse_functions(schema.metadata)

    async def count_rows(self, filter: Optional[str] = None) -> int:
        """
        Count the number of rows in the table.

        Parameters
        ----------
        filter: str, optional
            A SQL where clause to filter the rows to count.
        """
        return await self._inner.count_rows(filter)

    async def head(self, n=5) -> pa.Table:
        """
        Return the first `n` rows of the table.

        Parameters
        ----------
        n: int, default 5
            The number of rows to return.
        """
        return await self.query().limit(n).to_arrow()

    def query(self) -> AsyncQuery:
        """
        Returns an [AsyncQuery][lancedb.query.AsyncQuery] that can be used
        to search the table.

        Use methods on the returned query to control query behavior.  The query
        can be executed with methods like [to_arrow][lancedb.query.AsyncQuery.to_arrow],
        [to_pandas][lancedb.query.AsyncQuery.to_pandas] and more.
        """
        return AsyncQuery(self._inner.query(), self)

    async def to_lance(self, **kwargs) -> lance.LanceDataset:
        """Return the Lance dataset backing this table.

        Parameters
        ----------
        **kwargs
            Forwarded to `lance.dataset`.

        Returns
        -------
        lance.LanceDataset
            The Lance dataset at this table handle's version and branch.

        Examples
        --------
        >>> async def get_lance_dataset(table):
        ...     return await table.to_lance()
        """
        try:
            import lance
        except ImportError:
            raise ImportError(
                "The lance library is required to use this function. "
                "Please install with `pip install pylance`."
            )

        # lance.dataset() can't open a branch directly, so open the base table
        # and check out the branch ref (a None branch resolves to main).
        branch = self.current_branch()
        table_version = await self.version()
        version = None if branch is not None else table_version
        ds = lance.dataset(
            await self.uri(),
            version=version,
            storage_options=await self.latest_storage_options(),
            **kwargs,
        )
        if branch is not None:
            ds = ds.checkout_version((branch, table_version))
        return ds

    async def to_pandas(self, blob_mode: BlobMode = "lazy", **kwargs) -> "pd.DataFrame":
        """Return the table as a pandas DataFrame.

        Parameters
        ----------
        blob_mode: str, default "lazy"
            Controls how Lance blob columns are returned.
        **kwargs
            Forwarded to PyArrow / Lance pandas conversion.

        Returns
        -------
        pd.DataFrame
        """
        validate_blob_mode(blob_mode)
        schema = await self.schema()
        if blob_mode == "descriptions" or not schema_has_blob_field(schema):
            arrow_tbl = await self.to_arrow()
            if blob_mode == "descriptions":
                arrow_tbl = strip_auto_row_ids(
                    arrow_tbl, row_addressable_blob_v2_paths(schema)
                )
            return arrow_tbl.to_pandas(**kwargs)

        if blob_mode == "lazy" and get_uri_scheme(await self.uri()) == "memory":
            return (await self.to_arrow()).to_pandas(**kwargs)
        if blob_mode == "bytes" and blob_v2_column_paths(schema):
            return await self.query().to_pandas(blob_mode=blob_mode, **kwargs)
        return (await self.to_lance()).to_pandas(blob_mode=blob_mode, **kwargs)

    async def to_arrow(self) -> pa.Table:
        """Return the table as a pyarrow Table.

        Returns
        -------
        pa.Table
        """
        if (
            _should_push_down_query_table(
                self._namespace_client, self._pushdown_operations
            )
            and not self._route_pushdown_to_rust
        ):
            return (await self._execute_query(Query())).read_all()

        return await self.query().to_arrow()

    async def create_index(
        self,
        column: str,
        *,
        replace: Optional[bool] = None,
        config: Optional[
            Union[
                IvfFlat,
                IvfPq,
                IvfRq,
                HnswPq,
                HnswSq,
                HnswFlat,
                BTree,
                Bitmap,
                LabelList,
                Fm,
                FTS,
            ]
        ] = None,
        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
        train: bool = True,
    ):
        """Create an index to speed up queries

        Indices can be created on vector columns or scalar columns.
        Indices on vector columns will speed up vector searches.
        Indices on scalar columns will speed up filtering (in both
        vector and non-vector searches)

        Parameters
        ----------
        column: str
            The column to index.
        replace: bool, default True
            Whether to replace the existing index

            If this is false, and another index already exists on the same columns
            and the same name, then an error will be returned.  This is true even if
            that index is out of date.

            The default is True
        config: default None
            For advanced configuration you can specify the type of index you would
            like to create.   You can also specify index-specific parameters when
            creating an index object.
        wait_timeout: timedelta, optional
            The timeout to wait if indexing is asynchronous.
        name: str, optional
            The name of the index. If not provided, a default name will be generated.
        train: bool, default True
            Whether to train the index with existing data. Vector indices always train
            with existing data.
        """
        if config is not None:
            if not isinstance(
                config,
                (
                    IvfFlat,
                    IvfSq,
                    IvfPq,
                    IvfRq,
                    HnswPq,
                    HnswSq,
                    HnswFlat,
                    BTree,
                    Bitmap,
                    LabelList,
                    Fm,
                    FTS,
                ),
            ):
                raise TypeError(
                    "config must be an instance of IvfSq, IvfPq, IvfRq, HnswPq, HnswSq,"
                    " BTree, Bitmap, LabelList, Fm, or FTS, but got "
                    + str(type(config))
                )
        try:
            await self._inner.create_index(
                column,
                index=config,
                replace=replace,
                wait_timeout=wait_timeout,
                name=name,
                train=train,
            )
        except (ValueError, RuntimeError) as e:
            if isinstance(config, FTS):
                _maybe_add_fts_error_note(
                    e,
                    base_tokenizer=config.base_tokenizer,
                    language=config.language,
                )
            raise e

    async def create_index_async(
        self,
        column: str,
        *,
        replace: Optional[bool] = None,
        config: Optional[
            Union[
                IvfFlat,
                IvfPq,
                IvfRq,
                HnswPq,
                HnswSq,
                HnswFlat,
                BTree,
                Bitmap,
                LabelList,
                Fm,
                FTS,
            ]
        ] = None,
        wait_timeout: Optional[timedelta] = None,
        name: Optional[str] = None,
        train: bool = True,
    ) -> AsyncJob:
        """Create an index, returning a handle to the indexing job.

        Takes the same arguments as :meth:`create_index`. The job may already
        be complete when returned; callers must not assume the index exists
        until :meth:`AsyncJob.wait` resolves.
        """
        job = await self._inner.create_index_async(
            column,
            index=config,
            replace=replace,
            wait_timeout=wait_timeout,
            name=name,
            train=train,
        )
        return AsyncJob(job)

    async def drop_index(self, name: str) -> None:
        """
        Drop an index from the table.

        Parameters
        ----------
        name: str
            The name of the index to drop.

        Notes
        -----
        This does not delete the index from disk, it just removes it from the table.
        To delete the index, run [optimize][lancedb.table.AsyncTable.optimize]
        after dropping the index.

        Use [list_indices][lancedb.table.AsyncTable.list_indices] to find the names
        of the indices.
        """
        await self._inner.drop_index(name)

    async def prewarm_index(self, name: str) -> None:
        """
        Prewarm an index in the table.

        This is a hint to the database that the index will be accessed in the
        future and should be loaded into memory if possible.  This can reduce
        cold-start latency for subsequent queries.

        This call initiates prewarming and returns once the request is accepted.
        It is idempotent and safe to call from multiple clients concurrently.

        It is generally wasteful to call this if the index does not fit into the
        available cache.  Not all index types support prewarming; unsupported
        indices will silently ignore the request.

        Parameters
        ----------
        name: str
            The name of the index to prewarm
        """
        await self._inner.prewarm_index(name)

    async def prewarm_data(self, columns: Optional[List[str]] = None) -> None:
        """
        Prewarm data for the table.

        This is a hint to the database that the given columns will be accessed
        in the future and the database should prefetch the data if possible.
        Currently only supported on remote tables.

        This call initiates prewarming and returns once the request is accepted.
        It is idempotent and safe to call from multiple clients concurrently.

        This operation has a large upfront cost but can speed up future queries
        that need to fetch the given columns.  Large columns such as embeddings
        or binary data may not be practical to prewarm.  This feature is intended
        for workloads that issue many queries against the same columns.

        Parameters
        ----------
        columns: list of str, optional
            The columns to prewarm. If None, all columns are prewarmed.
        """
        await self._inner.prewarm_data(columns)

    async def wait_for_index(
        self, index_names: Iterable[str], timeout: timedelta = timedelta(seconds=300)
    ) -> None:
        """
        Wait for indexing to complete for the given index names.
        This will poll the table until all the indices are fully indexed,
        or raise a timeout exception if the timeout is reached.

        Parameters
        ----------
        index_names: str
            The name of the indices to poll
        timeout: timedelta
            Timeout to wait for asynchronous indexing. The default is 5 minutes.
        """
        await self._inner.wait_for_index(index_names, timeout)

    async def stats(self) -> TableStatistics:
        """
        Retrieve table and fragment statistics.
        """
        return await self._inner.stats()

    async def uri(self) -> str:
        """
        Get the table URI (storage location).

        For remote tables, this fetches the location from the server via describe.
        For local tables, this returns the dataset URI.

        Returns
        -------
        str
            The full storage location of the table (e.g., S3/GCS path).
        """
        return await self._inner.uri()

    async def initial_storage_options(self) -> Optional[Dict[str, str]]:
        """Get the initial storage options that were passed in when opening this table.

        For dynamically refreshed options (e.g., credential vending), use
        :meth:`latest_storage_options`.

        Warning: This is an internal API and the return value is subject to change.

        Returns
        -------
        Optional[Dict[str, str]]
            The storage options, or None if no storage options were configured.
        """
        return await self._inner.initial_storage_options()

    async def latest_storage_options(self) -> Optional[Dict[str, str]]:
        """Get the latest storage options, refreshing from provider if configured.

        This method is useful for credential vending scenarios where storage options
        may be refreshed dynamically. If no dynamic provider is configured, this
        returns the initial static options.

        Warning: This is an internal API and the return value is subject to change.

        Returns
        -------
        Optional[Dict[str, str]]
            The storage options, or None if no storage options were configured.
        """
        return await self._inner.latest_storage_options()

    async def add(
        self,
        data: DATA,
        *,
        mode: Optional[Literal["append", "overwrite"]] = "append",
        on_bad_vectors: Optional[OnBadVectorsType] = None,
        fill_value: Optional[float] = None,
        progress: Optional[Union[bool, Callable, Any]] = None,
        write_parallelism: Optional[int] = None,
        allow_external_blob_outside_bases: bool = False,
    ) -> AddResult:
        """Add more data to the [AsyncTable][lancedb.table.AsyncTable].

        Parameters
        ----------
        data: DATA
            The data to insert into the table. Acceptable types are:

            - list-of-dict

            - pandas.DataFrame

            - pyarrow.Table or pyarrow.RecordBatch
        mode: str
            The mode to use when writing the data. Valid values are
            "append" and "overwrite".
        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".
        fill_value: float, default 0.
            The value to use when filling vectors. Only used if on_bad_vectors="fill".
        progress: callable or tqdm-like, optional
            A callback or tqdm-compatible progress bar. See
            :meth:`Table.add` for details.
        write_parallelism: int, optional
            Number of partitions to write in parallel. Higher values increase
            throughput but also peak memory use, since each partition buffers
            data in flight. Defaults to an estimate based on the data size,
            capped at the number of CPU cores. Lower this if bulk ingestion is
            using too much memory.
        allow_external_blob_outside_bases: bool, default False
            Allow blob URIs outside registered bases. See :meth:`Table.add`.
            Local tables only.

        """
        schema = await self.schema()
        if on_bad_vectors is None:
            on_bad_vectors = "error"
        if fill_value is None:
            fill_value = 0.0

        # _santitize_data is an old code path, but we will use it until the
        # new code path is ready.
        if mode == "overwrite":
            # For overwrite, apply the same preprocessing as create_table
            # so vector columns are inferred as FixedSizeList.
            data, _ = sanitize_create_table(
                data, None, on_bad_vectors=on_bad_vectors, fill_value=fill_value
            )
        elif on_bad_vectors != "error" or (
            schema.metadata is not None and b"embedding_functions" in schema.metadata
        ):
            data = _sanitize_data(
                data,
                schema,
                metadata=schema.metadata,
                on_bad_vectors=on_bad_vectors,
                fill_value=fill_value,
                allow_subschema=True,
            )
        _register_optional_converters()
        data = to_scannable(data)
        progress, owns = _normalize_progress(progress)
        try:
            return await self._inner.add(
                data,
                mode or "append",
                progress=progress,
                write_parallelism=write_parallelism,
                allow_external_blob_outside_bases=allow_external_blob_outside_bases,
            )
        except RuntimeError as e:
            if "Cast error" in str(e):
                raise ValueError(e)
            elif "Vector column contains NaN" in str(e):
                raise ValueError(e)
            else:
                raise
        finally:
            if owns:
                progress.close()

    def merge_insert(self, on: Union[str, Iterable[str]]) -> LanceMergeInsertBuilder:
        """
        Returns a [`LanceMergeInsertBuilder`][lancedb.merge.LanceMergeInsertBuilder]
        that can be used to create a "merge insert" operation

        This operation can add rows, update rows, and remove rows all in a single
        transaction. It is a very generic tool that can be used to create
        behaviors like "insert if not exists", "update or insert (i.e. upsert)",
        or even replace a portion of existing data with new data (e.g. replace
        all data where month="january")

        The merge insert operation works by combining new data from a
        **source table** with existing data in a **target table** by using a
        join.  There are three categories of records.

        "Matched" records are records that exist in both the source table and
        the target table. "Not matched" records exist only in the source table
        (e.g. these are new data) "Not matched by source" records exist only
        in the target table (this is old data)

        The builder returned by this method can be used to customize what
        should happen for each category of data.

        Please note that the data may appear to be reordered as part of this
        operation.  This is because updated rows will be deleted from the
        dataset and then reinserted at the end with the new values.

        Parameters
        ----------

        on: Union[str, Iterable[str]]
            A column (or columns) to join on.  This is how records from the
            source table and target table are matched.  Typically this is some
            kind of key or id column.  Passing several columns matches on the
            composite key: a source row updates a target row only when it
            agrees on every one of them.

        Examples
        --------
        >>> import lancedb
        >>> data = pa.table({"a": [2, 1, 3], "b": ["a", "b", "c"]})
        >>> db = lancedb.connect("./.lancedb")
        >>> table = db.create_table("my_table", data)
        >>> new_data = pa.table({"a": [2, 3, 4], "b": ["x", "y", "z"]})
        >>> # Perform a "upsert" operation
        >>> res = table.merge_insert("a")     \\
        ...      .when_matched_update_all()     \\
        ...      .when_not_matched_insert_all() \\
        ...      .execute(new_data)
        >>> res
        MergeResult(version=2, num_updated_rows=2, num_inserted_rows=1, num_deleted_rows=0, num_attempts=1, num_rows=3)
        >>> # The order of new rows is non-deterministic since we use
        >>> # a hash-join as part of this operation and so we sort here
        >>> table.to_arrow().sort_by("a").to_pandas()
           a  b
        0  1  b
        1  2  x
        2  3  y
        3  4  z
        """  # noqa: E501
        on = [on] if isinstance(on, str) else list(iter(on))

        return LanceMergeInsertBuilder(self, on)

    @overload
    async def search(
        self,
        query: Optional[str] = None,
        vector_column_name: Optional[str] = None,
        query_type: Literal["auto"] = ...,
        ordering_field_name: Optional[str] = None,
        fts_columns: Optional[Union[str, List[str]]] = None,
    ) -> Union[AsyncHybridQuery, AsyncFTSQuery, AsyncVectorQuery]: ...

    @overload
    async def search(
        self,
        query: Optional[str] = None,
        vector_column_name: Optional[str] = None,
        query_type: Literal["hybrid"] = ...,
        ordering_field_name: Optional[str] = None,
        fts_columns: Optional[Union[str, List[str]]] = None,
    ) -> AsyncHybridQuery: ...

    @overload
    async def search(
        self,
        query: Optional[Union[VEC, "PIL.Image.Image", Tuple]] = None,
        vector_column_name: Optional[str] = None,
        query_type: Literal["auto"] = ...,
        ordering_field_name: Optional[str] = None,
        fts_columns: Optional[Union[str, List[str]]] = None,
    ) -> AsyncVectorQuery: ...

    @overload
    async def search(
        self,
        query: Optional[str] = None,
        vector_column_name: Optional[str] = None,
        query_type: Literal["fts"] = ...,
        ordering_field_name: Optional[str] = None,
        fts_columns: Optional[Union[str, List[str]]] = None,
    ) -> AsyncFTSQuery: ...

    @overload
    async def search(
        self,
        query: Optional[
            Union[VEC, str, "PIL.Image.Image", Tuple, FullTextQuery]
        ] = None,
        vector_column_name: Optional[str] = None,
        query_type: Literal["vector"] = ...,
        ordering_field_name: Optional[str] = None,
        fts_columns: Optional[Union[str, List[str]]] = None,
    ) -> AsyncVectorQuery: ...

    async def search(
        self,
        query: Optional[
            Union[VEC, str, "PIL.Image.Image", Tuple, FullTextQuery]
        ] = None,
        vector_column_name: Optional[str] = None,
        query_type: QueryType = "auto",
        ordering_field_name: Optional[str] = None,
        fts_columns: Optional[Union[str, List[str]]] = None,
    ) -> Union[AsyncHybridQuery, AsyncFTSQuery, AsyncVectorQuery]:
        """Create a search query to find the nearest neighbors
        of the given query vector. We currently support [vector search](https://lancedb.com/docs/search/vector-search/)
        and [full-text search](https://lancedb.com/docs/search/full-text-search/).

        All query options are defined in [AsyncQuery][lancedb.query.AsyncQuery].

        Parameters
        ----------
        query: list/np.ndarray/str/PIL.Image.Image, default None
            The targetted vector to search for.

            - *default None*.
            Acceptable types are: list, np.ndarray, PIL.Image.Image

            - If None then the select/where/limit clauses are applied to filter
            the table
        vector_column_name: str, optional
            The name of the vector column to search.

            The vector column needs to be a pyarrow fixed size list type

            - If not specified then the vector column is inferred from
            the table schema

            - If the table has multiple vector columns then the *vector_column_name*
            needs to be specified. Otherwise, an error is raised.
        query_type: str
            *default "auto"*.
            Acceptable types are: "vector", "fts", "hybrid", or "auto"

            - If "auto" then the query type is inferred from the query;

                - If `query` is a list/np.ndarray then the query type is
                "vector";

                - If `query` is a PIL.Image.Image then either do vector search,
                or raise an error if no corresponding embedding function is found.

            - If `query` is a string, then the query type is "vector" if the
              table has embedding functions else the query type is "fts"

        Returns
        -------
        LanceQueryBuilder
            A query builder object representing the query.
        """

        def is_embedding(query):
            return isinstance(query, (list, np.ndarray, pa.Array, pa.ChunkedArray))

        async def get_embedding_func(
            vector_column_name: Optional[str],
            query_type: QueryType,
            query: Optional[Union[VEC, str, "PIL.Image.Image", Tuple, FullTextQuery]],
        ) -> Tuple[str, EmbeddingFunctionConfig]:
            if isinstance(query, FullTextQuery):
                query_type = "fts"
            schema = await self.schema()
            vector_column_name = infer_vector_column_name(
                schema=schema,
                query_type=query_type,
                query=query,
                vector_column_name=vector_column_name,
            )
            funcs = EmbeddingFunctionRegistry.get_instance().parse_functions(
                schema.metadata
            )
            func = funcs.get(vector_column_name)
            if func is None:
                error = ValueError(
                    f"Column '{vector_column_name}' has no registered "
                    "embedding function."
                )
                if len(funcs) > 0:
                    add_note(
                        error,
                        "Embedding functions are registered for columns: "
                        f"{list(funcs.keys())}",
                    )
                else:
                    add_note(
                        error, "No embedding functions are registered for any columns."
                    )
                raise error
            return vector_column_name, func

        async def make_embedding(embedding, query):
            if embedding is not None:
                loop = asyncio.get_running_loop()
                # This function is likely to block, since it either calls an expensive
                # function or makes an HTTP request to an embeddings REST API. Run it
                # on a dedicated executor so it can't starve the default executor that
                # other blocking I/O shares. See
                # https://github.com/lancedb/lancedb/issues/3310.
                return (
                    await loop.run_in_executor(
                        embedding_executor(),
                        embedding.function.compute_query_embeddings_with_retry,
                        query,
                    )
                )[0]
            else:
                return None

        if query_type == "auto":
            # Infer the query type.
            if is_embedding(query):
                vector_query = query
                query_type = "vector"
            elif isinstance(query, FullTextQuery):
                query_type = "fts"
            elif isinstance(query, str):
                try:
                    (
                        indices,
                        (vector_column_name, embedding_conf),
                    ) = await asyncio.gather(
                        self.list_indices(),
                        get_embedding_func(vector_column_name, "auto", query),
                    )
                except ValueError as e:
                    if "Column" in str(
                        e
                    ) and "has no registered embedding function" in str(e):
                        # If the column has no registered embedding function,
                        # then it's an FTS query.
                        query_type = "fts"
                    else:
                        raise e
                else:
                    if embedding_conf is not None:
                        vector_query = await make_embedding(embedding_conf, query)
                        if any(
                            i.columns[0] == embedding_conf.source_column
                            and i.index_type == "FTS"
                            for i in indices
                        ):
                            query_type = "hybrid"
                        else:
                            query_type = "vector"
                    else:
                        query_type = "fts"
            else:
                # it's an image or something else embeddable.
                query_type = "vector"
        elif query_type == "vector":
            if is_embedding(query):
                vector_query = query
            else:
                vector_column_name, embedding_conf = await get_embedding_func(
                    vector_column_name, query_type, query
                )
                vector_query = await make_embedding(embedding_conf, query)
        elif query_type == "hybrid":
            if is_embedding(query):
                raise ValueError("Hybrid search requires a text query")
            else:
                vector_column_name, embedding_conf = await get_embedding_func(
                    vector_column_name, query_type, query
                )
                vector_query = await make_embedding(embedding_conf, query)

        if query_type == "vector":
            builder = self.query().nearest_to(vector_query)
            if vector_column_name:
                builder = builder.column(vector_column_name)
            return builder
        elif query_type == "fts":
            return self.query().nearest_to_text(query, columns=fts_columns)
        elif query_type == "hybrid":
            builder = self.query().nearest_to(vector_query)
            if vector_column_name:
                builder = builder.column(vector_column_name)
            return builder.nearest_to_text(query, columns=fts_columns)
        else:
            raise ValueError(f"Unknown query type: '{query_type}'")

    def vector_search(
        self,
        query_vector: Union[VEC, Tuple],
    ) -> AsyncVectorQuery:
        """
        Search the table with a given query vector.
        This is a convenience method for preparing a vector query and
        is the same thing as calling `nearestTo` on the builder returned
        by `query`.  Seer [nearest_to][lancedb.query.AsyncQuery.nearest_to] for more
        details.
        """
        return self.query().nearest_to(query_vector)

    def _sync_query_to_async(
        self, query: Query
    ) -> (
        AsyncHybridQuery
        | AsyncFTSQuery
        | AsyncVectorQuery
        | AsyncQuery
        | AsyncTakeQuery
    ):
        if query.take_offsets is not None:
            take_query = self.take_offsets(query.take_offsets)
            if query.columns:
                take_query = take_query.select(query.columns)
            if query.use_lsm is not None:
                take_query = take_query.use_lsm(query.use_lsm)
            if query.with_row_id:
                take_query = take_query.with_row_id()
            return take_query

        async_query = self.query()
        if query.limit is not None:
            async_query = async_query.limit(query.limit)
        if query.offset is not None:
            async_query = async_query.offset(query.offset)
        if query.columns:
            async_query = async_query.select(query.columns)
        if query.filter is not None:
            async_query = async_query.where(query.filter)
        if query.fast_search:
            async_query = async_query.fast_search()
        if query.use_lsm is not None:
            async_query = async_query.use_lsm(query.use_lsm)
        if query.with_row_id:
            async_query = async_query.with_row_id()
        if query.order_by:
            async_query = async_query.order_by(query.order_by)

        if query.vector:
            async_query = async_query.nearest_to(query.vector).distance_range(
                query.lower_bound, query.upper_bound
            )
            if query.distance_type is not None:
                async_query = async_query.distance_type(query.distance_type)
            if query.minimum_nprobes is not None and query.maximum_nprobes is not None:
                # Set both to the minimum first to avoid min > max error.
                async_query = async_query.nprobes(
                    query.minimum_nprobes
                ).maximum_nprobes(query.maximum_nprobes)
            elif query.minimum_nprobes is not None:
                async_query = async_query.minimum_nprobes(query.minimum_nprobes)
            elif query.maximum_nprobes is not None:
                async_query = async_query.maximum_nprobes(query.maximum_nprobes)
            if query.refine_factor is not None:
                async_query = async_query.refine_factor(query.refine_factor)
            if query.vector_column:
                async_query = async_query.column(query.vector_column)
            if query.ef:
                async_query = async_query.ef(query.ef)
            if query.bypass_vector_index:
                async_query = async_query.bypass_vector_index()

        if query.postfilter:
            async_query = async_query.postfilter()

        if query.full_text_query:
            async_query = async_query.nearest_to_text(
                query.full_text_query.query, query.full_text_query.columns
            )

        return async_query

    async def _execute_query(
        self,
        query: Query,
        *,
        batch_size: Optional[int] = None,
        timeout: Optional[timedelta] = None,
    ) -> pa.RecordBatchReader:
        if (
            _should_push_down_query_table(
                self._namespace_client, self._pushdown_operations
            )
            and not self._route_pushdown_to_rust
            and query.take_offsets is None
        ):
            from lancedb.namespace import _execute_server_side_query

            table_id = self._namespace_path + [self.name]
            return _execute_server_side_query(self._namespace_client, table_id, query)

        # The sync table calls into this method, so we need to map the
        # query to the async version of the query and run that here. This is only
        # used for that code path right now.

        async_query = self._sync_query_to_async(query)

        return await async_query.to_batches(
            max_batch_length=batch_size, timeout=timeout
        )

    async def _explain_plan(self, query: Query, verbose: Optional[bool]) -> str:
        # This method is used by the sync table
        async_query = self._sync_query_to_async(query)
        return await async_query.explain_plan(verbose)

    async def _analyze_plan(
        self,
        query: Query,
        *,
        distributed_metrics: AnalyzePlanDistributedMetrics = "aggregate",
    ) -> str:
        # This method is used by the sync table
        async_query = self._sync_query_to_async(query)
        return await async_query.analyze_plan(distributed_metrics)

    async def _output_schema(self, query: Query) -> pa.Schema:
        async_query = self._sync_query_to_async(query)
        return await async_query.output_schema()

    async def _do_merge(
        self,
        merge: LanceMergeInsertBuilder,
        new_data: DATA,
        on_bad_vectors: OnBadVectorsType,
        fill_value: float,
    ) -> MergeResult:
        schema = await self.schema()
        if on_bad_vectors is None:
            on_bad_vectors = "error"
        if fill_value is None:
            fill_value = 0.0
        data = _sanitize_data(
            new_data,
            schema,
            metadata=schema.metadata,
            on_bad_vectors=on_bad_vectors,
            fill_value=fill_value,
            allow_subschema=True,
        )
        if isinstance(data, pa.Table):
            data = pa.RecordBatchReader.from_batches(data.schema, data.to_batches())
        return await self._inner.execute_merge_insert(
            data,
            dict(
                on=merge._on,
                when_matched_update_all=merge._when_matched_update_all,
                when_matched_update_all_condition=merge._when_matched_update_all_condition,
                when_not_matched_insert_all=merge._when_not_matched_insert_all,
                when_not_matched_by_source_delete=merge._when_not_matched_by_source_delete,
                when_not_matched_by_source_condition=merge._when_not_matched_by_source_condition,
                when_not_matched_by_source_condition_expr=merge._when_not_matched_by_source_condition_expr,
                timeout=merge._timeout,
                use_index=merge._use_index,
                use_lsm=merge._use_lsm,
                validate_single_shard=merge._validate_single_shard,
            ),
        )

    async def delete(self, where: Union[str, Expr]) -> DeleteResult:
        """Delete rows from the table.

        This can be used to delete a single row, many rows, all rows, or
        sometimes no rows (if your predicate matches nothing).

        Parameters
        ----------
        where: str or :class:`~lancedb.expr.Expr`
            The filter condition. Can be a SQL string or a type-safe
            :class:`~lancedb.expr.Expr` built with :func:`~lancedb.expr.col`
            and :func:`~lancedb.expr.lit`.

            The filter must not be empty, or it will error.

        Examples
        --------
        >>> import lancedb
        >>> data = [
        ...    {"x": 1, "vector": [1.0, 2]},
        ...    {"x": 2, "vector": [3.0, 4]},
        ...    {"x": 3, "vector": [5.0, 6]}
        ... ]
        >>> db = lancedb.connect("./.lancedb")
        >>> table = db.create_table("my_table", data)
        >>> table.to_pandas()
           x      vector
        0  1  [1.0, 2.0]
        1  2  [3.0, 4.0]
        2  3  [5.0, 6.0]
        >>> table.delete("x = 2")
        DeleteResult(num_deleted_rows=1, version=2)
        >>> table.to_pandas()
           x      vector
        0  1  [1.0, 2.0]
        1  3  [5.0, 6.0]

        If you have a list of values to delete, you can combine them into a
        stringified list and use the `IN` operator:

        >>> to_remove = [1, 5]
        >>> to_remove = ", ".join([str(v) for v in to_remove])
        >>> to_remove
        '1, 5'
        >>> table.delete(f"x IN ({to_remove})")
        DeleteResult(num_deleted_rows=1, version=3)
        >>> table.to_pandas()
           x      vector
        0  3  [5.0, 6.0]
        """
        predicate = where._inner if isinstance(where, Expr) else where
        return await self._inner.delete(predicate)

    async def update(
        self,
        updates: Optional[Dict[str, Any]] = None,
        *,
        where: Optional[Union[str, Expr]] = None,
        updates_sql: Optional[Dict[str, str]] = None,
    ) -> UpdateResult:
        """
        This can be used to update zero to all rows in the table.

        If a filter is provided with `where` then only rows matching the
        filter will be updated.  Otherwise all rows will be updated.

        Parameters
        ----------
        updates: dict, optional
            The updates to apply.  The keys should be the name of the column to
            update.  The values should be the new values to assign.  This is
            required unless updates_sql is supplied.
        where: str or [Expr][lancedb.expr.Expr], optional
            The filter condition. Can be a SQL string or a type-safe
            [Expr][lancedb.expr.Expr] built with [col][lancedb.expr.col] and
            [lit][lancedb.expr.lit]. Only rows that satisfy this filter will
            be updated.
        updates_sql: dict, optional
            The updates to apply, expressed as SQL expression strings.  The keys should
            be column names. The values should be SQL expressions.  These can be SQL
            literals (e.g. "7" or "'foo'") or they can be expressions based on the
            previous value of the row (e.g. "x + 1" to increment the x column by 1)

        Returns
        -------
        UpdateResult
            An object containing:
            - rows_updated: The number of rows that were updated
            - version: The new version number of the table after the update

        Examples
        --------
        >>> import asyncio
        >>> import lancedb
        >>> from lancedb.expr import col
        >>> import pandas as pd
        >>> async def demo_update():
        ...     data = pd.DataFrame({"x": [1, 2], "vector": [[1, 2], [3, 4]]})
        ...     db = await lancedb.connect_async("./.lancedb")
        ...     table = await db.create_table("my_table", data)
        ...     # x is [1, 2], vector is [[1, 2], [3, 4]]
        ...     await table.update({"vector": [10, 10]}, where=col("x") == 2)
        ...     # x is [1, 2], vector is [[1, 2], [10, 10]]
        ...     await table.update(updates_sql={"x": "x + 1"})
        ...     # x is [2, 3], vector is [[1, 2], [10, 10]]
        >>> asyncio.run(demo_update())
        """
        if updates is not None and updates_sql is not None:
            raise ValueError("Only one of updates or updates_sql can be provided")
        if updates is None and updates_sql is None:
            raise ValueError("Either updates or updates_sql must be provided")

        if updates is not None:
            updates_sql = {k: value_to_sql(v) for k, v in updates.items()}

        predicate = where.to_sql() if isinstance(where, Expr) else where
        return await self._inner.update(updates_sql, predicate)

    async def add_columns(
        self,
        transforms: dict[str, str | FunctionApplication]
        | FunctionApplication
        | pa.Field
        | List[pa.Field]
        | pa.Schema
        | None = None,
        *,
        computed: dict[str, str] | None = None,
    ) -> AddColumnsResult:
        """
        Add new columns with defined values.

        Parameters
        ----------
        transforms: Dict[str, str | FunctionApplication] or FunctionApplication
            A map of column name to a SQL expression to use to calculate the
            value of the new column. These expressions will be evaluated for
            each row in the table, and can reference existing columns.
            Alternatively, you can pass a pyarrow field or schema to add
            new columns with NULLs.

            A mapping with one ``FunctionApplication`` value keeps its scalar
            or named-struct result in the named table column. A bare
            named-struct application expands its ordered result fields as one
            atomic binding; aliases come from ``rename(columns=...)``.
            Function columns are supported only on LanceDB Cloud and
            Enterprise.
        computed: Dict[str, str], optional
            A mapping from output column names to SQL expressions derives each
            output field from its expression. A direct projection of a Blob v2
            field inherits Blob v2 semantics; other expressions derive their
            ordinary Arrow type. Mapping order is declaration and dependency
            order.

            Unlike ``transforms``, the expression is stored rather than
            evaluated now: the column is committed with no values, and rows get
            them from
            [`refresh_column`][lancedb.table.AsyncTable.refresh_column].

            A refresh does not revisit rows it has already filled, so mutating
            an input leaves the value computed at fill time. While a
            declaration reads a column, that column cannot be renamed, retyped
            or dropped.

            On LanceDB Cloud and Enterprise the expression is planned by
            the server. Cannot be combined with ``transforms``.

        Returns
        -------
        AddColumnsResult
            version: the new version number of the table after adding columns.

        """
        function_application = None
        function_output_name = None
        if isinstance(transforms, FunctionApplication):
            function_application = transforms
        elif isinstance(transforms, dict) and any(
            isinstance(value, FunctionApplication) for value in transforms.values()
        ):
            if len(transforms) != 1 or not all(
                isinstance(value, FunctionApplication) for value in transforms.values()
            ):
                raise ValueError(
                    "one add_columns call declares exactly one Function binding"
                )
            function_output_name, function_application = next(iter(transforms.items()))

        if function_application is not None:
            if computed:
                raise ValueError(
                    "add_columns cannot mix a Function application with SQL "
                    "computed columns"
                )
            function_application._ensure_declarable()
            return await self._inner.add_function_columns(
                function_application.to_canonical_json(), function_output_name
            )

        if isinstance(transforms, pa.Field):
            transforms = [transforms]
        if isinstance(transforms, list) and all(
            {isinstance(f, pa.Field) for f in transforms}
        ):
            transforms = pa.schema(transforms)
        if computed:
            if transforms:
                raise ValueError(
                    "add_columns cannot take both transforms and computed columns"
                )
            return await self._inner.add_computed_columns(list(computed.items()))
        if transforms is None:
            raise ValueError("add_columns requires transforms or computed columns")
        if isinstance(transforms, pa.Schema):
            return await self._inner.add_columns_with_schema(transforms)
        else:
            return await self._inner.add_columns(list(transforms.items()))

    async def refresh_column(self, column: str) -> RefreshColumnResult:
        """
        Fill the rows of a computed column that hold no value yet.

        Declared with ``add_columns(computed=...)``, a column starts empty and
        gets its values here. Rows appended since the last refresh are filled
        by the next one; rows already filled are left as they are, so the call
        is idempotent and does not observe a mutated input.

        Local tables only: a remote refresh runs as a server job, through
        [`refresh_column_async`][lancedb.table.Table.refresh_column_async].

        Parameters
        ----------
        column: str
            The name of the computed column to fill.

        Returns
        -------
        RefreshColumnResult
            The number of rows filled and the new version of the table.
        """
        return await self._inner.refresh_column(column)

    async def refresh_column_async(
        self, column: str
    ) -> AsyncJob[RefreshColumnJobResult]:
        """
        Like :meth:`refresh_column`, but returns a handle to the refresh job
        instead of blocking until it completes.

        The job may already be complete when returned; callers must not assume
        the column is filled until :meth:`AsyncJob.wait` resolves. Invalid
        input -- an unknown column, or one that is not computed -- raises here
        rather than failing the job. On local tables the job runs
        in-process; on LanceDB Cloud and Enterprise it is the server's
        backfill job.

        Returns
        -------
        AsyncJob[RefreshColumnResult]
            A job whose successful ``wait`` returns row counts plus the source
            and published table versions.

        Examples
        --------
        >>> import asyncio
        >>> import lancedb
        >>> async def refresh_in_background():
        ...     db = await lancedb.connect_async("./.lancedb")
        ...     table = await db.create_table("computed_job_async_demo", [{"x": 1}])
        ...     await table.add_columns(computed={"doubled": "x * 2"})
        ...     job = await table.refresh_column_async("doubled")
        ...     result = await job.wait()
        ...     assert result.rows_assigned == 1
        ...     return await job.status()
        >>> asyncio.run(refresh_in_background())
        'finished'
        """
        return _typed_job(
            await self._inner.refresh_column_async(column),
            RefreshColumnJobResult.from_json,
        )

    async def alter_columns(
        self, *alterations: Iterable[dict[str, Any]]
    ) -> AlterColumnsResult:
        """
        Alter column names and nullability.

        alterations : Iterable[Dict[str, Any]]
            A sequence of dictionaries, each with the following keys:
            - "path": str
                The column path to alter. For a top-level column, this is the name.
                For a nested column, this is the dot-separated path, e.g. "a.b.c".
            - "rename": str, optional
                The new name of the column. If not specified, the column name is
                not changed.
            - "data_type": pyarrow.DataType, optional
               The new data type of the column. Existing values will be casted
               to this type. If not specified, the column data type is not changed.
            - "nullable": bool, optional
                Whether the column should be nullable. If not specified, the column
                nullability is not changed. Only non-nullable columns can be changed
                to nullable. Currently, you cannot change a nullable column to
                non-nullable.

        Returns
        -------
        AlterColumnsResult
            version: the new version number of the table after the alteration.
        """
        return await self._inner.alter_columns(alterations)

    async def update_field_metadata(
        self, *updates: dict[str, Any]
    ) -> UpdateFieldMetadataResult:
        """Update per-field metadata. See
        [`Table.update_field_metadata`][lancedb.table.Table.update_field_metadata]."""
        return await self._inner.update_field_metadata(updates)

    async def drop_columns(self, columns: Iterable[str]):
        """
        Drop columns from the table.

        Parameters
        ----------
        columns : Iterable[str]
            The names of the columns to drop.
        """
        return await self._inner.drop_columns(columns)

    async def version(self) -> int:
        """
        Retrieve the version of the table

        LanceDb supports versioning.  Every operation that modifies the table increases
        version.  As long as a version hasn't been deleted you can `[Self::checkout]`
        that version to view the data at that point.  In addition, you can
        `[Self::restore]` the version to replace the current table with a previous
        version.
        """
        return await self._inner.version()

    async def list_versions(self):
        """
        List all versions of the table
        """
        versions = await self._inner.list_versions()
        for v in versions:
            ts_nanos = v["timestamp"]
            v["timestamp"] = datetime.fromtimestamp(ts_nanos // 1e9) + timedelta(
                microseconds=(ts_nanos % 1e9) // 1e3
            )

        return versions

    async def checkout(self, version: int | str):
        """
        Checks out a specific version of the Table

        Any read operation on the table will now access the data at the checked out
        version. As a consequence, calling this method will disable any read consistency
        interval that was previously set.

        This is a read-only operation that turns the table into a sort of "view"
        or "detached head".  Other table instances will not be affected.  To make the
        change permanent you can use the `[Self::restore]` method.

        Any operation that modifies the table will fail while the table is in a checked
        out state. To return the table to a normal state use
        `[Self::checkout_latest]`.

        Parameters
        ----------
        version: int | str,
            The version to check out. A version number (`int`) or a tag
            (`str`) can be provided.
        """
        try:
            await self._inner.checkout(version)
        except RuntimeError as e:
            if "not found" in str(e):
                raise ValueError(
                    f"Version {version} no longer exists. Was it cleaned up?"
                )
            else:
                raise

    async def checkout_latest(self):
        """
        Ensures the table is pointing at the latest version

        This can be used to manually update a table when the read_consistency_interval
        is None
        It can also be used to undo a `[Self::checkout]` operation
        """
        await self._inner.checkout_latest()

    async def restore(self, version: Optional[int | str] = None):
        """
        Restore the table to the currently checked out version

        This operation will fail if checkout has not been called previously

        This operation will overwrite the latest version of the table with a
        previous version.  Any changes made since the checked out version will
        no longer be visible.

        Once the operation concludes the table will no longer be in a checked
        out state and the read_consistency_interval, if any, will apply.
        """
        await self._inner.restore(version)

    def take_offsets(self, offsets: list[int]) -> AsyncTakeQuery:
        """
        Take a list of offsets from the table.

        Offsets are 0-indexed and relative to the current version of the table.  Offsets
        are not stable.  A row with an offset of N may have a different offset in a
        different version of the table (e.g. if an earlier row is deleted).

        Offsets are mostly useful for sampling as the set of all valid offsets is easily
        known in advance to be [0, len(table)).

        No guarantees are made regarding the order in which results are returned.
        Repeated offsets produce repeated rows.

        Parameters
        ----------
        offsets: list[int]
            The offsets to take.

        Returns
        -------
        pa.RecordBatch
            A record batch containing the rows at the given offsets.
        """
        return AsyncTakeQuery(self._inner.take_offsets(offsets), self)

    def take_row_ids(self, row_ids: list[int]) -> AsyncTakeQuery:
        """
        Take a list of row ids from the table.

        Row ids are not stable and are relative to the current version of the table.
        They can change due to compaction and updates.

        Unlike offsets, row ids are not 0-indexed and no assumptions should be made
        about the possible range of row ids.  In order to use this method you must
        first obtain the row ids by scanning or searching the table.

        Even so, row ids are more stable than offsets and can be useful in some
        situations.

        There is an ongoing effort to make row ids stable which is tracked at
        https://github.com/lancedb/lancedb/issues/1120

        Parameters
        ----------
        row_ids: list[int]
            The row ids to take.

        Returns
        -------
        AsyncTakeQuery
            A query object that can be executed to get the rows.
        """
        return AsyncTakeQuery(self._inner.take_row_ids(row_ids), self)

    async def blob_columns(self) -> list[str]:
        return await self._inner.blob_columns()

    async def fetch_blobs(
        self, column: str, row_ids: Union[list[int], pa.Table]
    ) -> pa.LargeBinaryArray:
        return await self._inner.fetch_blobs(
            column, _normalize_blob_row_ids(row_ids, column)
        )

    async def fetch_blob_ranges(
        self,
        column: str,
        requests: Sequence[Tuple[int, int, int]],
    ) -> pa.LargeBinaryArray:
        return await self._inner.fetch_blob_ranges(column, list(requests))

    async def fetch_blob_files(
        self, column: str, row_ids: Union[list[int], pa.Table]
    ) -> "list[Optional[BlobFile]]":
        handles = await self._inner.fetch_blob_files(
            column, _normalize_blob_row_ids(row_ids, column)
        )
        return _wrap_blob_files(handles)

    @property
    def tags(self) -> AsyncTags:
        """Tag management for the dataset.

        Similar to Git, tags are a way to add metadata to a specific version of the
        dataset.

        .. warning::

            Tagged versions are exempted from the
            :py:meth:`optimize(cleanup_older_than)` process.

            To remove a version that has been tagged, you must first
            :py:meth:`~Tags.delete` the associated tag.

        """
        return AsyncTags(self._inner)

    @property
    def branches(self) -> AsyncBranches:
        """Branch management for the table.

        Branches are isolated, writable lines of history forked from another
        branch (or version). Writes on a branch do not affect ``main``.
        """
        return AsyncBranches(self._inner)

    def current_branch(self) -> Optional[str]:
        """The branch this table handle is scoped to, or ``None`` for ``main``."""
        return self._inner.current_branch()

    async def optimize(
        self,
        *,
        cleanup_older_than: Optional[timedelta] = None,
        delete_unverified: bool = False,
        retrain=False,
    ) -> OptimizeStats:
        """
        Optimize the on-disk data and indices for better performance.

        Modeled after ``VACUUM`` in PostgreSQL.

        Optimization covers three operations:

         * Compaction: Merges small files into larger ones
         * Prune: Removes old versions of the dataset
         * Index: Optimizes the indices, adding new data to existing indices

        Parameters
        ----------
        cleanup_older_than: timedelta, optional default 7 days
            All files belonging to versions older than this will be removed.  Set
            to 0 days to remove all versions except the latest.  The latest version
            is never removed.
        delete_unverified: bool, default False
            Files leftover from a failed transaction may appear to be part of an
            in-progress operation (e.g. appending new data) and these files will not
            be deleted unless they are at least 7 days old. If delete_unverified is True
            then these files will be deleted regardless of their age.

            .. warning::

                This should only be set to True if you can guarantee that no other
                process is currently working on this dataset. Otherwise the dataset
                could be put into a corrupted state.

        retrain: bool, default False
            This parameter is no longer used and is deprecated.

        Notes
        -----
        The frequency an application should call optimize is based on the frequency of
        data modifications.  If data is frequently added, deleted, or updated then
        optimize should be run frequently.  A good rule of thumb is to run optimize if
        you have added or modified 100,000 or more records or run more than 20 data
        modification operations.
        """
        cleanup_since_ms: Optional[int] = None
        if cleanup_older_than is not None:
            cleanup_since_ms = round(cleanup_older_than.total_seconds() * 1000)

        if retrain:
            import warnings

            warnings.warn(
                "The 'retrain' parameter is deprecated and will be removed in a "
                "future version.",
                DeprecationWarning,
                stacklevel=2,
            )

        return await self._inner.optimize(
            cleanup_since_ms=cleanup_since_ms,
            delete_unverified=delete_unverified,
        )

    async def list_indices(self) -> Iterable[IndexConfig]:
        """
        List all indices that have been created with Self::create_index
        """
        return await self._inner.list_indices()

    async def tokenize(
        self,
        query: str,
        *,
        column: Optional[str] = None,
        index_name: Optional[str] = None,
    ) -> Iterable[FtsToken]:
        """
        Tokenize a query using the tokenizer configured on an FTS index.

        Specify exactly one of ``column`` or ``index_name``.

        Model-backed tokenizers such as ``jieba/*`` and ``lindera/*`` are
        rebuilt in the client process from index metadata. For remote tables,
        this means the same tokenizer model files must also exist locally.
        """
        return await self._inner.tokenize(query, column=column, index_name=index_name)

    async def index_stats(self, index_name: str) -> Optional[IndexStatistics]:
        """
        Retrieve statistics about an index

        Parameters
        ----------
        index_name: str
            The name of the index to retrieve statistics for

        Returns
        -------
        IndexStatistics or None
            The statistics about the index. Returns None if the index does not exist.
        """
        stats = await self._inner.index_stats(index_name)
        if stats is None:
            return None
        else:
            return IndexStatistics(**stats)

    async def uses_v2_manifest_paths(self) -> bool:
        """
        Check if the table is using the new v2 manifest paths.

        Returns
        -------
        bool
            True if the table is using the new v2 manifest paths, False otherwise.
        """
        return await self._inner.uses_v2_manifest_paths()

    async def migrate_manifest_paths_v2(self):
        """
        Migrate the manifest paths to the new format.

        This will update the manifest to use the new v2 format for paths.

        This function is idempotent, and can be run multiple times without
        changing the state of the object store.

        !!! danger

            This should not be run while other concurrent operations are happening.
            And it should also run until completion before resuming other operations.

        You can use
        [AsyncTable.uses_v2_manifest_paths][lancedb.table.AsyncTable.uses_v2_manifest_paths]
        to check if the table is already using the new path style.
        """
        await self._inner.migrate_manifest_paths_v2()

    @deprecation.deprecated(
        deprecated_in="0.33.1",
        current_version=__version__,
        details="Use update_field_metadata() instead.",
    )
    async def replace_field_metadata(
        self, field_name: str, new_metadata: dict[str, str]
    ):
        """
        Replace the metadata of a field in the schema

        .. deprecated:: 0.33.1
            Use :func:`update_field_metadata` instead.

        Parameters
        ----------
        field_name: str
            The name of the field to replace the metadata for
        new_metadata: dict
            The new metadata to set
        """
        await self._inner.replace_field_metadata(field_name, new_metadata)


@dataclass
class IndexStatistics:
    """
    Statistics about an index.

    Attributes
    ----------
    num_indexed_rows: int
        The number of rows that are covered by this index.
    num_unindexed_rows: int
        The number of rows that are not covered by this index.
    index_type: str
        The type of index that was created.
    distance_type: Optional[str]
        The distance type used by the index.
    num_indices: Optional[int]
        The number of parts the index is split into.
    """

    num_indexed_rows: int
    num_unindexed_rows: int
    index_type: Literal[
        "IVF_FLAT",
        "IVF_SQ",
        "IVF_PQ",
        "IVF_RQ",
        "IVF_HNSW_SQ",
        "IVF_HNSW_PQ",
        "IVF_HNSW_FLAT",
        "FTS",
        "BTREE",
        "BITMAP",
        "LABEL_LIST",
    ]
    distance_type: Optional[Literal["l2", "cosine", "dot"]] = None
    num_indices: Optional[int] = None

    # This exists for backwards compatibility with an older API, which returned
    # a dictionary instead of a class.
    def __getitem__(self, key):
        return getattr(self, key)


@dataclass
class TableStatistics:
    """
    Statistics about a table and fragments.

    Attributes
    ----------
    total_bytes: int
        The total size, in bytes, of the table's data files, index files, and
        overlay files. Read from the manifest, so this excludes deletion files
        and manifests.
    num_rows: int
        The total number of rows in the table.
    num_indices: int
        The total number of indices in the table.
    fragment_stats: FragmentStatistics
        Statistics about fragments in the table.
    """

    total_bytes: int
    num_rows: int
    num_indices: int
    fragment_stats: FragmentStatistics


@dataclass
class FragmentStatistics:
    """
    Statistics about fragments.

    Attributes
    ----------
    num_fragments: int
        The total number of fragments in the table.
    num_small_fragments: int
        The total number of small fragments in the table.
        Small fragments have low row counts and may need to be compacted.
    lengths: FragmentSummaryStats
        Statistics about the number of rows in the table fragments.
    """

    num_fragments: int
    num_small_fragments: int
    lengths: FragmentSummaryStats


@dataclass
class FragmentSummaryStats:
    """
    Statistics about fragments sizes

    Attributes
    ----------
    min: int
        The number of rows in the fragment with the fewest rows.
    max: int
        The number of rows in the fragment with the most rows.
    mean: int
        The mean number of rows in the fragments.
    p25: int
        The 25th percentile of number of rows in the fragments.
    p50: int
        The 50th percentile of number of rows in the fragments.
    p75: int
        The 75th percentile of number of rows in the fragments.
    p99: int
        The 99th percentile of number of rows in the fragments.
    """

    min: int
    max: int
    mean: int
    p25: int
    p50: int
    p75: int
    p99: int


class Tags:
    """
    Table tag manager.
    """

    def __init__(self, table):
        self._table = table

    def list(self) -> Dict[str, Tag]:
        """
        List all table tags.

        Returns
        -------
        dict[str, Tag]
            A dictionary mapping tag names to version numbers.
        """
        return LOOP.run(self._table.tags.list())

    def get_version(self, tag: str) -> int:
        """
        Get the version of a tag.

        Parameters
        ----------
        tag: str,
            The name of the tag to get the version for.
        """
        return LOOP.run(self._table.tags.get_version(tag))

    def create(self, tag: str, version: int) -> None:
        """
        Create a tag for a given table version.

        Parameters
        ----------
        tag: str,
            The name of the tag to create. This name must be unique among all tag
            names for the table.
        version: int,
            The table version to tag.
        """
        LOOP.run(self._table.tags.create(tag, version))

    def delete(self, tag: str) -> None:
        """
        Delete tag from the table.

        Parameters
        ----------
        tag: str,
            The name of the tag to delete.
        """
        LOOP.run(self._table.tags.delete(tag))

    def update(self, tag: str, version: int) -> None:
        """
        Update tag to a new version.

        Parameters
        ----------
        tag: str,
            The name of the tag to update.
        version: int,
            The new table version to tag.
        """
        LOOP.run(self._table.tags.update(tag, version))


class Branches:
    """
    Table branch manager.
    """

    def __init__(self, parent: "LanceTable"):
        self._parent = parent
        self._table = parent._table

    def list(self) -> Dict[str, Any]:
        """List all branches, mapping name to branch metadata."""
        return LOOP.run(self._table.branches.list())

    def create(
        self,
        name: str,
        from_ref: Optional[str] = None,
        from_version: Optional[int] = None,
    ) -> "Table":
        """Create a branch and return a handle scoped to it.

        Parameters
        ----------
        name: str
            Name of the new branch.
        from_ref: str, optional
            Source branch to fork from. Defaults to ``main``.
        from_version: int, optional
            A specific version on ``from_ref`` to fork from. Defaults to latest.
        """
        async_table = LOOP.run(
            self._table.branches.create(name, from_ref, from_version)
        )
        return self._wrap(async_table)

    def checkout(self, name: str, version: Optional[int] = None) -> "Table":
        """Check out an existing branch and return a handle scoped to it.

        Parameters
        ----------
        name: str
            Name of the branch to check out.
        version: int, optional
            A specific version on the branch to pin. When set, the returned
            handle is a read-only view of that version; when omitted it tracks
            the branch's latest and stays writable.
        """
        async_table = LOOP.run(self._table.branches.checkout(name, version))
        return self._wrap(async_table, version)

    def delete(self, name: str) -> None:
        """Delete a branch."""
        LOOP.run(self._table.branches.delete(name))

    def diff(self, from_branch: str) -> Dict[str, Any]:
        """Diff a branch against main."""
        return LOOP.run(self._table.branches.diff(from_branch))

    def cherry_pick(self, from_branch: str, dry_run: bool = False) -> Dict[str, Any]:
        """Cherry-pick a branch onto main, or dry-run.

        Parameters
        ----------
        from_branch: str
            Branch to cherry-pick from.
        dry_run: bool, default False
            When True, only preview. When False, attempt the cherry-pick.

        Notes
        -----
        A failed cherry-pick returns ``status="failed"`` instead of raising.
        """
        return LOOP.run(self._table.branches.cherry_pick(from_branch, dry_run))

    def _wrap(
        self, async_table: "AsyncTable", version: Optional[int] = None
    ) -> "Table":
        # Delegate to the parent so the branch handle keeps its concrete type
        # (LanceTable / RemoteTable) and connection context; `version` is the
        # explicit pin so a remote handle can restore branch+version on reopen.
        return self._parent._wrap_branch_handle(async_table, version)


class AsyncTags:
    """
    Async table tag manager.
    """

    def __init__(self, table):
        self._table = table

    async def list(self) -> Dict[str, Tag]:
        """
        List all table tags.

        Returns
        -------
        dict[str, Tag]
            A dictionary mapping tag names to version numbers.
        """
        return await self._table.tags.list()

    async def get_version(self, tag: str) -> int:
        """
        Get the version of a tag.

        Parameters
        ----------
        tag: str,
            The name of the tag to get the version for.
        """
        return await self._table.tags.get_version(tag)

    async def create(self, tag: str, version: int) -> None:
        """
        Create a tag for a given table version.

        Parameters
        ----------
        tag: str,
            The name of the tag to create. This name must be unique among all tag
            names for the table.
        version: int,
            The table version to tag.
        """
        await self._table.tags.create(tag, version)

    async def delete(self, tag: str) -> None:
        """
        Delete tag from the table.

        Parameters
        ----------
        tag: str,
            The name of the tag to delete.
        """
        await self._table.tags.delete(tag)

    async def update(self, tag: str, version: int) -> None:
        """
        Update tag to a new version.

        Parameters
        ----------
        tag: str,
            The name of the tag to update.
        version: int,
            The new table version to tag.
        """
        await self._table.tags.update(tag, version)


class AsyncBranches:
    """Async table branch manager."""

    def __init__(self, table):
        self._table = table

    async def list(self) -> Dict[str, Any]:
        """List all branches, mapping name to branch metadata."""
        return await self._table.branches.list()

    async def create(
        self,
        name: str,
        from_ref: Optional[str] = None,
        from_version: Optional[int] = None,
    ) -> "AsyncTable":
        """Create a branch and return a handle scoped to it.

        Parameters
        ----------
        name: str
            Name of the new branch.
        from_ref: str, optional
            Source branch to fork from. Defaults to ``main``.
        from_version: int, optional
            A specific version on ``from_ref`` to fork from. Defaults to latest.
        """
        # "main" and None are two spellings of the root branch in lance; normalize
        # so from_ref="main" behaves identically to the default.
        if from_ref == "main":
            from_ref = None
        inner = await self._table.branches.create(name, from_ref, from_version)
        return AsyncTable(inner)

    async def checkout(self, name: str, version: Optional[int] = None) -> "AsyncTable":
        """Check out an existing branch and return a handle scoped to it.

        Parameters
        ----------
        name: str
            Name of the branch to check out.
        version: int, optional
            A specific version on the branch to pin. When set, the returned
            handle is a read-only view of that version; when omitted it tracks
            the branch's latest and stays writable.
        """
        return AsyncTable(await self._table.branches.checkout(name, version))

    async def delete(self, name: str) -> None:
        """Delete a branch."""
        await self._table.branches.delete(name)

    async def diff(self, from_branch: str) -> Dict[str, Any]:
        """Diff a branch against main."""
        return await self._table.branches.diff(from_branch)

    async def cherry_pick(
        self, from_branch: str, dry_run: bool = False
    ) -> Dict[str, Any]:
        """Cherry-pick a branch onto main, or dry-run.

        A failed cherry-pick returns ``status="failed"`` instead of raising.
        """
        return await self._table.branches.cherry_pick(from_branch, dry_run)
