Issue 1:
Apache Airflow version

2.3.3

What happened

stuck in the middle of airflow db upgrade in one of my Airflow’s clusters.

I am trying to upgrade from 2.2.3 --> 2.3.3, using MySql.

Upgrade failed in Running upgrade c97c2ab6aa23 -> 4eaab2fe6582, Migrate RTIF to use run_id and map_index.

Actually, it also failed before that (on previous steps) and I had to run some commands manually to proceed until this point. Here is what I have done so far:

USE airflow;
ALTER TABLE task_reschedule
ADD CONSTRAINT `task_reschedule_ti_fkey` FOREIGN KEY (`dag_id`, `task_id`, `run_id`) REFERENCES `task_instance` (`dag_id`, `task_id`, `run_id`) ON DELETE CASCADE;
ALTER TABLE task_reschedule
ADD KEY `idx_task_reschedule_dag_task_run` (`dag_id`,`task_id`,`run_id`);
ALTER TABLE `task_instance`
ADD PRIMARY KEY (`dag_id`,`task_id`,`run_id`);
ALTER TABLE task_instance DROP COLUMN map_index;
ALTER TABLE task_reschedule DROP COLUMN map_index;
ALTER TABLE rendered_task_instance_fields DROP COLUMN map_index;
ALTER TABLE task_instance MODIFY `task_id` VARCHAR(250) CHARACTER SET utf8 COLLATE utf8_bin;
ALTER TABLE task_reschedule MODIFY `task_id` VARCHAR(250) CHARACTER SET utf8 COLLATE utf8_bin;
ALTER TABLE rendered_task_instance_fields MODIFY `dag_id` VARCHAR(250) CHARACTER SET utf8 COLLATE utf8_bin;

The log in this step:

INFO [alembic.runtime.migration] Context impl MySQLImpl.
INFO [alembic.runtime.migration] Will assume non-transactional DDL.
INFO [alembic.runtime.migration] Running upgrade c97c2ab6aa23 -> 4eaab2fe6582, Migrate RTIF to use run_id and map_index
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1705, in _execute_context
self.dialect.do_execute(
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute
cursor.execute(statement, parameters)
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/cursors.py", line 148, in execute
result = self._query(query)
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/cursors.py", line 310, in _query
conn.query(q)
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 548, in query
self._affected_rows = self._read_query_result(unbuffered=unbuffered)
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 775, in _read_query_result
result.read()
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 1156, in read
first_packet = self.connection._read_packet()
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 725, in _read_packet
packet.raise_for_error()
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/protocol.py", line 221, in raise_for_error
err.raise_mysql_exception(self._data)
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/err.py", line 143, in raise_mysql_exception
raise errorclass(errno, errval)
pymysql.err.OperationalError: (1060, "Duplicate column name 'run_id'")

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

Traceback (most recent call last):
File "/home/airflow/.local/bin/airflow", line 8, in <module>
sys.exit(main())
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/__main__.py", line 38, in main
args.func(args)
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/cli_parser.py", line 51, in command
return func(*args, **kwargs)
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/cli.py", line 99, in wrapper
return f(*args, **kwargs)
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/cli/commands/db_command.py", line 82, in upgradedb
db.upgradedb(to_revision=to_revision, from_revision=from_revision, show_sql_only=args.show_sql_only)
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/session.py", line 71, in wrapper
return func(*args, session=session, **kwargs)
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/utils/db.py", line 1463, in upgradedb
command.upgrade(config, revision=to_revision or 'heads')
File "/home/airflow/.local/lib/python3.9/site-packages/alembic/command.py", line 322, in upgrade
script.run_env()
File "/home/airflow/.local/lib/python3.9/site-packages/alembic/script/base.py", line 569, in run_env
util.load_python_file(self.dir, "env.py")
File "/home/airflow/.local/lib/python3.9/site-packages/alembic/util/pyfiles.py", line 94, in load_python_file
module = load_module_py(module_id, path)
File "/home/airflow/.local/lib/python3.9/site-packages/alembic/util/pyfiles.py", line 110, in load_module_py
spec.loader.exec_module(module) # type: ignore
File "<frozen importlib._bootstrap_external>", line 850, in exec_module
File "<frozen importlib._bootstrap>", line 228, in _call_with_frames_removed
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/migrations/env.py", line 107, in <module>
run_migrations_online()
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/migrations/env.py", line 101, in run_migrations_online
context.run_migrations()
File "<string>", line 8, in run_migrations
File "/home/airflow/.local/lib/python3.9/site-packages/alembic/runtime/environment.py", line 853, in run_migrations
self.get_context().run_migrations(**kw)
File "/home/airflow/.local/lib/python3.9/site-packages/alembic/runtime/migration.py", line 623, in run_migrations
step.migration_fn(**kw)
File "/home/airflow/.local/lib/python3.9/site-packages/airflow/migrations/versions/0104_2_3_0_migrate_rtif_to_use_run_id_and_map_index.py", line 118, in upgrade
batch_op.add_column(sa.Column('run_id', type_=StringID(), nullable=True))
File "/usr/local/lib/python3.9/contextlib.py", line 126, in __exit__
next(self.gen)
File "/home/airflow/.local/lib/python3.9/site-packages/alembic/operations/base.py", line 376, in batch_alter_table
impl.flush()
File "/home/airflow/.local/lib/python3.9/site-packages/alembic/operations/batch.py", line 111, in flush
fn(*arg, **kw)
File "/home/airflow/.local/lib/python3.9/site-packages/alembic/ddl/impl.py", line 322, in add_column
self._exec(base.AddColumn(table_name, column, schema=schema))
File "/home/airflow/.local/lib/python3.9/site-packages/alembic/ddl/impl.py", line 195, in _exec
return conn.execute(construct, multiparams)
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1200, in execute
return meth(self, multiparams, params, _EMPTY_EXECUTION_OPTS)
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/sql/ddl.py", line 77, in _execute_on_connection
return connection._execute_ddl(
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1290, in _execute_ddl
ret = self._execute_context(
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1748, in _execute_context
self._handle_dbapi_exception(
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1929, in _handle_dbapi_exception
util.raise_(
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
raise exception
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 1705, in _execute_context
self.dialect.do_execute(
File "/home/airflow/.local/lib/python3.9/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute
cursor.execute(statement, parameters)
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/cursors.py", line 148, in execute
result = self._query(query)
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/cursors.py", line 310, in _query
conn.query(q)
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 548, in query
self._affected_rows = self._read_query_result(unbuffered=unbuffered)
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 775, in _read_query_result
result.read()
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 1156, in read
first_packet = self.connection._read_packet()
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/connections.py", line 725, in _read_packet
packet.raise_for_error()
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/protocol.py", line 221, in raise_for_error
err.raise_mysql_exception(self._data)
File "/home/airflow/.local/lib/python3.9/site-packages/pymysql/err.py", line 143, in raise_mysql_exception
raise errorclass(errno, errval)
sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (1060, "Duplicate column name 'run_id'")
[SQL: ALTER TABLE rendered_task_instance_fields ADD COLUMN run_id VARCHAR(250) COLLATE utf8mb3_bin]
(Background on this error at: http://sqlalche.me/e/14/e3q8)

The problem is that the table’s schema now looks like:

CREATE TABLE `rendered_task_instance_fields` (
`dag_id` varchar(250) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL,
`task_id` varchar(250) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL,
`rendered_fields` json NOT NULL,
`k8s_pod_yaml` json DEFAULT NULL,
`run_id` varchar(250) CHARACTER SET utf8 COLLATE utf8_bin NOT NULL,
PRIMARY KEY (`dag_id`,`task_id`,`run_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

Seems like the execution_date was dropped from rendered_task_instance_fields table. I don't see this column now.

To ensure it happened during the migration, I tried to bring up Airflow 2.2.5 in docker - and I see the execution_date. Then, I tried to bring up Airflow 2.3.3 in docker - and I DON'T see this column.

What you think should happen instead

The migration should succeed smoothly as happened in the previous Airflow upgrades.

How to reproduce

I guess you should have data in MySql tables created by version 2.2.3 (and before, we are running Airflow since 1.10.x) and try to upgrade to 2.3.3

Operating System

using docker image - apache/airflow:2.3.3-python3.9

Versions of Apache Airflow Providers

No response

Deployment

Other 3rd-party Helm chart

Deployment details

running MySql 8 in RDS (AWS)

Anything else

No response

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: 1.7.1.2

Kubernetes version (if you are using kubernetes) (use kubectl version):

Environment:

Cloud provider or hardware configuration:

OS (e.g. from /etc/os-release):

Kernel (e.g. uname -a):

Install tools:

Others:

What happened:

Environment setup

We are running Airflow 1.7.1.2 with MySQL 5.6. The wait_timeout of MySQL is set at 300 seconds, which means that idle connections will go away after 300 seconds of inactivity. To reflect this, we set SQLAlchemy's pool_recycle in airflow.cfg to 290 seconds, which should force Airflow/SQLAlchemy to recycle/discard connections after 290 seconds. Thus Airflow shouldn't try to use an already-dead connection.

Symptom

When running a PythonOperator that takes more than 300 seconds to execute, the task would finish executing the Python callable, but ends up with error: https://gist.github.com/garthcn/cd7bcdec12748406506f2b0710655c8b

It seems that after the Python callable finishes executing, Airflow tries to push its return value to XCom. However, the SQL connection has gone away while Airflow/SQLAlchemy think it's still there.

Hypothesis

I did some investigation and think that it might be caused by not calling session.commit() or session.close() for the DB operations before the XCom push. As far as I know, in SQLAlchemy, if you don't close a connection and let it be in checked-out state, it won't be recycled by connection pool, and thus SQLAlchemy will try to use it again after >300 seconds (which is the wait_timeout for MySQL in our case). This will result in a "MySQL connection has gone away" issue.

It seems that Airflow codebase uses @provide_context decorator to help with session open/close, and my hunch is that some functions are not using it or misusing it.

What you expected to happen:

How to reproduce it:

Anything else we need to know:

Moved here from https://issues.apache.org/jira/browse/AIRFLOW-405