<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.3</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">I have a DAG that contains <a href="https://airflow.apache.org/docs/apache-airflow/stable/concepts/dynamic-task-mapping.html" rel="nofollow">dynamic tasks</a>, and when I try to <code class="notranslate">backfill</code> it I consistently run into a errors like the one below:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1799, in _execute_context
    self.dialect.do_executemany(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py&quot;, line 953, in do_executemany
    context._psycopg2_fetched_rows = xtras.execute_values(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/psycopg2/extras.py&quot;, line 1270, in execute_values
    cur.execute(b''.join(parts))
psycopg2.errors.DeadlockDetected: deadlock detected
DETAIL:  Process 3398082 waits for ShareLock on transaction 356299; blocked by process 3403341.
Process 3403341 waits for ShareLock on transaction 356228; blocked by process 3398082.
HINT:  See server log for query details.
CONTEXT:  while locking tuple (0,58) in relation &quot;dag_run&quot;
SQL statement &quot;SELECT 1 FROM ONLY &quot;public&quot;.&quot;dag_run&quot; x WHERE &quot;dag_id&quot;::pg_catalog.text OPERATOR(pg_catalog.=) $1::pg_catalog.text AND &quot;run_id&quot;::pg_catalog.text OPERATOR(pg_catalog.=) $2::pg_catalog.text FOR KEY SHARE OF x&quot;


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py&quot;, line 847, in _execute
    self._execute_dagruns(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py&quot;, line 737, in _execute_dagruns
    processed_dag_run_dates = self._process_backfill_task_instances(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py&quot;, line 637, in _process_backfill_task_instances
    run.update_state(session=session)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 524, in update_state
    info = self.task_instance_scheduling_decisions(session)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 647, in task_instance_scheduling_decisions
    self.verify_integrity(missing_indexes=missing_indexes, session=session)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 853, in verify_integrity
    self._create_task_instances(dag.dag_id, tasks, created_counts, hook_is_noop, session=session)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 1046, in _create_task_instances
    session.bulk_insert_mappings(TI, tasks)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 3736, in bulk_insert_mappings
    self._bulk_save_mappings(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 3843, in _bulk_save_mappings
    transaction.rollback(_capture_exception=True)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/langhelpers.py&quot;, line 70, in __exit__
    compat.raise_(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/compat.py&quot;, line 207, in raise_
    raise exception
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 3831, in _bulk_save_mappings
    persistence._bulk_insert(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py&quot;, line 107, in _bulk_insert
    _emit_insert_statements(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py&quot;, line 1097, in _emit_insert_statements
    c = connection._execute_20(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1631, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/sql/elements.py&quot;, line 325, in _execute_on_connection
    return connection._execute_clauseelement(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1498, in _execute_clauseelement
    ret = self._execute_context(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1862, in _execute_context
    self._handle_dbapi_exception(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 2043, in _handle_dbapi_exception
    util.raise_(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/compat.py&quot;, line 207, in raise_
    raise exception
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1799, in _execute_context
    self.dialect.do_executemany(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py&quot;, line 953, in do_executemany
    context._psycopg2_fetched_rows = xtras.execute_values(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/psycopg2/extras.py&quot;, line 1270, in execute_values
    cur.execute(b''.join(parts))
sqlalchemy.exc.OperationalError: (psycopg2.errors.DeadlockDetected) deadlock detected
DETAIL:  Process 3398082 waits for ShareLock on transaction 356299; blocked by process 3403341.
Process 3403341 waits for ShareLock on transaction 356228; blocked by process 3398082.
HINT:  See server log for query details.
CONTEXT:  while locking tuple (0,58) in relation &quot;dag_run&quot;
SQL statement &quot;SELECT 1 FROM ONLY &quot;public&quot;.&quot;dag_run&quot; x WHERE &quot;dag_id&quot;::pg_catalog.text OPERATOR(pg_catalog.=) $1::pg_catalog.text AND &quot;run_id&quot;::pg_catalog.text OPERATOR(pg_catalog.=) $2::pg_catalog.text FOR KEY SHARE OF x&quot;

[SQL: INSERT INTO task_instance (task_id, dag_id, run_id, map_index, try_number, max_tries, hostname, unixname, pool, pool_slots, queue, priority_weight, operator, executor_config) VALUES (%(task_id)s, %(dag_id)s, %(run_id)s, %(map_index)s, %(try_number)s, %(max_tries)s, %(hostname)s, %(unixname)s, %(pool)s, %(pool_slots)s, %(queue)s, %(priority_weight)s, %(operator)s, %(executor_config)s)]
[parameters: ({'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 8, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920720&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 9, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920f90&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 10, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920c30&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 11, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb9200f0&gt;})]"><pre class="notranslate"><code class="notranslate">Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1799, in _execute_context
    self.dialect.do_executemany(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py", line 953, in do_executemany
    context._psycopg2_fetched_rows = xtras.execute_values(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/psycopg2/extras.py", line 1270, in execute_values
    cur.execute(b''.join(parts))
psycopg2.errors.DeadlockDetected: deadlock detected
DETAIL:  Process 3398082 waits for ShareLock on transaction 356299; blocked by process 3403341.
Process 3403341 waits for ShareLock on transaction 356228; blocked by process 3398082.
HINT:  See server log for query details.
CONTEXT:  while locking tuple (0,58) in relation "dag_run"
SQL statement "SELECT 1 FROM ONLY "public"."dag_run" x WHERE "dag_id"::pg_catalog.text OPERATOR(pg_catalog.=) $1::pg_catalog.text AND "run_id"::pg_catalog.text OPERATOR(pg_catalog.=) $2::pg_catalog.text FOR KEY SHARE OF x"


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py", line 847, in _execute
    self._execute_dagruns(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
    return func(*args, **kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py", line 737, in _execute_dagruns
    processed_dag_run_dates = self._process_backfill_task_instances(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
    return func(*args, **kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py", line 637, in _process_backfill_task_instances
    run.update_state(session=session)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
    return func(*args, **kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py", line 524, in update_state
    info = self.task_instance_scheduling_decisions(session)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
    return func(*args, **kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py", line 647, in task_instance_scheduling_decisions
    self.verify_integrity(missing_indexes=missing_indexes, session=session)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
    return func(*args, **kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py", line 853, in verify_integrity
    self._create_task_instances(dag.dag_id, tasks, created_counts, hook_is_noop, session=session)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py", line 1046, in _create_task_instances
    session.bulk_insert_mappings(TI, tasks)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 3736, in bulk_insert_mappings
    self._bulk_save_mappings(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 3843, in _bulk_save_mappings
    transaction.rollback(_capture_exception=True)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
    compat.raise_(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
    raise exception
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 3831, in _bulk_save_mappings
    persistence._bulk_insert(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py", line 107, in _bulk_insert
    _emit_insert_statements(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py", line 1097, in _emit_insert_statements
    c = connection._execute_20(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1631, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/sql/elements.py", line 325, in _execute_on_connection
    return connection._execute_clauseelement(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1498, in _execute_clauseelement
    ret = self._execute_context(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1862, in _execute_context
    self._handle_dbapi_exception(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 2043, in _handle_dbapi_exception
    util.raise_(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
    raise exception
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1799, in _execute_context
    self.dialect.do_executemany(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py", line 953, in do_executemany
    context._psycopg2_fetched_rows = xtras.execute_values(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/psycopg2/extras.py", line 1270, in execute_values
    cur.execute(b''.join(parts))
sqlalchemy.exc.OperationalError: (psycopg2.errors.DeadlockDetected) deadlock detected
DETAIL:  Process 3398082 waits for ShareLock on transaction 356299; blocked by process 3403341.
Process 3403341 waits for ShareLock on transaction 356228; blocked by process 3398082.
HINT:  See server log for query details.
CONTEXT:  while locking tuple (0,58) in relation "dag_run"
SQL statement "SELECT 1 FROM ONLY "public"."dag_run" x WHERE "dag_id"::pg_catalog.text OPERATOR(pg_catalog.=) $1::pg_catalog.text AND "run_id"::pg_catalog.text OPERATOR(pg_catalog.=) $2::pg_catalog.text FOR KEY SHARE OF x"

[SQL: INSERT INTO task_instance (task_id, dag_id, run_id, map_index, try_number, max_tries, hostname, unixname, pool, pool_slots, queue, priority_weight, operator, executor_config) VALUES (%(task_id)s, %(dag_id)s, %(run_id)s, %(map_index)s, %(try_number)s, %(max_tries)s, %(hostname)s, %(unixname)s, %(pool)s, %(pool_slots)s, %(queue)s, %(priority_weight)s, %(operator)s, %(executor_config)s)]
[parameters: ({'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 8, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920720&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 9, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920f90&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 10, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920c30&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 11, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb9200f0&gt;})]
</code></pre></div>
<details>
  <summary>More log context</summary>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[2022-08-15 16:10:38,545] {dagbag.py:508} INFO - Filling up the DagBag from /home/tsanders/airflow_standalone_sqlite/dags/test_backfill_races.py
[2022-08-15 16:10:38,740] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=15 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,741] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=14 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,747] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=10 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,747] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=9 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,748] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=16 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,755] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=12 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,756] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=8 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,761] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=11 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,767] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=18 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,768] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=17 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,771] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=19 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,776] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=13 [queued]&gt; on host nkclintjs.tradebot.com
Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1799, in _execute_context
    self.dialect.do_executemany(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py&quot;, line 953, in do_executemany
    context._psycopg2_fetched_rows = xtras.execute_values(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/psycopg2/extras.py&quot;, line 1270, in execute_values
    cur.execute(b''.join(parts))
psycopg2.errors.DeadlockDetected: deadlock detected
DETAIL:  Process 3398082 waits for ShareLock on transaction 356299; blocked by process 3403341.
Process 3403341 waits for ShareLock on transaction 356228; blocked by process 3398082.
HINT:  See server log for query details.
CONTEXT:  while locking tuple (0,58) in relation &quot;dag_run&quot;
SQL statement &quot;SELECT 1 FROM ONLY &quot;public&quot;.&quot;dag_run&quot; x WHERE &quot;dag_id&quot;::pg_catalog.text OPERATOR(pg_catalog.=) $1::pg_catalog.text AND &quot;run_id&quot;::pg_catalog.text OPERATOR(pg_catalog.=) $2::pg_catalog.text FOR KEY SHARE OF x&quot;


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py&quot;, line 847, in _execute
    self._execute_dagruns(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py&quot;, line 737, in _execute_dagruns
    processed_dag_run_dates = self._process_backfill_task_instances(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py&quot;, line 637, in _process_backfill_task_instances
    run.update_state(session=session)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 524, in update_state
    info = self.task_instance_scheduling_decisions(session)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 647, in task_instance_scheduling_decisions
    self.verify_integrity(missing_indexes=missing_indexes, session=session)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 68, in wrapper
    return func(*args, **kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 853, in verify_integrity
    self._create_task_instances(dag.dag_id, tasks, created_counts, hook_is_noop, session=session)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 1046, in _create_task_instances
    session.bulk_insert_mappings(TI, tasks)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 3736, in bulk_insert_mappings
    self._bulk_save_mappings(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 3843, in _bulk_save_mappings
    transaction.rollback(_capture_exception=True)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/langhelpers.py&quot;, line 70, in __exit__
    compat.raise_(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/compat.py&quot;, line 207, in raise_
    raise exception
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 3831, in _bulk_save_mappings
    persistence._bulk_insert(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py&quot;, line 107, in _bulk_insert
    _emit_insert_statements(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py&quot;, line 1097, in _emit_insert_statements
    c = connection._execute_20(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1631, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/sql/elements.py&quot;, line 325, in _execute_on_connection
    return connection._execute_clauseelement(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1498, in _execute_clauseelement
    ret = self._execute_context(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1862, in _execute_context
    self._handle_dbapi_exception(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 2043, in _handle_dbapi_exception
    util.raise_(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/compat.py&quot;, line 207, in raise_
    raise exception
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1799, in _execute_context
    self.dialect.do_executemany(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py&quot;, line 953, in do_executemany
    context._psycopg2_fetched_rows = xtras.execute_values(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/psycopg2/extras.py&quot;, line 1270, in execute_values
    cur.execute(b''.join(parts))
sqlalchemy.exc.OperationalError: (psycopg2.errors.DeadlockDetected) deadlock detected
DETAIL:  Process 3398082 waits for ShareLock on transaction 356299; blocked by process 3403341.
Process 3403341 waits for ShareLock on transaction 356228; blocked by process 3398082.
HINT:  See server log for query details.
CONTEXT:  while locking tuple (0,58) in relation &quot;dag_run&quot;
SQL statement &quot;SELECT 1 FROM ONLY &quot;public&quot;.&quot;dag_run&quot; x WHERE &quot;dag_id&quot;::pg_catalog.text OPERATOR(pg_catalog.=) $1::pg_catalog.text AND &quot;run_id&quot;::pg_catalog.text OPERATOR(pg_catalog.=) $2::pg_catalog.text FOR KEY SHARE OF x&quot;

[SQL: INSERT INTO task_instance (task_id, dag_id, run_id, map_index, try_number, max_tries, hostname, unixname, pool, pool_slots, queue, priority_weight, operator, executor_config) VALUES (%(task_id)s, %(dag_id)s, %(run_id)s, %(map_index)s, %(try_number)s, %(max_tries)s, %(hostname)s, %(unixname)s, %(pool)s, %(pool_slots)s, %(queue)s, %(priority_weight)s, %(operator)s, %(executor_config)s)]
[parameters: ({'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 8, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920720&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 9, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920f90&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 10, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920c30&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 11, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb9200f0&gt;})]
(Background on this error at: https://sqlalche.me/e/14/e3q8)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/bin/airflow&quot;, line 11, in &lt;module&gt;
    sys.exit(main())
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/__main__.py&quot;, line 38, in main
    args.func(args)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/cli/cli_parser.py&quot;, line 51, in command
    return func(*args, **kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 99, in wrapper
    return f(*args, **kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/cli/commands/dag_command.py&quot;, line 107, in dag_backfill
    dag.run(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dag.py&quot;, line 2288, in run
    job.run()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/base_job.py&quot;, line 244, in run
    self._execute()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 71, in wrapper
    return func(*args, session=session, **kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py&quot;, line 876, in _execute
    session.commit()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 1435, in commit
    self._transaction.commit(_to_root=self.future)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 827, in commit
    self._assert_active(prepared_ok=True)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 601, in _assert_active
    raise sa_exc.PendingRollbackError(
sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (psycopg2.errors.DeadlockDetected) deadlock detected
DETAIL:  Process 3398082 waits for ShareLock on transaction 356299; blocked by process 3403341.
Process 3403341 waits for ShareLock on transaction 356228; blocked by process 3398082.
HINT:  See server log for query details.
CONTEXT:  while locking tuple (0,58) in relation &quot;dag_run&quot;
SQL statement &quot;SELECT 1 FROM ONLY &quot;public&quot;.&quot;dag_run&quot; x WHERE &quot;dag_id&quot;::pg_catalog.text OPERATOR(pg_catalog.=) $1::pg_catalog.text AND &quot;run_id&quot;::pg_catalog.text OPERATOR(pg_catalog.=) $2::pg_catalog.text FOR KEY SHARE OF x&quot;

[SQL: INSERT INTO task_instance (task_id, dag_id, run_id, map_index, try_number, max_tries, hostname, unixname, pool, pool_slots, queue, priority_weight, operator, executor_config) VALUES (%(task_id)s, %(dag_id)s, %(run_id)s, %(map_index)s, %(try_number)s, %(max_tries)s, %(hostname)s, %(unixname)s, %(pool)s, %(pool_slots)s, %(queue)s, %(priority_weight)s, %(operator)s, %(executor_config)s)]
[parameters: ({'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 8, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920720&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 9, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920f90&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 10, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920c30&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 11, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb9200f0&gt;})]
(Background on this error at: https://sqlalche.me/e/14/e3q8) (Background on this error at: https://sqlalche.me/e/14/7s2a)
Process QueuedLocalWorker-4:
Process QueuedLocalWorker-5:
Process QueuedLocalWorker-3:
Process QueuedLocalWorker-2:
Process QueuedLocalWorker-13:
Process QueuedLocalWorker-6:
Process QueuedLocalWorker-7:
Process QueuedLocalWorker-8:
Process QueuedLocalWorker-12:
Process QueuedLocalWorker-11:
Process QueuedLocalWorker-9:
Process QueuedLocalWorker-10:
Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 190, in do_work
    self.execute_work(key=key, command=command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 271, in sigint_handler
    sys.exit(0)
Traceback (most recent call last):
SystemExit: 0

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 190, in do_work
    self.execute_work(key=key, command=command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 271, in sigint_handler
    sys.exit(0)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 190, in do_work
    self.execute_work(key=key, command=command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 67, in run
    return super().run()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)
SystemExit: 0
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 192, in do_work
    self.task_queue.task_done()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 271, in sigint_handler
    sys.exit(0)

During handling of the above exception, another exception occurred:

  File &quot;&lt;string&gt;&quot;, line 2, in task_done
Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py&quot;, line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
SystemExit: 0
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 416, in _send_bytes
    self._send(header + buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 373, in _send
    n = write(self._handle, buf)

During handling of the above exception, another exception occurred:

  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
BrokenPipeError: [Errno 32] Broken pipe
Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 67, in run
    return super().run()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 192, in do_work
    self.task_queue.task_done()
  File &quot;&lt;string&gt;&quot;, line 2, in task_done
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py&quot;, line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 416, in _send_bytes
    self._send(header + buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 373, in _send
    n = write(self._handle, buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 67, in run
    return super().run()
BrokenPipeError: [Errno 32] Broken pipe
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 192, in do_work
    self.task_queue.task_done()
  File &quot;&lt;string&gt;&quot;, line 2, in task_done
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py&quot;, line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 416, in _send_bytes
    self._send(header + buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 373, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 190, in do_work
    self.execute_work(key=key, command=command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 190, in do_work
    self.execute_work(key=key, command=command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 190, in do_work
    self.execute_work(key=key, command=command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 190, in do_work
    self.execute_work(key=key, command=command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 190, in do_work
    self.execute_work(key=key, command=command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 190, in do_work
    self.execute_work(key=key, command=command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 190, in do_work
    self.execute_work(key=key, command=command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 271, in sigint_handler
    sys.exit(0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 271, in sigint_handler
    sys.exit(0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
SystemExit: 0
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 271, in sigint_handler
    sys.exit(0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
SystemExit: 0
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 271, in sigint_handler
    sys.exit(0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 271, in sigint_handler
    sys.exit(0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 271, in sigint_handler
    sys.exit(0)

During handling of the above exception, another exception occurred:

  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 271, in sigint_handler
    sys.exit(0)
SystemExit: 0

During handling of the above exception, another exception occurred:

SystemExit: 0
Traceback (most recent call last):
Traceback (most recent call last):
SystemExit: 0
Traceback (most recent call last):
SystemExit: 0

During handling of the above exception, another exception occurred:


During handling of the above exception, another exception occurred:

SystemExit: 0
Traceback (most recent call last):

During handling of the above exception, another exception occurred:


During handling of the above exception, another exception occurred:

Traceback (most recent call last):
Traceback (most recent call last):

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 190, in do_work
    self.execute_work(key=key, command=command)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 67, in run
    return super().run()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 190, in do_work
    self.execute_work(key=key, command=command)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 67, in run
    return super().run()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 67, in run
    return super().run()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 271, in sigint_handler
    sys.exit(0)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 67, in run
    return super().run()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 192, in do_work
    self.task_queue.task_done()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 67, in run
    return super().run()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 67, in run
    return super().run()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
SystemExit: 0
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 192, in do_work
    self.task_queue.task_done()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 67, in run
    return super().run()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)
  File &quot;&lt;string&gt;&quot;, line 2, in task_done
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 192, in do_work
    self.task_queue.task_done()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 271, in sigint_handler
    sys.exit(0)
  File &quot;&lt;string&gt;&quot;, line 2, in task_done
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)

During handling of the above exception, another exception occurred:

  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 192, in do_work
    self.task_queue.task_done()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py&quot;, line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File &quot;&lt;string&gt;&quot;, line 2, in task_done
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 192, in do_work
    self.task_queue.task_done()
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 192, in do_work
    self.task_queue.task_done()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py&quot;, line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 192, in do_work
    self.task_queue.task_done()
Traceback (most recent call last):
  File &quot;&lt;string&gt;&quot;, line 2, in task_done
SystemExit: 0
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py&quot;, line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File &quot;&lt;string&gt;&quot;, line 2, in task_done
  File &quot;&lt;string&gt;&quot;, line 2, in task_done
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File &quot;&lt;string&gt;&quot;, line 2, in task_done
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py&quot;, line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 416, in _send_bytes
    self._send(header + buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))

During handling of the above exception, another exception occurred:

  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py&quot;, line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py&quot;, line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 416, in _send_bytes
    self._send(header + buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py&quot;, line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 373, in _send
    n = write(self._handle, buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 416, in _send_bytes
    self._send(header + buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 416, in _send_bytes
    self._send(header + buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 416, in _send_bytes
    self._send(header + buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 373, in _send
    n = write(self._handle, buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 416, in _send_bytes
    self._send(header + buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 373, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe
BrokenPipeError: [Errno 32] Broken pipe
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 67, in run
    return super().run()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 373, in _send
    n = write(self._handle, buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 373, in _send
    n = write(self._handle, buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 416, in _send_bytes
    self._send(header + buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 315, in _bootstrap
    self.run()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 373, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)
BrokenPipeError: [Errno 32] Broken pipe
BrokenPipeError: [Errno 32] Broken pipe
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 373, in _send
    n = write(self._handle, buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 67, in run
    return super().run()
BrokenPipeError: [Errno 32] Broken pipe
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 192, in do_work
    self.task_queue.task_done()
BrokenPipeError: [Errno 32] Broken pipe
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py&quot;, line 108, in run
    self._target(*self._args, **self._kwargs)
  File &quot;&lt;string&gt;&quot;, line 2, in task_done
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py&quot;, line 192, in do_work
    self.task_queue.task_done()
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py&quot;, line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File &quot;&lt;string&gt;&quot;, line 2, in task_done
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py&quot;, line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 416, in _send_bytes
    self._send(header + buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 373, in _send
    n = write(self._handle, buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 416, in _send_bytes
    self._send(header + buf)
  File &quot;/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py&quot;, line 373, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe
BrokenPipeError: [Errno 32] Broken pipe"><pre class="notranslate"><code class="notranslate">[2022-08-15 16:10:38,545] {dagbag.py:508} INFO - Filling up the DagBag from /home/tsanders/airflow_standalone_sqlite/dags/test_backfill_races.py
[2022-08-15 16:10:38,740] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=15 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,741] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=14 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,747] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=10 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,747] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=9 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,748] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=16 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,755] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=12 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,756] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=8 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,761] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=11 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,767] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=18 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,768] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=17 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,771] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=19 [queued]&gt; on host nkclintjs.tradebot.com
[2022-08-15 16:10:38,776] {task_command.py:371} INFO - Running &lt;TaskInstance: test_backfill_races.t1 backfill__2022-02-04T06:00:00+00:00 map_index=13 [queued]&gt; on host nkclintjs.tradebot.com
Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1799, in _execute_context
    self.dialect.do_executemany(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py", line 953, in do_executemany
    context._psycopg2_fetched_rows = xtras.execute_values(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/psycopg2/extras.py", line 1270, in execute_values
    cur.execute(b''.join(parts))
psycopg2.errors.DeadlockDetected: deadlock detected
DETAIL:  Process 3398082 waits for ShareLock on transaction 356299; blocked by process 3403341.
Process 3403341 waits for ShareLock on transaction 356228; blocked by process 3398082.
HINT:  See server log for query details.
CONTEXT:  while locking tuple (0,58) in relation "dag_run"
SQL statement "SELECT 1 FROM ONLY "public"."dag_run" x WHERE "dag_id"::pg_catalog.text OPERATOR(pg_catalog.=) $1::pg_catalog.text AND "run_id"::pg_catalog.text OPERATOR(pg_catalog.=) $2::pg_catalog.text FOR KEY SHARE OF x"


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py", line 847, in _execute
    self._execute_dagruns(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
    return func(*args, **kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py", line 737, in _execute_dagruns
    processed_dag_run_dates = self._process_backfill_task_instances(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
    return func(*args, **kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py", line 637, in _process_backfill_task_instances
    run.update_state(session=session)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
    return func(*args, **kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py", line 524, in update_state
    info = self.task_instance_scheduling_decisions(session)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
    return func(*args, **kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py", line 647, in task_instance_scheduling_decisions
    self.verify_integrity(missing_indexes=missing_indexes, session=session)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py", line 68, in wrapper
    return func(*args, **kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py", line 853, in verify_integrity
    self._create_task_instances(dag.dag_id, tasks, created_counts, hook_is_noop, session=session)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py", line 1046, in _create_task_instances
    session.bulk_insert_mappings(TI, tasks)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 3736, in bulk_insert_mappings
    self._bulk_save_mappings(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 3843, in _bulk_save_mappings
    transaction.rollback(_capture_exception=True)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
    compat.raise_(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
    raise exception
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 3831, in _bulk_save_mappings
    persistence._bulk_insert(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py", line 107, in _bulk_insert
    _emit_insert_statements(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py", line 1097, in _emit_insert_statements
    c = connection._execute_20(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1631, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/sql/elements.py", line 325, in _execute_on_connection
    return connection._execute_clauseelement(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1498, in _execute_clauseelement
    ret = self._execute_context(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1862, in _execute_context
    self._handle_dbapi_exception(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 2043, in _handle_dbapi_exception
    util.raise_(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
    raise exception
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1799, in _execute_context
    self.dialect.do_executemany(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py", line 953, in do_executemany
    context._psycopg2_fetched_rows = xtras.execute_values(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/psycopg2/extras.py", line 1270, in execute_values
    cur.execute(b''.join(parts))
sqlalchemy.exc.OperationalError: (psycopg2.errors.DeadlockDetected) deadlock detected
DETAIL:  Process 3398082 waits for ShareLock on transaction 356299; blocked by process 3403341.
Process 3403341 waits for ShareLock on transaction 356228; blocked by process 3398082.
HINT:  See server log for query details.
CONTEXT:  while locking tuple (0,58) in relation "dag_run"
SQL statement "SELECT 1 FROM ONLY "public"."dag_run" x WHERE "dag_id"::pg_catalog.text OPERATOR(pg_catalog.=) $1::pg_catalog.text AND "run_id"::pg_catalog.text OPERATOR(pg_catalog.=) $2::pg_catalog.text FOR KEY SHARE OF x"

[SQL: INSERT INTO task_instance (task_id, dag_id, run_id, map_index, try_number, max_tries, hostname, unixname, pool, pool_slots, queue, priority_weight, operator, executor_config) VALUES (%(task_id)s, %(dag_id)s, %(run_id)s, %(map_index)s, %(try_number)s, %(max_tries)s, %(hostname)s, %(unixname)s, %(pool)s, %(pool_slots)s, %(queue)s, %(priority_weight)s, %(operator)s, %(executor_config)s)]
[parameters: ({'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 8, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920720&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 9, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920f90&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 10, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920c30&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 11, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb9200f0&gt;})]
(Background on this error at: https://sqlalche.me/e/14/e3q8)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/opt/conda/envs/production/bin/airflow", line 11, in &lt;module&gt;
    sys.exit(main())
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/__main__.py", line 38, in main
    args.func(args)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/cli/cli_parser.py", line 51, in command
    return func(*args, **kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py", line 99, in wrapper
    return f(*args, **kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/cli/commands/dag_command.py", line 107, in dag_backfill
    dag.run(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dag.py", line 2288, in run
    job.run()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/base_job.py", line 244, in run
    self._execute()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/session.py", line 71, in wrapper
    return func(*args, session=session, **kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/jobs/backfill_job.py", line 876, in _execute
    session.commit()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 1435, in commit
    self._transaction.commit(_to_root=self.future)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 827, in commit
    self._assert_active(prepared_ok=True)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 601, in _assert_active
    raise sa_exc.PendingRollbackError(
sqlalchemy.exc.PendingRollbackError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (psycopg2.errors.DeadlockDetected) deadlock detected
DETAIL:  Process 3398082 waits for ShareLock on transaction 356299; blocked by process 3403341.
Process 3403341 waits for ShareLock on transaction 356228; blocked by process 3398082.
HINT:  See server log for query details.
CONTEXT:  while locking tuple (0,58) in relation "dag_run"
SQL statement "SELECT 1 FROM ONLY "public"."dag_run" x WHERE "dag_id"::pg_catalog.text OPERATOR(pg_catalog.=) $1::pg_catalog.text AND "run_id"::pg_catalog.text OPERATOR(pg_catalog.=) $2::pg_catalog.text FOR KEY SHARE OF x"

[SQL: INSERT INTO task_instance (task_id, dag_id, run_id, map_index, try_number, max_tries, hostname, unixname, pool, pool_slots, queue, priority_weight, operator, executor_config) VALUES (%(task_id)s, %(dag_id)s, %(run_id)s, %(map_index)s, %(try_number)s, %(max_tries)s, %(hostname)s, %(unixname)s, %(pool)s, %(pool_slots)s, %(queue)s, %(priority_weight)s, %(operator)s, %(executor_config)s)]
[parameters: ({'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 8, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920720&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 9, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920f90&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 10, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb920c30&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-02-04T06:00:00+00:00', 'map_index': 11, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tsanders', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7fb6bb9200f0&gt;})]
(Background on this error at: https://sqlalche.me/e/14/e3q8) (Background on this error at: https://sqlalche.me/e/14/7s2a)
Process QueuedLocalWorker-4:
Process QueuedLocalWorker-5:
Process QueuedLocalWorker-3:
Process QueuedLocalWorker-2:
Process QueuedLocalWorker-13:
Process QueuedLocalWorker-6:
Process QueuedLocalWorker-7:
Process QueuedLocalWorker-8:
Process QueuedLocalWorker-12:
Process QueuedLocalWorker-11:
Process QueuedLocalWorker-9:
Process QueuedLocalWorker-10:
Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 190, in do_work
    self.execute_work(key=key, command=command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py", line 271, in sigint_handler
    sys.exit(0)
Traceback (most recent call last):
SystemExit: 0

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 190, in do_work
    self.execute_work(key=key, command=command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py", line 271, in sigint_handler
    sys.exit(0)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 190, in do_work
    self.execute_work(key=key, command=command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 67, in run
    return super().run()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
SystemExit: 0
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 192, in do_work
    self.task_queue.task_done()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py", line 271, in sigint_handler
    sys.exit(0)

During handling of the above exception, another exception occurred:

  File "&lt;string&gt;", line 2, in task_done
Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py", line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
SystemExit: 0
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 416, in _send_bytes
    self._send(header + buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 373, in _send
    n = write(self._handle, buf)

During handling of the above exception, another exception occurred:

  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
BrokenPipeError: [Errno 32] Broken pipe
Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 67, in run
    return super().run()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 192, in do_work
    self.task_queue.task_done()
  File "&lt;string&gt;", line 2, in task_done
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py", line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 416, in _send_bytes
    self._send(header + buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 373, in _send
    n = write(self._handle, buf)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 67, in run
    return super().run()
BrokenPipeError: [Errno 32] Broken pipe
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 192, in do_work
    self.task_queue.task_done()
  File "&lt;string&gt;", line 2, in task_done
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py", line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 416, in _send_bytes
    self._send(header + buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 373, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 190, in do_work
    self.execute_work(key=key, command=command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 190, in do_work
    self.execute_work(key=key, command=command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 190, in do_work
    self.execute_work(key=key, command=command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 190, in do_work
    self.execute_work(key=key, command=command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 190, in do_work
    self.execute_work(key=key, command=command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 190, in do_work
    self.execute_work(key=key, command=command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 190, in do_work
    self.execute_work(key=key, command=command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py", line 271, in sigint_handler
    sys.exit(0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py", line 271, in sigint_handler
    sys.exit(0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
SystemExit: 0
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py", line 271, in sigint_handler
    sys.exit(0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
SystemExit: 0
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py", line 271, in sigint_handler
    sys.exit(0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py", line 271, in sigint_handler
    sys.exit(0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py", line 271, in sigint_handler
    sys.exit(0)

During handling of the above exception, another exception occurred:

  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py", line 271, in sigint_handler
    sys.exit(0)
SystemExit: 0

During handling of the above exception, another exception occurred:

SystemExit: 0
Traceback (most recent call last):
Traceback (most recent call last):
SystemExit: 0
Traceback (most recent call last):
SystemExit: 0

During handling of the above exception, another exception occurred:


During handling of the above exception, another exception occurred:

SystemExit: 0
Traceback (most recent call last):

During handling of the above exception, another exception occurred:


During handling of the above exception, another exception occurred:

Traceback (most recent call last):
Traceback (most recent call last):

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 190, in do_work
    self.execute_work(key=key, command=command)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 67, in run
    return super().run()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 190, in do_work
    self.execute_work(key=key, command=command)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 67, in run
    return super().run()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 67, in run
    return super().run()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 84, in execute_work
    state = self._execute_work_in_fork(command)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py", line 271, in sigint_handler
    sys.exit(0)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 67, in run
    return super().run()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 192, in do_work
    self.task_queue.task_done()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 67, in run
    return super().run()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 67, in run
    return super().run()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 102, in _execute_work_in_fork
    pid, ret = os.waitpid(pid, 0)
SystemExit: 0
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 192, in do_work
    self.task_queue.task_done()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 67, in run
    return super().run()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "&lt;string&gt;", line 2, in task_done
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 192, in do_work
    self.task_queue.task_done()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/utils/cli.py", line 271, in sigint_handler
    sys.exit(0)
  File "&lt;string&gt;", line 2, in task_done
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)

During handling of the above exception, another exception occurred:

  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 192, in do_work
    self.task_queue.task_done()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py", line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "&lt;string&gt;", line 2, in task_done
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 192, in do_work
    self.task_queue.task_done()
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 192, in do_work
    self.task_queue.task_done()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py", line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 192, in do_work
    self.task_queue.task_done()
Traceback (most recent call last):
  File "&lt;string&gt;", line 2, in task_done
SystemExit: 0
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py", line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "&lt;string&gt;", line 2, in task_done
  File "&lt;string&gt;", line 2, in task_done
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "&lt;string&gt;", line 2, in task_done
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py", line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 416, in _send_bytes
    self._send(header + buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))

During handling of the above exception, another exception occurred:

  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py", line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py", line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 416, in _send_bytes
    self._send(header + buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py", line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 373, in _send
    n = write(self._handle, buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 416, in _send_bytes
    self._send(header + buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 416, in _send_bytes
    self._send(header + buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 416, in _send_bytes
    self._send(header + buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 373, in _send
    n = write(self._handle, buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 416, in _send_bytes
    self._send(header + buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 373, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe
BrokenPipeError: [Errno 32] Broken pipe
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 67, in run
    return super().run()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 373, in _send
    n = write(self._handle, buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 373, in _send
    n = write(self._handle, buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 416, in _send_bytes
    self._send(header + buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 373, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
BrokenPipeError: [Errno 32] Broken pipe
BrokenPipeError: [Errno 32] Broken pipe
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 373, in _send
    n = write(self._handle, buf)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 67, in run
    return super().run()
BrokenPipeError: [Errno 32] Broken pipe
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 192, in do_work
    self.task_queue.task_done()
BrokenPipeError: [Errno 32] Broken pipe
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "&lt;string&gt;", line 2, in task_done
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/executors/local_executor.py", line 192, in do_work
    self.task_queue.task_done()
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py", line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "&lt;string&gt;", line 2, in task_done
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/managers.py", line 809, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 416, in _send_bytes
    self._send(header + buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 211, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 373, in _send
    n = write(self._handle, buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 416, in _send_bytes
    self._send(header + buf)
  File "/opt/conda/envs/production/lib/python3.9/multiprocessing/connection.py", line 373, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe
BrokenPipeError: [Errno 32] Broken pipe
</code></pre></div>
</details>
<p dir="auto">When I tried to run this backfill on our production Airflow deployment, I got the following error instead:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1799, in _execute_context
    self.dialect.do_executemany(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py&quot;, line 953, in do_executemany
    context._psycopg2_fetched_rows = xtras.execute_values(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/psycopg2/extras.py&quot;, line 1270, in execute_values
    cur.execute(b''.join(parts))
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint &quot;task_instance_pkey&quot;
DETAIL:  Key (dag_id, task_id, run_id, map_index)=(test_backfill_races, t2, backfill__2022-01-04T00:00:00-06:00, 24) already exists.


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py&quot;, line 1048, in _create_task_instances
    session.bulk_save_objects(tasks)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 3627, in bulk_save_objects
    self._bulk_save_mappings(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 3843, in _bulk_save_mappings
    transaction.rollback(_capture_exception=True)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/langhelpers.py&quot;, line 70, in __exit__
    compat.raise_(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/compat.py&quot;, line 207, in raise_
    raise exception
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 3831, in _bulk_save_mappings
    persistence._bulk_insert(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py&quot;, line 107, in _bulk_insert
    _emit_insert_statements(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py&quot;, line 1097, in _emit_insert_statements
    c = connection._execute_20(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1631, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/sql/elements.py&quot;, line 325, in _execute_on_connection
    return connection._execute_clauseelement(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1498, in _execute_clauseelement
    ret = self._execute_context(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1862, in _execute_context
    self._handle_dbapi_exception(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 2043, in _handle_dbapi_exception
    util.raise_(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/compat.py&quot;, line 207, in raise_
    raise exception
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py&quot;, line 1799, in _execute_context
    self.dialect.do_executemany(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py&quot;, line 953, in do_executemany
    context._psycopg2_fetched_rows = xtras.execute_values(
  File &quot;/opt/conda/envs/production/lib/python3.9/site-packages/psycopg2/extras.py&quot;, line 1270, in execute_values
    cur.execute(b''.join(parts))
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint &quot;task_instance_pkey&quot;
DETAIL:  Key (dag_id, task_id, run_id, map_index)=(test_backfill_races, t2, backfill__2022-01-04T00:00:00-06:00, 24) already exists.

[SQL: INSERT INTO task_instance (task_id, dag_id, run_id, map_index, try_number, max_tries, hostname, unixname, pool, pool_slots, queue, priority_weight, operator, executor_config) VALUES (%(task_id)s, %(dag_id)s, %(run_id)s, %(map_index)s, %(try_number)s, %(max_tries)s, %(hostname)s, %(unixname)s, %(pool)s, %(pool_slots)s, %(queue)s, %(priority_weight)s, %(operator)s, %(executor_config)s)]
[parameters: ({'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-01-04T00:00:00-06:00', 'map_index': 24, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tradebot', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7f2522faf2d0&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-01-04T00:00:00-06:00', 'map_index': 25, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tradebot', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7f2522faf0f0&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-01-04T00:00:00-06:00', 'map_index': 26, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tradebot', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7f2522faf8d0&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-01-04T00:00:00-06:00', 'map_index': 27, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tradebot', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7f2522fafbd0&gt;})]
(Background on this error at: https://sqlalche.me/e/14/gkpj)"><pre class="notranslate"><code class="notranslate">Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1799, in _execute_context
    self.dialect.do_executemany(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py", line 953, in do_executemany
    context._psycopg2_fetched_rows = xtras.execute_values(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/psycopg2/extras.py", line 1270, in execute_values
    cur.execute(b''.join(parts))
psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "task_instance_pkey"
DETAIL:  Key (dag_id, task_id, run_id, map_index)=(test_backfill_races, t2, backfill__2022-01-04T00:00:00-06:00, 24) already exists.


The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/opt/conda/envs/production/lib/python3.9/site-packages/airflow/models/dagrun.py", line 1048, in _create_task_instances
    session.bulk_save_objects(tasks)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 3627, in bulk_save_objects
    self._bulk_save_mappings(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 3843, in _bulk_save_mappings
    transaction.rollback(_capture_exception=True)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
    compat.raise_(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
    raise exception
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 3831, in _bulk_save_mappings
    persistence._bulk_insert(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py", line 107, in _bulk_insert
    _emit_insert_statements(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/orm/persistence.py", line 1097, in _emit_insert_statements
    c = connection._execute_20(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1631, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/sql/elements.py", line 325, in _execute_on_connection
    return connection._execute_clauseelement(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1498, in _execute_clauseelement
    ret = self._execute_context(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1862, in _execute_context
    self._handle_dbapi_exception(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 2043, in _handle_dbapi_exception
    util.raise_(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 207, in raise_
    raise exception
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1799, in _execute_context
    self.dialect.do_executemany(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py", line 953, in do_executemany
    context._psycopg2_fetched_rows = xtras.execute_values(
  File "/opt/conda/envs/production/lib/python3.9/site-packages/psycopg2/extras.py", line 1270, in execute_values
    cur.execute(b''.join(parts))
sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "task_instance_pkey"
DETAIL:  Key (dag_id, task_id, run_id, map_index)=(test_backfill_races, t2, backfill__2022-01-04T00:00:00-06:00, 24) already exists.

[SQL: INSERT INTO task_instance (task_id, dag_id, run_id, map_index, try_number, max_tries, hostname, unixname, pool, pool_slots, queue, priority_weight, operator, executor_config) VALUES (%(task_id)s, %(dag_id)s, %(run_id)s, %(map_index)s, %(try_number)s, %(max_tries)s, %(hostname)s, %(unixname)s, %(pool)s, %(pool_slots)s, %(queue)s, %(priority_weight)s, %(operator)s, %(executor_config)s)]
[parameters: ({'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-01-04T00:00:00-06:00', 'map_index': 24, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tradebot', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7f2522faf2d0&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-01-04T00:00:00-06:00', 'map_index': 25, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tradebot', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7f2522faf0f0&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-01-04T00:00:00-06:00', 'map_index': 26, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tradebot', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7f2522faf8d0&gt;}, {'task_id': 't2', 'dag_id': 'test_backfill_races', 'run_id': 'backfill__2022-01-04T00:00:00-06:00', 'map_index': 27, 'try_number': 0, 'max_tries': 0, 'hostname': '', 'unixname': 'tradebot', 'pool': 'default_pool', 'pool_slots': 1, 'queue': 'default', 'priority_weight': 1, 'operator': '_PythonDecoratedOperator', 'executor_config': &lt;psycopg2.extensions.Binary object at 0x7f2522fafbd0&gt;})]
(Background on this error at: https://sqlalche.me/e/14/gkpj)
</code></pre></div>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">Mapped tasks should be able to be backfilled like "normal" tasks.</p>
<h3 dir="auto">How to reproduce</h3>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="#!/usr/bin/env python3
import datetime
import itertools
import logging
import string

from airflow.decorators import dag, task


logger = logging.getLogger(__name__)

TASK_COUNT = 70  # turning this up increases the likelihood of issues


@dag(
    schedule_interval='@daily',
    start_date=datetime.datetime(2022, 8, 15),
    default_args={
        'retries': 0,
    },
)
def test_backfill_races():
    @task
    def get_tasks():
        return list(itertools.islice(itertools.cycle(string.ascii_letters), TASK_COUNT))

    @task
    def t1(arg):
        logger.info(f'{arg=}')
        return arg

    @task
    def t2(arg):
        logger.info(f'{arg=}')


    t2.expand(arg=t1.expand(arg=get_tasks()))


dag = test_backfill_races()


if __name__ == '__main__':
    dag.cli()"><pre class="notranslate"><code class="notranslate">#!/usr/bin/env python3
import datetime
import itertools
import logging
import string

from airflow.decorators import dag, task


logger = logging.getLogger(__name__)

TASK_COUNT = 70  # turning this up increases the likelihood of issues


@dag(
    schedule_interval='@daily',
    start_date=datetime.datetime(2022, 8, 15),
    default_args={
        'retries': 0,
    },
)
def test_backfill_races():
    @task
    def get_tasks():
        return list(itertools.islice(itertools.cycle(string.ascii_letters), TASK_COUNT))

    @task
    def t1(arg):
        logger.info(f'{arg=}')
        return arg

    @task
    def t2(arg):
        logger.info(f'{arg=}')


    t2.expand(arg=t1.expand(arg=get_tasks()))


dag = test_backfill_races()


if __name__ == '__main__':
    dag.cli()
</code></pre></div>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="airflow dags backfill test_backfill_races -s 2022-08-01 -e 2022-08-10"><pre class="notranslate"><code class="notranslate">airflow dags backfill test_backfill_races -s 2022-08-01 -e 2022-08-10
</code></pre></div>
<h3 dir="auto">Operating System</h3>
<p dir="auto">CentOS Stream 8</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">None</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other</p>
<h3 dir="auto">Deployment details</h3>
<ol dir="auto">
<li>
<p dir="auto">Standalone w/ Postgres DB backend.  Relevant configs:<br>
a. <code class="notranslate">core.executor = LocalExecutor</code><br>
b. <code class="notranslate">core.parallelism = 12</code><br>
c. <code class="notranslate">core.max_active_tasks_per_dag = 16</code><br>
d. <code class="notranslate">core.max_active_runs_per_dag = 16</code></p>
</li>
<li>
<p dir="auto">Self-hosted w/ Postgres DB backend.  Relevant configs:<br>
a. <code class="notranslate">core.executor = CeleryExecutor</code><br>
b. <code class="notranslate">core.parallelism = 256</code><br>
c. <code class="notranslate">core.max_active_tasks_per_dag = 128</code><br>
d. <code class="notranslate">core.max_active_runs_per_dag = 16</code><br>
e. <code class="notranslate">celery.worker_concurrency = 16</code></p>
</li>
</ol>
<h3 dir="auto">Anything else</h3>
<p dir="auto">Possibly related: <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="944046171" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/16982" data-hovercard-type="issue" data-hovercard-url="/apache/airflow/issues/16982/hovercard" href="https://github.com/apache/airflow/issues/16982">#16982</a></p>
<p dir="auto">Turning down the effective parallelism of the DAG (e.g. <code class="notranslate">max_active_tasks=4</code>) sometimes helps to avoid this, but (obviously) puts a significant bottleneck on the backfill.</p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">Other Airflow 2 version (please specify below)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">Airflow 2.4.3, running on Kubernetes with High-Availability Scheduler (2 pods). Database is postgres 10.21</p>
<p dir="auto">I have a dag that has been running for two months with this timetable: <code class="notranslate">timetable=CronTriggerTimetable("40 5 * * *", timezone = 'UTC')</code> (which has not changed). It has run reliably every day as expected however yesterday it simply did not run. I have checked for errors and there were none, and checking the scheduler logs (on both pods) it appears no evaluation of the dag run was done in either scheduler instance. in the cli inspecting the next 60 run execution dates indicates there is no problem with the timetable's interpretation</p>
<p dir="auto">My suspiscion is that a race condition in the <code class="notranslate">SELECT ... FOR UPDATE</code> of the database may have prevented either scheduler executing the DAG. Is this possible? Otherwise how can this have happened?</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">The dag should have run as it always has previously</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">I wish I knew! I can't work it out.</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">RedHat Enterprise Linux 8</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">2.4.3</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other Docker-based deployment</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">High availability schedulers (n=2) on OpenShift Kubernetes. Postgres database (version 10.21</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">This has happened once before, just under a month ago, on the same DAG.</p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>