diff --git a/src/vidxp/cli_commands/index.py b/src/vidxp/cli_commands/index.py index d51a1593..14fee847 100644 --- a/src/vidxp/cli_commands/index.py +++ b/src/vidxp/cli_commands/index.py @@ -478,12 +478,14 @@ def index_clear( typer.Option("--json", help="Emit machine-readable JSON."), ] = False, ) -> None: - """Publish an empty active snapshot without deleting retained generations.""" + """Remove generated index data and publish an empty active snapshot.""" state = state_from_context(ctx) if not yes: typer.confirm( - f"Clear the active index at {state.service.index_directory}?", + f"Clear the active index at {state.service.index_directory}? " + "This removes generated index data and vector collections; " + "imported source media is preserved.", abort=True, ) cleared = state.service.clear_index() @@ -493,4 +495,9 @@ def index_clear( if effective_output_format(state, json_output) == OutputFormat.json: emit_json(payload) else: - typer.echo("Index cleared." if cleared else "No index was found.") + typer.echo( + "Index cleared. Generated index data and vector collections " + "were removed; imported source media is preserved." + if cleared + else "No index was found." + ) diff --git a/src/vidxp/infrastructure/local_index.py b/src/vidxp/infrastructure/local_index.py index 982e4199..3fcc4cda 100644 --- a/src/vidxp/infrastructure/local_index.py +++ b/src/vidxp/infrastructure/local_index.py @@ -582,4 +582,38 @@ def clear(self, config: IndexConfig) -> bool: self._require_index_directory(config.index_directory) repository = self.repository with repository.lease(): - return repository.clear() + cleared = repository.clear() + self._discard_vector_collections( + repository, + config, + client_factory=self.chroma_clients, + ) + return cleared + + def _discard_vector_collections( + self, + repository: LocalSnapshotRepository, + config: IndexConfig, + *, + client_factory: ChromaClientFactory | None = None, + ) -> None: + clients = client_factory or ChromaClientFactory() + cleanup_config = replace( + config, + storage_directory=repository.store, + generation_directory=None, + video_id=None, + generation_id=None, + snapshot_id=None, + snapshot_sha256=None, + ) + if clients.remote or repository.store.is_dir(): + try: + with IndexStorage( + cleanup_config, + create=False, + client_factory=clients, + ) as storage: + storage.clear() + except FileNotFoundError: + pass diff --git a/src/vidxp/infrastructure/local_snapshots.py b/src/vidxp/infrastructure/local_snapshots.py index e364f27f..8c2221ad 100644 --- a/src/vidxp/infrastructure/local_snapshots.py +++ b/src/vidxp/infrastructure/local_snapshots.py @@ -116,14 +116,23 @@ def generation_directory(self, generation_id: str) -> Path: def _snapshot_path(self, snapshot_id: str) -> Path: return self.snapshots / f"{snapshot_id}.json" - def read_active(self, *, required: bool = False) -> IndexSnapshot | None: - resolved = self._read_active(required=required) + def read_active( + self, + *, + required: bool = False, + validate_generations: bool = True, + ) -> IndexSnapshot | None: + resolved = self._read_active( + required=required, + validate_generations=validate_generations, + ) return None if resolved is None else resolved[1] def _read_active( self, *, required: bool = False, + validate_generations: bool = True, ) -> tuple[ActiveSnapshotPointer, IndexSnapshot] | None: if not self.active_pointer.is_file(): if required: @@ -138,6 +147,7 @@ def _read_active( snapshot = self.read_snapshot( pointer.snapshot_id, expected_sha256=pointer.snapshot_sha256, + validate_generations=validate_generations, ) return pointer, snapshot except IndexSchemaError: @@ -152,6 +162,7 @@ def read_snapshot( snapshot_id: str, *, expected_sha256: str | None = None, + validate_generations: bool = True, ) -> IndexSnapshot: snapshot_path = self._snapshot_path(snapshot_id) if not snapshot_path.is_file(): @@ -177,7 +188,8 @@ def read_snapshot( raise IndexSchemaError( "The snapshot filename and document identifier differ." ) - self._validate_generations(snapshot) + if validate_generations: + self._validate_generations(snapshot) return snapshot def _validate_generations(self, snapshot: IndexSnapshot) -> None: @@ -338,9 +350,10 @@ def remove(self, media_id: str) -> bool: return True def clear(self) -> bool: - active = self.read_active() - if active is None or not active.generations: + resolved = self._read_active(validate_generations=False) + if resolved is None or not resolved[1].generations: return False + active = resolved[1] self._publish( generations={}, config_fingerprint=active.config_fingerprint, diff --git a/src/vidxp/infrastructure/sql_snapshots.py b/src/vidxp/infrastructure/sql_snapshots.py index 43a19473..4dc7dfd9 100644 --- a/src/vidxp/infrastructure/sql_snapshots.py +++ b/src/vidxp/infrastructure/sql_snapshots.py @@ -104,7 +104,12 @@ def _ensure_index_state(connection: Connection) -> None: except IntegrityError: pass - def read_active(self, *, required: bool = False) -> IndexSnapshot | None: + def read_active( + self, + *, + required: bool = False, + validate_generations: bool = True, + ) -> IndexSnapshot | None: with self.engine.connect() as connection: row = connection.execute( select( @@ -122,6 +127,7 @@ def read_active(self, *, required: bool = False) -> IndexSnapshot | None: connection, row.active_snapshot_id, expected_sha256=row.active_snapshot_sha256, + validate_generations=validate_generations, ) def read_snapshot( @@ -129,12 +135,14 @@ def read_snapshot( snapshot_id: str, *, expected_sha256: str | None = None, + validate_generations: bool = True, ) -> IndexSnapshot: with self.engine.connect() as connection: return self._read_snapshot( connection, snapshot_id, expected_sha256=expected_sha256, + validate_generations=validate_generations, ) def _read_snapshot( @@ -143,6 +151,7 @@ def _read_snapshot( snapshot_id: str, *, expected_sha256: str | None, + validate_generations: bool = True, ) -> IndexSnapshot: row = connection.execute( select( @@ -166,7 +175,8 @@ def _read_snapshot( raise IndexSchemaError( f"Index snapshot {snapshot_id} failed integrity validation." ) - self._validate_generations(snapshot) + if validate_generations: + self._validate_generations(snapshot) return snapshot def validate_generation( @@ -247,7 +257,7 @@ def remove(self, media_id: str) -> bool: return True def clear(self) -> bool: - active = self.read_active() + active = self.read_active(validate_generations=False) if active is None or not active.generations: return False self._publish( @@ -255,6 +265,7 @@ def clear(self) -> bool: remove_media_id="*", config_fingerprint=active.config_fingerprint, configuration=dict(active.configuration), + validate_generations=False, ) return True @@ -265,6 +276,7 @@ def _publish( remove_media_id: str | None, config_fingerprint: str, configuration: dict[str, Any], + validate_generations: bool = True, ) -> IndexSnapshot: with self.engine.begin() as connection: self._ensure_index_state(connection) @@ -283,6 +295,7 @@ def _publish( connection, state.active_snapshot_id, expected_sha256=state.active_snapshot_sha256, + validate_generations=validate_generations, ) ) generations = dict(active.generations) if active is not None else {} diff --git a/tests/test_local_snapshots.py b/tests/test_local_snapshots.py index 091060c8..ce198e47 100644 --- a/tests/test_local_snapshots.py +++ b/tests/test_local_snapshots.py @@ -16,8 +16,9 @@ IndexSchemaError, StorageRecord, ) +from vidxp.core.manifest import MANIFEST_FILE, sha256_file, write_json_atomic +from vidxp.core.snapshots import GenerationReference from vidxp.core.storage import IndexStorage -from vidxp.core.manifest import MANIFEST_FILE, write_json_atomic from vidxp.infrastructure.local_snapshots import LocalSnapshotRepository from vidxp.infrastructure.local_index import LocalIndexBackend @@ -66,11 +67,95 @@ def write_generation( input_sha: str, record_counts: dict[str, int] | None = None, store_size_bytes_at_commit: int | None = 123, + ): + manifest = self._manifest_payload( + config, + media_id=media_id, + input_sha=input_sha, + record_counts=record_counts, + store_size_bytes_at_commit=store_size_bytes_at_commit, + schema_version=INDEX_SCHEMA_VERSION, + ) + write_json_atomic( + config.run_directory / MANIFEST_FILE, + manifest, + ) + return self.repository.generation_reference( + generation_id=str(config.generation_id), + media_id=media_id, + ) + + def legacy_generation( + self, + media_id: str, + *, + input_sha: str, + schema_version: int, + ): + generation_id = self.repository.new_generation_id() + config = replace( + self.config, + video_id=media_id, + generation_id=generation_id, + generation_directory=self.repository.generation_directory( + generation_id + ), + ) + return config, self.write_legacy_generation( + config, + media_id=media_id, + input_sha=input_sha, + schema_version=schema_version, + ) + + def write_legacy_generation( + self, + config: IndexConfig, + *, + media_id: str, + input_sha: str, + schema_version: int, + record_counts: dict[str, int] | None = None, + store_size_bytes_at_commit: int | None = 123, + ): + manifest = self._manifest_payload( + config, + media_id=media_id, + input_sha=input_sha, + record_counts=record_counts, + store_size_bytes_at_commit=store_size_bytes_at_commit, + schema_version=schema_version, + ) + manifest_path = config.run_directory / MANIFEST_FILE + write_json_atomic(manifest_path, manifest) + return GenerationReference( + generation_id=str(config.generation_id), + media_id=media_id, + manifest_sha256=sha256_file(manifest_path), + input_sha256=input_sha, + config_fingerprint=config.fingerprint(), + modalities=tuple(config.enabled_modalities), + record_counts={ + modality: manifest["record_counts"][modality] + for modality in config.enabled_modalities + }, + store_size_bytes_at_commit=store_size_bytes_at_commit, + ) + + def _manifest_payload( + self, + config: IndexConfig, + *, + media_id: str, + input_sha: str, + record_counts: dict[str, int] | None, + store_size_bytes_at_commit: int | None, + schema_version: int, ): now = datetime.now(timezone.utc).isoformat() - manifest = { + return { "manifest_schema_version": MANIFEST_SCHEMA_VERSION, - "index_schema_version": INDEX_SCHEMA_VERSION, + "index_schema_version": schema_version, "dataset": config.dataset, "split": config.split, "run_id": config.run_id, @@ -114,14 +199,6 @@ def write_generation( }, "store_size_bytes_at_commit": store_size_bytes_at_commit, } - write_json_atomic( - config.run_directory / MANIFEST_FILE, - manifest, - ) - return self.repository.generation_reference( - generation_id=str(config.generation_id), - media_id=media_id, - ) def test_unknown_store_size_round_trips_through_snapshot_metadata(self): config, reference = self.generation( @@ -185,6 +262,28 @@ def test_add_reindex_remove_and_clear_publish_immutable_snapshots(self): ) self.assertFalse(self.repository.clear()) + def test_clear_recovers_from_incompatible_generation_schema(self): + config, reference = self.legacy_generation( + "a", + input_sha="a" * 64, + schema_version=INDEX_SCHEMA_VERSION - 1, + ) + self.repository._publish( + generations={"a": reference}, + config_fingerprint=config.fingerprint(), + configuration=self.repository.snapshot_configuration(config), + ) + + with self.assertRaisesRegex(IndexSchemaError, "invalid"): + self.repository.read_active(required=True) + + self.assertTrue(self.repository.clear()) + self.assertEqual( + self.repository.status()["state"], + "empty", + ) + self.assertFalse(self.repository.clear()) + def test_pointer_failure_preserves_previous_active_snapshot(self): config_a1, a1 = self.generation("a", input_sha="a" * 64) previous = self.repository.publish_generation(a1, config_a1) @@ -793,6 +892,65 @@ def test_real_chroma_reader_remains_pinned_across_reindex(self): {second_reference.generation_id}, ) + def test_clear_discards_incompatible_vector_collections(self): + generation_id = self.repository.new_generation_id() + config = replace( + self.config, + video_id="a", + generation_id=generation_id, + generation_directory=self.repository.generation_directory( + generation_id + ), + ) + with IndexStorage(config) as storage: + storage.upsert( + "scene", + [ + StorageRecord( + source_id=f"source-{generation_id}", + embedding=[1.0, 0.0], + metadata={ + **config.record_identity( + "scene", + f"source-{generation_id}", + ), + }, + ) + ], + batch_size=1, + cancellation=CancellationToken(), + ) + reference = self.write_legacy_generation( + config, + media_id="a", + input_sha="a" * 64, + schema_version=INDEX_SCHEMA_VERSION - 1, + record_counts={"scene": 1}, + ) + self.repository._publish( + generations={"a": reference}, + config_fingerprint=config.fingerprint(), + configuration=self.repository.snapshot_configuration(config), + ) + runtime = Mock() + runtime.backends.torch_device = "cpu" + backend = LocalIndexBackend( + Mock(), + runtime, + self.repository.layout, + ) + clear_config = replace( + self.config, + storage_directory=self.repository.indexes, + ) + + self.assertTrue(backend.clear(clear_config)) + + self.assertEqual(self.repository.status()["state"], "empty") + with self.assertRaises(FileNotFoundError): + with IndexStorage(self.config, create=False): + pass + if __name__ == "__main__": unittest.main()