feat: add column-level lineage to the OpenLineage - #19643
Conversation
FrankChen021
left a comment
There was a problem hiding this comment.
| Severity | Findings |
|---|---|
| P0 | 0 |
| P1 | 0 |
| P2 | 3 |
| P3 | 0 |
| Total | 3 |
| Severity | Findings |
|---|---|
| P0 | 0 |
| P1 | 0 |
| P2 | 3 |
| P3 | 0 |
| Total | 3 |
Found 3 issues.
Reviewed 5 of 5 changed files.
This is an automated review by Codex GPT-5.5
|
Drive-by comment: if we can, I'd like to expose this column extraction logic within Druid itself. A few reasons for this:
|
|
@gianm @clintropolis would like to get your thoughts here on the column extraction logic – do we want to make use of any Calcite functionality here? |
FrankChen021
left a comment
There was a problem hiding this comment.
| Severity | Findings |
|---|---|
| P0 | 0 |
| P1 | 0 |
| P2 | 2 |
| P3 | 0 |
| Total | 2 |
Reviewed 8 of 8 changed files.
Found 2 issues in the updated lineage implementation. The prior datasource-filter and filtered-aggregator concerns look addressed.
This is an automated review by Codex GPT-5.5
FrankChen021
left a comment
There was a problem hiding this comment.
| Severity | Findings |
|---|---|
| P0 | 0 |
| P1 | 0 |
| P2 | 1 |
| P3 | 0 |
| Total | 1 |
Reviewed 8 of 8 changed files.
This is an automated review by Codex GPT-5.6-Sol
| { | ||
| Set<String> tables = new LinkedHashSet<>(); | ||
| if (query instanceof UnionQuery) { | ||
| for (DataSource dataSource : ((UnionQuery) query).getDataSources()) { |
There was a problem hiding this comment.
[P2] Recurse before extracting union data sources
UnionQuery accepts arbitrary query branches, and UnionQueryLogic recursively executes a branch that is itself a UnionQuery. However, getDataSources() immediately calls getDataSource() on every branch; a nested UnionQuery throws there by design. Consequently, a valid nested union still causes this logger to drop its lineage event and log an error. Walk getQueries() recursively, as QueryColumnUsageAnalyzer.collectInto already does, before collecting each leaf datasource's table names.
|
Removing from Druid 38 milestone since this PR is not fully reviewed yet. |
Fixes #19314
Description
Builds on #19107 (the OpenLineage request-logger extension) to add column-level lineage for native queries. Each input dataset now carries which columns the query referenced and how they were used.
Two facets are attached per input dataset:
schema— the standard OpenLineageSchemaDatasetFacetlisting the referenced input column names (names only, sorted).druid_columnUsage— a Druid-specific dataset facet mapping each referenced column to the role(s) it was used in:PROJECTION,GROUP_BY,AGGREGATION,FILTER,JOIN.Validated in docker:
SELECT page, "user" FROM wikipedia WHERE countryName = '…' LIMIT 10scanpage=PROJECTION, user=PROJECTION, countryName=FILTERSELECT countryName, SUM(added) FROM wikipedia WHERE channel = '…' GROUP BY countryNamegroupBycountryName=GROUP_BY, added=AGGREGATION, channel=FILTERSELECT page, SUM(added) s FROM wikipedia GROUP BY page ORDER BY s DESC LIMIT 5topNpage=GROUP_BY, added=AGGREGATIONSELECT SUM(added) FROM wikipedia WHERE isRobot = 'false'timeseriesadded=AGGREGATION, isRobot=FILTERSELECT w1.page, w2.channel FROM wikipedia w1 JOIN wikipedia w2 ON w1.page = w2.page WHERE w1.countryName = '…'scan(join)page=[PROJECTION, JOIN], channel=PROJECTION, countryName=FILTER— right-sidew2.channelcorrectly un-prefixedSELECT countryName FROM (SELECT countryName, SUM(added) s FROM wikipedia GROUP BY countryName) GROUP BY countryNamegroupBycountryName=GROUP_BY— no fabricated sub-query-output columnsExample emitted facets for the join query's input dataset:
Release note
The OpenLineage emitter now emits column-level lineage for native queries as
schemaanddruid_columnUsagefacets on input datasets. This can be disabled withdruid.request.logging.columnLineageEnabled=false.Key changed/added classes in this PR
OpenLineageRequestLoggerOpenLineageRequestLoggerProviderDruidColumnUsageDatasetFacet.jsonThis PR has: