sqlalchemy.exc.IntegrityError: (psycopg2.errors.ForeignKeyViolation) update or delete on table "dag_run" violates foreign key constraint "task_instance_dag_run_fkey" on table "task_instance" DETAIL: Key (dag_id, run_id)=(abo72_dedicated_brm_for_net_revenue, manual__2023-04-12T14:03:50.679746+00:00) is still referenced from table "task_instance".

[anka0618]

I am facing this issue while executing start task in one of my airflow dags. Can you please help here?

[anka0618]

I am using airflow 2.4.3 version

[potiuk]

Likely you are trying to run the same tasks again (for example by triggering it wiith the same run_id) - but it's hard to say withtout more details - without explaining the DAG code of yours and explaining what you are doing step-by-step.

You might want to review the code of yours and search for some strange unusual ways of triggering tasks.

Also it might be result of some bad db crash or programmatic error somewhere and maybe you will need to nuke your database or those individual records mentioned in the error.

[anka0618]

Hi Potiuk,
Please find the below details.

Dag code :

from datetime import datetime, timedelta
from pathlib import Path

from airflow import DAG
from airflow.models import Variable
#from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.dummy import DummyOperator

custom utils

from audit_plugin.safe_start_operator import StartOperator
from audit_plugin.safe_dependent_trigger_dag_run_operator import DependentTriggerDagRunOperator
from audit_plugin.ods_dag_build_operator import OdsDagBuildOperator
from utils.job_config import JobConfig

from utils.dagrun_utils import dagrun_utils as dagUtils
#from airflow.operators.python_operator import PythonOperator
from audit_plugin.end_dag_operator import EndDagOperator
from airflow.models import Variable
import logging
from airflow.utils.trigger_rule import TriggerRule

JOB_ARGS = JobConfig.get_config()
landing_project = (JOB_ARGS['landing_project'] + Variable.get('env')).replace('', '') audit_project = (JOB_ARGS['landing_project'] + Variable.get('env')).replace('', '')
dag_run_id = '{{ ts_nodash }}'
ods_name = 'cms_ods'

create DAG object by instantiating class

DAG = DAG(
Path(file).stem,
default_args=None,
start_date=datetime(2020, 1, 1),
schedule_interval=JOB_ARGS['schedule_interval'],
max_active_runs=2,
catchup=False)

def start_main_controller_dag(**context):
dagUtils.dag_run_id = context['ts_nodash']
print('pass')

start_dag = StartOperator(task_id='start_dags',
python_callable=start_main_controller_dag,
provide_context=True,
dag_run_id=dagUtils.batch_id+dag_run_id,
#batch_id=dagUtils.batch_id,
ods_name=ods_name,
dag=DAG)

build_dag = OdsDagBuildOperator(task_id='build_dags',
#provide_context=True,
audit_project=audit_project,
ods_name=ods_name,
dag=DAG)

main_controller = Variable.get(f"{ods_name}main_controller", default_var='{}')
print(main_controller)
logging.info(main_controller)
if main_controller == '{}':
Variable.set("main_controller{}".format(ods_name), '{}')
else:
main_controller = eval(main_controller)

if 'state' in main_controller:
if main_controller['state']:
taskinstances = main_controller['taskinstances']
task_dag_dict = {}

    trigger_dags = DummyOperator(task_id='trigger_dags',
                                 dag=DAG)
    task_dag_dict['trigger_dags'] = trigger_dags

    for index, taskinstance in enumerate(taskinstances):
        # Reading fields names from sql query
        parent_run_dag_id = taskinstance[2]
        child_run_dag_id = taskinstance[3]
        if taskinstance[4]:
            maxtime_minutes = taskinstance[4]
        else:
            maxtime_minutes = 60

        if not parent_run_dag_id in task_dag_dict:
            triggerParentTargetDag = DependentTriggerDagRunOperator(
                task_id=parent_run_dag_id,
                trigger_dag_id=parent_run_dag_id,
                external_dag_id=parent_run_dag_id,
                #python_callable=dagUtils.conditionally_trigger,
                params={'trigger_run_dag_id': Path(__file__).stem + "." + parent_run_dag_id,
                        'batch_id': dagUtils.batch_id},
                #timedelta=timedelta(minutes=1),
                execution_timeout=timedelta(minutes=maxtime_minutes),
                #dagrun_timeout=timedelta(minutes=maxtime_minutes),
                run_id=Path(__file__).stem + "." +
                parent_run_dag_id + "_" + dag_run_id,
                dag=DAG
            )

            task_dag_dict[parent_run_dag_id] = triggerParentTargetDag

        if child_run_dag_id != 'end_dags':
            if not child_run_dag_id in task_dag_dict:
                triggerChildTargetDag = DependentTriggerDagRunOperator(
                    task_id=child_run_dag_id,
                    trigger_dag_id=child_run_dag_id,
                    external_dag_id=child_run_dag_id,
                    #python_callable=dagUtils.conditionally_trigger,
                    params={'trigger_run_dag_id': Path(__file__).stem + "." + child_run_dag_id,
                            'batch_id': dagUtils.batch_id},
                    #timedelta=timedelta(minutes=1),
                    execution_timeout=timedelta(minutes=maxtime_minutes),
                    #dagrun_timeout=timedelta(minutes=maxtime_minutes),
                    run_id=Path(__file__).stem + "." +
                    child_run_dag_id + "_" + dag_run_id,
                    dag=DAG
                    # trigger_rule=TriggerRule.ONE_SUCCESS
                )
                task_dag_dict[child_run_dag_id] = triggerChildTargetDag

            task_dag_dict[parent_run_dag_id].set_downstream(
                task_dag_dict[child_run_dag_id])

        elif child_run_dag_id == 'end_dags':
            if not child_run_dag_id in task_dag_dict:
                end_dags = EndDagOperator(task_id='end_dags',
                                          trigger_rule="all_done",
                                          recipient_email_ids=JOB_ARGS['recipient_email_ids'],
                                          dag=DAG
                                          )
                task_dag_dict[child_run_dag_id] = end_dags
            task_dag_dict[parent_run_dag_id].set_downstream(end_dags)

    build_dag.set_downstream(task_dag_dict['trigger_dags'])

for initialising

start_dag.set_downstream(build_dag)

StartOperator is making use of dag_run_id and trying update the same in dag_run table in airflow db.

Below is the code for start operator.
##from airflow.utils.decorators import apply_defaults
import logging
#from airflow.exceptions import AirflowException
from airflow.models import DagBag, DagModel, DagRun, TaskInstance
#from airflow.sensors.base_sensor_operator import BaseSensorOperator
from airflow.utils.db import provide_session
#from airflow.operators.python_operator import PythonOperator
from airflow.operators.python import PythonOperator
from airflow.models import Variable

class StartOperator(PythonOperator):
"""
Usage Example:

StartOperator(task_id='start_dags'
                       ,python_callable=start_task
                       ,provide_context=True
                       ,dag_run_id=dag_run_id
                       ,dag=DAG)
:parameter
dag_run_id: if you do not pass this values custom run_id will not apply.

)
"""
template_fields = ['dag_run_id']

##@apply_defaults
def __init__(self, dag_run_id=None, ods_name=None,  *args, **kwargs):
    super(StartOperator, self).__init__(*args, **kwargs)
    self.dag_run_id = dag_run_id
    self.ods_name = ods_name

##@apply_defaults
def start_task(**context):
    print("startOperator context")
    print(context)
    if bool(context['dag_run'].conf):
        # if context['dag_run'].conf['batch_id']:
        print("batch_id finding")
        print(context['dag_run'].conf['batch_id'])
        context['params']['batch_id'] = context['dag_run'].conf['batch_id']

    print(context['params']['batch_id'])
    print(context)

@provide_session
def set_run_id_for_dag(self, context, session):
    logging.info("setting custom run_id for DAG set_run_id_for_dag")
    print(context)
    logging.info(context['dag'].dag_id)

    dag_run = (
        session.query(DagRun)
        .filter_by(
            dag_id=context['dag'].dag_id,
            execution_date=context['execution_date'])
        .first()
    )
    if self.dag_run_id:
        dag_run_id = self.dag_run_id
    else:
        dag_run_id = str(context['ts_nodash'])

    if dag_run:
        print(dag_run.run_id)
        if self.ods_name:
            self.ods_name = '{}_main_controller'.format(self.ods_name)
        else:
            self.ods_name = 'main_controller.'
        logging.info(self.ods_name)
        logging.info(dag_run.run_id)
        if not dag_run.run_id.__contains__(self.ods_name):
            dag_run.run_id = dag_run.run_id[:dag_run.run_id.find('__')+2] + dag_run_id
            logging.info("custom run id" + dag_run.run_id)
            logging.info(session.query)
            session.merge(dag_run)
            session.commit()
            #Variable.set('main_control_run_id', context['ts_nodash'])

@provide_session
def execute(self, context, session=None):
    print('StartOperator in safe_StartOperator')
    print(context)
    if self.dag_run_id:
        self.set_run_id_for_dag(context, session=session)
    PythonOperator.execute(self, context)

    # dag_run.refresh_from_db(self)

---

Please note that same piece of code is working fine in airflow 1.10.15 we are facing this issue while upgrading to 2.4.3.

Please let me know if you need any other details.

[anka0618]

Please let us know if we can meet on zoom call or google meet. So that i can explain all the details.

[potiuk]

You do realize that this is a free help that people here do in their free time?

However, if you want paid consultancy you can find help setup a call here with experts (for example) https://www.ringerhq.com/

[potiuk]

I suggest you start from scratch or try to fix your DB manually - there is something wrong with your DB. Unfortunately if you waited 3 years with migration and migrate from 1.10 that has reached End of Life more than 3 years ago you can expect similar issues.

I strongly suggest to start from a clean state. This is - unfortunately a price for being so late and not migrating before.

[anka0618]

Hi Potiuk

Thanks for the suggestions. I am cleaning up the DB and checking on the issue.