Fixed issue of new dag getting old dataset events. #39603
Conversation
|
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 Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
|
tests/jobs/test_scheduler_job.py
Outdated
| @@ -3776,6 +3776,11 @@ def test_create_dag_runs_datasets(self, session, dag_maker): | |||
| dataset1 = Dataset(uri="ds1") | |||
| dataset2 = Dataset(uri="ds2") | |||
|
|
|||
| # Create DAG before the arrival of dataset events. | |||
There was a problem hiding this comment.
Should this be part of its own, new test?
test_new_dagrun_ignores_old_datasets or similar?
There was a problem hiding this comment.
@RNHTTR I've added a new unit test. Please review it and let me know if it needs any adjustments.
Unit test is checking newly added Dag should just get the Dataset arrived after it was created(event3) and not the dataset events(event 1 and event2)
tests/jobs/test_scheduler_job.py
Outdated
| @@ -3811,20 +3816,40 @@ def test_create_dag_runs_datasets(self, session, dag_maker): | |||
| ) | |||
| session.add(event2) | |||
|
|
|||
| # Create a third event, creation time is more recent, but data interval is even older | |||
There was a problem hiding this comment.
Given that this is reproducible with a single, previous DAG run having a dataset event, shouldn't the test only require 3 total DAG runs?
- One DAG run from the original DAG (let's call this Run A)
- One more DAG run (Run B) after the new DAG is introduced, which triggers the new DAG's DAG run which should only have dataset info from Run B.
There was a problem hiding this comment.
The newly designed test case perfectly addresses the scenario you have mentioned. Let me know if i missing anything.
|
cc @uranusjr |
|
To be honest I don’t really consider the current behaviour (in 2.9.1) a bug, but I can see why people expect it this way. I think we probably need an entry in |
|
Also the tests need improvements, as mentioned in above reviews. |
airflow/jobs/scheduler_job_runner.py
Outdated
| if previous_dag_run: | ||
| dataset_event_filters.append(DatasetEvent.timestamp > previous_dag_run.execution_date) | ||
| else: | ||
| dataset_event_filters.append( | ||
| DatasetEvent.timestamp >= DagScheduleDatasetReference.created_at | ||
| ) |
There was a problem hiding this comment.
Hmm just wondering if the difference between >= and > is something we should worry about.
If a new dataset triggreed DAG is created for an already existing dataset. (Dataset has already existing dataset events) DAG see all dataset events from very first event for dataset.
Fixes: #39456