You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "prabhusneha (via GitHub)" <gi...@apache.org> on 2025/01/29 16:18:53 UTC
[PR] AIP-83 amendment -- update backfill [airflow]
prabhusneha opened a new pull request, #46248:
URL: https://github.com/apache/airflow/pull/46248
Closes: [#46202](https://github.com/apache/airflow/issues/46202)
Once uniqueness on logical date is restored, the change in backfill logic handles the removal of locking of dagrun that was added to avoid the creation of multiple concurrent dag runs for a given logical 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] AIP-83 amendment -- update backfill [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1952761037
##########
airflow/models/backfill.py:
##########
@@ -263,57 +286,54 @@ def _create_backfill_dag_run(
backfill_sort_ordinal,
session,
):
- with session.begin_nested() as nested:
- dr = session.scalar(
- with_row_locks(
- query=_get_latest_dag_run_row_query(info, session),
- session=session,
- ),
+ with session.begin_nested():
+ should_skip_create_backfill = should_create_backfill_dag_run(
+ info, reprocess_behavior, backfill_id, backfill_sort_ordinal, session
)
- if dr:
- non_create_reason = _get_dag_run_no_create_reason(dr, reprocess_behavior)
- if non_create_reason:
- # rolling back here restores to start of this nested tran
- # which releases the lock on the latest dag run, since we
- # are not creating a new one
- nested.rollback()
- session.add(
- BackfillDagRun(
- backfill_id=backfill_id,
- dag_run_id=None,
- logical_date=info.logical_date,
- exception_reason=non_create_reason,
- sort_ordinal=backfill_sort_ordinal,
- )
- )
- return
+ if should_skip_create_backfill:
+ return
+
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
- dr = dag.create_dagrun(
- run_id=dag.timetable.generate_run_id(
- run_type=DagRunType.BACKFILL_JOB,
+ try:
+ dr = dag.create_dagrun(
+ run_id=dag.timetable.generate_run_id(
+ run_type=DagRunType.BACKFILL_JOB,
+ logical_date=info.logical_date,
+ data_interval=info.data_interval,
+ ),
logical_date=info.logical_date,
data_interval=info.data_interval,
- ),
- logical_date=info.logical_date,
- data_interval=info.data_interval,
- run_after=info.run_after,
- conf=dag_run_conf,
- run_type=DagRunType.BACKFILL_JOB,
- triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
- state=DagRunState.QUEUED,
- start_date=timezone.utcnow(),
- backfill_id=backfill_id,
- session=session,
- )
- session.add(
- BackfillDagRun(
+ run_after=info.run_after,
+ conf=dag_run_conf,
+ run_type=DagRunType.BACKFILL_JOB,
+ triggered_by=DagRunTriggeredByType.BACKFILL,
+ dag_version=dag_version,
+ state=DagRunState.QUEUED,
+ start_date=timezone.utcnow(),
backfill_id=backfill_id,
- dag_run_id=dr.id,
- sort_ordinal=backfill_sort_ordinal,
- logical_date=info.logical_date,
+ session=session,
+ )
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill_id,
+ dag_run_id=dr.id,
+ sort_ordinal=backfill_sort_ordinal,
+ logical_date=info.logical_date,
+ )
+ )
+ except IntegrityError:
+ log.info(
+ "Skipped creating backfill dag run for dag_id=%s backfill_id=%s, logical_date=%s (already exists)",
+ dr.dag_id,
+ dr.id,
+ info.logical_date,
+ )
+ log.info("Doing session rollback.")
+ session.rollback()
+
+ should_create_backfill_dag_run(
Review Comment:
Here `should_create_backfill_dag_run` is called to create backfill dag run record if the insert fails.
Created a separate function for this to reuse in try and error handling
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1965244888
##########
airflow/models/backfill.py:
##########
@@ -263,57 +286,54 @@ def _create_backfill_dag_run(
backfill_sort_ordinal,
session,
):
- with session.begin_nested() as nested:
- dr = session.scalar(
- with_row_locks(
- query=_get_latest_dag_run_row_query(info, session),
- session=session,
- ),
+ with session.begin_nested():
+ should_skip_create_backfill = should_create_backfill_dag_run(
+ info, reprocess_behavior, backfill_id, backfill_sort_ordinal, session
)
- if dr:
- non_create_reason = _get_dag_run_no_create_reason(dr, reprocess_behavior)
- if non_create_reason:
- # rolling back here restores to start of this nested tran
- # which releases the lock on the latest dag run, since we
- # are not creating a new one
- nested.rollback()
- session.add(
- BackfillDagRun(
- backfill_id=backfill_id,
- dag_run_id=None,
- logical_date=info.logical_date,
- exception_reason=non_create_reason,
- sort_ordinal=backfill_sort_ordinal,
- )
- )
- return
+ if should_skip_create_backfill:
+ return
+
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
- dr = dag.create_dagrun(
- run_id=dag.timetable.generate_run_id(
- run_type=DagRunType.BACKFILL_JOB,
+ try:
+ dr = dag.create_dagrun(
+ run_id=dag.timetable.generate_run_id(
+ run_type=DagRunType.BACKFILL_JOB,
+ logical_date=info.logical_date,
+ data_interval=info.data_interval,
+ ),
logical_date=info.logical_date,
data_interval=info.data_interval,
- ),
- logical_date=info.logical_date,
- data_interval=info.data_interval,
- run_after=info.run_after,
- conf=dag_run_conf,
- run_type=DagRunType.BACKFILL_JOB,
- triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
- state=DagRunState.QUEUED,
- start_date=timezone.utcnow(),
- backfill_id=backfill_id,
- session=session,
- )
- session.add(
- BackfillDagRun(
+ run_after=info.run_after,
+ conf=dag_run_conf,
+ run_type=DagRunType.BACKFILL_JOB,
+ triggered_by=DagRunTriggeredByType.BACKFILL,
+ dag_version=dag_version,
+ state=DagRunState.QUEUED,
+ start_date=timezone.utcnow(),
backfill_id=backfill_id,
- dag_run_id=dr.id,
- sort_ordinal=backfill_sort_ordinal,
- logical_date=info.logical_date,
+ session=session,
+ )
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill_id,
+ dag_run_id=dr.id,
+ sort_ordinal=backfill_sort_ordinal,
+ logical_date=info.logical_date,
+ )
+ )
+ except IntegrityError:
+ log.info(
+ "Skipped creating backfill dag run for dag_id=%s backfill_id=%s, logical_date=%s (already exists)",
+ dr.dag_id,
+ dr.id,
+ info.logical_date,
+ )
+ log.info("Doing session rollback.")
+ session.rollback()
+
+ should_create_backfill_dag_run(
Review Comment:
yeah the problem here is in part naming.
a function called "should_create_x" reads like it will just give you information. so you would not expect it to be creating records, or to depend on it to create a record.
otherwise it's confusing. you could either get rid of this method, or limit it to only give info (e.g. maybe just return `BackfillDagRunExceptionReason | None` and do the insert conditionally at call site)
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1952725320
##########
airflow/models/backfill.py:
##########
@@ -263,57 +286,54 @@ def _create_backfill_dag_run(
backfill_sort_ordinal,
session,
):
- with session.begin_nested() as nested:
- dr = session.scalar(
- with_row_locks(
- query=_get_latest_dag_run_row_query(info, session),
- session=session,
- ),
+ with session.begin_nested():
+ should_skip_create_backfill = should_create_backfill_dag_run(
+ info, reprocess_behavior, backfill_id, backfill_sort_ordinal, session
)
- if dr:
- non_create_reason = _get_dag_run_no_create_reason(dr, reprocess_behavior)
- if non_create_reason:
- # rolling back here restores to start of this nested tran
- # which releases the lock on the latest dag run, since we
- # are not creating a new one
- nested.rollback()
- session.add(
- BackfillDagRun(
- backfill_id=backfill_id,
- dag_run_id=None,
- logical_date=info.logical_date,
- exception_reason=non_create_reason,
- sort_ordinal=backfill_sort_ordinal,
- )
- )
- return
+ if should_skip_create_backfill:
+ return
+
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
- dr = dag.create_dagrun(
- run_id=dag.timetable.generate_run_id(
- run_type=DagRunType.BACKFILL_JOB,
+ try:
+ dr = dag.create_dagrun(
+ run_id=dag.timetable.generate_run_id(
+ run_type=DagRunType.BACKFILL_JOB,
+ logical_date=info.logical_date,
+ data_interval=info.data_interval,
+ ),
logical_date=info.logical_date,
data_interval=info.data_interval,
- ),
- logical_date=info.logical_date,
- data_interval=info.data_interval,
- run_after=info.run_after,
- conf=dag_run_conf,
- run_type=DagRunType.BACKFILL_JOB,
- triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
- state=DagRunState.QUEUED,
- start_date=timezone.utcnow(),
- backfill_id=backfill_id,
- session=session,
- )
- session.add(
- BackfillDagRun(
+ run_after=info.run_after,
+ conf=dag_run_conf,
+ run_type=DagRunType.BACKFILL_JOB,
+ triggered_by=DagRunTriggeredByType.BACKFILL,
+ dag_version=dag_version,
+ state=DagRunState.QUEUED,
+ start_date=timezone.utcnow(),
backfill_id=backfill_id,
- dag_run_id=dr.id,
- sort_ordinal=backfill_sort_ordinal,
- logical_date=info.logical_date,
+ session=session,
+ )
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill_id,
+ dag_run_id=dr.id,
+ sort_ordinal=backfill_sort_ordinal,
+ logical_date=info.logical_date,
+ )
+ )
+ except IntegrityError:
+ log.info(
+ "Skipped creating backfill dag run for dag_id=%s backfill_id=%s, logical_date=%s (already exists)",
+ dr.dag_id,
+ dr.id,
+ info.logical_date,
+ )
+ log.info("Doing session rollback.")
+ session.rollback()
+
+ should_create_backfill_dag_run(
Review Comment:
do you know why we are calling `should_create_backfill_dag_run` here?
this seems might be a mistake?
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1953884418
##########
airflow/models/backfill.py:
##########
@@ -263,57 +286,54 @@ def _create_backfill_dag_run(
backfill_sort_ordinal,
session,
):
- with session.begin_nested() as nested:
- dr = session.scalar(
- with_row_locks(
- query=_get_latest_dag_run_row_query(info, session),
- session=session,
- ),
+ with session.begin_nested():
+ should_skip_create_backfill = should_create_backfill_dag_run(
+ info, reprocess_behavior, backfill_id, backfill_sort_ordinal, session
)
- if dr:
- non_create_reason = _get_dag_run_no_create_reason(dr, reprocess_behavior)
- if non_create_reason:
- # rolling back here restores to start of this nested tran
- # which releases the lock on the latest dag run, since we
- # are not creating a new one
- nested.rollback()
- session.add(
- BackfillDagRun(
- backfill_id=backfill_id,
- dag_run_id=None,
- logical_date=info.logical_date,
- exception_reason=non_create_reason,
- sort_ordinal=backfill_sort_ordinal,
- )
- )
- return
+ if should_skip_create_backfill:
+ return
+
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
- dr = dag.create_dagrun(
- run_id=dag.timetable.generate_run_id(
- run_type=DagRunType.BACKFILL_JOB,
+ try:
+ dr = dag.create_dagrun(
+ run_id=dag.timetable.generate_run_id(
+ run_type=DagRunType.BACKFILL_JOB,
+ logical_date=info.logical_date,
+ data_interval=info.data_interval,
+ ),
logical_date=info.logical_date,
data_interval=info.data_interval,
- ),
- logical_date=info.logical_date,
- data_interval=info.data_interval,
- run_after=info.run_after,
- conf=dag_run_conf,
- run_type=DagRunType.BACKFILL_JOB,
- triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
- state=DagRunState.QUEUED,
- start_date=timezone.utcnow(),
- backfill_id=backfill_id,
- session=session,
- )
- session.add(
- BackfillDagRun(
+ run_after=info.run_after,
+ conf=dag_run_conf,
+ run_type=DagRunType.BACKFILL_JOB,
+ triggered_by=DagRunTriggeredByType.BACKFILL,
+ dag_version=dag_version,
+ state=DagRunState.QUEUED,
+ start_date=timezone.utcnow(),
backfill_id=backfill_id,
- dag_run_id=dr.id,
- sort_ordinal=backfill_sort_ordinal,
- logical_date=info.logical_date,
+ session=session,
+ )
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill_id,
+ dag_run_id=dr.id,
+ sort_ordinal=backfill_sort_ordinal,
+ logical_date=info.logical_date,
+ )
+ )
+ except IntegrityError:
+ log.info(
+ "Skipped creating backfill dag run for dag_id=%s backfill_id=%s, logical_date=%s (already exists)",
+ dr.dag_id,
+ dr.id,
+ info.logical_date,
+ )
+ log.info("Doing session rollback.")
+ session.rollback()
+
+ should_create_backfill_dag_run(
Review Comment:
So the logic is (simplified)
1. Check if there’s an existing run
2. If so, return
3. Otherwise, try to create a new run
4. If that succeeds, good.
5. Otherwise, race condition? Go to `should_create_backfill_dag_run` again; in this case, `_get_latest_dag_run_row_query` should return that conflicting run instead, and we only insert a new BackfillDagRun without a new DagRun.
It might be better if it’s clearer that the DagRun fetched in the second `should_create_backfill_dag_run` call should be a different one, but I’m not exactly sure how. Maybe call `_get_latest_dag_run_row_query` in `_create_backfill_dag_run` instead? Not sure.
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1953834113
##########
airflow/models/backfill.py:
##########
@@ -263,57 +286,54 @@ def _create_backfill_dag_run(
backfill_sort_ordinal,
session,
):
- with session.begin_nested() as nested:
- dr = session.scalar(
- with_row_locks(
- query=_get_latest_dag_run_row_query(info, session),
- session=session,
- ),
+ with session.begin_nested():
+ should_skip_create_backfill = should_create_backfill_dag_run(
+ info, reprocess_behavior, backfill_id, backfill_sort_ordinal, session
)
- if dr:
- non_create_reason = _get_dag_run_no_create_reason(dr, reprocess_behavior)
- if non_create_reason:
- # rolling back here restores to start of this nested tran
- # which releases the lock on the latest dag run, since we
- # are not creating a new one
- nested.rollback()
- session.add(
- BackfillDagRun(
- backfill_id=backfill_id,
- dag_run_id=None,
- logical_date=info.logical_date,
- exception_reason=non_create_reason,
- sort_ordinal=backfill_sort_ordinal,
- )
- )
- return
+ if should_skip_create_backfill:
+ return
+
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
- dr = dag.create_dagrun(
- run_id=dag.timetable.generate_run_id(
- run_type=DagRunType.BACKFILL_JOB,
+ try:
+ dr = dag.create_dagrun(
+ run_id=dag.timetable.generate_run_id(
+ run_type=DagRunType.BACKFILL_JOB,
+ logical_date=info.logical_date,
+ data_interval=info.data_interval,
+ ),
logical_date=info.logical_date,
data_interval=info.data_interval,
- ),
- logical_date=info.logical_date,
- data_interval=info.data_interval,
- run_after=info.run_after,
- conf=dag_run_conf,
- run_type=DagRunType.BACKFILL_JOB,
- triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
- state=DagRunState.QUEUED,
- start_date=timezone.utcnow(),
- backfill_id=backfill_id,
- session=session,
- )
- session.add(
- BackfillDagRun(
+ run_after=info.run_after,
+ conf=dag_run_conf,
+ run_type=DagRunType.BACKFILL_JOB,
+ triggered_by=DagRunTriggeredByType.BACKFILL,
+ dag_version=dag_version,
+ state=DagRunState.QUEUED,
+ start_date=timezone.utcnow(),
backfill_id=backfill_id,
- dag_run_id=dr.id,
- sort_ordinal=backfill_sort_ordinal,
- logical_date=info.logical_date,
+ session=session,
+ )
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill_id,
+ dag_run_id=dr.id,
+ sort_ordinal=backfill_sort_ordinal,
+ logical_date=info.logical_date,
+ )
+ )
+ except IntegrityError:
+ log.info(
+ "Skipped creating backfill dag run for dag_id=%s backfill_id=%s, logical_date=%s (already exists)",
+ dr.dag_id,
+ dr.id,
+ info.logical_date,
+ )
+ log.info("Doing session rollback.")
+ session.rollback()
+
+ should_create_backfill_dag_run(
Review Comment:
If there are any concerns I can address them in another PR along with adding a test
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1966843909
##########
airflow/models/backfill.py:
##########
@@ -263,57 +286,54 @@ def _create_backfill_dag_run(
backfill_sort_ordinal,
session,
):
- with session.begin_nested() as nested:
- dr = session.scalar(
- with_row_locks(
- query=_get_latest_dag_run_row_query(info, session),
- session=session,
- ),
+ with session.begin_nested():
+ should_skip_create_backfill = should_create_backfill_dag_run(
+ info, reprocess_behavior, backfill_id, backfill_sort_ordinal, session
)
- if dr:
- non_create_reason = _get_dag_run_no_create_reason(dr, reprocess_behavior)
- if non_create_reason:
- # rolling back here restores to start of this nested tran
- # which releases the lock on the latest dag run, since we
- # are not creating a new one
- nested.rollback()
- session.add(
- BackfillDagRun(
- backfill_id=backfill_id,
- dag_run_id=None,
- logical_date=info.logical_date,
- exception_reason=non_create_reason,
- sort_ordinal=backfill_sort_ordinal,
- )
- )
- return
+ if should_skip_create_backfill:
+ return
+
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
- dr = dag.create_dagrun(
- run_id=dag.timetable.generate_run_id(
- run_type=DagRunType.BACKFILL_JOB,
+ try:
+ dr = dag.create_dagrun(
+ run_id=dag.timetable.generate_run_id(
+ run_type=DagRunType.BACKFILL_JOB,
+ logical_date=info.logical_date,
+ data_interval=info.data_interval,
+ ),
logical_date=info.logical_date,
data_interval=info.data_interval,
- ),
- logical_date=info.logical_date,
- data_interval=info.data_interval,
- run_after=info.run_after,
- conf=dag_run_conf,
- run_type=DagRunType.BACKFILL_JOB,
- triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
- state=DagRunState.QUEUED,
- start_date=timezone.utcnow(),
- backfill_id=backfill_id,
- session=session,
- )
- session.add(
- BackfillDagRun(
+ run_after=info.run_after,
+ conf=dag_run_conf,
+ run_type=DagRunType.BACKFILL_JOB,
+ triggered_by=DagRunTriggeredByType.BACKFILL,
+ dag_version=dag_version,
+ state=DagRunState.QUEUED,
+ start_date=timezone.utcnow(),
backfill_id=backfill_id,
- dag_run_id=dr.id,
- sort_ordinal=backfill_sort_ordinal,
- logical_date=info.logical_date,
+ session=session,
+ )
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill_id,
+ dag_run_id=dr.id,
+ sort_ordinal=backfill_sort_ordinal,
+ logical_date=info.logical_date,
+ )
+ )
+ except IntegrityError:
+ log.info(
+ "Skipped creating backfill dag run for dag_id=%s backfill_id=%s, logical_date=%s (already exists)",
+ dr.dag_id,
+ dr.id,
+ info.logical_date,
+ )
+ log.info("Doing session rollback.")
+ session.rollback()
+
+ should_create_backfill_dag_run(
Review Comment:
resolved in https://github.com/apache/airflow/pull/46959
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1942700714
##########
airflow/models/backfill.py:
##########
@@ -389,34 +391,36 @@ def _create_backfill(
dag=dag,
)
- log.info("obtaining lock on dag %s", dag_id)
- # we obtain a lock on dag model so that nothing else will create
- # dag runs at the same time. mainly this is required by non-uniqueness
- # of logical_date. otherwise we could just create run in a try-except.
- dag_model = session.scalar(
- with_row_locks(
- select(DagModel).where(DagModel.dag_id == dag_id),
- session=session,
- )
- )
+ dag_model = session.scalar(select(DagModel).where(DagModel.dag_id == dag_id))
+
if not dag_model:
raise RuntimeError(f"Dag {dag_id} not found")
for info in dagrun_info_list:
backfill_sort_ordinal += 1
- _create_backfill_dag_run(
- dag=dag,
- info=info,
- backfill_id=br.id,
- dag_run_conf=br.dag_run_conf,
- reprocess_behavior=br.reprocess_behavior,
- backfill_sort_ordinal=backfill_sort_ordinal,
- session=session,
- )
- log.info(
- "created backfill dag run dag_id=%s backfill_id=%s, info=%s",
- dag.dag_id,
- br.id,
- info,
- )
+ try:
Review Comment:
Agreed. Removed the the error handling at the call site.
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on PR #46248:
URL: https://github.com/apache/airflow/pull/46248#issuecomment-2655518042
> I think we should add a test for this in future PRs.
Adding a test in a separate PR
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1941709575
##########
airflow/models/backfill.py:
##########
@@ -389,34 +391,36 @@ def _create_backfill(
dag=dag,
)
- log.info("obtaining lock on dag %s", dag_id)
- # we obtain a lock on dag model so that nothing else will create
- # dag runs at the same time. mainly this is required by non-uniqueness
- # of logical_date. otherwise we could just create run in a try-except.
- dag_model = session.scalar(
- with_row_locks(
- select(DagModel).where(DagModel.dag_id == dag_id),
- session=session,
- )
- )
+ dag_model = session.scalar(select(DagModel).where(DagModel.dag_id == dag_id))
+
if not dag_model:
raise RuntimeError(f"Dag {dag_id} not found")
for info in dagrun_info_list:
backfill_sort_ordinal += 1
- _create_backfill_dag_run(
- dag=dag,
- info=info,
- backfill_id=br.id,
- dag_run_conf=br.dag_run_conf,
- reprocess_behavior=br.reprocess_behavior,
- backfill_sort_ordinal=backfill_sort_ordinal,
- session=session,
- )
- log.info(
- "created backfill dag run dag_id=%s backfill_id=%s, info=%s",
- dag.dag_id,
- br.id,
- info,
- )
+ try:
Review Comment:
hmm so .... you are adding integrity error handling both inside `_create_backfill_dag_run` and at the call site of `_create_backfill_dag_run`. i don't think you need it in both places. you have to decide where it makes sense to add.
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1942699575
##########
airflow/models/backfill.py:
##########
@@ -288,32 +280,42 @@ def _create_backfill_dag_run(
)
return
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
- dr = dag.create_dagrun(
- run_id=dag.timetable.generate_run_id(
- run_type=DagRunType.BACKFILL_JOB,
+ try:
+ dr = dag.create_dagrun(
+ run_id=dag.timetable.generate_run_id(
+ run_type=DagRunType.BACKFILL_JOB,
+ logical_date=info.logical_date,
+ data_interval=info.data_interval,
+ ),
logical_date=info.logical_date,
data_interval=info.data_interval,
- ),
- logical_date=info.logical_date,
- data_interval=info.data_interval,
- run_after=info.run_after,
+ run_after=info.run_after,
conf=dag_run_conf,
- run_type=DagRunType.BACKFILL_JOB,
- triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
- state=DagRunState.QUEUED,
- start_date=timezone.utcnow(),
- backfill_id=backfill_id,
- session=session,
- )
- session.add(
- BackfillDagRun(
+ run_type=DagRunType.BACKFILL_JOB,
+ triggered_by=DagRunTriggeredByType.BACKFILL,
+ dag_version=dag_version,
+ state=DagRunState.QUEUED,
+ start_date=timezone.utcnow(),
backfill_id=backfill_id,
- dag_run_id=dr.id,
- sort_ordinal=backfill_sort_ordinal,
- logical_date=info.logical_date,
+ session=session,
)
- )
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill_id,
+ dag_run_id=dr.id,
+ sort_ordinal=backfill_sort_ordinal,
+ logical_date=info.logical_date,
+ )
+ )
+ except IntegrityError:
+ log.info(
Review Comment:
Handled this in exception handling. Creating BDR if the insert fails.
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1941703951
##########
airflow/models/backfill.py:
##########
@@ -288,32 +280,42 @@ def _create_backfill_dag_run(
)
return
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
- dr = dag.create_dagrun(
- run_id=dag.timetable.generate_run_id(
- run_type=DagRunType.BACKFILL_JOB,
+ try:
+ dr = dag.create_dagrun(
+ run_id=dag.timetable.generate_run_id(
+ run_type=DagRunType.BACKFILL_JOB,
+ logical_date=info.logical_date,
+ data_interval=info.data_interval,
+ ),
logical_date=info.logical_date,
data_interval=info.data_interval,
- ),
- logical_date=info.logical_date,
- data_interval=info.data_interval,
- run_after=info.run_after,
+ run_after=info.run_after,
conf=dag_run_conf,
- run_type=DagRunType.BACKFILL_JOB,
- triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
- state=DagRunState.QUEUED,
- start_date=timezone.utcnow(),
- backfill_id=backfill_id,
- session=session,
- )
- session.add(
- BackfillDagRun(
+ run_type=DagRunType.BACKFILL_JOB,
+ triggered_by=DagRunTriggeredByType.BACKFILL,
+ dag_version=dag_version,
+ state=DagRunState.QUEUED,
+ start_date=timezone.utcnow(),
backfill_id=backfill_id,
- dag_run_id=dr.id,
- sort_ordinal=backfill_sort_ordinal,
- logical_date=info.logical_date,
+ session=session,
)
- )
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill_id,
+ dag_run_id=dr.id,
+ sort_ordinal=backfill_sort_ordinal,
+ logical_date=info.logical_date,
+ )
+ )
+ except IntegrityError:
+ log.info(
Review Comment:
if the insert fails, we should probably still create the BDR record
we would need to do it after the rollback (which would roll back the BDR insert)
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1941703951
##########
airflow/models/backfill.py:
##########
@@ -288,32 +280,42 @@ def _create_backfill_dag_run(
)
return
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
- dr = dag.create_dagrun(
- run_id=dag.timetable.generate_run_id(
- run_type=DagRunType.BACKFILL_JOB,
+ try:
+ dr = dag.create_dagrun(
+ run_id=dag.timetable.generate_run_id(
+ run_type=DagRunType.BACKFILL_JOB,
+ logical_date=info.logical_date,
+ data_interval=info.data_interval,
+ ),
logical_date=info.logical_date,
data_interval=info.data_interval,
- ),
- logical_date=info.logical_date,
- data_interval=info.data_interval,
- run_after=info.run_after,
+ run_after=info.run_after,
conf=dag_run_conf,
- run_type=DagRunType.BACKFILL_JOB,
- triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
- state=DagRunState.QUEUED,
- start_date=timezone.utcnow(),
- backfill_id=backfill_id,
- session=session,
- )
- session.add(
- BackfillDagRun(
+ run_type=DagRunType.BACKFILL_JOB,
+ triggered_by=DagRunTriggeredByType.BACKFILL,
+ dag_version=dag_version,
+ state=DagRunState.QUEUED,
+ start_date=timezone.utcnow(),
backfill_id=backfill_id,
- dag_run_id=dr.id,
- sort_ordinal=backfill_sort_ordinal,
- logical_date=info.logical_date,
+ session=session,
)
- )
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill_id,
+ dag_run_id=dr.id,
+ sort_ordinal=backfill_sort_ordinal,
+ logical_date=info.logical_date,
+ )
+ )
+ except IntegrityError:
+ log.info(
Review Comment:
if the insert fails, we should probably still create the BDR record
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv merged PR #46248:
URL: https://github.com/apache/airflow/pull/46248
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "sunank200 (via GitHub)" <gi...@apache.org>.
sunank200 commented on PR #46248:
URL: https://github.com/apache/airflow/pull/46248#issuecomment-2655645231
> Add test for AIP-83 amendment - update backfill #46605
@prabhusneha I think Rahul has added test here: [46605](https://github.com/apache/airflow/issues/46605). Please check.
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1953821639
##########
airflow/models/backfill.py:
##########
@@ -263,57 +286,54 @@ def _create_backfill_dag_run(
backfill_sort_ordinal,
session,
):
- with session.begin_nested() as nested:
- dr = session.scalar(
- with_row_locks(
- query=_get_latest_dag_run_row_query(info, session),
- session=session,
- ),
+ with session.begin_nested():
+ should_skip_create_backfill = should_create_backfill_dag_run(
+ info, reprocess_behavior, backfill_id, backfill_sort_ordinal, session
)
- if dr:
- non_create_reason = _get_dag_run_no_create_reason(dr, reprocess_behavior)
- if non_create_reason:
- # rolling back here restores to start of this nested tran
- # which releases the lock on the latest dag run, since we
- # are not creating a new one
- nested.rollback()
- session.add(
- BackfillDagRun(
- backfill_id=backfill_id,
- dag_run_id=None,
- logical_date=info.logical_date,
- exception_reason=non_create_reason,
- sort_ordinal=backfill_sort_ordinal,
- )
- )
- return
+ if should_skip_create_backfill:
+ return
+
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
- dr = dag.create_dagrun(
- run_id=dag.timetable.generate_run_id(
- run_type=DagRunType.BACKFILL_JOB,
+ try:
+ dr = dag.create_dagrun(
+ run_id=dag.timetable.generate_run_id(
+ run_type=DagRunType.BACKFILL_JOB,
+ logical_date=info.logical_date,
+ data_interval=info.data_interval,
+ ),
logical_date=info.logical_date,
data_interval=info.data_interval,
- ),
- logical_date=info.logical_date,
- data_interval=info.data_interval,
- run_after=info.run_after,
- conf=dag_run_conf,
- run_type=DagRunType.BACKFILL_JOB,
- triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
- state=DagRunState.QUEUED,
- start_date=timezone.utcnow(),
- backfill_id=backfill_id,
- session=session,
- )
- session.add(
- BackfillDagRun(
+ run_after=info.run_after,
+ conf=dag_run_conf,
+ run_type=DagRunType.BACKFILL_JOB,
+ triggered_by=DagRunTriggeredByType.BACKFILL,
+ dag_version=dag_version,
+ state=DagRunState.QUEUED,
+ start_date=timezone.utcnow(),
backfill_id=backfill_id,
- dag_run_id=dr.id,
- sort_ordinal=backfill_sort_ordinal,
- logical_date=info.logical_date,
+ session=session,
+ )
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill_id,
+ dag_run_id=dr.id,
+ sort_ordinal=backfill_sort_ordinal,
+ logical_date=info.logical_date,
+ )
+ )
+ except IntegrityError:
+ log.info(
+ "Skipped creating backfill dag run for dag_id=%s backfill_id=%s, logical_date=%s (already exists)",
+ dr.dag_id,
+ dr.id,
+ info.logical_date,
+ )
+ log.info("Doing session rollback.")
+ session.rollback()
+
+ should_create_backfill_dag_run(
Review Comment:
But if insert fails, wouldn’t a new insert still fail?
##########
airflow/models/backfill.py:
##########
@@ -263,57 +286,54 @@ def _create_backfill_dag_run(
backfill_sort_ordinal,
session,
):
- with session.begin_nested() as nested:
- dr = session.scalar(
- with_row_locks(
- query=_get_latest_dag_run_row_query(info, session),
- session=session,
- ),
+ with session.begin_nested():
+ should_skip_create_backfill = should_create_backfill_dag_run(
+ info, reprocess_behavior, backfill_id, backfill_sort_ordinal, session
)
- if dr:
- non_create_reason = _get_dag_run_no_create_reason(dr, reprocess_behavior)
- if non_create_reason:
- # rolling back here restores to start of this nested tran
- # which releases the lock on the latest dag run, since we
- # are not creating a new one
- nested.rollback()
- session.add(
- BackfillDagRun(
- backfill_id=backfill_id,
- dag_run_id=None,
- logical_date=info.logical_date,
- exception_reason=non_create_reason,
- sort_ordinal=backfill_sort_ordinal,
- )
- )
- return
+ if should_skip_create_backfill:
+ return
+
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
- dr = dag.create_dagrun(
- run_id=dag.timetable.generate_run_id(
- run_type=DagRunType.BACKFILL_JOB,
+ try:
+ dr = dag.create_dagrun(
+ run_id=dag.timetable.generate_run_id(
+ run_type=DagRunType.BACKFILL_JOB,
+ logical_date=info.logical_date,
+ data_interval=info.data_interval,
+ ),
logical_date=info.logical_date,
data_interval=info.data_interval,
- ),
- logical_date=info.logical_date,
- data_interval=info.data_interval,
- run_after=info.run_after,
- conf=dag_run_conf,
- run_type=DagRunType.BACKFILL_JOB,
- triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
- state=DagRunState.QUEUED,
- start_date=timezone.utcnow(),
- backfill_id=backfill_id,
- session=session,
- )
- session.add(
- BackfillDagRun(
+ run_after=info.run_after,
+ conf=dag_run_conf,
+ run_type=DagRunType.BACKFILL_JOB,
+ triggered_by=DagRunTriggeredByType.BACKFILL,
+ dag_version=dag_version,
+ state=DagRunState.QUEUED,
+ start_date=timezone.utcnow(),
backfill_id=backfill_id,
- dag_run_id=dr.id,
- sort_ordinal=backfill_sort_ordinal,
- logical_date=info.logical_date,
+ session=session,
+ )
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill_id,
+ dag_run_id=dr.id,
+ sort_ordinal=backfill_sort_ordinal,
+ logical_date=info.logical_date,
+ )
+ )
+ except IntegrityError:
+ log.info(
+ "Skipped creating backfill dag run for dag_id=%s backfill_id=%s, logical_date=%s (already exists)",
+ dr.dag_id,
+ dr.id,
+ info.logical_date,
+ )
+ log.info("Doing session rollback.")
+ session.rollback()
+
+ should_create_backfill_dag_run(
Review Comment:
Yeah this looks like a mistake… I’ll submit a PR to 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] AIP-83 amendment -- update backfill [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1953827273
##########
airflow/models/backfill.py:
##########
@@ -263,57 +286,54 @@ def _create_backfill_dag_run(
backfill_sort_ordinal,
session,
):
- with session.begin_nested() as nested:
- dr = session.scalar(
- with_row_locks(
- query=_get_latest_dag_run_row_query(info, session),
- session=session,
- ),
+ with session.begin_nested():
+ should_skip_create_backfill = should_create_backfill_dag_run(
+ info, reprocess_behavior, backfill_id, backfill_sort_ordinal, session
)
- if dr:
- non_create_reason = _get_dag_run_no_create_reason(dr, reprocess_behavior)
- if non_create_reason:
- # rolling back here restores to start of this nested tran
- # which releases the lock on the latest dag run, since we
- # are not creating a new one
- nested.rollback()
- session.add(
- BackfillDagRun(
- backfill_id=backfill_id,
- dag_run_id=None,
- logical_date=info.logical_date,
- exception_reason=non_create_reason,
- sort_ordinal=backfill_sort_ordinal,
- )
- )
- return
+ if should_skip_create_backfill:
+ return
+
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
- dr = dag.create_dagrun(
- run_id=dag.timetable.generate_run_id(
- run_type=DagRunType.BACKFILL_JOB,
+ try:
+ dr = dag.create_dagrun(
+ run_id=dag.timetable.generate_run_id(
+ run_type=DagRunType.BACKFILL_JOB,
+ logical_date=info.logical_date,
+ data_interval=info.data_interval,
+ ),
logical_date=info.logical_date,
data_interval=info.data_interval,
- ),
- logical_date=info.logical_date,
- data_interval=info.data_interval,
- run_after=info.run_after,
- conf=dag_run_conf,
- run_type=DagRunType.BACKFILL_JOB,
- triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
- state=DagRunState.QUEUED,
- start_date=timezone.utcnow(),
- backfill_id=backfill_id,
- session=session,
- )
- session.add(
- BackfillDagRun(
+ run_after=info.run_after,
+ conf=dag_run_conf,
+ run_type=DagRunType.BACKFILL_JOB,
+ triggered_by=DagRunTriggeredByType.BACKFILL,
+ dag_version=dag_version,
+ state=DagRunState.QUEUED,
+ start_date=timezone.utcnow(),
backfill_id=backfill_id,
- dag_run_id=dr.id,
- sort_ordinal=backfill_sort_ordinal,
- logical_date=info.logical_date,
+ session=session,
+ )
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill_id,
+ dag_run_id=dr.id,
+ sort_ordinal=backfill_sort_ordinal,
+ logical_date=info.logical_date,
+ )
+ )
+ except IntegrityError:
+ log.info(
+ "Skipped creating backfill dag run for dag_id=%s backfill_id=%s, logical_date=%s (already exists)",
+ dr.dag_id,
+ dr.id,
+ info.logical_date,
+ )
+ log.info("Doing session rollback.")
+ session.rollback()
+
+ should_create_backfill_dag_run(
Review Comment:
The try block creates a backfill by inserting into `dag_run` table and if that fails, the error handling here is inserting a row in `backfill_dag_run` table with `non_create_reason`.
This was suggested here - https://github.com/apache/airflow/pull/46248#discussion_r1941703951
--
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] AIP-83 amendment -- update backfill [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #46248:
URL: https://github.com/apache/airflow/pull/46248#discussion_r1953817570
##########
airflow/models/backfill.py:
##########
@@ -263,57 +286,54 @@ def _create_backfill_dag_run(
backfill_sort_ordinal,
session,
):
- with session.begin_nested() as nested:
- dr = session.scalar(
- with_row_locks(
- query=_get_latest_dag_run_row_query(info, session),
- session=session,
- ),
+ with session.begin_nested():
+ should_skip_create_backfill = should_create_backfill_dag_run(
+ info, reprocess_behavior, backfill_id, backfill_sort_ordinal, session
)
- if dr:
- non_create_reason = _get_dag_run_no_create_reason(dr, reprocess_behavior)
- if non_create_reason:
- # rolling back here restores to start of this nested tran
- # which releases the lock on the latest dag run, since we
- # are not creating a new one
- nested.rollback()
- session.add(
- BackfillDagRun(
- backfill_id=backfill_id,
- dag_run_id=None,
- logical_date=info.logical_date,
- exception_reason=non_create_reason,
- sort_ordinal=backfill_sort_ordinal,
- )
- )
- return
+ if should_skip_create_backfill:
+ return
+
dag_version = DagVersion.get_latest_version(dag.dag_id, session=session)
- dr = dag.create_dagrun(
- run_id=dag.timetable.generate_run_id(
- run_type=DagRunType.BACKFILL_JOB,
+ try:
+ dr = dag.create_dagrun(
+ run_id=dag.timetable.generate_run_id(
+ run_type=DagRunType.BACKFILL_JOB,
+ logical_date=info.logical_date,
+ data_interval=info.data_interval,
+ ),
logical_date=info.logical_date,
data_interval=info.data_interval,
- ),
- logical_date=info.logical_date,
- data_interval=info.data_interval,
- run_after=info.run_after,
- conf=dag_run_conf,
- run_type=DagRunType.BACKFILL_JOB,
- triggered_by=DagRunTriggeredByType.BACKFILL,
- dag_version=dag_version,
- state=DagRunState.QUEUED,
- start_date=timezone.utcnow(),
- backfill_id=backfill_id,
- session=session,
- )
- session.add(
- BackfillDagRun(
+ run_after=info.run_after,
+ conf=dag_run_conf,
+ run_type=DagRunType.BACKFILL_JOB,
+ triggered_by=DagRunTriggeredByType.BACKFILL,
+ dag_version=dag_version,
+ state=DagRunState.QUEUED,
+ start_date=timezone.utcnow(),
backfill_id=backfill_id,
- dag_run_id=dr.id,
- sort_ordinal=backfill_sort_ordinal,
- logical_date=info.logical_date,
+ session=session,
+ )
+ session.add(
+ BackfillDagRun(
+ backfill_id=backfill_id,
+ dag_run_id=dr.id,
+ sort_ordinal=backfill_sort_ordinal,
+ logical_date=info.logical_date,
+ )
+ )
+ except IntegrityError:
+ log.info(
+ "Skipped creating backfill dag run for dag_id=%s backfill_id=%s, logical_date=%s (already exists)",
+ dr.dag_id,
+ dr.id,
+ info.logical_date,
+ )
+ log.info("Doing session rollback.")
+ session.rollback()
+
+ should_create_backfill_dag_run(
Review Comment:
Yeah this looks like a mistake… I’ll submit a PR to 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