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

tis_with_right_state.append(ti_key)
        
        if not tis_with_right_state:
            return len(event_buffer)
        
        filter_for_tis = TI.filter_for_tis(tis_with_right_state)
        query = session.query(TI).filter(filter_for_tis).options(selectinload("dag_model"))
        
        
        tis: Iterator[TI] = with_row_locks(
            query,
            of=TI,
            session=session,
            **skip_locked(session=session),
        )
        for ti in tis:
            try_number = ti_primary_key_to_try_number_map[ti.key.primary]
            buffer_key = ti.key.with_try_number(try_number)
            state, info = event_buffer.pop(buffer_key)
            if state == TaskInstanceState.QUEUED:
                ti.external_executor_id = info
                self.log.info("Setting external_id for %s to %s", ti, info)
                continue
            msg = (
                "TaskInstance Finished: dag_id=%s, task_id=%s, run_id=%s, map_index=%s, "
                "run_start_date=%s, run_end_date=%s, "
                "run_duration=%s, state=%s, executor_state=%s, try_number=%s, max_tries=%s, job_id=%s, "
                "pool=%s, queue=%s, priority_weight=%d, operator=%s, queued_dttm=%s, "
                "queued_by_job_id=%s, pid=%s"
            )