diff --git a/README.md b/README.md index 661de0d..6f632c2 100644 --- a/README.md +++ b/README.md @@ -74,12 +74,38 @@ no `fragment_ids` instead). > **Breaking change (from 0.5.0):** the `segmented` parameter was removed — -> the distributed segment-index workflow is now the only code path, and the -> single-node fallbacks are gone. `replace` now defaults to `True` (matching +> the distributed segment-index workflow is now the only code path, and +> the single-node fallbacks are gone. `replace` now defaults to `True` (matching > pylance). Indexes created by older versions of > `create_scalar_index` on `INVERTED` columns may carry empty index metadata > (see #69); rebuilding them with `replace=True` records full metadata. +#### Index Maintenance + +Appended data is not indexed automatically — queries stay correct (uncovered +fragments fall back to scans) but slow down as the unindexed share grows. +`optimize_indices` restores index health on the dataset's latest version: +it indexes newly appended fragments, merges small segments, and heals stale +fragment IDs left inside mixed segments by deletes as part of a commit that +indexes or merges new data. It commits no new version when there is no new +data to index and no segments to merge. + +```python +from daft_lance import optimize_indices + +stats = optimize_indices("s3://bucket/my_dataset") +stats = optimize_indices("s3://bucket/my_dataset", indices=["name_idx"], num_indices_to_merge=4) +``` + +Like lance-ray, `optimize_indices` delegates to pylance's +`DatasetOptimizer.optimize_indices` and runs in the coordinator process; for +a distributed rebuild use `create_scalar_index(..., replace=True)`. It +returns `OptimizeIndicesStats` with the dataset versions immediately before +and after the call, the duration, and per-index segment/coverage counts +(coverage counts only fragments still live in the manifest); unknown or +empty `indices` raise `ValueError`, and duplicates are ignored. + + ### Column Merging ```python diff --git a/daft_lance/__init__.py b/daft_lance/__init__.py index 1a0323d..033c085 100644 --- a/daft_lance/__init__.py +++ b/daft_lance/__init__.py @@ -9,15 +9,19 @@ create_scalar_index, merge_columns, merge_columns_df, + optimize_indices, read_lance, write_lance, ) +from .lance_scalar_index import OptimizeIndicesStats __all__ = [ + "OptimizeIndicesStats", "compact_files", "create_scalar_index", "merge_columns", "merge_columns_df", + "optimize_indices", "read_lance", "take_blobs", "write_lance", diff --git a/daft_lance/_lance.py b/daft_lance/_lance.py index 32aa78c..216b02b 100644 --- a/daft_lance/_lance.py +++ b/daft_lance/_lance.py @@ -16,7 +16,7 @@ from .lance_compaction import compact_files_internal from .lance_data_sink import LanceDataSink from .lance_merge_column import merge_columns_from_df, merge_columns_internal -from .lance_scalar_index import create_scalar_index_internal +from .lance_scalar_index import OptimizeIndicesStats, create_scalar_index_internal, optimize_indices_internal from .lance_scan import LanceScanOperator from .namespace import validate_uri_or_namespace from .utils import construct_lance_dataset_handle @@ -550,6 +550,110 @@ def create_scalar_index( ) +@PublicAPI +def optimize_indices( + uri: str | pathlib.Path | None = None, + io_config: IOConfig | None = None, + *, + table_id: list[str] | None = None, + namespace_impl: str | None = None, + namespace_properties: dict[str, str] | None = None, + indices: list[str] | None = None, + num_indices_to_merge: int | None = None, + storage_options: dict[str, Any] | None = None, + block_size: int | None = None, + commit_lock: Any | None = None, + index_cache_size: int | None = None, + default_scan_options: dict[str, Any] | None = None, + metadata_cache_size_bytes: int | None = None, +) -> OptimizeIndicesStats: + """Incrementally optimize existing indexes. + + As data is appended it is not added to existing indexes automatically: + queries keep working (uncovered fragments fall back to scans) but they + get slower as the unindexed share grows. This function restores index + health on the dataset's latest version: newly appended fragments are + indexed, small segments are merged, and stale fragment IDs left inside + mixed segments by deletes are healed as part of a commit that indexes + or merges new data. It commits no new version when there is no new + data to index and no segments to merge. + + Delegates to pylance's ``DatasetOptimizer.optimize_indices`` — the same + choice lance-ray makes — so it runs in the coordinator process (Lance + core parallelizes the underlying scans). A full distributed rebuild is + ``create_scalar_index(..., replace=True)``. + + Args: + uri: The URI of the Lance table (supports remote URLs to object stores such as `s3://` or `gs://`) + io_config: A custom IOConfig to use when accessing Lance data. Defaults to None. + table_id: Table identifier within the namespace, e.g. ["catalog", "schema", "table"]. + Mutually exclusive with ``uri``. + namespace_impl: Lance Namespace implementation, e.g. "dir" or "rest". + namespace_properties: Properties for connecting to the namespace, e.g. + {"root": "/data"} for "dir" or {"uri": "http://host:port"} for "rest". + indices: Names of the indexes to optimize. ``None`` (the default) + optimizes every index on the dataset. Unknown names and an empty + list raise ``ValueError``; duplicates are ignored. + num_indices_to_merge: How many segments to merge when compacting an + index (passed to pylance). ``0`` indexes the new data into a new + segment instead of merging; ``None`` uses pylance's default. + storage_options: Storage options for the dataset. + block_size: Block size for the dataset. + commit_lock: Commit lock for the dataset. + index_cache_size: Size of the index cache. + default_scan_options: Default scan options for the dataset. + metadata_cache_size_bytes: Size of the metadata cache in bytes. + + Returns: + OptimizeIndicesStats: versions of the dataset's latest snapshot + immediately before and after the call, wall-clock duration, and + per-index segment/coverage counts (coverage counts only fragments + still live in the manifest). ``changed`` is ``True`` when a new + version became visible during the call — with no concurrent + writers, exactly when this run committed one. + + Raises: + ValueError: If ``indices`` is empty or names indexes that do not + exist on the dataset. + + Examples: + >>> import daft_lance + >>> stats = daft_lance.optimize_indices("s3://my-bucket/dataset/") # doctest: +SKIP + >>> stats.changed # doctest: +SKIP + True + >>> [i.name for i in stats.indices] # doctest: +SKIP + ['name_idx'] + + Optimize one index and merge its small segments: + + >>> daft_lance.optimize_indices( # doctest: +SKIP + ... "s3://my-bucket/dataset/", indices=["name_idx"], num_indices_to_merge=4 + ... ) + """ + io_config = context.get_context().daft_planning_config.default_io_config if io_config is None else io_config + + dataset_handle = construct_lance_dataset_handle( + uri, + storage_options=storage_options, + io_config=io_config, + namespace_impl=namespace_impl, + namespace_properties=namespace_properties, + table_id=table_id, + block_size=block_size, + commit_lock=commit_lock, + index_cache_size=index_cache_size, + default_scan_options=default_scan_options, + metadata_cache_size_bytes=metadata_cache_size_bytes, + ) + + return optimize_indices_internal( + dataset_handle.dataset, + dataset_handle.worker_open_context(), + indices=indices, + num_indices_to_merge=num_indices_to_merge, + ) + + @PublicAPI def compact_files( uri: str | pathlib.Path | None = None, diff --git a/daft_lance/lance_scalar_index.py b/daft_lance/lance_scalar_index.py index 85d1509..e66c646 100644 --- a/daft_lance/lance_scalar_index.py +++ b/daft_lance/lance_scalar_index.py @@ -1,7 +1,9 @@ from __future__ import annotations +import dataclasses import logging import pickle +import time from typing import TYPE_CHECKING, Any, cast import daft @@ -464,3 +466,163 @@ def _validate_segments_against_manifest( "not covered by any built segment (a worker result was likely lost). Re-run the " "index build." ) + + +@dataclasses.dataclass(frozen=True) +class OptimizedIndexStats: + """Per-index outcome of an ``optimize_indices`` run. + + ``fragments_covered_*`` count only fragments that still exist in the + manifest: stale IDs left inside a segment by deletes do not count as + coverage. ``*_after`` fields are zero when the index is absent from the + after-snapshot (e.g. a concurrent writer dropped it). + """ + + name: str + segments_before: int + segments_after: int + fragments_covered_before: int + fragments_covered_after: int + + +@dataclasses.dataclass(frozen=True) +class OptimizeIndicesStats: + """Outcome of an ``optimize_indices`` run over one dataset. + + The versions are sampled from the dataset's latest version immediately + before and after the optimize call, so they describe the same lineage a + single writer sees. + """ + + version_before: int + version_after: int + duration_seconds: float + indices: list[OptimizedIndexStats] + + @property + def changed(self) -> bool: + """Whether a new dataset version became visible during the run. + + With no concurrent writers this is exactly whether this run + committed a version; a concurrent commit during the call also makes + it ``True``. + """ + return self.version_after != self.version_before + + +def _index_snapshot(lance_ds: lance.LanceDataset) -> dict[str, tuple[int, int]]: + """Map index name to ``(segment count, live covered-fragment count)``. + + Coverage counts only fragment IDs that still exist in the manifest, so + stale IDs left by deletes do not masquerade as coverage. Datasets with + legacy manifests that ``describe_indices`` cannot parse fall back to + ``list_indices`` names with unknown counts, the same degradation + ``create_scalar_index`` uses (``_existing_index_names``). + """ + live_fragments = {fragment.fragment_id for fragment in lance_ds.get_fragments()} + try: + descriptions = lance_ds.describe_indices() + except Exception: + logger.warning("describe_indices() failed; reporting index names only", exc_info=True) + try: + return {cast(dict[str, Any], idx)["name"]: (0, 0) for idx in lance_ds.list_indices()} + except Exception: + return {} + + snapshot: dict[str, tuple[int, int]] = {} + for desc in descriptions: + segments = desc.segments or [] + covered = {fid for segment in segments for fid in (segment.fragment_ids or ())} + snapshot[desc.name] = (len(segments), len(covered & live_fragments)) + return snapshot + + +def optimize_indices_internal( + lance_ds: lance.LanceDataset, + open_context: DatasetOpenContext, + *, + indices: list[str] | None = None, + num_indices_to_merge: int | None = None, +) -> OptimizeIndicesStats: + """Incrementally maintain existing indexes. + + Delegates to pylance's ``DatasetOptimizer.optimize_indices`` — the same + choice lance-ray makes — because Lance core owns the delta-index + semantics. One run indexes newly appended fragments, merges small + segments (``num_indices_to_merge``) when there are deltas to merge, and + heals stale fragment IDs left inside mixed segments **as part of a + commit that indexes or merges new data**; with no new data to index it + commits nothing and stale-only coverage is left as-is. Heavier changes + are a distributed rebuild: ``create_scalar_index(..., replace=True)``. + + ``indices`` is our parameter and gets deterministic semantics here + because pylance silently ignores unknown names: an empty list raises, + unknown names raise listing the available indexes, and duplicates are + ignored (each index is optimized and reported once). Merge-count + validation and everything about index internals belong to Lance. + + Both stat snapshots are sampled from the dataset's latest version, so + they are never mixed across snapshots even if the caller pinned the + handle's version. + """ + if indices is not None: + if len(indices) == 0: + raise ValueError("indices must be a non-empty list of index names; pass None to optimize all indexes.") + unique_indices = list(dict.fromkeys(indices)) + duplicates = sorted({name for name in unique_indices if indices.count(name) > 1}) + if duplicates: + logger.warning("Duplicate index names %s were given; each index is optimized once.", duplicates) + indices = unique_indices + + base = open_context.open_latest() + before = _index_snapshot(base) + if indices is not None: + unknown = sorted(set(indices) - before.keys()) + if unknown: + raise ValueError(f"indices {unknown} do not exist on the dataset. Available index names: {sorted(before)}") + + call_kwargs: dict[str, Any] = {} + if indices is not None: + call_kwargs["index_names"] = list(indices) + if num_indices_to_merge is not None: + call_kwargs["num_indices_to_merge"] = num_indices_to_merge + + logger.info( + "Optimizing indices: uri=%s, indices=%s, num_indices_to_merge=%s", + open_context.uri, + indices if indices is not None else "(all)", + num_indices_to_merge, + ) + version_before = base.version + start = time.monotonic() + lance_ds.optimize.optimize_indices(**call_kwargs) # type: ignore[no-untyped-call] + duration = time.monotonic() - start + + latest = open_context.open_latest() + after = _index_snapshot(latest) + + selected = indices if indices is not None else sorted(before) + per_index = [ + OptimizedIndexStats( + name=name, + segments_before=before.get(name, (0, 0))[0], + segments_after=after.get(name, (0, 0))[0], + fragments_covered_before=before.get(name, (0, 0))[1], + fragments_covered_after=after.get(name, (0, 0))[1], + ) + for name in selected + ] + + logger.info( + "Optimized %d indices in %.2fs: version %d -> %d", + len(per_index), + duration, + version_before, + latest.version, + ) + return OptimizeIndicesStats( + version_before=version_before, + version_after=latest.version, + duration_seconds=duration, + indices=per_index, + ) diff --git a/tests/io/lance/test_lance_optimize_indices.py b/tests/io/lance/test_lance_optimize_indices.py new file mode 100644 index 0000000..612f7c3 --- /dev/null +++ b/tests/io/lance/test_lance_optimize_indices.py @@ -0,0 +1,321 @@ +from __future__ import annotations + +import warnings +from pathlib import Path +from typing import Any + +import lance +import pytest + +import daft +import daft_lance +from daft.dependencies import pa +from daft_lance import OptimizeIndicesStats, create_scalar_index, optimize_indices + +warnings.filterwarnings("ignore", category=DeprecationWarning, module="lance") + + +def _make_dataset(path: Path, n_rows: int = 80, rows_per_file: int = 20) -> str: + table = pa.table( + { + "id": list(range(n_rows)), + "name": [f"name-{i % 8}" for i in range(n_rows)], + } + ) + lance.write_dataset(table, str(path), mode="create", max_rows_per_file=rows_per_file) + return str(path) + + +def _fragment_ids(uri: str) -> list[int]: + return sorted(f.fragment_id for f in lance.dataset(uri).get_fragments()) + + +def _covered_fragments(uri: str, index_name: str) -> set[int]: + for desc in lance.dataset(uri).describe_indices(): + if desc.name == index_name: + covered: set[int] = set() + for segment in desc.segments or []: + covered.update(segment.fragment_ids or []) + return covered + raise AssertionError(f"index {index_name} not found") + + +def _segment_count(uri: str, index_name: str) -> int: + for desc in lance.dataset(uri).describe_indices(): + if desc.name == index_name: + return len(desc.segments or []) + raise AssertionError(f"index {index_name} not found") + + +def _sorted_rows(uri: str, predicate: str) -> dict[str, list[Any]]: + table = lance.dataset(uri).scanner(filter=predicate).to_table().sort_by("id") + return table.to_pydict() + + +PREDICATES = ["name = 'name-3'", "name in ('name-0', 'name-7')", "id >= 40", "id = 55", "name = 'nope'"] + + +def test_optimize_covers_new_fragments_and_keeps_query_correct(tmp_path: Path) -> None: + """Appended fragments are indexed in one transaction and results stay identical.""" + uri = _make_dataset(tmp_path / "incremental.lance") + reference = _make_dataset(tmp_path / "reference.lance") + + create_scalar_index(uri, column="name", index_type="INVERTED") + extra = pa.table({"id": list(range(100, 140)), "name": [f"name-{100 + i}" for i in range(40)]}) + lance.write_dataset(extra, uri, mode="append", max_rows_per_file=20) + lance.write_dataset(extra, reference, mode="append", max_rows_per_file=20) + + assert _covered_fragments(uri, "name_idx") < set(_fragment_ids(uri)) + + stats = optimize_indices(uri) + + assert isinstance(stats, OptimizeIndicesStats) + assert stats.changed + assert stats.version_after == stats.version_before + 1 + assert stats.duration_seconds >= 0 + idx = next(i for i in stats.indices if i.name == "name_idx") + assert idx.fragments_covered_after == len(_fragment_ids(uri)) + assert idx.fragments_covered_after > idx.fragments_covered_before + assert _covered_fragments(uri, "name_idx") == set(_fragment_ids(uri)) + + for predicate in PREDICATES: + assert _sorted_rows(uri, predicate) == _sorted_rows(reference, predicate), predicate + + +def test_optimize_merges_small_segments(tmp_path: Path) -> None: + """num_indices_to_merge compacts fragmented segment coverage.""" + uri = _make_dataset(tmp_path / "merge.lance") + + create_scalar_index(uri, column="name", index_type="INVERTED", fragment_group_size=1) + assert _segment_count(uri, "name_idx") == 4 + + stats = optimize_indices(uri, indices=["name_idx"], num_indices_to_merge=4) + + idx = stats.indices[0] + assert idx.segments_after < idx.segments_before + assert _segment_count(uri, "name_idx") == 1 + assert lance.dataset(uri).scanner(filter="name = 'name-3'").to_table().num_rows == 10 + + +def test_optimize_noop_commits_no_version(tmp_path: Path) -> None: + """A healthy index and an index-free dataset both optimize to a no-op.""" + uri = _make_dataset(tmp_path / "noop.lance") + create_scalar_index(uri, column="name", index_type="INVERTED") + version_before = lance.dataset(uri).version + + stats = optimize_indices(uri) + + assert not stats.changed + assert stats.version_after == version_before + + plain = _make_dataset(tmp_path / "plain.lance") + stats = optimize_indices(plain) + assert not stats.changed + assert stats.indices == [] + + +def test_optimize_heals_stale_fragment_ids_after_delete(tmp_path: Path) -> None: + """A fully deleted fragment inside a mixed segment is dropped from coverage. + + Deletes retire fully-dead segments but cannot remove one dead fragment + from a segment that also covers live ones; the optimizer heals it — but + only as part of a commit that indexes or merges new data, so this test + appends first (test_stale_coverage_without_new_data_is_not_healed pins + the other side). + """ + uri = str(tmp_path / "heal.lance") + lance.write_dataset( + pa.table({"id": list(range(40)), "name": [f"row {i}" for i in range(40)]}), + uri, + mode="create", + max_rows_per_file=10, # fragments 0..3 + ) + create_scalar_index(uri, column="name", index_type="INVERTED", name="s_idx", fragment_group_size=4) + + rows_of_fragment_0 = [f"row {i}" for i in range(10)] + in_list = ", ".join(f"'{r}'" for r in rows_of_fragment_0) + lance.dataset(uri).delete(f"name in ({in_list})") + assert 0 not in [f.fragment_id for f in lance.dataset(uri).get_fragments()] + assert 0 in _covered_fragments(uri, "s_idx") # stale id still in coverage + + lance.write_dataset( + pa.table({"id": list(range(100, 110)), "name": [f"row {100 + i}" for i in range(10)]}), + uri, + mode="append", + max_rows_per_file=10, + ) + + optimize_indices(uri, indices=["s_idx"]) + + covered = _covered_fragments(uri, "s_idx") + assert 0 not in covered + assert covered == set(_fragment_ids(uri)) + ds = lance.dataset(uri) + assert ds.count_rows() == 40 + assert ds.scanner(filter="name = 'row 25'").to_table().num_rows == 1 + + +def test_indices_filter_targets_one_index(tmp_path: Path) -> None: + """The name filter optimizes only the named index.""" + uri = _make_dataset(tmp_path / "filter.lance") + create_scalar_index(uri, column="name", index_type="INVERTED", name="name_idx") + create_scalar_index(uri, column="id", index_type="BTREE", name="id_idx") + extra = pa.table({"id": list(range(100, 120)), "name": [f"name-{100 + i}" for i in range(20)]}) + lance.write_dataset(extra, uri, mode="append", max_rows_per_file=20) + + stats = optimize_indices(uri, indices=["id_idx"]) + + assert [i.name for i in stats.indices] == ["id_idx"] + assert _covered_fragments(uri, "id_idx") == set(_fragment_ids(uri)) + assert _covered_fragments(uri, "name_idx") < set(_fragment_ids(uri)) # untouched + assert lance.dataset(uri).scanner(filter="id = 105").to_table().num_rows == 1 + + +def test_unknown_index_names_raise(tmp_path: Path) -> None: + uri = _make_dataset(tmp_path / "unknown.lance") + create_scalar_index(uri, column="name", index_type="INVERTED") + version_before = lance.dataset(uri).version + + with pytest.raises(ValueError, match=r"\['ghost_idx'\].*Available index names: \['name_idx'\]"): + optimize_indices(uri, indices=["ghost_idx"]) + + # Rejected before any work: the index and version are untouched. + assert lance.dataset(uri).version == version_before + assert _covered_fragments(uri, "name_idx") == set(_fragment_ids(uri)) + + +def test_empty_indices_raise(tmp_path: Path) -> None: + uri = _make_dataset(tmp_path / "empty.lance") + with pytest.raises(ValueError, match="non-empty"): + optimize_indices(uri, indices=[]) + + +def test_optimize_via_namespace_entry(tmp_path: Path) -> None: + """The namespace entry resolves the same dataset as the URI entry.""" + ns = {"namespace_impl": "dir", "namespace_properties": {"root": str(tmp_path)}} + table_id = ["tbl"] + + daft_lance.write_lance( + daft.from_pydict({"id": list(range(40)), "name": [f"name-{i % 8}" for i in range(40)]}), + table_id=table_id, + mode="create", + **ns, + ).collect() + uri = str(tmp_path / "tbl.lance") + create_scalar_index(table_id=table_id, column="name", index_type="INVERTED", **ns) + daft_lance.write_lance( + daft.from_pydict({"id": list(range(100, 120)), "name": [f"name-{100 + i}" for i in range(20)]}), + table_id=table_id, + mode="append", + **ns, + ).collect() + + stats = daft_lance.optimize_indices(table_id=table_id, **ns) + + assert stats.changed + assert _covered_fragments(uri, "name_idx") == set(_fragment_ids(uri)) + + +def test_stats_report_versions_duration_and_per_index_counts(tmp_path: Path) -> None: + uri = _make_dataset(tmp_path / "stats.lance") + create_scalar_index(uri, column="name", index_type="INVERTED", name="name_idx") + create_scalar_index(uri, column="id", index_type="BTREE", name="id_idx") + extra = pa.table({"id": list(range(100, 140)), "name": [f"name-{100 + i}" for i in range(40)]}) + lance.write_dataset(extra, uri, mode="append", max_rows_per_file=20) + all_fragments = set(_fragment_ids(uri)) + version_before = lance.dataset(uri).version + + stats = optimize_indices(uri) + + assert stats.version_before == version_before + assert stats.version_after == version_before + 1 + assert stats.duration_seconds > 0 + assert {i.name for i in stats.indices} == {"name_idx", "id_idx"} + for i in stats.indices: + assert i.fragments_covered_before < len(all_fragments) + assert i.fragments_covered_after == len(all_fragments) + + +def test_stale_coverage_without_new_data_is_not_healed(tmp_path: Path) -> None: + """Healing rides along with commits that index or merge new data. + + With nothing new to index, optimize commits nothing and leaves the + stale-only coverage as-is (documented behavior: use a replace=True + rebuild to clean it up deterministically). + """ + uri = str(tmp_path / "stale.lance") + lance.write_dataset( + pa.table({"id": list(range(40)), "name": [f"row {i}" for i in range(40)]}), + uri, + mode="create", + max_rows_per_file=10, + ) + create_scalar_index(uri, column="name", index_type="INVERTED", name="s_idx", fragment_group_size=4) + rows_of_fragment_0 = [f"row {i}" for i in range(10)] + in_list = ", ".join(f"'{r}'" for r in rows_of_fragment_0) + lance.dataset(uri).delete(f"name in ({in_list})") + version_before = lance.dataset(uri).version + assert 0 in _covered_fragments(uri, "s_idx") + + stats = optimize_indices(uri) + + assert not stats.changed + assert lance.dataset(uri).version == version_before + assert 0 in _covered_fragments(uri, "s_idx") + + +def test_optimize_after_delete_all_is_noop_with_live_only_coverage(tmp_path: Path) -> None: + """After deleting every row the index and its stale ids stay as-is. + + Coverage stats count only live fragments, so a fully-dead dataset + reports zero coverage instead of the stale IDs. + """ + uri = str(tmp_path / "gone.lance") + lance.write_dataset(pa.table({"name": [f"row {i}" for i in range(20)]}), uri, mode="create", max_rows_per_file=10) + create_scalar_index(uri, column="name", index_type="INVERTED", name="s_idx") + lance.dataset(uri).delete("name != ''") + assert lance.dataset(uri).count_rows() == 0 + version_before = lance.dataset(uri).version + + stats = optimize_indices(uri) + + assert not stats.changed + assert lance.dataset(uri).version == version_before + idx = stats.indices[0] + assert idx.fragments_covered_before == 0 + assert idx.fragments_covered_after == 0 + assert idx.segments_before == idx.segments_after # index not retired + + +def test_duplicate_indices_are_deduplicated(tmp_path: Path) -> None: + """Duplicate names optimize and report each index exactly once.""" + uri = _make_dataset(tmp_path / "dupes.lance") + create_scalar_index(uri, column="name", index_type="INVERTED") + extra = pa.table({"id": list(range(100, 120)), "name": [f"name-{100 + i}" for i in range(20)]}) + lance.write_dataset(extra, uri, mode="append", max_rows_per_file=20) + + stats = optimize_indices(uri, indices=["name_idx", "name_idx"]) + + assert [i.name for i in stats.indices] == ["name_idx"] + assert _covered_fragments(uri, "name_idx") == set(_fragment_ids(uri)) + + +def test_index_snapshot_falls_back_to_list_indices() -> None: + """Legacy manifests that describe_indices cannot parse still report names. + + Same degradation create_scalar_index uses (_existing_index_names); + counts are unknown in that case. + """ + from daft_lance.lance_scalar_index import _index_snapshot + + class FakeLanceDataset: + def describe_indices(self): + raise RuntimeError("missing index_details") + + def list_indices(self): + return [{"name": "legacy_idx"}] + + def get_fragments(self): + return [] + + assert _index_snapshot(FakeLanceDataset()) == {"legacy_idx": (0, 0)}