fix(ingest/adf): recurse into ForEach/IfCondition/Until container activities for lineage extraction - #19488
fix(ingest/adf): recurse into ForEach/IfCondition/Until container activities for lineage extraction#19488brock-acryl wants to merge 2 commits into
Conversation
…ner activities for lineage extraction Container activities got a DataJob node but their nested activities list was never visited, so any Copy/Lookup activity nested inside a ForEach, IfCondition, Until, or Switch produced no lineage edges. Also add Lookup activity input extraction, which exposes its dataset via typeProperties.dataset.referenceName rather than the inputs attribute used by Copy. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
github-actions
Bot
added
the
ingestion
PR SummaryOverview Pipeline processing no longer stops at top-level activities: it walks nested children with a BFS queue, enqueues children from a new Integration golden files Reviewed by Cursor Bugbot for commit 6e6630d. Bugbot is set up for automated code reviews on this repo. Configure here. |
Codecov Report❌ Patch coverage is
❌ Your patch check has failed because the patch coverage (6.45%) is below the target coverage (75.00%). You can increase the patch coverage or adjust the target coverage. 📢 Thoughts on this report? Let us know! |
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
2 issues found across 3 files
You’re at about 90% of the monthly reviewed-line limit. You may want to disable incremental reviews to conserve quota. Reviews will continue until that limit is exceeded. If you need help avoiding interruptions, please contact contact@cubic.dev.
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="metadata-ingestion/tests/integration/azure_data_factory/adf_foreach_golden.json">
<violation number="1" location="metadata-ingestion/tests/integration/azure_data_factory/adf_foreach_golden.json:227">
P2: The new GetTableList dataJobInputOutput aspect records systemMetadata runId 'adf-test-foreach', while the other three aspects of the same dataJob (dataPlatformInstance, dataJobInfo, subTypes) and the test's configured run_id all use 'adf-foreach-test'. Fix the added aspect to use 'adf-foreach-test' so system metadata stays consistent within the job and matches the golden snapshot's run.</violation>
</file>
<file name="metadata-ingestion/tests/integration/azure_data_factory/adf_branching_golden.json">
<violation number="1" location="metadata-ingestion/tests/integration/azure_data_factory/adf_branching_golden.json:239">
P2: The new nested-activity DataJob entities are recorded with runId "adf-test-branching", but this golden file is also validated by test_branching_pipeline in test_complex_pipelines.py, which runs with run_id "adf-branching-test". After this PR that test also recurses into the Switch/IfCondition branches, so its regenerated nested entities will carry "adf-branching-test" and will not match the entries added here. The golden appears to have been regenerated from only one of the two tests that share it. Verify that test_branching_pipeline still passes against this fixture (its output nested DataJobs carry a different run id) rather than relying on a stale regeneration.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| }, | ||
| "systemMetadata": { | ||
| "lastObserved": 1705320000000, | ||
| "runId": "adf-test-foreach", |
There was a problem hiding this comment.
P2: The new GetTableList dataJobInputOutput aspect records systemMetadata runId 'adf-test-foreach', while the other three aspects of the same dataJob (dataPlatformInstance, dataJobInfo, subTypes) and the test's configured run_id all use 'adf-foreach-test'. Fix the added aspect to use 'adf-foreach-test' so system metadata stays consistent within the job and matches the golden snapshot's run.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At metadata-ingestion/tests/integration/azure_data_factory/adf_foreach_golden.json, line 227:
<comment>The new GetTableList dataJobInputOutput aspect records systemMetadata runId 'adf-test-foreach', while the other three aspects of the same dataJob (dataPlatformInstance, dataJobInfo, subTypes) and the test's configured run_id all use 'adf-foreach-test'. Fix the added aspect to use 'adf-foreach-test' so system metadata stays consistent within the job and matches the golden snapshot's run.</comment>
<file context>
@@ -209,6 +209,25 @@
+ },
+ "systemMetadata": {
+ "lastObserved": 1705320000000,
+ "runId": "adf-test-foreach",
+ "lastRunId": "no-run-id-provided"
+ }
</file context>
| "runId": "adf-test-foreach", | |
| "runId": "adf-foreach-test", |
| }, | ||
| "systemMetadata": { | ||
| "lastObserved": 1705320000000, | ||
| "runId": "adf-test-branching", |
There was a problem hiding this comment.
P2: The new nested-activity DataJob entities are recorded with runId "adf-test-branching", but this golden file is also validated by test_branching_pipeline in test_complex_pipelines.py, which runs with run_id "adf-branching-test". After this PR that test also recurses into the Switch/IfCondition branches, so its regenerated nested entities will carry "adf-branching-test" and will not match the entries added here. The golden appears to have been regenerated from only one of the two tests that share it. Verify that test_branching_pipeline still passes against this fixture (its output nested DataJobs carry a different run id) rather than relying on a stale regeneration.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At metadata-ingestion/tests/integration/azure_data_factory/adf_branching_golden.json, line 239:
<comment>The new nested-activity DataJob entities are recorded with runId "adf-test-branching", but this golden file is also validated by test_branching_pipeline in test_complex_pipelines.py, which runs with run_id "adf-branching-test". After this PR that test also recurses into the Switch/IfCondition branches, so its regenerated nested entities will carry "adf-branching-test" and will not match the entries added here. The golden appears to have been regenerated from only one of the two tests that share it. Verify that test_branching_pipeline still passes against this fixture (its output nested DataJobs carry a different run id) rather than relying on a stale regeneration.</comment>
<file context>
@@ -221,6 +221,25 @@
+ },
+ "systemMetadata": {
+ "lastObserved": 1705320000000,
+ "runId": "adf-test-branching",
+ "lastRunId": "no-run-id-provided"
+ }
</file context>
Connector Tests ResultsAll connector tests passed for commit To skip connector tests, add the Autogenerated by the connector-tests CI pipeline. |
| for activity in pipeline.activities or []: | ||
| # Emit activities as DataJobs, using BFS to recurse into container | ||
| # activities (ForEach, IfCondition, Until, Switch) so nested activities | ||
| # like Copy also get DataJobs with proper lineage. |
There was a problem hiding this comment.
🟡 W1: list.pop(0) is O(n) per element — use collections.deque
BFS with list.pop(0) shifts the entire list left on each iteration, making this O(n²) for pipelines with many activities. deque.popleft() is O(1).
| # like Copy also get DataJobs with proper lineage. | |
| activities_to_process: deque[Activity] = deque(pipeline.activities or []) | |
| while activities_to_process: | |
| activity = activities_to_process.popleft() |
(Add from collections import deque at the top of the file.)
| ref_name = getattr(dataset_ref, "reference_name", None) | ||
| if ref_name: | ||
| dataset_urn = self._resolve_dataset_urn(ref_name, factory_key) | ||
| if dataset_urn: |
There was a problem hiding this comment.
🟡 W2: Expression-based dataset references produce phantom URNs
The Lookup extraction (and the existing Copy extraction) resolves reference_name literally. When ADF datasets use parameterized references like @dataset().table_name (a very common ADF pattern), the resulting URN is:
urn:li:dataset:(urn:li:dataPlatform:mssql,{'value': '@dataset().table_name', 'type': 'Expression'},PROD)
These phantom datasets do not match any real ingested datasets, so the lineage graph shows edges to nonexistent entities — the UI renders no lineage.
Consider detecting expressions (reference names containing @ or {) and either:
- Skipping lineage emission with a
self.report.warning(title="Expression-based dataset reference", message="...", context=f"dataset={ref_name}"), or - Attempting to resolve the expression against the ADF dataset definitions to find the actual backing table name.
|
|
||
| Returns a flat list of child activities. | ||
| """ | ||
| nested: list[Activity] = [] |
There was a problem hiding this comment.
ℹ️ S1: No cycle protection in BFS
If the Azure SDK ever returns a malformed pipeline where a container activity references itself (or two containers reference each other), this loop runs forever. A visited set is cheap defensive programming:
| nested: list[Activity] = [] | |
| def _get_nested_activities(self, activity: Activity) -> list[Activity]: | |
| """Extract nested child activities from container activities.""" |
And in the BFS loop:
visited: set[str] = set()
while activities_to_process:
activity = activities_to_process.popleft()
activity_id = f"{activity.name}:{activity.type}"
if activity_id in visited:
continue
visited.add(activity_id)
...
maggiehays
added
the
pending-submitter-response
…versal Addresses review feedback on PR #19488: list.pop(0) made the BFS queue O(n^2) for pipelines with many nested activities, and there was no protection against a malformed/self-referencing pipeline definition looping forever. Switches to collections.deque with popleft() (O(1)) and adds a visited-activities guard. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
maggiehays
added
needs-review
3 participants
Summary
_process_pipelines()only iterated top-level pipeline activities. Container activity types (ForEach,IfCondition,Until,Switch) got a DataJob node but their nestedactivitieslist was never visited — so anyCopyActivity/LookupActivitynested inside aForEachproduced no lineage edges._extract_activity_inputs()had no handling forLookupActivity, which exposes its dataset viatypeProperties.dataset.referenceNamerather than theinputsattribute used byCopyActivity.What changed:
activity.typeisForEachorUntil, recurse intoactivity.activities; forIfCondition, recurse into bothif_true_activitiesandif_false_activities; forSwitch, recurse into each case'sactivitiesplusdefault_activities.LookupActivityinput extraction viadataset.reference_name.adf_foreach_golden.jsonandadf_branching_golden.json— both now include the previously-invisible inner DataJob nodes and their lineage edges.Before: A
Databricks → ForEach[Copy] → MSSQLpipeline showed no lineage (theCopyActivityinsideForEachwas never processed)After:
CopyTableToStagingDataJob is emitted with full input/output lineageTest plan
test_adf_source.py: 14/14 passedtest_complex_pipelines.py: 13/13 passedadf_foreach_golden.jsonandadf_branching_golden.jsonupdated to reflect new lineage emissions./gradlew :metadata-ingestion:lintFix— clean, no additional changesChecklist
🤖 Generated with Claude Code
Summary by cubic
Fixes Azure Data Factory lineage extraction so Copy and Lookup activities nested inside container activities (
ForEach,IfCondition,Until,Switch) now emit DataJob nodes with lineage edges. Previously only top-level pipeline activities were processed, so any nested activity produced no lineage.ForEach/Untilviaactivities,IfConditionviaif_true_activities/if_false_activities, andSwitchvia each case'sactivitiesplusdefault_activities.dequeBFS and a visited-set guard, so malformed or self-referencing pipeline definitions can't loop forever.dataset.reference_nameinstead of theinputsattribute used by Copy activities.adf_foreach_golden.jsonandadf_branching_golden.jsonto include the previously missing inner DataJobs.Written for commit 6e6630d. Summary will update on new commits.