kacpermuda commented on code in PR #73655:
URL: https://github.com/apache/airflow/pull/73655#discussion_r4258365552
##########
providers/snowflake/src/airflow/providers/snowflake/utils/openlineage.py:
##########
@@ -189,12 +138,20 @@ def _process_data_from_api(data: list[dict[str, Any]]) ->
list[dict[str, Any]]:
def _get_queries_details_from_snowflake(
- hook: SnowflakeHook | SnowflakeSqlApiHook, query_ids: list[str]
+ hook: SnowflakeHook | SnowflakeSqlApiHook,
+ query_ids: list[str],
+ task_start_time: datetime.datetime | None = None,
) -> dict[str, dict[str, Any]]:
"""Retrieve execution details for specific queries from Snowflake's query
history."""
if not query_ids:
return {}
query_condition = f"IN {tuple(query_ids)}" if len(query_ids) > 1 else f"=
'{query_ids[0]}'"
+ # QUERY_HISTORY() defaults to scanning the last 7 days
(`END_TIME_RANGE_START`/`END_TIME_RANGE_END`
+ # default to now / now - 10080 minutes), which is a needlessly expensive
account-wide scan when we
+ # already know roughly when these queries ran. Bound it to the task's own
execution window instead,
+ # with a small buffer on both sides for clock skew between Airflow and
Snowflake.
+ range_start = (task_start_time or timezone.utcnow()) -
datetime.timedelta(minutes=5)
+ range_end = timezone.utcnow() + datetime.timedelta(minutes=5)
Review Comment:
Make sense, thanks, removed the range_end entirely
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
