### 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)

from airflow.task.task_runner import get_task_runner
        self.task_runner = get_task_runner(self)
        def signal_handler(signum, frame):
            """Setting kill signal handler."""
            self.log.error("Received SIGTERM. Terminating subprocesses")
            self.task_runner.terminate()
            self.handle_task_exit(128 + signum)
        def segfault_signal_handler(signum, frame):
            """Setting sigmentation violation signal handler"""
            self.log.critical(SIGSEGV_MESSAGE)
            self.task_runner.terminate()
            self.handle_task_exit(128 + signum)
            raise AirflowException("Segmentation Fault detected.")
        def sigusr2_debug_handler(signum, frame):
            import sys
            import threading
            import traceback
            id2name = {th.ident: th.name for th in threading.enumerate()}
            for threadId, stack in sys._current_frames().items():
                print(id2name[threadId])
                traceback.print_stack(f=stack)
        signal.signal(signal.SIGSEGV, segfault_signal_handler)
        signal.signal(signal.SIGTERM, signal_handler)
        if not IS_WINDOWS:
            
            signal.signal(signal.SIGUSR2, sigusr2_debug_handler)
        if not self.task_instance.check_and_change_state_before_execution(