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.
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: