Skip to content

Commit 5a0e71f

Browse files
author
Jonathan P Moraes
authored
Refreshing table function review (#54)
* Fixing downstream column-lineage bug
1 parent 3536d8a commit 5a0e71f

2 files changed

Lines changed: 285 additions & 140 deletions

File tree

‎api/src/main/java/marquez/db/ColumnLineageDao.java‎

Lines changed: 130 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -253,60 +253,147 @@ Set<ColumnLineageNodeData> getLineageRowsForDatasets(
253253
List<Pair<String, String>> datasets);
254254

255255
/**
256-
* Fetch all of the column lineage nodes that are directly connected to the input dataset fields.
257-
* This returns a single layer of lineage using column lineage as edges. Fields that have
258-
* no input or output lineage will have no results.
256+
* Fetch upstream lineage nodes that directly produce data TO the input dataset fields.
257+
* This returns fields that are direct producers of the given fields.
258+
* Only follows INPUT edges to the given fields (direct upstream producers).
259259
*
260-
* @param datasetFieldUuids The UUIDs of the dataset fields to get lineage for
261-
* @param withDownstream Whether to include downstream lineage
260+
* @param datasetFieldUuids The UUIDs of the dataset fields to get upstream producers for
262261
* @param createdAtUntil The point in time to get lineage for
263-
* @return Set of ColumnLineageNodeData representing the direct lineage
262+
* @return Set of ColumnLineageNodeData representing the direct upstream producer fields
264263
*/
265-
@SqlQuery(
266-
"""
267-
WITH dataset_fields_view AS (
268-
SELECT d.namespace_name as namespace_name, d.name as dataset_name, df.name as field_name, df.type, df.uuid, d.namespace_uuid
264+
@SqlQuery("""
265+
WITH dataset_fields_view AS (
266+
SELECT
267+
d.namespace_name as namespace_name,
268+
d.name as dataset_name,
269+
df.name as field_name,
270+
df.type,
271+
df.uuid,
272+
d.namespace_uuid
269273
FROM dataset_fields df
270274
INNER JOIN datasets_view d ON d.uuid = df.dataset_uuid
271-
)
272-
SELECT
273-
output_fields.namespace_name,
274-
output_fields.dataset_name,
275-
output_fields.field_name,
276-
output_fields.type,
277-
ARRAY_AGG(DISTINCT ARRAY[
275+
),
276+
-- Find upstream producer fields: fields that produce data TO our target fields
277+
upstream_producers AS (
278+
SELECT DISTINCT
278279
input_fields.namespace_name,
279280
input_fields.dataset_name,
280-
CAST(cl.input_dataset_version_uuid AS VARCHAR),
281281
input_fields.field_name,
282-
cl.transformation_description,
283-
cl.transformation_type
284-
]) AS inputFields,
285-
ARRAY_AGG(DISTINCT ARRAY[
282+
input_fields.type,
283+
ARRAY[]::text[][] AS inputFields,
284+
ARRAY_AGG(DISTINCT ARRAY[
285+
output_fields.namespace_name,
286+
output_fields.dataset_name,
287+
CAST(cl.output_dataset_version_uuid AS VARCHAR),
288+
output_fields.field_name,
289+
cl.transformation_description,
290+
cl.transformation_type
291+
]) AS outputFields,
292+
cl.input_dataset_version_uuid as dataset_version_uuid
293+
FROM tmp_column_lineage_latest cl
294+
INNER JOIN dataset_fields_view input_fields
295+
ON cl.input_dataset_field_uuid = input_fields.uuid
296+
INNER JOIN dataset_symlinks ds_input
297+
ON ds_input.namespace_uuid = input_fields.namespace_uuid
298+
AND ds_input.name = input_fields.dataset_name
299+
INNER JOIN dataset_fields_view output_fields
300+
ON cl.output_dataset_field_uuid = output_fields.uuid
301+
INNER JOIN dataset_symlinks ds_output
302+
ON ds_output.namespace_uuid = output_fields.namespace_uuid
303+
AND ds_output.name = output_fields.dataset_name
304+
WHERE cl.output_dataset_field_uuid IN (<datasetFieldUuids>)
305+
AND cl.created_at <= :createdAtUntil
306+
AND ds_input.is_primary is true
307+
AND ds_output.is_primary is true
308+
GROUP BY
309+
input_fields.namespace_name,
310+
input_fields.dataset_name,
311+
input_fields.field_name,
312+
input_fields.type,
313+
cl.input_dataset_version_uuid
314+
)
315+
SELECT * FROM upstream_producers
316+
""")
317+
Set<ColumnLineageNodeData> getUpstreamColumnLineage(
318+
@BindList(onEmpty = NULL_STRING) List<UUID> datasetFieldUuids,
319+
Instant createdAtUntil);
320+
321+
/**
322+
* Fetch downstream lineage nodes that directly consume data FROM the input dataset fields.
323+
* This returns fields that are direct consumers of the given fields.
324+
* Only follows OUTPUT edges from the given fields (direct downstream consumers).
325+
*
326+
* @param datasetFieldUuids The UUIDs of the dataset fields to get downstream consumers for
327+
* @param createdAtUntil The point in time to get lineage for
328+
* @return Set of ColumnLineageNodeData representing the direct downstream consumer fields
329+
*/
330+
@SqlQuery("""
331+
WITH dataset_fields_view AS (
332+
SELECT
333+
d.namespace_name as namespace_name,
334+
d.name as dataset_name,
335+
df.name as field_name,
336+
df.type,
337+
df.uuid,
338+
d.namespace_uuid
339+
FROM dataset_fields df
340+
INNER JOIN datasets_view d ON d.uuid = df.dataset_uuid
341+
),
342+
-- Find downstream consumer fields: fields that consume data FROM our target fields
343+
downstream_consumers AS (
344+
SELECT DISTINCT
286345
output_fields.namespace_name,
287346
output_fields.dataset_name,
288-
CAST(cl.output_dataset_version_uuid AS VARCHAR),
289347
output_fields.field_name,
290-
cl.transformation_description,
291-
cl.transformation_type
292-
]) FILTER (WHERE cl.output_dataset_field_uuid IS NOT NULL) AS outputFields,
293-
cl.output_dataset_version_uuid as dataset_version_uuid
294-
FROM tmp_column_lineage_latest cl
295-
INNER JOIN dataset_fields_view output_fields ON cl.output_dataset_field_uuid = output_fields.uuid
296-
INNER JOIN dataset_symlinks ds_output ON ds_output.namespace_uuid = output_fields.namespace_uuid AND ds_output.name = output_fields.dataset_name
297-
LEFT JOIN dataset_fields_view input_fields ON cl.input_dataset_field_uuid = input_fields.uuid
298-
INNER JOIN dataset_symlinks ds_input ON ds_input.namespace_uuid = input_fields.namespace_uuid AND ds_input.name = input_fields.dataset_name
299-
WHERE output_fields.uuid IN (<datasetFieldUuids>)
300-
AND ds_output.is_primary is true
301-
AND ds_input.is_primary is true
302-
AND cl.created_at <= :createdAtUntil
303-
GROUP BY
304-
output_fields.namespace_name,
305-
output_fields.dataset_name,
306-
output_fields.field_name,
307-
output_fields.type,
308-
cl.output_dataset_version_uuid
309-
""")
348+
output_fields.type,
349+
ARRAY_AGG(DISTINCT ARRAY[
350+
input_fields.namespace_name,
351+
input_fields.dataset_name,
352+
CAST(cl.input_dataset_version_uuid AS VARCHAR),
353+
input_fields.field_name,
354+
cl.transformation_description,
355+
cl.transformation_type
356+
]) AS inputFields,
357+
ARRAY[]::text[][] AS outputFields,
358+
cl.output_dataset_version_uuid as dataset_version_uuid
359+
FROM tmp_column_lineage_latest cl
360+
INNER JOIN dataset_fields_view output_fields
361+
ON cl.output_dataset_field_uuid = output_fields.uuid
362+
INNER JOIN dataset_symlinks ds_output
363+
ON ds_output.namespace_uuid = output_fields.namespace_uuid
364+
AND ds_output.name = output_fields.dataset_name
365+
INNER JOIN dataset_fields_view input_fields
366+
ON cl.input_dataset_field_uuid = input_fields.uuid
367+
INNER JOIN dataset_symlinks ds_input
368+
ON ds_input.namespace_uuid = input_fields.namespace_uuid
369+
AND ds_input.name = input_fields.dataset_name
370+
WHERE cl.input_dataset_field_uuid IN (<datasetFieldUuids>)
371+
AND cl.created_at <= :createdAtUntil
372+
AND ds_output.is_primary is true
373+
AND ds_input.is_primary is true
374+
GROUP BY
375+
output_fields.namespace_name,
376+
output_fields.dataset_name,
377+
output_fields.field_name,
378+
output_fields.type,
379+
cl.output_dataset_version_uuid
380+
)
381+
SELECT * FROM downstream_consumers
382+
""")
383+
Set<ColumnLineageNodeData> getDownstreamColumnLineage(
384+
@BindList(onEmpty = NULL_STRING) List<UUID> datasetFieldUuids,
385+
Instant createdAtUntil);
386+
387+
/**
388+
* Fetch all of the column lineage nodes that are directly connected to the input dataset fields.
389+
* This returns a single layer of lineage using column lineage as edges. Fields that have
390+
* no input or output lineage will have no results.
391+
*
392+
* @param datasetFieldUuids The UUIDs of the dataset fields to get lineage for
393+
* @param withDownstream Whether to include downstream lineage
394+
* @param createdAtUntil The point in time to get lineage for
395+
* @return Set of ColumnLineageNodeData representing the direct lineage
396+
*/
310397
Set<ColumnLineageNodeData> getDirectColumnLineage(
311398
@BindList(onEmpty = NULL_STRING) List<UUID> datasetFieldUuids,
312399
boolean withDownstream,

0 commit comments

Comments
 (0)