### Description

expanding tasks in batches to allow mapped tasks spawn more than 1024 processes.

### Use case/motivation

Maximum length of a list is limited to 1024 by `max_map_length (AIRFLOW__CORE__MAX_MAP_LENGTH)`.
during scheduling of the new tasks, an UPDATE query is ran that tries to set all the new tasks at once.  Increasing `max_map_length` more than 4K makes airflow scheduler completely unresponsive.

Also, Postgres throws `stack depth limit exceeded` error which can be fixed by updating to a newer version and setting `max_stack_depth` higher. But it doesn't really matter because airflow scheduler freezes up.
 
As a workaround, I split the dag runs into subdag runs which works but it would be much nicer if we didn't have to worry about exceeding  `max_map_length`. 

### Related issues

It was discussed here:
[Increasing 'max_map_length' leads to SQL 'max_stack_depth' error with 5000 dags to be spawned #28478](https://github.com/apache/airflow/discussions/28478)

### Are you willing to submit a PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

self._send_sla_callbacks_to_processor(dag)
        if callback:
            self.job.executor.send_callback(callback)
        else:
            self.log.debug("callback is empty")
    def _send_sla_callbacks_to_processor(self, dag: DAG) -> None:
        """Sends SLA Callbacks to DagFileProcessor if tasks have SLAs set and check_slas=True."""
        if not settings.CHECK_SLAS:
            return
        if not any(isinstance(task.sla, timedelta) for task in dag.tasks):
            self.log.debug("Skipping SLA check for %s because no tasks in DAG have SLAs", dag)
            return
        if not dag.timetable.periodic:
            self.log.debug("Skipping SLA check for %s because DAG is not scheduled", dag)
            return
        dag_model = DagModel.get_dagmodel(dag.dag_id)
        if not dag_model:
            self.log.error("Couldn't find DAG %s in database!", dag.dag_id)
            return
        request = SlaCallbackRequest(
            full_filepath=dag.fileloc,
            dag_id=dag.dag_id,
            processor_subdir=dag_model.processor_subdir,
        )
        self.job.executor.send_callback(request)
    @provide_session
    def _fail_tasks_stuck_in_queued(self, session: Session = NEW_SESSION) -> None: