Fixes 31278: Extract lineage from SQL-defined Databricks DLT pipelines - #31654
Merged
Conversation
SQL DLT transformations parsed to nothing because kafka_parser only understood the Python DLT API, so those pipelines produced no lineage. Parse CREATE [OR REFRESH] MATERIALIZED VIEW / STREAMING TABLE through the existing LineageParser, resolve qualified upstreams against their own catalog and schema, read `file` libraries, and recurse into workspace subdirectories.
…/databricks-sql-dlt-lineage
Adding SQL support as a fallback inside extract_dlt_table_dependencies left an if/elif chain that dispatched on failure rather than language, against the registry convention in docs/design-patterns.md. Introduce dlt_parsers.py with a DltSourceParser protocol and registry. Python and SQL each declare handles/extract and an explicit priority, so a Python notebook embedding SQL strings is not misread and a new language needs no dispatch edit. Move the parse-result dataclasses to models.py and reduce kafka_parser.py to Kafka configuration. Fix three Databricks SQL shapes found while probing real constructs: APPLY CHANGES INTO lost its source, the LIVE. prefix was resolved as a schema, and table-valued functions were returned as tables. Also use the previously dead get_pipeline_libraries helper instead of a duplicated inline copy.
Contributor
✅ PR checks passedThe linked issue has a description and all required Shipping project fields set. Thanks! |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Pull request overview
This PR extends the Databricks DLT (Delta Live Tables) pipeline ingestion to extract lineage from SQL-authored DLT pipelines (e.g., transformations defined as .sql files) while refactoring the existing Kafka/DLT parsing code into clearer, more modular components within the ingestion framework.
Changes:
- Added a new DLT parser registry with dedicated SQL and Python parsers, enabling SQL-defined DLT dataset dependency extraction via the shared
LineageParser. - Refactored Databricks pipeline lineage extraction to correctly resolve qualified upstream table references and to discover pipeline sources declared via
fileand recursivegloblibrary entries. - Moved connector-specific dataclasses (
KafkaSourceConfig,DLTTableDependency) intomodels.pyand updated unit tests, including a new SQL lineage test suite.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| ingestion/src/metadata/ingestion/source/pipeline/databrickspipeline/dlt_parsers.py | Introduces parser registry + SQL DLT parsing (via LineageParser) and migrates the Python DLT parsing behind the same interface. |
| ingestion/src/metadata/ingestion/source/pipeline/databrickspipeline/kafka_parser.py | Narrows to Kafka source parsing + consolidates pipeline library path extraction (notebook/file/glob). |
| ingestion/src/metadata/ingestion/source/pipeline/databrickspipeline/metadata.py | Uses new DLT dependency extraction, fixes dataset qualification for qualified upstreams, and expands workspace directory traversal recursively for glob sources. |
| ingestion/src/metadata/ingestion/source/pipeline/databrickspipeline/models.py | Houses Kafka/DLT dataclasses alongside existing pipeline models for consistent connector structure. |
| ingestion/tests/unit/topology/pipeline/test_databricks_kafka_parser.py | Updates tests to use the new parser module structure (e.g., PythonDltParser). |
| ingestion/tests/unit/topology/pipeline/test_databricks_dlt_sql_lineage.py | Adds a focused unit test suite validating SQL DLT dependency extraction + end-to-end lineage edge emission. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
![gitar-bot[bot]](https://avatars.githubusercontent.com/in/827041?s=60&v=4){kind=small}
Glob includes were only normalised for the `/**` form, so `*.sql` and `**/*.sql` kept their wildcard and were exported as literal paths, missing every transformation. Reduce any include to the directory before its first wildcard, always ending in a slash. Stripping the LIVE. prefix could collapse two references onto one dataset, so de-duplicate upstreams while preserving order. Drop table-valued functions only when the statement invokes them, so a dataset named `range` still resolves. Strip whitespace and stray dots when qualifying a name. Remove the dead DLT_TABLE_PATTERN and the unused get_pipeline_libraries client parameter.
_qualify_dlt_table_name returned a bare (catalog, schema, table) tuple, so callers had to know what each position meant. Return a DLTTableReference instead, alongside the other connector models. Replace the parser registry, the Protocol priority field and the per-call sort with an explicit ordered DLT_PARSERS list. Two implementations did not justify the indirection, and the previous approach also misused enum_register, which everywhere else keys functions by an enum value rather than classes by an ad-hoc string. Order is now stated where the list is defined, and a test asserts it rather than assuming it. Bound the entity caches with LRUCache, which CLAUDE.md requires and the sibling Databricks ownership source already uses. Typing the cache surfaced that fqn.build can return None and the old code cached under a None key, so collapse the two duplicated cache blocks into a single _lookup_table helper that guards it once. Build the source through the real create() in tests instead of assigning its internals. The hand-assigned caches had already drifted from production once without any test failing. Cache bounds, eviction, miss caching and the None guard are now asserted against the real object.
github-actions
Bot
added
Ingestion
safe to test
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 7 out of 7 changed files in this pull request and generated 1 comment.
Suppressed comments (1)
ingestion/src/metadata/ingestion/source/pipeline/databrickspipeline/dlt_parsers.py:92
SqlDltParser.handles()relies on patterns that matchCREATE .../APPLY CHANGES INTOanywhere in the source. This can misclassify non-SQL sources (e.g., a Python notebook containing those keywords inside aspark.sql("...")string) as SQL and attempt lineage extraction. Anchoring these patterns to the start of a statement (allowing leading SQL comments) reduces false positives without affecting real SQL transformations.
SQL_DLT_CREATE_PATTERN = re.compile(
r"\bCREATE\s+(?:OR\s+REFRESH\s+)?(?:TEMPORARY\s+|PRIVATE\s+)?"
r"(?:MATERIALIZED\s+VIEW|STREAMING\s+(?:LIVE\s+)?TABLE|LIVE\s+TABLE)\b",
re.IGNORECASE,
)
SQL_DLT_APPLY_CHANGES_PATTERN = re.compile(r"\bAPPLY\s+CHANGES\s+INTO\b", re.IGNORECASE)
glob_base_directory reduced an include to its base directory and the caller then expanded that whole tree, so a selective include such as /transformations/*.sql pulled in unrelated files and attributed their lineage to the pipeline. That is worse than the wildcard bug it replaced, because the edges are wrong rather than absent. Model a library entry as DLTLibrarySource, carrying the directory to list and the pattern its contents must match. Recurse only for **, and check every candidate with glob_matches, where * stays inside a path segment and ** spans them. fnmatch is unsuitable because its * also crosses /. Tests now assert which files an include selects rather than which directory it reduces to, which is what the earlier assertion missed.
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 7 out of 7 changed files in this pull request and generated 1 comment.
Suppressed comments (2)
ingestion/src/metadata/ingestion/source/pipeline/databrickspipeline/kafka_parser.py:211
glob_matches()supports?wildcards, butglob_base_directory()only treats*as a wildcard. A pattern like/tx/file?.sqlwould be treated as a concrete file path (so no directory expansion/filtering), which is inconsistent with the matcher and the docstring (“first wildcard onward”). Consider treating?as a wildcard too when computing the base directory.
if "*" not in include:
return include
base = include.split("*", 1)[0]
ingestion/src/metadata/ingestion/source/pipeline/databrickspipeline/dlt_parsers.py:522
- PR description mentions parsers carrying an explicit
priority, but dispatch is currently order-dependent via the hard-codedDLT_PARSERSlist. That’s fine functionally, but it contradicts the stated design and makes future additions rely on list ordering (which can be easy to miss in review). Either implement an explicit priority+sort inextract_dlt_table_dependencies, or update the design/description to match the order-based dispatch.
# Checked in order. Python comes first because a `@dlt.` decorator is unambiguous,
# while the SQL keywords can also appear inside a string literal in a Python
# notebook that calls spark.sql(...).
DLT_PARSERS: List[Type[DltSourceParser]] = [PythonDltParser, SqlDltParser] # noqa: UP006
get_pipeline_libraries returns DLTLibrarySource, but the spec.configuration and spec.development source_path fallback still appended a raw string, so iterating the list hit AttributeError on library.is_directory and lineage died for pipelines that use those fields. glob_base_directory only looked for *, while glob_matches also honours ?, so an include such as /transformations/file_?.sql was treated as a concrete path and sent to the export API literally. Both now share GLOB_WILDCARDS so they cannot disagree again.
…neage test The test stubbed _table_lookup_cache with a plain dict, which satisfied main's implementation but not this branch's LRUCache. The lookup raised, the exception was swallowed, and the two tests expecting an edge failed once main was merged in.
…ging The libraries key can be present and null, and testing for the key rather than reading it made len() raise before the source paths were collected, which cost the pipeline its lineage. Per-path discovery also logged at INFO, which buries the run summary now that expansion is recursive.
Code Review ✅ Approved 5 resolved / 5 findingsAdds SQL-defined Databricks DLT lineage extraction by introducing a parser registry that handles both Python and SQL sources, with Databricks-specific normalizations for ✅ 5 resolved✅ Edge Case: Real tables named like TVFs are dropped as upstreams
✅ Quality: Unused
|
| Compact |
|
Was this helpful? React with 👍 / 👎 | Powered by Gitar — free for open source
|
mohittilala
enabled auto-merge
September 1, 2026 00:39
harshsoni2024
approved these changes
Sep 1, 2026
shrabantipaul-collate
pushed a commit
that referenced
this pull request
Sep 1, 2026
#31654) * Fixes 31278: Extract lineage from SQL-defined Databricks DLT pipelines SQL DLT transformations parsed to nothing because kafka_parser only understood the Python DLT API, so those pipelines produced no lineage. Parse CREATE [OR REFRESH] MATERIALIZED VIEW / STREAMING TABLE through the existing LineageParser, resolve qualified upstreams against their own catalog and schema, read `file` libraries, and recurse into workspace subdirectories. * refactor(databricks): dispatch DLT parsing through a language registry Adding SQL support as a fallback inside extract_dlt_table_dependencies left an if/elif chain that dispatched on failure rather than language, against the registry convention in docs/design-patterns.md. Introduce dlt_parsers.py with a DltSourceParser protocol and registry. Python and SQL each declare handles/extract and an explicit priority, so a Python notebook embedding SQL strings is not misread and a new language needs no dispatch edit. Move the parse-result dataclasses to models.py and reduce kafka_parser.py to Kafka configuration. Fix three Databricks SQL shapes found while probing real constructs: APPLY CHANGES INTO lost its source, the LIVE. prefix was resolved as a schema, and table-valued functions were returned as tables. Also use the previously dead get_pipeline_libraries helper instead of a duplicated inline copy. * fix(databricks): address PR review on DLT source discovery Glob includes were only normalised for the `/**` form, so `*.sql` and `**/*.sql` kept their wildcard and were exported as literal paths, missing every transformation. Reduce any include to the directory before its first wildcard, always ending in a slash. Stripping the LIVE. prefix could collapse two references onto one dataset, so de-duplicate upstreams while preserving order. Drop table-valued functions only when the statement invokes them, so a dataset named `range` still resolves. Strip whitespace and stray dots when qualifying a name. Remove the dead DLT_TABLE_PATTERN and the unused get_pipeline_libraries client parameter. * refactor(databricks): model DLT references and simplify parser dispatch _qualify_dlt_table_name returned a bare (catalog, schema, table) tuple, so callers had to know what each position meant. Return a DLTTableReference instead, alongside the other connector models. Replace the parser registry, the Protocol priority field and the per-call sort with an explicit ordered DLT_PARSERS list. Two implementations did not justify the indirection, and the previous approach also misused enum_register, which everywhere else keys functions by an enum value rather than classes by an ad-hoc string. Order is now stated where the list is defined, and a test asserts it rather than assuming it. Bound the entity caches with LRUCache, which CLAUDE.md requires and the sibling Databricks ownership source already uses. Typing the cache surfaced that fqn.build can return None and the old code cached under a None key, so collapse the two duplicated cache blocks into a single _lookup_table helper that guards it once. Build the source through the real create() in tests instead of assigning its internals. The hand-assigned caches had already drifted from production once without any test failing. Cache bounds, eviction, miss caching and the None guard are now asserted against the real object. * fix(databricks): select only the files a DLT glob actually matches glob_base_directory reduced an include to its base directory and the caller then expanded that whole tree, so a selective include such as /transformations/*.sql pulled in unrelated files and attributed their lineage to the pipeline. That is worse than the wildcard bug it replaced, because the edges are wrong rather than absent. Model a library entry as DLTLibrarySource, carrying the directory to list and the pattern its contents must match. Recurse only for **, and check every candidate with glob_matches, where * stays inside a path segment and ** spans them. fnmatch is unsuitable because its * also crosses /. Tests now assert which files an include selects rather than which directory it reduces to, which is what the earlier assertion missed. * fix(databricks): keep library entries modelled and treat ? as a wildcard get_pipeline_libraries returns DLTLibrarySource, but the spec.configuration and spec.development source_path fallback still appended a raw string, so iterating the list hit AttributeError on library.is_directory and lineage died for pipelines that use those fields. glob_base_directory only looked for *, while glob_matches also honours ?, so an include such as /transformations/file_?.sql was treated as a concrete path and sent to the export API literally. Both now share GLOB_WILDCARDS so they cannot disagree again. * fix(databricks): expand a pattern-less library directory in full A configuration or development source_path produces a library with no pattern, and treating that as non-recursive skipped every nested transformation. No pattern means no filter, so the whole tree is in scope. An include that names a bare directory was also stored as its own pattern, which then matched nothing and dropped the directory entirely. Only attach a pattern when the include actually contains a wildcard. Cover every spec.libraries shape end to end so the reduction and the matching cannot drift apart again. * fix(databricks): state directory-ness instead of inferring it from the path is_directory was read off a trailing slash, so a source_path that Databricks spells without one was exported as a file and its transformations never listed. The producer knows what each entry is, so it now says so: notebook and file entries are files, a wildcard include is always a directory, and a source_path is normalised to a directory. is_recursive only looked for **, so a wildcard in a directory segment such as /tx/2024_?/file.sql reduced to /tx/ and then refused to descend, selecting nothing. Decide from the part of the pattern below the base directory, where either ** or a / means traversal is required. * fix(databricks): let the workspace settle an ambiguous glob include A glob include with neither a wildcard nor a trailing slash may name a file or a directory, and the string cannot tell them apart: assuming a file leaves a directory unlisted, assuming a directory swallows a concrete .sql. Model the gap instead, with is_directory None meaning the spec did not say, and resolve it by listing the path. Anything that lists is a directory, anything that does not is read as a single source. * fix(databricks): read the legacy and private DLT create forms CREATE LIVE TABLE, CREATE STREAMING LIVE TABLE and the PRIVATE modifier are spellings DLT still accepts but the query parser cannot read. It falls back to a bare command and reports no tables, so the dataset was dropped with neither an error nor a warning. Rewrite them to the modern equivalent before parsing, anchored to the CREATE clause so a dataset whose own name contains "live" is left alone. * refactor(databricks): drop the glob machinery Databricks cannot produce The pipelines API accepts only an exact file, a directory, or a directory with a trailing **, and rejects every other wildcard form outright. The pattern matcher, the wildcard set and the recursive flag were therefore unreachable, along with the tests asserting on them. A library is now just a file or a directory taken in full. * fix(databricks): describe library discovery as it behaves and lift the depth cap The get_pipeline_libraries docstring still promised pattern filtering that no longer exists, and the expansion cap of 5 silently dropped transformations nested deeper than that, which costs the lineage depending on them. Correct the docstring and raise the cap to 20. A directory that is not there costs nothing to look for. * fix(databricks): use the real lookup cache in the self-referencing lineage test The test stubbed _table_lookup_cache with a plain dict, which satisfied main's implementation but not this branch's LRUCache. The lookup raised, the exception was swallowed, and the two tests expecting an edge failed once main was merged in. * fix(databricks): read a null libraries value and quieten per-file logging The libraries key can be present and null, and testing for the key rather than reading it made len() raise before the source paths were collected, which cost the pipeline its lineage. Per-path discovery also logged at INFO, which buries the run summary now that expansion is recursive.
This was referenced Sep 1, 2026
Merged
Merged
Closed
mohittilala
added a commit
that referenced
this pull request
Sep 2, 2026
#31654) * Fixes 31278: Extract lineage from SQL-defined Databricks DLT pipelines SQL DLT transformations parsed to nothing because kafka_parser only understood the Python DLT API, so those pipelines produced no lineage. Parse CREATE [OR REFRESH] MATERIALIZED VIEW / STREAMING TABLE through the existing LineageParser, resolve qualified upstreams against their own catalog and schema, read `file` libraries, and recurse into workspace subdirectories. * refactor(databricks): dispatch DLT parsing through a language registry Adding SQL support as a fallback inside extract_dlt_table_dependencies left an if/elif chain that dispatched on failure rather than language, against the registry convention in docs/design-patterns.md. Introduce dlt_parsers.py with a DltSourceParser protocol and registry. Python and SQL each declare handles/extract and an explicit priority, so a Python notebook embedding SQL strings is not misread and a new language needs no dispatch edit. Move the parse-result dataclasses to models.py and reduce kafka_parser.py to Kafka configuration. Fix three Databricks SQL shapes found while probing real constructs: APPLY CHANGES INTO lost its source, the LIVE. prefix was resolved as a schema, and table-valued functions were returned as tables. Also use the previously dead get_pipeline_libraries helper instead of a duplicated inline copy. * fix(databricks): address PR review on DLT source discovery Glob includes were only normalised for the `/**` form, so `*.sql` and `**/*.sql` kept their wildcard and were exported as literal paths, missing every transformation. Reduce any include to the directory before its first wildcard, always ending in a slash. Stripping the LIVE. prefix could collapse two references onto one dataset, so de-duplicate upstreams while preserving order. Drop table-valued functions only when the statement invokes them, so a dataset named `range` still resolves. Strip whitespace and stray dots when qualifying a name. Remove the dead DLT_TABLE_PATTERN and the unused get_pipeline_libraries client parameter. * refactor(databricks): model DLT references and simplify parser dispatch _qualify_dlt_table_name returned a bare (catalog, schema, table) tuple, so callers had to know what each position meant. Return a DLTTableReference instead, alongside the other connector models. Replace the parser registry, the Protocol priority field and the per-call sort with an explicit ordered DLT_PARSERS list. Two implementations did not justify the indirection, and the previous approach also misused enum_register, which everywhere else keys functions by an enum value rather than classes by an ad-hoc string. Order is now stated where the list is defined, and a test asserts it rather than assuming it. Bound the entity caches with LRUCache, which CLAUDE.md requires and the sibling Databricks ownership source already uses. Typing the cache surfaced that fqn.build can return None and the old code cached under a None key, so collapse the two duplicated cache blocks into a single _lookup_table helper that guards it once. Build the source through the real create() in tests instead of assigning its internals. The hand-assigned caches had already drifted from production once without any test failing. Cache bounds, eviction, miss caching and the None guard are now asserted against the real object. * fix(databricks): select only the files a DLT glob actually matches glob_base_directory reduced an include to its base directory and the caller then expanded that whole tree, so a selective include such as /transformations/*.sql pulled in unrelated files and attributed their lineage to the pipeline. That is worse than the wildcard bug it replaced, because the edges are wrong rather than absent. Model a library entry as DLTLibrarySource, carrying the directory to list and the pattern its contents must match. Recurse only for **, and check every candidate with glob_matches, where * stays inside a path segment and ** spans them. fnmatch is unsuitable because its * also crosses /. Tests now assert which files an include selects rather than which directory it reduces to, which is what the earlier assertion missed. * fix(databricks): keep library entries modelled and treat ? as a wildcard get_pipeline_libraries returns DLTLibrarySource, but the spec.configuration and spec.development source_path fallback still appended a raw string, so iterating the list hit AttributeError on library.is_directory and lineage died for pipelines that use those fields. glob_base_directory only looked for *, while glob_matches also honours ?, so an include such as /transformations/file_?.sql was treated as a concrete path and sent to the export API literally. Both now share GLOB_WILDCARDS so they cannot disagree again. * fix(databricks): expand a pattern-less library directory in full A configuration or development source_path produces a library with no pattern, and treating that as non-recursive skipped every nested transformation. No pattern means no filter, so the whole tree is in scope. An include that names a bare directory was also stored as its own pattern, which then matched nothing and dropped the directory entirely. Only attach a pattern when the include actually contains a wildcard. Cover every spec.libraries shape end to end so the reduction and the matching cannot drift apart again. * fix(databricks): state directory-ness instead of inferring it from the path is_directory was read off a trailing slash, so a source_path that Databricks spells without one was exported as a file and its transformations never listed. The producer knows what each entry is, so it now says so: notebook and file entries are files, a wildcard include is always a directory, and a source_path is normalised to a directory. is_recursive only looked for **, so a wildcard in a directory segment such as /tx/2024_?/file.sql reduced to /tx/ and then refused to descend, selecting nothing. Decide from the part of the pattern below the base directory, where either ** or a / means traversal is required. * fix(databricks): let the workspace settle an ambiguous glob include A glob include with neither a wildcard nor a trailing slash may name a file or a directory, and the string cannot tell them apart: assuming a file leaves a directory unlisted, assuming a directory swallows a concrete .sql. Model the gap instead, with is_directory None meaning the spec did not say, and resolve it by listing the path. Anything that lists is a directory, anything that does not is read as a single source. * fix(databricks): read the legacy and private DLT create forms CREATE LIVE TABLE, CREATE STREAMING LIVE TABLE and the PRIVATE modifier are spellings DLT still accepts but the query parser cannot read. It falls back to a bare command and reports no tables, so the dataset was dropped with neither an error nor a warning. Rewrite them to the modern equivalent before parsing, anchored to the CREATE clause so a dataset whose own name contains "live" is left alone. * refactor(databricks): drop the glob machinery Databricks cannot produce The pipelines API accepts only an exact file, a directory, or a directory with a trailing **, and rejects every other wildcard form outright. The pattern matcher, the wildcard set and the recursive flag were therefore unreachable, along with the tests asserting on them. A library is now just a file or a directory taken in full. * fix(databricks): describe library discovery as it behaves and lift the depth cap The get_pipeline_libraries docstring still promised pattern filtering that no longer exists, and the expansion cap of 5 silently dropped transformations nested deeper than that, which costs the lineage depending on them. Correct the docstring and raise the cap to 20. A directory that is not there costs nothing to look for. * fix(databricks): use the real lookup cache in the self-referencing lineage test The test stubbed _table_lookup_cache with a plain dict, which satisfied main's implementation but not this branch's LRUCache. The lookup raised, the exception was swallowed, and the two tests expecting an edge failed once main was merged in. * fix(databricks): read a null libraries value and quieten per-file logging The libraries key can be present and null, and testing for the key rather than reading it made len() raise before the source paths were collected, which cost the pipeline its lineage. Per-path discovery also logged at INFO, which buries the run summary now that expansion is recursive.
Rohit0301
pushed a commit
that referenced
this pull request
Sep 3, 2026
#31654) * Fixes 31278: Extract lineage from SQL-defined Databricks DLT pipelines SQL DLT transformations parsed to nothing because kafka_parser only understood the Python DLT API, so those pipelines produced no lineage. Parse CREATE [OR REFRESH] MATERIALIZED VIEW / STREAMING TABLE through the existing LineageParser, resolve qualified upstreams against their own catalog and schema, read `file` libraries, and recurse into workspace subdirectories. * refactor(databricks): dispatch DLT parsing through a language registry Adding SQL support as a fallback inside extract_dlt_table_dependencies left an if/elif chain that dispatched on failure rather than language, against the registry convention in docs/design-patterns.md. Introduce dlt_parsers.py with a DltSourceParser protocol and registry. Python and SQL each declare handles/extract and an explicit priority, so a Python notebook embedding SQL strings is not misread and a new language needs no dispatch edit. Move the parse-result dataclasses to models.py and reduce kafka_parser.py to Kafka configuration. Fix three Databricks SQL shapes found while probing real constructs: APPLY CHANGES INTO lost its source, the LIVE. prefix was resolved as a schema, and table-valued functions were returned as tables. Also use the previously dead get_pipeline_libraries helper instead of a duplicated inline copy. * fix(databricks): address PR review on DLT source discovery Glob includes were only normalised for the `/**` form, so `*.sql` and `**/*.sql` kept their wildcard and were exported as literal paths, missing every transformation. Reduce any include to the directory before its first wildcard, always ending in a slash. Stripping the LIVE. prefix could collapse two references onto one dataset, so de-duplicate upstreams while preserving order. Drop table-valued functions only when the statement invokes them, so a dataset named `range` still resolves. Strip whitespace and stray dots when qualifying a name. Remove the dead DLT_TABLE_PATTERN and the unused get_pipeline_libraries client parameter. * refactor(databricks): model DLT references and simplify parser dispatch _qualify_dlt_table_name returned a bare (catalog, schema, table) tuple, so callers had to know what each position meant. Return a DLTTableReference instead, alongside the other connector models. Replace the parser registry, the Protocol priority field and the per-call sort with an explicit ordered DLT_PARSERS list. Two implementations did not justify the indirection, and the previous approach also misused enum_register, which everywhere else keys functions by an enum value rather than classes by an ad-hoc string. Order is now stated where the list is defined, and a test asserts it rather than assuming it. Bound the entity caches with LRUCache, which CLAUDE.md requires and the sibling Databricks ownership source already uses. Typing the cache surfaced that fqn.build can return None and the old code cached under a None key, so collapse the two duplicated cache blocks into a single _lookup_table helper that guards it once. Build the source through the real create() in tests instead of assigning its internals. The hand-assigned caches had already drifted from production once without any test failing. Cache bounds, eviction, miss caching and the None guard are now asserted against the real object. * fix(databricks): select only the files a DLT glob actually matches glob_base_directory reduced an include to its base directory and the caller then expanded that whole tree, so a selective include such as /transformations/*.sql pulled in unrelated files and attributed their lineage to the pipeline. That is worse than the wildcard bug it replaced, because the edges are wrong rather than absent. Model a library entry as DLTLibrarySource, carrying the directory to list and the pattern its contents must match. Recurse only for **, and check every candidate with glob_matches, where * stays inside a path segment and ** spans them. fnmatch is unsuitable because its * also crosses /. Tests now assert which files an include selects rather than which directory it reduces to, which is what the earlier assertion missed. * fix(databricks): keep library entries modelled and treat ? as a wildcard get_pipeline_libraries returns DLTLibrarySource, but the spec.configuration and spec.development source_path fallback still appended a raw string, so iterating the list hit AttributeError on library.is_directory and lineage died for pipelines that use those fields. glob_base_directory only looked for *, while glob_matches also honours ?, so an include such as /transformations/file_?.sql was treated as a concrete path and sent to the export API literally. Both now share GLOB_WILDCARDS so they cannot disagree again. * fix(databricks): expand a pattern-less library directory in full A configuration or development source_path produces a library with no pattern, and treating that as non-recursive skipped every nested transformation. No pattern means no filter, so the whole tree is in scope. An include that names a bare directory was also stored as its own pattern, which then matched nothing and dropped the directory entirely. Only attach a pattern when the include actually contains a wildcard. Cover every spec.libraries shape end to end so the reduction and the matching cannot drift apart again. * fix(databricks): state directory-ness instead of inferring it from the path is_directory was read off a trailing slash, so a source_path that Databricks spells without one was exported as a file and its transformations never listed. The producer knows what each entry is, so it now says so: notebook and file entries are files, a wildcard include is always a directory, and a source_path is normalised to a directory. is_recursive only looked for **, so a wildcard in a directory segment such as /tx/2024_?/file.sql reduced to /tx/ and then refused to descend, selecting nothing. Decide from the part of the pattern below the base directory, where either ** or a / means traversal is required. * fix(databricks): let the workspace settle an ambiguous glob include A glob include with neither a wildcard nor a trailing slash may name a file or a directory, and the string cannot tell them apart: assuming a file leaves a directory unlisted, assuming a directory swallows a concrete .sql. Model the gap instead, with is_directory None meaning the spec did not say, and resolve it by listing the path. Anything that lists is a directory, anything that does not is read as a single source. * fix(databricks): read the legacy and private DLT create forms CREATE LIVE TABLE, CREATE STREAMING LIVE TABLE and the PRIVATE modifier are spellings DLT still accepts but the query parser cannot read. It falls back to a bare command and reports no tables, so the dataset was dropped with neither an error nor a warning. Rewrite them to the modern equivalent before parsing, anchored to the CREATE clause so a dataset whose own name contains "live" is left alone. * refactor(databricks): drop the glob machinery Databricks cannot produce The pipelines API accepts only an exact file, a directory, or a directory with a trailing **, and rejects every other wildcard form outright. The pattern matcher, the wildcard set and the recursive flag were therefore unreachable, along with the tests asserting on them. A library is now just a file or a directory taken in full. * fix(databricks): describe library discovery as it behaves and lift the depth cap The get_pipeline_libraries docstring still promised pattern filtering that no longer exists, and the expansion cap of 5 silently dropped transformations nested deeper than that, which costs the lineage depending on them. Correct the docstring and raise the cap to 20. A directory that is not there costs nothing to look for. * fix(databricks): use the real lookup cache in the self-referencing lineage test The test stubbed _table_lookup_cache with a plain dict, which satisfied main's implementation but not this branch's LRUCache. The lookup raised, the exception was swallowed, and the two tests expecting an edge failed once main was merged in. * fix(databricks): read a null libraries value and quieten per-file logging The libraries key can be present and null, and testing for the key rather than reading it made len() raise before the source paths were collected, which cost the pipeline its lineage. Per-path discovery also logged at INFO, which buries the run summary now that expansion is recursive.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Describe your changes:
Fixes #31278
The Databricks Pipeline connector discovers DLT lineage by exporting each pipeline's source and parsing it in
kafka_parser.py. That parser only understood the Python DLT API, so a pipeline whose transformations are.sqlfiles produced no lineage at all. Every file was skipped withSkipping lineage for this notebook - no DLT tables found, even though the SQL contained fully qualified, resolvable upstream references and both endpoints already existed in OpenMetadata.This adds SQL support and fixes two neighbouring gaps in source discovery.
dlt_parsers.py(new)DltSourceParserprotocol plus adlt_parser_registry, following the registry convention indocs/design-patterns.mdrather than anif/elifchain. Each language declares whether ithandlesa source and how toextractdatasets from it, so a future language is one new class with no edits to the dispatch.priorityinstead of relying on registration order. Python is checked first because a@dlt.decorator is unambiguous, whereas SQL keywords can appear inside a string literal in a Python notebook that callsspark.sql(...).SqlDltParserrecognisesCREATE [OR REFRESH] [TEMPORARY|PRIVATE] MATERIALIZED VIEW,STREAMING TABLE, the legacyLIVE TABLE, andAPPLY CHANGES INTOfor CDC. Statements are split withsqlparseand handed to the existingLineageParser, so dialect handling, masking and timeouts are reused rather than reimplemented.STREAM(table)is unwrapped (a DLT marker, otherwise reported as a table calledstream), theLIVE.prefix is stripped (a pipeline namespace, not a schema), and table-valued functions such asread_filesandcloud_filesare dropped since no entity backs them.PythonDltParseris the existing implementation moved unchanged behind the same contract.kafka_parser.pyget_pipeline_librariesnow handlesnotebook,fileandglobin one place and is finally used by the connector, replacing a duplicated inline copy that was previously dead test-only code.models.pyKafkaSourceConfigandDLTTableDependencymoved here, matching the convention that connector types live inmodels.py.metadata.py_qualify_dlt_table_name()splits a dataset reference into catalog, schema and table. SQL upstreams are either bare (a sibling dataset in the same pipeline, resolved against the pipeline's target catalog and schema) or fully qualified (resolved where they actually live). Previously a qualified reference was nested under the pipeline target, producing an unresolvable FQN likeservice.pipeline_catalog.pipeline_schema."raw_catalog.raw_schema.orders_raw".spec.librariesentries of the form{"file": {"path": ...}}are now read. Pipelines sourced from Git folders or Asset Bundles declare files rather than notebooks, and those were previously ignored entirely._expand_workspace_directory()walks subdirectories.workspace/listreturns immediate children only, so aglobsuch as/Repos/project/transformations/**previously missed anything nested. Recursion is depth-capped.Type of change:
High-level design:
The parsing work is delegated to the existing
LineageParserrather than a new regex.LineageParseralready dispatches across the SqlFluff, SqlGlot and SqlParse analyzers and is used by other pipeline connectors (dbtcloud,airflow), so this extends the established pattern instead of adding a parallel one.sqlparse.split()handles files that declare several datasets, giving one target plus its upstreams per statement, which preserves the per-dataset dependency mapping that a whole-file parse would flatten.Qualifier resolution is deliberately a small pure helper (
_qualify_dlt_table_name) so it is directly testable and so the Python path, which only ever yields bare names, keeps its existing behaviour.Backward compatibility: the SQL path is a fallback that only fires when the Python parser returns nothing, and the added library and directory handling are additive. No schema changes, no migrations.
Tests:
Use cases covered
.sqlfiles now produces table-to-table lineage with the pipeline attached inlineageDetails.CREATE OR REFRESH MATERIALIZED VIEW.Unit tests
ingestion/tests/unit/topology/pipeline/test_databricks_dlt_sql_lineage.py(25 tests)105 passedacross the Databricks pipeline suite (80 pre-existing plus 25 new).51/51identical for dataset dependencies and51/51identical for Kafka sources.802 passedacross all pipeline connector tests. Threetest_airflow_connection.pyfailures and 13 collection errors across the wider unit suite are pre-existing in this environment (missingairflow,paramiko,pydorisand a missing generatedopenlineagemodule). The erroring module set is byte-identical with and without this change.Backend integration tests
Ingestion integration tests
Playwright (UI) tests
Manual testing performed
Reproduced against payloads captured verbatim from a real failing agent run, replaying the recorded pipeline spec,
workspace/listresponse and exported sources through the shipped connector with only the Databricks client stubbed:Edge cases were probed against real Databricks SQL rather than assumed: CDC via
APPLY CHANGES INTO, theLIVE.prefix, Auto Loader table functions, backticked identifiers, CTEs,UNION, multi-statement files, and unparseable input. Each is covered by a test.Also ran
make py_format_check(clean) andmake static-checks. Static checks report the same38 errors, 75 warningswith and without this change, so no new type errors are introduced.UI screen recording / screenshots:
Not applicable.
Checklist:
Fixes <issue-number>: <short explanation>Fixes #<issue-number>above.Greptile Summary
The PR adds SQL-defined Databricks DLT lineage extraction, consolidates pipeline-library discovery, qualifies SQL table references, and recursively discovers workspace sources.
Confidence Score: 3/5
The PR is not yet safe to merge because selective globs can add lineage from unrelated transformations and question-mark globs can omit all selected lineage.
The current normalization discards selective
*suffixes and never recognizes?, while directory expansion applies no original-pattern filter; these previously reported source-selection failures therefore remain at HEAD.Files Needing Attention: ingestion/src/metadata/ingestion/source/pipeline/databrickspipeline/kafka_parser.py and ingestion/src/metadata/ingestion/source/pipeline/databrickspipeline/metadata.py
Important Files Changed
Flowchart
%%{init: {'theme': 'neutral'}}%% flowchart LR Spec[Databricks pipeline spec] --> Libraries[Resolve notebook, file, glob, or source_path] Libraries --> Expand[Expand workspace directories] Expand --> Export[Export source code] Export --> Dispatch{Recognized source language} Dispatch -->|Python DLT| Python[PythonDltParser] Dispatch -->|SQL DLT| SQL[SqlDltParser and LineageParser] Python --> Dependencies[DLT table dependencies] SQL --> Dependencies Dependencies --> Qualify[Qualify catalog, schema, and table] Qualify --> Lineage[Publish OpenMetadata lineage]Reviews (13): Last reviewed commit: "fix(databricks): read a null libraries v..." | Re-trigger Greptile