Add try_number to log table

airflow/api_connexion/openapi/v1.yaml

@@ -1243,6 +1243,8 @@ paths:
         - $ref: "#/components/parameters/FilterDAGID"
         - $ref: "#/components/parameters/FilterTaskID"
         - $ref: "#/components/parameters/FilterRunID"
+        - $ref: "#/components/parameters/FilterMapIndex"
+        - $ref: "#/components/parameters/FilterTryNumber"
         - $ref: "#/components/parameters/Event"
         - $ref: "#/components/parameters/Owner"
         - $ref: "#/components/parameters/Before"
@@ -5587,6 +5589,13 @@ components:
         type: integer
       description: Filter on map index for mapped task.
 
+    FilterTryNumber:
+      in: query
+      name: try_number
+      schema:
+        type: integer
+      description: Filter on try_number for task instance.
+
     OrderBy:
       in: query
       name: order_by

airflow/api_connexion/schemas/event_log_schema.py

@@ -37,6 +37,8 @@ class Meta:
     dag_id = auto_field(dump_only=True)
     task_id = auto_field(dump_only=True)
     run_id = auto_field(dump_only=True)
+    map_index = auto_field(dump_only=True)
+    try_number = auto_field(dump_only=True)
     event = auto_field(dump_only=True)
     execution_date = auto_field(dump_only=True)
     owner = auto_field(dump_only=True)

airflow/migrations/versions/0149_2_10_0_add_try_number_to_audit_log.py

@@ -0,0 +1,54 @@
+#
+# 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 try_number to audit log.
+
+Revision ID: 41b3bc7c0272
+Revises: ec3471c1e067
+Create Date: 2024-07-11 14:48:58.998259
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "41b3bc7c0272"
+down_revision = "ec3471c1e067"
+branch_labels = None
+depends_on = None
+airflow_version = "2.10.0"
+
+
+def upgrade():
+    """Apply add try_number to audit log."""
+    with op.batch_alter_table("log") as batch_op:
+        batch_op.add_column(sa.Column("try_number", sa.Integer(), nullable=True))
+        batch_op.create_index(
+            "idx_log_task_instance", ["dag_id", "task_id", "run_id", "map_index", "try_number"], unique=False
+        )
+
+
+def downgrade():
+    """Unapply add try_number to audit log."""
+    with op.batch_alter_table("log") as batch_op:
+        batch_op.drop_index("idx_log_task_instance")
+        batch_op.drop_column("try_number")

airflow/models/log.py

@@ -40,11 +40,13 @@ class Log(Base):
     owner = Column(String(500))
     owner_display_name = Column(String(500))
     extra = Column(Text)
+    try_number = Column(Integer)
 
     __table_args__ = (
         Index("idx_log_dag", dag_id),
         Index("idx_log_dttm", dttm),
         Index("idx_log_event", event),
+        Index("idx_log_task_instance", dag_id, task_id, run_id, map_index, try_number),
     )
 
     def __init__(self, event, task_instance=None, owner=None, owner_display_name=None, extra=None, **kwargs):
@@ -59,6 +61,7 @@ def __init__(self, event, task_instance=None, owner=None, owner_display_name=Non
             self.task_id = task_instance.task_id
             self.execution_date = task_instance.execution_date
             self.run_id = task_instance.run_id
+            self.try_number = task_instance.try_number
             self.map_index = task_instance.map_index
             if getattr(task_instance, "task", None):
                 task_owner = task_instance.task.owner
@@ -73,6 +76,8 @@ def __init__(self, event, task_instance=None, owner=None, owner_display_name=Non
             self.run_id = kwargs["run_id"]
         if "map_index" in kwargs:
             self.map_index = kwargs["map_index"]
+        if "try_number" in kwargs:
+            self.try_number = kwargs["try_number"]
 
         self.owner = owner or task_owner
         self.owner_display_name = owner_display_name or None

airflow/utils/db.py

@@ -118,7 +118,7 @@ class MappedClassProtocol(Protocol):
     "2.8.1": "88344c1d9134",
     "2.9.0": "1949afb29106",
     "2.9.2": "686269002441",
-    "2.10.0": "ec3471c1e067",
+    "2.10.0": "41b3bc7c0272",
 }
 
 

airflow/www/static/js/types/api-generated.ts

@@ -2598,6 +2598,8 @@ export interface components {
     FilterSourceMapIndex: number;
     /** @description Filter on map index for mapped task. */
     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.
@@ -3676,6 +3678,10 @@ export interface operations {
         task_id?: components["parameters"]["FilterTaskID"];
         /** Returns objects matched by the Run ID. */
         run_id?: components["parameters"]["FilterRunID"];
+        /** Filter on map index for mapped task. */
+        map_index?: components["parameters"]["FilterMapIndex"];
+        /** Filter on try_number for task instance. */
+        try_number?: components["parameters"]["FilterTryNumber"];
         /** The name of event log. */
         event?: components["parameters"]["Event"];
         /** The owner's name of event log. */

docs/apache-airflow/img/airflow_erd.sha256

@@ -1 +1 @@
-88c9ce0742c54f376a3c600600d06ac9a6f80a3a2cb85dfcf2472d1c77ed75db
+3afa3d98632dcfe0f585f8777d1bfcb06a4d4f1df20838ee019dc94fa05ef73d