# Story 4.5: Incremental Indexing

Status: done

## Story

As a **system**,
I want **to add new documents without re-indexing everything**,
So that **the knowledge base grows efficiently**.

## Acceptance Criteria

1. **AC4.5.1:** Add new document records incrementally
   - Insert new records into PostgreSQL without affecting existing records
   - Add new vectors to Milvus collection without rebuilding
   - Maintain referential integrity (document_id links)
   - No full re-indexing required

2. **AC4.5.2:** Delete document and all its records
   - Remove records from PostgreSQL using CASCADE on document_id
   - Remove vectors from Milvus using document_id filter expression
   - Both operations must complete atomically (all or nothing)
   - Return count of deleted records/vectors

3. **AC4.5.3:** Re-process document (delete + re-index)
   - Delete old records/vectors first
   - Insert new records/vectors
   - Operation must be atomic (no partial state visible)
   - Handle case where document_id doesn't exist

4. **AC4.5.4:** Integration with document lifecycle
   - Call incremental indexing from document upload workflow
   - Call delete from document deletion endpoint (Story 2.6)
   - Support both single and batch document operations
   - Maintain sync between PostgreSQL and Milvus

5. **AC4.5.5:** Error handling and rollback
   - If PostgreSQL delete fails, don't delete from Milvus
   - If Milvus delete fails, rollback PostgreSQL transaction
   - Log all operations with document_id for debugging
   - Raise appropriate exceptions (IndexingError, DocumentNotFoundError)

## Tasks / Subtasks

- [ ] **Task 1: Extend VectorStore with delete operations** (AC: 4.5.2)
  - [ ] Add `delete_vectors_by_document_id(document_id)` method to VectorStore
  - [ ] Implement Milvus delete expression: `document_id == "uuid"`
  - [ ] Return count of deleted vectors
  - [ ] Handle case where no vectors exist for document
  - [ ] Add structured logging for delete operations

- [ ] **Task 2: Extend StructuredStore with delete operations** (AC: 4.5.2)
  - [ ] Add `delete_records_by_document_id(document_id)` method to StructuredStore
  - [ ] Use DELETE CASCADE on document_id foreign key
  - [ ] Return count of deleted records
  - [ ] Handle case where no records exist for document
  - [ ] Add structured logging for delete operations

- [ ] **Task 3: Create DocumentDeletionService** (AC: 4.5.2, 4.5.5)
  - [ ] Create `src/services/document_deletion_service.py`
  - [ ] Implement `delete_document(document_id)` method
  - [ ] Delete from both PostgreSQL and Milvus atomically
  - [ ] Use database transaction with rollback on Milvus failure
  - [ ] Log success/failure with counts
  - [ ] Raise appropriate exceptions

- [ ] **Task 4: Implement re-indexing operation** (AC: 4.5.3, 4.5.5)
  - [ ] Add `reindex_document(document_id, records)` to DocumentDeletionService
  - [ ] Delete old records/vectors first
  - [ ] Insert new records/vectors
  - [ ] Use database transaction for atomicity
  - [ ] Handle case where document doesn't exist (first-time index)
  - [ ] Log operation with before/after counts

- [ ] **Task 5: Integrate with document upload workflow** (AC: 4.5.4)
  - [ ] Update document upload endpoint to call indexing
  - [ ] Sequence: validate → extract → index (PostgreSQL + Milvus)
  - [ ] Update document status to "completed" after successful indexing
  - [ ] Handle indexing failures gracefully (mark document as "failed")

- [ ] **Task 6: Integrate with document deletion endpoint** (AC: 4.5.4)
  - [ ] Update DELETE /documents/{document_id} to call DocumentDeletionService
  - [ ] Return 404 if document doesn't exist
  - [ ] Return 200 with deletion counts if successful
  - [ ] Handle partial deletion failures

- [ ] **Task 7: Unit tests for delete operations** (AC: All)
  - [ ] Test VectorStore.delete_vectors_by_document_id()
  - [ ] Test StructuredStore.delete_records_by_document_id()
  - [ ] Test DocumentDeletionService.delete_document()
  - [ ] Test error handling and rollback
  - [ ] Mock Milvus and PostgreSQL

- [ ] **Task 8: Integration tests for incremental operations** (AC: All)
  - [ ] Test adding new document to existing knowledge base
  - [ ] Test deleting document and verifying removal from both stores
  - [ ] Test re-indexing document (delete + add)
  - [ ] Test atomic rollback on failure
  - [ ] Use real PostgreSQL and Milvus instances

- [ ] **Task 9: Update examples and documentation** (AC: All)
  - [ ] Add incremental indexing examples to `examples/vector_store_usage.py`
  - [ ] Demonstrate add, delete, and reindex operations
  - [ ] Show integration with document lifecycle
  - [ ] Document error handling patterns

## Dev Notes

### Architecture Patterns and Constraints

**From architecture.md:**
- **Service layer:** Create `src/services/document_deletion_service.py` for deletion orchestration
- **Database transactions:** Use SQLAlchemy async transactions for atomicity
- **Error handling:** Use `IndexingError`, `DocumentNotFoundError` from `src.core.exceptions`
- **Logging:** Structured logging with structlog

**From PRD (prd.md):**
- **FR5:** Users can delete documents and their associated extracted data
- **FR18:** System stores extracted content with full structural metadata
- **NFR5:** Data consistency between PostgreSQL and Milvus must be maintained

### Learnings from Previous Story

**From Story 4.4: Vector Embedding and Indexing (Status: done)**

**New Services Created:**
- `EmbeddingService` at `src/knowledge/embeddings.py` - Use for embedding new records
- `VectorStore` at `src/knowledge/vector_store.py` - Extend with delete methods
- `record_to_text()` function - Reuse for text representation

**Key Patterns:**
- **Batch processing:** 100 records per operation for performance
- **Async patterns:** Use async/await throughout
- **Retry logic:** Exponential backoff (1s, 2s, 4s) for transient errors
- **Milvus operations:** Use `insert()` not `upsert()` for auto_id collections, call `flush()` after insert
- **Structured logging:** Log all operations with contextual fields

**Technical Details:**
- **Milvus collection:** `dsol_records` with HNSW index
- **Vector dimensions:** 3072 (text-embedding-3-large)
- **Distance metric:** COSINE similarity
- **Text format:** "Key: Value. Key: Value. From: file, Sheet: sheet, Row: row"

**Files to Extend:**
- `src/knowledge/vector_store.py` - Add delete_vectors_by_document_id() method
- `src/knowledge/structured_store.py` - Add delete_records_by_document_id() method

**Integration Points:**
- Document upload workflow (POST /documents)
- Document deletion endpoint (DELETE /documents/{id}) from Story 2.6
- Use `StructuredStore` and `VectorStore` together for dual-store operations

[Source: docs/sprint-artifacts/4-4-vector-embedding-indexing.md]

### Milvus Delete Operation

**Delete by Expression:**
```python
from pymilvus import Collection, connections
from src.core.config import settings

# Connect to Milvus
connections.connect(
    alias="default",
    host=settings.milvus_host,
    port=settings.milvus_port
)

# Get collection
collection = Collection(settings.milvus_collection)

# Delete by document_id filter expression
expr = f'document_id == "{str(document_id)}"'
result = collection.delete(expr)

# Get delete count
delete_count = result.delete_count
```

### PostgreSQL CASCADE Delete

**Foreign Key Constraint:**
```sql
ALTER TABLE extracted_records
ADD CONSTRAINT fk_document
FOREIGN KEY (document_id)
REFERENCES documents(id)
ON DELETE CASCADE;
```

**Delete Operation:**
```python
from src.db.models import ExtractedRecord

# Delete records (CASCADE will handle it)
result = await session.execute(
    delete(ExtractedRecord).where(
        ExtractedRecord.document_id == document_id
    )
)

deleted_count = result.rowcount
await session.commit()
```

### Atomic Deletion Service

**Service Pattern:**
```python
from src.knowledge.structured_store import StructuredStore
from src.knowledge.vector_store import delete_vectors_by_document_id
from src.core.exceptions import IndexingError
from uuid import UUID

class DocumentDeletionService:
    """Orchestrates deletion from both PostgreSQL and Milvus."""

    def __init__(self, db_session):
        self.db_session = db_session
        self.structured_store = StructuredStore(db_session)
        self.logger = structlog.get_logger()

    async def delete_document(self, document_id: UUID) -> dict:
        """Delete document and all associated records atomically.

        Args:
            document_id: Document UUID to delete

        Returns:
            Dict with deletion counts: {"records": int, "vectors": int}

        Raises:
            DocumentNotFoundError: If document doesn't exist
            IndexingError: If deletion fails
        """
        try:
            # Start transaction
            async with self.db_session.begin():
                # Delete from PostgreSQL (CASCADE to records)
                records_deleted = await self.structured_store.delete_records_by_document_id(
                    document_id
                )

                # Delete from Milvus
                vectors_deleted = await delete_vectors_by_document_id(document_id)

                self.logger.info(
                    "document_deleted",
                    document_id=str(document_id),
                    records_deleted=records_deleted,
                    vectors_deleted=vectors_deleted
                )

                return {
                    "records": records_deleted,
                    "vectors": vectors_deleted
                }

        except Exception as e:
            # Transaction will auto-rollback
            self.logger.error(
                "document_deletion_failed",
                document_id=str(document_id),
                error=str(e)
            )
            raise IndexingError(f"Failed to delete document: {e}")
```

### Re-indexing Pattern

```python
async def reindex_document(
    self,
    document_id: UUID,
    records: list[ExtractedRecord]
) -> dict:
    """Re-index document by deleting old and inserting new.

    Args:
        document_id: Document UUID
        records: New extracted records to index

    Returns:
        Dict with operation counts
    """
    # Delete old records/vectors
    deletion_result = await self.delete_document(document_id)

    # Index new records
    record_ids = await self.structured_store.store_records(records, document_id)
    vector_store = VectorStore()
    vector_count = await vector_store.index_records(records, document_id, record_ids)

    return {
        "deleted_records": deletion_result["records"],
        "deleted_vectors": deletion_result["vectors"],
        "indexed_records": len(record_ids),
        "indexed_vectors": vector_count
    }
```

### Integration with Document Lifecycle

**Upload Workflow:**
```python
# POST /documents endpoint (Story 2.2)
async def upload_document(file: UploadFile):
    # 1. Validate file (Story 2.3)
    validator.validate(file)

    # 2. Extract records (Epic 3)
    records = await extraction_pipeline.process(file)

    # 3. Index in both stores (Stories 4.1, 4.4, 4.5)
    async with get_db_session() as session:
        structured_store = StructuredStore(session)
        record_ids = await structured_store.store_records(records, document_id)

        vector_store = VectorStore()
        await vector_store.index_records(records, document_id, record_ids)

    # 4. Mark document as completed
    await update_document_status(document_id, "completed")
```

**Delete Workflow:**
```python
# DELETE /documents/{document_id} endpoint (Story 2.6)
async def delete_document(document_id: UUID):
    async with get_db_session() as session:
        deletion_service = DocumentDeletionService(session)
        result = await deletion_service.delete_document(document_id)

    return {
        "message": "Document deleted",
        "records_deleted": result["records"],
        "vectors_deleted": result["vectors"]
    }
```

### Error Scenarios

**Scenario 1: Milvus delete fails**
- PostgreSQL transaction will rollback
- No changes committed
- User sees error message

**Scenario 2: Document doesn't exist**
- PostgreSQL delete returns 0 rows
- Milvus delete returns 0 vectors
- Still consider success (idempotent)

**Scenario 3: Partial extraction failure on reindex**
- Original records/vectors already deleted
- New indexing fails partway
- Document left in incomplete state
- **Mitigation:** Mark document status as "failed" for manual review

### Prerequisites

- **Story 4.1 COMPLETE:** StructuredStore with store_records()
- **Story 4.4 COMPLETE:** VectorStore with index_records()
- **Story 2.6 COMPLETE:** DELETE /documents/{id} endpoint exists
- **Database:** CASCADE constraint on extracted_records.document_id

### Success Criteria

- ✅ New documents can be added without re-indexing existing ones
- ✅ Documents can be deleted from both PostgreSQL and Milvus
- ✅ Deletion is atomic (both stores or neither)
- ✅ Re-indexing works (delete old + insert new)
- ✅ Integration with document upload and deletion endpoints
- ✅ Error handling with appropriate rollback
- ✅ Unit tests pass (15+ tests)
- ✅ Integration tests pass with real stores
- ✅ Examples demonstrate all operations

### Out of Scope

- **Collection partitioning:** Advanced Milvus optimization for large-scale data
- **Soft deletes:** Mark as deleted without removing data (not in requirements)
- **Bulk re-indexing:** Re-index all documents (future maintenance operation)
- **Version control:** Track document versions over time
- **Partial updates:** Update individual records (not in requirements)

### Next Steps

After Story 4.5:
- **Epic 4 COMPLETE:** Knowledge base fully functional
- **Epic 5:** Q&A & Answer Generation
  - Story 5.1: Query endpoint
  - Story 5.2: Query router (exact vs semantic)
  - Story 5.3: Answer generation with sources

## Technical Design

### Data Flow Diagrams

**Add New Document:**
```
Upload → Validate → Extract → [PostgreSQL + Milvus] → Complete
                                  ↓            ↓
                            store_records  index_records
```

**Delete Document:**
```
DELETE API → DocumentDeletionService → [PostgreSQL + Milvus] → Response
                                          ↓            ↓
                                    DELETE CASCADE  delete(expr)
```

**Re-index Document:**
```
Re-index → Delete Old → Index New → Complete
             ↓              ↓
        [PG + Milvus]  [PG + Milvus]
```

### Database Schema Reference

**Foreign Key Constraint (extracted_records table):**
```sql
-- Migration to add CASCADE constraint
ALTER TABLE extracted_records
ADD CONSTRAINT fk_extracted_records_document
FOREIGN KEY (document_id)
REFERENCES documents(id)
ON DELETE CASCADE;

-- Test CASCADE behavior
-- When document is deleted, all its extracted_records are automatically deleted
```

### Module Structure

```
src/
├── services/
│   └── document_deletion_service.py  # NEW - Orchestrates deletion
├── knowledge/
│   ├── structured_store.py          # EXTEND - Add delete method
│   └── vector_store.py               # EXTEND - Add delete method
└── api/
    └── routes/
        └── documents.py              # MODIFY - Integrate deletion service
```

### API Response Examples

**Successful Deletion:**
```json
{
  "message": "Document deleted successfully",
  "document_id": "550e8400-e29b-41d4-a716-446655440000",
  "records_deleted": 1250,
  "vectors_deleted": 1250
}
```

**Successful Reindex:**
```json
{
  "message": "Document re-indexed successfully",
  "document_id": "550e8400-e29b-41d4-a716-446655440000",
  "deleted": {
    "records": 1250,
    "vectors": 1250
  },
  "indexed": {
    "records": 1300,
    "vectors": 1300
  }
}
```

### Example Usage

```python
from src.services.document_deletion_service import DocumentDeletionService
from src.db.session import get_db_session
from uuid import UUID

# Example 1: Delete document
async def delete_example():
    document_id = UUID("550e8400-e29b-41d4-a716-446655440000")

    async with get_db_session() as session:
        service = DocumentDeletionService(session)
        result = await service.delete_document(document_id)

    print(f"Deleted {result['records']} records and {result['vectors']} vectors")

# Example 2: Re-index document
async def reindex_example():
    document_id = UUID("550e8400-e29b-41d4-a716-446655440000")

    # Get new extracted records from pipeline
    new_records = await extraction_pipeline.process(document_path)

    async with get_db_session() as session:
        service = DocumentDeletionService(session)
        result = await service.reindex_document(document_id, new_records)

    print(f"Reindexed: {result}")
```

---

## Definition of Done

- [ ] All acceptance criteria (AC4.5.1 - AC4.5.5) implemented and tested
- [ ] `DocumentDeletionService` created in `src/services/document_deletion_service.py`
- [ ] `delete_vectors_by_document_id()` added to VectorStore
- [ ] `delete_records_by_document_id()` added to StructuredStore
- [ ] CASCADE constraint verified on extracted_records table
- [ ] Re-indexing operation working (delete + add)
- [ ] Integration with document upload workflow
- [ ] Integration with document deletion endpoint (Story 2.6)
- [ ] Unit tests pass (15+ tests)
- [ ] Integration tests pass with real PostgreSQL and Milvus
- [ ] Atomic deletion working (rollback on failure)
- [ ] Error handling with appropriate exceptions
- [ ] Examples updated in `examples/vector_store_usage.py`
- [ ] Code follows project style (Black, Ruff, mypy strict)
- [ ] Structured logging implemented
- [ ] Documentation strings (docstrings) for all public methods
- [ ] Sprint status updated to "done"
- [ ] Git commit with proper commit message

## Dev Agent Record

### Context Reference

<!-- Path(s) to story context XML will be added here by context workflow -->

### Agent Model Used

{{agent_model_name_version}}

### Debug Log References

### Completion Notes List

### File List