Hi Dennis,
Thanks for writing this up -- result caching is something I have been
thinking about as a following to AIP-103 (and hence listed as follow-up AIP
in AIP-103 doc
<
https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-103%3A+Task+State+Management#:~:text=AIP%2Dxx%20(Task%20Result%20Caching)
)
in for the same: ML and reproducible-pipeline workloads. A few thoughts on
shape, scope, and how this fits with what's already in flight.
*Yes, this needs an AIP*
I'd push back on "this doesn't require an AIP". The proposal touches many
critical paths, especially the scheduler's execution path (skipping
queue/executor on a cache hit).
That's a public-surface, cross-cutting change with behaviour that's hard to
walk back once shipped. The AIP doesn't have to be heavy -- but a written
design that the community can vote on (and that ties down the edge cases
below) will save us from "we shipped it experimental and now four people
interpret cache semantics differently". The discussion thread alone won't
capture the decisions we need to make.
*Relationship to AIP-103 (Task State Management)*
This is the part I think needs the most thought before you start
implementing.
AIP-103 is already landing a pluggable, scoped, JWT-authorized state
backend with:
- A `BaseTaskStateBackend` interface with sync + async
get/set/delete/clear
- A `StateScope` abstraction (`TaskScope`, `AssetScope` today)
- Retention/GC policy at deployment + operator level
- A UI panel for inspecting/clearing entries
- Execution API endpoints under `/state/...`
Result caching needs essentially the same infrastructure: a pluggable
backend, retention (TTL is just a different retention policy), UI for
inspecting/clearing entries, multi-team isolation, garbage collection. If
we ship a parallel `BaseTaskCacheBackend` with its own UI, its own config
section, its own retention story, we end up with two near-identical
subsystems that diverge over time.
My strong preference: extend AIP-103's scope model with a `CacheScope` (or
similar) keyed by `(team, dag_id, task_id, source_hash, input_hash)`, and
reuse the backend abstraction. Result caching then becomes "AIP-103 + a
scheduler-side lookup + a TTL semantics layer". Hence, caching was already
listed as a follow-up future AIP in AIP-103 doc
<
https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-103%3A+Task+State+Management#:~:text=AIP%2Dxx%20(Task%20Result%20Caching)
.
*Specific concerns worth pinning down in the AIP*
A few things I'd want explicitly answered before voting:
- *Observability on cache hits*. If a TI is marked SUCCESS without a
worker run, what does the UI show? `start_date` / `end_date` /
`duration`?
Logs? Does `dag.task.duration` stat get emitted? "Skipped because
cached"
needs to be visually distinct from a real success -- otherwise debugging
"why is my pipeline producing stale data" gets very hard. I'd argue for
a
new TI state (`CACHED` or a sub-state) rather than overloading SUCCESS.
- *Task source hash definition*. You say "computed from the serialized
DAG JSON". The serialized DAG includes `retries`, `pool`, `queue`,
`default_args`, owner, etc. -- none of which affect output. If a user
bumps
`retries` from 2 to 3, the cache invalidates. The hash needs a narrower
definition: the callable's bytecode/source + `op_args` + `op_kwargs`
only.
Worth specifying exactly what's in the hash.
- *XCom backend interaction.* If the XCom backend is S3 and we're
caching the XCom value, a cache hit pushes the *same* pointer to the new
TI's XCom row. Is that pointer still alive? Do we re-validate on hit?
Do we
copy the underlying object? This matters when the XCom backend has its
own
retention separate from the cache TTL.
- *Mapped tasks.* Is the cache key per `map_index`? What happens on
partial hits (3 of 10 mapped tasks hit cache, 7 miss)? Does the executor
still get N slots reserved? This needs to be in the design, not "we'll
figure it out".
- *Deferred / async tasks.* Cache the final result, or skip the deferral
entirely? What about sensors a user has explicitly opted into caching --
that's a foot-gun and I'd consider blocking it outright with a `cache`
param on `BaseSensorOperator` that raises if set to `True`, rather than
relying on "free" inheritance.
- *Determinism assumption.* This is opt-in, so technically user-owned.
But a `@task` calling `datetime.utcnow()` or hitting a non-idempotent
API
will silently produce wrong results across runs once cached. Worth a
prominent doc warning and maybe a static-analysis lint in `airflow dags
reserialize` that flags obviously-non-deterministic callables.
- *Scheduler critical-path latency.* 500ms fail-open timeout * N
cache-eligible tasks per scheduler loop = a noticeable scheduler stall
if
the backend degrades. The lookup needs to be async/batched, and we
should
be explicit about what happens to scheduling throughput when the cache
backend is slow but not failing. Otherwise the "we avoid 100ms
subprocess
overhead" win gets eaten by lookup overhead in the steady state. And can
cause worse effect that we want to avoid -- especially if we have a
pluggable backend that runs on scheduler -- we end up running user code
there.
- *Manual invalidation.* "I'm not sure on exactly how that would work"
-- this is a hard requirement, not a follow-up. Operators will need to
clear the cache for a task/dag after a bug fix, schema change, or
upstream
data correction. Needs a CLI command (`airflow tasks clear-cache ...`)
and
a UI button, both authz'd. Without it, a stale cached result is
unrecoverable except by waiting for TTL or changing the source.
re prior art: Flyte's design is a reasonable reference but their execution
model is materially different -- their tasks are containers with declared
inputs/outputs afaik, so source-hash and input-hash are first-class. We
have neither, which is why source-hash definition (above) matters so much.
The other thing I have been playing around is caching parts of Task
Execution and not the entire task results. Not certain yet but playing
around.
Cheers,
Kaxil
On Fri, 22 May 2026 at 18:14, Ferruzzi, Dennis <[email protected]>
wrote:
Hi all,
I've heard a few people discussing the idea and with the
Deadlines/Callbacks/Workloads project wrapping up, I'd like to work on
Task
Result Caching as my next project. I'm looking for feedback before I
start
implementation.
The short version: for ML and other repetitive workloads, tasks often run
with identical inputs and produce identical outputs across Dag runs.
Currently we spawn a subprocess every time regardless. I'd like to add
an
optional, per-task caching mechanism that lets the scheduler
short-circuit
execution when a valid cached result already exists. No subprocess
spawned, no executor slot consumed.
I don't feel like this requires an AIP, but I'm happy to write one if the
community feels otherwise.
THE BASICS
Users opt in per-task:
@task(cache=True)
def train_model(data_path: str): ...
@task(cache=CacheConfig(ttl=3600, exclude_inputs=["account_id"]))
def fetch_data(account_id: str, date_range: str): ...
# Classic operators
PythonOperator(..., cache=True)
I propose a new "cache" parameter to BaseOperator:
cache: CacheConfig | bool | None = None
cache=True is sugar for CacheConfig() with all defaults. Tasks
without cache set are completely unaffected; the scheduler skips the
check
entirely.
HOW IT WORKS
1. Scheduler is about to queue a cache-enabled task.
2. Computes a cache key from the task's identity and inputs.
3. Checks the cache backend (with a fail-open timeout).
4. Cache hit: mark SUCCESS, push cached result to XCom, done.
5. Cache miss or timeout: execute normally, store result afterward.
The cache check happens in the scheduler before spawning the subprocess.
This is the whole point; we avoid the 100ms+ (estimated, I don't have
hard
numbers on this) subprocess overhead entirely on cache hits. If the
cache
backend is slow or unreachable, we timeout and execute normally.
CACHE KEY STRUCTURE
The cache key is a hash of:
- Always included (system-managed): team_name, dag_id, task_id, task
source hash
- Included by default: all kwargs passed to the task
- User-subtractable via exclude_inputs: kwargs irrelevant to the result
- User-addable via include_inputs: context fields like dag_run.conf
keys
The task source hash is computed on the fly from the serialized Dag JSON
(already loaded in the scheduler) and calculating it shouldn't take more
than a couple microseconds. If the task's code changes, the hash
changes,
and the cache is invalid.
CACHE INVALIDATION
- include_inputs / exclude_inputs determine identity ("Are these the
same inputs?")
- TTL determines validity ("Are these results still fresh?")
A cache hit requires both. TTL is configurable with a global default and
per-task override. Manual invalidation may also be available or may be a
future addition, I'm not sure on exactly how that would work.
STORAGE BACKEND
Pluggable interface with a metadata DB default. Same pattern as XCom
backends, secrets backends, etc. Providers can implement alternatives
(S3,
Redis, dedicated DB).
MULTI-TEAM ISOLATION
team_name is pulled from the Dag's bundle-to-team association, not
user-provided. It will be baked into the cache key hash and stored as a
DB
column for query filtering. When multi_team is enabled, no cross-team
cache hits are possible.
TASK TYPE SUPPORT
- Operators and Transfers: cacheable, opt-in
- Sensors: BaseSensorOperator defaults to cache=False since the point
of
a sensor is to check the world right now, not return a stored answer.
Users could explicitly re-enable it on a per-sensor basis if they want to
for some reason since BaseSensorOperator inherits BaseOperator, but
that's
a "free" feature and not something extra I intend to put time into
supporting or explicitly blocking.
- Waiters and Notifiers: not applicable (don't inherit BaseOperator)
ADMIN CONTROLS
- Global kill switch: [core] task_caching_enabled
- Global TTL default in airflow.cfg
- Cache backend class path: [core] task_cache_backend
- Fail-open timeout (default 500ms?)
PRIOR ART
I could not find any prior discussion of this in the Airflow community
but
I checked Prefect and Flyte; both of them have task caching. This design
is closest to Flyte's (opt-in, source-code-aware, project/domain
isolation)
with stronger multi-team isolation and TTL support.
LAUNCH PLAN
This would launch as an experimental feature behind a feature flag.
OPEN QUESTIONS
- Does this need an AIP?
- Any concerns with adding a "cache" param to BaseOperator?
- Thoughts on the cache key structure (include/exclude approach)?
- Anything I'm missing?
Looking forward to your thoughts.
- ferruzzi
