diff --git a/airflow-core/src/airflow/models/dagrun.py b/airflow-core/src/airflow/models/dagrun.py
index 89999bdbf29f0..5da5d9c856525 100644
--- a/airflow-core/src/airflow/models/dagrun.py
+++ b/airflow-core/src/airflow/models/dagrun.py
@@ -119,6 +119,9 @@
 
 log = structlog.get_logger(__name__)
 
+# Create a set without None for the IN clause
+NON_NULL_SCHEDULABLE_STATES = SCHEDULEABLE_STATES - {None}
+
 
 class TISchedulingDecision(NamedTuple):
     """Type of return for DagRun.task_instance_scheduling_decisions."""
@@ -2081,6 +2084,7 @@ def schedule_tis(
                 result = session.execute(
                     update(TI)
                     .where(TI.id.in_(id_chunk))
+                    .where(or_(TI.state.is_(None), TI.state.in_(NON_NULL_SCHEDULABLE_STATES)))
                     .values(
                         state=TaskInstanceState.SCHEDULED,
                         scheduled_dttm=timezone.utcnow(),