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