Speed up repeated latest Dag lookups#69815
Conversation
viiccwen
changed the title
|
cc @guan404ming, @jason810496. 🙌 |
API requests repeatedly resolve the latest Dag and should benefit from the configured cache without sacrificing freshness. Signed-off-by: viiccwen <vicwen@apache.org>
0686ad9 to
d108c81
Compare
There was a problem hiding this comment.
Pull request overview
This PR optimizes API-server latest Dag resolution by reusing an already-cached deserialized Dag when the database “latest serialized row” query returns the same dag_version_id and dag_hash, avoiding repeated deserialization while still preserving freshness via the latest-row query.
Changes:
- Reuse cached deserialized Dag in
DBDagBag.get_latest_version_of_dag()when version ID + hash match; evict and reload on in-place hash changes. - Add unit tests covering cache-hit reuse, miss behavior, and validation timestamp refresh + metrics.
- Update cache configuration/docs wording to clarify freshness revalidation vs. eviction (LRU/TTL).
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| airflow-core/src/airflow/models/dagbag.py | Add hash/version-aware reuse path for latest-Dag lookups while keeping the latest-row DB query. |
| airflow-core/tests/unit/models/test_dagbag.py | Add regression/unit tests for cache hit/miss behavior and metrics for latest-Dag lookups. |
| airflow-core/src/airflow/config_templates/config.yml | Clarify cache behavior in config reference text. |
| airflow-core/docs/faq.rst | Clarify API server cache freshness checking vs. eviction/TTL in user documentation. |
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
jason810496
left a comment
jason810496
left a comment
There was a problem hiding this comment.
I'd prefer to wait until @ephraimbuddy confirm on this case.
| def get_latest_version_of_dag(self, dag_id: str, *, session: Session) -> SerializedDAG | None: | ||
| """Get the latest version of a dag by its id.""" | ||
| from airflow.models.serialized_dag import SerializedDagModel | ||
|
|
||
| if not (serdag := SerializedDagModel.get(dag_id, session=session)): | ||
| return None | ||
|
|
||
| with self._lock: | ||
| cached = self._dags.get(serdag.dag_version_id) | ||
| if cached is not None: | ||
| # The latest-row query already loaded dag_hash, so no revalidation query is needed. | ||
| if cached.dag_hash == serdag.dag_hash: | ||
| self._dags[serdag.dag_version_id] = cached._replace(last_validated=time.monotonic()) | ||
| else: | ||
| # A serialized Dag can be updated in place without changing its version ID. | ||
| self._dags.pop(serdag.dag_version_id, None) | ||
| cached = None | ||
|
|
||
| if cached is not None: | ||
| if self._use_cache: | ||
| stats.incr("api_server.dag_bag.cache_hit") | ||
| return cached.dag | ||
| if self._use_cache: | ||
| stats.incr("api_server.dag_bag.cache_miss") | ||
| return self._read_dag(serdag) |
There was a problem hiding this comment.
Are we intentionally to retrieve the latest Dag instance from Metadata DB without cache for the get_latest_version_of_dag method introduced in #53153 ?
There's a similar method -- get_dag that always read from the cache first.
airflow/airflow-core/src/airflow/models/dagbag.py
Lines 177 to 179 in 50f90d1
Would like to double check the usage and the intention back then, thanks.
closes #69814
Repeated API requests that resolve the latest Dag always need to query the latest serialized row to preserve freshness. However, they also deserialized that row on every request, even when the same Dag version and hash were already cached.
This change reuses the cached deserialized Dag when the queried version ID and hash are unchanged.
It keeps the existing latest-row query, reloads serialized Dags that were updated in place, and refreshes the cache validation timestamp. The cache documentation now distinguishes freshness revalidation from LRU/TTL eviction.
A real-path benchmark measured the ORM latest-row query, Dag deserialization, and task route handler. Each result is the median of three rounds with 50 lookups per round.
The benchmark uses a new SQLAlchemy session for each lookup and leaves
[core] compress_serialized_dagsat its default value (False). It excludes HTTP middleware, authentication, and response encoding, so the table isolates the route-handler path affected by this change rather than claiming the same percentage improvement for total HTTP latency.Tests cover matching cache hits, validation timestamp refresh, cache metrics, cache misses, and in-place hash changes. The DagBag test file, related API/query-count tests, Ruff checks, pre-commit checks, manual checks, and Apache Airflow documentation build pass.
Was generative AI tooling used to co-author this PR?
{pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.