dag_runs_query = session.query(DagRun.id).filter(DagRun.dag_id == self.dag_id).with_for_update()
        if start_date is None and end_date is None:
            dag_runs_query = dag_runs_query.filter(DagRun.execution_date == start_date)
        else:
            if start_date is not None:
                dag_runs_query = dag_runs_query.filter(DagRun.execution_date >= start_date)
            if end_date is not None:
                dag_runs_query = dag_runs_query.filter(DagRun.execution_date <= end_date)
        locked_dag_run_ids = dag_runs_query.all()
        altered = set_state(
            tasks=tasks_to_set_state,
            execution_date=execution_date,
            run_id=run_id,
            upstream=upstream,
            downstream=downstream,
            future=future,
            past=past,
            state=state,
            commit=commit,
            session=session,
        )
        if not commit:
            del locked_dag_run_ids
            return altered
        
        
        session.flush()
        task_subset = self.partial_subset(
            task_ids_or_regex=task_ids,
            include_downstream=True,