diff --git a/airflow/migrations/versions/0127_2_7_0_add_index_on_last_scheduling_decision_.py b/airflow/migrations/versions/0127_2_7_0_add_index_on_last_scheduling_decision_.py new file mode 100644 index 0000000000000..785fa43258ba3 --- /dev/null +++ b/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") diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py index 42845b34bdd0e..dba6fbf748376 100644 --- a/airflow/models/dagrun.py +++ b/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", diff --git a/docs/apache-airflow/img/airflow_erd.sha256 b/docs/apache-airflow/img/airflow_erd.sha256 index 2405a400dbc83..9cb0d7f0da47d 100644 --- a/docs/apache-airflow/img/airflow_erd.sha256 +++ b/docs/apache-airflow/img/airflow_erd.sha256 @@ -1 +1 @@ -2d0924c9f5c471214953113e8830b842fc45e9344ff6d67b46267cac99e2cdef \ No newline at end of file +a79d2ad704b4b2e88a834676ba0eec16175db945acad03404c9b73311636b464 \ No newline at end of file diff --git a/docs/apache-airflow/img/airflow_erd.svg b/docs/apache-airflow/img/airflow_erd.svg index 142c05dfd84b4..8439c226f8b98 100644 --- a/docs/apache-airflow/img/airflow_erd.svg +++ b/docs/apache-airflow/img/airflow_erd.svg @@ -1225,28 +1225,28 @@ task_instance--xcom -0..N +1 1 task_instance--xcom -1 +0..N 1 task_instance--xcom -1 +0..N 1 task_instance--xcom -0..N +1 1 diff --git a/docs/apache-airflow/migrations-ref.rst b/docs/apache-airflow/migrations-ref.rst index 7b09a46f8c02c..2bcd9929e4691 100644 --- a/docs/apache-airflow/migrations-ref.rst +++ b/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 | +---------------------------------+-------------------+-------------------+--------------------------------------------------------------+