This is an automated email from the ASF dual-hosted git repository.
Lee-W pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 1ff21149246 Reject inverted date windows in airflow partitions clear
(#69454)
1ff21149246 is described below
commit 1ff21149246aced7c681e0f0daa7d9833278557a
Author: Wei Lee <[email protected]>
AuthorDate: Tue Jul 7 21:35:43 2026 +0800
Reject inverted date windows in airflow partitions clear (#69454)
---
.../src/airflow/cli/commands/partition_command.py | 29 ++++-
.../unit/cli/commands/test_partition_command.py | 136 +++++++++++++++++++++
2 files changed, 160 insertions(+), 5 deletions(-)
diff --git a/airflow-core/src/airflow/cli/commands/partition_command.py
b/airflow-core/src/airflow/cli/commands/partition_command.py
index c948e7f4b16..13443c42490 100644
--- a/airflow-core/src/airflow/cli/commands/partition_command.py
+++ b/airflow-core/src/airflow/cli/commands/partition_command.py
@@ -44,7 +44,9 @@ def clear(args, *, session: Session = NEW_SESSION) -> None:
the matching partitions; a date-only value (no time) is treated as local
midnight.
"""
- has_range = args.start_date is not None or args.end_date is not None or
args.date is not None
+ has_start = args.start_date is not None
+ has_end = args.end_date is not None
+ has_range = has_start or has_end or args.date is not None
selectors_used = sum([args.run_id is not None, args.partition_key is not
None, has_range])
if selectors_used != 1:
raise SystemExit(
@@ -52,8 +54,10 @@ def clear(args, *, session: Session = NEW_SESSION) -> None:
"(--start-date/--end-date or --date)."
)
- if args.date is not None:
- if args.start_date is not None or args.end_date is not None:
+ from_date_flag = args.date is not None
+
+ if from_date_flag:
+ if has_start or has_end:
raise SystemExit("--date cannot be combined with --start-date /
--end-date.")
raw = args.date
parts = raw.split("~", 1)
@@ -64,9 +68,24 @@ def clear(args, *, session: Session = NEW_SESSION) -> None:
args.end_date = parsedate(parts[1].strip())
except ValueError:
raise SystemExit("--date sides must be parseable as a date or
datetime.")
+ has_start = has_end = True
+
+ has_date_window = has_start or has_end
+ if has_date_window:
+ dag = get_db_dag(bundle_names=None, dag_id=args.dag_id)
+ if has_start and has_end:
+ lower = dag.timetable.localize_partition_datetime(args.start_date)
+ upper = dag.timetable.localize_partition_datetime(args.end_date)
+ if lower > upper:
+ if from_date_flag:
+ raise SystemExit(
+ f"--date: the start of the range
({parts[0].strip()!r}) must be on or before "
+ f"the end ({parts[1].strip()!r})."
+ )
+ raise SystemExit("--start-date must be on or before
--end-date.")
+ else:
+ dag = None
- has_date_window = args.start_date is not None or args.end_date is not None
- dag = get_db_dag(bundle_names=None, dag_id=args.dag_id) if has_date_window
else None
clear_tis = bool(args.clear_task_instances)
processed_any = False
diff --git a/airflow-core/tests/unit/cli/commands/test_partition_command.py
b/airflow-core/tests/unit/cli/commands/test_partition_command.py
index efcbc9afe5f..7892c5bd74d 100644
--- a/airflow-core/tests/unit/cli/commands/test_partition_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_partition_command.py
@@ -760,6 +760,142 @@ class TestPartitionsClear:
)
assert "--date must be in the form 'a~b'" in str(excinfo.value.code)
+ @pytest.mark.parametrize(
+ ("cli_args", "expected_message"),
+ [
+ (
+ ["--start-date", "2026-01-03", "--end-date", "2026-01-01"],
+ "--start-date must be on or before --end-date.",
+ ),
+ (
+ ["--date", "2026-01-03~2026-01-01"],
+ "--date: the start of the range ('2026-01-03') must be on or
before the end ('2026-01-01').",
+ ),
+ ],
+ )
+ def test_inverted_date_window_is_rejected(self, parser, cli_args,
expected_message):
+ with pytest.raises(SystemExit) as excinfo:
+ partition_command.clear(parser.parse_args(["partitions", "clear",
"--dag-id", DAG_ID, *cli_args]))
+ assert excinfo.value.code == expected_message
+
+ def test_equal_start_and_end_date_is_accepted(self, parser, capsys):
+ partition_command.clear(
+ parser.parse_args(
+ [
+ "partitions",
+ "clear",
+ "--dag-id",
+ DAG_ID,
+ "--start-date",
+ "2026-01-02",
+ "--end-date",
+ "2026-01-02",
+ ]
+ )
+ )
+ captured = capsys.readouterr()
+ assert captured.out == (
+ "DagRun part_run_2: partition_key='2026-01-02T00:00:00' -> None, "
+ "partition_date=2026-01-02T00:00:00+00:00 -> None\n"
+ "Cleared partition fields on 1 DagRun(s).\n"
+ )
+
+ def test_offset_aware_window_false_rejection_is_accepted(self, parser,
dag_maker):
+ """Absolute-instant order disagrees with Asia/Taipei-localized
wall-clock order.
+
+ --start-date 2026-01-01T00:00:00+00:00 (instant Jan1 00:00Z) is
*after* the
+ instant of --end-date 2026-01-01T01:00:00+10:00 (Dec31 15:00Z), so a
raw
+ instant comparison would wrongly flag this as an inverted window. Once
each
+ bound's wall-clock reading is localized to the Dag's Asia/Taipei
timetable
+ timezone (matching downstream ``apply_partition_date_window``), the
window
+ becomes [Dec31 16:00Z, Dec31 17:00Z) -- lower before upper, valid --
and the
+ run seeded inside it must be cleared.
+ """
+ dag_id = "dag_taipei_offset_false_rejection"
+ clear_db_runs()
+ clear_db_dags()
+ with dag_maker(
+ dag_id,
+ schedule=CronPartitionTimetable("0 0 * * *",
timezone="Asia/Taipei"),
+ start_date=datetime(2025, 1, 1, tzinfo=pendulum.UTC),
+ catchup=True,
+ serialized=True,
+ ):
+ EmptyOperator(task_id="t1")
+ dag_maker.create_dagrun(
+ run_id="in_window_run",
+ state=DagRunState.SUCCESS,
+ logical_date=None,
+ partition_date=datetime(2025, 12, 31, 16, 30, 0,
tzinfo=pendulum.UTC),
+ partition_key="in-window",
+ )
+ dag_maker.sync_dagbag_to_db()
+
+ partition_command.clear(
+ parser.parse_args(
+ [
+ "partitions",
+ "clear",
+ "--dag-id",
+ dag_id,
+ "--start-date",
+ "2026-01-01T00:00:00+00:00",
+ "--end-date",
+ "2026-01-01T01:00:00+10:00",
+ ]
+ )
+ )
+
+ run = _get_run("in_window_run")
+ assert run.partition_key is None
+ assert run.partition_date is None
+
+ clear_db_runs()
+ clear_db_dags()
+
+ def test_offset_aware_window_false_acceptance_is_rejected(self, parser,
dag_maker):
+ """Absolute-instant order disagrees with Asia/Taipei-localized
wall-clock order.
+
+ --start-date 2026-01-01T05:00:00+00:00 (instant Jan1 05:00Z) is
*before* the
+ instant of --end-date 2026-01-01T04:00:00-10:00 (Jan1 14:00Z), so a raw
+ instant comparison would wrongly accept this window. Once each bound's
+ wall-clock reading is localized to the Dag's Asia/Taipei timetable
timezone,
+ the window is [Dec31 21:00Z, Dec31 20:00Z) -- lower after upper,
inverted --
+ and clear() must reject it with the same message as the UTC-only case.
+ """
+ dag_id = "dag_taipei_offset_false_acceptance"
+ clear_db_runs()
+ clear_db_dags()
+ with dag_maker(
+ dag_id,
+ schedule=CronPartitionTimetable("0 0 * * *",
timezone="Asia/Taipei"),
+ start_date=datetime(2025, 1, 1, tzinfo=pendulum.UTC),
+ catchup=True,
+ serialized=True,
+ ):
+ EmptyOperator(task_id="t1")
+ dag_maker.sync_dagbag_to_db()
+
+ with pytest.raises(SystemExit) as excinfo:
+ partition_command.clear(
+ parser.parse_args(
+ [
+ "partitions",
+ "clear",
+ "--dag-id",
+ dag_id,
+ "--start-date",
+ "2026-01-01T05:00:00+00:00",
+ "--end-date",
+ "2026-01-01T04:00:00-10:00",
+ ]
+ )
+ )
+ assert excinfo.value.code == "--start-date must be on or before
--end-date."
+
+ clear_db_runs()
+ clear_db_dags()
+
def test_date_range_syntax_mutually_exclusive_with_start_end(self, parser):
with pytest.raises(SystemExit) as excinfo:
partition_command.clear(
