Make prev_logical_date variable offset-aware

[akrava]

Closes: #29227

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[jacobhurlburt]

Testing locally prev_logical_date = DateTime.min is also an option. Might be preferable since it is consistent with the move to use pendulum going forward.

[dimberman]

Hi @akrava can you please add a failable test to this? IT would be great if we can have some way to ensure there isn't a regression in the future.

[josh-fell]

+1 on adding a test to prevent future regression that @dimberman mentioned.

I don't have a preference on using Datetime instead. I'll defer to @pierrejeambrun and @bbovenzi there.

[akrava]

Thanks @jacobhurlburt and @dimberman for suggestions. I've changed to DateTime.min from pendulum library - and it works fine. Also I've added test for this case - this error is reproducing only when DAG has non-cron schedule (like time interval). Here is code:

airflow/airflow/www/views.py

Lines 2853 to 2869 in 47b67f1

	if isinstance(dag.timetable, CronMixin):
	# Optimized calendar generation for timetables based on a cron expression.
	dates_iter: Iterator[datetime | None] = croniter(
	dag.timetable._expression,
	start_time=last_automated_data_interval.end,
	ret_type=datetime,
	)
	for dt in dates_iter:
	if dt is None:
	break
	if dt.year != year:
	break
	if dag.end_date and dt > dag.end_date:
	break
	dates[dt.date()] += 1
	else:
	prev_logical_date = datetime.min

From the code above we can see that non offset-aware datetime.min is running when isinstance(dag.timetable, CronMixin) is False. That's why I've added test case with latest_only DAG from example folder, because the schedule of this DAG is time interval, not cron:

airflow/airflow/example_dags/example_latest_only.py

Lines 27 to 29 in 47b67f1

	with DAG(
	dag_id="latest_only",
	schedule=dt.timedelta(hours=4),

[rdzsp]

Hello, I'm struggling with this issue. I'm installing Airflow by following this instruction https://airflow.apache.org/docs/apache-airflow/stable/installation/installing-from-pypi.html. But I see on this pull request seems fix the problem. Is your documentation not update yet?

[potiuk]

Hello, I'm struggling with this issue. I'm installing Airflow by following this instruction https://airflow.apache.org/docs/apache-airflow/stable/installation/installing-from-pypi.html. But I see on this pull request seems fix the problem. Is your documentation not update yet?

If you want an official release, you have to wait for the new release to be published - now it is being voted - but you can help with testing it by installing 2.5.2rc2 version - it would be great if you test it and post information about it in #30028 - like others did.