Issue 1:
Apache Airflow version: 1.8.0

Ticket was created 23/Nov/16 01:12

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:

We ported over a number of our cronjobs to run using Airflow. To achieve the desired behavior, we use the LatestOnlyOperator that was merged to master by pull request 1752: #1752.

When we moved over our cronjobs, we migrated many at a time (using ansible). These DAGs had a start date going back a few days.

The first thing I noticed is that it seemed to take a long time to process the backfilled DAGs. They were being processed correctly, in the sense that the 'latest_only' operator was being completely successfully, and the downstream BashOperator was marked as skipped. It also appeared that the DAG run completed successfully in the tree view. However, when I searched the DAG runs for state contains running, I saw that they were present.

One thing I noticed was that in the logs for one of the "stuck" DAG runs, it appeared that the 'latest_only' task was processed multiple times.

[2016-11-22 12:26:27,701] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:28:50,335] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:28:53,288] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:28:58,400] {models.py:1196} INFO -

Starting attempt 1 of 1

[2016-11-22 12:28:59,334] {models.py:1219} INFO - Executing <Task(LatestOnlyOperator): latest_only> on 2016-11-20 04:00:00

[2016-11-22 12:29:00,671] {airflow_next.py:27} INFO - Checking latest only with left_window: 2016-11-21 04:00:00 right_window: 2016-11-22 04:00:00 now: 2016-11-22 12:29:00.670321

[2016-11-22 12:29:00,671] {airflow_next.py:29} INFO - Not latest execution, skipping downstream.

[2016-11-22 12:29:00,672] {airflow_next.py:34} INFO - Skipping task: my_dag

[2016-11-22 12:29:01,397] {airflow_next.py:41} INFO - Done.

[2016-11-22 12:31:13,055] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:31:17,899] {models.py:1150} INFO - Task <TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]> previously succeeded on 2016-11-22 12:29:01.397870

[2016-11-22 12:32:31,907] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:34:56,522] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:35:00,975] {models.py:1150} INFO - Task <TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]> previously succeeded on 2016-11-22 12:29:01.397870

[2016-11-22 12:35:36,323] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:38:00,140] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:38:05,057] {models.py:1150} INFO - Task <TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]> previously succeeded on 2016-11-22 12:29:01.397870

[2016-11-22 12:38:50,014] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:41:07,609] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:41:12,232] {models.py:1150} INFO - Task <TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]> previously succeeded on 2016-11-22 12:29:01.397870

[2016-11-22 12:41:45,857] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:44:05,354] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:44:09,635] {models.py:1150} INFO - Task <TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]> previously succeeded on 2016-11-22 12:29:01.397870

[2016-11-22 12:44:30,851] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:46:58,977] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:47:02,836] {models.py:1150} INFO - Task <TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]> previously succeeded on 2016-11-22 12:29:01.397870

[2016-11-22 12:48:27,571] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:50:54,034] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:50:57,951] {models.py:1150} INFO - Task <TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]> previously succeeded on 2016-11-22 12:29:01.397870

[2016-11-22 12:51:21,442] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:53:44,461] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:53:48,392] {models.py:1150} INFO - Task <TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]> previously succeeded on 2016-11-22 12:29:01.397870

[2016-11-22 12:54:28,745] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:56:50,740] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:56:54,382] {models.py:1150} INFO - Task <TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]> previously succeeded on 2016-11-22 12:29:01.397870

[2016-11-22 12:57:59,881] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:59:04,245] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

[2016-11-22 12:59:05,666] {models.py:1150} INFO - Task <TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]> previously succeeded on 2016-11-22 12:29:01.397870

[2016-11-22 13:02:18,434] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py

We use CeleryExecutor backed by Redis. When inspecting the Redis key itself that held the queue, I verified that duplicate tasks were in there; there were thousnands of tasks in the queue. Here are the tasks that were at the head of the task list (the dag names are changed for readability):

[

u"['airflow run DAG_01 latest_only 2016-11-18T09:23:00 --local -sd DAGS_FOLDER/DAG_01.py ']",

u"['airflow run DAG_02 latest_only 2016-11-18T03:45:00 --local -sd DAGS_FOLDER/DAG_02.py ']",

u"['airflow run DAG_03 latest_only 2016-11-18T00:08:00 --local -sd DAGS_FOLDER/DAG_03.py ']",

u"['airflow run DAG_04 latest_only 2016-11-22T10:40:00 --local -sd DAGS_FOLDER/DAG_04.py ']",

u"['airflow run DAG_05 latest_only 2016-11-18T05:00:00 --local -sd DAGS_FOLDER/DAG_05.py ']",

u"['airflow run DAG_06 latest_only 2016-11-22T11:20:00 --local -sd DAGS_FOLDER/DAG_06.py ']",

u"['airflow run DAG_07 latest_only 2016-11-18T10:14:00 --local -sd DAGS_FOLDER/DAG_07.py ']",

u"['airflow run DAG_01 latest_only 2016-11-18T09:23:00 --local -sd DAGS_FOLDER/DAG_01.py ']",

u"['airflow run DAG_02 latest_only 2016-11-18T03:45:00 --local -sd DAGS_FOLDER/DAG_02.py ']",

u"['airflow run DAG_03 latest_only 2016-11-18T00:08:00 --local -sd DAGS_FOLDER/DAG_03.py ']",

u"['airflow run DAG_04 latest_only 2016-11-22T10:40:00 --local -sd DAGS_FOLDER/DAG_04.py ']",

u"['airflow run DAG_05 latest_only 2016-11-18T05:00:00 --local -sd DAGS_FOLDER/DAG_05.py ']",

u"['airflow run DAG_06 latest_only 2016-11-22T11:20:00 --local -sd DAGS_FOLDER/DAG_06.py ']",

u"['airflow run DAG_07 latest_only 2016-11-18T10:14:00 --local -sd DAGS_FOLDER/DAG_07.py ']"

]

Grepping the scheduler's logs, here is one of the instances I see of the scheduler enqueuing a duplicate task:

$ grep -A2 "DAG_01" /var/log/airflow/airflow-scheduler.log | grep -A2 "09:23"

[2016-11-22 13:24:26,660] {models.py:2660} INFO - Checking state for <DagRun DAG_01 @ 2016-11-18 09:23:00: scheduled__2016-11-18T09:23:00, externally triggered: False>

[2016-11-22 13:24:26,672] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.

[2016-11-22 13:24:26,678] {jobs.py:514} INFO - Checking dependencies on 2 tasks instances, minus 0 skippable ones

[2016-11-22 13:24:26,726] {base_executor.py:36} INFO - Adding to queue: airflow run DAG_01 latest_only 2016-11-18T09:23:00 --local -sd DAGS_FOLDER/DAG_01.py

[2016-11-22 13:24:26,769] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.

[2016-11-22 13:24:26,769] {jobs.py:514} INFO - Checking dependencies on 0 tasks instances, minus 0 skippable ones

[2016-11-22 13:24:31,830] {models.py:2660} INFO - Checking state for <DagRun DAG_01 @ 2016-11-18 09:23:00: scheduled__2016-11-18T09:23:00, externally triggered: False>

[2016-11-22 13:24:31,832] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.

[2016-11-22 13:24:31,832] {jobs.py:514} INFO - Checking dependencies on 0 tasks instances, minus 0 skippable ones

[2016-11-22 13:24:37,238] {models.py:2660} INFO - Checking state for <DagRun DAG_01 @ 2016-11-18 09:23:00: scheduled__2016-11-18T09:23:00, externally triggered: False>

[2016-11-22 13:24:37,240] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.

[2016-11-22 13:24:37,252] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.

[2016-11-22 13:24:45,736] {models.py:2660} INFO - Checking state for <DagRun DAG_01 @ 2016-11-18 09:23:00: scheduled__2016-11-18T09:23:00, externally triggered: False>

[2016-11-22 13:24:45,744] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.

[2016-11-22 13:24:45,756] {jobs.py:514} INFO - Checking dependencies on 2 tasks instances, minus 0 skippable ones

[2016-11-22 13:24:56,613] {models.py:2660} INFO - Checking state for <DagRun DAG_01 @ 2016-11-18 09:23:00: scheduled__2016-11-18T09:23:00, externally triggered: False>

[2016-11-22 13:24:56,624] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.

[2016-11-22 13:24:56,638] {jobs.py:514} INFO - Checking dependencies on 2 tasks instances, minus 0 skippable ones

[2016-11-22 13:24:56,680] {base_executor.py:36} INFO - Adding to queue: airflow run DAG_01 latest_only 2016-11-18T09:23:00 --local -sd DAGS_FOLDER/DAG_01.py

[2016-11-22 13:24:56,823] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.

[2016-11-22 13:24:56,824] {jobs.py:514} INFO - Checking dependencies on 0 tasks instances, minus 0 skippable ones

Eventually, we ended up just creating new DAG definitions with future start dates and manually cleared the Redis queue.

Additional Context:

Our scheduler is daemonized by upstart and runs with -n 5

Here is the template we use for our cron DAGs (note it's a jinja2 template):

{{ansible_managed}}

from dateutil import parser

from airflow.operators import LatestOnlyOperator

from airflow.operators import BashOperator

from airflow.models import DAG

args = {

'owner': 'airflow',

'start_date': parser.parse('{{item.start_date}}'),

'retries': 0,

}

dag = DAG(

dag_id='{{item.name}}',

default_args=args,

schedule_interval='{{item.schedule}}',

max_active_runs=1,

)

latest_only = LatestOnlyOperator(task_id='latest_only', dag=dag)

script = BashOperator(

task_id='{{item.name}}',

bash_command='{{item.command}}',

default_args=args,

dag=dag,

)

script.set_upstream(latest_only)

One thing to note: We are on Airflow 2.7.1.3; however, we brought in the operator through a plugin:

import datetime

import logging

from airflow.models import BaseOperator, TaskInstance

from airflow.plugins_manager import AirflowPlugin

from airflow.utils.state import State

from airflow import settings

class LatestOnlyOperator(BaseOperator):

"""

Allows a workflow to skip tasks that are not running during the most

recent schedule interval.

If the task is run outside of the latest schedule interval, all

directly downstream tasks will be skipped.

"""

ui_color = '#e9ffdb' # nyanza

def execute(self, context):
now = datetime.datetime.now()
left_window = context['dag'].following_schedule(
context['execution_date'])
right_window = context['dag'].following_schedule(left_window)
logging.info(
'Checking latest only with left_window: %s right_window: %s '
'now: %s', left_window, right_window, now)
if not left_window < now <= right_window:
logging.info('Not latest execution, skipping downstream.')
session = settings.Session()
for task in context['task'].downstream_list:
ti = TaskInstance(
task, execution_date=context['ti'].execution_date)
logging.info('Skipping task: %s', ti.task_id)
ti.state = State.SKIPPED
ti.start_date = now
ti.end_date = now
session.merge(ti)
session.commit()
session.close()
logging.info('Done.')
else:
logging.info('Latest, allowing execution to proceed.')

class AirflowNextPlugin(AirflowPlugin):

name = "airflow_next"

operators = [LatestOnlyOperator]

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-648

Issue 2:
Apache Airflow version

2.3.1

What happened

The execution of of the command "airflow db upgrade" to upgrade the version from 2.2.5 to 2.3.0 or 2.3.1 fails with the following error:

(airflow) [airflow@rabbit01 airflow]$ export LD_LIBRARY_PATH=/opt/python-bit/lib:$LD_LIBRARY_PATH

(airflow) [airflow@rabbit01 airflow]$ airflow db upgrade

DB: mysql+mysqldb://airflow:@localdb.localdomain.it:3306/airflow

Performing upgrade with database mysql+mysqldb://airflow:@localdb.localdomain.it:3306/airflow

[2022-06-06 10:18:48,523] {db.py:1445} INFO - Creating tables

INFO [alembic.runtime.migration] Context impl MySQLImpl.

INFO [alembic.runtime.migration] Will assume non-transactional DDL.

INFO [alembic.runtime.migration] Running upgrade 45ba3f1493b9 -> 849da589634d, Prefix DAG permissions.

Traceback (most recent call last):

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1783, in _cursor_execute

self.dialect.do_execute(cursor, statement, parameters, context)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute

cursor.execute(statement, parameters)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/cursors.py", line 206, in execute

res = self._query(query)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/cursors.py", line 319, in _query

db.query(q)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/connections.py", line 254, in query

_mysql.connection.query(self, query)

MySQLdb._exceptions.ProgrammingError: (1146, "Table 'airflow.ab_permission_id_seq' doesn't exist")

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

Traceback (most recent call last):

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1608, in _execute_context

context = constructor(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 998, in _init_compiled

self._process_executesingle_defaults()

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1866, in _process_executesingle_defaults

val = self.get_insert_default(c)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1811, in get_insert_default

return self._exec_default(column, column.default, column.type)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1672, in exec_default

return self.fire_sequence(default, type)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/dialects/mysql/base.py", line 1401, in fire_sequence

return self._execute_scalar(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1269, in _execute_scalar

conn._cursor_execute(self.cursor, stmt, parameters, context=self)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1785, in _cursor_execute

self.handle_dbapi_exception(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1929, in handle_dbapi_exception

util.raise(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise

raise exception

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1783, in _cursor_execute

self.dialect.do_execute(cursor, statement, parameters, context)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute

cursor.execute(statement, parameters)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/cursors.py", line 206, in execute

res = self._query(query)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/cursors.py", line 319, in _query

db.query(q)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/connections.py", line 254, in query

_mysql.connection.query(self, query)

sqlalchemy.exc.ProgrammingError: (MySQLdb._exceptions.ProgrammingError) (1146, "Table 'airflow.ab_permission_id_seq' doesn't exist")

[SQL: select nextval(ab_permission_id_seq)]

(Background on this error at: http://sqlalche.me/e/14/f405)

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

Traceback (most recent call last):

File "/app/x01/airflow/environments/airflow/bin/airflow", line 8, in

sys.exit(main())

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/main.py", line 38, in main

args.func(args)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/cli/cli_parser.py", line 51, in command

return func(*args, **kwargs)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/utils/cli.py", line 99, in wrapper

return f(*args, **kwargs)

File "/app/x01/airflow/environments/airflow/lib/python3.8/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 "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/utils/session.py", line 77, in wrapper

return func(*args, session=session, **kwargs)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/utils/db.py", line 1446, in upgradedb

command.upgrade(config, revision=to_revision or 'heads')

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/alembic/command.py", line 320, in upgrade

script.run_env()

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/alembic/script/base.py", line 563, in run_env

util.load_python_file(self.dir, "env.py")

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/alembic/util/pyfiles.py", line 92, in load_python_file

module = load_module_py(module_id, path)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/alembic/util/pyfiles.py", line 108, in load_module_py

spec.loader.exec_module(module) # type: ignore

File "", line 783, in exec_module

File "", line 219, in _call_with_frames_removed

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/migrations/env.py", line 107, in

run_migrations_online()

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/migrations/env.py", line 101, in run_migrations_online

context.run_migrations()

File "", line 8, in run_migrations

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/alembic/runtime/environment.py", line 851, in run_migrations

self.get_context().run_migrations(**kw)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/alembic/runtime/migration.py", line 620, in run_migrations

step.migration_fn(**kw)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/migrations/versions/0073_2_0_0_prefix_dag_permissions.py", line 231, in upgrade

migrate_to_new_dag_permissions(db)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/migrations/versions/0073_2_0_0_prefix_dag_permissions.py", line 161, in migrate_to_new_dag_permissions

can_read_action = get_or_create_action(db.session, 'can_read')

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/migrations/versions/0073_2_0_0_prefix_dag_permissions.py", line 109, in get_or_create_action

session.commit()

File "", line 2, in commit

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 1423, in commit

self._transaction.commit(_to_root=self.future)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 829, in commit

self._prepare_impl()

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 808, in _prepare_impl

self.session.flush()

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3255, in flush

self._flush(objects)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3395, in flush

transaction.rollback(capture_exception=True)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 70, in exit

compat.raise(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise

raise exception

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3355, in _flush

flush_context.execute()

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 453, in execute

rec.execute(self)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 627, in execute

util.preloaded.orm_persistence.save_obj(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 242, in save_obj

_emit_insert_statements(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 1219, in _emit_insert_statements

result = connection._execute_20(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1520, in _execute_20

return meth(self, args_10style, kwargs_10style, execution_options)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/sql/elements.py", line 313, in _execute_on_connection

return connection._execute_clauseelement(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1389, in _execute_clauseelement

ret = self._execute_context(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1614, in _execute_context

self.handle_dbapi_exception(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1929, in handle_dbapi_exception

util.raise(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise

raise exception

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1608, in _execute_context

context = constructor(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 998, in _init_compiled

self._process_executesingle_defaults()

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1866, in _process_executesingle_defaults

val = self.get_insert_default(c)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1811, in get_insert_default

return self._exec_default(column, column.default, column.type)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1672, in exec_default

return self.fire_sequence(default, type)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/dialects/mysql/base.py", line 1401, in fire_sequence

return self._execute_scalar(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1269, in _execute_scalar

conn._cursor_execute(self.cursor, stmt, parameters, context=self)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1785, in _cursor_execute

self.handle_dbapi_exception(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1929, in handle_dbapi_exception

util.raise(

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise

raise exception

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1783, in _cursor_execute

self.dialect.do_execute(cursor, statement, parameters, context)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute

cursor.execute(statement, parameters)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/cursors.py", line 206, in execute

res = self._query(query)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/cursors.py", line 319, in _query

db.query(q)

File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/connections.py", line 254, in query

_mysql.connection.query(self, query)

sqlalchemy.exc.StatementError: (sqlalchemy.exc.ProgrammingError) (MySQLdb._exceptions.ProgrammingError) (1146, "Table 'airflow.ab_permission_id_seq' doesn't exist")

It seems that the upgrade looks for a table that is not present and not created by the upgrade itself. The same error is returned when the command "airflow db init" is executed on a clean environment.

What you think should happen instead

The db upgrade should create first the missing table and any other related (and missing) object.

How to reproduce

Just running the command using a MariaDB (or Mysql maybe) database.

Operating System

CentOS Linux release 7.9.2009

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==3.4.0

apache-airflow-providers-celery==2.1.4

apache-airflow-providers-elasticsearch==3.0.3

apache-airflow-providers-ftp==2.1.2

apache-airflow-providers-http==2.1.2

apache-airflow-providers-imap==2.2.3

apache-airflow-providers-jdbc==2.1.3

apache-airflow-providers-jenkins==2.1.0

apache-airflow-providers-jira==2.0.5

apache-airflow-providers-mysql==2.2.3

apache-airflow-providers-odbc==2.0.4

apache-airflow-providers-oracle==2.2.3

apache-airflow-providers-postgres==4.1.0

apache-airflow-providers-redis==2.0.4

apache-airflow-providers-samba==3.0.4

apache-airflow-providers-sftp==2.6.0

apache-airflow-providers-sqlite==2.1.3

apache-airflow-providers-ssh==2.4.4

Deployment

Virtualenv installation

Deployment details

Virtualenv based on Python 3.8.7

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