Databricks Lineage Scanner

WITH filtered_column_lineage AS (
  SELECT
    account_id,
    metastore_id,
    workspace_id,
    entity_type,
    entity_id,
    entity_run_id,
    source_table_full_name,
    source_table_catalog,
    source_table_schema,
    source_table_name,
    source_path,
    source_type,
    lower(source_column_name) AS source_column_name,
    target_table_full_name,
    target_table_catalog,
    target_table_schema,
    target_table_name,
    target_path,
    target_type,
    lower(target_column_name) AS target_column_name,
    event_time
  FROM system.access.column_lineage l
  WHERE source_column_name IS not null
    AND target_column_name IS not null
    AND event_time > current_date() - INTERVAL 7 DAYS
  QUALIFY
    (source_table_name is null OR target_table_name is null OR
    (row_number() OVER (PARTITION BY source_table_full_name, target_table_full_name, lower(source_column_name), lower(target_column_name) ORDER BY event_time DESC) = 1))
)
SELECT lin.*,
  dense_rank() OVER (ORDER BY lin.source_table_full_name, lin.target_table_full_name, lin.entity_id, lin.entity_run_id, lin.event_time DESC) as unique_transformation_id,
  src_tab.table_name as source_table_name_from_data_dict,
  lower(src_col.column_name) as source_column_name_from_data_dict,
  trg_tab.table_name as target_table_name_from_data_dict,
  lower(trg_col.column_name) as target_column_name_from_data_dict
FROM filtered_column_lineage lin
LEFT JOIN system.information_schema.tables src_tab ON
  src_tab.table_catalog = lin.source_table_catalog AND
  src_tab.table_schema = lin.source_table_schema AND
  src_tab.table_name = lin.source_table_name
LEFT JOIN system.information_schema.columns src_col ON
  src_col.table_catalog = lin.source_table_catalog AND
  src_col.table_schema = lin.source_table_schema AND
  src_col.table_name = lin.source_table_name AND
  lower(src_col.column_name) = lin.source_column_name
LEFT JOIN system.information_schema.tables trg_tab ON
  trg_tab.table_catalog = lin.target_table_catalog AND
  trg_tab.table_schema = lin.target_table_schema AND
  trg_tab.table_name = lin.target_table_name
LEFT JOIN system.information_schema.columns trg_col ON
  trg_col.table_catalog = lin.target_table_catalog AND
  trg_col.table_schema = lin.target_table_schema AND
  trg_col.table_name = lin.target_table_name AND
  lower(trg_col.column_name) = lin.target_column_name;