Merge branch 'main' into feature/remove-large-in-clause-in-assets

Author: Nataneljpwd

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

@@ -296,7 +296,7 @@ class DagRun(StrictBaseModel):
     data_interval_start: UtcDateTime | None
     data_interval_end: UtcDateTime | None
     run_after: UtcDateTime
-    start_date: UtcDateTime
+    start_date: UtcDateTime | None
     end_date: UtcDateTime | None
     clear_number: int = 0
     run_type: DagRunType

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

@@ -35,13 +35,20 @@
 )
 from airflow.api_fastapi.execution_api.versions.v2026_03_31 import (
     AddNoteField,
+    MakeDagRunStartDateNullable,
     ModifyDeferredTaskKwargsToJsonValue,
     RemoveUpstreamMapIndexesField,
 )
 
 bundle = VersionBundle(
     HeadVersion(),
-    Version("2026-03-31", ModifyDeferredTaskKwargsToJsonValue, RemoveUpstreamMapIndexesField, AddNoteField),
+    Version(
+        "2026-03-31",
+        MakeDagRunStartDateNullable,
+        ModifyDeferredTaskKwargsToJsonValue,
+        RemoveUpstreamMapIndexesField,
+        AddNoteField,
+    ),
     Version("2025-12-08", MovePreviousRunEndpoint, AddDagRunDetailEndpoint),
     Version("2025-11-07", AddPartitionKeyField),
     Version("2025-11-05", AddTriggeringUserNameField),

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

@@ -21,6 +21,7 @@
 
 from cadwyn import ResponseInfo, VersionChange, convert_response_to_previous_version_for, schema
 
+from airflow.api_fastapi.common.types import UtcDateTime
 from airflow.api_fastapi.execution_api.datamodels.taskinstance import (
     DagRun,
     TIDeferredStatePayload,
@@ -68,3 +69,29 @@ def remove_note_field(response: ResponseInfo) -> None:  # type: ignore[misc]
         """Remove note field for older API versions."""
         if "dag_run" in response.body and isinstance(response.body["dag_run"], dict):
             response.body["dag_run"].pop("note", None)
+
+
+class MakeDagRunStartDateNullable(VersionChange):
+    """Make DagRun.start_date field nullable for runs that haven't started yet."""
+
+    description = __doc__
+
+    instructions_to_migrate_to_previous_version = (schema(DagRun).field("start_date").had(type=UtcDateTime),)
+
+    @convert_response_to_previous_version_for(TIRunContext)  # type: ignore[arg-type]
+    def ensure_start_date_in_ti_run_context(response: ResponseInfo) -> None:  # type: ignore[misc]
+        """
+        Ensure start_date is never None in DagRun for previous API versions.
+
+        Older Task SDK clients expect start_date to be non-nullable. When the
+        DagRun hasn't started yet (e.g. queued), fall back to run_after.
+        """
+        dag_run = response.body.get("dag_run")
+        if isinstance(dag_run, dict) and dag_run.get("start_date") is None:
+            dag_run["start_date"] = dag_run.get("run_after")
+
+    @convert_response_to_previous_version_for(DagRun)  # type: ignore[arg-type]
+    def ensure_start_date_in_dag_run(response: ResponseInfo) -> None:  # type: ignore[misc]
+        """Ensure start_date is never None in direct DagRun responses for previous API versions."""
+        if response.body.get("start_date") is None:
+            response.body["start_date"] = response.body.get("run_after")

airflow-core/src/airflow/models/connection.py

@@ -226,7 +226,7 @@ def _normalize_conn_type(conn_type):
     def _parse_from_uri(self, uri: str):
         schemes_count_in_uri = uri.count("://")
         if schemes_count_in_uri > 2:
-            raise AirflowException(f"Invalid connection string: {uri}.")
+            raise AirflowException("Invalid connection string.")
         host_with_protocol = schemes_count_in_uri == 2
         uri_parts = urlsplit(uri)
         conn_type = uri_parts.scheme
@@ -235,7 +235,7 @@ def _parse_from_uri(self, uri: str):
         if host_with_protocol:
             uri_splits = rest_of_the_url.split("://", 1)
             if "@" in uri_splits[0] or ":" in uri_splits[0]:
-                raise AirflowException(f"Invalid connection string: {uri}.")
+                raise AirflowException("Invalid connection string.")
         uri_parts = urlsplit(rest_of_the_url)
         protocol = uri_parts.scheme if host_with_protocol else None
         host = _parse_netloc_to_hostname(uri_parts)

airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_03_31/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

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

@@ -0,0 +1,127 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import pytest
+
+from airflow._shared.timezones import timezone
+from airflow.utils.state import DagRunState, State
+
+from tests_common.test_utils.db import clear_db_runs
+
+pytestmark = pytest.mark.db_test
+
+TIMESTAMP_STR = "2024-09-30T12:00:00Z"
+TIMESTAMP = timezone.parse(TIMESTAMP_STR)
+
+RUN_PATCH_BODY = {
+    "state": "running",
+    "hostname": "test-hostname",
+    "unixname": "test-user",
+    "pid": 12345,
+    "start_date": TIMESTAMP_STR,
+}
+
+
+@pytest.fixture
+def old_ver_client(client):
+    """Client configured to use API version before start_date nullable change."""
+    client.headers["Airflow-API-Version"] = "2025-12-08"
+    return client
+
+
+class TestDagRunStartDateNullableBackwardCompat:
+    """Test that older API versions get a non-null start_date fallback."""
+
+    @pytest.fixture(autouse=True)
+    def _freeze_time(self, time_machine):
+        time_machine.move_to(TIMESTAMP_STR, tick=False)
+
+    def setup_method(self):
+        clear_db_runs()
+
+    def teardown_method(self):
+        clear_db_runs()
+
+    def test_old_version_gets_run_after_when_start_date_is_null(
+        self,
+        old_ver_client,
+        session,
+        create_task_instance,
+    ):
+        ti = create_task_instance(
+            task_id="test_start_date_nullable",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.QUEUED,
+            session=session,
+            start_date=TIMESTAMP,
+        )
+        ti.dag_run.start_date = None  # DagRun has not started yet
+        session.commit()
+
+        response = old_ver_client.patch(f"/execution/task-instances/{ti.id}/run", json=RUN_PATCH_BODY)
+        dag_run = response.json()["dag_run"]
+
+        assert response.status_code == 200
+        assert dag_run["start_date"] is not None
+        assert dag_run["start_date"] == dag_run["run_after"]
+
+    def test_head_version_allows_null_start_date(
+        self,
+        client,
+        session,
+        create_task_instance,
+    ):
+        ti = create_task_instance(
+            task_id="test_start_date_null_head",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.QUEUED,
+            session=session,
+            start_date=TIMESTAMP,
+        )
+        ti.dag_run.start_date = None  # DagRun has not started yet
+        session.commit()
+
+        response = client.patch(f"/execution/task-instances/{ti.id}/run", json=RUN_PATCH_BODY)
+        dag_run = response.json()["dag_run"]
+
+        assert response.status_code == 200
+        assert dag_run["start_date"] is None
+
+    def test_old_version_preserves_real_start_date(
+        self,
+        old_ver_client,
+        session,
+        create_task_instance,
+    ):
+        ti = create_task_instance(
+            task_id="test_start_date_preserved",
+            state=State.QUEUED,
+            dagrun_state=DagRunState.RUNNING,
+            session=session,
+            start_date=TIMESTAMP,
+        )
+        assert ti.dag_run.start_date == TIMESTAMP  # DagRun has already started
+        session.commit()
+
+        response = old_ver_client.patch(f"/execution/task-instances/{ti.id}/run", json=RUN_PATCH_BODY)
+        dag_run = response.json()["dag_run"]
+
+        assert response.status_code == 200
+        assert dag_run["start_date"] is not None, "start_date should not be None when DagRun has started"
+        assert dag_run["start_date"] == TIMESTAMP.isoformat().replace("+00:00", "Z")

airflow-core/tests/unit/models/test_connection.py

@@ -138,7 +138,7 @@ def clear_fernet_cache(self):
                 {"param": "value"},
                 None,
             ),
-            (
+            (  # Test password sanitisation
                 "type://user:pass@protocol://host:port?param=value",
                 None,
                 None,
@@ -147,7 +147,29 @@ def clear_fernet_cache(self):
                 None,
                 None,
                 None,
-                r"Invalid connection string: type://user:pass@protocol://host:port?param=value.",
+                r"Invalid connection string.",
+            ),
+            (
+                "foo:pwd@host://://",
+                None,
+                None,
+                None,
+                None,
+                None,
+                None,
+                None,
+                r"Invalid connection string.",
+            ),
+            (
+                "type://:foo@host/://://",
+                None,
+                None,
+                None,
+                None,
+                None,
+                None,
+                None,
+                r"Invalid connection string.",
             ),
             (
                 "type://host?int_param=123&bool_param=true&float_param=1.5&str_param=some_str",

providers/amazon/tests/system/amazon/aws/example_emr_eks.py

@@ -18,6 +18,7 @@
 
 import json
 import subprocess
+import time
 from datetime import datetime
 
 import boto3
@@ -57,13 +58,15 @@
 JOB_ROLE_ARN_KEY = "JOB_ROLE_ARN"
 JOB_ROLE_NAME_KEY = "JOB_ROLE_NAME"
 SUBNETS_KEY = "SUBNETS"
+UPDATE_TRUST_POLICY_WAIT_TIME_KEY = "UPDATE_TRUST_POLICY_WAIT_TIME_KEY"
 
 sys_test_context_task = (
     SystemTestContextBuilder()
     .add_variable(ROLE_ARN_KEY)
     .add_variable(JOB_ROLE_ARN_KEY)
     .add_variable(JOB_ROLE_NAME_KEY)
     .add_variable(SUBNETS_KEY, split_string=True)
+    .add_variable(UPDATE_TRUST_POLICY_WAIT_TIME_KEY, optional=True, default_value=10)
     .build()
 )
 
@@ -137,7 +140,7 @@ def delete_iam_oidc_identity_provider(cluster_name):
 
 
 @task
-def update_trust_policy_execution_role(cluster_name, cluster_namespace, role_name):
+def update_trust_policy_execution_role(cluster_name, cluster_namespace, role_name, wait_time):
     # Remove any already existing trusted entities added with "update-role-trust-policy"
     # Prevent getting an error "Cannot exceed quota for ACLSizePerRole"
     client = boto3.client("iam")
@@ -173,6 +176,9 @@ def update_trust_policy_execution_role(cluster_name, cluster_namespace, role_nam
     if build.returncode != 0:
         raise RuntimeError(err)
 
+    # Wait for IAM changes to propagate to avoid authentication failures
+    time.sleep(wait_time)
+
 
 @task(trigger_rule=TriggerRule.ALL_DONE)
 def delete_virtual_cluster(virtual_cluster_id):
@@ -193,6 +199,7 @@ def delete_virtual_cluster(virtual_cluster_id):
     subnets = test_context[SUBNETS_KEY]
     job_role_arn = test_context[JOB_ROLE_ARN_KEY]
     job_role_name = test_context[JOB_ROLE_NAME_KEY]
+    update_trust_policy_wait_time = test_context[UPDATE_TRUST_POLICY_WAIT_TIME_KEY]
 
     s3_bucket_name = f"{env_id}-bucket"
     eks_cluster_name = f"{env_id}-cluster"
@@ -316,7 +323,9 @@ def delete_virtual_cluster(virtual_cluster_id):
         create_cluster_and_nodegroup,
         await_create_nodegroup,
         run_eksctl_commands(eks_cluster_name, eks_namespace),
-        update_trust_policy_execution_role(eks_cluster_name, eks_namespace, job_role_name),
+        update_trust_policy_execution_role(
+            eks_cluster_name, eks_namespace, job_role_name, update_trust_policy_wait_time
+        ),
         # TEST BODY
         create_emr_eks_cluster,
         job_starter,

providers/apache/hive/src/airflow/providers/apache/hive/hooks/hive.py

@@ -877,8 +877,8 @@ def get_conn(self, schema: str | None = None) -> Any:
             auth_mechanism = db.extra_dejson.get("auth_mechanism", "KERBEROS")
             kerberos_service_name = db.extra_dejson.get("kerberos_service_name", "hive")
 
-        # Password should be set if and only if in LDAP or CUSTOM mode
-        if auth_mechanism in ("LDAP", "CUSTOM"):
+        # Password should be set if in LDAP, CUSTOM or PLAIN mode
+        if auth_mechanism in ("LDAP", "CUSTOM", "PLAIN"):
             password = db.password
 
         from pyhive.hive import connect

providers/apache/hive/tests/unit/apache/hive/hooks/test_hive.py

@@ -677,6 +677,26 @@ def test_get_conn_with_password(self, mock_connect):
                 database="default",
             )
 
+    @mock.patch("pyhive.hive.connect")
+    def test_get_conn_with_password_plain(self, mock_connect):
+        conn_id = "conn_plain_with_password"
+        conn_env = CONN_ENV_PREFIX + conn_id.upper()
+
+        with mock.patch.dict(
+            "os.environ",
+            {conn_env: "jdbc+hive2://login:password@localhost:10000/default?auth_mechanism=PLAIN"},
+        ):
+            HiveServer2Hook(hiveserver2_conn_id=conn_id).get_conn()
+            mock_connect.assert_called_once_with(
+                host="localhost",
+                port=10000,
+                auth="PLAIN",
+                kerberos_service_name=None,
+                username="login",
+                password="password",
+                database="default",
+            )
+
     @pytest.mark.parametrize(
         ("host", "port", "schema", "message"),
         [