Issue 1:
Apache Airflow version

Other Airflow 2 version (please specify below)

What happened

Airflow 2.2.5

Running a DAG and got this error for a task.

[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']

[2023-02-22, 09:38:16 EST] {standard_task_runner.py:80} INFO - Job 36461: Subtask find_service_volume_for_fastqsr

[2023-02-22, 09:38:17 EST] {task_command.py:298} INFO - Running <TaskInstance: icwg_fastqsr_ftp_service_volume.find_service_volume_for_fastqsr manual__2023-02-22T14:38:11.752718+00:00 [running]> on host airflow-worker-vtgg4

[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 https://github.com/apache/airflow/issues/new

Traceback (most recent call last):

What you think should happen instead

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.

How to reproduce

Copy a DAG with Big Query operator from one composer to another running in same project and start running it.

Operating System

Ubuntu

Versions of Apache Airflow Providers

Big Query Operator related

Deployment

Google Cloud Composer

Deployment details

Airflow on Google Cloud Composer 2.1.3

Anything else

NA

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Issue 2:
Apache Airflow version

2.4.1

What happened

Running airflow db upgrade on an Airflow installation with 100 DAGs fails with this error:

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, ...

What you think should happen instead

airflow db upgrade should successfully reserialize DAGs at the end of the upgrade just like the airflow dags reserialize command.

How to reproduce

Upgrade to airflow 2.4.1 on an existing codebase

Run airflow db upgrade

Operating System

Debian GNU/Linux 10 (buster)

Versions of Apache Airflow Providers

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

Deployment

Other Docker-based deployment

Deployment details

k8s deployment

Anything else

Fails consistently in these two scenarios:

Run db upgrade only:

airflow db upgrade

Run along with reserialize

airflow dags reserialize --clear-only
airflow db upgrade

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct