Save scheduler execution time by adding new Index idea for dag_run SECOND TRY

airflow/migrations/versions/0127_2_7_0_add_index_on_last_scheduling_decision_.py

@@ -0,0 +1,56 @@
+#
+# 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.
+
+"""Add index on last_scheduling_decision NULLS FIRST, execution_date, state for queued dagrun
+
+Revision ID: b1c2bca6c666
+Revises: 937cbd173ca1
+Create Date: 2023-05-25 13:01:27.446971
+
+"""
+from __future__ import annotations
+
+from alembic import op
+from sqlalchemy import text
+
+# revision identifiers, used by Alembic.
+revision = "b1c2bca6c666"
+down_revision = "937cbd173ca1"
+branch_labels = None
+depends_on = None
+airflow_version = "2.7.0"
+
+
+def upgrade():
+    """Apply Add index on last_scheduling_decision NULLS FIRST, execution_date, state for queued dagrun"""
+    conn = op.get_bind()
+    if conn.dialect.name == "postgresql":
+        with op.batch_alter_table("dag_run") as batch_op:
+            batch_op.create_index(
+                "idx_last_scheduling_decision_queued",
+                [text("last_scheduling_decision NULLS FIRST"), "execution_date", "state"],
+                postgresql_where=text("state='queued'"),
+            )
+
+
+def downgrade():
+    """Unapply Add index on last_scheduling_decision NULLS FIRST, execution_date, state for queued dagrun"""
+    conn = op.get_bind()
+    if conn.dialect.name == "postgresql":
+        with op.batch_alter_table("dag_run") as batch_op:
+            batch_op.drop_index("idx_last_scheduling_decision_queued")

airflow/models/dagrun.py

@@ -144,6 +144,15 @@ class DagRun(Base, LoggingMixin):
         UniqueConstraint("dag_id", "execution_date", name="dag_run_dag_id_execution_date_key"),
         UniqueConstraint("dag_id", "run_id", name="dag_run_dag_id_run_id_key"),
         Index("idx_last_scheduling_decision", last_scheduling_decision),
+        Index(
+            "idx_last_scheduling_decision_queued",
+            # Not possible to add .nulls_first(), because only postgresql can handle Index like that.
+            # Migration script which contains postgres dialect check adds NULLS FIST to index.
+            last_scheduling_decision,
+            execution_date,
+            _state,
+            postgresql_where=text("state='queued'"),
+        ),
         Index("idx_dag_run_dag_id", dag_id),
         Index(
             "idx_dag_run_running_dags",

docs/apache-airflow/img/airflow_erd.sha256

@@ -1 +1 @@
-2d0924c9f5c471214953113e8830b842fc45e9344ff6d67b46267cac99e2cdef
+a79d2ad704b4b2e88a834676ba0eec16175db945acad03404c9b73311636b464

docs/apache-airflow/migrations-ref.rst

@@ -39,7 +39,10 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
 | Revision ID                     | Revises ID        | Airflow Version   | Description                                                  |
 +=================================+===================+===================+==============================================================+
-| ``937cbd173ca1`` (head)         | ``c804e5c76e3e``  | ``2.7.0``         | Add index to task_instance table                             |
+| ``b1c2bca6c666`` (head)         | ``937cbd173ca1``  | ``2.7.0``         | Add index on last_scheduling_decision NULLS FIRST,           |
+|                                 |                   |                   | execution_date, state for queued dagrun                      |
++---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
+| ``937cbd173ca1``                | ``c804e5c76e3e``  | ``2.7.0``         | Add index to task_instance table                             |
 +---------------------------------+-------------------+-------------------+--------------------------------------------------------------+
 | ``c804e5c76e3e``                | ``98ae134e6fff``  | ``2.6.2``         | Add ``onupdate`` cascade to ``task_map`` table               |
 +---------------------------------+-------------------+-------------------+--------------------------------------------------------------+