# New Airflow Document Extraction Pipeline

## Overview

This document describes the new Airflow-based document extraction pipeline architecture. The entry point is the `document_extraction_dag.py` DAG in Airflow, which orchestrates multi-format document processing through a series of tasks running in isolated Python environments.

## Key Architectural Changes from Legacy

### 1. **Orchestration**: Airflow DAG (Event-Driven)
- **Old**: Direct Python service calls
- **New**: Apache Airflow DAG with task dependencies and branching

### 2. **Execution Environment**: External Python (uv venv)
- **Old**: Single Python environment
- **New**: Heavy processing tasks run in `/opt/airflow/.venv` managed by `uv`
- Benefits: Dependency isolation, better control over package versions

### 3. **Document Parsing**: Docling-based
- **Old**: openpyxl + custom image/shape extraction
- **New**: Docling framework for all formats + large table detection

### 4. **Data Transfer**: Pickle files + XCom
- **Old**: Direct in-memory data passing
- **New**: Large data saved to pickle files in `/opt/airflow/logs`, small metadata via XCom

### 5. **Multi-Format Support**
- **Old**: Excel-only
- **New**: Excel, Word (.docx, .doc, .docm), PDF, PowerPoint (.pptx, .ppt, .pptm)

---

## Architecture Diagram

```
┌──────────────────────────────────────────────────────────────────────┐
│                      Airflow Document Extraction DAG                  │
│                    (document_extraction_dag.py)                       │
└────────────────────┬─────────────────────────────────────────────────┘
                     │
      ┌──────────────┴──────────────┐
      │   Event Trigger (API/CLI)   │
      │   Params:                   │
      │   - document_id (UUID)      │
      │   - object_key (path)       │
      │   - content_type (MIME)     │
      └──────────────┬──────────────┘
                     │
┌────────────────────▼────────────────────┐
│ Task 1: ValidateEventOperator           │
│ - Validates event structure             │
│ - Extracts document_id, object_key      │
└────────────────────┬────────────────────┘
                     │
┌────────────────────▼────────────────────┐
│ Task 2: FetchDocumentOperator           │
│ - Downloads from MinIO to local disk    │
│ - Returns local_path for processing     │
└────────────────────┬────────────────────┘
                     │
┌────────────────────▼────────────────────┐
│ Task 3: detect_format (Branch)          │
│ - Detects file extension                │
│ - Routes to appropriate parse task      │
└────┬────────┬────────┬────────┬─────────┘
     │        │        │        │
     │ .xlsx  │ .docx  │ .pdf   │ .pptx
     │        │        │        │
┌────▼──┐ ┌───▼───┐ ┌──▼──┐ ┌───▼────┐
│Excel  │ │Word   │ │PDF  │ │PowerPt │
│Parse  │ │Parse  │ │Parse│ │Parse   │
└────┬──┘ └───┬───┘ └──┬──┘ └───┬────┘
     │        │        │        │
     │        └────┬───┴────────┘
     │             │ (text docs merge)
     │             │
┌────▼────────┐    │
│ Detect      │    │
│ Headers     │    │
│ (LLM)       │    │
└────┬────────┘    │
     │             │
┌────▼────────┐ ┌──▼─────────────┐
│ Chunk       │ │ Chunk          │
│ Excel       │ │ Text           │
│ (Table)     │ │ (Semantic)     │
└────┬────────┘ └──┬─────────────┘
     │             │
┌────▼────────┐ ┌──▼─────────────┐
│ Generate    │ │ Generate       │
│ Embeddings  │ │ Embeddings     │
│ (Excel)     │ │ (Text)         │
└────┬────────┘ └──┬─────────────┘
     │             │
┌────▼────────┐ ┌──▼─────────────┐
│ Store       │ │ Store          │
│ Vectors     │ │ Vectors        │
│ (Milvus)    │ │ (Milvus)       │
└─────────────┘ └────────────────┘
```

---

## Pipeline Stages

### Stage 1: Event Validation & Document Fetch

#### Task 1.1: ValidateEventOperator
**File**: `airflow/plugins/operators/validate_event.py`

**Responsibilities**:
- Validates incoming event structure
- Ensures required fields: `document_id`, `object_key`, `content_type`
- Returns validated event dict

**Output (XCom)**:
```python
{
    'document_id': 'uuid-string',
    'object_key': 'path/to/file.xlsx',
    'content_type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
}
```

#### Task 1.2: FetchDocumentOperator
**File**: `airflow/plugins/operators/fetch_document.py`

**Responsibilities**:
- Downloads document from MinIO storage
- Saves to `/opt/airflow/logs/` (shared volume)
- Returns local path

**Output (XCom)**:
```python
{
    'local_path': '/opt/airflow/logs/temp_abc123.xlsx',
    'document_id': 'uuid-string',
    'original_filename': 'report.xlsx'
}
```

---

### Stage 2: Format Detection & Branching

#### Task 2: detect_format (TaskFlow Branch)
**File**: `document_extraction_dag.py:56`

**Logic**:
```python
Extension → Task ID
.xlsx, .xls, .xlsm → 'parse_excel'
.docx, .doc, .docm → 'parse_word'
.pdf → 'parse_pdf'
.pptx, .ppt, .pptm → 'parse_powerpoint'
Other → 'unsupported_format'
```

---

### Stage 3: Document Parsing (External Python)

All parse tasks run in `/opt/airflow/.venv` using `@task.external_python` decorator.

#### Task 3.1: parse_excel_document
**File**: `airflow/dags/tasks/parse_tasks.py:22`

**Processing Flow**:

```
1. Large Table Detection
   ├─ Use LargeTableExtractor.detect_tables()
   ├─ Classify: large (>1000 rows) vs normal tables
   └─ Result: detected_tables with is_large flag

2. Large Table Processing (Chunked Extraction)
   ├─ For each large table:
   │  ├─ LargeTableExtractor.extract_table()
   │  ├─ Chunks table into manageable pieces
   │  └─ Returns pre-processed records
   └─ Output: large_table_records[]

3. Normal Table Processing (Docling Flow)
   ├─ DocumentToHtmlConverter.convert_to_html()
   ├─ HtmlTableExtractor.extract_all_tables()
   ├─ Returns: valid_tables[] + raw_text_tables[]
   └─ Each table includes sheet_name mapping

4. Save to Pickle
   ├─ File: /opt/airflow/logs/parse_excel_{document_id}.pkl
   └─ Data: {html, tables, raw_text_tables, large_table_records}
```

**Output (XCom)**:
```python
{
    'data_file': '/opt/airflow/logs/parse_excel_abc123.pkl',
    'doc_type': 'excel',
    'document_id': 'uuid',
    'original_filename': 'report.xlsx',
    'local_path': '/opt/airflow/logs/temp_abc123.xlsx',
    'table_count': 15,
    'large_table_record_count': 500
}
```

#### Task 3.2: parse_word_document
**File**: `airflow/dags/tasks/parse_tasks.py:161`

**Processing Flow**:

```
1. Format Conversion (if needed)
   ├─ If .doc or .docm:
   │  ├─ DocumentToHtmlConverter._convert_doc_to_docx()
   │  ├─ Save converted .docx for reuse
   │  └─ Cleanup temp conversion directory
   └─ Else: Use original .docx

2. Markdown Conversion
   ├─ DocumentToHtmlConverter.convert_to_html()
   └─ Returns: markdown text

3. Save to Pickle
   ├─ File: /opt/airflow/logs/parse_word_{document_id}.pkl
   └─ Data: {markdown_content, converted_path, ...}
```

**Output (XCom)**:
```python
{
    'data_file': '/opt/airflow/logs/parse_word_abc123.pkl',
    'doc_type': 'word',
    'document_id': 'uuid',
    'original_filename': 'document.docx',
    'local_path': '/opt/airflow/logs/temp_abc123.docx',
    'converted_path': '/opt/airflow/logs/temp_abc123_converted.docx',  # If .doc/.docm
    'content_length': 50000
}
```

#### Task 3.3: parse_pdf_document
**File**: `airflow/dags/tasks/parse_tasks.py:266`

**Processing Flow**:
```
1. Markdown Conversion
   └─ DocumentToHtmlConverter.convert_to_html()

2. Save to Pickle
   └─ File: /opt/airflow/logs/parse_pdf_{document_id}.pkl
```

#### Task 3.4: parse_powerpoint_document
**File**: `airflow/dags/tasks/parse_tasks.py:330`

**Processing Flow**: Same as PDF

---

### Stage 4: Excel Header Detection (LLM-based)

#### Task 4: detect_headers_task
**File**: `airflow/dags/tasks/processing_tasks.py:16`

**Only runs for Excel documents**

**Processing Flow**:

```
1. Load Parse Data
   └─ Read from pickle file (data_file from parse_excel)

2. LLM Header Detection
   ├─ For each table:
   │  ├─ LlmHeaderDetector.detect_headers(table_html)
   │  ├─ Returns: HeaderStructure {header_rows, headers}
   │  └─ Fallback: {header_rows: [0], headers: []} if LLM fails
   └─ Output: all_headers[]

3. Return Header Data
   └─ Via XCom (no pickle needed, small data)
```

**Output (XCom)**:
```python
{
    'tables_processed': 15,
    'header_data': [
        {
            'table_index': 0,
            'header_rows': [0, 1],  # Multi-row headers
            'headers': ['Name', 'Q1', 'Q2', 'Q3', 'Q4']
        },
        ...
    ]
}
```

---

### Stage 5: Chunking

Two parallel paths based on document type.

#### Task 5.1: chunk_excel_task (Table-based Chunking)
**File**: `airflow/dags/tasks/processing_tasks.py:105`

**Processing Flow**:

```
1. Load Parse + Header Data
   ├─ Load from parse pickle file
   ├─ Extract: tables, raw_text_tables, large_table_records
   └─ Load header_data from headers_result

2. Build Records
   ├─ A. Pre-processed Large Tables
   │  └─ Include large_table_records directly
   │
   ├─ B. Normal Tables
   │  ├─ For each table:
   │  │  ├─ Match headers from header_data
   │  │  ├─ Create HtmlTable object
   │  │  ├─ RecordBuilder.build_records(table, headers)
   │  │  └─ Returns: structured records (dict per row)
   │  └─ Output: all_records[]
   │
   └─ C. Raw Text Tables
      ├─ Group by sheet_name
      ├─ RecordBuilder.build_combined_raw_text_records()
      ├─ Handles token limits (max 8000)
      ├─ Merges small tables (min 100 tokens)
      └─ Output: text records

3. Add Metadata
   ├─ Add filename to each record's _source
   ├─ Ensure col_range exists
   └─ Ready for embedding

4. Save to Pickle
   ├─ File: /opt/airflow/logs/chunks_excel_{document_id}.pkl
   └─ Data: {chunks: all_records, chunking_strategy: 'table'}
```

**Output (XCom)**:
```python
{
    'data_file': '/opt/airflow/logs/chunks_excel_abc123.pkl',
    'chunk_count': 523,
    'chunking_strategy': 'table',
    'doc_type': 'excel',
    'original_filename': 'report.xlsx'
}
```

#### Task 5.2: chunk_text_task (Semantic Chunking)
**File**: `airflow/dags/tasks/processing_tasks.py:285`

**Processing Flow**:

```
1. Load Parse Data
   └─ Read from pickle file

2. Handle Legacy Formats
   ├─ Check for pre-converted file (from parse task)
   ├─ If .doc/.docm and no converted file:
   │  ├─ DocumentToHtmlConverter._convert_doc_to_docx()
   │  └─ Use converted .docx
   └─ Else: Use original file

3. Re-convert with Docling
   ├─ DocumentConverter().convert(file_path)
   └─ Get full document structure

4. Semantic Chunking
   ├─ SemanticChunker(max_tokens=8000)
   ├─ chunk_document(doc, file_id, filename)
   ├─ Intelligent boundary detection
   └─ Returns: semantic chunks[]

5. Add Metadata
   └─ Add doc_type, original_filename to each chunk

6. Save to Pickle
   └─ File: /opt/airflow/logs/chunks_text_{document_id}.pkl
```

**Output (XCom)**:
```python
{
    'data_file': '/opt/airflow/logs/chunks_text_abc123.pkl',
    'chunk_count': 42,
    'chunking_strategy': 'semantic',
    'doc_type': 'word',  # or 'pdf', 'powerpoint'
    'original_filename': 'document.docx'
}
```

---

### Stage 6: Embedding Generation

#### Task 6: generate_embeddings_task
**File**: `airflow/dags/tasks/processing_tasks.py:440`

**Processing Flow**:

```
1. Load Chunk Data
   ├─ Read from pickle file
   └─ Extract chunks[]

2. Prepare Texts
   ├─ For dict chunks (Excel):
   │  └─ Format as "key: value | key: value ..."
   ├─ For string chunks:
   │  └─ Extract content/text field
   └─ Filter out empty texts

3. Generate Embeddings (Batched)
   ├─ EmbeddingService.embed_batch(batch)
   ├─ Batch size: min(batch_size, 100) (Azure OpenAI limit)
   ├─ Process in batches to avoid rate limits
   └─ Returns: embeddings[] (1536-dim vectors)

4. Save to Pickle
   ├─ File: /opt/airflow/logs/embedding_data/embeddings_{uuid}.pkl
   └─ Data: {chunks, embeddings, chunk_count, doc_type}
```

**Output (XCom)**:
```python
{
    'data_file': '/opt/airflow/logs/embedding_data/embeddings_abc123.pkl',
    'chunk_count': 523,
    'doc_type': 'excel',
    'original_filename': 'report.xlsx'
}
```

---

### Stage 7: Vector Storage

#### Task 7: store_vectors_task
**File**: `airflow/dags/tasks/processing_tasks.py:575`

**Processing Flow**:

```
1. Load Embedded Data
   ├─ Read from pickle file
   ├─ Extract: chunks[], embeddings[]
   └─ Cleanup pickle file after loading

2. Prepare Records
   ├─ If Excel (doc_type='excel'):
   │  ├─ Convert dict chunks to ExtractedRecord objects
   │  ├─ Ensure complete _source metadata:
   │  │  - document_id, filename, sheet, table_id, row, col_range
   │  └─ Returns: records[]
   └─ Else: Use chunks directly

3. Store in Milvus
   ├─ Generate record_ids (UUID for each chunk)
   ├─ VectorStore.index_with_vectors()
   │  ├─ records: data chunks
   │  ├─ vectors: embeddings
   │  ├─ document_id: file UUID
   │  ├─ record_ids: chunk UUIDs
   │  ├─ document_type: excel/word/pdf/powerpoint
   │  └─ filename: original filename
   └─ Returns: stored_count

4. Return Result
   └─ {stored_count, success: True}
```

**Output (XCom)**:
```python
{
    'stored_count': 523,
    'success': True
}
```

---

## Data Structures

### Parse Result (Excel - in pickle file)
```python
{
    'doc_type': 'excel',
    'html_content': str,  # Full HTML from Docling
    'markdown_content': '',
    'tables': [
        {
            'html': str,  # Table HTML
            'rows': List[List[str]],  # 2D array
            'sheet_name': str  # Sheet mapping
        }
    ],
    'raw_text_tables': [
        {
            'rows': List[List[str]],
            'sheet_name': str
        }
    ],
    'large_table_records': [
        {
            'content': Dict[str, Any],  # Pre-built record
            'headers': List[str],
            '_source': {...}
        }
    ],
    'local_path': str,
    'document_id': str,
    'original_filename': str
}
```

### Parse Result (Text Documents - in pickle file)
```python
{
    'doc_type': 'word' | 'pdf' | 'powerpoint',
    'html_content': '',
    'markdown_content': str,  # Full markdown
    'tables': [],
    'raw_text_tables': [],
    'local_path': str,
    'converted_path': str,  # Only for .doc/.docm
    'document_id': str,
    'original_filename': str
}
```

### Header Detection Result
```python
{
    'tables_processed': int,
    'header_data': [
        {
            'table_index': int,
            'header_rows': List[int],  # e.g., [0] or [0, 1]
            'headers': List[str]  # Column names
        }
    ]
}
```

### Chunk Result (Excel - in pickle file)
```python
{
    'chunks': [
        {
            'content': Dict[str, Any],  # Row data as dict
            'headers': List[str],
            '_source': {
                'document_id': str,
                'filename': str,
                'sheet': str,
                'table_id': int,
                'row': int,
                'col_range': str
            },
            '_merge_metadata': Optional[Dict],
            'resolved_content': Optional[Dict]
        }
    ],
    'chunk_count': int,
    'chunking_strategy': 'table',
    'doc_type': 'excel',
    'original_filename': str
}
```

### Chunk Result (Text - in pickle file)
```python
{
    'chunks': [
        {
            'content': str,  # Semantic chunk text
            'doc_type': str,
            'original_filename': str,
            # Additional semantic metadata from SemanticChunker
        }
    ],
    'chunk_count': int,
    'chunking_strategy': 'semantic',
    'doc_type': 'word' | 'pdf' | 'powerpoint',
    'original_filename': str
}
```

### Embedded Result (in pickle file)
```python
{
    'chunks': List[Dict],  # Original chunks
    'embeddings': List[List[float]],  # 1536-dim vectors
    'chunk_count': int,
    'doc_type': str,
    'original_filename': str
}
```

---

## Key Components

### 1. Document Converter
**Module**: `src.extraction_v2.document_converter`

**Responsibilities**:
- Unified interface for all document formats
- Docling integration for conversion
- LibreOffice integration for .doc/.docm conversion
- Sheet name mapping for Excel files

### 2. Large Table Extractor
**Module**: `src.extraction_v2.large_table_extractor`

**Responsibilities**:
- Detect large tables (>1000 rows) via border detection
- Chunked extraction to avoid memory issues
- Pre-process large tables into records before Docling

### 3. HTML Table Extractor
**Module**: `src.extraction_v2.html_table_extractor`

**Responsibilities**:
- Extract tables from Docling HTML output
- Classify: data tables vs raw text tables
- Parse HTML table structure into 2D arrays

### 4. LLM Header Detector
**Module**: `src.extraction_v2.llm_header_detector`

**Responsibilities**:
- Use Azure OpenAI to detect header rows in tables
- Handle multi-row headers
- Extract column names
- Fallback: assume row 0 is header

### 5. Record Builder
**Module**: `src.extraction_v2.record_builder`

**Responsibilities**:
- Convert table rows to structured records
- Apply headers to create dict-based records
- Handle token limits for raw text tables
- Merge small tables to optimize storage

### 6. Semantic Chunker
**Module**: `src.extraction_v2.semantic_chunker`

**Responsibilities**:
- Intelligent boundary detection for text documents
- Respects semantic structure (paragraphs, sections)
- Max tokens: 8000 (text-embedding-3-large limit)
- Uses Docling document structure

### 7. Embedding Service
**Module**: `src.knowledge.embeddings`

**Responsibilities**:
- Azure OpenAI text-embedding-3-large integration
- Batch processing (max 100 per call)
- Async embedding generation
- Error handling and retries

### 8. Vector Store
**Module**: `src.knowledge.vector_store`

**Responsibilities**:
- Milvus integration
- Index management
- Vector insertion with metadata
- Document type handling

---

## Configuration

### Airflow Environment Variables
```bash
# Task timeouts (in minutes)
AIRFLOW_PARSE_TIMEOUT_MINUTES=120
AIRFLOW_EMBEDDING_TIMEOUT_MINUTES=60
AIRFLOW_STORE_TIMEOUT_MINUTES=30
```

### External Python Environment
```
Location: /opt/airflow/.venv
Manager: uv (fast Python package manager)
Created by: `uv sync` during Docker build
Isolated from: Airflow's main Python environment
```

### Data Storage Locations
```
Parse results: /opt/airflow/logs/parse_{type}_{doc_id}.pkl
Chunk results: /opt/airflow/logs/chunks_{type}_{doc_id}.pkl
Embedding results: /opt/airflow/logs/embedding_data/embeddings_{uuid}.pkl
Temp documents: /opt/airflow/logs/temp_{uuid}.{ext}
```

---

## Performance Optimizations

### 1. Large Table Handling
- **Detection**: Border-based detection before Docling
- **Chunked Processing**: Process >1000-row tables in chunks
- **Pre-processing**: Convert to records before embedding

### 2. XCom Size Management
- **Large Data**: Saved to pickle files
- **Small Metadata**: Passed via XCom
- **Cleanup**: Pickle files deleted after consumption

### 3. Format Conversion Caching
- **Word .doc/.docm**: Convert once in parse task
- **Reuse**: Pass converted path to chunk task
- **Avoid**: Double conversion overhead

### 4. Batch Embedding
- **Batch Size**: 16 (configurable, max 100)
- **Parallel Processing**: Multiple chunks per API call
- **Rate Limiting**: Respects Azure OpenAI limits

### 5. Token Limit Handling
- **Excel Raw Text**: Max 8000 tokens per record
- **Small Table Merging**: Min 100 tokens threshold
- **Semantic Chunking**: Max 8000 tokens per chunk

---

## Error Handling

### Task-Level Retries
- **Retries**: 0 (disabled by default)
- **Timeout**: Per-task execution timeouts
- **Trigger Rules**: `none_failed_min_one_success` for text merge

### Graceful Degradation
- **Header Detection**: Fallback to row 0 if LLM fails
- **Empty Content**: Filter out empty chunks before embedding
- **Conversion Failures**: Raise ValueError with context

### Cleanup
- **Pickle Files**: Deleted after consumption
- **Temp Files**: Cleaned up in finally blocks
- **Converted Files**: Preserved during task chain, cleaned at end

---

## DAG Trigger

### Event-Based Triggering
```bash
# Via Airflow CLI
airflow dags trigger document_extraction_dag \
  --conf '{
    "document_id": "550e8400-e29b-41d4-a716-446655440000",
    "object_key": "uploads/report.xlsx",
    "content_type": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
  }'
```

### Via API
```python
import requests

response = requests.post(
    'http://localhost:8080/api/v1/dags/document_extraction_dag/dagRuns',
    json={
        'conf': {
            'document_id': '550e8400-e29b-41d4-a716-446655440000',
            'object_key': 'uploads/report.xlsx',
            'content_type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
        }
    },
    auth=('airflow', 'airflow')
)
```

---

## Comparison with Legacy Pipeline

| Aspect | Legacy Pipeline | New Airflow Pipeline |
|--------|----------------|---------------------|
| **Orchestration** | Direct Python calls | Airflow DAG with dependencies |
| **Formats** | Excel only | Excel, Word, PDF, PowerPoint |
| **Parsing** | openpyxl + custom | Docling framework |
| **Images** | Extracted with AI descriptions | Handled by Docling |
| **Shapes/Comments** | Custom extraction + annotation | Not implemented (Docling limitation) |
| **Chunking** | LangChain text splitter | Table-based (Excel) + Semantic (text) |
| **Headers** | None | LLM-based detection |
| **Large Tables** | Memory issues | Chunked extraction |
| **Execution** | Single Python env | Isolated uv venv |
| **Data Transfer** | In-memory | Pickle files + XCom |
| **Keyword/Entity** | KeyBERT + NER | Not implemented |
| **MinIO Images** | Base64 → S3 upload | Handled by Docling |
| **Retry Logic** | Manual | Airflow task retries |
| **Monitoring** | Application logs | Airflow UI + logs |

---

## Migration Considerations

### Features Lost from Legacy

1. **Shape Extraction**: Text boxes, connectors not supported
2. **Comment Extraction**: Excel comments not extracted
3. **Cell Annotations**: Shape/comment context not embedded in cells
4. **Keyword/Entity Extraction**: KeyBERT and NER not integrated
5. **Image Storage**: Images not separately stored in MinIO
6. **Image Descriptions**: AI-generated descriptions not created

### Features Gained

1. **Multi-Format Support**: Word, PDF, PowerPoint processing
2. **Large Table Handling**: Chunked extraction for >1000-row tables
3. **LLM Header Detection**: Intelligent header row identification
4. **Semantic Chunking**: Context-aware text chunking
5. **Workflow Orchestration**: Airflow monitoring and management
6. **Environment Isolation**: Dependency conflicts eliminated
7. **Scalability**: Distributed task execution possible

### Integration Points

- **Input**: Same (document_id, object_key, content_type)
- **Output**: Milvus vector storage (compatible schema)
- **Metadata**: Different structure (no keyword_text, shape, comment chunks)

---

## File References

| File | Purpose | Location |
|------|---------|----------|
| `document_extraction_dag.py` | Main DAG definition | airflow/dags/ |
| `parse_tasks.py` | Parse task implementations | airflow/dags/tasks/ |
| `processing_tasks.py` | Processing task implementations | airflow/dags/tasks/ |
| `validate_event.py` | Event validation operator | airflow/plugins/operators/ |
| `fetch_document.py` | Document fetch operator | airflow/plugins/operators/ |

---

## Related Documentation

- [Old Excel Ingestion Pipeline](./old_excel_ingestion_pipeline.md)
- [Architecture Overview](./architecture.md)
- [Large Table Parsing Strategy](./large-table-parsing-strategy.md)

---

**Last Updated**: 2024-12-30
**Pipeline Version**: v2 (Docling-based, Airflow-orchestrated)