This is an automated email from the ASF dual-hosted git repository.
ephraimbuddy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new e24f5294645 Add indexes on serialized_dag and dag_code (#69391)
e24f5294645 is described below
commit e24f5294645c20f55069c2f2a1b890eb3ff8cc2a
Author: Justin Pakzad <[email protected]>
AuthorDate: Fri Aug 7 04:14:57 2026 -0400
Add indexes on serialized_dag and dag_code (#69391)
---
airflow-core/docs/migrations-ref.rst | 4 +-
...3_4_0_add_indexes_on_serialized_dag_and_dag_.py | 55 ++++++++++++++++++++++
airflow-core/src/airflow/models/dagcode.py | 3 +-
airflow-core/src/airflow/models/serialized_dag.py | 3 +-
airflow-core/src/airflow/utils/db.py | 2 +-
...9_add_indexes_on_serialized_dag_and_dag_code.py | 55 ++++++++++++++++++++++
6 files changed, 118 insertions(+), 4 deletions(-)
diff --git a/airflow-core/docs/migrations-ref.rst
b/airflow-core/docs/migrations-ref.rst
index e3b33cab185..691720f6508 100644
--- a/airflow-core/docs/migrations-ref.rst
+++ b/airflow-core/docs/migrations-ref.rst
@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are
executed via when you ru
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| Revision ID | Revises ID | Airflow Version | Description
|
+=========================+==================+===================+==============================================================+
-| ``b2f1a9c7d4e0`` (head) | ``7a98f1b7dbd3`` | ``3.4.0`` | Reference
the asset event from asset_dag_run_queue (consume- |
+| ``3c525f44bea8`` (head) | ``b2f1a9c7d4e0`` | ``3.4.0`` | Add indexes
on serialized_dag and dag_code. |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``b2f1a9c7d4e0`` | ``7a98f1b7dbd3`` | ``3.4.0`` | Reference
the asset event from asset_dag_run_queue (consume- |
| | | |
by-reference). |
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| ``7a98f1b7dbd3`` | ``c4e7a1f9b2d0`` | ``3.4.0`` | Add index
on asset_event (asset_id, partition_key). |
diff --git
a/airflow-core/src/airflow/migrations/versions/0129_3_4_0_add_indexes_on_serialized_dag_and_dag_.py
b/airflow-core/src/airflow/migrations/versions/0129_3_4_0_add_indexes_on_serialized_dag_and_dag_.py
new file mode 100644
index 00000000000..5adb80033dd
--- /dev/null
+++
b/airflow-core/src/airflow/migrations/versions/0129_3_4_0_add_indexes_on_serialized_dag_and_dag_.py
@@ -0,0 +1,55 @@
+#
+# 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 indexes on serialized_dag and dag_code.
+
+Revision ID: 3c525f44bea8
+Revises: b2f1a9c7d4e0
+Create Date: 2026-07-04 18:18:11.140047
+
+"""
+
+from __future__ import annotations
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "3c525f44bea8"
+down_revision = "b2f1a9c7d4e0"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+
+def upgrade():
+ """Apply Add indexes on serialized_dag and dag_code."""
+ with op.batch_alter_table("dag_code", schema=None) as batch_op:
+ batch_op.create_index("idx_dag_code_dag_id_last_updated", ["dag_id",
"last_updated"], unique=False)
+
+ with op.batch_alter_table("serialized_dag", schema=None) as batch_op:
+ batch_op.create_index("idx_serialized_dag_dag_id_created_at",
["dag_id", "created_at"], unique=False)
+
+
+def downgrade():
+ """Unapply Add indexes on serialized_dag and dag_code."""
+ with op.batch_alter_table("serialized_dag", schema=None) as batch_op:
+ batch_op.drop_index("idx_serialized_dag_dag_id_created_at")
+
+ with op.batch_alter_table("dag_code", schema=None) as batch_op:
+ batch_op.drop_index("idx_dag_code_dag_id_last_updated")
diff --git a/airflow-core/src/airflow/models/dagcode.py
b/airflow-core/src/airflow/models/dagcode.py
index d591ddb25b7..c693f0384ce 100644
--- a/airflow-core/src/airflow/models/dagcode.py
+++ b/airflow-core/src/airflow/models/dagcode.py
@@ -23,7 +23,7 @@ from uuid import UUID
import sqlalchemy as sa
import uuid6
-from sqlalchemy import ForeignKey, String, Text, select
+from sqlalchemy import ForeignKey, Index, String, Text, select
from sqlalchemy.dialects.mysql import MEDIUMTEXT
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.sql.expression import literal
@@ -70,6 +70,7 @@ class DagCode(Base):
sa.Uuid(), ForeignKey("dag_version.id", ondelete="CASCADE"),
nullable=False, unique=True
)
dag_version = relationship("DagVersion", back_populates="dag_code",
uselist=False)
+ __table_args__ = (Index("idx_dag_code_dag_id_last_updated", dag_id,
last_updated),)
def __init__(self, dag_version, full_filepath: str, source_code: str |
None = None):
self.dag_version = dag_version
diff --git a/airflow-core/src/airflow/models/serialized_dag.py
b/airflow-core/src/airflow/models/serialized_dag.py
index f3f5b783096..fa7d80335a3 100644
--- a/airflow-core/src/airflow/models/serialized_dag.py
+++ b/airflow-core/src/airflow/models/serialized_dag.py
@@ -27,7 +27,7 @@ from typing import TYPE_CHECKING, Any, Literal, NamedTuple
from uuid import UUID
import uuid6
-from sqlalchemy import JSON, ForeignKey, LargeBinary, String, Uuid, exists,
select, tuple_, update
+from sqlalchemy import JSON, ForeignKey, Index, LargeBinary, String, Uuid,
exists, select, tuple_, update
from sqlalchemy.dialects.postgresql import ARRAY, JSONB
from sqlalchemy.orm import Mapped, backref, foreign, mapped_column,
relationship
from sqlalchemy.sql.expression import func, literal
@@ -344,6 +344,7 @@ class SerializedDagModel(Base):
)
load_op_links = True
+ __table_args__ = (Index("idx_serialized_dag_dag_id_created_at", dag_id,
created_at),)
def __init__(self, dag: LazyDeserializedDAG) -> None:
self.dag_id = dag.dag_id
diff --git a/airflow-core/src/airflow/utils/db.py
b/airflow-core/src/airflow/utils/db.py
index 21ccd5ea3bc..7dcb15c7423 100644
--- a/airflow-core/src/airflow/utils/db.py
+++ b/airflow-core/src/airflow/utils/db.py
@@ -117,7 +117,7 @@ _REVISION_HEADS_MAP: dict[str, str] = {
"3.1.8": "509b94a1042d",
"3.2.0": "1d6611b6ab7c",
"3.3.0": "d2f4e1b3c5a7",
- "3.4.0": "b2f1a9c7d4e0",
+ "3.4.0": "3c525f44bea8",
}
# Prefix used to identify tables holding data moved during migration.
diff --git
a/airflow-core/tests/unit/migrations/test_0129_add_indexes_on_serialized_dag_and_dag_code.py
b/airflow-core/tests/unit/migrations/test_0129_add_indexes_on_serialized_dag_and_dag_code.py
new file mode 100644
index 00000000000..4774a7711dd
--- /dev/null
+++
b/airflow-core/tests/unit/migrations/test_0129_add_indexes_on_serialized_dag_and_dag_code.py
@@ -0,0 +1,55 @@
+#
+# 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.
+from __future__ import annotations
+
+import pytest
+import sqlalchemy as sa
+
+from airflow import settings
+from airflow.utils.db import downgrade, upgradedb
+
+pytestmark = pytest.mark.db_test
+
+_REVISION = "3c525f44bea8"
+_DOWN_REVISION = "b2f1a9c7d4e0"
+
+_EXPECTED_INDEXES = {
+ "idx_serialized_dag_dag_id_created_at": ("serialized_dag", ["dag_id",
"created_at"]),
+ "idx_dag_code_dag_id_last_updated": ("dag_code", ["dag_id",
"last_updated"]),
+}
+
+
+class TestMigration0124AddDagIdIndexes:
+ @pytest.fixture(autouse=True)
+ def _restore_head(self):
+ yield
+ upgradedb()
+
+ @staticmethod
+ def _get_indexes(table):
+ with settings.engine.connect() as conn:
+ return {ix["name"]: ix["column_names"] for ix in
sa.inspect(conn).get_indexes(table)}
+
+ def test_upgrade_creates_indexes_and_downgrade_drops_them(self):
+ downgrade(to_revision=_DOWN_REVISION)
+ for index, (table, _) in _EXPECTED_INDEXES.items():
+ assert index not in self._get_indexes(table)
+
+ upgradedb(to_revision=_REVISION)
+ for index, (table, columns) in _EXPECTED_INDEXES.items():
+ assert self._get_indexes(table).get(index) == columns
