Задача берёт дату из datetime.now(), а не из логического интервала

def load():
    day = datetime.now().date()
    run_sql(f"""
        delete from dm.sales where order_date = '{day}';
        insert into dm.sales
        select order_date, branch_id, sum(amount)
        from dds.orders where order_date = '{day}'
        group by order_date, branch_id
    """)

• Загрузятся данные за 3 августа: Airflow при перезапуске подставляет в задачу логическую дату рана.
• Задача упадёт: данных за 5 августа в dds.orders ещё нет, потому что суточный ран за 5-е не отрабатывал.
• Задача отработает и заполнит все пропущенные даты с 3 по 5 августа, так как delete затронет только устаревшие строки.
• Задача перезапишет данные за 5 августа, а дырка за 3 августа останется — при этом ран в интерфейсе станет зелёным.