Avoid materializing DagRun conf before task execution

airflow-core/newsfragments/+avoid-dagrun-conf-materialization.bugfix.rst

@@ -0,0 +1 @@
+Avoid materializing DagRun configuration in the Execution API server and task supervisor.

airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py

@@ -414,6 +414,9 @@ class TIRunContext(BaseModel):
     dag_run: DagRun
     """DAG run information for the task instance."""
 
+    dag_run_conf_json: str | None = None
+    """Compact DagRun configuration JSON transported only during task startup."""
+
     task_reschedule_count: int = 0
     """How many times the task has been rescheduled."""
 

airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py

@@ -35,10 +35,11 @@
 from opentelemetry.trace import StatusCode
 from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 from pydantic import JsonValue, ValidationError
-from sqlalchemy import and_, func, or_, tuple_, update
+from sqlalchemy import Text, and_, cast as sa_cast, func, or_, tuple_, update
 from sqlalchemy.engine import CursorResult
 from sqlalchemy.exc import DataError, NoResultFound, SQLAlchemyError
-from sqlalchemy.orm import contains_eager, joinedload
+from sqlalchemy.orm import contains_eager, defer, joinedload, selectinload
+from sqlalchemy.orm.attributes import set_committed_value
 from sqlalchemy.sql import select
 from sqlalchemy.sql.dml import Update
 from structlog.contextvars import bind_contextvars
@@ -279,17 +280,21 @@ def ti_run(
         result = session.execute(query)
         log.info("Task instance state updated", rows_affected=getattr(result, "rowcount", 0))
 
-        dr = (
-            session.scalars(
-                select(DR)
+        dr_row = (
+            session.execute(
+                select(DR, sa_cast(DR.conf, Text).label("serialized_conf"))
                 .filter_by(dag_id=ti.dag_id, run_id=ti.run_id)
-                .options(joinedload(DR.consumed_asset_events), *eager_load_teams(DR.dag_model))
+                .options(
+                    defer(DR.conf),
+                    selectinload(DR.consumed_asset_events),
+                    *eager_load_teams(DR.dag_model),
+                )
             )
             .unique()
             .one_or_none()
         )
 
-        if not dr:
+        if not dr_row:
             log.error("DagRun not found", dag_id=ti.dag_id, run_id=ti.run_id)
             raise HTTPException(
                 status_code=status.HTTP_404_NOT_FOUND,
@@ -298,6 +303,8 @@ def ti_run(
                     "message": f"DagRun with dag_id={ti.dag_id} and run_id={ti.run_id} not found",
                 },
             )
+        dr, serialized_conf = dr_row
+        set_committed_value(dr, "conf", None)
 
         # Send the keys to the SDK so that the client requests to clear those XComs from the server.
         # The reason we cannot do this here in the server is because we need to issue a purge on custom XCom backends
@@ -319,6 +326,7 @@ def ti_run(
 
         context = TIRunContext(
             dag_run=dr,
+            dag_run_conf_json=None if serialized_conf == "null" else serialized_conf,
             task_reschedule_count=task_reschedule_count,
             max_tries=ti.max_tries,
             # TODO: Add variables and connections that are needed (and has perms) for the task

airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py

@@ -55,6 +55,7 @@
 from airflow.api_fastapi.execution_api.versions.v2026_10_30 import (
     AddArgBindingsToTIRunContext,
     AddCallbackRunEndpoint,
+    AddDagRunConfJsonToTIRunContext,
     AddDagRunNoteUpdateEndpoint,
     AddMultiTeamToTIRunContext,
     AddStoppedTaskReport,
@@ -67,6 +68,7 @@
     Version(
         "2026-10-30",
         AddArgBindingsToTIRunContext,
+        AddDagRunConfJsonToTIRunContext,
         AddCallbackRunEndpoint,
         AddDagRunNoteUpdateEndpoint,
         AddTerminalStateRetryReasonField,

airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_10_30.py

@@ -17,6 +17,8 @@
 
 from __future__ import annotations
 
+import json
+
 from cadwyn import (
     ResponseInfo,
     VersionChange,
@@ -34,6 +36,24 @@
 )
 
 
+class AddDagRunConfJsonToTIRunContext(VersionChange):
+    """Add compact DagRun configuration JSON to the task startup context."""
+
+    description = __doc__
+    instructions_to_migrate_to_previous_version = (
+        schema(TIRunContext).field("dag_run_conf_json").didnt_exist,
+    )
+
+    @convert_response_to_previous_version_for(TIRunContext)  # type: ignore[arg-type]
+    def deserialize_conf_for_previous_versions(response: ResponseInfo) -> None:  # type: ignore[misc]
+        """Preserve the dictionary contract for older task SDK clients."""
+        if "dag_run" not in response.body or not isinstance(response.body["dag_run"], dict):
+            return
+        dag_run_conf_json = response.body.pop("dag_run_conf_json", None)
+        if isinstance(dag_run_conf_json, str):
+            response.body["dag_run"]["conf"] = json.loads(dag_run_conf_json)
+
+
 class AddStoppedTaskReport(VersionChange):
     """Allow supervisors to report server-requested termination after the child exits."""
 

airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py

@@ -17,6 +17,7 @@
 
 from __future__ import annotations
 
+import json
 import re
 from contextlib import nullcontext
 from datetime import datetime
@@ -4980,6 +4981,35 @@ def test_ti_run_with_null_conf(self, client, session, create_task_instance):
         context = response.json()
         assert context["dag_run"]["conf"] is None
 
+    def test_ti_run_transports_conf_as_opaque_json(self, client, session, create_task_instance):
+        """Task startup does not materialize DagRun.conf in the API server or supervisor."""
+        ti = create_task_instance(
+            task_id="test_ti_run_with_opaque_conf",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+        )
+        ti.dag_run.conf = {"records": [{"id": 1, "value": "one"}, {"id": 2, "value": "two"}]}
+        session.commit()
+
+        response = client.patch(
+            f"/execution/task-instances/{ti.id}/run",
+            json={
+                "state": "running",
+                "pid": 100,
+                "hostname": "test-hostname",
+                "unixname": "test-user",
+                "start_date": timezone.utcnow().isoformat(),
+            },
+        )
+
+        assert response.status_code == 200, f"Response: {response.text}"
+        context = response.json()
+        assert context["dag_run"]["conf"] is None
+        assert json.loads(context["dag_run_conf_json"]) == {
+            "records": [{"id": 1, "value": "one"}, {"id": 2, "value": "two"}]
+        }
+
 
 class TestTIPatchRenderedMapIndex:
     def setup_method(self):

devel-common/src/tests_common/pytest_plugin.py

@@ -2746,8 +2746,10 @@ def set_dag(what: StartupDetails, dag_id: str, task: BaseOperator) -> RuntimeTas
             task = dag.task_dict[task.task_id]  # type: ignore[assignment]
         else:
             dag = task.dag
-        if what.ti_context.dag_run.conf:
-            dag.params = what.ti_context.dag_run.conf  # type: ignore[assignment]
+        if dag_run_conf := what.ti_context.dag_run_conf_json:
+            dag.params = json.loads(dag_run_conf)
+        elif what.ti_context.dag_run.conf:
+            dag.params = what.ti_context.dag_run.conf
         ti = RuntimeTaskInstance.model_construct(
             **what.ti.model_dump(exclude_unset=True),
             task=task,
@@ -2873,6 +2875,7 @@ def _create_task_instance(
         task_reschedule_count: int = 0,
         ti_id: UUID | None = None,
         conf: dict[str, Any] | None = None,
+        dag_run_conf_json: str | None = None,
         should_retry: bool | None = None,
         max_tries: int | None = None,
     ) -> RuntimeTaskInstance:
@@ -2950,6 +2953,7 @@ def _create_task_instance(
                     **({"state": DagRunState.RUNNING} if "state" in DagRun.model_fields else {}),
                 }
             ),
+            dag_run_conf_json=dag_run_conf_json,
             task_reschedule_count=task_reschedule_count,
             max_tries=task_retries if max_tries is None else max_tries,
             should_retry=should_retry if should_retry is not None else try_number <= task_retries,

task-sdk/src/airflow/sdk/api/datamodels/_generated.py

@@ -838,6 +838,7 @@ class TIRunContext(BaseModel):
     """
 
     dag_run: DagRun
+    dag_run_conf_json: Annotated[str | None, Field(title="Dag Run Conf Json")] = None
     task_reschedule_count: Annotated[int | None, Field(title="Task Reschedule Count")] = 0
     max_tries: Annotated[int, Field(title="Max Tries")]
     variables: Annotated[list[VariableResponse] | None, Field(title="Variables")] = None

task-sdk/src/airflow/sdk/execution_time/schema/schema.json

@@ -5108,6 +5108,18 @@
         "dag_run": {
           "$ref": "#/$defs/DagRun"
         },
+        "dag_run_conf_json": {
+          "anyOf": [
+            {
+              "type": "string"
+            },
+            {
+              "type": "null"
+            }
+          ],
+          "default": null,
+          "title": "Dag Run Conf Json"
+        },
         "task_reschedule_count": {
           "default": 0,
           "title": "Task Reschedule Count",

task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py

@@ -40,6 +40,7 @@ def get_bundle() -> VersionBundle:
     from airflow.sdk.execution_time.schema.versions.v2026_10_30 import (
         AddArgBindingsToSupervisorTIRunContext,
         AddDagDefinitionsToDagFileParsingResult,
+        AddDagRunConfJsonToSupervisorTIRunContext,
         AddRetryReasonToTaskState,
     )
 
@@ -50,6 +51,7 @@ def get_bundle() -> VersionBundle:
             AddArgBindingsToSupervisorTIRunContext,
             AddRetryReasonToTaskState,
             AddDagDefinitionsToDagFileParsingResult,
+            AddDagRunConfJsonToSupervisorTIRunContext,
         ),
         Version("2026-06-16"),
     )

task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py

@@ -26,11 +26,13 @@
 
 from __future__ import annotations
 
-from cadwyn import VersionChange, schema
+import json
+
+from cadwyn import ResponseInfo, VersionChange, convert_response_to_previous_version_for, schema
 
 from airflow.dag_processing.processor import DagFileParsingResult  # noqa: SDK002
 from airflow.sdk.api.datamodels._generated import TIRunContext
-from airflow.sdk.execution_time.comms import TaskState
+from airflow.sdk.execution_time.comms import StartupDetails, TaskState
 
 
 class AddArgBindingsToSupervisorTIRunContext(VersionChange):
@@ -47,6 +49,28 @@ class AddArgBindingsToSupervisorTIRunContext(VersionChange):
     instructions_to_migrate_to_previous_version = (schema(TIRunContext).field("arg_bindings").didnt_exist,)
 
 
+class AddDagRunConfJsonToSupervisorTIRunContext(VersionChange):
+    """Add compact DagRun configuration JSON to the supervisor task context."""
+
+    description = __doc__
+    instructions_to_migrate_to_previous_version = (
+        schema(TIRunContext).field("dag_run_conf_json").didnt_exist,
+    )
+
+    @convert_response_to_previous_version_for(StartupDetails)  # type: ignore[arg-type]
+    def deserialize_conf_for_previous_versions(response: ResponseInfo) -> None:  # type: ignore[misc]
+        """Preserve the dictionary contract for older language SDK clients."""
+        ti_context = response.body.get("ti_context")
+        if not isinstance(ti_context, dict):
+            return
+        dag_run = ti_context.get("dag_run")
+        if not isinstance(dag_run, dict):
+            return
+        dag_run_conf_json = ti_context.pop("dag_run_conf_json", None)
+        if isinstance(dag_run_conf_json, str):
+            dag_run["conf"] = json.loads(dag_run_conf_json)
+
+
 class AddRetryReasonToTaskState(VersionChange):
     """Add `retry_reason` to `TaskState`."""
 

task-sdk/src/airflow/sdk/execution_time/task_runner.py

@@ -22,6 +22,7 @@
 import contextvars
 import functools
 import inspect
+import json
 import os
 import sys
 import time
@@ -289,7 +290,12 @@ def get_template_context(self) -> Context:
         dag_run_conf: dict[str, Any] | None = None
         macros_accessor = MacrosAccessor()
         if from_server := self._ti_context_from_server:
-            dag_run_conf = from_server.dag_run.conf or dag_run_conf
+            if from_server.dag_run.conf is not None:
+                dag_run_conf = from_server.dag_run.conf
+            elif from_server.dag_run_conf_json is not None:
+                dag_run_conf = json.loads(from_server.dag_run_conf_json)
+                from_server.dag_run.conf = dag_run_conf
+                from_server.dag_run_conf_json = None
             macros_accessor = MacrosAccessor(
                 team_name=from_server.dag_run.team_name, multi_team=bool(from_server.multi_team)
             )

task-sdk/tests/conftest.py

@@ -240,6 +240,7 @@ def __call__(
         run_type: str = ...,
         task_reschedule_count: int = ...,
         conf: dict[str, Any] | None = ...,
+        dag_run_conf_json: str | None = ...,
         should_retry: bool = ...,
         max_tries: int = ...,
         consumed_asset_events: Sequence[AssetEventDagRunReference] = ...,
@@ -282,6 +283,7 @@ def _make_context(
         run_type: str = "manual",
         task_reschedule_count: int = 0,
         conf: dict[str, Any] | None = None,
+        dag_run_conf_json: str | None = None,
         should_retry: bool = False,
         max_tries: int = 0,
         consumed_asset_events: Sequence[AssetEventDagRunReference] = (),
@@ -303,6 +305,7 @@ def _make_context(
                 end_date=None,
                 partition_key=None,
             ),
+            dag_run_conf_json=dag_run_conf_json,
             task_reschedule_count=task_reschedule_count,
             max_tries=max_tries,
             should_retry=should_retry,

task-sdk/tests/task_sdk/execution_time/schema/test_migrator.py

@@ -424,6 +424,7 @@ def startup_details(self):
                     consumed_asset_events=[],
                     partition_key=None,
                 ),
+                dag_run_conf_json='{"records":[{"id":1,"value":"one"},{"id":2,"value":"two"}]}',
                 max_tries=1,
                 arg_bindings=[
                     # No value_schema: the unconstrained ("any") case rides through the migrator too.
@@ -453,6 +454,10 @@ def real_migrator(self) -> SchemaVersionMigrator:
     def test_downgrade_strips_arg_bindings_for_previous_version(self, real_migrator, startup_details):
         out = real_migrator.downgrade(startup_details, "2026-06-16").model_dump()
         assert "arg_bindings" not in out["ti_context"]
+        assert "dag_run_conf_json" not in out["ti_context"]
+        assert out["ti_context"]["dag_run"]["conf"] == {
+            "records": [{"id": 1, "value": "one"}, {"id": 2, "value": "two"}]
+        }
 
     def test_head_version_keeps_arg_bindings(self, real_migrator, startup_details):
         from airflow.sdk.api.datamodels._generated import LiteralArgBinding, XComArgBinding

task-sdk/tests/task_sdk/execution_time/test_comms.py

@@ -99,6 +99,7 @@ def test_recv_StartupDetails(self, socket_pair):
                     "consumed_asset_events": [],
                     "partition_key": None,
                 },
+                "dag_run_conf_json": '{"records":[{"id":1,"value":"one"},{"id":2,"value":"two"}]}',
                 "max_tries": 0,
                 "should_retry": False,
                 "variables": None,
@@ -123,6 +124,9 @@ def test_recv_StartupDetails(self, socket_pair):
         assert msg.dag_rel_path == "/dev/null"
         assert msg.bundle_info == BundleInfo(name="any-name", version="any-version")
         assert msg.start_date == timezone.datetime(2024, 12, 1, 1)
+        assert msg.ti_context.dag_run.conf is None
+        assert msg.ti_context.dag_run_conf_json is not None
+        assert '"records"' in msg.ti_context.dag_run_conf_json
 
     def test_huge_payload(self, socket_pair):
         r, w = socket_pair

task-sdk/tests/task_sdk/execution_time/test_task_runner.py

@@ -591,6 +591,26 @@ def send_side_effect(msg=None, **kwargs):
     assert error is None
 
 
+def test_template_context_materializes_opaque_dag_run_conf_in_task(create_runtime_ti):
+    """Serialized DagRun.conf is decoded once, in the task process."""
+    task = BaseOperator(task_id="opaque-conf")
+    runtime_ti = create_runtime_ti(
+        task=task,
+        dag_run_conf_json='{"records":[{"id":1,"value":"one"},{"id":2,"value":"two"}]}',
+    )
+
+    context = runtime_ti.get_template_context()
+    decoded_conf = context["dag_run"].conf
+    repeated_context = runtime_ti.get_template_context()
+
+    assert decoded_conf == {
+        "records": [{"id": 1, "value": "one"}, {"id": 2, "value": "two"}]
+    }
+    assert repeated_context["dag_run"].conf is decoded_conf
+    assert runtime_ti._ti_context_from_server is not None
+    assert runtime_ti._ti_context_from_server.dag_run_conf_json is None
+
+
 def test_run_signals_fail_closed_when_failure_terminal_send_fails(create_runtime_ti, mock_supervisor_comms):
     """
     When the task FAILS and the terminal-state send to the supervisor fails too