┌──────────────────────────────────────────────────────────┐
│  1. LineageSource                                        │
│  source/database/lineage_source.py                       │
│                                                          │
│  Fetches query logs filtered for lineage-relevant types: │
│  CREATE TABLE AS SELECT, INSERT...SELECT, MERGE,         │
│  UPDATE...FROM, view definitions, stored procedures      │
│                                                          │
│  Yields: Either[AddLineageRequest]                       │
└──────────────────────┬───────────────────────────────────┘
                       │
                       ▼
┌──────────────────────────────────────────────────────────┐
│  MetadataRestSink                                        │
│  sink/metadata_rest.py                                   │
│                                                          │
│  PUT /api/v1/lineage                                     │
│  (merges column lineage with existing edges)             │
└──────────────────────────────────────────────────────────┘

# 1. SOURCE fetches and chunks query logs
LineageSource._iter()
    └── yield_query_lineage()
        ├── Fetch query logs from database (or file)
        ├── Chunk into batches of 200
        └── Process chunks in parallel threads

# 2. PROCESSOR parses each query (runs in thread pool)
    └── query_lineage_processor(query)           # lineage_processors.py
        └── get_lineage_by_query(query, ...)     # lineage/sql_lineage.py
            ├── LineageParser(query, dialect)     # lineage/parser.py
            │   ├── Try SqlGlot (30s timeout)
            │   ├── Fallback SqlFluff (30s timeout)
            │   └── Fallback SqlParse
            │
            ├── Extract: source_tables, target_tables, intermediate_tables
            ├── Extract: column_lineage (column-to-column mappings)
            │
            └── For each (source, target) pair:
                ├── search_table_entities(source)  # ES + API lookup
                ├── search_table_entities(target)
                ├── Build column lineage mappings
                └── yield AddLineageRequest

# 3. VIEW LINEAGE (separate pass)
    └── yield_view_lineage()
        └── view_lineage_processor(view)
            └── get_view_lineage(view_definition, ...)
                ├── Parse CREATE VIEW ... AS SELECT ...
                ├── Extract source tables from SELECT
                └── yield AddLineageRequest (source tables → view)

parser = LineageParser(query="INSERT INTO target SELECT * FROM source", dialect=Dialect.ANSI)

parser.source_tables         # ["source"]       — tables being read
parser.target_tables         # ["target"]       — tables being written
parser.intermediate_tables   # []               — staging/temp tables
parser.column_lineage        # [(source.col1, target.col1), ...]
parser.table_aliases         # {"s": "source"}

source ──→ target

CREATE TEMP TABLE staging AS SELECT * FROM source;
INSERT INTO target SELECT * FROM staging;

source ──→ staging ──→ target

column_lineage = [
    ColumnLineage(
        fromColumns=["service.db.schema.source.col1"],
        toColumn="service.db.schema.target.col1"
    ),
    ColumnLineage(
        fromColumns=["service.db.schema.source.col2"],
        toColumn="service.db.schema.target.col2"
    ),
]

search_table_entities(service_name, database, schema, table)
  1. Search Elasticsearch (fast) for table FQN
  2. If not found, try API search (reliable)
  3. If schema not found, retry without schema qualifier
  4. Cache results (1,000-entry LRU cache)

# Query references tables from multiple services:
INSERT INTO warehouse.analytics.summary
SELECT * FROM production.sales.orders

# Resolution searches across multiple services:
search_table_entities(
    service_names=["warehouse_service", "production_service"],
    ...
)

view_lineage_processor(view_entity)
  1. Get view definition SQL
  2. Parse with LineageParser
  3. Extract source tables from the SELECT
  4. Create lineage: source tables → view
  5. Override existing view lineage (views can be redefined)

Query logs → Chunk into batches of 200
                    │
     ┌──────────────┼──────────────┐
     ▼              ▼              ▼
  Thread 1       Thread 2       Thread 3
  (chunk 1)      (chunk 2)      (chunk 3)
     │              │              │
     └──────────────┼──────────────┘
                    ▼
            Result queue → yield AddLineageRequest

AddLineageRequest(
    edge=EntitiesEdge(
        fromEntity=EntityReference(id=source_id, type="table"),
        toEntity=EntityReference(id=target_id, type="table"),
        lineageDetails=LineageDetails(
            sqlQuery="INSERT INTO target SELECT ...",  # Masked
            source=LineageSource.QueryLineage,
            columnsLineage=[
                ColumnLineage(
                    fromColumns=["db.schema.source.col1"],
                    toColumn="db.schema.target.col1"
                ),
            ],
            pipeline=EntityReference(...)  # Optional: stored procedure
        )
    )
)

add_lineage(request, check_patch=False)
  1. Check if edge already exists
  2. If exists: merge column lineage (union of old + new columns)
  3. If new: create edge
  4. Cache the result

source:
  type: postgres-lineage          # Database-specific lineage source
  serviceName: my_postgres
  sourceConfig:
    config:
      type: DatabaseLineage
      queryLogDuration: 1           # Days to look back
      resultLimit: 100000           # Max queries to fetch
      parsingTimeoutLimit: 30       # Seconds per query
      processCrossDatabaseLineage: false
      crossDatabaseServiceNames: []
      threads: 4                    # Parallel processing threads
      queryLogFilePath: ""          # Optional: CSV file path