Merge branch 'main' into deserialize-xcom-native

Author: amoghrajesh

.github/CODEOWNERS

@@ -35,11 +35,12 @@
 # UI
 /airflow-core/src/airflow/ui/ @bbovenzi @pierrejeambrun @ryanahamilton @jscheffl @shubhamraj-git
 
-# Translations
+# Translations (i18n)
 airflow-core/src/airflow/ui/src/i18n/locales/de/ @jscheffl
 airflow-core/src/airflow/ui/src/i18n/locales/zh-TW/ @Lee-W
 airflow-core/src/airflow/ui/src/i18n/locales/nl/ @BasPH # not codeowner but engaged: @DjVinnii
 airflow-core/src/airflow/ui/src/i18n/locales/pl/ @potiuk @mobuchowski # not codeowner but engaged: @kacpermuda
+airflow-core/src/airflow/ui/src/i18n/locales/he/ @eladkal @shahar1 @romsharon98
 
 # Security/Permissions
 /airflow-core/src/airflow/security/permissions.py @vincbeck

.github/boring-cyborg.yml

@@ -111,9 +111,6 @@ labelPRBasedOnFilePath:
   provider:common-sql:
     - providers/common/sql/**
 
-  provider:standard:
-    - providers/standard/**
-
   provider:databricks:
     - providers/databricks/**
 
@@ -180,6 +177,9 @@ labelPRBasedOnFilePath:
   provider:jenkins:
     - providers/jenkins/**
 
+  provider:keycloak:
+    - providers/keycloak/**
+
   provider:microsoft-azure:
     - providers/microsoft/azure/**
 
@@ -219,7 +219,7 @@ labelPRBasedOnFilePath:
   provider:opsgenie:
     - providers/opsgenie/**
 
-  provider:Oracle:
+  provider:oracle:
     - providers/oracle/**
 
   provider:pagerduty:
@@ -279,6 +279,9 @@ labelPRBasedOnFilePath:
   provider:ssh:
     - providers/ssh/**
 
+  provider:standard:
+    - providers/standard/**
+
   provider:tableau:
     - providers/tableau/**
 
@@ -344,6 +347,30 @@ labelPRBasedOnFilePath:
     - airflow-core/docs/ui.rst
     - airflow-core/src/airflow/ui/**/*
 
+  area:translations:
+    - airflow-core/src/airflow/ui/src/i18n/**/*
+
+  translation:default:
+    - airflow-core/src/airflow/ui/src/i18n/locales/en/*
+
+  translation:de:
+    - airflow-core/src/airflow/ui/src/i18n/locales/de/*
+
+  translation:he:
+    - airflow-core/src/airflow/ui/src/i18n/locales/he/*
+
+  translation:ko:
+    - airflow-core/src/airflow/ui/src/i18n/locales/ko/*
+
+  translation:nl:
+    - airflow-core/src/airflow/ui/src/i18n/locales/nl/*
+
+  translation:pl:
+    - airflow-core/src/airflow/ui/src/i18n/locales/pl/*
+
+  translation:zh-TW:
+    - airflow-core/src/airflow/ui/src/i18n/locales/zh-TW/*
+
   area:CLI:
     - airflow-core/src/airflow/cli/**/*.py
     - airflow-core/tests/unit/cli/**/*.py

.pre-commit-config.yaml

@@ -629,6 +629,7 @@ repos:
         pass_filenames: true
         exclude: >
           (?x)
+          ^airflow-core/src/airflow/ui/src/i18n/config\.ts$|
           ^airflow-core/src/airflow/ui/openapi-gen/|
           ^airflow-core/src/airflow/ui/src/i18n/locales/de/README\.md$|
           ^airflow-core/src/airflow/cli/commands/local_commands/fastapi_api_command\.py$|

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

@@ -65,11 +65,12 @@
 from airflow.api_fastapi.core_api.security import GetUserDep, ReadableTIFilterDep, requires_access_dag
 from airflow.api_fastapi.core_api.services.public.task_instances import (
     BulkTaskInstanceService,
+    _patch_task_instance_note,
+    _patch_task_instance_state,
     _patch_ti_validate_request,
 )
 from airflow.api_fastapi.logging.decorators import action_logging
 from airflow.exceptions import TaskNotFound
-from airflow.listeners.listener import get_listener_manager
 from airflow.models import Base, DagRun
 from airflow.models.taskinstance import TaskInstance as TI, clear_task_instances
 from airflow.models.taskinstancehistory import TaskInstanceHistory as TIH
@@ -848,46 +849,22 @@ def patch_task_instance(
 
     for key, _ in data.items():
         if key == "new_state":
-            tis = dag.set_task_instance_state(
+            _patch_task_instance_state(
                 task_id=task_id,
-                run_id=dag_run_id,
-                map_indexes=[map_index] if map_index is not None else None,
-                state=data["new_state"],
-                upstream=body.include_upstream,
-                downstream=body.include_downstream,
-                future=body.include_future,
-                past=body.include_past,
-                commit=True,
+                dag_run_id=dag_run_id,
+                dag=dag,
+                task_instance_body=body,
+                data=data,
                 session=session,
             )
-            if not tis:
-                raise HTTPException(
-                    status.HTTP_409_CONFLICT, f"Task id {task_id} is already in {data['new_state']} state"
-                )
-
-            for ti in tis:
-                try:
-                    if data["new_state"] == TaskInstanceState.SUCCESS:
-                        get_listener_manager().hook.on_task_instance_success(
-                            previous_state=None, task_instance=ti
-                        )
-                    elif data["new_state"] == TaskInstanceState.FAILED:
-                        get_listener_manager().hook.on_task_instance_failed(
-                            previous_state=None,
-                            task_instance=ti,
-                            error=f"TaskInstance's state was manually set to `{TaskInstanceState.FAILED}`.",
-                        )
-                except Exception:
-                    log.exception("error calling listener")
 
         elif key == "note":
-            for ti in tis:
-                if update_mask or body.note is not None:
-                    if ti.task_instance_note is None:
-                        ti.note = (body.note, user.get_id())
-                    else:
-                        ti.task_instance_note.content = body.note
-                        ti.task_instance_note.user_id = user.get_id()
+            _patch_task_instance_note(
+                task_instance_body=body,
+                tis=tis,
+                user=user,
+                update_mask=update_mask,
+            )
 
     return TaskInstanceCollectionResponse(
         task_instances=[

airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py

@@ -94,6 +94,64 @@ def _patch_ti_validate_request(
     return dag, list(tis), body.model_dump(include=fields_to_update, by_alias=True)
 
 
+def _patch_task_instance_state(
+    task_id: str,
+    dag_run_id: str,
+    dag: DAG,
+    task_instance_body: BulkTaskInstanceBody | PatchTaskInstanceBody,
+    data: dict,
+    session: Session,
+) -> None:
+    map_index = getattr(task_instance_body, "map_index", None)
+    map_indexes = None if map_index is None else [map_index]
+
+    updated_tis = dag.set_task_instance_state(
+        task_id=task_id,
+        run_id=dag_run_id,
+        map_indexes=map_indexes,
+        state=data["new_state"],
+        upstream=task_instance_body.include_upstream,
+        downstream=task_instance_body.include_downstream,
+        future=task_instance_body.include_future,
+        past=task_instance_body.include_past,
+        commit=True,
+        session=session,
+    )
+    if not updated_tis:
+        raise HTTPException(
+            status.HTTP_409_CONFLICT,
+            f"Task id {task_id} is already in {data['new_state']} state",
+        )
+
+    for ti in updated_tis:
+        try:
+            if data["new_state"] == TaskInstanceState.SUCCESS:
+                get_listener_manager().hook.on_task_instance_success(previous_state=None, task_instance=ti)
+            elif data["new_state"] == TaskInstanceState.FAILED:
+                get_listener_manager().hook.on_task_instance_failed(
+                    previous_state=None,
+                    task_instance=ti,
+                    error=f"TaskInstance's state was manually set to `{TaskInstanceState.FAILED}`.",
+                )
+        except Exception:
+            log.exception("error calling listener")
+
+
+def _patch_task_instance_note(
+    task_instance_body: BulkTaskInstanceBody | PatchTaskInstanceBody,
+    tis: list[TI],
+    user: GetUserDep,
+    update_mask: list[str] | None = Query(None),
+) -> None:
+    for ti in tis:
+        if update_mask or task_instance_body.note is not None:
+            if ti.task_instance_note is None:
+                ti.note = (task_instance_body.note, user.get_id())
+            else:
+                ti.task_instance_note.content = task_instance_body.note
+                ti.task_instance_note.user_id = user.get_id()
+
+
 class BulkTaskInstanceService(BulkService[BulkTaskInstanceBody]):
     """Service for handling bulk operations on task instances."""
 
@@ -134,55 +192,6 @@ def categorize_task_instances(
         not_found_task_keys = {(task_id, map_index) for task_id, map_index in task_ids} - matched_task_keys
         return task_instances_map, matched_task_keys, not_found_task_keys
 
-    def _patch_task_instance_state(
-        self,
-        dag: DAG,
-        task_instance_body: BulkTaskInstanceBody,
-        data: dict,
-    ) -> None:
-        map_indexes = None if task_instance_body.map_index is None else [task_instance_body.map_index]
-
-        updated_tis = dag.set_task_instance_state(
-            task_id=task_instance_body.task_id,
-            run_id=self.dag_run_id,
-            map_indexes=map_indexes,
-            state=data["new_state"],
-            upstream=task_instance_body.include_upstream,
-            downstream=task_instance_body.include_downstream,
-            future=task_instance_body.include_future,
-            past=task_instance_body.include_past,
-            commit=True,
-            session=self.session,
-        )
-        if not updated_tis:
-            raise HTTPException(
-                status.HTTP_409_CONFLICT,
-                f"Task id {task_instance_body.task_id} is already in {data['new_state']} state",
-            )
-        for ti in updated_tis:
-            try:
-                if data["new_state"] == TaskInstanceState.SUCCESS:
-                    get_listener_manager().hook.on_task_instance_success(
-                        previous_state=None, task_instance=ti
-                    )
-                elif data["new_state"] == TaskInstanceState.FAILED:
-                    get_listener_manager().hook.on_task_instance_failed(
-                        previous_state=None,
-                        task_instance=ti,
-                        error=f"TaskInstance's state was manually set to `{TaskInstanceState.FAILED}`.",
-                    )
-            except Exception:
-                log.exception("error calling listener")
-
-    def _patch_task_instance_note(self, task_instance_body: BulkTaskInstanceBody, tis: list[TI]) -> None:
-        for ti in tis:
-            if task_instance_body.note is not None:
-                if ti.task_instance_note is None:
-                    ti.note = (task_instance_body.note, self.user.get_id())
-                else:
-                    ti.task_instance_note.content = task_instance_body.note
-                    ti.task_instance_note.user_id = self.user.get_id()
-
     def handle_bulk_create(
         self, action: BulkCreateAction[BulkTaskInstanceBody], results: BulkActionResponse
     ) -> None:
@@ -232,13 +241,18 @@ def handle_bulk_update(
 
                 for key, _ in data.items():
                     if key == "new_state":
-                        self._patch_task_instance_state(
+                        _patch_task_instance_state(
+                            task_id=task_instance_body.task_id,
+                            dag_run_id=self.dag_run_id,
                             dag=dag,
                             task_instance_body=task_instance_body,
+                            session=self.session,
                             data=data,
                         )
                     elif key == "note":
-                        self._patch_task_instance_note(task_instance_body=task_instance_body, tis=tis)
+                        _patch_task_instance_note(
+                            task_instance_body=task_instance_body, tis=tis, user=self.user
+                        )
 
                 results.success.append(task_instance_body.task_id)
         except ValidationError as e:

airflow-core/src/airflow/cli/cli_config.py

@@ -483,6 +483,15 @@ def string_lower_type(val):
     type=positive_int(allow_zero=False),
     help="Wait time between retries in seconds",
 )
+ARG_DB_BATCH_SIZE = Arg(
+    ("--batch-size",),
+    default=None,
+    type=positive_int(allow_zero=False),
+    help=(
+        "Maximum number of rows to delete or archive in a single transaction.\n"
+        "Lower values reduce long-running locks but increase the number of batches."
+    ),
+)
 
 # pool
 ARG_POOL_NAME = Arg(("pool",), metavar="NAME", help="Pool name")
@@ -1452,6 +1461,7 @@ class GroupCommand(NamedTuple):
             ARG_VERBOSE,
             ARG_YES,
             ARG_DB_SKIP_ARCHIVE,
+            ARG_DB_BATCH_SIZE,
         ),
     ),
     ActionCommand(

airflow-core/src/airflow/cli/commands/db_command.py

@@ -283,6 +283,7 @@ def cleanup_tables(args):
         verbose=args.verbose,
         confirm=not args.yes,
         skip_archive=args.skip_archive,
+        batch_size=args.batch_size,
     )
 
 

airflow-core/src/airflow/ui/index.html

@@ -1,5 +1,5 @@
 <!doctype html>
-<html lang="en" style="height: 100%">
+<html style="height: 100%">
   <head>
     <meta charset="UTF-8" />
     <base href="{{ backend_server_base_url }}" />

airflow-core/src/airflow/ui/package.json

@@ -17,7 +17,7 @@
   },
   "dependencies": {
     "@chakra-ui/anatomy": "^2.3.4",
-    "@chakra-ui/react": "^3.17.0",
+    "@chakra-ui/react": "^3.20.0",
     "@codemirror/lang-json": "^6.0.1",
     "@emotion/react": "^11.14.0",
     "@tanstack/react-query": "^5.75.1",