Dynamically Mapped Tasks: DB performance issues about airflow HOT 17 CLOSED

yeah, would be great to avoid DB internal error

Feel free to fix it. Apache Airflow it is open source project, and every one could contribute changes (fixes/features) back, especially if they know what then outcome achievement/benefits of the changes.

So I would recommend try to patch this part into your side and check is it still work as expected, and without additional error logs in DB backend and without side effects.

diff --git a/airflow/models/taskinstance.py b/airflow/models/taskinstance.py
index f154461a77..a08b9cd94a 100644
--- a/airflow/models/taskinstance.py
+++ b/airflow/models/taskinstance.py
@@ -3454,8 +3454,12 @@ class TaskInstance(Base, LoggingMixin):
                     run_id=ti.run_id,
                 ),
                 session=session,
-                nowait=True,
-            ).one()
+                skip_locked=True,
+            ).one_or_none()
+            if not dag_run:
+                # Need to log something?
+                session.rollback()
+                return
 
             task = ti.task
             if TYPE_CHECKING:

from airflow.

thanks @Taragolis, I will try with my first PR)

from airflow.

I tested the image from PR on our environment and this error disappeared.

from airflow.

Correction: SELECT .. FOR UPDATE NOWAIT raise this error

from airflow.

This one use in mini scheduler mechanism, which might fail and it is fine and by design, this just an optimisation mechanism.
It might also tried to work concurrently, so the idea (as I could understand) it try to obtain row lock level lock and if it failed (EAFP) than it fine because other mini-scheduler already care about it.

from airflow.

As measure might be use skip_locked instead of no_wait both of them non blocking, with differences how it handled internally in DB backend:

• SKIP LOCKED: return non locked records
• NO WAIT: raise an error

from airflow.

I've added couple additional reviewers which might more familiar with the mini scheduler. But it might take a time - all review happen by people on their free time, and some reviews require a bit more time that others.

You could also ask a round in #new-contributors or #contributors (pick one channel) in Apache Airflow Community Slack Workspace

from airflow.

Does #39745 solves this issue or are there additional tasks?

It does, we can close it.

from airflow.

The Postgres instance started reporting many could not obtain lock on row in relation "dag_run" errors:

This pretty fine, because Airflow tried to obtain row level lock by utilise statement SELECT .. FOR UPDATE SKIP LOCKED

from airflow.

But it seems what has been added in #38914. I'm not sure that ignoring this error can be considered a solution. If it always results in an ERROR when trying to obtain a lock, why do we even need to attempt the query?

from airflow.

if it always results in an ERROR

Are you sure that it is always return an error?

from airflow.

I'm not sure. But then should be kind of retry mechanism(I'm also not sure if we have it). Anyway I just see a lot of errors in DB and my first guess: something is not correct with transaction level.

from airflow.

As measure might be use skip_locked instead of no_wait both of them non blocking, with differences how it handled internally in DB backend:

• SKIP LOCKED: return non locked records
• NO WAIT: raise an error

yeah, would be great to avoid DB internal error

from airflow.

@VladimirYushkevich thanks for the PR, I am seeing the same errors since updating to 2.9.1 when running mapped tasks on a LocalExecutor.

from airflow.

@Taragolis, do I need to ask for the review explicitly? I believe the PR is ready from my side, but maybe I'm missing something.

from airflow.

Does #39745 solves this issue or are there additional tasks?

from airflow.

related

#38914
#39745

from airflow.