Rename execution_date to logical_date across codebase #43902

sunank200 · 2024-11-11T23:47:17Z

Motivation

This PR renames execution_date to logical_date across the codebase. The shift towards logical_date helps move away from the limitations of execution_date, particularly with dynamic DAG runs and cases where multiple runs occur at the same time.

Key Changes

Replaced all instances of execution_date with logical_date including models.
Updated references in database models, templates, and functions.
Removed unique constraints on execution_date in the database to allow multiple DAG runs with the same logical time.

How `execution_date` and `logical_date` Work

Logical date is equivalent to execution date: The two are just different names for the same value.
Timetable controls logical date: The logical date can be set to any value, not necessarily tied to the data interval's start or end.
Schedules dictate behavior: For value-based schedules (like cron), the logical date is set by the timetable class used.

Transitioning from `execution_date`

Airflow 3.0 will fully remove execution_date.
For uniquely identifying runs:
- Use run_id for human-readable identifiers.
- Use data_interval_start/data_interval_end for data partitioning in future.
- id (auto-incremented) is recommended for ordering DAG runs.
If your existing logic relies on execution_date, switch to using data_interval_start or data_interval_end for identifying time ranges.

Testing

Updated unit tests to reflect the changes.

^ 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.

uranusjr · 2024-11-12T04:39:29Z

This needs a news fragment. Rewording the above summary would work fine. See existing files in newsfragments (the significant ones) for some examples.

uranusjr · 2024-11-12T04:40:43Z

For templates, replace {{ ds }} with {{ data_interval_start | ds }}.

~~This should probably use logical_date for max compatibility. Also should mention ts (same rewrite).~~

Oh wait, we’re not removing those in this PR yet, so this should not be mentioned at all.

airflow/api_connexion/openapi/v1.yaml

airflow/api_connexion/schemas/dag_run_schema.py

uranusjr · 2024-11-12T05:06:34Z

airflow/cli/cli_config.py

@@ -410,7 +410,7 @@ def string_lower_type(val):
 # trigger_dag
 ARG_RUN_ID = Arg(("-r", "--run-id"), help="Helps to identify this run")
 ARG_CONF = Arg(("-c", "--conf"), help="JSON string that gets pickled into the DagRun's conf attribute")
-ARG_EXEC_DATE = Arg(("-e", "--exec-date"), help="The execution date of the DAG", type=parsedate)
+ARG_EXEC_DATE = Arg(("-e", "--exec-date"), help="The logical date of the DAG", type=parsedate)


We should also rename this flag in a future PR.

airflow/models/taskreschedule.py

airflow/models/xcom.py

airflow/utils/context.py

airflow/models/taskinstance.py

airflow/utils/context.pyi

docs/apache-airflow/templates-ref.rst

providers/src/airflow/providers/cncf/kubernetes/cli/kubernetes_command.py

providers/src/airflow/providers/google/cloud/operators/gcs.py

providers/src/airflow/providers/standard/sensors/date_time.py

providers/src/airflow/providers/openlineage/plugins/adapter.py

uranusjr · 2024-11-14T03:52:53Z

newsfragments/43902.significant.rst

This also needs to mention that some keys in the context have been removed.

uranusjr · 2024-11-14T03:54:07Z

airflow/utils/context.py

@@ -451,7 +426,6 @@ def context_copy_partial(source: Context, keys: Container[str]) -> Context:
    :meta private:
    """
    new = Context({k: v for k, v in source._context.items() if k in keys})
-    new._deprecation_replacements = source._deprecation_replacements.copy()


This line should not be removed.

uranusjr · 2024-11-14T03:55:12Z

providers/src/airflow/providers/elasticsearch/log/es_task_handler.py

+            logical_date = (
+                dag_run.logical_date.isoformat() if AIRFLOW_V_3_0_PLUS else dag_run.execution_date.isoformat()
+            )


Suggested change

logical_date = (

dag_run.logical_date.isoformat() if AIRFLOW_V_3_0_PLUS else dag_run.execution_date.isoformat()

)

logical_date = dag_run.logical_date.isoformat()

dag_run.logical_date is available on Airflow 2.

uranusjr · 2024-11-14T03:58:09Z

providers/src/airflow/providers/google/cloud/operators/gcs.py

+AIRFLOW_VERSION = Version(airflow_version)
+AIRFLOW_V_3_0_PLUS = Version(AIRFLOW_VERSION.base_version) >= Version("3.0.0")


Not needed anymore?

uranusjr · 2024-11-14T03:58:21Z

providers/src/airflow/providers/google/cloud/sensors/gcs.py

+AIRFLOW_VERSION = Version(airflow_version)
+AIRFLOW_V_3_0_PLUS = Version(AIRFLOW_VERSION.base_version) >= Version("3.0.0")


Not needed?

uranusjr · 2024-11-14T03:59:37Z

providers/src/airflow/providers/opensearch/log/os_task_handler.py

+            logical_date = (
+                dag_run.logical_date.isoformat() if AIRFLOW_V_3_0_PLUS else dag_run.execution_date.isoformat()
+            )


Suggested change

logical_date = (

dag_run.logical_date.isoformat() if AIRFLOW_V_3_0_PLUS else dag_run.execution_date.isoformat()

)

logical_date = dag_run.logical_date.isoformat()

uranusjr · 2024-11-14T04:00:49Z

providers/src/airflow/providers/opensearch/log/os_task_handler.py

+        return (
+            log_id_template.format(
+                dag_id=ti.dag_id,
+                task_id=ti.task_id,
+                run_id=getattr(ti, "run_id", ""),
+                data_interval_start=data_interval_start,
+                data_interval_end=data_interval_end,
+                logical_date=logical_date,
+                try_number=try_number,
+                map_index=getattr(ti, "map_index", ""),
+            )
+            if AIRFLOW_V_3_0_PLUS
+            else log_id_template.format(
+                dag_id=ti.dag_id,
+                task_id=ti.task_id,
+                run_id=getattr(ti, "run_id", ""),
+                data_interval_start=data_interval_start,
+                data_interval_end=data_interval_end,
+                execution_date=logical_date,
+                try_number=try_number,
+                map_index=getattr(ti, "map_index", ""),
+            )
        )


Suggested change

return (

log_id_template.format(

dag_id=ti.dag_id,

task_id=ti.task_id,

run_id=getattr(ti, "run_id", ""),

data_interval_start=data_interval_start,

data_interval_end=data_interval_end,

logical_date=logical_date,

try_number=try_number,

map_index=getattr(ti, "map_index", ""),

)

if AIRFLOW_V_3_0_PLUS

else log_id_template.format(

dag_id=ti.dag_id,

task_id=ti.task_id,

run_id=getattr(ti, "run_id", ""),

data_interval_start=data_interval_start,

data_interval_end=data_interval_end,

execution_date=logical_date,

try_number=try_number,

map_index=getattr(ti, "map_index", ""),

)

)

return log_id_template.format(

dag_id=ti.dag_id,

task_id=ti.task_id,

run_id=getattr(ti, "run_id", ""),

data_interval_start=data_interval_start,

data_interval_end=data_interval_end,

execution_date=logical_date, # Airlow 2 Compatibility.

logical_date=logical_date,

try_number=try_number,

map_index=ti.map_index,

)

uranusjr · 2024-11-14T04:02:33Z

providers/src/airflow/providers/presto/hooks/presto.py

    task_info = {
        "dag_id": context_var["dag_id"],
        "task_id": context_var["task_id"],
-        "execution_date": context_var["execution_date"],
+        date_key: context_var["logical_date"] if AIRFLOW_V_3_0_PLUS else context_var["execution_date"],


Suggested change

date_key: context_var["logical_date"] if AIRFLOW_V_3_0_PLUS else context_var["execution_date"],

date_key: context_var["logical_date"],

uranusjr · 2024-11-14T04:03:10Z

providers/src/airflow/providers/standard/operators/python.py

@@ -310,7 +309,7 @@ def get_tasks_to_skip():
            self.skip(
                dag_run=dag_run,
                tasks=to_skip,
-                execution_date=cast("DateTime", dag_run.execution_date),  # type: ignore[call-arg]
+                execution_date=cast("DateTime", dag_run.execution_date),  # type: ignore[call-arg, union-attr]


Suggested change

execution_date=cast("DateTime", dag_run.execution_date), # type: ignore[call-arg, union-attr]

execution_date=cast("DateTime", dag_run.logical_date), # type: ignore[call-arg, union-attr]

uranusjr · 2024-11-14T04:04:38Z

providers/src/airflow/providers/standard/operators/python.py

        "prev_data_interval_end_success",
        "prev_data_interval_start_success",
-        "prev_execution_date",
-        "prev_execution_date_success",
        "prev_start_date_success",
        "prev_end_date_success",
    }


These can not be deleted; the operator will not work correctly on Airflow 2. We can add comments these should be deleted when we drop Airflow 2 support on the provider.

uranusjr · 2024-11-14T04:05:08Z

providers/src/airflow/providers/trino/hooks/trino.py

    task_info = {
        "dag_id": context_var["dag_id"],
        "task_id": context_var["task_id"],
-        "execution_date": context_var["execution_date"],
+        date_key: context_var["logical_date"] if AIRFLOW_V_3_0_PLUS else context_var["execution_date"],


Suggested change

date_key: context_var["logical_date"] if AIRFLOW_V_3_0_PLUS else context_var["execution_date"],

date_key: context_var["logical_date"],

uranusjr · 2024-11-14T04:05:36Z

providers/tests/amazon/aws/links/test_base_aws.py

+AIRFLOW_VERSION = Version(airflow_version)
+AIRFLOW_V_3_0_PLUS = Version(AIRFLOW_VERSION.base_version) >= Version("3.0.0")


Not needed?

boring-cyborg bot added area:API Airflow's REST/HTTP API area:CLI area:Scheduler including HA (high availability) scheduler provider:cncf-kubernetes Kubernetes provider related issues labels Nov 11, 2024

sunank200 mentioned this pull request Nov 11, 2024

Remove execution_date and logical_date from DAG Run APIs and Functions, transition to run_id as sole identifier for Airflow 3.0 #42404

Open

4 tasks

sunank200 added legacy api Whether legacy API changes should be allowed in PR legacy ui Whether legacy UI change should be allowed in PR labels Nov 11, 2024

sunank200 force-pushed the rename-execution-date-to-logical-date branch 9 times, most recently from ea36b6c to 65f9b8e Compare November 12, 2024 02:54

sunank200 requested review from Lee-W and phanikumv November 12, 2024 02:54