[airflow] branch main updated: Filter dataset dependency data on webserver (#27046)

[bb...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

bbovenzi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 1c9a87e0c3 Filter dataset dependency data on webserver (#27046)
1c9a87e0c3 is described below

commit 1c9a87e0c3e792e0fcd0e53f77b2b0158f94f02e
Author: Brent Bovenzi <br...@astronomer.io>
AuthorDate: Fri Oct 14 13:34:19 2022 -0400

    Filter dataset dependency data on webserver (#27046)
    
    * filter dataset dependency data on webserver
    
    * rename u,v to source,target
---
 .../www/static/js/api/useDatasetDependencies.ts    | 13 ++----------
 airflow/www/static/js/types/index.ts               |  4 ++--
 airflow/www/views.py                               | 24 ++++++++++++----------
 3 files changed, 17 insertions(+), 24 deletions(-)

diff --git a/airflow/www/static/js/api/useDatasetDependencies.ts b/airflow/www/static/js/api/useDatasetDependencies.ts
index 104a7025ba..d7ec3260e6 100644
--- a/airflow/www/static/js/api/useDatasetDependencies.ts
+++ b/airflow/www/static/js/api/useDatasetDependencies.ts
@@ -72,7 +72,7 @@ const generateGraph = ({ nodes, edges, font }: GenerateProps) => ({
     height: 40,
     value,
   })),
-  edges: edges.map((e) => ({ id: `${e.u}-${e.v}`, sources: [e.u], targets: [e.v] })),
+  edges: edges.map((e) => ({ id: `${e.source}-${e.target}`, sources: [e.source], targets: [e.target] })),
 });
 
 const formatDependencies = async ({ edges, nodes }: DatasetDependencies) => {
@@ -81,17 +81,8 @@ const formatDependencies = async ({ edges, nodes }: DatasetDependencies) => {
   // get computed style to calculate how large each node should be
   const font = `bold ${16}px ${window.getComputedStyle(document.body).fontFamily}`;
 
-  // Make sure we only show edges that are connected to two nodes.
-  const newEdges = edges.filter((e) => {
-    const edgeNodes = nodes.filter((n) => n.id === e.u || n.id === e.v);
-    return edgeNodes.length === 2;
-  });
-
-  // Then filter out any nodes without an edge.
-  const newNodes = nodes.filter((n) => newEdges.some((e) => e.u === n.id || e.v === n.id));
-
   // Finally generate the graph data with elk
-  const data = await elk.layout(generateGraph({ nodes: newNodes, edges: newEdges, font }));
+  const data = await elk.layout(generateGraph({ nodes, edges, font }));
   return data as Data;
 };
 
diff --git a/airflow/www/static/js/types/index.ts b/airflow/www/static/js/types/index.ts
index 0eca149980..d82f76b804 100644
--- a/airflow/www/static/js/types/index.ts
+++ b/airflow/www/static/js/types/index.ts
@@ -109,8 +109,8 @@ interface DepNode {
 }
 
 interface DepEdge {
-  u: string;
-  v: string;
+  source: string;
+  target: string;
 }
 
 interface DatasetListItem extends API.Dataset {
diff --git a/airflow/www/views.py b/airflow/www/views.py
index ad7cc5104c..f514637ba1 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -3496,19 +3496,21 @@ class Airflow(AirflowBaseView):
 
         for dag, dependencies in SerializedDagModel.get_dag_dependencies().items():
             dag_node_id = f"dag:{dag}"
-            if dag_node_id not in nodes_dict:
-                nodes_dict[dag_node_id] = node_dict(dag_node_id, dag, "dag")
-
-            for dep in dependencies:
-                if dep.node_id not in nodes_dict and (
-                    dep.dependency_type == 'dag' or dep.dependency_type == 'dataset'
-                ):
-                    nodes_dict[dep.node_id] = node_dict(dep.node_id, dep.dependency_id, dep.dependency_type)
-                edge_tuples.add((f"dag:{dep.source}", dep.node_id))
-                edge_tuples.add((dep.node_id, f"dag:{dep.target}"))
+            if dag_node_id not in nodes_dict and len(dependencies) > 0:
+                for dep in dependencies:
+                    if dep.dependency_type == 'dag' or dep.dependency_type == 'dataset':
+                        nodes_dict[dag_node_id] = node_dict(dag_node_id, dag, 'dag')
+                        if dep.node_id not in nodes_dict:
+                            nodes_dict[dep.node_id] = node_dict(
+                                dep.node_id, dep.dependency_id, dep.dependency_type
+                            )
+                        if dep.source != 'dataset':
+                            edge_tuples.add((f"dag:{dep.source}", dep.node_id))
+                        if dep.target != 'dataset':
+                            edge_tuples.add((dep.node_id, f"dag:{dep.target}"))
 
         nodes = list(nodes_dict.values())
-        edges = [{"u": u, "v": v} for u, v in edge_tuples]
+        edges = [{"source": source, "target": target} for source, target in edge_tuples]
 
         data = {
             'nodes': nodes,