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

found_new_filters = (
                len(starved_pools) > num_starved_pools
                or len(starved_dags) > num_starved_dags
                or len(starved_tasks) > num_starved_tasks
            )
            if is_done or not found_new_filters:
                break
            self.log.debug(
                "Found no task instances to queue on the %s. iteration "
                "but there could be more candidate task instances to check.",
                loop_count,
            )
        for pool_name, num_starving_tasks in pool_num_starving_tasks.items():
            Stats.gauge(f"pool.starving_tasks.{pool_name}", num_starving_tasks)
        Stats.gauge("scheduler.tasks.starving", num_starving_tasks_total)
        Stats.gauge("scheduler.tasks.running", num_tasks_in_executor)
        Stats.gauge("scheduler.tasks.executable", len(executable_tis))
        if len(executable_tis) > 0:
            task_instance_str = "\n\t".join(repr(x) for x in executable_tis)
            self.log.info("Setting the following tasks to queued state:\n\t%s", task_instance_str)
            
            filter_for_tis = TI.filter_for_tis(executable_tis)
            session.query(TI).filter(filter_for_tis).update(
                
                
                {
                    TI.state: TaskInstanceState.QUEUED,
                    TI.queued_dttm: timezone.utcnow(),
                    TI.queued_by_job_id: self.id,
                },