### Description

A count of datasets that are currently registered/declared in an Airflow deployment.

### Use case/motivation

Would be nice to see how deployments are adopting datasets.

### Related issues

_No response_

### 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, states: list[TaskInstanceState], session: Session
    ) -> tuple[DefaultDict[str, int], DefaultDict[tuple[str, str], int]]:
        """
        Get the concurrency maps.
        :param states: List of states to query for
        :return: A map from (dag_id, task_id) to # of task instances and
         a map from (dag_id, task_id) to # of task instances in the given state list
        """
        ti_concurrency_query: list[tuple[str, str, int]] = (
            session.query(TI.task_id, TI.dag_id, func.count("*"))
            .filter(TI.state.in_(states))
            .group_by(TI.task_id, TI.dag_id)
        ).all()
        dag_map: DefaultDict[str, int] = defaultdict(int)
        task_map: DefaultDict[tuple[str, str], int] = defaultdict(int)
        for result in ti_concurrency_query:
            task_id, dag_id, count = result
            dag_map[dag_id] += count
            task_map[(dag_id, task_id)] = count
        return dag_map, task_map
    def _executable_task_instances_to_queued(self, max_tis: int, session: Session) -> list[TI]: