Fix sqlite constraints update and ignore session_session_id_uq index

airflow/migrations/versions/0142_2_9_2_update_missing_constraints.py

@@ -76,10 +76,9 @@ def upgrade():
         # SQLite does not support DROP CONSTRAINT
         # We have to recreate the table without the constraint
         conn.execute(sa.text("PRAGMA foreign_keys=off"))
-        conn.execute(sa.text("ALTER TABLE connection RENAME TO connection_old"))
         conn.execute(
             sa.text("""
-        CREATE TABLE "connection" (
+        CREATE TABLE connection_new (
                 id INTEGER NOT NULL,
                 conn_id VARCHAR(250) NOT NULL,
                 conn_type VARCHAR(500) NOT NULL,
@@ -97,8 +96,9 @@ def upgrade():
         )
         """)
         )
-        conn.execute(sa.text("INSERT INTO connection SELECT * FROM connection_old"))
-        conn.execute(sa.text("DROP TABLE connection_old"))
+        conn.execute(sa.text("INSERT INTO connection_new SELECT * FROM connection"))
+        conn.execute(sa.text("DROP TABLE connection"))
+        conn.execute(sa.text("ALTER TABLE connection_new RENAME TO connection"))
         conn.execute(sa.text("PRAGMA foreign_keys=on"))
     else:
         op.execute("ALTER TABLE connection DROP CONSTRAINT IF EXISTS unique_conn_id")
@@ -179,10 +179,9 @@ def upgrade():
         # SQLite does not support DROP CONSTRAINT
         # We have to recreate the table without the constraint
         conn.execute(sa.text("PRAGMA foreign_keys=off"))
-        conn.execute(sa.text("ALTER TABLE dag_run RENAME TO dag_run_old"))
         conn.execute(
             sa.text("""
-            CREATE TABLE dag_run (
+            CREATE TABLE dag_run_new (
                 id INTEGER NOT NULL,
                 dag_id VARCHAR(250) NOT NULL,
                 queued_at TIMESTAMP,
@@ -210,8 +209,9 @@ def upgrade():
         """)
         )
 
-        conn.execute(sa.text("INSERT INTO dag_run SELECT * FROM dag_run_old"))
-        conn.execute(sa.text("DROP TABLE dag_run_old"))
+        conn.execute(sa.text("INSERT INTO dag_run_new SELECT * FROM dag_run"))
+        conn.execute(sa.text("DROP TABLE dag_run"))
+        conn.execute(sa.text("ALTER TABLE dag_run_new RENAME TO dag_run"))
         conn.execute(sa.text("PRAGMA foreign_keys=on"))
         with op.batch_alter_table("dag_run") as batch_op:
             batch_op.create_index("dag_id_state", ["dag_id", "state"], if_not_exists=True)

docs/apache-airflow/img/airflow_erd.sha256

@@ -1 +1 @@
-8e508a6b71c59805a60b6aebc9cb2aff1d76a16ed0d7419dc3a9a32721ea2ef9
+8dbea540d4bb502765e91cc229ddfc2fe1e175ffd8418a6141a6cbb866f30952

tests/utils/test_db.py

@@ -84,6 +84,7 @@ def test_database_schema_and_sqlalchemy_model_are_in_sync(self):
             # Ignore flask-session table/index
             lambda t: (t[0] == "remove_table" and t[1].name == "session"),
             lambda t: (t[0] == "remove_index" and t[1].name == "session_id"),
+            lambda t: (t[0] == "remove_index" and t[1].name == "session_session_id_uq"),
             # sqlite sequence is used for autoincrementing columns created with `sqlite_autoincrement` option
             lambda t: (t[0] == "remove_table" and t[1].name == "sqlite_sequence"),
         ]