Fixes 32021: read native Unity Catalog lineage and its SQL in one streamed query - #33141
Fixes 32021: read native Unity Catalog lineage and its SQL in one streamed query#33141ulixius9 wants to merge 4 commits into
Conversation
Native lineage edges could appear without the SQL that produced them. Attach the statement by joining system.access.table_lineage to system.query.history, and read the column mappings from system.access.column_lineage in the same query instead of a second result set. Drive the emission from the lineage rows rather than from the catalog: the system tables name every table an edge can end at, so those are the tables looked up. Listing every table of every schema of every catalog spent a paginated request per hundred tables in the service, nearly all of them for tables no edge mentions, and an upstream was re-resolved once per edge naming it. Both endpoints now resolve through one LRU cache of hits and definite misses, and an external table is only resolved once its storage is known to be an ingested container. system.query.history is probed once. When it cannot be read, the statement columns and the join are left out of the query entirely, so a missing grant costs the SQL text rather than all of the lineage. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Read the rows ordered by target and emit a target's edges as soon as the next target appears, so what is held is one target rather than every edge in the catalog. Drops table_lineage_map, column_lineage_map, path_lineage_map and the per-edge SQL map; the grouping that remains is one target wide. Rows are still grouped rather than emitted one by one because Databricks reports an edge twice when it names a side by table on one row and by path on another, and addLineage replaces an edge's details instead of merging them. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
github-actions
Bot
added
Ingestion
safe to test
![gitar-bot[bot]](https://avatars.githubusercontent.com/in/827041?s=60&v=4){kind=small}
| UNITY_CATALOG_QUERY_HISTORY_PROBE = textwrap.dedent( | ||
| """ | ||
| SELECT | ||
| source_table_full_name, | ||
| source_path, | ||
| target_table_full_name, | ||
| target_path | ||
| FROM system.access.table_lineage | ||
| WHERE event_time >= current_date() - INTERVAL {query_log_duration} DAYS | ||
| AND (source_table_full_name IS NOT NULL OR source_path IS NOT NULL) | ||
| AND (target_table_full_name IS NOT NULL OR target_path IS NOT NULL) | ||
| GROUP BY source_table_full_name, source_path, target_table_full_name, target_path | ||
| SELECT lineage.statement_id, history.statement_text | ||
| FROM system.access.table_lineage lineage | ||
| JOIN system.query.history history | ||
| ON lineage.statement_id = history.statement_id | ||
| AND lineage.workspace_id = history.workspace_id | ||
| WHERE 1=0 | ||
| """ | ||
| ) |
There was a problem hiding this comment.
💡 Edge Case: Probe does not validate history.start_time used by main query
The probe query (UNITY_CATALOG_QUERY_HISTORY_PROBE) exercises statement_id, workspace_id and statement_text on system.query.history, but the actual join in UNITY_CATALOG_QUERY_HISTORY_JOIN additionally references history.start_time. The stated purpose of the probe is to guarantee the main query won't fail on the history join and take the lineage down with it; since start_time is not covered, a schema mismatch on that column would pass the probe yet fail the whole native-lineage query, losing all lineage rather than just the SQL text. Consider referencing start_time (and statement columns) in the probe so its guarantee matches what the main query actually needs.
Was this helpful? React with 👍 / 👎
|
✅ Playwright Results — workflow succeededValidated commit ✅ 4500 passed · ❌ 0 failed · 🟡 5 flaky · ⏭️ 1 skipped · 🧰 0 lifecycle flaky PerformanceBlocking targets: ✅ met · Optimization targets: 🟡 in progress Shard-job maxima below are not the full workflow wall time; the linked run includes build, fixture, planning, and reporting. 🕒 Full workflow signal wall (to summary) 1h 7m 50s ⏱️ Max setup 4m 50s · max shard execution 23m 40s · max shard-job elapsed before upload 26m 38s · reporting 22s 🌐 216.93 requests/attempt · 2.31 app boots/UI scenario · 35.85% common-shard skew Optimization targets still in progress:
🟡 5 flaky test(s) (passed on retry)
How to debug locally# Download playwright-test-results-<shard> artifact and unzip
npx playwright show-trace path/to/trace.zip # view trace |
Unity Catalog normalises its identifiers, so the three parts the system tables report are the three parts the table was ingested under and the Elasticsearch search fqn.build runs per table returns that same string. Verified against a live workspace: all 444 edges still resolve. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Code Review 👍 Approved with suggestions 1 resolved / 2 findingsOptimizes Unity Catalog native lineage ingestion from two system-table queries plus paginated lookups to a single streamed query that joins lineage with query history and SQL in one pass, eliminating the unbounded map caching and expensive per-edge resolution. Comprehensive test coverage (99% of changed lines, 84 tests passing) validates edge cases including path-based sources, merged column mappings, and graceful degradation when query history is unavailable. Consider extending the 💡 Edge Case: Probe does not validate history.start_time used by main query📄 ingestion/src/metadata/ingestion/source/database/unitycatalog/queries.py:55-64 📄 ingestion/src/metadata/ingestion/source/database/unitycatalog/queries.py:128 The probe query ( ✅ 1 resolved✅ Quality: License header removed from self-referencing lineage test
🤖 Prompt for agentsOptionsDisplay: compact → Counting what did not apply, without listing it. Comment with these commands to change the behavior for this request:
Was this helpful? React with 👍 / 👎 | Powered by Gitar — free for open source |
Describe your changes:
Fixes #32021
Native Unity Catalog lineage edges carry no SQL even when the statement that wrote
them is available. I attached it by joining
system.access.table_lineagetosystem.query.historyinside the query that already reads the lineage, and whilethere I removed the two things that made that lookup expensive: the walk over every
table in the service, and the in-memory copy of every edge in the catalog.
Before: two system-table queries cached into unbounded maps, then one paginated
list_all_entitiesrequest per hundred tables in the service (nearly all of them fortables no edge mentions) plus one uncached
get_by_nameper edge. Now: one query fortable edges, column mappings and SQL together, streamed in target order, with the
tables an edge actually names resolved once each behind an LRU cache.
This supersedes #33002, which fixes the same issue by batching a second
table_lineage ⋈ query.historyquery per 100 edges. Measured on the same liveworkspace and window, that costs 53.2s of warehouse time against 7.1s here
(numbers below). This branch is cut from current
main, which #33002's branchpredates by ~150 lines in this file (path lineage, self-reference dedupe).
Type of change:
High-level design:
One query instead of two plus one per batch.
unity_catalog_native_lineage_query(query_log_duration, include_query_history)aggregates each edge's column mappings server-side
(
to_json(collect_set(struct(...)))) and left-joinssystem.query.historyon theedge's latest
statement_id/workspace_id(max_by(struct(...), event_time)).Column mappings join on
<=>, not=:source_pathis NULL on every edge reportedby table name, and
=on NULL never matches, so a plain join would silently dropthose edges' mappings.
Graceful degradation.
statement_idis only populated for statements run on aSQL warehouse and
system.query.historyis separately granted, so a query namingthem would fail as a whole and take the lineage with it. A single
WHERE 1=0probedecides once per run; on failure the statement columns and the join are left out
entirely and lineage is ingested without SQL.
Streaming, one target at a time. The query is
ORDER BY target, so a target'sedges are complete as soon as the next target appears.
_stream_native_lineagebuffers only the target in hand and hands it to
_process_target_lineage.Why the rows are grouped at all (rather than emitted one by one, Snowflake
ACCESS_HISTORYstyle): Databricks reports one edge twice when it names a side bytable on one row and by path on another, and
LineageRepository.addLineagereplacesan edge's
lineageDetailsinstead of merging them — a second request for a pairalready sent would drop the column mappings of the first. That replace-not-merge
behaviour was confirmed live: running
main's connector over the same servicestripped
sqlQueryfrom all 444 edges, and re-running this one put it back.Grouping per target covers every duplication observed so far, since both rows share
the target. A target reported both by name and by path would still emit twice; on
the live dataset 0 of 239 targets had more than one raw key. Closing that case
too would mean pushing path→table resolution (scheme aliases, trailing slashes) into
SQL, moving the
path_utilsbehaviours out of unit tests and into something onlyverifiable on a live workspace. Rejected on that basis.
Iteration is driven by the lineage rows, not the catalog: targets come from the
system tables and external-location edges from
system.information_schema.tables. Nolist_all_entitiescall remains. The FQN of a table is built rather than searchedfor — Unity Catalog normalises identifiers, so
fqn.build's Elasticsearch round tripper table returns the same string
fqn._buildproduces (same approach assnowflake/lineage.py); all 444 live edges still resolve. Filter patterns apply tothe
catalog.schema.tablenames the system tables report, and a table that is both alineage target and an external table is reported once in the run summary.
Backward compatibility: no schema, API or config change. Same edges and column
mappings as before, plus SQL.
queryLogDurationkeeps its meaning; history reachesone day further back so a statement that ran just before the oldest lineage day is
still found.
Tests:
Use cases covered
Information panel (the issue).
system.query.historyis not readable.rows' column mappings merged.
Unit tests
ingestion/tests/unit/topology/database/test_unity_catalog_lineage.py(rewrittento assert on emitted
AddLineageRequests rather than internal maps; 57 tests)ingestion/tests/unit/topology/database/test_unitycatalog_self_referencing_lineage.py(retargeted to the method that now owns the rule; 4 tests)
lineage.py90%,
queries.py100%.61 passedin those two files;140 passedacross all collectibletest_unity*files.Backend integration tests
Ingestion integration tests
metastore is the SQL of the query itself, which no fixture reproduces. Validated
manually against a live workspace instead; see below.
Playwright (UI) tests
locales/en-US/Database/UnityCatalog.md).Manual testing performed
Live Databricks workspace, catalog
demo, schemaom_lineage_acceptance(205 batch targets +
my_source/my_target), 7-day lookback.uc_lineage_live_test): 245 tables, 0 errors.Reading native lineage from system tables (lookback: 7 days, SQL text: yes)→emitted 444 edges over 239 targets,444 processed, 0 errors, 0 warnings, 100% success.
/v1/lineage/getLineageEdge/...: every edge has anon-empty
sqlQuerybyte-identical to the statement insystem.query.history,and its column-mapping count matches the warehouse's mappings for that pair exactly.
Example (
my_source → my_target):INSERT INTO demo.om_lineage_acceptance.my_target SELECT id, amount * 2 AS total FROM demo.om_lineage_acceptance.my_sourcewith2 column mappings and
source: QueryLineage.main's two queries over the same window: 444 edges and 683column mappings on both sides, 0 missing, 0 extra — the single query loses nothing.
one-target buffer never sees a target twice.
include_query_history=Falsevariant runs on the warehouse andreturns the same 444 rows with
statement_textNULL and column mappings intact.sqlQuerypreserved.Warehouse cost, same window, warm warehouse:
main(no SQL text)table_lineage+column_lineageSource-side wall clock is a wash at this size — 10.2s here (245
get_by_name, 1.5sprobe, 7.1s query) against
main's 10.6s (450get_by_name+ catalog pagination,7.5s+ of queries) — because this service has 245 tables and 239 of them are lineage
targets, the best possible case for the catalog walk. The saving scales with
tables-in-service ÷ tables-in-lineage. End to end the run is 39s against
main's 23s;that difference is the sink writing the statement text onto 444 edges, i.e. the
feature, not the restructure.
Static checks:
ruff check,ruff format --check,check_ruff_suppressions.py --check(the 5 obsolete G004 baseline entries this change retires are pruned), and
basedpyright --baselinemode=discard— its tworeportOptionalIterablehits onfor row in rowsalso occur onmain's copy of this file in the same environment(3 there, 2 here), so they are the local partial venv, not a regression.
UI screen recording / screenshots:
Not applicable.
Checklist:
I have read the CONTRIBUTING document.
My PR title is
Fixes <issue-number>: <short explanation>My PR is linked to a GitHub issue via
Fixes #<issue-number>above.I have commented on my code, particularly in hard-to-understand areas.
For JSON Schema changes: not applicable, no schema change.
For UI changes: not applicable.
I have added tests (unit / integration / Playwright as applicable) and listed them above.
I have added tests around the new logic.
For connector/ingestion changes: I updated the documentation.
🤖 Generated with Claude Code