Fix triggerer query where limit is not supported in some MySQL ve…

…rsion (apache#17601)

This PR fixes the triggerrer query where limit is not supported in some DB versions and also fixed the issue where total_hours was used on a timedelta.

airflow/models/trigger.py

@@ -184,12 +184,10 @@ def assign_unassigned(cls, triggerer_id, capacity, session=None):
         # Find triggers who do NOT have an alive triggerer_id, and then assign
         # up to `capacity` of those to us.
         trigger_ids_query = (
-            session.query(cls.id)
-            .filter(cls.triggerer_id.notin_(alive_triggerer_ids))
-            .limit(capacity)
-            .subquery()
+            session.query(cls.id).filter(cls.triggerer_id.notin_(alive_triggerer_ids)).limit(capacity).all()
         )
-        session.query(cls).filter(cls.id.in_(trigger_ids_query)).update(
+        session.query(cls).filter(cls.id.in_([i.id for i in trigger_ids_query])).update(
             {cls.triggerer_id: triggerer_id},
             synchronize_session=False,
         )
+        session.commit()

airflow/triggers/temporal.py

@@ -54,7 +54,7 @@ async def run(self):
         unexpectedly, or handles a DST change poorly.
         """
         # Sleep an hour at a time while it's more than 2 hours away
-        while (self.moment - timezone.utcnow()).total_hours() > 2:
+        while (self.moment - timezone.utcnow()).total_seconds() > 2 * 3600:
             await asyncio.sleep(3600)
         # Sleep a second at a time otherwise
         while self.moment > timezone.utcnow():