实例sql:
WITH RECURSIVE
table_lineage AS (
SELECT DISTINCT
SUBSTRING_INDEX(field_id, '.', 2) AS downstream_tbl,
SUBSTRING_INDEX(depend_field_id, '.', 2) AS upstream_tbl
FROM dgp_mysql_redan.redan.dgp_dataset_field_lineage
),
upstream_closure AS (
SELECT downstream_tbl AS table_name,
upstream_tbl AS depend_table,
CAST(1 AS BIGINT) AS hop
FROM table_lineage
UNION ALL
SELECT u.table_name,
tl.upstream_tbl,
u.hop + 1
FROM upstream_closure u
JOIN table_lineage tl ON u.depend_table = tl.downstream_tbl
),
downstream_closure AS (
SELECT upstream_tbl AS table_name,
downstream_tbl AS depend_table,
CAST(1 AS BIGINT) AS hop
FROM table_lineage
UNION ALL
SELECT d.table_name,
tl.downstream_tbl,
d.hop + 1
FROM downstream_closure d
JOIN table_lineage tl ON d.depend_table = tl.upstream_tbl
)
SELECT table_name, depend_table, hop, 'up' AS direction
FROM upstream_closure
UNION ALL
SELECT table_name, depend_table, hop, 'down' AS direction
FROM downstream_closure
ORDER BY table_name, direction, hop, depend_table;