[v3-3-test] Fix asset materialization dropping partition date on partitioned Dag runs (#69314)

airflow-core/src/airflow/api_fastapi/core_api/routes/public/assets.py

@@ -464,6 +464,7 @@ def materialize_asset(
             triggering_user_name=user.get_name(),
             state=DagRunState.QUEUED,
             partition_key=params["partition_key"],
+            partition_date=params["partition_date"],
             note=params["note"],
             session=session,
         )

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py

@@ -41,7 +41,9 @@
 from airflow.models.serialized_dag import SerializedDagModel
 from airflow.models.trigger import Trigger
 from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.sdk import Asset
 from airflow.timetables.simple import PartitionedAtRuntime
+from airflow.timetables.trigger import CronPartitionTimetable
 from airflow.utils.session import provide_session
 from airflow.utils.state import DagRunState
 from airflow.utils.types import DagRunType
@@ -1684,6 +1686,40 @@ def test_should_respond_400_on_invalid_dag_run_id(self, test_client):
         assert response.status_code == 400
         assert "must not contain '..'" in response.json()["detail"]
 
+    @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+    def test_should_respond_200_with_partition_date_for_partitioned_dag(
+        self, test_client, dag_maker, session
+    ):
+        """Materializing a Dag with a real partitioned timetable must populate partition_date.
+
+        Regression guard: before this fix, `partition_date` resolved by `validate_context` was
+        dropped when creating the run, unlike the sibling `/dags/{dag_id}/dagRuns` trigger route.
+        """
+        partitioned_dag_id = "test_materialize_populates_partition_date"
+        asset = Asset(name="materialize_partition_date_asset", uri="s3://bucket/materialize-partition-date")
+        with dag_maker(
+            dag_id=partitioned_dag_id,
+            schedule=CronPartitionTimetable("0 0 * * *", timezone="UTC"),
+            start_date=DEFAULT_DATE,
+            session=session,
+            serialized=True,
+        ):
+            EmptyOperator(task_id="task", outlets=[asset])
+        session.commit()
+
+        asset_id = session.scalar(select(AssetModel.id).where(AssetModel.uri == asset.uri))
+
+        response = test_client.post(
+            f"/assets/{asset_id}/materialize",
+            json={"partition_key": "2025-06-01T00:00:00"},
+        )
+        assert response.status_code == 200
+
+        dag_run = session.scalar(select(DagRun).where(DagRun.dag_id == partitioned_dag_id))
+        assert dag_run is not None
+        assert dag_run.partition_key == "2025-06-01T00:00:00"
+        assert dag_run.partition_date == timezone.datetime(2025, 6, 1)
+
 
 class TestGetAssetQueuedEvents(TestQueuedEventEndpoint):
     @pytest.mark.usefixtures("time_freezer")