From 2300c231691ba8fd264e35d9e841a2b5822addb7 Mon Sep 17 00:00:00 2001 From: wangzheyan Date: Mon, 21 Sep 2026 12:26:38 +0800 Subject: [PATCH 1/3] feat: add optimize_indices for incremental index maintenance Delegates to pylance's DatasetOptimizer.optimize_indices (the same choice lance-ray makes) to index newly appended fragments, merge small segments, and heal stale coverage left by deletes, committing at most one new version and none when there is nothing to do. The indices name filter is our parameter and validated deterministically (empty list and unknown names raise); everything else belongs to Lance. Returns OptimizeIndicesStats with versions, duration, and per-index segment/coverage counts. --- README.md | 23 ++ daft_lance/__init__.py | 5 + daft_lance/_lance.py | 106 +++++++- daft_lance/lance_scalar_index.py | 120 +++++++++ tests/io/lance/test_lance_optimize_indices.py | 233 ++++++++++++++++++ 5 files changed, 486 insertions(+), 1 deletion(-) create mode 100644 tests/io/lance/test_lance_optimize_indices.py diff --git a/README.md b/README.md index 661de0d..6fe2b8d 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,29 @@ index type — mismatches are rejected by Lance's build/commit APIs. Backfill ap that leaves the dataset version unchanged (rebuild with `replace=True` and no `fragment_ids` instead). +#### 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 in one Lance transaction: it indexes +newly appended fragments, merges small segments, and heals stale coverage +left by deletes. It is a no-op that commits no new version when there is +nothing to do. + +```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 versions before and after, the +duration, and per-index segment/coverage counts; unknown or empty `indices` +raise `ValueError`. + > **Breaking change (from 0.5.0):** the `segmented` parameter was removed — > the distributed segment-index workflow is now the only code path, and the diff --git a/daft_lance/__init__.py b/daft_lance/__init__.py index 1a0323d..a776647 100644 --- a/daft_lance/__init__.py +++ b/daft_lance/__init__.py @@ -9,15 +9,20 @@ create_scalar_index, merge_columns, merge_columns_df, + optimize_indices, read_lance, write_lance, ) +from .lance_scalar_index import OptimizedIndexStats, OptimizeIndicesStats __all__ = [ + "OptimizeIndicesStats", + "OptimizedIndexStats", "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..a8c29bb 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, + version: int | str | None = None, + asof: str | 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 in one Lance transaction: newly appended fragments are indexed, + small segments are merged, and stale fragment IDs left inside mixed + segments by deletes are healed. It is a no-op that commits no new + version when every index already covers all fragments. + + 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``. + 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. + version: Version of the dataset to use as the starting snapshot. + asof: Timestamp to use for time travel queries. + 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: version numbers before and after, wall-clock + duration, and per-index segment/coverage counts before and after. + ``changed`` is ``True`` only when a new version was committed. + + 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/") + >>> stats.changed + True + >>> [i.name for i in stats.indices] + ['name_idx'] + + Optimize one index and merge its small segments: + + >>> daft_lance.optimize_indices("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, + version=version, + asof=asof, + 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..68e0ff3 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,121 @@ 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. + + An index that the optimizer retires entirely (all of its fragments were + deleted) reports zeros for the ``*_after`` fields. + """ + + 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.""" + + version_before: int + version_after: int + duration_seconds: float + indices: list[OptimizedIndexStats] + + @property + def changed(self) -> bool: + """Whether the run committed a new dataset version.""" + 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, covered-fragment count)``.""" + snapshot: dict[str, tuple[int, int]] = {} + for desc in lance_ds.describe_indices(): + segments = desc.segments or [] + covered = {fid for segment in segments for fid in (segment.fragment_ids or ())} + snapshot[desc.name] = (len(segments), len(covered)) + 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: it extends coverage over newly appended fragments, merges + small segments (``num_indices_to_merge``), and heals stale fragment IDs + left inside mixed segments by deletes. It commits at most one new + version and is a no-op (no new version) when every index already covers + all fragments. 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, + and unknown names raise listing the available indexes. Merge-count + validation and everything about index internals belong to Lance. + """ + before = _index_snapshot(lance_ds) + 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.") + 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 = lance_ds.version + start = time.monotonic() + lance_ds.optimize.optimize_indices(**call_kwargs) + 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..58dcd71 --- /dev/null +++ b/tests/io/lance/test_lance_optimize_indices.py @@ -0,0 +1,233 @@ +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. + """ + 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) From 654aab50163edb151f06bce12a6a12e630eef776 Mon Sep 17 00:00:00 2001 From: wangzheyan Date: Mon, 21 Sep 2026 14:21:06 +0800 Subject: [PATCH 2/3] fix: silence no-untyped-call for pylance optimize_indices pylance's DatasetOptimizer.optimize_indices is declared as def optimize_indices(self, **kwargs) with no annotations, so mypy --strict rejects the call from typed code. Add the targeted type-ignore with the error code, matching the existing pattern for untyped pylance APIs (e.g. compact_files in lance_compaction.py). --- daft_lance/lance_scalar_index.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/daft_lance/lance_scalar_index.py b/daft_lance/lance_scalar_index.py index 68e0ff3..9a2cf54 100644 --- a/daft_lance/lance_scalar_index.py +++ b/daft_lance/lance_scalar_index.py @@ -553,7 +553,7 @@ def optimize_indices_internal( ) version_before = lance_ds.version start = time.monotonic() - lance_ds.optimize.optimize_indices(**call_kwargs) + lance_ds.optimize.optimize_indices(**call_kwargs) # type: ignore[no-untyped-call] duration = time.monotonic() - start latest = open_context.open_latest() From a5a15539aa6a2509cd1932d00f3da8a0fc7bf736 Mon Sep 17 00:00:00 2001 From: wangzheyan Date: Mon, 21 Sep 2026 14:37:25 +0800 Subject: [PATCH 3/3] fix: address review findings on optimize_indices - Drop version/asof: pylance plans from the pinned manifest while committing on latest, so pinning a maintenance op silently leaves newer fragments unindexed and corrupted the stats; lance-ray's optimize_indices has neither parameter. - Sample both stat snapshots from the dataset's latest version (the before-baseline no longer mixes the handle's snapshot with open_latest), so changed/version_* no longer report concurrent or pinned commits as this run's work. - Count only live fragments as coverage, so stale IDs left by deletes no longer masquerade as coverage (also makes healing visible in the stats). - Ignore duplicate names in indices, matching fragment_ids semantics. - Fall back to list_indices when describe_indices cannot parse a legacy manifest, mirroring _existing_index_names. - Precise docs: healing rides along with commits that index or merge new data; a no-new-data run commits nothing; a delete-all leaves the index and its stale ids as-is. README section moved after the 0.5.0 note, doctest examples marked +SKIP. - Un-export OptimizedIndexStats (still importable from the module) to keep one top-level stats name. 4 new tests pin these behaviors (13 total; suite 426 passed). --- README.md | 31 ++++--- daft_lance/__init__.py | 3 +- daft_lance/_lance.py | 36 ++++---- daft_lance/lance_scalar_index.py | 74 +++++++++++---- tests/io/lance/test_lance_optimize_indices.py | 90 ++++++++++++++++++- 5 files changed, 183 insertions(+), 51 deletions(-) diff --git a/README.md b/README.md index 6fe2b8d..6f632c2 100644 --- a/README.md +++ b/README.md @@ -72,14 +72,23 @@ index type — mismatches are rejected by Lance's build/commit APIs. Backfill ap that leaves the dataset version unchanged (rebuild with `replace=True` and 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 +> 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 in one Lance transaction: it indexes -newly appended fragments, merges small segments, and heals stale coverage -left by deletes. It is a no-op that commits no new version when there is -nothing to do. +`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 @@ -91,18 +100,12 @@ stats = optimize_indices("s3://bucket/my_dataset", indices=["name_idx"], num_ind 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 versions before and after, the -duration, and per-index segment/coverage counts; unknown or empty `indices` -raise `ValueError`. +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. -> **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 -> 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. - ### Column Merging ```python diff --git a/daft_lance/__init__.py b/daft_lance/__init__.py index a776647..033c085 100644 --- a/daft_lance/__init__.py +++ b/daft_lance/__init__.py @@ -13,11 +13,10 @@ read_lance, write_lance, ) -from .lance_scalar_index import OptimizedIndexStats, OptimizeIndicesStats +from .lance_scalar_index import OptimizeIndicesStats __all__ = [ "OptimizeIndicesStats", - "OptimizedIndexStats", "compact_files", "create_scalar_index", "merge_columns", diff --git a/daft_lance/_lance.py b/daft_lance/_lance.py index a8c29bb..216b02b 100644 --- a/daft_lance/_lance.py +++ b/daft_lance/_lance.py @@ -561,8 +561,6 @@ def optimize_indices( indices: list[str] | None = None, num_indices_to_merge: int | None = None, storage_options: dict[str, Any] | None = None, - version: int | str | None = None, - asof: str | None = None, block_size: int | None = None, commit_lock: Any | None = None, index_cache_size: int | None = None, @@ -574,10 +572,11 @@ def optimize_indices( 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 in one Lance transaction: newly appended fragments are indexed, - small segments are merged, and stale fragment IDs left inside mixed - segments by deletes are healed. It is a no-op that commits no new - version when every index already covers all fragments. + 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 @@ -594,13 +593,11 @@ def optimize_indices( {"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``. + 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. - version: Version of the dataset to use as the starting snapshot. - asof: Timestamp to use for time travel queries. block_size: Block size for the dataset. commit_lock: Commit lock for the dataset. index_cache_size: Size of the index cache. @@ -608,9 +605,12 @@ def optimize_indices( metadata_cache_size_bytes: Size of the metadata cache in bytes. Returns: - OptimizeIndicesStats: version numbers before and after, wall-clock - duration, and per-index segment/coverage counts before and after. - ``changed`` is ``True`` only when a new version was committed. + 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 @@ -618,15 +618,17 @@ def optimize_indices( Examples: >>> import daft_lance - >>> stats = daft_lance.optimize_indices("s3://my-bucket/dataset/") - >>> stats.changed + >>> stats = daft_lance.optimize_indices("s3://my-bucket/dataset/") # doctest: +SKIP + >>> stats.changed # doctest: +SKIP True - >>> [i.name for i in stats.indices] + >>> [i.name for i in stats.indices] # doctest: +SKIP ['name_idx'] Optimize one index and merge its small segments: - >>> daft_lance.optimize_indices("s3://my-bucket/dataset/", indices=["name_idx"], num_indices_to_merge=4) + >>> 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 @@ -637,8 +639,6 @@ def optimize_indices( namespace_impl=namespace_impl, namespace_properties=namespace_properties, table_id=table_id, - version=version, - asof=asof, block_size=block_size, commit_lock=commit_lock, index_cache_size=index_cache_size, diff --git a/daft_lance/lance_scalar_index.py b/daft_lance/lance_scalar_index.py index 9a2cf54..e66c646 100644 --- a/daft_lance/lance_scalar_index.py +++ b/daft_lance/lance_scalar_index.py @@ -472,8 +472,10 @@ def _validate_segments_against_manifest( class OptimizedIndexStats: """Per-index outcome of an ``optimize_indices`` run. - An index that the optimizer retires entirely (all of its fragments were - deleted) reports zeros for the ``*_after`` fields. + ``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 @@ -485,7 +487,12 @@ class OptimizedIndexStats: @dataclasses.dataclass(frozen=True) class OptimizeIndicesStats: - """Outcome of an ``optimize_indices`` run over one dataset.""" + """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 @@ -494,17 +501,39 @@ class OptimizeIndicesStats: @property def changed(self) -> bool: - """Whether the run committed a new dataset version.""" + """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, covered-fragment count)``.""" + """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 lance_ds.describe_indices(): + 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)) + snapshot[desc.name] = (len(segments), len(covered & live_fragments)) return snapshot @@ -519,22 +548,35 @@ def optimize_indices_internal( Delegates to pylance's ``DatasetOptimizer.optimize_indices`` — the same choice lance-ray makes — because Lance core owns the delta-index - semantics: it extends coverage over newly appended fragments, merges - small segments (``num_indices_to_merge``), and heals stale fragment IDs - left inside mixed segments by deletes. It commits at most one new - version and is a no-op (no new version) when every index already covers - all fragments. Heavier changes are a distributed rebuild: - ``create_scalar_index(..., replace=True)``. + 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, - and unknown names raise listing the available indexes. Merge-count + 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. """ - before = _index_snapshot(lance_ds) 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)}") @@ -551,7 +593,7 @@ def optimize_indices_internal( indices if indices is not None else "(all)", num_indices_to_merge, ) - version_before = lance_ds.version + version_before = base.version start = time.monotonic() lance_ds.optimize.optimize_indices(**call_kwargs) # type: ignore[no-untyped-call] duration = time.monotonic() - start diff --git a/tests/io/lance/test_lance_optimize_indices.py b/tests/io/lance/test_lance_optimize_indices.py index 58dcd71..612f7c3 100644 --- a/tests/io/lance/test_lance_optimize_indices.py +++ b/tests/io/lance/test_lance_optimize_indices.py @@ -118,7 +118,10 @@ 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. + 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( @@ -231,3 +234,88 @@ def test_stats_report_versions_duration_and_per_index_counts(tmp_path: Path) -> 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)}