You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dstandish (via GitHub)" <gi...@apache.org> on 2025/05/22 19:26:04 UTC
[PR] Fix get dags query to not have join explosion [airflow]
dstandish opened a new pull request, #50984:
URL: https://github.com/apache/airflow/pull/50984
Previously it was missing dag_id filter, but joining on start date would still be problematic. In this PR I refactor the query a bit so that all joins are guaranteed 1-1. To get "latest" DagRun I sort by the DagRun.id column. This is a simplifying assumption that would be more performant than sorting by start_date, since there could be more than 1 dag run with a given start date.
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2110368454
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
+ .where(DagRun.start_date.is_not(null()))
+ .group_by(DagRun.dag_id)
+ .subquery(name="mrq")
Review Comment:
Actually when you name the subquery, it is used in the joins and it makes the sql more readable. If you do _not_ provide a name, you see `anon_1` etc, which, is not very good. Further it's another clue when tracing a query back from the database to the python code.
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2110382381
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
Review Comment:
Yes, it is possible to write such a query, but it would be expensive. This is a simplification that would generally be true. It would always show the latest created run. But if you cleared an old dag run, still the latest created dag run.
So yeah we could do the more complicated query, but to me, it doesn't really seem worth it. What do you think. @jedcunningham ?
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on PR #50984:
URL: https://github.com/apache/airflow/pull/50984#issuecomment-2914270383
> I didn't dive too deep, but `generate_dag_with_latest_run_query` is also using a `latest_dag_run_per_dag_id_cte` that seems worong, most likely also has the problem (joining over dag_run latest date, but actually multiple run with the same start date can happen, resulting in multiple rows per Dag which is not handled).
>
> ```python
> latest_dag_run_per_dag_id_cte = (
> select(DagRun.dag_id, func.max(DagRun.start_date).label("start_date"))
> .where()
> .group_by(DagRun.dag_id)
> .cte()
> )
> ```
Right this function is wrong -- but I am not touching that one here -- one at a time.
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2103838546
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py:
##########
@@ -220,10 +220,10 @@ class TestGetDags(TestDagEndpoint):
"dag_run_state": "failed",
"exclude_stale": False,
},
- 1,
- [DAG3_ID],
+ 0,
+ [],
Review Comment:
Why does this change?
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2104534039
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
+ .where(DagRun.start_date.is_not(null()))
+ .group_by(DagRun.dag_id)
+ .subquery(name="mrq")
+ )
+
+ has_max_run_filter = (
+ dag_run_state.value
+ or last_dag_run_state.value
+ or dag_run_start_date_range.is_active()
+ or dag_run_end_date_range.is_active()
+ )
+
+ if has_max_run_filter or order_by.value in (
+ "last_run_state",
+ "last_run_start_date",
+ "-last_run_state",
+ "-last_run_start_date",
+ ):
+ query = query.join(
+ max_run_id_query,
+ DagModel.dag_id == max_run_id_query.c.dag_id,
+ isouter=True,
+ ).join(DagRun, DagRun.id == max_run_id_query.c.max_dag_run_id, isouter=True)
Review Comment:
That's a nice optimization, but I think it's really minor, at a cost of maintainability. This hardcode a bunch of filters, and I'm sure in a bit when we edit this endpoint a few times, we will end up with filters missing in here.
But that's not a big deal, I'm ok merging this.
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py:
##########
@@ -220,10 +220,10 @@ class TestGetDags(TestDagEndpoint):
"dag_run_state": "failed",
"exclude_stale": False,
},
- 1,
- [DAG3_ID],
+ 0,
+ [],
Review Comment:
+1
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
Review Comment:
I would love to avoid doing this, we can add an extra column on an aggregated `DagRun.id`, for instance keep it on `func.max(DagRun.start_date)` and add `func.max(DagRun.id)`. This way when we join, we do that both on the `start_date` + `DagRun.id` + `Dag.id`. This will ensure only 1 - 1 row matching.
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2107016770
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py:
##########
@@ -220,10 +220,10 @@ class TestGetDags(TestDagEndpoint):
"dag_run_state": "failed",
"exclude_stale": False,
},
- 1,
- [DAG3_ID],
+ 0,
+ [],
Review Comment:
Also ideally we would want a test that creates the join explosion to assert that it's not happening again. (something with multiple dag_run with the same start_date)
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2104827807
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py:
##########
@@ -220,10 +220,10 @@ class TestGetDags(TestDagEndpoint):
"dag_run_state": "failed",
"exclude_stale": False,
},
- 1,
- [DAG3_ID],
+ 0,
+ [],
Review Comment:
The reason is that, dag_run_state behavior was a bit nonsensical, as I mentioned on slack. I will be removing dag_run_state in followup so this is moot.
But the answer is, it meant "list me the dags that have _any_ dag run in state X". The reason it changed in the test, is for now it means same as last dag run state.
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2110437104
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
Review Comment:
So, this is the query for getting the latest dag run by start_date
```
SELECT
dag.dag_display_name,
dag.dag_id,
dag.is_paused,
dag.is_stale,
dag.last_parsed_time,
dag.last_expired,
dag.fileloc,
dag.relative_fileloc,
dag.bundle_name,
dag.bundle_version,
dag.owners,
dag.description,
dag.timetable_summary,
dag.timetable_description,
dag.asset_expression,
dag.deadline,
dag.max_active_tasks,
dag.max_active_runs,
dag.max_consecutive_failed_dag_runs,
dag.has_task_concurrency_limits,
dag.has_import_errors,
dag.next_dagrun,
dag.next_dagrun_data_interval_start,
dag.next_dagrun_data_interval_end,
dag.next_dagrun_create_after
FROM
dag
LEFT OUTER JOIN (SELECT
mrq_inner.dag_id AS dag_id,
mrq_inner.max_run_id AS max_run_id
FROM
(SELECT
dag_run.id AS max_run_id,
dag_run.dag_id AS dag_id,
row_number() OVER (PARTITION BY dag_run.dag_id ORDER BY dag_run.start_date DESC) AS rn
FROM dag_run
WHERE
dag_run.start_date IS NOT NULL
) AS mrq_inner
WHERE
mrq_inner.rn = % (rn_1) s
) AS mrq ON dag.dag_id = mrq.dag_id
LEFT OUTER JOIN dag_run ON dag_run.id = mrq.max_run_id
```
We could do it but, not sure how well it will perform
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2111471169
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
Review Comment:
We can go like this and improve later, 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] Fix get dags query to not have join explosion [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #50984:
URL: https://github.com/apache/airflow/pull/50984#issuecomment-2917159802
### Backport successfully created: v3-0-test
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>✅</td>
<td>v3-0-test</td>
<td><a href="https://github.com/apache/airflow/pull/51172"><img src="https://img.shields.io/badge/PR-51172-blue" alt="PR Link"></a></td>
</tr>
</table>
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2104862388
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
+ .where(DagRun.start_date.is_not(null()))
+ .group_by(DagRun.dag_id)
+ .subquery(name="mrq")
+ )
+
+ has_max_run_filter = (
+ dag_run_state.value
+ or last_dag_run_state.value
+ or dag_run_start_date_range.is_active()
+ or dag_run_end_date_range.is_active()
+ )
+
+ if has_max_run_filter or order_by.value in (
+ "last_run_state",
+ "last_run_start_date",
+ "-last_run_state",
+ "-last_run_start_date",
+ ):
+ query = query.join(
+ max_run_id_query,
+ DagModel.dag_id == max_run_id_query.c.dag_id,
+ isouter=True,
+ ).join(DagRun, DagRun.id == max_run_id_query.c.max_dag_run_id, isouter=True)
Review Comment:
oh you mean just leave the group by even when not used? yeah think that's prohibitive from a performance perspective. this endpoint is run a _lot_. and anyway, if one is missed, it should be caught by tests -- it will try and sort by a reference that is not there.
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2104857959
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
+ .where(DagRun.start_date.is_not(null()))
+ .group_by(DagRun.dag_id)
+ .subquery(name="mrq")
+ )
+
+ has_max_run_filter = (
+ dag_run_state.value
+ or last_dag_run_state.value
+ or dag_run_start_date_range.is_active()
+ or dag_run_end_date_range.is_active()
+ )
+
+ if has_max_run_filter or order_by.value in (
+ "last_run_state",
+ "last_run_start_date",
+ "-last_run_state",
+ "-last_run_start_date",
+ ):
+ query = query.join(
+ max_run_id_query,
+ DagModel.dag_id == max_run_id_query.c.dag_id,
+ isouter=True,
+ ).join(DagRun, DagRun.id == max_run_id_query.c.max_dag_run_id, isouter=True)
Review Comment:
what's the alternative?
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2106984039
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
Review Comment:
We shouldn't have to do this because it can yield wrong results. We are capable of emitting a query that will select 1 Run per dag_id, the one that has the max `start_date` and in case of multiple rows for the same `dag_id` and same max `start_date` will then choose the single row with the `max_dag_run_id` as a second criteria. (maybe with a window function over the partition of dagrun with the latest_start date or two nested subqueries)
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2110437104
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
Review Comment:
So, this is the query for getting the latest dag run by start_date
```
SELECT
dag.dag_display_name,
dag.dag_id,
dag.is_paused,
dag.is_stale,
dag.last_parsed_time,
dag.last_expired,
dag.fileloc,
dag.relative_fileloc,
dag.bundle_name,
dag.bundle_version,
dag.owners,
dag.description,
dag.timetable_summary,
dag.timetable_description,
dag.asset_expression,
dag.deadline,
dag.max_active_tasks,
dag.max_active_runs,
dag.max_consecutive_failed_dag_runs,
dag.has_task_concurrency_limits,
dag.has_import_errors,
dag.next_dagrun,
dag.next_dagrun_data_interval_start,
dag.next_dagrun_data_interval_end,
dag.next_dagrun_create_after
FROM
dag
LEFT OUTER JOIN (SELECT
mrq_inner.dag_id AS dag_id,
mrq_inner.max_run_id AS max_run_id
FROM
(SELECT
dag_run.id AS max_run_id,
dag_run.dag_id AS dag_id,
row_number() OVER (PARTITION BY dag_run.dag_id ORDER BY dag_run.start_date DESC) AS rn
FROM dag_run
WHERE
dag_run.start_date IS NOT NULL
) AS mrq_inner
WHERE
mrq_inner.rn = 1
) AS mrq ON dag.dag_id = mrq.dag_id
LEFT OUTER JOIN dag_run ON dag_run.id = mrq.max_run_id
```
We could do it but, not sure how well it will perform
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2106984039
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
Review Comment:
We shouldn't have to do this because it can yield wrong results. We are capable of emitting a query that will select 1 Run per dag_id, the one that has the max `start_date` and in case of multiple rows for the same `dag_id` and same max `start_date` will then choose the single row with the `max_dag_run_id` as a second criteria.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
+ .where(DagRun.start_date.is_not(null()))
+ .group_by(DagRun.dag_id)
+ .subquery(name="mrq")
+ )
+
+ has_max_run_filter = (
+ dag_run_state.value
+ or last_dag_run_state.value
+ or dag_run_start_date_range.is_active()
+ or dag_run_end_date_range.is_active()
+ )
+
+ if has_max_run_filter or order_by.value in (
+ "last_run_state",
+ "last_run_start_date",
+ "-last_run_state",
+ "-last_run_start_date",
+ ):
+ query = query.join(
+ max_run_id_query,
+ DagModel.dag_id == max_run_id_query.c.dag_id,
+ isouter=True,
+ ).join(DagRun, DagRun.id == max_run_id_query.c.max_dag_run_id, isouter=True)
Review Comment:
I mean always do the join without condition.
But disregard my comment that's not important.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
+ .where(DagRun.start_date.is_not(null()))
+ .group_by(DagRun.dag_id)
+ .subquery(name="mrq")
Review Comment:
The subquery name doesn't seem used. Also just from a db debugging perspective (inspected the output sql statement)`mrq` without context doesn't mean anything.
I think we should remove it.
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2110430718
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
Review Comment:
I think a "most of the time right, always fast and simple" answer is a good one here - and much better than a join explosion. A follow up to make it more "right" is always welcome, but we can look at that complexity and maintenance and performance then.
This is less bad than reusing dagruns, ultimately :) that'd be my vote for the long term fix.
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2104862388
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
+ .where(DagRun.start_date.is_not(null()))
+ .group_by(DagRun.dag_id)
+ .subquery(name="mrq")
+ )
+
+ has_max_run_filter = (
+ dag_run_state.value
+ or last_dag_run_state.value
+ or dag_run_start_date_range.is_active()
+ or dag_run_end_date_range.is_active()
+ )
+
+ if has_max_run_filter or order_by.value in (
+ "last_run_state",
+ "last_run_start_date",
+ "-last_run_state",
+ "-last_run_start_date",
+ ):
+ query = query.join(
+ max_run_id_query,
+ DagModel.dag_id == max_run_id_query.c.dag_id,
+ isouter=True,
+ ).join(DagRun, DagRun.id == max_run_id_query.c.max_dag_run_id, isouter=True)
Review Comment:
oh you mean just leave the group by every time? yeah think that's prohibitive from a performance perspective. this endpoint is run a _lot_. and anyway, if one is missed, it should be caught by tests -- it will try and sort by a reference that is not there.
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2106939610
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
+ .where(DagRun.start_date.is_not(null()))
+ .group_by(DagRun.dag_id)
+ .subquery(name="mrq")
Review Comment:
The subquery name doesn't seem used. Also just from a db debugging perspective (inspect the output sql statement)`mrq` without context doesn't mean anything.
I think we should remove it.
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
+ .where(DagRun.start_date.is_not(null()))
+ .group_by(DagRun.dag_id)
+ .subquery(name="mrq")
Review Comment:
nit: The subquery name doesn't seem used. Also just from a db debugging perspective (inspect the output sql statement)`mrq` without context doesn't mean anything.
I think we should remove it.
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2104856182
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
Review Comment:
i'm not following.
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish merged PR #50984:
URL: https://github.com/apache/airflow/pull/50984
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on PR #50984:
URL: https://github.com/apache/airflow/pull/50984#issuecomment-2914266370
@pierrejeambrun re
> We shouldn't have to do this because it can yield wrong results. We are capable of emitting a query that will select 1 Run per dag_id, the one that has the max start_date and in case of multiple rows for the same dag_id and same max start_date will then choose the single row with the max_dag_run_id as a second criteria. (maybe with a window function over the partition of dagrun with the latest_start date or two nested subqueries)
Yes, it is _possible_ to write such a query, but it would be expensive. This is a simplification that would generally be true. It would always show the latest created run. But if you cleared an old dag run, still the latest created dag run.
So yeah we could do the more complicated query, but to me, it doesn't really seem worth it. What do you think. @jedcunningham ?
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2110387978
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py:
##########
@@ -115,30 +116,54 @@ def get_dags(
session: SessionDep,
) -> DAGCollectionResponse:
"""Get all DAGs."""
- dag_runs_select = None
+ query = select(DagModel)
- if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
- dag_runs_select, _ = paginated_select(
- statement=select(DagRun),
+ max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
+ select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
Review Comment:
What do you think about this @pierrejeambrun .
For now let's go with just max run id since it should be performant and it's simple and will in most cases give the same answer / be good enough.
Then, if desired, at some point in the future we can add a column to dag model like `last_run` or something, and it can be updated whenever a dag run starts, thus avoiding this expensive and complicated query? we would not even need to join to dag run table.
--
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] Fix get dags query to not have join explosion [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #50984:
URL: https://github.com/apache/airflow/pull/50984#discussion_r2106923874
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py:
##########
@@ -220,10 +220,10 @@ class TestGetDags(TestDagEndpoint):
"dag_run_state": "failed",
"exclude_stale": False,
},
- 1,
- [DAG3_ID],
+ 0,
+ [],
Review Comment:
Oh ok, so follow up PR then, got it
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py:
##########
@@ -220,10 +220,10 @@ class TestGetDags(TestDagEndpoint):
"dag_run_state": "failed",
"exclude_stale": False,
},
- 1,
- [DAG3_ID],
+ 0,
+ [],
Review Comment:
Follow up PR then, got it
--
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