# Hybrid Excel Pipeline Integration Analysis

## Executive Summary

This document provides a detailed analysis for merging the **old Excel pipeline** (openpyxl-based) with the **new Airflow pipeline** (Docling-based) to create a hybrid solution where:

1. **New Pipeline** handles ONLY large tables (>1000 rows)
2. **Old Pipeline** processes the ENTIRE file unchanged (including images, shapes, comments)
3. **Both pipelines** use the **NEW Milvus schema** defined in `src/knowledge/vector_store.py`
4. **Keyword/Entity extraction is REMOVED** for simplicity
5. **Both pipeline outputs are merged** without deduplication (some text overlap may occur)

---

## Integration Strategy

### Routing Logic (Simplified)

```
Excel File Input
    ↓
[Large Table Detection]
    ↓
    ├─ Large Tables (>1000 rows)
    │  └─→ NEW Pipeline (Airflow/Docling)
    │     Returns: large_table_records[]
    │
    └─ ENTIRE File (No Exclusions)
       └─→ OLD Pipeline (openpyxl-based, UNCHANGED)
          Returns: all_chunks[] (includes large tables)
    ↓
[Merge]
    └─ Combine both outputs directly
       Note: Some text overlap may occur for large table content
```

### High-Level Architecture (Simplified)

```
┌─────────────────────────────────────────────────────────┐
│              Excel Document Ingestion Service           │
│                   (Entry Point)                         │
└────────────────┬────────────────────────────────────────┘
                 │
                 ├─ Step 1: Large Table Detection
                 │  └─ LargeTableExtractor.detect_tables()
                 │     Returns: large_table_metadata[]
                 │
                 ├─ Step 2A: Large Tables → NEW Pipeline
                 │  ├─ Trigger Airflow DAG (API call)
                 │  ├─ Wait for completion (polling/webhook)
                 │  └─ Returns: large_table_records[] + embeddings[]
                 │
                 ├─ Step 2B: ENTIRE File → OLD Pipeline (UNCHANGED)
                 │  ├─ ExcelParserService (NO MODIFICATIONS)
                 │  ├─ Processes: tables, images, shapes, comments
                 │  ├─ NO keyword/entity extraction
                 │  └─ Returns: all_chunks[] (includes large tables)
                 │
                 ├─ Step 3: Schema Adaptation
                 │  ├─ Convert OLD chunks → NEW Milvus schema
                 │  ├─ Generate embeddings for OLD chunks
                 │  └─ Merge: large_table_records + adapted_chunks
                 │
                 └─ Step 4: Unified Storage
                    └─ VectorStore.index_with_vectors() (NEW schema)
```

---

## Detailed Phase Analysis

### Phase 1: Large Table Detection

**Input**: Excel file (bytes)

**Process**:
```python
from src.extraction_v2.large_table_extractor import LargeTableExtractor

extractor = LargeTableExtractor()
detected_tables = extractor.detect_tables(file_path)

# Classify tables
large_tables = [t for t in detected_tables if t.get('is_large', False)]
normal_tables = [t for t in detected_tables if not t.get('is_large', False)]
```

**Output Schema**:
```python
{
    'large_tables': [
        {
            'sheet': str,  # Sheet name
            'table_index': int,
            'row_count': int,
            'col_count': int,
            'is_large': True,
            'start_row': int,
            'end_row': int,
            'start_col': int,
            'end_col': int
        }
    ],
    'normal_tables': [
        # Same structure but is_large: False
    ]
}
```

---

### Phase 2A: NEW Pipeline (Large Tables Only)

#### Input Schema
```python
{
    'document_id': str,  # UUID
    'object_key': str,   # MinIO path
    'content_type': str, # MIME type
    'large_tables_only': True,  # Flag to skip normal processing
    'large_table_metadata': [
        {
            'sheet': str,
            'table_index': int,
            'start_row': int,
            'end_row': int,
            # ... from detection phase
        }
    ]
}
```

#### Processing Flow
```
Airflow DAG: document_extraction_dag
    ↓
1. parse_excel_document
   └─ Output: large_table_records[]
    ↓
2. Skip detect_headers (already chunked)
    ↓
3. Skip chunk_excel (records pre-built)
    ↓
4. generate_embeddings_task
   └─ Output: embeddings[]
    ↓
5. Return to caller (NOT stored yet)
```

#### Output Schema (NEW Pipeline)
```python
{
    'records': [
        {
            'content': {
                'column1': 'value1',
                'column2': 'value2',
                # ... dict of column: value
            },
            'headers': ['column1', 'column2', ...],
            '_source': {
                'document_id': str,
                'filename': str,
                'sheet': str,
                'table_id': int,
                'row': int,
                'col_range': str
            },
            '_merge_metadata': Optional[dict],
            'resolved_content': Optional[dict]
        }
    ],
    'embeddings': [
        [float, ...],  # 3072-dimensional vectors
        # ... one per record
    ],
    'record_count': int
}
```

---

### Phase 2B: OLD Pipeline (ENTIRE File, UNCHANGED)

#### Input
- Excel file (bytes) - **ENTIRE file, no exclusions**
- **No modifications to OLD pipeline code required**

#### Processing Flow
```python
# OLD pipeline processes ENTIRE file (including large tables)
# NO CHANGES to existing code
chunks = await excel_parser_service.parse_excel_file(
    file_content=file_content,
    file_id=file_id,
    original_filename=original_filename,
    replace_images_with_descriptions=False,
    max_characters=2000,
)

# Process chunks WITHOUT keyword/entity extraction (removed for simplicity)
milvus_records = await excel_ingestion_service._process_chunks_for_milvus(
    chunks=chunks,
    extract_keywords_and_entities=False,  # DISABLED
    user_metadata=user_metadata
)
```

#### Output Schema (OLD Pipeline)
```python
[
    {
        'file_id': str,
        'element_type': str,  # "Text", "Image", "Shape", "Comment"
        'file_type': str,     # "xlsx"
        'text': str,          # Content or description
        'keywords_text': '',  # EMPTY (no keyword extraction)
        'metadata': {
            'internal_key': {
                # Element-specific metadata
                'sheet_name': str,
                'sheet_number': int,
                'image_url': str,       # For Image type
                'image_s3_key': str,    # For Image type
                'position': dict,       # For Shape/Image
                'connected_target': dict,  # For Shape
                'comment_id': str,      # For Comment
                'author': str,          # For Comment
                # ... more fields
            },
            # User metadata at root level
        },
        'created_at': str
    },
    # ... one per chunk (includes Text chunks from large tables)
]
```

**Note**: Output includes Text chunks from large tables which will also be present in NEW pipeline output (overlap accepted).

---

### Phase 3: Schema Adaptation

This is the CRITICAL phase where we:
1. **Adapt**: Convert OLD chunks to NEW Milvus schema
2. **Embed**: Generate embeddings for adapted chunks
3. **Merge**: Combine with NEW pipeline output

#### Step 3.1: Schema Adaptation

##### NEW Milvus Schema (Target)

From `src/knowledge/vector_store.py:27-50`:

```python
@dataclass
class MilvusEntity:
    """Milvus entity structure for vector storage.

    Unified schema supporting both Excel and Word documents.
    """
    vector: list[float]          # 3072-dimensional embedding
    record_id: str               # UUID linking to PostgreSQL
    document_id: str             # UUID for filtering by document
    filename: str                # For display in search results
    document_type: str           # "excel" or "word"
    text_content: str            # Original text (max 8000 chars for Excel)
    metadata: dict               # JSON dict with type-specific fields
        # For Excel: {
        #     "sheet_name": str,
        #     "row_number": int,
        #     "has_symbols": bool
        # }
```

#### Adapter Design

**File**: `src/adapters/legacy_to_milvus_adapter.py`

```python
"""Adapter to convert legacy pipeline chunks to new Milvus schema."""

from typing import Any, Dict, List
from uuid import UUID, uuid4
from dataclasses import dataclass

from src.knowledge.embeddings import EmbeddingService


@dataclass
class LegacyChunk:
    """Legacy pipeline chunk structure."""
    file_id: str
    element_type: str
    file_type: str
    text: str
    keywords_text: str
    metadata: Dict[str, Any]
    created_at: str


class LegacyToMilvusAdapter:
    """Converts legacy Excel pipeline chunks to new Milvus schema."""

    def __init__(self):
        self.embedding_service = EmbeddingService()

    async def adapt_chunks(
        self,
        legacy_chunks: List[Dict[str, Any]],
        document_id: UUID,
        filename: str,
    ) -> Dict[str, Any]:
        """Convert legacy chunks to new Milvus entity format.

        Args:
            legacy_chunks: List of chunks from old pipeline
            document_id: Document UUID
            filename: Original filename

        Returns:
            Dict with 'records' and 'embeddings' ready for Milvus
        """
        records = []
        texts_to_embed = []

        for chunk in legacy_chunks:
            # Convert to new format
            record = self._convert_chunk_to_record(chunk, document_id, filename)
            records.append(record)

            # Prepare text for embedding
            text = self._prepare_embedding_text(chunk)
            texts_to_embed.append(text)

        # Generate embeddings
        embeddings = await self.embedding_service.embed_batch(texts_to_embed)

        return {
            'records': records,
            'embeddings': embeddings,
            'record_count': len(records)
        }

    def _convert_chunk_to_record(
        self,
        chunk: Dict[str, Any],
        document_id: UUID,
        filename: str,
    ) -> Dict[str, Any]:
        """Convert a single legacy chunk to new record format.

        Maps old schema fields to new ExtractedRecord-like structure.
        """
        element_type = chunk.get('element_type', 'Text')
        metadata_key = chunk.get('metadata', {})
        internal_metadata = metadata_key.get('internal_key', {})

        # Extract common fields
        sheet_name = internal_metadata.get('sheet_name', 'Sheet1')
        sheet_number = internal_metadata.get('sheet_number', 1)

        # Build content based on element type
        if element_type == 'Text':
            content = self._build_text_content(chunk, internal_metadata)
        elif element_type == 'Image':
            content = self._build_image_content(chunk, internal_metadata)
        elif element_type == 'Shape':
            content = self._build_shape_content(chunk, internal_metadata)
        elif element_type == 'Comment':
            content = self._build_comment_content(chunk, internal_metadata)
        else:
            content = {'text': chunk.get('text', '')}

        # Build _source metadata
        source = {
            'document_id': str(document_id),
            'filename': filename,
            'sheet': sheet_name,
            'table_id': 0,  # Legacy chunks don't have table_id
            'row': self._extract_row_number(internal_metadata, element_type),
            'col_range': self._extract_col_range(internal_metadata, element_type),
        }

        # Create record in ExtractedRecord-like format
        record = {
            'content': content,
            'headers': list(content.keys()),
            '_source': source,
            '_legacy_metadata': {
                'element_type': element_type,
                'keywords_text': chunk.get('keywords_text', ''),
                'created_at': chunk.get('created_at', ''),
                'full_metadata': internal_metadata,
            }
        }

        return record

    def _build_text_content(
        self,
        chunk: Dict[str, Any],
        internal_metadata: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Build content dict for Text element type."""
        return {
            'text': chunk.get('text', ''),
            'sheet_name': internal_metadata.get('sheet_name', ''),
            'sheet_number': internal_metadata.get('sheet_number', 0),
        }

    def _build_image_content(
        self,
        chunk: Dict[str, Any],
        internal_metadata: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Build content dict for Image element type."""
        return {
            'description': chunk.get('text', ''),  # AI-generated description
            'image_url': internal_metadata.get('image_url', ''),
            'image_s3_key': internal_metadata.get('image_s3_key', ''),
            'image_format': internal_metadata.get('image_format', ''),
            'width': internal_metadata.get('width', 0),
            'height': internal_metadata.get('height', 0),
            'position_key': internal_metadata.get('position_key', ''),
            'cell_reference': internal_metadata.get('cell_reference', ''),
        }

    def _build_shape_content(
        self,
        chunk: Dict[str, Any],
        internal_metadata: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Build content dict for Shape element type."""
        connected_target = internal_metadata.get('connected_target', {})
        return {
            'shape_text': chunk.get('text', ''),
            'shape_type': internal_metadata.get('shape_type', ''),
            'target_cell': connected_target.get('cell_ref', ''),
            'target_value': connected_target.get('cell_value', ''),
            'position': str(internal_metadata.get('position', {})),
        }

    def _build_comment_content(
        self,
        chunk: Dict[str, Any],
        internal_metadata: Dict[str, Any]
    ) -> Dict[str, Any]:
        """Build content dict for Comment element type."""
        return {
            'comment_text': chunk.get('text', ''),
            'comment_id': internal_metadata.get('comment_id', ''),
            'author': internal_metadata.get('author', ''),
            'cell_address': internal_metadata.get('cell_address', ''),
            'cell_value': internal_metadata.get('cell_value', ''),
        }

    def _extract_row_number(
        self,
        internal_metadata: Dict[str, Any],
        element_type: str
    ) -> int:
        """Extract row number from metadata based on element type."""
        if element_type == 'Comment':
            return internal_metadata.get('row', 0)
        elif element_type in ['Image', 'Shape']:
            # Try to extract from position
            position = internal_metadata.get('position', {})
            from_cell = position.get('from_cell', '')
            if from_cell:
                # Extract row number from cell ref (e.g., "A10" -> 10)
                import re
                match = re.search(r'\d+', from_cell)
                return int(match.group()) if match else 0
        return 0

    def _extract_col_range(
        self,
        internal_metadata: Dict[str, Any],
        element_type: str
    ) -> str:
        """Extract column range from metadata."""
        if element_type in ['Image', 'Shape']:
            position = internal_metadata.get('position', {})
            from_cell = position.get('from_cell', '')
            to_cell = position.get('to_cell', '')
            if from_cell and to_cell:
                return f"{from_cell}:{to_cell}"
            elif from_cell:
                return from_cell
        elif element_type == 'Comment':
            return internal_metadata.get('cell_address', '')
        return ''

    def _prepare_embedding_text(self, chunk: Dict[str, Any]) -> str:
        """Prepare text for embedding generation.

        Since keyword/entity extraction is removed, simply return the text.
        """
        return chunk.get('text', '')

    def convert_to_milvus_entities(
        self,
        records: List[Dict[str, Any]],
        embeddings: List[List[float]],
        document_id: UUID,
        filename: str,
    ) -> List[Dict[str, Any]]:
        """Convert adapted records to final Milvus entity format.

        This creates the exact structure expected by VectorStore._prepare_milvus_entities.

        Args:
            records: Adapted records (ExtractedRecord-like)
            embeddings: Pre-computed embeddings
            document_id: Document UUID
            filename: Original filename

        Returns:
            List of dicts matching MilvusEntity structure
        """
        entities = []

        for record, embedding in zip(records, embeddings):
            # Extract sheet and row from _source
            source = record.get('_source', {})
            sheet_name = source.get('sheet', 'Sheet1')
            row_number = source.get('row', 0)

            # Check if has symbols (legacy metadata might indicate this)
            legacy_meta = record.get('_legacy_metadata', {})
            has_symbols = legacy_meta.get('element_type') in ['Shape', 'Image', 'Comment']

            # Prepare text_content (truncate for Milvus VARCHAR limit)
            content_text = self._record_to_text(record)
            if len(content_text) > 8000:
                content_text = content_text[:7997] + '...'

            # Build metadata for Milvus (following Excel format)
            metadata = {
                'sheet_name': sheet_name,
                'row_number': row_number,
                'has_symbols': has_symbols,
                # Add legacy-specific fields for filtering
                'element_type': legacy_meta.get('element_type', 'Text'),
                # 'has_keywords': Removed (no keyword extraction)
            }

            entity = {
                'vector': embedding,
                'record_id': str(uuid4()),  # Generate new UUID for each entity
                'document_id': str(document_id),
                'filename': filename,
                'document_type': 'excel',
                'text_content': content_text,
                'metadata': metadata,
            }

            entities.append(entity)

        return entities

    def _record_to_text(self, record: Dict[str, Any]) -> str:
        """Convert record content to text representation for storage."""
        content = record.get('content', {})

        # Format as key: value pairs
        parts = []
        for key, value in content.items():
            if value:
                parts.append(f"{key}: {value}")

        return ' | '.join(parts) if parts else ''
```

---

### Phase 4: Merging and Storage

#### Merging Strategy

```python
async def hybrid_excel_ingestion(
    file_content: bytes,
    file_id: UUID,
    original_filename: str,
    user_metadata: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    """Hybrid Excel ingestion using both pipelines.

    Both pipelines receive the ENTIRE file in parallel.
    NEW pipeline does large table detection internally.

    Returns:
        Combined ingestion result from both pipelines
    """
    # Initialize collections
    all_records = []
    all_embeddings = []
    large_table_metadata = []  # Will be populated by NEW pipeline

    # Step 1A: Trigger NEW Pipeline (Airflow) with ENTIRE file
    logger.info("Triggering NEW pipeline (Airflow) with entire file")

    # Trigger Airflow DAG (API call)
    # Airflow will detect large tables internally and process only those
    airflow_result = await trigger_airflow_excel_processing(
        document_id=str(file_id),
        file_content=file_content,
        object_key=object_key,  # MinIO path
    )

    # Extract records, embeddings, and large table metadata from Airflow result
    all_records.extend(airflow_result['records'])
    all_embeddings.extend(airflow_result['embeddings'])
    large_table_metadata = airflow_result.get('large_table_metadata', [])

    logger.info(
        f"NEW pipeline: {len(airflow_result['records'])} records from "
        f"{len(large_table_metadata)} large tables"
    )

    # Step 2B: Process ENTIRE file via OLD pipeline (UNCHANGED)
    logger.info("Processing ENTIRE file via OLD pipeline")

    # Get OLD pipeline service
    from excel_document_ingestion_service import get_excel_document_ingestion_service
    old_service = get_excel_document_ingestion_service()

    # Parse ENTIRE file (NO exclusions, NO modifications to OLD pipeline)
    chunks = await old_service.excel_parser_service.parse_excel_file(
        file_content=file_content,
        file_id=file_id,
        original_filename=original_filename,
        replace_images_with_descriptions=False,
    )

    # Process chunks WITHOUT keyword/entity extraction
    legacy_milvus_records = await old_service._process_chunks_for_milvus(
        chunks=chunks,
        extract_keywords_and_entities=False,  # DISABLED
        user_metadata=user_metadata,
    )

    logger.info(f"OLD pipeline: {len(legacy_milvus_records)} records")

    # Step 3: Adapt ALL chunks to new schema (no deduplication)
    from src.adapters.legacy_to_milvus_adapter import LegacyToMilvusAdapter

    adapter = LegacyToMilvusAdapter()
    adapted_result = await adapter.adapt_chunks(
        legacy_chunks=legacy_milvus_records,
        document_id=file_id,
        filename=original_filename,
    )

    all_records.extend(adapted_result['records'])
    all_embeddings.extend(adapted_result['embeddings'])

    logger.info(f"Total records: {len(all_records)}")

    # Step 4: Store in Milvus using NEW schema
    from src.knowledge.vector_store import VectorStore

    # Generate record IDs
    record_ids = [uuid4() for _ in all_records]

    vector_store = VectorStore()
    await vector_store.create_collection()

    stored_count = await vector_store.index_with_vectors(
        records=all_records,
        vectors=all_embeddings,
        document_id=file_id,
        record_ids=record_ids,
        document_type='excel',
        filename=original_filename,
    )

    return {
        'file_id': str(file_id),
        'original_filename': original_filename,
        'large_table_count': len(large_tables),
        'total_records': len(all_records),
        'stored_count': stored_count,
        'pipeline_breakdown': {
            'new_pipeline_records': len(airflow_result['records']) if large_tables else 0,
            'old_pipeline_records': len(adapted_result['records']),
        }
    }
```

---

## Schema Mapping Reference

### OLD Pipeline → NEW Milvus Schema Mapping

| OLD Field | NEW Field | Transformation | Notes |
|-----------|-----------|----------------|-------|
| `file_id` | `document_id` | Direct copy (as string) | UUID → str |
| `element_type` | `metadata.element_type` | Move to metadata | "Text", "Image", "Shape", "Comment" |
| `file_type` | `document_type` | Fixed value `"excel"` | Always "excel" for this pipeline |
| `text` | `text_content` | Direct copy, truncate to 8000 chars | Enforces Milvus VARCHAR limit |
| `keywords_text` | Embedded in `text_content` | Append as "\n\nKeywords: ..." | Used for embedding generation |
| `metadata.internal_key.sheet_name` | `metadata.sheet_name` | Direct copy | Core Excel metadata |
| `metadata.internal_key.sheet_number` | N/A | Stored in `_legacy_metadata` | Not in new schema |
| Extract row from position/cell | `metadata.row_number` | Parse from cell reference | For Images/Shapes/Comments |
| `metadata.internal_key.*` | `_legacy_metadata.full_metadata` | Preserve in record | For backward compatibility |
| `keywords_text` | ~~Removed~~ | N/A | Keyword extraction disabled |
| Element type | `metadata.has_symbols` | Boolean | True for Image/Shape/Comment |
| N/A (generate) | `record_id` | `uuid4()` | New UUID per entity |
| N/A (generate) | `vector` | Embed `text` only | 3072-dim from Azure OpenAI |
| `created_at` | N/A | Stored in `_legacy_metadata` | Not in new schema |

### NEW Pipeline → NEW Milvus Schema Mapping

| NEW Pipeline Field | NEW Milvus Field | Transformation | Notes |
|--------------------|------------------|----------------|-------|
| `content` (dict) | `text_content` | Format as "key: value \| key: value" | ExtractedRecord content |
| `_source.document_id` | `document_id` | Direct copy | Already in correct format |
| `_source.filename` | `filename` | Direct copy | Original filename |
| `_source.sheet` | `metadata.sheet_name` | Direct copy | Sheet name |
| `_source.row` | `metadata.row_number` | Direct copy | Row number |
| `resolved_content` (if exists) | `metadata.has_symbols` | `bool(resolved_content)` | Indicates symbol resolution |
| Embedding (pre-computed) | `vector` | Direct copy | Already 3072-dim |
| Record ID (generated) | `record_id` | From `record_ids` list | UUID from upstream |

---

## Implementation Checklist - Airflow DAG Tasks

### Architecture: Airflow Tasks, Not Microservices

This is an **Airflow DAG implementation** where each service class becomes an **Airflow task**. The hybrid pipeline will be implemented as a series of Airflow tasks using `@task.external_python` decorators.

### Current Excel Flow in Airflow DAG

```python
# airflow/dags/document_extraction_dag.py
branch >> parse_excel >> detect_headers >> chunk_excel >> embeddings_excel >> store_vectors_excel
```

### New Hybrid Excel Flow in Airflow DAG

```python
# New parallel flow for Excel documents
branch >> parse_excel_new_pipeline (existing: parse_excel task)
      >> parse_excel_old_pipeline (NEW TASK: runs old pipeline services)
      >> adapt_old_pipeline_schema (NEW TASK: schema adaptation)
      >> merge_pipeline_results (NEW TASK: combines both outputs)
      >> generate_embeddings_merged (existing: embeddings task)
      >> store_vectors_excel (existing: store task)
```

### 1. Migrate and Fix OLD Pipeline Services to `src/extraction_v2/`

**Why Migration is Needed:**
The OLD pipeline services were copied from a different project and have incorrect imports/dependencies:
- Import paths reference old project structure (`src.commons.configs.settings`)
- Dependencies may not exist in this project
- Services are currently in root directory instead of proper package structure

**Services to Migrate:**
1. **`azure_chat_service.py` → `src/extraction_v2/azure_chat_service.py`**
   - Fix: `from src.commons.configs.settings import get_settings` → `from src.core.config import settings`
   - Fix: Remove "chat" from setting names (e.g., `azure_openai_chat_deployment` → `azure_openai_deployment`)
   - Add missing settings to `src/core/config.py` if needed (temperature, max_tokens)

2. **`excel_image_extraction_service.py` → `src/extraction_v2/excel_image_extraction_service.py`**
   - Fix: Update import for `AzureChatService` to use new location
   - Fix: Update logger imports if needed
   - Remove: Any dependencies that don't exist in this project

3. **`excel_parser_service.py` → `src/extraction_v2/excel_parser_service.py`**
   - Fix: Update import for `ExcelImageExtractionService`
   - Fix: Update settings import path
   - Fix: Update DTO imports if needed
   - Remove: `ExcelShapeExtractionService` dependency (not needed in new pipeline)
   - Remove: `replace_images_with_descriptions` parameter (always replace with descriptions)
   - Remove: Any other unused dependencies

**After Migration:**
- Delete old files from root directory
- Update Airflow task imports to use new locations
- Services will be properly organized in `src/extraction_v2/` package

### 2. Create Airflow Task: `parse_excel_old_pipeline`

**File**: `airflow/dags/tasks/parse_tasks.py` (add new task)

```python
@task.external_python(python=EXTERNAL_PYTHON)
def parse_excel_old_pipeline(fetch_result: Dict[str, Any]) -> Dict[str, Any]:
    """
    Parse Excel using OLD pipeline (openpyxl-based).

    Runs the complete OLD pipeline:
    - ExcelParserService (text chunks, tables)
    - ExcelImageExtractionService (images + AI descriptions)
    - ExcelShapeExtractionService (shapes + connectors)
    - Comment extraction

    All imports must be inside function for external_python.

    Args:
        fetch_result: Result from fetch_document task

    Returns:
        Dict with data_file path containing all OLD pipeline chunks
    """
    import sys
    sys.path.insert(0, '/opt/airflow')

    from src.extraction_v2.excel_parser_service import get_excel_parser_service
    import asyncio
    import logging
    import pickle

    log = logging.getLogger(__name__)

    local_path = fetch_result['local_path']
    document_id = fetch_result['document_id']
    filename = fetch_result.get('original_filename', '')

    log.info(f"Running OLD pipeline for Excel document {document_id}")

    # Run OLD pipeline services
    async def run_old_pipeline():
        # Read file
        with open(local_path, 'rb') as f:
            file_content = f.read()

        # Parse Excel (text + tables + images + shapes + comments)
        # ExcelParserService internally handles all extraction:
        # - Text/tables via openpyxl
        # - Images replaced with AI-generated descriptions
        # - Shapes and comments are extracted directly
        parser_service = get_excel_parser_service()
        chunks = await parser_service.parse_excel_file(
            file_content=file_content,
            file_id=document_id,
            original_filename=filename,
            max_characters=2000,
        )

        log.info(f"OLD pipeline produced {len(chunks)} chunks")
        return chunks

    # Run async function
    chunks = asyncio.run(run_old_pipeline())

    # Save to pickle file
    data_file = f"/opt/airflow/logs/old_pipeline_{document_id}.pkl"
    with open(data_file, 'wb') as f:
        pickle.dump(chunks, f)

    log.info(f"Saved OLD pipeline result to {data_file}")

    return {
        'data_file': data_file,
        'document_id': document_id,
        'filename': filename,
        'chunk_count': len(chunks),
    }
```

### 3. Create Airflow Task: `adapt_old_pipeline_schema`

**File**: `airflow/dags/tasks/processing_tasks.py` (add new task)

```python
@task.external_python(python=EXTERNAL_PYTHON)
def adapt_old_pipeline_schema(old_pipeline_result: Dict[str, Any]) -> Dict[str, Any]:
    """
    Adapt OLD pipeline chunks to NEW Milvus schema.

    Uses LegacyToMilvusAdapter to convert OLD format to NEW format
    and generate embeddings.

    Args:
        old_pipeline_result: Result from parse_excel_old_pipeline task

    Returns:
        Dict with adapted records + embeddings (in pickle file)
    """
    import sys
    sys.path.insert(0, '/opt/airflow')

    from src.adapters.legacy_to_milvus_adapter import LegacyToMilvusAdapter
    from uuid import UUID
    import logging
    import pickle
    import os
    import asyncio

    log = logging.getLogger(__name__)

    # Load OLD pipeline chunks
    data_file = old_pipeline_result['data_file']
    document_id = old_pipeline_result['document_id']
    filename = old_pipeline_result['filename']

    log.info(f"Loading OLD pipeline chunks from {data_file}")
    with open(data_file, 'rb') as f:
        old_chunks = pickle.load(f)

    log.info(f"Adapting {len(old_chunks)} OLD pipeline chunks to NEW schema")

    # Run adapter
    async def adapt_chunks():
        adapter = LegacyToMilvusAdapter()
        return await adapter.adapt_chunks(
            legacy_chunks=old_chunks,
            document_id=UUID(document_id),
            filename=filename,
        )

    adapted_result = asyncio.run(adapt_chunks())

    log.info(
        f"Adaptation complete: {adapted_result['record_count']} records, "
        f"{len(adapted_result['embeddings'])} embeddings"
    )

    # Save adapted result
    adapted_file = f"/opt/airflow/logs/adapted_{document_id}.pkl"
    with open(adapted_file, 'wb') as f:
        pickle.dump(adapted_result, f)

    # Clean up OLD pipeline file
    try:
        os.remove(data_file)
        log.info(f"Cleaned up OLD pipeline file: {data_file}")
    except Exception as e:
        log.warning(f"Failed to clean up {data_file}: {e}")

    return {
        'data_file': adapted_file,
        'document_id': document_id,
        'filename': filename,
        'record_count': adapted_result['record_count'],
    }
```

### 4. Create Airflow Task: `merge_pipeline_results`

**File**: `airflow/dags/tasks/processing_tasks.py` (add new task)

```python
@task.external_python(python=EXTERNAL_PYTHON)
def merge_pipeline_results(
    new_pipeline_result: Dict[str, Any],
    adapted_old_result: Dict[str, Any]
) -> Dict[str, Any]:
    """
    Merge NEW pipeline (large tables) + adapted OLD pipeline results.

    Simply combines both outputs - no deduplication.

    Args:
        new_pipeline_result: Embeddings result from NEW pipeline (existing flow)
        adapted_old_result: Adapted OLD pipeline result

    Returns:
        Combined result ready for storage
    """
    import logging
    import pickle
    import os

    log = logging.getLogger(__name__)

    document_id = new_pipeline_result.get('document_id')

    # Load NEW pipeline embeddings result
    new_data_file = new_pipeline_result.get('data_file')
    log.info(f"Loading NEW pipeline result from {new_data_file}")
    with open(new_data_file, 'rb') as f:
        new_data = pickle.load(f)

    # Load adapted OLD pipeline result
    old_data_file = adapted_old_result['data_file']
    log.info(f"Loading adapted OLD pipeline result from {old_data_file}")
    with open(old_data_file, 'rb') as f:
        old_data = pickle.load(f)

    # Merge records and embeddings
    merged_records = new_data.get('records', []) + old_data.get('records', [])
    merged_embeddings = new_data.get('embeddings', []) + old_data.get('embeddings', [])

    log.info(
        f"Merged results: {len(new_data.get('records', []))} NEW + "
        f"{len(old_data.get('records', []))} OLD = {len(merged_records)} total"
    )

    # Save merged result
    merged_data = {
        'records': merged_records,
        'embeddings': merged_embeddings,
        'record_count': len(merged_records),
        'document_id': document_id,
    }

    merged_file = f"/opt/airflow/logs/merged_{document_id}.pkl"
    with open(merged_file, 'wb') as f:
        pickle.dump(merged_data, f)

    # Clean up intermediate files
    for temp_file in [new_data_file, old_data_file]:
        try:
            os.remove(temp_file)
            log.info(f"Cleaned up: {temp_file}")
        except Exception as e:
            log.warning(f"Failed to clean up {temp_file}: {e}")

    return {
        'data_file': merged_file,
        'document_id': document_id,
        'record_count': len(merged_records),
    }
```

### 5. Update DAG Dependencies

**File**: `airflow/dags/document_extraction_dag.py`

Replace the Excel flow section with:

```python
# OLD: Single pipeline flow
# branch >> parse_excel >> detect_headers >> chunk_excel >> embeddings_excel >> store_vectors_excel

# NEW: Hybrid parallel flow
parse_old = parse_excel_old_pipeline.override(
    task_id='parse_excel_old_pipeline',
    execution_timeout=PARSE_TIMEOUT,
)(fetch_result=fetch.output)

adapt_old = adapt_old_pipeline_schema.override(
    task_id='adapt_old_pipeline_schema',
)(old_pipeline_result=parse_old)

# Merge NEW pipeline embeddings + adapted OLD pipeline
merge_results = merge_pipeline_results.override(
    task_id='merge_results',
)(
    new_pipeline_result=embeddings_excel,
    adapted_old_result=adapt_old
)

# Store merged results
store_vectors_excel_merged = store_vectors_task.override(
    task_id='store_vectors_excel',
    execution_timeout=STORE_TIMEOUT,
)(
    document_id=doc_id,
    embedded_result=merge_results,
)

# Dependencies:
# Excel path (parallel NEW + OLD):
#   NEW: branch >> parse_excel >> detect_headers >> chunk_excel >> embeddings_excel ─┐
#                                                                                      ├─> merge_results >> store
#   OLD: branch >> fetch >> parse_old >> adapt_old ──────────────────────────────────┘

branch >> parse_excel >> detect_headers >> chunk_excel >> embeddings_excel
branch >> parse_old >> adapt_old
[embeddings_excel, adapt_old] >> merge_results >> store_vectors_excel_merged
```

### 6. Create Legacy Adapter Module

**File**: `src/adapters/legacy_to_milvus_adapter.py`

This module is already defined in the Schema Adaptation section above (lines 251-639).

Key points:
- Converts OLD pipeline chunk format to NEW Milvus entity format
- Handles all element types: Text, Image, Shape, Comment
- Generates embeddings for adapted chunks
- No deduplication logic (simplified approach)

---

## Data Flow Diagrams

### Complete Hybrid Flow (Corrected)

```
┌──────────────────────────────────────────────────────────────┐
│                     Excel File Upload                         │
│                    (file_content: bytes)                      │
└──────────────┬───────────────────────────────────────────────┘
               │
               │ File sent to BOTH pipelines in PARALLEL
               │
       ┌───────┴──────────┐
       │                  │
       ▼                  ▼
┌──────────────┐   ┌─────────────────────┐
│ NEW Pipeline │   │   OLD Pipeline      │
│ (Airflow)    │   │   (openpyxl)        │
│              │   │                     │
│ ENTIRE FILE  │   │   ENTIRE FILE       │
│              │   │   (UNCHANGED)       │
└──────┬───────┘   └──────┬──────────────┘
       │                  │
       │ Step 1:          │ Step 1:
       │ Detect Large     │ Parse Everything
       │ Tables           │ • Text chunks
       │ (internal)       │ • Images + AI desc
       │                  │ • Shapes + connectors
       │ Step 2:          │ • Comments
       │ Process ONLY     │ • MinIO upload
       │ Large Tables     │
       │ • Docling        │ (NO keyword/entity)
       │ • LLM headers    │
       │ • Chunked        │
       │   extract        │
       │ • RecordBuild    │
       │ • Embed          │
       │                  │
       ▼                  ▼
┌──────────────┐   ┌─────────────────────┐
│ Output:      │   │ Output:             │
│              │   │                     │
│ records[]    │   │ all_chunks[]        │
│ embeddings[] │   │ (includes large     │
│              │   │  table chunks)      │
│ (NEW schema) │   │                     │
└──────┬───────┘   └──────┬──────────────┘
       │                  │
       │                  ▼
       │           ┌─────────────────────┐
       │           │ Schema Adapter      │
       │           │                     │
       │           │ • Convert ALL to    │
       │           │   NEW Milvus schema │
       │           │ • Generate          │
       │           │   embeddings        │
       │           └──────┬──────────────┘
       │                  │
       │                  ▼
       │           ┌─────────────────────┐
       │           │ Output:             │
       │           │                     │
       │           │ records[]           │
       │           │ embeddings[]        │
       │           │ (NEW schema)        │
       │           └──────┬──────────────┘
       │                  │
       └──────┬───────────┘
              │
              ▼
┌──────────────────────────────────────────────────────────────┐
│  Merge Results                                                │
│  • Combine NEW pipeline records + OLD pipeline records       │
│  • All in NEW Milvus schema format                           │
│  • Some text overlap from large tables (acceptable)          │
└──────────────┬───────────────────────────────────────────────┘
               │
               ▼
┌──────────────────────────────────────────────────────────────┐
│  Unified Storage (VectorStore.index_with_vectors)            │
│  • Single batch insert to Milvus                             │
│  • NEW schema for all records                                │
│  • Supports filtering by element_type, sheet_name, etc.      │
└──────────────────────────────────────────────────────────────┘
```

**Key Points:**
- ✅ BOTH pipelines receive the ENTIRE file in parallel
- ✅ NEW pipeline does large table detection internally
- ✅ OLD pipeline processes everything unchanged
- ✅ Outputs are merged directly (some text overlap accepted)
- ✅ Simple and clean separation of concerns

---

## Testing Strategy

### Unit Tests

1. **LegacyToMilvusAdapter Tests**
   - Test conversion for each element type (Text, Image, Shape, Comment)
   - Validate field mappings
   - Verify embedding text preparation
   - Check truncation logic

2. **AirflowClient Tests**
   - Mock DAG trigger
   - Mock polling logic
   - Test timeout handling
   - Validate result parsing

3. **Schema Validation Tests**
   - Verify NEW schema compatibility
   - Test Milvus entity creation
   - Validate metadata structure

### Integration Tests

1. **End-to-End Hybrid Flow**
   - Upload test Excel file with large table
   - Verify routing to both pipelines
   - Validate merged results in Milvus
   - Query and verify all element types

2. **Large Table Only**
   - Test file with only large tables
   - Verify OLD pipeline skipped

3. **Normal Content Only**
   - Test file with no large tables
   - Verify NEW pipeline skipped

### Performance Tests

1. **Large File Processing**
   - 10,000+ row tables
   - Multiple large tables
   - Measure processing time per pipeline

2. **Memory Usage**
   - Monitor OLD pipeline memory (images, shapes)
   - Monitor NEW pipeline memory (chunked extraction)

3. **Milvus Storage**
   - Validate batch insertion performance
   - Verify query performance with overlapping records

---

## Rollout Plan

### Phase 1: Development (Weeks 1-2)
- [ ] **Step 1: Migrate OLD Pipeline Services**
  - [ ] Migrate `azure_chat_service.py` → `src/extraction_v2/azure_chat_service.py`
  - [ ] Migrate `excel_image_extraction_service.py` → `src/extraction_v2/excel_image_extraction_service.py`
  - [ ] Migrate `excel_parser_service.py` → `src/extraction_v2/excel_parser_service.py`
  - [ ] Fix all imports to use correct paths (`src.core.config`, `src.extraction_v2.*`)
  - [ ] Add missing settings to `src/core/config.py` (temperature, max_tokens)
  - [ ] Remove "chat" from setting names in azure_chat_service.py
  - [ ] Test migrated services can be imported successfully
- [ ] **Step 2: Implement Adapter**
  - [ ] Implement `src/adapters/legacy_to_milvus_adapter.py`
- [ ] **Step 3: Create Airflow Tasks**
  - [ ] Create `parse_excel_old_pipeline` task in `airflow/dags/tasks/parse_tasks.py`
  - [ ] Create `adapt_old_pipeline_schema` task in `airflow/dags/tasks/processing_tasks.py`
  - [ ] Create `merge_pipeline_results` task in `airflow/dags/tasks/processing_tasks.py`
- [ ] **Step 4: Update DAG**
  - [ ] Update `airflow/dags/document_extraction_dag.py` with hybrid Excel flow

### Phase 2: Testing (Week 3)
- [ ] Unit tests for all adapters
- [ ] Integration tests with sample files
- [ ] Performance benchmarking

### Phase 3: Staging Deployment (Week 4)
- [ ] Deploy to staging environment
- [ ] Run regression tests
- [ ] Validate Milvus schema compatibility

### Phase 4: Production Rollout (Week 5)
- [ ] Gradual rollout (10% → 50% → 100%)
- [ ] Monitor error rates
- [ ] Performance monitoring
- [ ] Rollback plan ready

---

## Backward Compatibility

### Query Compatibility

Queries must work for both old and new records:

```python
# Search query example
results = vector_store.search(
    query_vector=embedding,
    filter_expr='document_id == "uuid" and metadata["element_type"] == "Image"',
    limit=10
)

# Works for:
# - NEW pipeline records (no element_type in metadata)
# - OLD pipeline records (element_type in metadata)
```

### Metadata Filtering

Enable filtering by legacy fields:

```python
# Filter by element type (legacy field)
filter_expr = 'metadata["element_type"] == "Shape"'

# Filter by has_symbols (new field, derived from element_type)
filter_expr = 'metadata["has_symbols"] == true'

# Combined filter
filter_expr = (
    'document_type == "excel" and '
    'metadata["sheet_name"] == "Sheet1" and '
    'metadata["element_type"] == "Image"'
)
```

---

## Monitoring and Observability

### Metrics to Track

1. **Pipeline Routing**
   - % files routed to NEW pipeline
   - % files routed to OLD pipeline
   - % files using both pipelines

2. **Processing Time**
   - NEW pipeline avg time
   - OLD pipeline avg time
   - Combined time for hybrid files

3. **Storage Metrics**
   - Records per pipeline
   - Embedding generation time
   - Milvus insertion rate

4. **Error Rates**
   - Adapter conversion errors
   - Airflow DAG failures
   - Milvus storage errors

### Logging

```python
logger.info(
    "hybrid_ingestion_complete",
    document_id=str(file_id),
    filename=original_filename,
    large_table_count=len(large_tables),
    new_pipeline_records=new_count,
    old_pipeline_records=old_count,
    total_records=total_count,
    processing_time_seconds=elapsed_time,
)
```

---

## Migration from Pure OLD Pipeline

If you have existing documents in the OLD schema:

### Option 1: Dual Schema Support
- Keep both schemas in Milvus (separate collections)
- Update queries to search both collections
- Gradually migrate documents

### Option 2: Batch Migration
- Run adapter on all existing documents
- Create new collection with NEW schema
- Atomic swap collections

### Option 3: On-Demand Migration
- Lazy migration during re-indexing
- Keep OLD schema for existing documents
- Use NEW schema for new documents

---

## Summary

This **simplified hybrid approach** provides:

✅ **Minimal Code Changes**
- **ZERO modifications to OLD pipeline** (used as-is)
- Schema adapter for format conversion
- Simple merge without deduplication
- Easy to implement and maintain

✅ **Best of Both Worlds**
- Large table handling from NEW pipeline (Docling/LLM headers)
- Rich features from OLD pipeline (images, shapes, comments, AI descriptions)

✅ **Unified Storage**
- Single Milvus schema for all data
- Consistent query interface
- Future-proof architecture

✅ **Simplified Processing**
- No keyword/entity extraction (removed for simplicity)
- OLD pipeline processes entire file without changes
- Direct merge (some text overlap acceptable)

✅ **Scalable**
- Large tables (>1000 rows) processed efficiently via Airflow
- Normal content with rich metadata via OLD pipeline
- Parallel processing of both paths

---

**Implementation Complexity**: ⭐ (Very Low)
- No OLD pipeline modifications required
- 1 adapter module + 3 Airflow tasks
- No deduplication logic needed
- Straightforward testing and rollout

**Next Steps:**
1. Review and approve Airflow DAG design
2. **Phase 1: Migrate OLD Pipeline Services**
   - Migrate 3 service files to `src/extraction_v2/`
   - Fix imports (remove `src.commons.*`, use `src.core.config`)
   - Remove "chat" from setting names
   - Add missing Azure OpenAI settings to config
3. **Phase 2: Implement Components**
   - `src/adapters/legacy_to_milvus_adapter.py` (adapter module)
   - `parse_excel_old_pipeline` (Airflow task)
   - `adapt_old_pipeline_schema` (Airflow task)
   - `merge_pipeline_results` (Airflow task)
   - Update DAG dependencies in `document_extraction_dag.py`
4. Create integration tests
5. Deploy to Airflow staging for validation