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

"""
        Find all dags that were not updated by Dag Processor recently and mark them as inactive.
        In case one of DagProcessors is stopped (in case there are multiple of them
        for different dag folders), it's dags are never marked as inactive.
        Also remove dags from SerializedDag table.
        Executed on schedule only if [scheduler]standalone_dag_processor is True.
        """
        self.log.debug("Checking dags not parsed within last %s seconds.", self._dag_stale_not_seen_duration)
        limit_lpt = timezone.utcnow() - timedelta(seconds=self._dag_stale_not_seen_duration)
        stale_dags = (
            session.query(DagModel).filter(DagModel.is_active, DagModel.last_parsed_time < limit_lpt).all()
        )
        if not stale_dags:
            self.log.debug("Not stale dags found.")
            return
        self.log.info("Found (%d) stales dags not parsed after %s.", len(stale_dags), limit_lpt)
        for dag in stale_dags:
            dag.is_active = False
            SerializedDagModel.remove_dag(dag_id=dag.dag_id, session=session)
        session.flush()
    def _set_orphaned(self, dataset: DatasetModel) -> int:
        self.log.info("Orphaning unreferenced dataset '%s'", dataset.uri)
        dataset.is_orphaned = expression.true()
        return 1
    @provide_session
    def _orphan_unreferenced_datasets(self, session: Session = NEW_SESSION) -> None: