Resolve SA warnings in migrations scripts (#39418)

(cherry picked from commit 5e58723)

airflow/migrations/versions/0093_2_2_0_taskinstance_keyed_to_dagrun.py

@@ -224,8 +224,7 @@ def upgrade():
                 constraints = get_mssql_table_constraints(conn, "task_instance")
                 pk, _ = constraints["PRIMARY KEY"].popitem()
                 batch_op.drop_constraint(pk, type_="primary")
-            elif dialect_name not in ("sqlite"):
-                batch_op.drop_constraint("task_instance_pkey", type_="primary")
+            batch_op.drop_constraint("task_instance_pkey", type_="primary")
             batch_op.drop_index("ti_dag_date")
             batch_op.drop_index("ti_state_lkp")
             batch_op.drop_column("execution_date")
@@ -401,7 +400,7 @@ def _multi_table_update(dialect_name, target, column):
     if dialect_name == "sqlite":
         # Most SQLite versions don't support multi table update (and SQLA doesn't know about it anyway), so we
         # need to do a Correlated subquery update
-        sub_q = select(dag_run.c[column.name]).where(condition)
+        sub_q = select(dag_run.c[column.name]).where(condition).scalar_subquery()
 
         return target.update().values({column: sub_q})
     else:

airflow/migrations/versions/0102_2_3_0_switch_xcom_table_to_use_run_id.py

@@ -92,16 +92,14 @@ def upgrade():
     xcom = Table("xcom", metadata, *_get_old_xcom_columns())
     dagrun = _get_dagrun_table()
     query = select(
-        [
-            dagrun.c.id,
-            xcom.c.task_id,
-            xcom.c.key,
-            xcom.c.value,
-            xcom.c.timestamp,
-            xcom.c.dag_id,
-            dagrun.c.run_id,
-            literal_column("-1"),
-        ],
+        dagrun.c.id,
+        xcom.c.task_id,
+        xcom.c.key,
+        xcom.c.value,
+        xcom.c.timestamp,
+        xcom.c.dag_id,
+        dagrun.c.run_id,
+        literal_column("-1"),
     ).select_from(
         xcom.join(
             right=dagrun,

airflow/migrations/versions/0104_2_3_0_migrate_rtif_to_use_run_id_and_map_index.py

@@ -99,7 +99,7 @@ def _multi_table_update(dialect_name, target, column):
     if dialect_name == "sqlite":
         # Most SQLite versions don't support multi table update (and SQLA doesn't know about it anyway), so we
         # need to do a Correlated subquery update
-        sub_q = select(dag_run.c[column.name]).where(condition)
+        sub_q = select(dag_run.c[column.name]).where(condition).scalar_subquery()
 
         return target.update().values({column: sub_q})
     else:

airflow/migrations/versions/0105_2_3_0_add_map_index_to_taskfail.py

@@ -95,7 +95,7 @@ def _update_value_from_dag_run(
     if dialect_name == "sqlite":
         # Most SQLite versions don't support multi table update (and SQLA doesn't know about it anyway), so we
         # need to do a Correlated subquery update
-        sub_q = select(dag_run.c[target_column.name]).where(condition)
+        sub_q = select(dag_run.c[target_column.name]).where(condition).scalar_subquery()
 
         return target_table.update().values({target_column: sub_q})
     else:

docs/apache-airflow/img/airflow_erd.sha256

@@ -1 +1 @@
-00b4a76587ea1f5f2f317a555a083a0b4964c727d0a3c8c2044ac4f19c58abd4
+a82d08fd5d725c40dab022228ac57a514851c1718c21fe8712ff5f12698b5d1f