fix(index): filter stale vector segment rows before top-k

protos/index.proto

@@ -225,6 +225,12 @@ message VectorIndexDetails {
   // Keys use reverse-DNS namespacing (e.g., "lance.ivf.max_iters", "lancedb.accelerator").
   // Unrecognized keys must be silently ignored by all runtimes.
   map<string, string> runtime_hints = 9;
+
+  // The fragments whose rows are physically present in this segment. This is
+  // immutable even when fragment ownership is later pruned after an in-place
+  // update. It contains a serialized 32-bit Roaring bitmap when known and is
+  // absent for segments written before physical coverage was recorded.
+  optional bytes physical_fragment_bitmap = 10;
 }
 
 // Hierarchical Navigable Small World (HNSW) parameters, used as an optional configuration for IVF indexes.
@@ -248,4 +254,4 @@ message BloomFilterIndexDetails {}
 
 message RTreeIndexDetails {}
 
-message FMIndexDetails {}
+message FMIndexDetails {}

python/python/tests/test_vector_index.py

@@ -1961,6 +1961,76 @@ def test_optimize_indices(indexed_dataset):
     assert stats["num_indices"] == 2
 
 
+@pytest.mark.parametrize("enable_stable_row_ids", [False, True])
+def test_segment_ownership_filter_precedes_partition_topk(
+    tmp_path, enable_stable_row_ids
+):
+    ndim = 4
+
+    def table(ids, value):
+        vectors = np.full((len(ids), ndim), value, dtype=np.float32)
+        return pa.table(
+            {
+                "id": pa.array(ids, type=pa.int64()),
+                "vector": pa.FixedSizeListArray.from_arrays(
+                    pa.array(vectors.reshape(-1), type=pa.float32()), ndim
+                ),
+            }
+        )
+
+    dataset = lance.write_dataset(
+        table(range(20), 1.0),
+        tmp_path,
+        mode="create",
+        enable_stable_row_ids=enable_stable_row_ids,
+    )
+    dataset = lance.write_dataset(
+        table(range(100, 120), 0.0), dataset.uri, mode="append"
+    )
+    dataset = dataset.create_index(
+        "vector", index_type="IVF_FLAT", metric="l2", num_partitions=1
+    )
+
+    fragment = dataset.get_fragment(1)
+    row_ids = fragment.to_table(columns=["id"], with_row_id=True)["_rowid"].to_pylist()
+    update_data = pa.table(
+        {
+            "_rowid": pa.array(row_ids, type=pa.uint64()),
+            "vector": pa.array(
+                [[10.0] * ndim] * len(row_ids), type=pa.list_(pa.float32(), ndim)
+            ),
+        }
+    )
+    updated_fragment, fields_modified = fragment.update_columns(update_data)
+    dataset = lance.LanceDataset.commit(
+        dataset.uri,
+        lance.LanceOperation.Update(
+            updated_fragments=[updated_fragment], fields_modified=fields_modified
+        ),
+        read_version=dataset.version,
+    )
+    dataset.optimize.optimize_indices(num_indices_to_merge=0)
+    dataset = lance.dataset(dataset.uri)
+
+    def assert_current_nearest_rows():
+        result = dataset.to_table(
+            columns=["id"],
+            nearest={
+                "column": "vector",
+                "q": np.zeros(ndim, dtype=np.float32),
+                "k": 5,
+            },
+        )
+
+        assert all(row_id < 20 for row_id in result["id"].to_pylist())
+        assert result["_distance"].to_pylist() == pytest.approx([4.0] * 5)
+
+    assert_current_nearest_rows()
+    dataset.optimize.optimize_indices(num_indices_to_merge=2)
+    dataset = lance.dataset(dataset.uri)
+    assert_current_nearest_rows()
+
+
 @pytest.mark.skip(reason="retrain is deprecated")
 def test_retrain_indices(indexed_dataset):
     data = create_table()