<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.2.5<br>
Running a DAG and got this error for a task.<br>
[2023-02-22, 09:38:16 EST] {standard_task_runner.py:79} INFO - Running: ['airflow', 'tasks', 'run', 'icwg_fastqsr_ftp_service_volume', 'find_service_volume_for_fastqsr', 'manual__2023-02-22T14:38:11.752718+00:00', '--job-id', '36461', '--raw', '--subdir', 'DAGS_FOLDER/icwg_fastqsr/icwg_fastqsr_ftp_service_volume.py', '--cfg-path', '/tmp/tmp5lgl246c', '--error-file', '/tmp/tmpc2q98hw9']<br>
[2023-02-22, 09:38:16 EST] {standard_task_runner.py:80} INFO - Job 36461: Subtask find_service_volume_for_fastqsr<br>
[2023-02-22, 09:38:17 EST] {task_command.py:298} INFO - Running &lt;TaskInstance: icwg_fastqsr_ftp_service_volume.find_service_volume_for_fastqsr manual__2023-02-22T14:38:11.752718+00:00 [running]&gt; on host airflow-worker-vtgg4<br>
[2023-02-22, 09:38:17 EST] {taskinstance.py:1745} WARNING - We expected to get frame set in local storage but it was not. Please report this as an issue with full logs at <a href="https://github.com/apache/airflow/issues/new">https://github.com/apache/airflow/issues/new</a><br>
Traceback (most recent call last):</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">No idea as the same DAG was running fine in another composer in same project. We just copied this DAG from there and ran first time in order to migrate the DAG to this composer.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Copy a DAG with Big Query operator  from one composer to another running in same project and start running it.</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Ubuntu</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">Big Query Operator related</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Google Cloud Composer</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">Airflow on Google Cloud Composer 2.1.3</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">NA</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">2.4.1</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">Running <code class="notranslate">airflow db upgrade</code> on an Airflow installation with 100 DAGs fails with this error:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="ERROR [airflow.models.dagbag.DagBag] Failed to write serialized DAG: /usr/local/airflow/dags/REDACTED.py
Traceback (most recent call last):
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/models/dagbag.py&quot;, line 615, in _serialize_dag_capturing_errors
    dag_was_updated = SerializedDagModel.write_dag(
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 72, in wrapper
    return func(*args, **kwargs)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/models/serialized_dag.py&quot;, line 146, in write_dag
    session.query(literal(True))
  File &quot;/usr/local/lib/python3.9/site-packages/sqlalchemy/orm/query.py&quot;, line 2810, in first
    return self.limit(1)._iter().first()
  File &quot;/usr/local/lib/python3.9/site-packages/sqlalchemy/orm/query.py&quot;, line 2894, in _iter
    result = self.session.execute(
  File &quot;/usr/local/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 1688, in execute
    conn = self._connection_for_bind(bind)
  File &quot;/usr/local/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 1529, in _connection_for_bind
    return self._transaction._connection_for_bind(
  File &quot;/usr/local/lib/python3.9/site-packages/sqlalchemy/orm/session.py&quot;, line 721, in _connection_for_bind
    self._assert_active()
  File &quot;/usr/local/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.UniqueViolation) duplicate key value violates unique constraint &quot;serialized_dag_pkey&quot;
DETAIL:  Key (dag_id)=(REDACTED) already exists.

[SQL: INSERT INTO serialized_dag (dag_id, fileloc, fileloc_hash, data, data_compressed, last_updated, dag_hash, ..."><pre class="notranslate"><code class="notranslate">ERROR [airflow.models.dagbag.DagBag] Failed to write serialized DAG: /usr/local/airflow/dags/REDACTED.py
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/airflow/models/dagbag.py", line 615, in _serialize_dag_capturing_errors
    dag_was_updated = SerializedDagModel.write_dag(
  File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py", line 72, in wrapper
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/serialized_dag.py", line 146, in write_dag
    session.query(literal(True))
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/orm/query.py", line 2810, in first
    return self.limit(1)._iter().first()
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/orm/query.py", line 2894, in _iter
    result = self.session.execute(
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 1688, in execute
    conn = self._connection_for_bind(bind)
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 1529, in _connection_for_bind
    return self._transaction._connection_for_bind(
  File "/usr/local/lib/python3.9/site-packages/sqlalchemy/orm/session.py", line 721, in _connection_for_bind
    self._assert_active()
  File "/usr/local/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.UniqueViolation) duplicate key value violates unique constraint "serialized_dag_pkey"
DETAIL:  Key (dag_id)=(REDACTED) already exists.

[SQL: INSERT INTO serialized_dag (dag_id, fileloc, fileloc_hash, data, data_compressed, last_updated, dag_hash, ...
</code></pre></div>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto"><code class="notranslate">airflow db upgrade</code> should successfully reserialize DAGs at the end of the upgrade just like the <code class="notranslate">airflow dags reserialize</code> command.</p>
<h3 dir="auto">How to reproduce</h3>
<ol dir="auto">
<li>Upgrade to <code class="notranslate">airflow 2.4.1</code> on an existing codebase</li>
<li>Run <code class="notranslate">airflow db upgrade</code></li>
</ol>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian GNU/Linux 10 (buster)</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="apache-airflow-providers-amazon==5.1.0
apache-airflow-providers-celery==3.0.0
apache-airflow-providers-cncf-kubernetes==4.3.0
apache-airflow-providers-common-sql==1.2.0
apache-airflow-providers-datadog==3.0.0
apache-airflow-providers-ftp==3.1.0
apache-airflow-providers-http==4.0.0
apache-airflow-providers-imap==3.0.0
apache-airflow-providers-postgres==5.2.1
apache-airflow-providers-redis==3.0.0
apache-airflow-providers-sendgrid==3.0.0
apache-airflow-providers-sftp==4.0.0
apache-airflow-providers-slack==5.1.0
apache-airflow-providers-sqlite==3.2.1
apache-airflow-providers-ssh==3.1.0"><pre class="notranslate"><code class="notranslate">apache-airflow-providers-amazon==5.1.0
apache-airflow-providers-celery==3.0.0
apache-airflow-providers-cncf-kubernetes==4.3.0
apache-airflow-providers-common-sql==1.2.0
apache-airflow-providers-datadog==3.0.0
apache-airflow-providers-ftp==3.1.0
apache-airflow-providers-http==4.0.0
apache-airflow-providers-imap==3.0.0
apache-airflow-providers-postgres==5.2.1
apache-airflow-providers-redis==3.0.0
apache-airflow-providers-sendgrid==3.0.0
apache-airflow-providers-sftp==4.0.0
apache-airflow-providers-slack==5.1.0
apache-airflow-providers-sqlite==3.2.1
apache-airflow-providers-ssh==3.1.0
</code></pre></div>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Other Docker-based deployment</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">k8s deployment</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">Fails consistently in these two scenarios:</p>
<ol dir="auto">
<li>
<p dir="auto">Run db upgrade only:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content=" airflow db upgrade"><pre class="notranslate"><code class="notranslate"> airflow db upgrade
</code></pre></div>
</li>
<li>
<p dir="auto">Run along with reserialize</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content=" airflow dags reserialize --clear-only
 airflow db upgrade"><pre class="notranslate"><code class="notranslate"> airflow dags reserialize --clear-only
 airflow db upgrade
</code></pre></div>
</li>
</ol>
<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" checked=""> 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>