Drop the LogTemplate DB model

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                                                  |
 +=========================+==================+===================+==============================================================+
-| ``436dc127462c`` (head) | ``5a5d3253e946`` | ``3.4.0``         | Drop span_status column.                                     |
+| ``4f6723e37686`` (head) | ``436dc127462c`` | ``3.4.0``         | Drop log_template table.                                     |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``436dc127462c``        | ``5a5d3253e946`` | ``3.4.0``         | Drop span_status column.                                     |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``5a5d3253e946``        | ``d2f4e1b3c5a7`` | ``3.4.0``         | Add GIN index on asset_event.extra for PostgreSQL.           |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

airflow-core/newsfragments/69520.significant.rst

@@ -0,0 +1,54 @@
+Drop the ``LogTemplate`` DB model
+
+Airflow tracked historical values of ``[logging] log_filename_template`` and
+``[elasticsearch] log_id_template`` in a ``log_template`` table so a ``DagRun``'s
+logs stayed retrievable after either config value changed. This model has been
+removed: the Elasticsearch/OpenSearch task handlers reached into the metadata
+database directly to read it, which conflicts with the Airflow 3 execution
+model where workers/handlers should not query the metadata DB directly, and it
+otherwise only existed to duplicate two strings already available from config.
+
+**What changed:**
+
+- The ``LogTemplate`` model, the ``log_template`` table, and ``DagRun.log_template_id``
+  are gone.
+- ``DagRun.get_log_template()`` has been removed.
+- Log filename and log ID templates are now read directly from
+  ``[logging] log_filename_template`` and ``[elasticsearch] log_id_template`` at
+  render time, instead of being pinned to the value in effect when the ``DagRun``
+  was created.
+
+**Behaviour changes:**
+
+- Changing ``log_filename_template`` or ``log_id_template`` now applies
+  immediately and retroactively to *all* ``DagRun``s, including ones created
+  before the change. There is no more per-run historical pinning: if you change
+  either setting, logs written under the previous template are no longer
+  reachable through the new one.
+
+**Migration:**
+
+- A database migration drops the ``log_template`` table and the
+  ``dag_run.log_template_id`` column. No user action is required.
+- Any custom log handler or plugin calling ``DagRun.get_log_template()`` or
+  importing ``airflow.models.tasklog.LogTemplate`` must be updated to read
+  ``[logging] log_filename_template`` / ``[elasticsearch] log_id_template``
+  from config instead.
+
+* Types of change
+
+  * [ ] Dag changes
+  * [x] Config changes
+  * [ ] API changes
+  * [ ] CLI changes
+  * [x] Behaviour changes
+  * [ ] Plugin changes
+  * [ ] Dependency changes
+  * [x] Code interface changes
+
+* Migration rules needed
+
+  * Remove any usage of ``DagRun.get_log_template()`` or
+    ``airflow.models.tasklog.LogTemplate`` in custom log handlers/plugins;
+    read ``[logging] log_filename_template`` / ``[elasticsearch] log_id_template``
+    from config instead.

airflow-core/src/airflow/migrations/versions/0126_3_4_0_drop_log_template.py

@@ -0,0 +1,70 @@
+#
+# 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.
+"""
+Drop log_template table.
+
+Revision ID: 4f6723e37686
+Revises: 436dc127462c
+Create Date: 2026-07-05 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+from airflow.migrations.utils import disable_sqlite_fkeys
+from airflow.utils.sqlalchemy import UtcDateTime
+
+# revision identifiers, used by Alembic.
+revision = "4f6723e37686"
+down_revision = "436dc127462c"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+
+def upgrade():
+    """Apply Drop log_template table."""
+    with disable_sqlite_fkeys(op):
+        with op.batch_alter_table("dag_run", schema=None) as batch_op:
+            batch_op.drop_constraint(batch_op.f("task_instance_log_template_id_fkey"), type_="foreignkey")
+            batch_op.drop_column("log_template_id")
+        op.drop_table("log_template")
+
+
+def downgrade():
+    """Unapply Drop log_template table."""
+    with disable_sqlite_fkeys(op):
+        op.create_table(
+            "log_template",
+            sa.Column("id", sa.Integer(), nullable=False, autoincrement=True),
+            sa.Column("filename", sa.Text(), nullable=False),
+            sa.Column("elasticsearch_id", sa.Text(), nullable=False),
+            sa.Column("created_at", UtcDateTime(timezone=True), nullable=False),
+            sa.PrimaryKeyConstraint("id", name="log_template_pkey"),
+        )
+        with op.batch_alter_table("dag_run", schema=None) as batch_op:
+            batch_op.add_column(sa.Column("log_template_id", sa.Integer(), nullable=True))
+            batch_op.create_foreign_key(
+                batch_op.f("task_instance_log_template_id_fkey"),
+                "log_template",
+                ["log_template_id"],
+                ["id"],
+                ondelete="NO ACTION",
+            )

airflow-core/src/airflow/models/__init__.py

@@ -75,7 +75,6 @@ def import_all_models():
     import airflow.models.serialized_dag
     import airflow.models.task_state_store
     import airflow.models.taskinstancehistory
-    import airflow.models.tasklog
     import airflow.models.team
     import airflow.models.xcom
 

airflow-core/src/airflow/models/dagrun.py

@@ -80,7 +80,6 @@
 from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel
 from airflow.models.taskinstance import TaskInstance as TI, clear_task_instances
 from airflow.models.taskinstancehistory import TaskInstanceHistory as TIH
-from airflow.models.tasklog import LogTemplate
 from airflow.models.taskmap import TaskMap
 from airflow.serialization.definitions.deadline import SerializedReferenceModels
 from airflow.serialization.definitions.notset import NOTSET, ArgNotSet, is_arg_set
@@ -231,15 +230,6 @@ class DagRun(Base, LoggingMixin):
     run_after: Mapped[datetime] = mapped_column(UtcDateTime, default=_default_run_after, nullable=False)
     # When a scheduler last attempted to schedule TIs for this DagRun
     last_scheduling_decision: Mapped[datetime | None] = mapped_column(UtcDateTime, nullable=True)
-    # Foreign key to LogTemplate. DagRun rows created prior to this column's
-    # existence have this set to NULL. Later rows automatically populate this on
-    # insert to point to the latest LogTemplate entry.
-    log_template_id: Mapped[int | None] = mapped_column(
-        Integer,
-        ForeignKey("log_template.id", name="task_instance_log_template_id_fkey", ondelete="NO ACTION"),
-        default=select(func.max(LogTemplate.__table__.c.id)),
-        nullable=True,
-    )
     # This is nullable because it's too costly to migrate dagruns created prior
     # to this column's addition (Airflow 3.2.0). If you want a reasonable
     # meaningful non-null value, use ``dr.created_at or dr.run_after``.
@@ -2217,25 +2207,6 @@ def schedule_tis(
 
         return count
 
-    @provide_session
-    def get_log_template(self, *, session: Session = NEW_SESSION) -> LogTemplate:
-        return DagRun._get_log_template(log_template_id=self.log_template_id, session=session)
-
-    @staticmethod
-    @provide_session
-    def _get_log_template(log_template_id: int | None, *, session: Session = NEW_SESSION) -> LogTemplate:
-        template: LogTemplate | None
-        if log_template_id is None:  # DagRun created before LogTemplate introduction.
-            template = session.scalar(select(LogTemplate).order_by(LogTemplate.id).limit(1))
-        else:
-            template = session.get(LogTemplate, log_template_id)
-        if template is None:
-            raise AirflowException(
-                f"No log_template entry found for ID {log_template_id!r}. "
-                f"Please make sure you set up the metadatabase correctly."
-            )
-        return template
-
     @staticmethod
     def _get_partial_task_ids(dag: SerializedDAG | None) -> list[str] | None:
         return dag.task_ids if dag and dag.partial else None

airflow-core/src/airflow/serialization/definitions/dag.py

@@ -48,7 +48,6 @@
 from airflow.models.deadline import Deadline
 from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel
 from airflow.models.taskinstancekey import TaskInstanceKey
-from airflow.models.tasklog import LogTemplate
 from airflow.sdk.definitions.deadline import VariableInterval
 from airflow.serialization.decoders import decode_deadline_alert
 from airflow.serialization.definitions.deadline import DeadlineAlertFields, SerializedReferenceModels
@@ -1425,9 +1424,7 @@ def _create_orm_dagrun(
         partition_date=partition_date,
         note=note,
     )
-    # Load defaults into the following two fields to ensure result can be serialized detached
-    max_log_template_id = session.scalar(select(func.max(LogTemplate.__table__.c.id)))
-    run.log_template_id = int(max_log_template_id) if max_log_template_id is not None else 0
+    # Load default into the following field to ensure result can be serialized detached
     run.created_dag_version = dag_version
     run.consumed_asset_events = []
     session.add(run)

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

@@ -105,11 +105,10 @@ def wrapper(*args, **kwargs):
                 # Check and run migrations if necessary
                 if check_db:
                     from airflow.configuration import conf
-                    from airflow.utils.db import check_and_run_migrations, synchronize_log_template
+                    from airflow.utils.db import check_and_run_migrations
 
                     if conf.getboolean("database", "check_migrations"):
                         check_and_run_migrations()
-                    synchronize_log_template()
                     # Set server context so validation logic (e.g. DagBundlesManager) runs correctly
                     original_ctx = os.environ.pop("_AIRFLOW_PROCESS_CONTEXT", None)
                     try:

airflow-core/src/airflow/utils/db.py

@@ -48,7 +48,6 @@
     func,
     inspect,
     literal,
-    or_,
     select,
     text,
 )
@@ -117,7 +116,7 @@ class MappedClassProtocol(Protocol):
     "3.1.8": "509b94a1042d",
     "3.2.0": "1d6611b6ab7c",
     "3.3.0": "d2f4e1b3c5a7",
-    "3.4.0": "436dc127462c",
+    "3.4.0": "4f6723e37686",
 }
 
 # Prefix used to identify tables holding data moved during migration.
@@ -864,9 +863,8 @@ def initdb(*, session: Session = NEW_SESSION, use_migration_files: bool = False)
             _create_db_from_orm(session=session)
 
     external_db_manager.initdb(session, use_migration_files=use_migration_files)
-    # Add default pool & sync log_template
+    # Add default pool
     add_default_pool_if_not_exists(session=session)
-    synchronize_log_template(session=session)
 
 
 def _get_alembic_config():
@@ -1004,82 +1002,6 @@ def check_and_run_migrations():
         sys.exit(1)
 
 
-@provide_session
-def synchronize_log_template(*, session: Session = NEW_SESSION) -> None:
-    """
-    Synchronize log template configs with table.
-
-    This checks if the last row fully matches the current config values, and
-    insert a new row if not.
-    """
-    # NOTE: SELECT queries in this function are INTENTIONALLY written with the
-    # SQL builder style, not the ORM query API. This avoids configuring the ORM
-    # unless we need to insert something, speeding up CLI in general.
-
-    from airflow.models.tasklog import LogTemplate
-
-    metadata = reflect_tables([LogTemplate], session)
-    log_template_table: Table | None = metadata.tables.get(LogTemplate.__tablename__)
-
-    if log_template_table is None:
-        log.info("Log template table does not exist (added in 2.3.0); skipping log template sync.")
-        return
-
-    filename = conf.get("logging", "log_filename_template")
-    elasticsearch_id = conf.get("elasticsearch", "log_id_template")
-
-    stored = session.execute(
-        select(
-            log_template_table.c.filename,
-            log_template_table.c.elasticsearch_id,
-        )
-        .order_by(log_template_table.c.id.desc())
-        .limit(1)
-    ).first()
-
-    # If we have an empty table, and the default values exist, we will seed the
-    # table with values from pre 2.3.0, so old logs will still be retrievable.
-    if not stored:
-        is_default_log_id = elasticsearch_id == conf.get_default_value("elasticsearch", "log_id_template")
-        is_default_filename = filename == conf.get_default_value("logging", "log_filename_template")
-        if is_default_log_id and is_default_filename:
-            session.add(
-                LogTemplate(
-                    filename="{{ ti.dag_id }}/{{ ti.task_id }}/{{ ts }}/{{ try_number }}.log",
-                    elasticsearch_id="{dag_id}-{task_id}-{logical_date}-{try_number}",
-                )
-            )
-
-    # Before checking if the _current_ value exists, we need to check if the old config value we upgraded in
-    # place exists!
-    pre_upgrade_filename = conf.upgraded_values.get(("logging", "log_filename_template"), filename)
-    pre_upgrade_elasticsearch_id = conf.upgraded_values.get(
-        ("elasticsearch", "log_id_template"), elasticsearch_id
-    )
-    if pre_upgrade_filename != filename or pre_upgrade_elasticsearch_id != elasticsearch_id:
-        # The previous non-upgraded value likely won't be the _latest_ value (as after we've recorded the
-        # recorded the upgraded value it will be second-to-newest), so we'll have to just search which is okay
-        # as this is a table with a tiny number of rows
-        row = session.execute(
-            select(log_template_table.c.id)
-            .where(
-                or_(
-                    log_template_table.c.filename == pre_upgrade_filename,
-                    log_template_table.c.elasticsearch_id == pre_upgrade_elasticsearch_id,
-                )
-            )
-            .order_by(log_template_table.c.id.desc())
-            .limit(1)
-        ).first()
-        if not row:
-            session.add(
-                LogTemplate(filename=pre_upgrade_filename, elasticsearch_id=pre_upgrade_elasticsearch_id)
-            )
-
-    if not stored or stored.filename != filename or stored.elasticsearch_id != elasticsearch_id:
-        session.add(LogTemplate(filename=filename, elasticsearch_id=elasticsearch_id))
-
-
 def reflect_tables(tables: list[MappedClassProtocol | str] | None, session):
     """
     When running checks prior to upgrades, we use reflection to determine current state of the database.
@@ -1217,7 +1139,6 @@ def _run_upgradedb(
             external_db_manager.upgradedb(work_session, use_migration_files=use_migration_files)
 
         add_default_pool_if_not_exists(session=work_session)
-        synchronize_log_template(session=work_session)
 
 
 @provide_session