Use execution_date to check for existing DagRun for TriggerDagRunOperator

[gulshngill]

A small suggestion to change DagRun.find in trigger_dag to use execution_date as a parameter rather than run_id.

I feel it would be better to use this rather than run_id as a parameter since using run_id will miss out checking for a scheduled run that ran at the same execution_date and throw the error below when it tries to create a new run with the same execution_date:

sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "dag_run_dag_id_execution_date_key"

There is a constraint in dag_run called dag_run_dag_id_execution_date_key which can be found here.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[uranusjr]

As a side note, we are moving toward decoupling DAG runs against execution_date and removing the unique constraint eventually, and when that happens this code will become (semantically) incorrect. But until that happens, this is the correct thing to do, and wouldn't break anything even after the constraint's removal since it only adds a superfulous restriction on when DAG runs can be created, which is more appropriate than the functionality throwing exceptions right now.

Hmm, on second thought…

[uranusjr]

Actually, DagRun has a unique constraint on both execution_date and run_id. So while this fixes the duplicated execution_date issue, it'd introduce a new issue of duplicated run_id (which is possible with manual runs, where the user can customise run_id). So this should do a SQL WHERE run_id = ? OR execution_date = ? query instead. DagRun.find() does not support this, but I think the lookup logic is common enough to warrant a new method, say DagRun.find_duplicate(dag_id, run_id, execution_date) (where all three arguments are required). There is actually already at least one occurence, maybe more (I didn't search very hard).

airflow/airflow/api_connexion/endpoints/dag_run_endpoint.py

Lines 256 to 263 in 86a2a19

	dagrun_instance = (
	session.query(DagRun)
	.filter(
	DagRun.dag_id == dag_id,
	or_(DagRun.run_id == run_id, DagRun.execution_date == logical_date),
	)
	.first()
	)

[gulshngill]

Not related to this PR but noticed that this did not match the type in the docstring

[uranusjr]

Sure, why not

[gulshngill]

Hi @uranusjr , thanks for taking the time to review this PR and for the suggestion.
I've made the changes as suggested and updated the unit tests as well, I'll continue with doing the full matrix of tests

[gulshngill]

@uranusjr , I'm thinking while we're updating this file, should we also convert this function into a classmethod?

[boring-cyborg]

Awesome work, congrats on your first merged pull request!