Move Arangodb to new provider structure

.github/boring-cyborg.yml

@@ -37,16 +37,10 @@ labelPRBasedOnFilePath:
     - providers/apache/cassandra/**
 
   provider:apache-drill:
-    - providers/src/airflow/providers/apache/drill/**/*
-    - docs/apache-airflow-providers-apache-drill/**/*
-    - providers/tests/apache/drill/**/*
-    - providers/tests/system/apache/drill/**/*
+    - providers/apache/drill/**
 
   provider:apache-druid:
-    - providers/src/airflow/providers/apache/druid/**/*
-    - docs/apache-airflow-providers-apache-druid/**/*
-    - providers/tests/apache/druid/**/*
-    - providers/tests/system/apache/druid/**/*
+    - providers/apache/druid/**
 
   provider:apache-flink:
     - providers/src/airflow/providers/apache/flink/**/*
@@ -94,9 +88,7 @@ labelPRBasedOnFilePath:
     - providers/apprise/**
 
   provider:arangodb:
-    - providers/src/airflow/providers/arangodb/**/*
-    - docs/apache-airflow-providers-arangodb/**/*
-    - providers/tests/arangodb/**/*
+    - providers/arangodb/**
 
   provider:asana:
     - providers/asana/**
@@ -147,10 +139,7 @@ labelPRBasedOnFilePath:
     - providers/datadog/**
 
   provider:dbt-cloud:
-    - providers/src/airflow/providers/dbt/cloud/**/*
-    - docs/apache-airflow-providers-dbt-cloud/**/*
-    - providers/tests/dbt/cloud/**/*
-    - providers/tests/system/dbt/cloud/**/*
+    - providers/dbt/cloud/**
 
   provider:dingding:
     - providers/src/airflow/providers/dingding/**/*
@@ -168,10 +157,7 @@ labelPRBasedOnFilePath:
     - providers/edge/**
 
   provider:elasticsearch:
-    - providers/src/airflow/providers/elasticsearch/**/*
-    - docs/apache-airflow-providers-elasticsearch/**/*
-    - providers/tests/elasticsearch/**/*
-    - providers/tests/system/elasticsearch/**/*
+    - providers/elasticsearch/**
 
   provider:exasol:
     - providers/exasol/**
@@ -188,10 +174,7 @@ labelPRBasedOnFilePath:
     - providers/ftp/**
 
   provider:github:
-    - providers/src/airflow/providers/github/**/*
-    - docs/apache-airflow-providers-github/**/*
-    - providers/tests/github/**/*
-    - providers/tests/system/github/**/*
+    - providers/github/**
 
   provider:google:
     - providers/src/airflow/providers/google/**/*
@@ -270,10 +253,7 @@ labelPRBasedOnFilePath:
     - providers/openlineage/**
 
   provider:opensearch:
-    - providers/src/airflow/providers/opensearch/**/*
-    - docs/apache-airflow-providers-opensearch/**/*
-    - providers/tests/opensearch/**/*
-    - providers/tests/system/opensearch/**/*
+    - providers/opensearch/**
 
   provider:opsgenie:
     - providers/opsgenie/**
@@ -322,7 +302,7 @@ labelPRBasedOnFilePath:
     - providers/segment/**
 
   provider:sendgrid:
-    - providers/segment/**
+    - providers/sendgrid/**
 
   provider:sftp:
     - providers/sftp/**

.github/workflows/ci-image-checks.yml

@@ -289,7 +289,9 @@ jobs:
           key: cache-docs-inventory-v1-${{ hashFiles('pyproject.toml') }}
           if-no-files-found: 'error'
           retention-days: '2'
-        if: steps.restore-docs-inventory-cache != 'true'
+        # If we upload from multiple matrix jobs we could end up with a race condition. so just pick one job
+        # to be responsible for updating it. https://github.com/actions/upload-artifact/issues/506
+        if: steps.restore-docs-inventory-cache != 'true' && matrix.flag == '--docs-only'
       - name: "Upload build docs"
         uses: actions/upload-artifact@v4
         with:

airflow/api_connexion/schemas/task_instance_schema.py

@@ -60,6 +60,7 @@ class Meta:
     priority_weight = auto_field()
     operator = auto_field()
     queued_dttm = auto_field(data_key="queued_when")
+    scheduled_dttm = auto_field(data_key="scheduled_when")
     pid = auto_field()
     executor = auto_field()
     executor_config = auto_field()
@@ -102,6 +103,7 @@ class Meta:
     priority_weight = auto_field()
     operator = auto_field()
     queued_dttm = auto_field(data_key="queued_when")
+    scheduled_dttm = auto_field(data_key="scheduled_when")
     pid = auto_field()
     executor = auto_field()
     executor_config = auto_field()

airflow/api_fastapi/core_api/datamodels/task_instances.py

@@ -64,6 +64,7 @@ class TaskInstanceResponse(BaseModel):
     priority_weight: int | None
     operator: str | None
     queued_dttm: datetime | None = Field(alias="queued_when")
+    scheduled_dttm: datetime | None = Field(alias="scheduled_when")
     pid: int | None
     executor: str | None
     executor_config: Annotated[str, BeforeValidator(str)]
@@ -147,6 +148,7 @@ class TaskInstanceHistoryResponse(BaseModel):
     priority_weight: int | None
     operator: str | None
     queued_dttm: datetime | None = Field(alias="queued_when")
+    scheduled_dttm: datetime | None = Field(alias="scheduled_when")
     pid: int | None
     executor: str | None
     executor_config: Annotated[str, BeforeValidator(str)]

airflow/api_fastapi/core_api/openapi/v1-generated.yaml

@@ -9643,6 +9643,12 @@ components:
             format: date-time
           - type: 'null'
           title: Queued When
+        scheduled_when:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: 'null'
+          title: Scheduled When
         pid:
           anyOf:
           - type: integer
@@ -9677,6 +9683,7 @@ components:
       - priority_weight
       - operator
       - queued_when
+      - scheduled_when
       - pid
       - executor
       - executor_config
@@ -9770,6 +9777,12 @@ components:
             format: date-time
           - type: 'null'
           title: Queued When
+        scheduled_when:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: 'null'
+          title: Scheduled When
         pid:
           anyOf:
           - type: integer
@@ -9828,6 +9841,7 @@ components:
       - priority_weight
       - operator
       - queued_when
+      - scheduled_when
       - pid
       - executor
       - executor_config

airflow/api_fastapi/execution_api/routes/xcoms.py

@@ -28,6 +28,7 @@
 from airflow.api_fastapi.execution_api import deps
 from airflow.api_fastapi.execution_api.datamodels.token import TIToken
 from airflow.api_fastapi.execution_api.datamodels.xcom import XComResponse
+from airflow.models.taskmap import TaskMap
 from airflow.models.xcom import BaseXCom
 
 # TODO: Add dependency on JWT token
@@ -55,7 +56,7 @@ def get_xcom(
     map_index: Annotated[int, Query()] = -1,
 ) -> XComResponse:
     """Get an Airflow XCom from database - not other XCom Backends."""
-    if not has_xcom_access(key, token):
+    if not has_xcom_access(dag_id, run_id, task_id, key, token):
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
             detail={
@@ -104,6 +105,8 @@ def get_xcom(
     return XComResponse(key=key, value=xcom_value)
 
 
+# TODO: once we have JWT tokens, then remove dag_id/run_id/task_id from the URL and just use the info in
+# the token
 @router.post(
     "/{dag_id}/{run_id}/{task_id}/{key}",
     status_code=status.HTTP_201_CREATED,
@@ -139,8 +142,23 @@ def set_xcom(
     token: deps.TokenDep,
     session: SessionDep,
     map_index: Annotated[int, Query()] = -1,
+    mapped_length: Annotated[
+        int | None, Query(description="Number of mapped tasks this value expands into")
+    ] = None,
 ):
     """Set an Airflow XCom."""
+    from airflow.configuration import conf
+
+    if not has_xcom_access(dag_id, run_id, task_id, key, token, write=True):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail={
+                "reason": "access_denied",
+                "message": f"Task does not have access to set XCom key '{key}'",
+            },
+        )
+
+    # TODO: This is in-efficient. We json.loads it here for BaseXCom.set to then json.dump it!
     try:
         json.loads(value)
     except json.JSONDecodeError:
@@ -152,14 +170,30 @@ def set_xcom(
             },
         )
 
-    if not has_xcom_access(key, token):
-        raise HTTPException(
-            status_code=status.HTTP_403_FORBIDDEN,
-            detail={
-                "reason": "access_denied",
-                "message": f"Task does not have access to set XCom key '{key}'",
-            },
+    if mapped_length is not None:
+        task_map = TaskMap(
+            dag_id=dag_id,
+            task_id=task_id,
+            run_id=run_id,
+            map_index=map_index,
+            length=mapped_length,
+            keys=None,
         )
+        max_map_length = conf.getint("core", "max_map_length", fallback=1024)
+        if task_map.length > max_map_length:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail={
+                    "reason": "unmappable_return_value_length",
+                    "message": "pushed value is too large to map as a downstream's dependency",
+                },
+            )
+        session.add(task_map)
+
+    # else:
+    # TODO: Can/should we check if a client _hasn't_ provided this for an upstream of a mapped task? That
+    # means loading the serialized dag and that seems like a relatively costly operation for minimal benefit
+    # (the mapped task would fail in a moment as it can't be expanded anyway.)
 
     # We use `BaseXCom.set` to set XComs directly to the database, bypassing the XCom Backend.
     try:
@@ -184,13 +218,16 @@ def set_xcom(
     return {"message": "XCom successfully set"}
 
 
-def has_xcom_access(xcom_key: str, token: TIToken) -> bool:
+def has_xcom_access(
+    dag_id: str, run_id: str, task_id: str, xcom_key: str, token: TIToken, write: bool = False
+) -> bool:
     """Check if the task has access to the XCom."""
     # TODO: Placeholder for actual implementation
 
     ti_key = token.ti_key
     log.debug(
-        "Checking access for task instance with key '%s' to XCom '%s'",
+        "Checking %s XCom access for xcom from TaskInstance with key '%s' to XCom '%s'",
+        "write" if write else "read",
         ti_key,
         xcom_key,
     )

airflow/cli/commands/remote_commands/task_command.py

@@ -46,8 +46,8 @@
 from airflow.models import TaskInstance
 from airflow.models.dag import DAG, _run_inline_trigger
 from airflow.models.dagrun import DagRun
-from airflow.models.param import ParamsDict
 from airflow.models.taskinstance import TaskReturnCode
+from airflow.sdk.definitions.param import ParamsDict
 from airflow.settings import IS_EXECUTOR_CONTAINER, IS_K8S_EXECUTOR_POD
 from airflow.ti_deps.dep_context import DepContext
 from airflow.ti_deps.dependencies_deps import SCHEDULER_QUEUED_DEPS

airflow/example_dags/example_params_trigger_ui.py

@@ -27,7 +27,7 @@
 
 from airflow.decorators import task
 from airflow.models.dag import DAG
-from airflow.models.param import Param, ParamsDict
+from airflow.sdk import Param, ParamsDict
 from airflow.utils.trigger_rule import TriggerRule
 
 # [START params_trigger]

airflow/example_dags/example_params_ui_tutorial.py

@@ -29,7 +29,7 @@
 
 from airflow.decorators import task
 from airflow.models.dag import DAG
-from airflow.models.param import Param, ParamsDict
+from airflow.sdk import Param, ParamsDict
 
 with (
     DAG(