Migrate from 2.1.4 to 2.2.0

Apache Airflow version

2.2.0

Operating System

Linux

Versions of Apache Airflow Providers

default.

Deployment

Docker-Compose

Deployment details

Using airflow-2.2.0python3.7

What happened

Upgrading image from apache/airflow:2.1.4-python3.7
to apache/airflow:2.2.0-python3.7
Cause this inside scheduler, which is not starting:

Python version: 3.7.12
Airflow version: 2.2.0
Node: 6dd55b0a5dd7
-------------------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context
    cursor, statement, parameters, context
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute
    cursor.execute(statement, parameters)
psycopg2.errors.UndefinedColumn: column dag.max_active_tasks does not exist
LINE 1: ..., dag.schedule_interval AS dag_schedule_interval, dag.max_ac...
                                                             ^

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

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.7/site-packages/flask/app.py", line 2447, in wsgi_app
    response = self.full_dispatch_request()
  File "/home/airflow/.local/lib/python3.7/site-packages/flask/app.py", line 1952, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/home/airflow/.local/lib/python3.7/site-packages/flask/app.py", line 1821, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/home/airflow/.local/lib/python3.7/site-packages/flask/_compat.py", line 39, in reraise
    raise value
  File "/home/airflow/.local/lib/python3.7/site-packages/flask/app.py", line 1950, in full_dispatch_request
    rv = self.dispatch_request()
  File "/home/airflow/.local/lib/python3.7/site-packages/flask/app.py", line 1936, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/www/auth.py", line 51, in decorated
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/www/views.py", line 588, in index
    filter_dag_ids = current_app.appbuilder.sm.get_accessible_dag_ids(g.user)
  File "/home/airflow/.local/lib/python3.7/site-packages/airflow/www/security.py", line 377, in get_accessible_dag_ids
    return {dag.dag_id for dag in accessible_dags}
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/query.py", line 3535, in __iter__
    return self._execute_and_instances(context)
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/orm/query.py", line 3560, in _execute_and_instances
    result = conn.execute(querycontext.statement, self._params)
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1011, in execute
    return meth(self, multiparams, params)
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/sql/elements.py", line 298, in _execute_on_connection
    return connection._execute_clauseelement(self, multiparams, params)
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1130, in _execute_clauseelement
    distilled_params,
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1317, in _execute_context
    e, statement, parameters, cursor, context
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1511, in _handle_dbapi_exception
    sqlalchemy_exception, with_traceback=exc_info[2], from_=e
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
    raise exception
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context
    cursor, statement, parameters, context
  File "/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/engine/default.py", line 608, in do_execute
    cursor.execute(statement, parameters)
sqlalchemy.exc.ProgrammingError: (psycopg2.errors.UndefinedColumn) column dag.max_active_tasks does not exist
LINE 1: ..., dag.schedule_interval AS dag_schedule_interval, dag.max_ac...
                                                             ^

[SQL: SELECT dag.dag_id AS dag_dag_id, dag.root_dag_id AS dag_root_dag_id, dag.is_paused AS dag_is_paused, dag.is_subdag AS dag_is_subdag, dag.is_active AS dag_is_active, dag.last_parsed_time AS dag_last_parsed_time, dag.last_pickled AS dag_last_pickled, dag.last_expired AS dag_last_expired, dag.scheduler_lock AS dag_scheduler_lock, dag.pickle_id AS dag_pickle_id, dag.fileloc AS dag_fileloc, dag.owners AS dag_owners, dag.description AS dag_description, dag.default_view AS dag_default_view, dag.schedule_interval AS dag_schedule_interval, dag.max_active_tasks AS dag_max_active_tasks, dag.max_active_runs AS dag_max_active_runs, dag.has_task_concurrency_limits AS dag_has_task_concurrency_limits, dag.next_dagrun AS dag_next_dagrun, dag.next_dagrun_data_interval_start AS dag_next_dagrun_data_interval_start, dag.next_dagrun_data_interval_end AS dag_next_dagrun_data_interval_end, dag.next_dagrun_create_after AS dag_next_dagrun_create_after 
FROM dag]
(Background on this error at: http://sqlalche.me/e/13/f405)

What you expected to happen

Automatic database migration and properly working scheduler.

How to reproduce

Ugrade from 2.1.4 to 2.2.0 with some dags history.

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

Thanks for opening your first issue here! Be sure to follow the issue template!

Did you run airflow db upgrade?

Yes, I did. Got this:

airflow@6dd55b0a5dd7:/opt/airflow$ airflow db upgrade
/home/airflow/.local/lib/python3.7/site-packages/airflow/configuration.py:357
DeprecationWarning: The dag_concurrency option in [core] has been renamed to max_active_tasks_per_dag - the old setting has been used, but please update your config.
DB: postgresql+psycopg2://airflow:***@postgres/airflow
/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/dialects/postgresql/base.py:3573 
SAWarning: Predicate of partial index idx_dag_run_queued_dags ignored during reflection
/home/airflow/.local/lib/python3.7/site-packages/sqlalchemy/dialects/postgresql/base.py:3573 
SAWarning: Predicate of partial index idx_dag_run_running_dags ignored during reflection
[2021-10-11 19:21:13,069] {db.py:815} ERROR - Automatic migration is not available
[2021-10-11 19:21:13,069] {db.py:817} ERROR - The task_instance table has 53 rows without a corresponding dag_run row. You must manually correct this problem (possibly by deleting the problem rows).
[2021-10-11 19:21:13,071] {db.py:817} ERROR - The task_fail table has 24 rows without a corresponding dag_run row. You must manually correct this problem (possibly by deleting the problem rows).

I have just killed my docker-compose with 2.1.4, changed image version to 2.2.0 and airflow never started again :)
Downgrading to 2.1.4 fixes problem :)

Did you do the suggested fix ?

The task_instance table has 53 rows without a corresponding dag_run row. You must manually correct this problem (possibly by deleting the problem rows).
The task_fail table has 24 rows without a corresponding dag_run row. You must manually correct this problem (possibly by deleting the problem rows).` ?

Not yet, but I'm upgrading each new release from version 2.0.2 and this is first time, I'm getting migration problem. So I thought it's bigger problem. Especially I want to upgrade version in my production environment and seams it will be problematic.

Well, I think you should folllow what the suggestion is. You seem to have some wrong entries in your DB (which migh be result of bugs or some very old versions of airflow or both).

Closing as invalid unless following the migration suggestion does not fix the problem (likely you will have to do similar fixes in prooduction BTW).

After deleting wrong rows airflow is working.
Thank you for help.
Can be closed.

@sbialkowski-pixel

Hi @sbialkowski-pixel do you have the query to delete those rows in the postgres console ? I'm also facing this issue.

I've deleted this query, sorry.
But it was inspired how creator throw this error. So I found where in airflow code this error is produced:

airflow/airflow/utils/db.py

Lines 766 to 774 in bc19ae7

    
           join_cond = and_(table.c.dag_id == target.c.dag_id, table.c.execution_date == target.c.execution_date) 
        
           if "task_id" in target.columns: 
        
               join_cond = and_(join_cond, table.c.task_id == target.c.task_id) 
        
           query = ( 
        
               session.query(table.c.dag_id, table.c.task_id, table.c.execution_date) 
        
               .select_from(outerjoin(table, target, join_cond)) 
        
               .filter(target.c.dag_id.is_(None)) 
        
           )  # type: ignore

And I've recreate this in SQL.
If some references are lost, this kind of error should be repaired automatically IHMO.

If some references are lost, this kind of error should be repaired automatically IHMO.

Airflow does not delete user-generated data because we don't know if they are important to the user. It would be a disaster if we automatically delete the offending records if the user delibrately added them and did not get a chance to move them somewhere.

For the same reason, please inspect and fully understand any SQL queries suggested in this issue before running them, to make sure you are actually fine with those rows being deleted.

Problem is that there is no clue, which rows need to be deleted. Also there is no straight logic behind "corresponding rows" for me.

This part of code for me is the only source of some guidelines.

If there is better way to deal with this kind of problem in future, I will appreciate to hear it :)

@uranusjr

slack doesn't keep historical records, i'm transferring my comments here:

i picked one task instance from the list Migrate from 2.1.4 to 2.2.0 #18894 (comment)
and i can still find it on the UI so i didn't hit this 🐛 Deleting task instance from UI deletes Dag Run too #16896
on the other hand i couldn't find any DagRun with those execution_date values, so i must have hit the 🐛 @uranusjr has mentioned Deleting a DagRun from UI does not delete its associated tasks #15986. i rarely delete DagRuns, it explains why i only have just over a hundred dangling TIs

my airflow journey starts with 1.10.4

1.10.4 -> 1.10.5 -> 1.10.7 -> 1.10.9 -> 1.10.10 -> 1.10.11 -> 1.10.12 -> 1.10.13 -> 1.10.14 -> 2.0.0 -> 2.0.1 -> 2.0.2 -> 2.1.0 -> 2.1.1 -> 2.1.2 -> 2.1.3 -> 2.1.4 -> 2.2.0

…
LEFT JOIN
dag_run ON
dag_run.dag_id = task_fail.dag_id
AND dr.execution_date = task_fail.execution_date
…

This should read »dag_run.execution_date«

This becomes a bit neater with `... JOIN ... using (dag_id, execution_date)`

…

Am 13.10.2021 in 18:06, konfusator ***@***.***> schrieb: > > > … > LEFT JOIN > dag_run ON > dag_run.dag_id = task_fail.dag_id > AND dr.execution_date = task_fail.execution_date > … > > This should read »dag_run.execution_date« — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub, or unsubscribe. Triage notifications on the go with GitHub Mobile for iOS or Android.

@uranusjr

slack doesn't keep historical records, i'm transferring my comments here:

i picked one task instance from the list Migrate from 2.1.4 to 2.2.0 #18894 (comment)
and i can still find it on the UI so i didn't hit this 🐛 Deleting task instance from UI deletes Dag Run too #16896

on the other hand i couldn't find any DagRun with those execution_date values, so i must have hit the 🐛 @uranusjr has mentioned Deleting a DagRun from UI does not delete its associated tasks #15986. i rarely delete DagRuns, it explains why i only have just over a hundred dangling TIs

my airflow journey starts with 1.10.4

1.10.4 -> 1.10.5 -> 1.10.7 -> 1.10.9 -> 1.10.10 -> 1.10.11 -> 1.10.12 -> 1.10.13 -> 1.10.14 -> 2.0.0 -> 2.0.1 -> 2.0.2 -> 2.1.0 -> 2.1.1 -> 2.1.2 -> 2.1.3 -> 2.1.4 -> 2.2.0

#15986 might indeed have something to do with it.

I just deleted a few DAG runs on an Airflow instance which is still running 2.1.0 and can confirm that the task instances from those runs remain in the DB and show up in the query used to identify dangling rows: SELECT task_instance.execution_date, task_instance.task_id, task_instance.dag_id FROM task_instance LEFT JOIN dag_run ON task_instance.dag_id = dag_run.dag_id AND task_instance.execution_date = dag_run.execution_date WHERE dag_run.run_id IS NULL

All rows showing up in the query for this instance are related to DAGs where I have deleted DAG runs in the past, including scheduled and manual runs.

⬆️ We are seeing this while trying to upgrade a (new) Airflow 2.1.4 project.

This impacted me. The one thing that stood out was that most of the execution dates were from a period I definitely remember deleting. There was a DAG which was downloading data incrementally every 30 minutes (always download data > previous max ID) and I needed to clear a month of the data so it could start downloading again from like August onwards. I deleted these runs using the UI, so perhaps that is why?

@ldacey

@ldacey, yep, deleting dagruns via the UI turns out to be a common way to hit this (it's been a bug for a while it seems: #15986).

For anyone experiencing this when testing standlone and using SQLite, delete these ids:


select fail_id from (
(select id as fail_id,dag_id,execution_date from task_fail ) as fail
left join
(
select dag_id,execution_date from dag_run 
) as run
on run.dag_id=fail.dag_id and fail.execution_date=run.execution_date
) where run.dag_id is null

@leonsmith

SQL we used from our deployment playbook to clean up these tables

Obviously a disclaimer on running this blindly. Validate its rows you can safely remove first by just executing the CTEs

BEGIN;

-- Remove dag runs without a valid run_id
DELETE FROM dag_run WHERE run_id is NULL;

-- Remove task fails without a run_id
WITH task_fails_to_remove AS (
  SELECT 
    task_fail.dag_id,
    task_fail.task_id,
    task_fail.execution_date
  FROM
    task_fail
  LEFT JOIN 
    dag_run ON 
    dag_run.dag_id = task_fail.dag_id 
    AND dag_run.execution_date = task_fail.execution_date
  WHERE
    dag_run.run_id IS NULL
)
DELETE FROM
    task_fail
USING 
    task_fails_to_remove
WHERE (
    task_fail.dag_id = task_fails_to_remove.dag_id
    AND task_fail.task_id = task_fails_to_remove.task_id
    AND task_fail.execution_date = task_fails_to_remove.execution_date
);

-- Remove task instances without a run_id
WITH task_instances_to_remove AS (
  SELECT
    task_instance.dag_id,
    task_instance.task_id,
    task_instance.execution_date
  FROM
    task_instance
  LEFT JOIN 
    dag_run 
    ON dag_run.dag_id = task_instance.dag_id
    AND dag_run.execution_date = task_instance.execution_date
  WHERE 
    dag_run.run_id is NULL
)
DELETE FROM 
    task_instance
USING
    task_instances_to_remove
WHERE (
    task_instance.dag_id = task_instances_to_remove.dag_id
    AND task_instance.task_id = task_instances_to_remove.task_id
    AND task_instance.execution_date = task_instances_to_remove.execution_date
);

COMMIT;

Thank you for this @leonsmith ! Made our team's upgrade from v2.1.4 to v2.2.3 very clean and simple! 🚀 🚀 🚀 🚀

Navigation Menu

Repository navigation

Migrate from 2.1.4 to 2.2.0 #18894

Apache Airflow version

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

What happened

What you expected to happen

How to reproduce

Anything else

Are you willing to submit PR?

Code of Conduct

15 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	join_cond = and_(table.c.dag_id == target.c.dag_id, table.c.execution_date == target.c.execution_date)
	if "task_id" in target.columns:
	join_cond = and_(join_cond, table.c.task_id == target.c.task_id)

	query = (
	session.query(table.c.dag_id, table.c.task_id, table.c.execution_date)
	.select_from(outerjoin(table, target, join_cond))
	.filter(target.c.dag_id.is_(None))
	) # type: ignore

Migrate from 2.1.4 to 2.2.0 #18894

Description

Apache Airflow version

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

What happened

What you expected to happen

How to reproduce

Anything else

Are you willing to submit PR?

Code of Conduct

Activity

boring-cyborg commented on Oct 11, 2021

ephraimbuddy commented on Oct 11, 2021

sbialkowski-pixel commented on Oct 11, 2021

potiuk commented on Oct 11, 2021

sbialkowski-pixel commented on Oct 11, 2021

potiuk commented on Oct 11, 2021

sbialkowski-pixel commented on Oct 11, 2021

RenGeng commented on Oct 12, 2021

sbialkowski-pixel commented on Oct 12, 2021

uranusjr commented on Oct 12, 2021

sbialkowski-pixel commented on Oct 12, 2021

15 remaining items

zachliu commented on Oct 13, 2021

konfusator commented on Oct 13, 2021

MM-Lehmann commented on Oct 13, 2021

mhaalme commented on Oct 14, 2021

troyharvey commented on Oct 14, 2021

ldacey commented on Oct 15, 2021

jedcunningham commented on Oct 17, 2021

jasonoberme commented on Nov 1, 2021

aprettyloner commented on Jan 25, 2022

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions