Fix DB migrations from 2.10.5 to 3.0.0 for SQlite

FK naming is so in-consistent. SQLite doesn't really "have" fk names, so this
is empirically what worked.

Fixes #49296

Author: ashb

airflow-core/docs/img/airflow_erd.sha256

@@ -1 +1 @@
-da000ad784f974dad63f6db08942d8e968242380f468bc43e35de5634960dcfc
+505e2e6c0f9c6988297f939ea005dde1b458e73b915fe33bc90e745f4498c7ef

airflow-core/src/airflow/migrations/versions/0041_3_0_0_rename_dataset_as_asset.py

@@ -118,6 +118,8 @@ def _rename_pk_constraint(
 def upgrade():
     """Rename dataset as asset."""
     dialect = op.get_bind().dialect.name
+    is_sqlite = dialect == "sqlite"
+
     # Rename tables
     for original_name, new_name in table_name_mappings:
         op.rename_table(original_name, new_name)
@@ -155,7 +157,10 @@ def upgrade():
         )
 
     with op.batch_alter_table("asset_alias_asset", schema=None) as batch_op:
-        batch_op.drop_constraint(op.f("dataset_alias_dataset_alias_id_fkey"), type_="foreignkey")
+        batch_op.drop_constraint(
+            "ds_dsa_alias_id" if is_sqlite else op.f("dataset_alias_dataset_dataset_id_fkey"),
+            type_="foreignkey",
+        )
         _rename_index(
             batch_op=batch_op,
             original_name="idx_dataset_alias_dataset_alias_id",
@@ -172,7 +177,10 @@ def upgrade():
             ondelete="CASCADE",
         )
 
-        batch_op.drop_constraint(op.f("dataset_alias_dataset_dataset_id_fkey"), type_="foreignkey")
+        batch_op.drop_constraint(
+            "ds_dsa_dataset_id" if is_sqlite else op.f("dataset_alias_dataset_alias_id_fkey"),
+            type_="foreignkey",
+        )
         if dialect == "postgresql":
             op.execute("ALTER TABLE asset_alias_asset DROP CONSTRAINT IF EXISTS ds_dsa_alias_id")
             op.execute("ALTER TABLE asset_alias_asset DROP CONSTRAINT IF EXISTS ds_dsa_dataset_id")
@@ -196,7 +204,10 @@ def upgrade():
         )
 
     with op.batch_alter_table("asset_alias_asset_event", schema=None) as batch_op:
-        batch_op.drop_constraint(op.f("dataset_alias_dataset_event_alias_id_fkey"), type_="foreignkey")
+        batch_op.drop_constraint(
+            "dss_de_alias_id" if is_sqlite else op.f("dataset_alias_dataset_event_alias_id_fkey"),
+            type_="foreignkey",
+        )
         _rename_index(
             batch_op=batch_op,
             original_name="idx_dataset_alias_dataset_event_alias_id",
@@ -212,7 +223,10 @@ def upgrade():
             ondelete="CASCADE",
         )
 
-        batch_op.drop_constraint(op.f("dataset_alias_dataset_event_event_id_fkey"), type_="foreignkey")
+        batch_op.drop_constraint(
+            "dss_de_event_id" if is_sqlite else op.f("dataset_alias_dataset_event_event_id_fkey"),
+            type_="foreignkey",
+        )
         if dialect == "postgresql":
             op.execute("ALTER TABLE asset_alias_asset_event DROP CONSTRAINT IF EXISTS dss_de_alias_id")
             op.execute("ALTER TABLE asset_alias_asset_event DROP CONSTRAINT IF EXISTS dss_de_event_id")
@@ -469,6 +483,8 @@ def upgrade():
 
 def downgrade():
     """Unapply Rename dataset as asset."""
+    dialect = op.get_bind().dialect.name
+    is_sqlite = dialect == "sqlite"
     # Rename tables
     for original_name, new_name in table_name_mappings:
         op.rename_table(new_name, original_name)
@@ -507,7 +523,7 @@ def downgrade():
             unique=False,
         )
         batch_op.create_foreign_key(
-            constraint_name=op.f("dataset_alias_dataset_alias_id_fkey"),
+            constraint_name="ds_dsa_dataset_id" if is_sqlite else op.f("dataset_alias_dataset_alias_id_fkey"),
             referent_table="dataset_alias",
             local_cols=["alias_id"],
             remote_cols=["id"],
@@ -523,7 +539,7 @@ def downgrade():
             unique=False,
         )
         batch_op.create_foreign_key(
-            constraint_name=op.f("dataset_alias_dataset_dataset_id_fkey"),
+            constraint_name="ds_dsa_alias_id" if is_sqlite else op.f("dataset_alias_dataset_dataset_id_fkey"),
             referent_table="dataset",
             local_cols=["dataset_id"],
             remote_cols=["id"],
@@ -540,7 +556,9 @@ def downgrade():
             unique=False,
         )
         batch_op.create_foreign_key(
-            constraint_name=op.f("dataset_alias_dataset_event_alias_id_fkey"),
+            constraint_name="dss_de_alias_id"
+            if is_sqlite
+            else op.f("dataset_alias_dataset_event_alias_id_fkey"),
             referent_table="dataset_alias",
             local_cols=["alias_id"],
             remote_cols=["id"],
@@ -556,7 +574,9 @@ def downgrade():
             unique=False,
         )
         batch_op.create_foreign_key(
-            constraint_name=op.f("dataset_alias_dataset_event_event_id_fkey"),
+            constraint_name="dss_de_event_id"
+            if is_sqlite
+            else op.f("dataset_alias_dataset_event_event_id_fkey"),
             referent_table="dataset_event",
             local_cols=["event_id"],
             remote_cols=["id"],

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

@@ -1219,12 +1219,6 @@ def downgrade(*, to_revision, from_revision=None, show_sql_only=False, session:
             except ImportError:
                 log.warning("Import error occurred while importing FABDBManager. Skipping the check.")
                 pass
-        if not inspect(settings.engine).has_table("ab_user"):
-            log.error(
-                "Downgrade to revision less than 3.0.0 requires that `ab_user` table is present. "
-                "Please add FabDBManager to [core] external_db_managers and run fab migrations before proceeding"
-            )
-            return
     with create_global_lock(session=session, lock=DBLocks.MIGRATIONS):
         if show_sql_only:
             log.warning("Generating sql scripts for manual migration.")