Repository: apache/airflow

Issue:
Airflow REST GET API call for TaskInstances returns unordered items, making paged retrieval impossible
### Apache Airflow version

Other Airflow 2 version (please specify below)

### If "Other Airflow 2 version" selected, which one?

2.9.1, but nothing in the release notes thereafter indicates this issue has been fixed.

### What happened?

When using paged retrieval:

```
https://airflow/api/v1/dags/DAG/dagRuns/rc--2024-08-02_01-30/taskInstances?limit=100&offset=0
https://airflow/api/v1/dags/DAG/dagRuns/rc--2024-08-02_01-30/taskInstances?limit=100&offset=100
https://airflow/api/v1/dags/DAG/dagRuns/rc--2024-08-02_01-30/taskInstances?limit=100&offset=200
```

the task instance results are unordered, ensuring with near certainty that all subsequent paged retrievals (with offset > 0) return some tasks already returned before, and fail to return tasks that should have been in the page.  Thus, concatenating the tasks yields the correct *number* of tasks, but some are repeated and some are missing.

Full retrieval as per

```
https://airflow/api/v1/dags/DAG/dagRuns/rc--2024-08-02_01-30/taskInstances?limit=300
```

returns each task instance only once, and returns all task instances exhaustively.

### What you think should happen instead?

Task instances from API call should either be returned in some specific order, or there should be a query string parameter allowing us users to specify a designated order (such as task instance creation / appearance date).



### How to reproduce

Just issue a taskInstances GET call with a small limit, then with an offset equivalent to the small limit.  Tasks that were on the first batch are likely to appear on the second batch too.

### Operating System

Ubuntu 22.04 container

### Versions of Apache Airflow Providers

Not relevant.

### Deployment

Official Apache Airflow Helm Chart

### Deployment details

We use a very vanilla Kubernetes.

### Anything else?

Always.

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

I cannot, but I can recommend the following: all REST endpoints that list items in a paged manner must return items in a stable sort order, or at least allow the end user to specify a stable sort order criteria and document that unordered items are returned by default.

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)


Discussion:
omkar-foss: Fixed all linting issues and added all your suggestions from comments above. Please review when you get a chance, thanks.

Patch:
diff --git a/airflow/api_connexion/endpoints/task_instance_endpoint.py b/airflow/api_connexion/endpoints/task_instance_endpoint.py
index a9213d2c24..23e5f5a6d1 100644
--- a/airflow/api_connexion/endpoints/task_instance_endpoint.py
+++ b/airflow/api_connexion/endpoints/task_instance_endpoint.py
@@ -62,6 +62,7 @@ from airflow.www.extensions.init_auth_manager import get_auth_manager
 if TYPE_CHECKING:
     from sqlalchemy.orm import Session
     from sqlalchemy.sql import ClauseElement, Select
+    from sqlalchemy.sql.expression import ColumnOperators
 
     from airflow.api_connexion.types import APIResponse
     from airflow.auth.managers.models.batch_apis import IsAuthorizedDagRequest
@@ -245,28 +246,11 @@ def get_mapped_task_instances(
         .options(joinedload(TI.rendered_task_instance_fields))
     )
 
-    if order_by is None:
-        entry_query = entry_query.order_by(TI.map_index.asc())
-    elif order_by == "state":
-        entry_query = entry_query.order_by(TI.state.asc(), TI.map_index.asc())
-    elif order_by == "-state":
-        entry_query = entry_query.order_by(TI.state.desc(), TI.map_index.asc())
-    elif order_by == "duration":
-        entry_query = entry_query.order_by(TI.duration.asc(), TI.map_index.asc())
-    elif order_by == "-duration":
-        entry_query = entry_query.order_by(TI.duration.desc(), TI.map_index.asc())
-    elif order_by == "start_date":
-        entry_query = entry_query.order_by(TI.start_date.asc(), TI.map_index.asc())
-    elif order_by == "-start_date":
-        entry_query = entry_query.order_by(TI.start_date.desc(), TI.map_index.asc())
-    elif order_by == "end_date":
-        entry_query = entry_query.order_by(TI.end_date.asc(), TI.map_index.asc())
-    elif order_by == "-end_date":
-        entry_query = entry_query.order_by(TI.end_date.desc(), TI.map_index.asc())
-    elif order_by == "-map_index":
-        entry_query = entry_query.order_by(TI.map_index.desc())
-    else:
-        raise BadRequest(detail=f"Ordering with '{order_by}' is not supported")
+    try:
+        order_by_params = _get_order_by_params(order_by)
+        entry_query = entry_query.order_by(*order_by_params)
+    except _UnsupportedOrderBy as e:
+        raise BadRequest(detail=f"Ordering with {e.order_by!r} is not supported")
 
     # using execute because we want the SlaMiss entity. Scalars don't return None for missing entities
     task_instances = session.execute(entry_query.offset(offset).limit(limit)).all()
@@ -297,6 +281,39 @@ def _apply_range_filter(query: Select, key: ClauseElement, value_range: tuple[T,
     return query
 
 
+class _UnsupportedOrderBy(ValueError):
+    def __init__(self, order_by: str) -> None:
+        super().__init__(order_by)
+        self.order_by = order_by
+
+
+def _get_order_by_params(order_by: str | None = None) -> tuple[ColumnOperators, ...]:
+    """Return a tuple with the order by params to be used in the query."""
+    if order_by is None:
+        return (TI.map_index.asc(),)
+    if order_by == "state":
+        return (TI.state.asc(), TI.map_index.asc())
+    if order_by == "-state":
+        return (TI.state.desc(), TI.map_index.asc())
+    if order_by == "duration":
+        return (TI.duration.asc(), TI.map_index.asc())
+    if order_by == "-duration":
+        return (TI.duration.desc(), TI.map_index.asc())
+    if order_by == "start_date":
+        return (TI.start_date.asc(), TI.map_index.asc())
+    if order_by == "-start_date":
+        return (TI.start_date.desc(), TI.map_index.asc())
+    if order_by == "end_date":
+        return (TI.end_date.asc(), TI.map_index.asc())
+    if order_by == "-end_date":
+        return (TI.end_date.desc(), TI.map_index.asc())
+    if order_by == "map_index":
+        return (TI.map_index.asc(),)
+    if order_by == "-map_index":
+        return (TI.map_index.desc(),)
+    raise _UnsupportedOrderBy(order_by)
+
+
 @format_parameters(
     {
         "execution_date_gte": format_datetime,
@@ -331,6 +348,7 @@ def get_task_instances(
     queue: list[str] | None = None,
     executor: list[str] | None = None,
     offset: int | None = None,
+    order_by: str | None = None,
     session: Session = NEW_SESSION,
 ) -> APIResponse:
     """Get list of task instances."""
@@ -378,11 +396,16 @@ def get_task_instances(
         )
         .add_columns(SlaMiss)
         .options(joinedload(TI.rendered_task_instance_fields))
-        .offset(offset)
-        .limit(limit)
     )
+
+    try:
+        order_by_params = _get_order_by_params(order_by)
+        entry_query = entry_query.order_by(*order_by_params)
+    except _UnsupportedOrderBy as e:
+        raise BadRequest(detail=f"Ordering with {e.order_by!r} is not supported")
+
     # using execute because we want the SlaMiss entity. Scalars don't return None for missing entities
-    task_instances = session.execute(entry_query).all()
+    task_instances = session.execute(entry_query.offset(offset).limit(limit)).all()
     return task_instance_collection_schema.dump(
         TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
     )
@@ -453,6 +476,13 @@ def get_task_instances_batch(session: Session = NEW_SESSION) -> APIResponse:
     ti_query = base_query.options(
         joinedload(TI.rendered_task_instance_fields), joinedload(TI.task_instance_note)
     )
+
+    try:
+        order_by_params = _get_order_by_params(data["order_by"])
+        ti_query = ti_query.order_by(*order_by_params)
+    except _UnsupportedOrderBy as e:
+        raise BadRequest(detail=f"Ordering with {e.order_by!r} is not supported")
+
     # using execute because we want the SlaMiss entity. Scalars don't return None for missing entities
     task_instances = session.execute(ti_query).all()
 
diff --git a/airflow/api_connexion/openapi/v1.yaml b/airflow/api_connexion/openapi/v1.yaml
index d0f5db43db..a0809f045d 100644
--- a/airflow/api_connexion/openapi/v1.yaml
+++ b/airflow/api_connexion/openapi/v1.yaml
@@ -1517,6 +1517,7 @@ paths:
       parameters:
         - $ref: "#/components/parameters/PageLimit"
         - $ref: "#/components/parameters/PageOffset"
+        - $ref: "#/components/parameters/TaskInstanceOrderBy"
       responses:
         "200":
           description: Success.
@@ -1675,7 +1676,7 @@ paths:
         - $ref: "#/components/parameters/FilterPool"
         - $ref: "#/components/parameters/FilterQueue"
         - $ref: "#/components/parameters/FilterExecutor"
-        - $ref: "#/components/parameters/OrderBy"
+        - $ref: "#/components/parameters/TaskInstanceOrderBy"
       responses:
         "200":
           description: Success.
@@ -5177,6 +5178,16 @@ components:
           items:
             type: string
           description: The value can be repeated to retrieve multiple matching values (OR condition).
+        order_by:
+          type: string
+          description: |
+            The name of the field to order the results by. Prefix a field name
+            with `-` to reverse the sort order. `order_by` defaults to
+            `map_index` when unspecified.
+            Supported field names: `state`, `duration`, `start_date`, `end_date`
+            and `map_index`.
+
+            *New in version 3.0.0*
 
     # Common data type
     ScheduleInterval:
@@ -5802,6 +5813,21 @@ components:
         type: integer
       description: Filter on try_number for task instance.
 
+    TaskInstanceOrderBy:
+      in: query
+      name: order_by
+      schema:
+        type: string
+      required: false
+      description: |
+        The name of the field to order the results by. Prefix a field name
+        with `-` to reverse the sort order. `order_by` defaults to
+        `map_index` when unspecified.
+        Supported field names: `state`, `duration`, `start_date`, `end_date`
+        and `map_index`.
+
+        *New in version 3.0.0*
+
     OrderBy:
       in: query
       name: order_by
diff --git a/airflow/api_connexion/schemas/task_instance_schema.py b/airflow/api_connexion/schemas/task_instance_schema.py
index 74cd0585dc..0c1daf6ce2 100644
--- a/airflow/api_connexion/schemas/task_instance_schema.py
+++ b/airflow/api_connexion/schemas/task_instance_schema.py
@@ -167,6 +167,7 @@ class TaskInstanceBatchFormSchema(Schema):
     pool = fields.List(fields.Str(), load_default=None)
     queue = fields.List(fields.Str(), load_default=None)
     executor = fields.List(fields.Str(), load_default=None)
+    order_by = fields.Str(load_default=None)
 
 
 class ClearTaskInstanceFormSchema(Schema):
diff --git a/airflow/cli/cli_config.py b/airflow/cli/cli_config.py
index 2699165484..43c853a060 100644
--- a/airflow/cli/cli_config.py
+++ b/airflow/cli/cli_config.py
@@ -617,12 +617,6 @@ ARG_IGNORE_DEPENDENCIES = Arg(
     help="Ignore task-specific dependencies, e.g. upstream, depends_on_past, and retry delay dependencies",
     action="store_true",
 )
-ARG_IGNORE_DEPENDS_ON_PAST = Arg(
-    ("-I", "--ignore-depends-on-past"),
-    help="Deprecated -- use `--depends-on-past ignore` instead. "
-    "Ignore depends_on_past dependencies (but respect upstream dependencies)",
-    action="store_true",
-)
 ARG_DEPENDS_ON_PAST = Arg(
     ("-d", "--depends-on-past"),
     help="Determine how Airflow should deal with past dependencies. The default action is `check`, Airflow "
@@ -1382,7 +1376,6 @@ TASKS_COMMANDS = (
             ARG_RAW,
             ARG_IGNORE_ALL_DEPENDENCIES,
             ARG_IGNORE_DEPENDENCIES,
-            ARG_IGNORE_DEPENDS_ON_PAST,
             ARG_DEPENDS_ON_PAST,
             ARG_SHIP_DAG,
             ARG_PICKLE,
diff --git a/airflow/cli/commands/task_command.py b/airflow/cli/commands/task_command.py
index c4c6db386a..ffb08ff66d 100644
--- a/airflow/cli/commands/task_command.py
+++ b/airflow/cli/commands/task_command.py
@@ -330,7 +330,6 @@ def _run_task_by_local_task_job(args, ti: TaskInstance | TaskInstancePydantic) -
 
 RAW_TASK_UNSUPPORTED_OPTION = [
     "ignore_all_dependencies",
-    "ignore_depends_on_past",
     "ignore_dependencies",
     "force",
 ]
diff --git a/airflow/utils/cli.py b/airflow/utils/cli.py
index 00e6e15ba3..e8b9e6f7d2 100644
--- a/airflow/utils/cli.py
+++ b/airflow/utils/cli.py
@@ -36,7 +36,7 @@ from sqlalchemy import select
 
 from airflow import settings
 from airflow.api_internal.internal_api_call import InternalApiConfig
-from airflow.exceptions import AirflowException, RemovedInAirflow3Warning
+from airflow.exceptions import AirflowException
 from airflow.utils import cli_action_loggers, timezone
 from airflow.utils.log.non_caching_file_handler import NonCachingFileHandler
 from airflow.utils.platform import getuser, is_terminal_support_colors
@@ -352,14 +352,6 @@ def should_use_colors(args) -> bool:
 
 
 def should_ignore_depends_on_past(args) -> bool:
-    if args.ignore_depends_on_past:
-        warnings.warn(
-            "Using `--ignore-depends-on-past` is Deprecated."
-            "Please use `--depends-on-past ignore` instead.",
-            RemovedInAirflow3Warning,
-            stacklevel=2,
-        )
-        return True
     return args.depends_on_past == "ignore"
 
 
diff --git a/airflow/www/static/js/types/api-generated.ts b/airflow/www/static/js/types/api-generated.ts
index d5c1e06b6e..1897fd90ce 100644
--- a/airflow/www/static/js/types/api-generated.ts
+++ b/airflow/www/static/js/types/api-generated.ts
@@ -2304,6 +2304,16 @@ export interface components {
       queue?: string[];
       /** @description The value can be repeated to retrieve multiple matching values (OR condition). */
       executor?: string[];
+      /**
+       * @description The name of the field to order the results by. Prefix a field name
+       * with `-` to reverse the sort order. `order_by` defaults to
+       * `map_index` when unspecified.
+       * Supported field names: `state`, `duration`, `start_date`, `end_date`
+       * and `map_index`.
+       *
+       * *New in version 3.0.0*
+       */
+      order_by?: string;
     };
     /**
      * @description Schedule interval. Defines how often DAG runs, this object gets added to your latest task instance's
@@ -2654,6 +2664,16 @@ export interface components {
     FilterMapIndex: number;
     /** @description Filter on try_number for task instance. */
     FilterTryNumber: number;
+    /**
+     * @description The name of the field to order the results by. Prefix a field name
+     * with `-` to reverse the sort order. `order_by` defaults to
+     * `map_index` when unspecified.
+     * Supported field names: `state`, `duration`, `start_date`, `end_date`
+     * and `map_index`.
+     *
+     * *New in version 3.0.0*
+     */
+    TaskInstanceOrderBy: string;
     /**
      * @description The name of the field to order the results by.
      * Prefix a field name with `-` to reverse the sort order.
@@ -4049,6 +4069,16 @@ export interface operations {
         limit?: components["parameters"]["PageLimit"];
         /** The number of items to skip before starting to collect the result set. */
         offset?: components["parameters"]["PageOffset"];
+        /**
+         * The name of the field to order the results by. Prefix a field name
+         * with `-` to reverse the sort order. `order_by` defaults to
+         * `map_index` when unspecified.
+         * Supported field names: `state`, `duration`, `start_date`, `end_date`
+         * and `map_index`.
+         *
+         * *New in version 3.0.0*
+         */
+        order_by?: components["parameters"]["TaskInstanceOrderBy"];
       };
     };
     responses: {
@@ -4276,12 +4306,15 @@ export interface operations {
         /** The value can be repeated to retrieve multiple matching values (OR condition). */
         executor?: components["parameters"]["FilterExecutor"];
         /**
-         * The name of the field to order the results by.
-         * Prefix a field name with `-` to reverse the sort order.
+         * The name of the field to order the results by. Prefix a field name
+         * with `-` to reverse the sort order. `order_by` defaults to
+         * `map_index` when unspecified.
+         * Supported field names: `state`, `duration`, `start_date`, `end_date`
+         * and `map_index`.
          *
-         * *New in version 2.1.0*
+         * *New in version 3.0.0*
          */
-        order_by?: components["parameters"]["OrderBy"];
+        order_by?: components["parameters"]["TaskInstanceOrderBy"];
       };
     };
     responses: {
diff --git a/newsfragments/41635.significant.rst b/newsfragments/41635.significant.rst
new file mode 100644
index 0000000000..0566d60747
--- /dev/null
+++ b/newsfragments/41635.significant.rst
@@ -0,0 +1,1 @@
+Removed deprecated ``--ignore-depends-on-past``  cli option from task command. Please use ``--depends-on-past ignore``.


Test patch:
diff --git a/tests/api_connexion/endpoints/test_task_instance_endpoint.py b/tests/api_connexion/endpoints/test_task_instance_endpoint.py
index 4fcd66affe..25ded6c814 100644
--- a/tests/api_connexion/endpoints/test_task_instance_endpoint.py
+++ b/tests/api_connexion/endpoints/test_task_instance_endpoint.py
@@ -782,6 +782,80 @@ class TestGetTaskInstances(TestTaskInstanceEndpoint):
         assert count == response.json["total_entries"]
         assert count == len(response.json["task_instances"])
 
+    def test_should_respond_200_for_order_by(self, session):
+        dag_id = "example_python_operator"
+        self.create_task_instances(
+            session,
+            task_instances=[
+                {"start_date": DEFAULT_DATETIME_1 + dt.timedelta(minutes=(i + 1))} for i in range(10)
+            ],
+            dag_id=dag_id,
+        )
+
+        ti_count = session.query(TaskInstance).filter(TaskInstance.dag_id == dag_id).count()
+
+        # Ascending order
+        response_asc = self.client.get(
+            "/api/v1/dags/~/dagRuns/~/taskInstances?order_by=start_date",
+            environ_overrides={"REMOTE_USER": "test"},
+        )
+        assert response_asc.status_code == 200
+        assert response_asc.json["total_entries"] == ti_count
+        assert len(response_asc.json["task_instances"]) == ti_count
+
+        # Descending order
+        response_desc = self.client.get(
+            "/api/v1/dags/~/dagRuns/~/taskInstances?order_by=-start_date",
+            environ_overrides={"REMOTE_USER": "test"},
+        )
+        assert response_desc.status_code == 200
+        assert response_desc.json["total_entries"] == ti_count
+        assert len(response_desc.json["task_instances"]) == ti_count
+
+        # Compare
+        start_dates_asc = [ti["start_date"] for ti in response_asc.json["task_instances"]]
+        assert len(start_dates_asc) == ti_count
+        start_dates_desc = [ti["start_date"] for ti in response_desc.json["task_instances"]]
+        assert len(start_dates_desc) == ti_count
+        assert start_dates_asc == list(reversed(start_dates_desc))
+
+    def test_should_respond_200_for_pagination(self, session):
+        dag_id = "example_python_operator"
+        self.create_task_instances(
+            session,
+            task_instances=[
+                {"start_date": DEFAULT_DATETIME_1 + dt.timedelta(minutes=(i + 1))} for i in range(10)
+            ],
+            dag_id=dag_id,
+        )
+
+        # First 5 items
+        response_batch1 = self.client.get(
+            "/api/v1/dags/~/dagRuns/~/taskInstances?limit=5&offset=0",
+            environ_overrides={"REMOTE_USER": "test"},
+        )
+        assert response_batch1.status_code == 200, response_batch1.json
+        num_entries_batch1 = len(response_batch1.json["task_instances"])
+        assert num_entries_batch1 == 5
+        assert len(response_batch1.json["task_instances"]) == 5
+
+        # 5 items after that
+        response_batch2 = self.client.get(
+            "/api/v1/dags/~/dagRuns/~/taskInstances?limit=5&offset=5",
+            environ_overrides={"REMOTE_USER": "test"},
+            json={"limit": 5, "offset": 0, "dag_ids": [dag_id]},
+        )
+        assert response_batch2.status_code == 200, response_batch2.json
+        num_entries_batch2 = len(response_batch2.json["task_instances"])
+        assert num_entries_batch2 > 0
+        assert len(response_batch2.json["task_instances"]) > 0
+
+        # Match
+        ti_count = session.query(TaskInstance).filter(TaskInstance.dag_id == dag_id).count()
+        assert response_batch1.json["total_entries"] == response_batch2.json["total_entries"] == ti_count
+        assert (num_entries_batch1 + num_entries_batch2) == ti_count
+        assert response_batch1 != response_batch2
+
     def test_should_raises_401_unauthenticated(self):
         response = self.client.get(
             "/api/v1/dags/example_python_operator/dagRuns/~/taskInstances",
@@ -971,6 +1045,45 @@ class TestGetTaskInstancesBatch(TestTaskInstanceEndpoint):
         assert expected_ti_count == response.json["total_entries"]
         assert expected_ti_count == len(response.json["task_instances"])
 
+    def test_should_respond_200_for_order_by(self, session):
+        dag_id = "example_python_operator"
+        self.create_task_instances(
+            session,
+            task_instances=[
+                {"start_date": DEFAULT_DATETIME_1 + dt.timedelta(minutes=(i + 1))} for i in range(10)
+            ],
+            dag_id=dag_id,
+        )
+
+        ti_count = session.query(TaskInstance).filter(TaskInstance.dag_id == dag_id).count()
+
+        # Ascending order
+        response_asc = self.client.post(
+            "/api/v1/dags/~/dagRuns/~/taskInstances/list",
+            environ_overrides={"REMOTE_USER": "test"},
+            json={"order_by": "start_date", "dag_ids": [dag_id]},
+        )
+        assert response_asc.status_code == 200, response_asc.json
+        assert response_asc.json["total_entries"] == ti_count
+        assert len(response_asc.json["task_instances"]) == ti_count
+
+        # Descending order
+        response_desc = self.client.post(
+            "/api/v1/dags/~/dagRuns/~/taskInstances/list",
+            environ_overrides={"REMOTE_USER": "test"},
+            json={"order_by": "-start_date", "dag_ids": [dag_id]},
+        )
+        assert response_desc.status_code == 200, response_desc.json
+        assert response_desc.json["total_entries"] == ti_count
+        assert len(response_desc.json["task_instances"]) == ti_count
+
+        # Compare
+        start_dates_asc = [ti["start_date"] for ti in response_asc.json["task_instances"]]
+        assert len(start_dates_asc) == ti_count
+        start_dates_desc = [ti["start_date"] for ti in response_desc.json["task_instances"]]
+        assert len(start_dates_desc) == ti_count
+        assert start_dates_asc == list(reversed(start_dates_desc))
+
     @pytest.mark.parametrize(
         "task_instances, payload, expected_ti_count",
         [
diff --git a/tests/cli/commands/test_task_command.py b/tests/cli/commands/test_task_command.py
index 869d5a5a22..4ad96942c0 100644
--- a/tests/cli/commands/test_task_command.py
+++ b/tests/cli/commands/test_task_command.py
@@ -415,7 +415,6 @@ class TestCliTasks:
         "option",
         [
             "--ignore-all-dependencies",
-            "--ignore-depends-on-past",
             "--ignore-dependencies",
             "--force",
         ],