Avoid duplicate trigger-rule upstream-count queries per scheduling pass

[kaxil]

Summary

TriggerRuleDep sizes a task's upstream set with a SELECT task_id, count(*) ... GROUP BY task_id, and it only runs that when one of the upstreams is mapped. The problem: if several downstream tasks all depend on the same mapped upstream, each of them runs the identical query during the same scheduling pass. This caches the result on the DepContext for the duration of a pass, so it runs once per distinct upstream set instead of once per downstream.

Impact

The win scales with how many downstream tasks sit behind a mapped upstream. Measured on a DAG with one mapped upstream (5 instances) feeding 60 plain downstream tasks, counting just this query during one scheduling pass:

	upstream-count queries per pass
before	60
after	1

That count recurs every update_state pass, so across a run it's roughly downstreams x passes queries collapsing to passes. Each query is individually cheap (sub-millisecond), so this is about cutting database round-trips, not query time. It happens in the per-run scheduling work, not the serialized critical section, so it's a latency win for mapping-heavy DAGs rather than a throughput-ceiling change.

How it works

• Key is (dag_id, run_id, frozenset(upstream_task_ids)), stored on DepContext, which is already built once per _get_ready_tis pass and reused for every TI in that pass (the same object that caches finished_tis).
• It only kicks in when the task isn't inside a mapped task group. That's the branch where the predicate is plain task_id IN (upstream_ids) and comes out the same for every downstream with the same upstreams. Tasks inside a mapped task group have per map-index predicates, so they keep running their own query.
• upstream_setup is still summed in the caller from the cached rows, so the setup count stays right per downstream.

Staleness

It follows finished_tis (a per-pass snapshot), and I clear it in _get_ready_tis on expansion, since that's when a mapped task's instance count actually changes mid-pass. The revise-map-index path can change counts too, but it only adds not-yet-finished instances, and an unfinished upstream keeps the count-based rules from going ready, so a stale value there doesn't change the outcome.