Does it make sense for DebugExecutor to create backfills?

[dstandish]

I tried out DebugExecutor with the test dag pasted below.

It created 3 runs of the dag. And also curiously it uses the backfill system.

This seems like pretty undesirable behavior. Usually if you're trying to test a dag end-to-end, you will want to run just a single dag run. And dealing with task state and backfills... when you're debugging something you just want it to run so I'm not sure why we do this.

Is there a better cleaner way to use debug executor, or to run a dag in the way I'm looking for?

For example, something that behaves like this would make more sense to me:

def run_task(task, execution_date):
    ti = TaskInstance(task, execution_date)
    ti.render_templates()
    ti._run_raw_task(test_mode=True)


def run_tasks(*tasks):
    execution_date = pendulum.utcnow()
    for task in tasks:
        run_task(task, execution_date)

if __name__ == '__main__':
    run_tasks(push, pull)

But it would be better if I did not have to manually specify task sequence, and better still if everything that ordinarily would go to airflow logs would instead be printed to stdout.

What do you think? @turbaszek I think you are author of this executor... you have any thoughts or advice?

Thanks

Test dag:

from datetime import timedelta

import pendulum
from airflow import DAG
from airflow.models import BaseOperator
from airflow.utils.dates import days_ago

DEFAULT_ARGS = {
    "owner": "airflow",
    "depends_on_past": False,
    "email": ["airflow@example.com"],
    "email_on_failure": False,
    "email_on_retry": False,
}


class XcomPushOp(BaseOperator):
    def execute(self, context):
        return 'hello'


class XcomPullOp(BaseOperator):
    template_fields = ('key',)

    def __init__(self, key, **kwargs):
        super().__init__(**kwargs)
        self.key = key

    def execute(self, context):
        print(self.key)


with DAG(
    dag_id="standish_debug_executor_test",
    default_args=DEFAULT_ARGS,
    dagrun_timeout=timedelta(hours=2),
    start_date=days_ago(2),
    catchup=False,
    schedule_interval="0 3 * * *",
    tags=["example"],
) as dag:
    push = XcomPushOp(task_id='push')

    pull = XcomPullOp(
        task_id="pull",
        key="{{ task_instance.xcom_pull(task_ids='push', key='return_value') }}",
        do_xcom_push=True,
    )

    push >> pull

if __name__ == '__main__':
    dag.schedule_interval = None
    dag.clear()
    dag.run()

[dstandish]

I guess we can set schedule interval to None to make it run only one instance

if __name__ == '__main__':
    dag.schedule_interval = None
    dag.clear()
    dag.run()

But but calling self.log.info does not go to stdout...

[turbaszek]

But but calling self.log.info does not go to stdout...

Yes, this is an issue in 2.0 - apart from that the executor works and I'm using it pretty often. And as you discovered - it's easiest to use it with schedule_interval=None. If you have some ideas how we can improve that - they will be more than welcome 👍

[dstandish]

another related question .... why does it conver sensors to be reschedule mode? i tried DE with a sensor and it does not seem to respect the poke interval but seems to poke as fast as it can...

[turbaszek]

Because DebugExecutor is single threaded and having sensor in poke mode may block whole DAG. See:
#6740 (comment)

[dstandish]

you think there is a way to make it respect the poke interval?

[turbaszek]

Reschedule mode should respect poke mode. That's at least what I understand from the implementation.