Skip to content

feat-use-retryable-query-executor #1941

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Sep 10, 2022
Merged

feat-use-retryable-query-executor #1941

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 34 additions & 21 deletions metadata/metadata_service/proxy/gremlin_proxy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1639,7 +1639,8 @@ def _get_dashboard_vertex(self,
:param dashboard_uri: dashboard URI that is sent from frontend
:return: Tuple of 3 String and 1 int, for dashbaord uri, url, name, and created timestamp
"""
dashboard = self.g.V().has("key", dashboard_uri).valueMap().by(__.unfold()).toList()
dashboard_query = self.g.V().has("key", dashboard_uri).valueMap().by(__.unfold())
dashboard = self.query_executor()(query=dashboard_query, get=FromResultSet.toList)
if len(dashboard) > 1:
raise Exception(f"More than one Dashboard Found with ID {dashboard_uri}.")
if len(dashboard) == 0:
Expand All @@ -1660,16 +1661,18 @@ def _get_dashboard_group_and_cluster(self,
:param dashboard_uri: dashboard URI that is sent from frontend
:return: Tuple of 4 String
"""
dashboard_group = self.g.V().has("key", dashboard_uri).out("DASHBOARD_OF").valueMap().by(__.unfold()).toList()
dashboard_group_query = self.g.V().has("key", dashboard_uri).out("DASHBOARD_OF").valueMap().by(__.unfold())
dashboard_group = self.query_executor()(query=dashboard_group_query, get=FromResultSet.toList)
if dashboard_group:
dashboard_group_name = dashboard_group[0].get("name", "")
dashboard_group_url = dashboard_group[0].get("dashboard_group_url", "")
else:
dashboard_group_name = ""
dashboard_group_url = ""

dashboard_cluster = self.g.V().has("key", dashboard_uri).out("DASHBOARD_OF")
dashboard_cluster = dashboard_cluster.out("DASHBOARD_GROUP_OF").valueMap().by(__.unfold()).toList()
dashboard_cluster_query = self.g.V().has("key", dashboard_uri).out("DASHBOARD_OF")
dashboard_cluster_query = dashboard_cluster_query.out("DASHBOARD_GROUP_OF").valueMap().by(__.unfold())
dashboard_cluster = self.query_executor()(query=dashboard_cluster_query, get=FromResultSet.toList)
if dashboard_cluster:
cluster_name = dashboard_cluster[0].get("name", "")
product_name = dashboard_cluster[0].get("key", "").split("_")[0]
Expand All @@ -1687,17 +1690,19 @@ def _get_dashboard_tables(self,
:return: List of PopularTable
"""

dashboard_tables = self.g.V().has("key", dashboard_uri).out("DASHBOARD_WITH_TABLE")
dashboard_tables = dashboard_tables.valueMap(True).by(__.unfold()).toList()
dashboard_tables_query = self.g.V().has("key", dashboard_uri).out("DASHBOARD_WITH_TABLE")
dashboard_tables_query = dashboard_tables_query.valueMap(True).by(__.unfold())
dashboard_tables = self.query_executor()(query=dashboard_tables_query, get=FromResultSet.toList)
tables = []
for table in dashboard_tables:
tabe_base = table.get('key').split("://")[1]
table_db = table.get('key').split("://")[0]
table_cluster = tabe_base.split(".")[0]
table_name = table.get('name')
table_schema = tabe_base.split("/")[0]
table_desc_vertex = self.g.V(table[T.id]).out("DESCRIPTION")
table_desc_vertex = table_desc_vertex.filter(__.hasLabel("Description")).valueMap().by(__.unfold()).toList()
table_desc_vertex_q = self.g.V(table[T.id]).out("DESCRIPTION")
table_desc_vertex_q = table_desc_vertex_q.filter(__.hasLabel("Description")).valueMap().by(__.unfold())
table_desc_vertex = self.query_executor()(query=table_desc_vertex_q, get=FromResultSet.toList)
if table_desc_vertex:
for desc in table_desc_vertex:
table_desc = desc.get('description')
Expand All @@ -1724,30 +1729,36 @@ def get_dashboard(self,
dashboard_group_name, dashboard_group_url, cluster_name, product_name = \
self._get_dashboard_group_and_cluster(dashboard_uri)

dashboard_desc = self.g.V().has("key", dashboard_uri).out("DESCRIPTION")
dashboard_desc = dashboard_desc.valueMap().by(__.unfold()).toList()
dashboard_desc_query = self.g.V().has("key", dashboard_uri).out("DESCRIPTION")
dashboard_desc_query = dashboard_desc_query.valueMap().by(__.unfold())
dashboard_desc = self.query_executor()(query=dashboard_desc_query, get=FromResultSet.toList)
if dashboard_desc:
dashboard_description = dashboard_desc[0].get("description", "")
else:
dashboard_description = ""

owners = []
dashboard_owners = self.g.V().has("key", dashboard_uri).out("READ_BY", "OWNER")
dashboard_owners = dashboard_owners.dedup().valueMap().by(__.unfold()).toList()
dashboard_owners_query = self.g.V().has("key", dashboard_uri).out("READ_BY", "OWNER")
dashboard_owners_query = dashboard_owners_query.dedup().valueMap().by(__.unfold())
dashboard_owners = self.query_executor()(query=dashboard_owners_query, get=FromResultSet.toList)
for owner in dashboard_owners:
owner_data = self._get_user_details(user_id=owner["email"], user_data=owner)
owners.append(self._build_user_from_record(record=owner_data))

dashboard_tags = self.g.V().has("key", dashboard_uri).out("TAGGED_BY").valueMap().by(__.unfold()).toList()
dashboard_tags_query = self.g.V().has("key", dashboard_uri).out("TAGGED_BY").valueMap().by(__.unfold())
dashboard_tags = self.query_executor()(query=dashboard_tags_query, get=FromResultSet.toList)
tags = [Tag(tag_type=tag['tag_type'], tag_name=tag['key']) for tag in dashboard_tags]

dashboard_badges = self.g.V().has("key", dashboard_uri).out("HAS_BADGE").valueMap().by(__.unfold()).toList()
dashboard_badges_query = self.g.V().has("key", dashboard_uri).out("HAS_BADGE").valueMap().by(__.unfold())
dashboard_badges = self.query_executor()(query=dashboard_badges_query, get=FromResultSet.toList)
badges = self._make_badges(dashboard_badges)

dashboard_charts = self.g.V().has("key", dashboard_uri).out("HAS_CHART").valueMap().by(__.unfold()).toList()
dashboard_charts_query = self.g.V().has("key", dashboard_uri).out("HAS_CHART").valueMap().by(__.unfold())
dashboard_charts = self.query_executor()(query=dashboard_charts_query, get=FromResultSet.toList)
chart = [chart['name'] for chart in dashboard_charts if 'name' in chart and chart['name']]

dashboard_query = self.g.V().has("key", dashboard_uri).out("HAS_QUERY").valueMap().by(__.unfold()).toList()
dashboard_query_query = self.g.V().has("key", dashboard_uri).out("HAS_QUERY").valueMap().by(__.unfold())
dashboard_query = self.query_executor()(query=dashboard_query_query, get=FromResultSet.toList)
query_names = [query['name'] for query in dashboard_query if 'name' in query and query['name']]
queries = []
for query in dashboard_query:
Expand All @@ -1760,8 +1771,8 @@ def get_dashboard(self,
last_successful_run_timestamp = None
last_run_timestamp = None
last_run_state = None
dashboard_execution = self.g.V().has("key", dashboard_uri).out("EXECUTED").valueMap().by(__.unfold()).toList()

dashboard_execution_query = self.g.V().has("key", dashboard_uri).out("EXECUTED").valueMap().by(__.unfold())
dashboard_execution = self.query_executor()(query=dashboard_execution_query, get=FromResultSet.toList)
for execution in dashboard_execution:
if "last_successful_execution" in execution.get("key"):
last_successful_run_timestamp = int(execution.get("timestamp"))
Expand All @@ -1770,13 +1781,15 @@ def get_dashboard(self,
last_run_state = execution.get("state")

updated_timestamp = None
dashboard_last_update = self.g.V().has("key", dashboard_uri).out("LAST_UPDATED_AT")
dashboard_last_update = dashboard_last_update.valueMap().by(__.unfold()).toList()
dashboard_last_update_query = self.g.V().has("key", dashboard_uri).out("LAST_UPDATED_AT")
dashboard_last_update_query = dashboard_last_update_query.valueMap().by(__.unfold())
dashboard_last_update = self.query_executor()(query=dashboard_last_update_query, get=FromResultSet.toList)
for last_update in dashboard_last_update:
updated_timestamp = int(last_update.get("timestamp"))

view_count = 0
dashboard_view_count = self.g.V().has("key", dashboard_uri).outE("READ_BY").valueMap().by(__.unfold()).toList()
dashboard_view_count = self.g.V().has("key", dashboard_uri).outE("READ_BY").valueMap().by(__.unfold())
dashboard_view_count = self.query_executor()(query=dashboard_view_count, get=FromResultSet.toList)
for view in dashboard_view_count:
view_count += view.get("read_count")

Expand Down