Remove dag_code records with no serdag

airflow-core/docs/img/airflow_erd.sha256

@@ -1 +1 @@
-bc93e7288a7a8355b15dc721accaf80260f370f3afa0d478248f9fe4692a1f1d
+2a3ce828be7c3800cb61de4c081fcf46f1f94e31ac732f831e65a8a9581b5bb0

airflow-core/docs/installation/upgrading_to_airflow3.rst

@@ -41,6 +41,11 @@ Step 2: Clean and back up your existing Airflow Instance
   upgrade process. These schema changes can take a long time if the database is large. For a faster, safer migration, we recommend that you clean up your Airflow meta-database before the upgrade.
   You can use the ``airflow db clean`` :ref:`Airflow CLI command<cli-db-clean>` to trim your Airflow database.
 
+- Ensure that there are no errors related to dag processing, such as ``AirflowDagDuplicatedIdException``.  You should
+  be able to run ``airflow dags reserialize`` with no errors.  If you have have to resolve errors from dag processing,
+  ensure you deploy your changes to your old instance prior to upgrade, and wait until your dags have all been reprocessed
+  (and all errors gone) before you proceed with upgrade.
+
 
 Step 3: DAG Authors - Check your Airflow DAGs for compatibility
 ----------------------------------------------------------------

airflow-core/src/airflow/migrations/versions/0047_3_0_0_add_dag_versioning.py

@@ -66,6 +66,9 @@ def _airflow_2_fileloc_hash(fileloc):
 def upgrade():
     """Apply add dag versioning."""
     conn = op.get_bind()
+
+    op.execute("DELETE FROM dag_code WHERE fileloc_hash NOT IN (SELECT fileloc_hash FROM serialized_dag)")
+
     op.create_table(
         "dag_version",
         sa.Column("id", UUIDType(binary=False), nullable=False),