Reconcile orphan holding table handling

Orphaned rows were moved into holding tables in 2.2.0-2+,
but OSS 2.2.1+ expects these in differently named tables.
We will rename the tables (if we can).

We will also customize the warning shown on the UI pointing our
customers to file a support request.

airflow/utils/db.py

@@ -828,6 +828,37 @@ def check_task_tables_without_matching_dagruns(session) -> Iterable[str]:
         session.commit()
 
 
+def rename_dangling_tables(session) -> Iterable[str]:
+    """
+    Rename tables holding dangling rows to their final names
+
+    Orphaned rows were moved into holding tables in 2.2.0-2+,
+    but OSS 2.2.1+ expects these in differently named tables.
+    We will rename the tables (if we can).
+    """
+    tables_to_rename = {
+        "_airflow_22_dag_run_dangling": "_airflow_moved__2_2__dag_run",
+        "_airflow_22_task_instance_dangling": "_airflow_moved__2_2__task_instance",
+        "_airflow_22_task_reschedule_dangling": "_airflow_moved__2_2__task_reschedule",
+    }
+
+    existing_table_names = set(inspect(session.get_bind()).get_table_names())
+
+    for old_table, new_table in tables_to_rename.items():
+        if old_table not in existing_table_names:
+            continue
+
+        if new_table in existing_table_names:
+            yield (
+                f"We could not rename table {old_table} to {new_table} because"
+                f"{new_table} table already exists in your database. Please either "
+                f"drop one or both of the tables."
+            )
+            continue  # Check the others as well
+
+        session.execute(text(f"ALTER TABLE {old_table} RENAME TO {new_table};"))
+
+
 @provide_session
 def _check_migration_errors(session=None) -> Iterable[str]:
     """
@@ -839,6 +870,7 @@ def _check_migration_errors(session=None) -> Iterable[str]:
         check_conn_type_null,
         check_run_id_null,
         check_task_tables_without_matching_dagruns,
+        rename_dangling_tables,
     ):
         yield from check_fn(session)
 

airflow/www/templates/airflow/dags.html

@@ -53,10 +53,11 @@
   {% endfor %}
   {% for original_table_name, moved_table_name in migration_moved_data_alerts %}
     {% call message(category='error', dismissable=false) %}
-      Airflow found incompatible data in the <code>{{ original_table_name }}</code> table in the
+      <p>Airflow found incompatible data in the <code>{{ original_table_name }}</code> table in the
       metadatabase, and has moved them to <code>{{ moved_table_name }}</code> during the database migration
-      to upgrade. Please inspect the moved data to decide whether you need to keep them, and manually drop
-      the <code>{{ moved_table_name }}</code> table to dismiss this warning.
+      to upgrade.</p>
+      <p>Please file a support request on <a href="https://support.astronomer.io/hc/en-us">
+      Astronomer's Support Portal</a> so we can assist with resolving this issue.</p>
     {% endcall %}
   {% endfor %}
   {{ super() }}