Do not backfill old DagRun.created_at

Filling this field in the migration is much too costly. This field is
now left as NULL for rows created prior to the migration.

I couldn't find this field used anywhere.

Author: uranusjr

airflow-core/src/airflow/migrations/versions/0106_3_2_0_add_partition_key_to_backfill_dag_run.py

@@ -44,15 +44,9 @@
 def upgrade():
     """Apply Add partition_key to backfill_dag_run."""
     op.add_column("dag_run", sa.Column("created_at", UtcDateTime(timezone=True), nullable=True))
-    op.execute("update dag_run set created_at = run_after;")
-
-    with disable_sqlite_fkeys(op):
-        with op.batch_alter_table("dag_run", schema=None) as batch_op:
-            batch_op.alter_column("created_at", existing_type=UtcDateTime(timezone=True), nullable=False)
-
-        with op.batch_alter_table("backfill_dag_run", schema=None) as batch_op:
-            batch_op.add_column(sa.Column("partition_key", StringID(), nullable=True))
-            batch_op.alter_column("logical_date", existing_type=sa.TIMESTAMP(), nullable=True)
+    with disable_sqlite_fkeys(op), op.batch_alter_table("backfill_dag_run", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("partition_key", StringID(), nullable=True))
+        batch_op.alter_column("logical_date", existing_type=sa.TIMESTAMP(), nullable=True)
 
 
 def downgrade():
@@ -62,6 +56,5 @@ def downgrade():
         with op.batch_alter_table("backfill_dag_run", schema=None) as batch_op:
             batch_op.alter_column("logical_date", existing_type=sa.TIMESTAMP(), nullable=False)
             batch_op.drop_column("partition_key")
-
-        with op.batch_alter_table("dag_run", schema=None) as batch_op:
-            batch_op.drop_column("created_at")
+    with op.batch_alter_table("dag_run", schema=None) as batch_op:
+        batch_op.drop_column("created_at")

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

@@ -191,7 +191,10 @@ class DagRun(Base, LoggingMixin):
         ForeignKey("log_template.id", name="task_instance_log_template_id_fkey", ondelete="NO ACTION"),
         default=select(func.max(LogTemplate.__table__.c.id)),
     )
-    created_at: Mapped[datetime] = mapped_column(UtcDateTime, default=timezone.utcnow)
+    # 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``.
+    created_at: Mapped[datetime] = mapped_column(UtcDateTime, nullable=True, default=timezone.utcnow)
     updated_at: Mapped[datetime] = mapped_column(
         UtcDateTime, default=timezone.utcnow, onupdate=timezone.utcnow
     )