<p dir="auto"><strong>Apache Airflow version</strong>: 1.8.0</p>
<p dir="auto">Ticket was created 23/Nov/16 01:12</p>
<p dir="auto"><strong>Kubernetes version (if you are using kubernetes)</strong> (use <code class="notranslate">kubectl version</code>):</p>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>:</li>
<li><strong>OS</strong> (e.g. from /etc/os-release):</li>
<li><strong>Kernel</strong> (e.g. <code class="notranslate">uname -a</code>):</li>
<li><strong>Install tools</strong>:</li>
<li><strong>Others</strong>:<br>
<strong>What happened</strong>:</li>
</ul>
<p dir="auto">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: <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="172245968" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/1752" data-hovercard-type="pull_request" data-hovercard-url="/apache/airflow/pull/1752/hovercard" href="https://github.com/apache/airflow/pull/1752">#1752</a>.</p>
<p dir="auto">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.</p>
<p dir="auto">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.</p>
<p dir="auto">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.</p>
<h2 dir="auto">[2016-11-22 12:26:27,701] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:28:50,335] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:28:53,288] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:28:58,400] {models.py:1196} INFO -</h2>
<h2 dir="auto">Starting attempt 1 of 1</h2>
<p dir="auto">[2016-11-22 12:28:59,334] {models.py:1219} INFO - Executing &lt;Task(LatestOnlyOperator): latest_only&gt; on 2016-11-20 04:00:00<br>
[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<br>
[2016-11-22 12:29:00,671] {airflow_next.py:29} INFO - Not latest execution, skipping downstream.<br>
[2016-11-22 12:29:00,672] {airflow_next.py:34} INFO - Skipping task: my_dag<br>
[2016-11-22 12:29:01,397] {airflow_next.py:41} INFO - Done.<br>
[2016-11-22 12:31:13,055] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:31:17,899] {models.py:1150} INFO - Task &lt;TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]&gt; previously succeeded on 2016-11-22 12:29:01.397870<br>
[2016-11-22 12:32:31,907] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:34:56,522] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:35:00,975] {models.py:1150} INFO - Task &lt;TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]&gt; previously succeeded on 2016-11-22 12:29:01.397870<br>
[2016-11-22 12:35:36,323] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:38:00,140] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:38:05,057] {models.py:1150} INFO - Task &lt;TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]&gt; previously succeeded on 2016-11-22 12:29:01.397870<br>
[2016-11-22 12:38:50,014] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:41:07,609] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:41:12,232] {models.py:1150} INFO - Task &lt;TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]&gt; previously succeeded on 2016-11-22 12:29:01.397870<br>
[2016-11-22 12:41:45,857] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:44:05,354] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:44:09,635] {models.py:1150} INFO - Task &lt;TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]&gt; previously succeeded on 2016-11-22 12:29:01.397870<br>
[2016-11-22 12:44:30,851] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:46:58,977] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:47:02,836] {models.py:1150} INFO - Task &lt;TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]&gt; previously succeeded on 2016-11-22 12:29:01.397870<br>
[2016-11-22 12:48:27,571] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:50:54,034] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:50:57,951] {models.py:1150} INFO - Task &lt;TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]&gt; previously succeeded on 2016-11-22 12:29:01.397870<br>
[2016-11-22 12:51:21,442] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:53:44,461] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:53:48,392] {models.py:1150} INFO - Task &lt;TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]&gt; previously succeeded on 2016-11-22 12:29:01.397870<br>
[2016-11-22 12:54:28,745] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:56:50,740] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:56:54,382] {models.py:1150} INFO - Task &lt;TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]&gt; previously succeeded on 2016-11-22 12:29:01.397870<br>
[2016-11-22 12:57:59,881] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:59:04,245] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py<br>
[2016-11-22 12:59:05,666] {models.py:1150} INFO - Task &lt;TaskInstance: my_dag.latest_only 2016-11-20 04:00:00 [success]&gt; previously succeeded on 2016-11-22 12:29:01.397870<br>
[2016-11-22 13:02:18,434] {models.py:154} INFO - Filling up the DagBag from /DAGDIR/my_dag.py</p>
<p dir="auto">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):</p>
<p dir="auto">[<br>
u"['airflow run DAG_01 latest_only 2016-11-18T09:23:00 --local -sd DAGS_FOLDER/DAG_01.py ']",<br>
u"['airflow run DAG_02 latest_only 2016-11-18T03:45:00 --local -sd DAGS_FOLDER/DAG_02.py ']",<br>
u"['airflow run DAG_03 latest_only 2016-11-18T00:08:00 --local -sd DAGS_FOLDER/DAG_03.py ']",<br>
u"['airflow run DAG_04 latest_only 2016-11-22T10:40:00 --local -sd DAGS_FOLDER/DAG_04.py ']",<br>
u"['airflow run DAG_05 latest_only 2016-11-18T05:00:00 --local -sd DAGS_FOLDER/DAG_05.py ']",<br>
u"['airflow run DAG_06 latest_only 2016-11-22T11:20:00 --local -sd DAGS_FOLDER/DAG_06.py ']",<br>
u"['airflow run DAG_07 latest_only 2016-11-18T10:14:00 --local -sd DAGS_FOLDER/DAG_07.py ']",<br>
u"['airflow run DAG_01 latest_only 2016-11-18T09:23:00 --local -sd DAGS_FOLDER/DAG_01.py ']",<br>
u"['airflow run DAG_02 latest_only 2016-11-18T03:45:00 --local -sd DAGS_FOLDER/DAG_02.py ']",<br>
u"['airflow run DAG_03 latest_only 2016-11-18T00:08:00 --local -sd DAGS_FOLDER/DAG_03.py ']",<br>
u"['airflow run DAG_04 latest_only 2016-11-22T10:40:00 --local -sd DAGS_FOLDER/DAG_04.py ']",<br>
u"['airflow run DAG_05 latest_only 2016-11-18T05:00:00 --local -sd DAGS_FOLDER/DAG_05.py ']",<br>
u"['airflow run DAG_06 latest_only 2016-11-22T11:20:00 --local -sd DAGS_FOLDER/DAG_06.py ']",<br>
u"['airflow run DAG_07 latest_only 2016-11-18T10:14:00 --local -sd DAGS_FOLDER/DAG_07.py ']"<br>
]</p>
<p dir="auto">Grepping the scheduler's logs, here is one of the instances I see of the scheduler enqueuing a duplicate task:</p>
<h2 dir="auto">$ grep -A2 "DAG_01" /var/log/airflow/airflow-scheduler.log | grep -A2 "09:23"<br>
[2016-11-22 13:24:26,660] {models.py:2660} INFO - Checking state for &lt;DagRun DAG_01 @ 2016-11-18 09:23:00: scheduled__2016-11-18T09:23:00, externally triggered: False&gt;<br>
[2016-11-22 13:24:26,672] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.<br>
[2016-11-22 13:24:26,678] {jobs.py:514} INFO - Checking dependencies on 2 tasks instances, minus 0 skippable ones<br>
[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<br>
[2016-11-22 13:24:26,769] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.<br>
[2016-11-22 13:24:26,769] {jobs.py:514} INFO - Checking dependencies on 0 tasks instances, minus 0 skippable ones</h2>
<h2 dir="auto">[2016-11-22 13:24:31,830] {models.py:2660} INFO - Checking state for &lt;DagRun DAG_01 @ 2016-11-18 09:23:00: scheduled__2016-11-18T09:23:00, externally triggered: False&gt;<br>
[2016-11-22 13:24:31,832] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.<br>
[2016-11-22 13:24:31,832] {jobs.py:514} INFO - Checking dependencies on 0 tasks instances, minus 0 skippable ones</h2>
<h2 dir="auto">[2016-11-22 13:24:37,238] {models.py:2660} INFO - Checking state for &lt;DagRun DAG_01 @ 2016-11-18 09:23:00: scheduled__2016-11-18T09:23:00, externally triggered: False&gt;<br>
[2016-11-22 13:24:37,240] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.<br>
[2016-11-22 13:24:37,252] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.</h2>
<h2 dir="auto">[2016-11-22 13:24:45,736] {models.py:2660} INFO - Checking state for &lt;DagRun DAG_01 @ 2016-11-18 09:23:00: scheduled__2016-11-18T09:23:00, externally triggered: False&gt;<br>
[2016-11-22 13:24:45,744] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.<br>
[2016-11-22 13:24:45,756] {jobs.py:514} INFO - Checking dependencies on 2 tasks instances, minus 0 skippable ones</h2>
<h2 dir="auto">[2016-11-22 13:24:56,613] {models.py:2660} INFO - Checking state for &lt;DagRun DAG_01 @ 2016-11-18 09:23:00: scheduled__2016-11-18T09:23:00, externally triggered: False&gt;<br>
[2016-11-22 13:24:56,624] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.<br>
[2016-11-22 13:24:56,638] {jobs.py:514} INFO - Checking dependencies on 2 tasks instances, minus 0 skippable ones</h2>
<p dir="auto">[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<br>
[2016-11-22 13:24:56,823] {jobs.py:498} INFO - Getting list of tasks to skip for active runs.<br>
[2016-11-22 13:24:56,824] {jobs.py:514} INFO - Checking dependencies on 0 tasks instances, minus 0 skippable ones</p>
<p dir="auto">Eventually, we ended up just creating new DAG definitions with future start dates and manually cleared the Redis queue.</p>
<p dir="auto"><em>Additional Context</em>:</p>
<p dir="auto">Our scheduler is daemonized by upstart and runs with -n 5</p>
<p dir="auto">Here is the template we use for our cron DAGs (note it's a jinja2 template):</p>
<h1 dir="auto">{{ansible_managed}}</h1>
<p dir="auto">from dateutil import parser</p>
<p dir="auto">from airflow.operators import LatestOnlyOperator<br>
from airflow.operators import BashOperator<br>
from airflow.models import DAG</p>
<p dir="auto">args = {<br>
'owner': 'airflow',<br>
'start_date': parser.parse('{{item.start_date}}'),<br>
'retries': 0,<br>
}</p>
<p dir="auto">dag = DAG(<br>
dag_id='{{item.name}}',<br>
default_args=args,<br>
schedule_interval='{{item.schedule}}',<br>
max_active_runs=1,<br>
)</p>
<p dir="auto">latest_only = LatestOnlyOperator(task_id='latest_only', dag=dag)</p>
<p dir="auto">script = BashOperator(<br>
task_id='{{item.name}}',<br>
bash_command='{{item.command}}',<br>
default_args=args,<br>
dag=dag,<br>
)</p>
<p dir="auto">script.set_upstream(latest_only)</p>
<p dir="auto">One thing to note: We are on Airflow 2.7.1.3; however, we brought in the operator through a plugin:</p>
<p dir="auto">import datetime<br>
import logging</p>
<p dir="auto">from airflow.models import BaseOperator, TaskInstance<br>
from airflow.plugins_manager import AirflowPlugin<br>
from airflow.utils.state import State<br>
from airflow import settings</p>
<p dir="auto">class LatestOnlyOperator(BaseOperator):<br>
"""<br>
Allows a workflow to skip tasks that are not running during the most<br>
recent schedule interval.<br>
If the task is run outside of the latest schedule interval, all<br>
directly downstream tasks will be skipped.<br>
"""</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="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 &lt; now &lt;= 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.')"><pre class="notranslate"><code class="notranslate">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 &lt; now &lt;= 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.')
</code></pre></div>
<p dir="auto">class AirflowNextPlugin(AirflowPlugin):<br>
name = "airflow_next"<br>
operators = [LatestOnlyOperator]</p>
<p dir="auto"><strong>What you expected to happen</strong>:</p>
<p dir="auto"><strong>How to reproduce it</strong>:</p>
<p dir="auto"><strong>Anything else we need to know</strong>:</p>
<p dir="auto">Moved here from <a href="https://issues.apache.org/jira/browse/AIRFLOW-648" rel="nofollow">https://issues.apache.org/jira/browse/AIRFLOW-648</a></p>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.1</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">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:</p>
<p dir="auto">(airflow) [airflow@rabbit01 airflow]$ export LD_LIBRARY_PATH=/opt/python-bit/lib:$LD_LIBRARY_PATH<br>
(airflow) [airflow@rabbit01 airflow]$ airflow db upgrade<br>
DB: mysql+mysqldb://airflow:<em><strong>@localdb.localdomain.it:3306/airflow<br>
Performing upgrade with database mysql+mysqldb://airflow:</strong></em>@localdb.localdomain.it:3306/airflow<br>
[2022-06-06 10:18:48,523] {db.py:1445} INFO - Creating tables<br>
INFO  [alembic.runtime.migration] Context impl MySQLImpl.<br>
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.<br>
INFO  [alembic.runtime.migration] Running upgrade 45ba3f1493b9 -&gt; 849da589634d, Prefix DAG permissions.<br>
Traceback (most recent call last):<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1783, in _cursor_execute<br>
self.dialect.do_execute(cursor, statement, parameters, context)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute<br>
cursor.execute(statement, parameters)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/cursors.py", line 206, in execute<br>
res = self._query(query)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/cursors.py", line 319, in _query<br>
db.query(q)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/connections.py", line 254, in query<br>
_mysql.connection.query(self, query)<br>
<strong>MySQLdb._exceptions.ProgrammingError: (1146, "Table 'airflow.ab_permission_id_seq' doesn't exist")</strong></p>
<p dir="auto">The above exception was the direct cause of the following exception:</p>
<p dir="auto">Traceback (most recent call last):<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1608, in _execute_context<br>
context = constructor(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 998, in _init_compiled<br>
self._process_executesingle_defaults()<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1866, in _process_executesingle_defaults<br>
val = self.get_insert_default(c)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1811, in get_insert_default<br>
return self._exec_default(column, column.default, column.type)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1672, in <em>exec_default<br>
return self.fire_sequence(default, type</em>)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/dialects/mysql/base.py", line 1401, in fire_sequence<br>
return self._execute_scalar(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1269, in _execute_scalar<br>
conn._cursor_execute(self.cursor, stmt, parameters, context=self)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1785, in _cursor_execute<br>
self.<em>handle_dbapi_exception(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1929, in <em>handle_dbapi_exception<br>
util.raise</em>(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise</em><br>
raise exception<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1783, in _cursor_execute<br>
self.dialect.do_execute(cursor, statement, parameters, context)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute<br>
cursor.execute(statement, parameters)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/cursors.py", line 206, in execute<br>
res = self._query(query)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/cursors.py", line 319, in _query<br>
db.query(q)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/connections.py", line 254, in query<br>
_mysql.connection.query(self, query)<br>
<strong>sqlalchemy.exc.ProgrammingError: (MySQLdb._exceptions.ProgrammingError) (1146, "Table 'airflow.ab_permission_id_seq' doesn't exist")</strong><br>
[SQL: select nextval(ab_permission_id_seq)]<br>
(Background on this error at: <a href="http://sqlalche.me/e/14/f405" rel="nofollow">http://sqlalche.me/e/14/f405</a>)</p>
<p dir="auto">The above exception was the direct cause of the following exception:</p>
<p dir="auto">Traceback (most recent call last):<br>
File "/app/x01/airflow/environments/airflow/bin/airflow", line 8, in <br>
sys.exit(main())<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/<strong>main</strong>.py", line 38, in main<br>
args.func(args)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/cli/cli_parser.py", line 51, in command<br>
return func(*args, **kwargs)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/utils/cli.py", line 99, in wrapper<br>
return f(*args, **kwargs)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/cli/commands/db_command.py", line 82, in upgradedb<br>
db.upgradedb(to_revision=to_revision, from_revision=from_revision, show_sql_only=args.show_sql_only)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/utils/session.py", line 77, in wrapper<br>
return func(*args, session=session, **kwargs)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/utils/db.py", line 1446, in upgradedb<br>
command.upgrade(config, revision=to_revision or 'heads')<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/alembic/command.py", line 320, in upgrade<br>
script.run_env()<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/alembic/script/base.py", line 563, in run_env<br>
util.load_python_file(self.dir, "env.py")<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/alembic/util/pyfiles.py", line 92, in load_python_file<br>
module = load_module_py(module_id, path)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/alembic/util/pyfiles.py", line 108, in load_module_py<br>
spec.loader.exec_module(module)  # type: ignore<br>
File "", line 783, in exec_module<br>
File "", line 219, in _call_with_frames_removed<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/migrations/env.py", line 107, in <br>
run_migrations_online()<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/airflow/migrations/env.py", line 101, in run_migrations_online<br>
context.run_migrations()<br>
File "", line 8, in run_migrations<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/alembic/runtime/environment.py", line 851, in run_migrations<br>
self.get_context().run_migrations(**kw)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/alembic/runtime/migration.py", line 620, in run_migrations<br>
step.migration_fn(**kw)<br>
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<br>
migrate_to_new_dag_permissions(db)<br>
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<br>
can_read_action = get_or_create_action(db.session, 'can_read')<br>
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<br>
session.commit()<br>
File "", line 2, in commit<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 1423, in commit<br>
self._transaction.commit(_to_root=self.future)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 829, in commit<br>
self._prepare_impl()<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 808, in _prepare_impl<br>
self.session.flush()<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3255, in flush<br>
self._flush(objects)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3395, in <em>flush<br>
transaction.rollback(<em>capture_exception=True)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/util/langhelpers.py", line 70, in <strong>exit</strong><br>
compat.raise</em>(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise</em><br>
raise exception<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/session.py", line 3355, in _flush<br>
flush_context.execute()<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 453, in execute<br>
rec.execute(self)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/unitofwork.py", line 627, in execute<br>
util.preloaded.orm_persistence.save_obj(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 242, in save_obj<br>
_emit_insert_statements(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/orm/persistence.py", line 1219, in _emit_insert_statements<br>
result = connection._execute_20(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1520, in _execute_20<br>
return meth(self, args_10style, kwargs_10style, execution_options)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/sql/elements.py", line 313, in _execute_on_connection<br>
return connection._execute_clauseelement(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1389, in _execute_clauseelement<br>
ret = self._execute_context(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1614, in _execute_context<br>
self.<em>handle_dbapi_exception(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1929, in <em>handle_dbapi_exception<br>
util.raise</em>(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise</em><br>
raise exception<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1608, in _execute_context<br>
context = constructor(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 998, in _init_compiled<br>
self._process_executesingle_defaults()<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1866, in _process_executesingle_defaults<br>
val = self.get_insert_default(c)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1811, in get_insert_default<br>
return self._exec_default(column, column.default, column.type)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1672, in <em>exec_default<br>
return self.fire_sequence(default, type</em>)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/dialects/mysql/base.py", line 1401, in fire_sequence<br>
return self._execute_scalar(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 1269, in _execute_scalar<br>
conn._cursor_execute(self.cursor, stmt, parameters, context=self)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1785, in _cursor_execute<br>
self.<em>handle_dbapi_exception(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1929, in <em>handle_dbapi_exception<br>
util.raise</em>(<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 211, in raise</em><br>
raise exception<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1783, in _cursor_execute<br>
self.dialect.do_execute(cursor, statement, parameters, context)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 716, in do_execute<br>
cursor.execute(statement, parameters)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/cursors.py", line 206, in execute<br>
res = self._query(query)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/cursors.py", line 319, in _query<br>
db.query(q)<br>
File "/app/x01/airflow/environments/airflow/lib/python3.8/site-packages/MySQLdb/connections.py", line 254, in query<br>
_mysql.connection.query(self, query)<br>
<strong>sqlalchemy.exc.StatementError: (sqlalchemy.exc.ProgrammingError) (MySQLdb._exceptions.ProgrammingError) (1146, "Table 'airflow.ab_permission_id_seq' doesn't exist")</strong></p>
<p dir="auto">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.</p>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">The db upgrade should create first the missing table and any other related (and missing) object.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">Just running the command using a MariaDB (or Mysql maybe) database.</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">CentOS Linux release 7.9.2009</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-amazon==3.4.0<br>
apache-airflow-providers-celery==2.1.4<br>
apache-airflow-providers-elasticsearch==3.0.3<br>
apache-airflow-providers-ftp==2.1.2<br>
apache-airflow-providers-http==2.1.2<br>
apache-airflow-providers-imap==2.2.3<br>
apache-airflow-providers-jdbc==2.1.3<br>
apache-airflow-providers-jenkins==2.1.0<br>
apache-airflow-providers-jira==2.0.5<br>
apache-airflow-providers-mysql==2.2.3<br>
apache-airflow-providers-odbc==2.0.4<br>
apache-airflow-providers-oracle==2.2.3<br>
apache-airflow-providers-postgres==4.1.0<br>
apache-airflow-providers-redis==2.0.4<br>
apache-airflow-providers-samba==3.0.4<br>
apache-airflow-providers-sftp==2.6.0<br>
apache-airflow-providers-sqlite==2.1.3<br>
apache-airflow-providers-ssh==2.4.4</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Virtualenv installation</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">Virtualenv based on Python 3.8.7</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></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>