Changing task from unmapped to mapped task with task instance note and task reschedule

[tirkarthi]

Apache Airflow version

main (development)

What happened

Changing a non-mapped task with task instance note and task reschedule to a mapped task crashes scheduler when the task is cleared for rerun. Related commit where a similar fix was done.

commit a770edf
Author: Ephraim Anierobi splendidzigy24@gmail.com
Date: Mon Feb 20 20:45:25 2023 +0100

Fix Scheduler crash when clear a previous run of a normal task that is now a mapped task (#29645)

The fix was to clear the db references of the taskinstances in XCom, RenderedTaskInstanceFields
and TaskFail. That way, we are able to run the mapped tasks

What you think should happen instead

No response

How to reproduce

1. Create below dag file with BashOperator non-mapped.
2. Schedule a dag run and wait for it to finish.
3. Add a task instance note to bash operator.
4. Change t1 to t1 = BashOperator.partial(task_id="bash").expand(bash_command=command) and return ["echo hello"] from get_command.
5. Restart the scheduler and clear the task.
6. scheduler crashes on trying to use map_index though foreign key reference exists to task instance note and task reschedule.

import datetime

from airflow.operators.bash import BashOperator
from airflow import DAG
from airflow.decorators import task

with DAG(dag_id="bash_simple", start_date=datetime.datetime(2022, 1, 1), schedule=None, catchup=False) as dag:

    @task
    def get_command(arg1, arg2):
        return "echo hello"

    command = get_command(1, 1)
    t1 = BashOperator(task_id="bash", bash_command=command)


if __name__ == '__main__':
    dag.test()

Operating System

Ubuntu

Versions of Apache Airflow Providers

No response

Deployment

Virtualenv installation

Deployment details

No response

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