FrankChen021 commented on code in PR #19643:
URL: https://github.com/apache/druid/pull/19643#discussion_r3570771066
##########
extensions-contrib/openlineage-emitter/src/main/java/org/apache/druid/extensions/openlineage/OpenLineageRequestLogger.java:
##########
@@ -215,14 +229,34 @@ public void logNativeQuery(RequestLogLine requestLogLine)
throws IOException
return;
}
- List<String> inputs = new ArrayList<>(new
LinkedHashSet<>(requestLogLine.getQuery().getDataSource().getTableNames()));
+ List<String> inputs = new
ArrayList<>(extractInputTables(requestLogLine.getQuery()));
String queryId = requestLogLine.getQuery().getId();
if (queryId == null) {
log.debug("Native query reached OpenLineage logger without a query ID");
queryId = UNKNOWN_QUERY_ID;
}
- emit(buildRunEvent(queryId, queryType, requestLogLine, inputs, null));
+ Map<String, Map<String, EnumSet<QueryColumnUsageAnalyzer.ColumnUsage>>>
columnsByTable =
+ columnLineageEnabled ?
extractColumnsByTable(requestLogLine.getQuery()) : null;
+ emit(buildRunEvent(queryId, queryType, requestLogLine, inputs,
columnsByTable, null));
+ }
+
+ /**
+ * Collects the input table names of a native query. A top-level {@link
UnionQuery} has no single
+ * datasource ({@link UnionQuery#getDataSource()} throws by design), so its
branches are unioned;
+ * every other query exposes its tables through {@code
getDataSource().getTableNames()}.
+ */
+ private static Set<String> extractInputTables(Query<?> query)
+ {
+ Set<String> tables = new LinkedHashSet<>();
+ if (query instanceof UnionQuery) {
+ for (DataSource dataSource : ((UnionQuery) query).getDataSources()) {
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
