You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "guan404ming (via GitHub)" <gi...@apache.org> on 2025/11/07 16:46:24 UTC
[PR] Add task dependency graph to AssetGraph [airflow]
guan404ming opened a new pull request, #58059:
URL: https://github.com/apache/airflow/pull/58059
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
## Related Issue
#50873
## Why
- The current asset graph doesn’t display task-level dependencies.
## How
- Separate the dependency graph into two levels:
- Task ↔ Asset
- DAG ↔ Asset
- Add two checkboxes to toggle the visibility of each dependency level.
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add producing/consuming task dependencies to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on PR #58059:
URL: https://github.com/apache/airflow/pull/58059#issuecomment-3821895987
> In another PR, I think we should update the assets list and asset details pages to add scheduled as a column in the table or header card.
Make sense to me. I could do that in a follow up.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add producing/consuming task dependencies to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2703733556
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dependencies.py:
##########
@@ -272,3 +272,35 @@ def test_with_node_id_filter_not_found(self, test_client):
assert response.json() == {
"detail": "Unique connected component not found, got [] for connected components of node missing_node_id, expected only 1 connected component.",
}
+
+ @pytest.mark.parametrize(
+ ("dependency_type", "has_dag", "has_task"),
+ [
+ (None, True, False), # default is scheduling
+ ("scheduling", True, False),
+ ("data", False, True),
+ ],
+ )
+ def test_dependency_type_filter(self, test_client, asset1_id, dependency_type, has_dag, has_task):
+ params = {"node_id": f"asset:{asset1_id}"}
+ if dependency_type is not None:
+ params["dependency_type"] = dependency_type
+
+ response = test_client.get("/dependencies", params=params)
+ assert response.status_code == 200
+
+ result = response.json()
+ node_types = {node["type"] for node in result["nodes"]}
+
+ assert "asset" in node_types
+ assert ("dag" in node_types) == has_dag
+ assert ("task" in node_types) == has_task
Review Comment:
> We probably want to expand this to assert what the graph exactly look like and verify that on a complex structure, nodes are the one expected. (Here returning a graph with no dag and some tasks are enough to success the test)
Sure, I've updated with more detailed tests.
> But we don't know the dag id this task is from. There could be multiple tasks with the same name accross different dags.
I've added a label below the task name for identifying where the task from.
##########
airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py:
##########
@@ -76,3 +82,91 @@ def extract_single_connected_component(
]
return {"nodes": nodes, "edges": edges}
+
+
+def get_data_dependencies(asset_id: int, session: Session) -> dict[str, list[dict]]:
+ """Get full data lineage for an asset."""
+ from sqlalchemy import select
+
+ from airflow.models.asset import TaskInletAssetReference, TaskOutletAssetReference
+
+ SEPARATOR = "__SEPARATOR__"
+
+ nodes_dict: dict[str, dict] = {}
+ edge_set: set[tuple[str, str]] = set()
+
+ # BFS to trace full lineage
+ assets_to_process: deque[int] = deque([asset_id])
+ processed_assets: set[int] = set()
+ processed_tasks: set[tuple[str, str]] = set() # (dag_id, task_id)
+
+ while assets_to_process:
+ current_asset_id = assets_to_process.popleft()
+ if current_asset_id in processed_assets:
+ continue
+ processed_assets.add(current_asset_id)
+
+ asset = session.get(AssetModel, current_asset_id)
Review Comment:
Updated.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add producing/consuming task dependencies to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming merged PR #58059:
URL: https://github.com/apache/airflow/pull/58059
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task dependency graph to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on PR #58059:
URL: https://github.com/apache/airflow/pull/58059#issuecomment-3563008810
Yes, that’s my current understanding. Based on the discussion in the issue, it seems the intention is to surface the tasks inside the dag. But I’m not completely certain, so I’m happy to adjust the implementation if needed.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add producing/consuming task dependencies to AssetGraph [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2746889368
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dependencies.py:
##########
Review Comment:
Ideally we could extract that second part into `get_scheduling_dependencies` to clean the view code and make it easier to read.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add producing/consuming task dependencies to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2747140819
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dependencies.py:
##########
Review Comment:
Just updated, thanks!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task dependency graph to AssetGraph [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2581232914
##########
airflow-core/src/airflow/ui/src/pages/Asset/AssetGraph.tsx:
##########
@@ -29,15 +29,42 @@ import { useGraphLayout } from "src/components/Graph/useGraphLayout";
import { useColorMode } from "src/context/colorMode";
import { useDependencyGraph } from "src/queries/useDependencyGraph";
import { getReactFlowThemeStyle } from "src/theme";
+import { getTaskLevelDependencies } from "src/utils/assetGraph";
-export const AssetGraph = ({ asset }: { readonly asset?: AssetResponse }) => {
+export const AssetGraph = ({
+ asset,
+ showDagLevelDependencies = true,
+ showTaskLevelDependencies = false,
+}: {
+ readonly asset?: AssetResponse;
+ readonly showDagLevelDependencies?: boolean;
+ readonly showTaskLevelDependencies?: boolean;
+}) => {
const { assetId } = useParams();
const { colorMode = "light" } = useColorMode();
- const { data = { edges: [], nodes: [] } } = useDependencyGraph(`asset:${assetId}`);
+ const { data: dagLevelGraphData = { edges: [], nodes: [] } } = useDependencyGraph(`asset:${assetId}`);
+ const taskLevelGraphData = getTaskLevelDependencies({ edges: [], nodes: [] }, asset);
+
+ const combinedGraphData = {
+ edges: [
+ ...(showDagLevelDependencies ? dagLevelGraphData.edges : []),
+ ...(showTaskLevelDependencies ? taskLevelGraphData.edges : []),
+ ],
+ nodes: [
+ // Always show the asset nodes from DAG level data
+ ...dagLevelGraphData.nodes.filter((node) => node.id === `asset:${assetId}`),
+ // Conditionally show dag nodes
+ ...(showDagLevelDependencies
+ ? dagLevelGraphData.nodes.filter((node) => node.id !== `asset:${assetId}`)
+ : []),
+ // Conditionally show task nodes
+ ...(showTaskLevelDependencies ? taskLevelGraphData.nodes : []),
+ ],
+ };
Review Comment:
All that logic should probably be done in the backend, what do you think ?
(We do something similar when resolving asset aliases and downstream assets) for structure_data endpoint https://github.com/apache/airflow/pull/51401/files
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task dependency graph to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2602354256
##########
airflow-core/src/airflow/ui/src/pages/Asset/AssetGraph.tsx:
##########
@@ -29,15 +29,42 @@ import { useGraphLayout } from "src/components/Graph/useGraphLayout";
import { useColorMode } from "src/context/colorMode";
import { useDependencyGraph } from "src/queries/useDependencyGraph";
import { getReactFlowThemeStyle } from "src/theme";
+import { getTaskLevelDependencies } from "src/utils/assetGraph";
-export const AssetGraph = ({ asset }: { readonly asset?: AssetResponse }) => {
+export const AssetGraph = ({
+ asset,
+ showDagLevelDependencies = true,
+ showTaskLevelDependencies = false,
+}: {
+ readonly asset?: AssetResponse;
+ readonly showDagLevelDependencies?: boolean;
+ readonly showTaskLevelDependencies?: boolean;
+}) => {
const { assetId } = useParams();
const { colorMode = "light" } = useColorMode();
- const { data = { edges: [], nodes: [] } } = useDependencyGraph(`asset:${assetId}`);
+ const { data: dagLevelGraphData = { edges: [], nodes: [] } } = useDependencyGraph(`asset:${assetId}`);
+ const taskLevelGraphData = getTaskLevelDependencies({ edges: [], nodes: [] }, asset);
+
+ const combinedGraphData = {
+ edges: [
+ ...(showDagLevelDependencies ? dagLevelGraphData.edges : []),
+ ...(showTaskLevelDependencies ? taskLevelGraphData.edges : []),
+ ],
+ nodes: [
+ // Always show the asset nodes from DAG level data
+ ...dagLevelGraphData.nodes.filter((node) => node.id === `asset:${assetId}`),
+ // Conditionally show dag nodes
+ ...(showDagLevelDependencies
+ ? dagLevelGraphData.nodes.filter((node) => node.id !== `asset:${assetId}`)
+ : []),
+ // Conditionally show task nodes
+ ...(showTaskLevelDependencies ? taskLevelGraphData.nodes : []),
+ ],
+ };
Review Comment:
Sure, I've implemented this one in backend with bfs to fetch full dependency
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add data lineage graph to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2606048881
##########
airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py:
##########
@@ -76,3 +82,99 @@ def extract_single_connected_component(
]
return {"nodes": nodes, "edges": edges}
+
+
+def get_data_dependencies(asset_id: int, session: Session) -> dict[str, list[dict]]:
+ """Get full data lineage for an asset."""
+ from sqlalchemy import select
+
+ from airflow.models.asset import TaskInletAssetReference, TaskOutletAssetReference
+
+ SEPARATOR = "__SEPARATOR__"
+
+ nodes: list[dict] = []
+ edges: list[dict] = []
+ node_ids: set[str] = set()
+ edge_set: set[tuple[str, str]] = set()
+
+ # BFS to trace full lineage
+ assets_to_process: list[int] = [asset_id]
Review Comment:
Nice suggestion, that would be much better!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add data lineage graph to AssetGraph [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2690768056
##########
airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py:
##########
@@ -76,3 +82,91 @@ def extract_single_connected_component(
]
return {"nodes": nodes, "edges": edges}
+
+
+def get_data_dependencies(asset_id: int, session: Session) -> dict[str, list[dict]]:
+ """Get full data lineage for an asset."""
+ from sqlalchemy import select
+
+ from airflow.models.asset import TaskInletAssetReference, TaskOutletAssetReference
+
+ SEPARATOR = "__SEPARATOR__"
+
+ nodes_dict: dict[str, dict] = {}
+ edge_set: set[tuple[str, str]] = set()
+
+ # BFS to trace full lineage
+ assets_to_process: deque[int] = deque([asset_id])
+ processed_assets: set[int] = set()
+ processed_tasks: set[tuple[str, str]] = set() # (dag_id, task_id)
+
+ while assets_to_process:
+ current_asset_id = assets_to_process.popleft()
+ if current_asset_id in processed_assets:
+ continue
+ processed_assets.add(current_asset_id)
+
+ asset = session.get(AssetModel, current_asset_id)
Review Comment:
Here we probably want to eagerload `producing_tasks` and `comsuming_task` to not emit a lot of lazy queries
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dependencies.py:
##########
@@ -272,3 +272,35 @@ def test_with_node_id_filter_not_found(self, test_client):
assert response.json() == {
"detail": "Unique connected component not found, got [] for connected components of node missing_node_id, expected only 1 connected component.",
}
+
+ @pytest.mark.parametrize(
+ ("dependency_type", "has_dag", "has_task"),
+ [
+ (None, True, False), # default is scheduling
+ ("scheduling", True, False),
+ ("data", False, True),
+ ],
+ )
+ def test_dependency_type_filter(self, test_client, asset1_id, dependency_type, has_dag, has_task):
+ params = {"node_id": f"asset:{asset1_id}"}
+ if dependency_type is not None:
+ params["dependency_type"] = dependency_type
+
+ response = test_client.get("/dependencies", params=params)
+ assert response.status_code == 200
+
+ result = response.json()
+ node_types = {node["type"] for node in result["nodes"]}
+
+ assert "asset" in node_types
+ assert ("dag" in node_types) == has_dag
+ assert ("task" in node_types) == has_task
Review Comment:
We probably want to expand this to assert what the graph exactly look like and verify that on a complex structure, nodes are the one expected. (Here returning a graph with no dag and some tasks are enough to success the test)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add data lineage graph to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2606051457
##########
airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py:
##########
@@ -76,3 +82,99 @@ def extract_single_connected_component(
]
return {"nodes": nodes, "edges": edges}
+
+
+def get_data_dependencies(asset_id: int, session: Session) -> dict[str, list[dict]]:
+ """Get full data lineage for an asset."""
+ from sqlalchemy import select
+
+ from airflow.models.asset import TaskInletAssetReference, TaskOutletAssetReference
+
+ SEPARATOR = "__SEPARATOR__"
+
+ nodes: list[dict] = []
+ edges: list[dict] = []
+ node_ids: set[str] = set()
+ edge_set: set[tuple[str, str]] = set()
Review Comment:
Make sense to me, I've updated.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add data lineage graph to AssetGraph [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2605815263
##########
airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py:
##########
@@ -76,3 +82,99 @@ def extract_single_connected_component(
]
return {"nodes": nodes, "edges": edges}
+
+
+def get_data_dependencies(asset_id: int, session: Session) -> dict[str, list[dict]]:
+ """Get full data lineage for an asset."""
+ from sqlalchemy import select
+
+ from airflow.models.asset import TaskInletAssetReference, TaskOutletAssetReference
+
+ SEPARATOR = "__SEPARATOR__"
+
+ nodes: list[dict] = []
+ edges: list[dict] = []
+ node_ids: set[str] = set()
+ edge_set: set[tuple[str, str]] = set()
Review Comment:
It seems we just need the set and convert it back to list at the end of the function. Or we could even just returning the set and change the `edges` and `nodes` to `Iterable` instead of `list` for `BaseGraphResponse`.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/ui/dependencies.py:
##########
@@ -41,8 +46,24 @@
),
dependencies=[Depends(requires_access_dag("GET", DagAccessEntity.DEPENDENCIES))],
)
-def get_dependencies(session: SessionDep, node_id: str | None = None) -> BaseGraphResponse:
+def get_dependencies(
+ session: SessionDep,
+ node_id: str | None = None,
+ dependency_type: Literal["scheduling", "data"] = "scheduling",
+) -> BaseGraphResponse:
"""Dependencies graph."""
+ if dependency_type == "data":
+ if node_id is None or not node_id.startswith("asset:"):
+ raise HTTPException(400, "Data dependencies require an asset node_id (e.g., 'asset:123')")
+
+ try:
+ asset_id = int(node_id.replace("asset:", ""))
+ except ValueError:
+ raise HTTPException(400, f"Invalid asset node_id: {node_id}")
Review Comment:
Just to align the `HTTPException` usage with the other routers.
```suggestion
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Data dependencies require an asset node_id (e.g., 'asset:123')")
try:
asset_id = int(node_id.replace("asset:", ""))
except ValueError:
raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid asset node_id: {node_id}")
```
##########
airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py:
##########
@@ -76,3 +82,99 @@ def extract_single_connected_component(
]
return {"nodes": nodes, "edges": edges}
+
+
+def get_data_dependencies(asset_id: int, session: Session) -> dict[str, list[dict]]:
+ """Get full data lineage for an asset."""
+ from sqlalchemy import select
+
+ from airflow.models.asset import TaskInletAssetReference, TaskOutletAssetReference
+
+ SEPARATOR = "__SEPARATOR__"
+
+ nodes: list[dict] = []
+ edges: list[dict] = []
+ node_ids: set[str] = set()
+ edge_set: set[tuple[str, str]] = set()
+
+ # BFS to trace full lineage
+ assets_to_process: list[int] = [asset_id]
Review Comment:
Would it be better to leverage `deque` instead of `list` here for the BFS?
Since the time complexity of `deque.popleft` will be `O(1)` and `list.pop(0)` will be `O(n)`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add producing/consuming task dependencies to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on PR #58059:
URL: https://github.com/apache/airflow/pull/58059#issuecomment-3842234394
I am going to merge and continue to work on the follow up. Thanks for all reviews!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task dependency graph to AssetGraph [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2581249892
##########
airflow-core/src/airflow/ui/src/pages/Asset/AssetGraph.tsx:
##########
@@ -29,15 +29,42 @@ import { useGraphLayout } from "src/components/Graph/useGraphLayout";
import { useColorMode } from "src/context/colorMode";
import { useDependencyGraph } from "src/queries/useDependencyGraph";
import { getReactFlowThemeStyle } from "src/theme";
+import { getTaskLevelDependencies } from "src/utils/assetGraph";
-export const AssetGraph = ({ asset }: { readonly asset?: AssetResponse }) => {
+export const AssetGraph = ({
+ asset,
+ showDagLevelDependencies = true,
+ showTaskLevelDependencies = false,
+}: {
+ readonly asset?: AssetResponse;
+ readonly showDagLevelDependencies?: boolean;
+ readonly showTaskLevelDependencies?: boolean;
+}) => {
const { assetId } = useParams();
const { colorMode = "light" } = useColorMode();
- const { data = { edges: [], nodes: [] } } = useDependencyGraph(`asset:${assetId}`);
+ const { data: dagLevelGraphData = { edges: [], nodes: [] } } = useDependencyGraph(`asset:${assetId}`);
+ const taskLevelGraphData = getTaskLevelDependencies({ edges: [], nodes: [] }, asset);
+
+ const combinedGraphData = {
+ edges: [
+ ...(showDagLevelDependencies ? dagLevelGraphData.edges : []),
+ ...(showTaskLevelDependencies ? taskLevelGraphData.edges : []),
+ ],
+ nodes: [
+ // Always show the asset nodes from DAG level data
+ ...dagLevelGraphData.nodes.filter((node) => node.id === `asset:${assetId}`),
+ // Conditionally show dag nodes
+ ...(showDagLevelDependencies
+ ? dagLevelGraphData.nodes.filter((node) => node.id !== `asset:${assetId}`)
+ : []),
+ // Conditionally show task nodes
+ ...(showTaskLevelDependencies ? taskLevelGraphData.nodes : []),
+ ],
+ };
Review Comment:
This probably shouldn't be done in the front-end.
Because here you'll get only 1 level depth of task dependencies I think. But if we want to trace the whole `data` lineage, we probably need to call that on neighbors tasks.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add producing/consuming task dependencies to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on PR #58059:
URL: https://github.com/apache/airflow/pull/58059#issuecomment-3767105929
> But I don't see any consuming tasks, even after running asset_alias_example_alias_consumer. But i'm not sure if we do have the information at this point.
You're right. The data model doesn't currently store task-to-asset-alias relationships (only TaskInletAssetReference / TaskOutletAssetReference for actual assets). This would need new tables or querying serialized DAG data. Should we bring these in in this pr or in a follow up?
> Nit. I'm not sure we want to call that 'lineage'. That could be 'task dependencies' or something else.
Make sense to me. I've updated to use the task dependencies wording.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task dependency graph to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2602192232
##########
airflow-core/src/airflow/ui/src/pages/Asset/AssetGraph.tsx:
##########
@@ -29,15 +29,42 @@ import { useGraphLayout } from "src/components/Graph/useGraphLayout";
import { useColorMode } from "src/context/colorMode";
import { useDependencyGraph } from "src/queries/useDependencyGraph";
import { getReactFlowThemeStyle } from "src/theme";
+import { getTaskLevelDependencies } from "src/utils/assetGraph";
-export const AssetGraph = ({ asset }: { readonly asset?: AssetResponse }) => {
+export const AssetGraph = ({
+ asset,
+ showDagLevelDependencies = true,
+ showTaskLevelDependencies = false,
+}: {
+ readonly asset?: AssetResponse;
+ readonly showDagLevelDependencies?: boolean;
+ readonly showTaskLevelDependencies?: boolean;
+}) => {
const { assetId } = useParams();
const { colorMode = "light" } = useColorMode();
- const { data = { edges: [], nodes: [] } } = useDependencyGraph(`asset:${assetId}`);
+ const { data: dagLevelGraphData = { edges: [], nodes: [] } } = useDependencyGraph(`asset:${assetId}`);
+ const taskLevelGraphData = getTaskLevelDependencies({ edges: [], nodes: [] }, asset);
+
+ const combinedGraphData = {
+ edges: [
+ ...(showDagLevelDependencies ? dagLevelGraphData.edges : []),
+ ...(showTaskLevelDependencies ? taskLevelGraphData.edges : []),
+ ],
+ nodes: [
+ // Always show the asset nodes from DAG level data
+ ...dagLevelGraphData.nodes.filter((node) => node.id === `asset:${assetId}`),
+ // Conditionally show dag nodes
+ ...(showDagLevelDependencies
+ ? dagLevelGraphData.nodes.filter((node) => node.id !== `asset:${assetId}`)
+ : []),
+ // Conditionally show task nodes
+ ...(showTaskLevelDependencies ? taskLevelGraphData.nodes : []),
+ ],
+ };
Review Comment:
Sure, I think move them to ui endpoint would be better.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task dependency graph to AssetGraph [airflow]
Posted by "bbovenzi (via GitHub)" <gi...@apache.org>.
bbovenzi commented on PR #58059:
URL: https://github.com/apache/airflow/pull/58059#issuecomment-3560119600
Wait, so is this to show the tasks inside the Dag instead?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add producing/consuming task dependencies to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2744669574
##########
airflow-core/src/airflow/ui/src/components/Graph/TaskNode.tsx:
##########
@@ -90,7 +111,7 @@ export const TaskNode = ({
isGroup={isGroup}
isMapped={isMapped}
isOpen={isOpen}
- label={label}
+ label={displayLabel}
setupTeardownType={setupTeardownType}
/>
Review Comment:
Sure, that is a nice suggestion. I've updated.
<img width="3024" height="1898" alt="image" src="https://github.com/user-attachments/assets/459e85dd-6c05-4a71-8d82-ed950cfc0027" />
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add data lineage graph to AssetGraph [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2690928812
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/ui/test_dependencies.py:
##########
@@ -272,3 +272,35 @@ def test_with_node_id_filter_not_found(self, test_client):
assert response.json() == {
"detail": "Unique connected component not found, got [] for connected components of node missing_node_id, expected only 1 connected component.",
}
+
+ @pytest.mark.parametrize(
+ ("dependency_type", "has_dag", "has_task"),
+ [
+ (None, True, False), # default is scheduling
+ ("scheduling", True, False),
+ ("data", False, True),
+ ],
+ )
+ def test_dependency_type_filter(self, test_client, asset1_id, dependency_type, has_dag, has_task):
+ params = {"node_id": f"asset:{asset1_id}"}
+ if dependency_type is not None:
+ params["dependency_type"] = dependency_type
+
+ response = test_client.get("/dependencies", params=params)
+ assert response.status_code == 200
+
+ result = response.json()
+ node_types = {node["type"] for node in result["nodes"]}
+
+ assert "asset" in node_types
+ assert ("dag" in node_types) == has_dag
+ assert ("task" in node_types) == has_task
Review Comment:
Also in the UI we see
<img width="1601" height="784" alt="Screenshot 2026-01-14 at 16 28 18" src="https://github.com/user-attachments/assets/dcddf3a4-65f1-4db5-9a4a-41ad10a88299" />
But we don't know the `dag id` this task is from. There could be multiple tasks with the same name accross different dags.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add producing/consuming task dependencies to AssetGraph [airflow]
Posted by "bbovenzi (via GitHub)" <gi...@apache.org>.
bbovenzi commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2743069263
##########
airflow-core/src/airflow/ui/src/components/Graph/TaskNode.tsx:
##########
@@ -90,7 +111,7 @@ export const TaskNode = ({
isGroup={isGroup}
isMapped={isMapped}
isOpen={isOpen}
- label={label}
+ label={displayLabel}
setupTeardownType={setupTeardownType}
/>
Review Comment:
I wonder if we should add the TaskIcon when this node is used in the Asset Graph to make it clear that we're showing a task instead of a dag?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task dependency graph to AssetGraph [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2581249892
##########
airflow-core/src/airflow/ui/src/pages/Asset/AssetGraph.tsx:
##########
@@ -29,15 +29,42 @@ import { useGraphLayout } from "src/components/Graph/useGraphLayout";
import { useColorMode } from "src/context/colorMode";
import { useDependencyGraph } from "src/queries/useDependencyGraph";
import { getReactFlowThemeStyle } from "src/theme";
+import { getTaskLevelDependencies } from "src/utils/assetGraph";
-export const AssetGraph = ({ asset }: { readonly asset?: AssetResponse }) => {
+export const AssetGraph = ({
+ asset,
+ showDagLevelDependencies = true,
+ showTaskLevelDependencies = false,
+}: {
+ readonly asset?: AssetResponse;
+ readonly showDagLevelDependencies?: boolean;
+ readonly showTaskLevelDependencies?: boolean;
+}) => {
const { assetId } = useParams();
const { colorMode = "light" } = useColorMode();
- const { data = { edges: [], nodes: [] } } = useDependencyGraph(`asset:${assetId}`);
+ const { data: dagLevelGraphData = { edges: [], nodes: [] } } = useDependencyGraph(`asset:${assetId}`);
+ const taskLevelGraphData = getTaskLevelDependencies({ edges: [], nodes: [] }, asset);
+
+ const combinedGraphData = {
+ edges: [
+ ...(showDagLevelDependencies ? dagLevelGraphData.edges : []),
+ ...(showTaskLevelDependencies ? taskLevelGraphData.edges : []),
+ ],
+ nodes: [
+ // Always show the asset nodes from DAG level data
+ ...dagLevelGraphData.nodes.filter((node) => node.id === `asset:${assetId}`),
+ // Conditionally show dag nodes
+ ...(showDagLevelDependencies
+ ? dagLevelGraphData.nodes.filter((node) => node.id !== `asset:${assetId}`)
+ : []),
+ // Conditionally show task nodes
+ ...(showTaskLevelDependencies ? taskLevelGraphData.nodes : []),
+ ],
+ };
Review Comment:
This probably shouldn't be done in the front-end.
Because here you'll get only 1 level depth of task dependencies I think. But if we want to trace the whole `data` lineage, we probably need to call that on neighbors tasks as well, to see the full dependency graph of `inlets/outlets`.
##########
airflow-core/src/airflow/ui/src/pages/Asset/AssetGraph.tsx:
##########
@@ -29,15 +29,42 @@ import { useGraphLayout } from "src/components/Graph/useGraphLayout";
import { useColorMode } from "src/context/colorMode";
import { useDependencyGraph } from "src/queries/useDependencyGraph";
import { getReactFlowThemeStyle } from "src/theme";
+import { getTaskLevelDependencies } from "src/utils/assetGraph";
-export const AssetGraph = ({ asset }: { readonly asset?: AssetResponse }) => {
+export const AssetGraph = ({
+ asset,
+ showDagLevelDependencies = true,
+ showTaskLevelDependencies = false,
+}: {
+ readonly asset?: AssetResponse;
+ readonly showDagLevelDependencies?: boolean;
+ readonly showTaskLevelDependencies?: boolean;
+}) => {
const { assetId } = useParams();
const { colorMode = "light" } = useColorMode();
- const { data = { edges: [], nodes: [] } } = useDependencyGraph(`asset:${assetId}`);
+ const { data: dagLevelGraphData = { edges: [], nodes: [] } } = useDependencyGraph(`asset:${assetId}`);
+ const taskLevelGraphData = getTaskLevelDependencies({ edges: [], nodes: [] }, asset);
+
+ const combinedGraphData = {
+ edges: [
+ ...(showDagLevelDependencies ? dagLevelGraphData.edges : []),
+ ...(showTaskLevelDependencies ? taskLevelGraphData.edges : []),
+ ],
+ nodes: [
+ // Always show the asset nodes from DAG level data
+ ...dagLevelGraphData.nodes.filter((node) => node.id === `asset:${assetId}`),
+ // Conditionally show dag nodes
+ ...(showDagLevelDependencies
+ ? dagLevelGraphData.nodes.filter((node) => node.id !== `asset:${assetId}`)
+ : []),
+ // Conditionally show task nodes
+ ...(showTaskLevelDependencies ? taskLevelGraphData.nodes : []),
+ ],
+ };
Review Comment:
This probably shouldn't be done in the front-end.
Because here you'll get only 1 level depth of task dependencies I think. But if we want to trace the whole `data` lineage, we probably need to call that on neighbors tasks as well, to see the full dependency graph of `inlets/outlets` tasks
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add data lineage graph to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on PR #58059:
URL: https://github.com/apache/airflow/pull/58059#issuecomment-3631927505
I've updated the implementation to split them into two views and add the button to switch between two different graph
https://github.com/user-attachments/assets/79504e56-0dfc-48d0-8c7c-4c485532e680
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add data lineage graph to AssetGraph [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #58059:
URL: https://github.com/apache/airflow/pull/58059#discussion_r2606051457
##########
airflow-core/src/airflow/api_fastapi/core_api/services/ui/dependencies.py:
##########
@@ -76,3 +82,99 @@ def extract_single_connected_component(
]
return {"nodes": nodes, "edges": edges}
+
+
+def get_data_dependencies(asset_id: int, session: Session) -> dict[str, list[dict]]:
+ """Get full data lineage for an asset."""
+ from sqlalchemy import select
+
+ from airflow.models.asset import TaskInletAssetReference, TaskOutletAssetReference
+
+ SEPARATOR = "__SEPARATOR__"
+
+ nodes: list[dict] = []
+ edges: list[dict] = []
+ node_ids: set[str] = set()
+ edge_set: set[tuple[str, str]] = set()
Review Comment:
Thanks that make sense to me, I've updated.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org