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

)
        if zombies:
            self.log.warning("Failing (%s) jobs without heartbeat after %s", len(zombies), limit_dttm)
        for ti, file_loc, processor_subdir in zombies:
            zombie_message_details = self._generate_zombie_message_details(ti)
            request = TaskCallbackRequest(
                full_filepath=file_loc,
                processor_subdir=processor_subdir,
                simple_task_instance=SimpleTaskInstance.from_ti(ti),
                msg=str(zombie_message_details),
            )
            self.log.error("Detected zombie job: %s", request)
            self.executor.send_callback(request)
            Stats.incr("zombies_killed")
    @staticmethod
    def _generate_zombie_message_details(ti: TaskInstance):
        zombie_message_details = {
            "DAG Id": ti.dag_id,
            "Task Id": ti.task_id,
            "Run Id": ti.run_id,
        }
        if ti.map_index != -1:
            zombie_message_details["Map Index"] = ti.map_index
        if ti.hostname:
            zombie_message_details["Hostname"] = ti.hostname
        if ti.external_executor_id:
            zombie_message_details["External Executor Id"] = ti.external_executor_id
        return zombie_message_details
    @provide_session
    def _cleanup_stale_dags(self, session: Session = NEW_SESSION) -> None: