This is an automated email from the ASF dual-hosted git repository.
amoghrajesh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 85843a30d5c AIP-103: Documentation for Task and Asset Store (#67299)
85843a30d5c is described below
commit 85843a30d5c79e491bec0187221e1b247067bcf6
Author: Jake McGrath <[email protected]>
AuthorDate: Sat Jun 6 13:14:24 2026 -0400
AIP-103: Documentation for Task and Asset Store (#67299)
Co-authored-by: Amogh Desai <[email protected]>
---
.../docs/administration-and-deployment/index.rst | 2 +
.../task-and-asset-store-cleanup.rst | 73 ++++++
.../task-and-asset-store.rst | 210 ++++++++++++++++
airflow-core/docs/core-concepts/asset-store.rst | 177 +++++++++++++
airflow-core/docs/core-concepts/index.rst | 9 +
.../docs/core-concepts/task-and-asset-store.rst | 80 ++++++
airflow-core/docs/core-concepts/task-store.rst | 278 +++++++++++++++++++++
7 files changed, 829 insertions(+)
diff --git a/airflow-core/docs/administration-and-deployment/index.rst
b/airflow-core/docs/administration-and-deployment/index.rst
index 812ae95d8bc..545ea1b6f3b 100644
--- a/airflow-core/docs/administration-and-deployment/index.rst
+++ b/airflow-core/docs/administration-and-deployment/index.rst
@@ -38,3 +38,5 @@ This section contains information about deploying Dags into
production and the a
priority-weight
web-stack
plugins
+ task-and-asset-store
+ task-and-asset-store-cleanup
diff --git
a/airflow-core/docs/administration-and-deployment/task-and-asset-store-cleanup.rst
b/airflow-core/docs/administration-and-deployment/task-and-asset-store-cleanup.rst
new file mode 100644
index 00000000000..11091431aec
--- /dev/null
+++
b/airflow-core/docs/administration-and-deployment/task-and-asset-store-cleanup.rst
@@ -0,0 +1,73 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _task-and-asset-store-cleanup:
+
+Task and Asset Store Cleanup
+=============================
+
+.. versionadded:: 3.3
+
+Airflow does not automatically purge task store rows on a schedule. Cleanup
(also known as "garbage collection") is the responsibility of the user (you)
and must be triggered explicitly via the CLI. This page explains what gets
cleaned up, how to run it, and how to integrate it into a recurring maintenance
workflow.
+
+
+What gets cleaned up
+--------------------
+
+The cleanup command operates only on **task store** rows in the
``MetastoreStoreBackend``. Asset store rows are never touched by this command.
Asset store rows are removed only by the orphan sweep when an asset is
deactivated (see :ref:`task-and-asset-store`).
+
+A task store row is eligible for deletion when its ``expires_at`` timestamp is
in the past. ``expires_at`` is computed on the worker at write time:
+
+* Keys written with an explicit ``retention=timedelta(...)`` expire after that
duration from the time of the write.
+* Keys written with ``retention=None`` (the default) pick up an expiry based
on ``[state_store] default_retention_days``. If that value is ``> 0``, the key
expires that many days after the write.
+* Keys written with ``retention=NEVER_EXPIRE`` have ``expires_at = NULL`` and
a flag that marks them as permanent. They are **never** deleted by this command
regardless of configuration.
+
+If ``[state_store] default_retention_days = 0``, keys written without an
explicit retention have ``expires_at = NULL`` (no expiry) and are also skipped.
Only keys with a non-null, past ``expires_at`` are removed.
+
+.. note::
+
+ Custom backends (``[state_store] backend`` set to anything other than the
default) are **explicitly skipped**. The cleanup command prints a message and
exits cleanly without deleting anything. If your custom backend needs its own
retention logic, implement it in ``BaseStoreBackend.cleanup()`` and call it
from your own maintenance process.
+
+
+Running cleanup
+---------------
+
+The command is::
+
+ airflow state-store cleanup-task-store
+
+It reads ``[state_store] default_retention_days`` and ``[state_store]
state_cleanup_batch_size`` from the ``airflow.cfg`` file, then deletes all
eligible rows.
+
+**Dry run**
+
+Use ``--dry-run`` to preview what would be deleted without removing anything::
+
+ airflow state-store cleanup-task-store --dry-run
+
+The output lists every row that would be deleted, grouped by dag, run, task,
map index, and key.
+
+**Batching**
+
+By default (``state_cleanup_batch_size = 0``) all eligible rows are deleted in
a single statement. On deployments with large ``task_store`` tables, set a
batch size to reduce lock duration per transaction::
+
+ # airflow.cfg
+ [state_store]
+ state_cleanup_batch_size = 10000
+
+The command then deletes rows in batches of 10,000, committing after each
batch, until no eligible rows remain.
+
+How often to run cleanup depends on your write volume and the value of
``default_retention_days``. A weekly cleanup may be sufficient for most
environments. For high-throughput pipelines that write task store entries on
every task execution, consider running cleanup more frequently to keep the
``task_store`` table small.
diff --git
a/airflow-core/docs/administration-and-deployment/task-and-asset-store.rst
b/airflow-core/docs/administration-and-deployment/task-and-asset-store.rst
new file mode 100644
index 00000000000..5cf54115e46
--- /dev/null
+++ b/airflow-core/docs/administration-and-deployment/task-and-asset-store.rst
@@ -0,0 +1,210 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _task-and-asset-store:
+
+Task and Asset Store Configuration
+====================================
+
+.. versionadded:: 3.3
+
+The task and asset store is the persistence layer for :doc:`task store
</core-concepts/task-store>` and :doc:`asset store
</core-concepts/asset-store>`. By default, both are stored in the Airflow
metadata database. This page describes the available configuration options,
garbage-collection semantics, and how to provide a custom backend.
Configuration (``[state_store]``), the CLI (``airflow state-store``), and the
backend base class (:class:`~airflow.sdk.state.BaseStoreBackend`) all use t
[...]
+
+Configuration reference
+-----------------------
+
+All options live under the ``[state_store]`` section of ``airflow.cfg``.
+
+.. note::
+
+ The config section is ``[state_store]``, **not** ``[task_store]``.
+
+``backend``
+~~~~~~~~~~~
+
+Full dotted path to a class that implements
:class:`~airflow.sdk.state.BaseStoreBackend`. Defaults to the built-in
metastore backend.
+
+.. code-block:: ini
+
+ [state_store]
+ backend = mypackage.state.CustomStateBackend
+
+``default_retention_days``
+~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Number of days after which task store rows expire. When a key is written with
no explicit retention, expires_at is computed on the worker as now +
default_retention_days. Changing this setting does not affect already-written
rows.
+
+* Set to ``0`` to disable time-based cleanup entirely.
+* Default: ``30``.
+* This setting does **not** apply to asset store rows.
+
+.. code-block:: ini
+
+ [state_store]
+ default_retention_days = 30
+
+``clear_on_success``
+~~~~~~~~~~~~~~~~~~~~
+
+When ``True``, all task store keys for a task instance are automatically
deleted when that task instance moves to the ``success`` state. Defaults to
``False``, which preserves task store entries after success for observability
(e.g. the submitted job ID or the last row count is still readable from the UI
or REST API after the run completes).
+
+.. important::
+
+ ``clear_on_success`` clears **task store only**. It has no effect on asset
store. Asset store is scoped to the asset rather than the task instance and
must be cleared explicitly.
+
+.. code-block:: ini
+
+ [state_store]
+ clear_on_success = False
+
+``state_cleanup_batch_size``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Number of rows deleted per batch during garbage collection cleanup. Set to
``0`` (default) to delete all matching rows in a single statement. Tune this on
deployments with large ``task_store`` tables to reduce lock contention.
+
+.. code-block:: ini
+
+ [state_store]
+ state_cleanup_batch_size = 10000
+
+.. _task-and-asset-store:worker-backends:
+
+Worker-side backend (``[workers] state_backend``)
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+A separate, optional config key under ``[workers]`` lets you route task store
and asset store values through a worker-side backend before they reach the API
server.
+
+.. code-block:: ini
+
+ [workers]
+ state_backend = mypackage.state.S3StateBackend
+
+When this is set, ``TaskStoreAccessor.set()`` calls
``serialize_task_store_to_ref()`` on the worker-side backend before sending the
returned value (a reference to the actual storage) to the Execution API, and
``get()`` calls ``deserialize_task_store_from_ref()`` after receiving the
stored reference from the Execution API. See `Custom worker-side backends`_
below.
+
+
+Garbage collection semantics
+-----------------------------
+
+The cleanup task, also known as "garbage collection" is triggered using the
Airflow CLI. The command to trigger the cleanup task is ``airflow state-store
cleanup-task-store``. This process removes store rows according to the
following rules:
+
+**Time-based expiry (task store only)**
+ Rows whose ``expires_at < now()`` are deleted. ``expires_at`` is computed on
the *worker* at write time, not by the server.
+
+**``default_retention_days`` fallback (task store only)**
+ Keys written with no explicit retention get an ``expires_at`` of now +
default_retention_days computed at write time. Garbage collection deletes rows
where ``expires_at < now()``."
+
+**``NEVER_EXPIRE`` keys**
+ Keys set with ``retention=NEVER_EXPIRE`` are stored with ``expires_at =
NULL`` and a flag that tells the garbage collection to skip them
unconditionally. They are never deleted by time-based cleanup, regardless of
``default_retention_days``.
+
+**``on_delete=CASCADE`` (asset store)**
+ When an asset is deleted, all corresponding asset store rows for that asset
are deleted.
+
+.. important::
+
+ Garbage collection only works for the ``MetastoreStateBackend``. Custom
backends are explicitly skipped.
+
+
+
+Custom backends
+---------------
+
+A custom backend must subclass :class:`~airflow.sdk.state.BaseStoreBackend`
and implement its abstract methods: ``get``, ``set``, ``delete``, and ``clear``
for synchronous callers and the ``aget``, ``aset``, ``adelete``, and ``aclear``
async equivalents. Refer to :class:`~airflow.sdk.state.BaseStoreBackend` for
the full API.
+
+Each method receives a ``scope`` argument that is either a
:class:`~airflow.sdk.state.TaskScope` or an
:class:`~airflow.sdk.state.AssetScope`. Use ``isinstance`` to dispatch:
+
+.. code-block:: python
+
+ from airflow.sdk.state import BaseStoreBackend, TaskScope, AssetScope
+
+
+ class MyBackend(BaseStoreBackend):
+ def get(self, scope, key, *, session=None):
+ if isinstance(scope, TaskScope):
+ return self._task_store.get(scope, key)
+ elif isinstance(scope, AssetScope):
+ return self._asset_store.get(scope, key)
+
+:class:`~airflow.sdk.state.AssetScope` has three optional fields: ``asset_id``
(integer, server-side only), ``name``, and ``uri``. At least one must be set.
Server-side operations (REST API calls) provide ``asset_id``. Worker-side
operations provide ``name`` or ``uri`` (workers do not have access to the
integer ``asset_id``).
+
+Configure the class via ``[state_store] backend``:
+
+.. code-block:: ini
+
+ [state_store]
+ backend = mypackage.state.MyBackend
+
+
+Custom worker-side backends
+----------------------------
+
+Worker-side backends extend ``BaseStoreBackend`` with two pairs of
serialization hooks. They are configured separately via ``[workers]
state_backend`` and run *on the worker process*, not on the API server. This
lets you store large payloads or credentialed data directly using worker
infrastructure while only a compact reference string is kept in the database.
+
+Override four serialization hooks from
:class:`~airflow.sdk.state.BaseStoreBackend`:
+
+* ``serialize_task_store_to_ref``: called by ``TaskStoreAccessor.set()``
before the value is sent to the Execution API; return a compact reference
string (e.g. an S3 key) to be stored in the database instead of the raw value.
+* ``deserialize_task_store_from_ref``: called by ``TaskStoreAccessor.get()``
after retrieving the reference from the backend; return the actual value.
+* ``serialize_asset_store_to_ref``: same as the task variant but for asset
store; receives the asset name or URI as ``asset_ref``.
+* ``deserialize_asset_store_from_ref``: called by ``AssetStoreAccessor.get()``
to resolve the stored reference back to the actual value.
+
+.. important::
+
+ **References must be deterministic.** Given the same inputs (``ti_id`` +
``key`` for task store; ``asset_ref`` + ``key`` for asset store), the
serialization method must always return the same reference string. Do not embed
timestamps, random UUIDs, or any other non-deterministic component in the
reference path.
+
+ When a key is deleted or cleared, Airflow clears the database reference
*first*, then calls the backend's ``delete()`` or ``clear()`` method. If
backend cleanup fails after the DB row is gone, the external object is
orphaned. Because the reference is deterministic, a subsequent ``set()`` for
the same key will overwrite the orphaned object, making the failure
recoverable. A non-deterministic reference would leave the external object
permanently orphaned with no way to locate it.
+
+Example skeleton:
+
+.. code-block:: python
+
+ from airflow.sdk.state import BaseStoreBackend, TaskScope, AssetScope
+
+ if TYPE_CHECKING:
+ from pydantic import JsonValue
+
+
+ class S3StateBackend(BaseStoreBackend):
+
+ def _task_ref(self, ti_id: str, key: str) -> str:
+ return f"airflow/task-store/{ti_id}/{key}"
+
+ def _asset_ref(self, asset_ref: str, key: str) -> str:
+ import hashlib
+
+ safe = hashlib.sha256(asset_ref.encode()).hexdigest()[:16]
+ return f"airflow/asset-store/{safe}/{key}"
+
+ def serialize_task_store_to_ref(self, *, value: JsonValue, key: str,
ti_id: str) -> str:
+ s3_key = self._task_ref(ti_id, key)
+ s3_client.put_object(Bucket=BUCKET, Key=s3_key,
Body=json.dumps(value).encode())
+ return s3_key
+
+ def deserialize_task_store_from_ref(self, stored: str) -> JsonValue:
+ s3_object = s3_client.get_object(Bucket=BUCKET, Key=stored)
+ return json.loads(s3_object["Body"].read().decode())
+
+ def serialize_asset_store_to_ref(self, *, value: JsonValue, key: str,
asset_ref: str) -> str:
+ s3_key = self._asset_ref(asset_ref, key)
+ s3_client.put_object(Bucket=BUCKET, Key=s3_key,
Body=json.dumps(value).encode())
+ return s3_key
+
+ def deserialize_asset_store_from_ref(self, stored: str) -> JsonValue:
+ s3_object = s3_client.get_object(Bucket=BUCKET, Key=stored)
+ return json.loads(s3_object["Body"].read().decode())
+
+ # Implement the remaining abstract methods as pass-throughs or
delegating to the
+ # default MetastoreStateBackend for the DB side
+ ...
diff --git a/airflow-core/docs/core-concepts/asset-store.rst
b/airflow-core/docs/core-concepts/asset-store.rst
new file mode 100644
index 00000000000..e5d15051ff1
--- /dev/null
+++ b/airflow-core/docs/core-concepts/asset-store.rst
@@ -0,0 +1,177 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _concepts:asset-store:
+
+.. spelling:word-list::
+
+ subscripted
+ subscripting
+
+Asset Store
+===========
+
+.. versionadded:: 3.3
+
+Asset store is a persistent key/value store scoped to an *asset*, independent
of any particular DAG run. Unlike :doc:`task store
</core-concepts/task-store>`, which is tied to a single task instance, asset
store persists across runs and is logically owned by the asset itself. It is
the natural home for cross-run metadata such as watermarks, incremental-load
cursors, and per-asset configuration.
+
+Asset store is accessed through the task context via
``context["asset_store"]``.
+
+
+When is ``asset_store`` available?
+----------------------------------
+
+When using asset store within a task, ``context["asset_store"]`` is populated
for **concrete** :class:`~airflow.sdk.definitions.asset.Asset` inlets and
outlets. A task must declare at least one concrete inlet or outlet for
``asset_store`` to contain any entries.
+
+Accessing asset store using ``context``
+---------------------------------------
+
+An asset becomes available through context["asset_store"] when it is included
in inlets or outlets. You can then retrieve its asset store by subscripting
context["asset_store"] with the asset object.
+
+.. code-block:: python
+
+ from airflow.sdk import Asset, DAG, task
+
+ my_asset = Asset("my_data", uri="s3://bucket/my_data")
+
+ with DAG("example_asset_store", schedule=None):
+
+ @task(inlets=[my_asset], outlets=[my_asset])
+ def process(**context):
+ asset_store = context["asset_store"][my_asset]
+ watermark = asset_store.get("watermark")
+ asset_store.set("watermark", "2024-06-01")
+
+To see asset store in-action in a real DAG, checkout the DAG in
`example_asset_store.py
<https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/example_dags/example_asset_store.py>`_.
+
+Single-inlet shorthand
+~~~~~~~~~~~~~~~~~~~~~~~
+
+For tasks with exactly **one** concrete inlet or outlet, you can call ``get``,
``set``, ``delete``, and ``clear`` directly on ``context["asset_store"]``
without subscripting.
+
+.. code-block:: python
+
+ @task(inlets=[my_asset], outlets=[my_asset])
+ def process_single(**context):
+ asset_store = context["asset_store"]
+ watermark = asset_store.get("watermark")
+ asset_store.set("watermark", "2024-06-01")
+
+If the task has more than one concrete inlet or outlet, calling the shorthand
raises a ``ValueError``. Use the subscript form
(``context["asset_store"][my_asset]``) whenever a task has multiple inlets.
+
+
+API reference
+-------------
+
+The following methods are available on both the per-asset accessor
(``context["asset_store"][my_asset]``) and the shorthand
(``context["asset_store"]``) when the task has exactly one inlet.
+
+``get(key, default)``
+~~~~~~~~~~~~~~~~~~~~~
+
+Returns the stored JSON value, or the ``default`` value if the key does not
exist.
+
+.. code-block:: python
+
+ # Using context
+ watermark = context["asset_store"][my_asset].get("watermark",
default="initial_watermark")
+
+``set(key, value)``
+~~~~~~~~~~~~~~~~~~~
+
+Writes or overwrites a key-value pair. Unlike task store, asset store has no
``retention`` parameter. Values persist until explicitly deleted or until the
asset is deactivated. Like with task store, ``value`` can be any
JSON-compatible type, except for ``None``. This includes:
+
+* ``str``
+* ``int``
+* ``float``
+* ``bool``
+* ``list``
+* ``dict``
+
+.. code-block:: python
+
+ # Using context
+ context["asset_store"][my_asset].set("watermark", "2024-06-01T00:00:00Z")
+
+``delete(key)``
+~~~~~~~~~~~~~~~
+
+Deletes a single key. No-op if the key does not exist.
+
+.. code-block:: python
+
+ # Using context
+ context["asset_store"][my_asset].delete("watermark")
+
+``clear()``
+~~~~~~~~~~~
+
+Deletes *all* asset store keys for the asset.
+
+.. code-block:: python
+
+ # Using context
+ context["asset_store"][my_asset].clear()
+
+Some Example Use cases
+----------------------
+
+Watermark pattern
+~~~~~~~~~~~~~~~~~
+
+The canonical use case for asset store is an incremental-load task that
advances a watermark on each run. The watermark is stored on the asset itself
so any task that reads or writes that asset can access it. This use case is
especially applicable when building things like asset "watchers" using
``BaseEventTrigger``'s.
+
+.. code-block:: python
+
+ from airflow.sdk import Asset, DAG, task
+
+ orders = Asset("orders", uri="s3://data/orders/")
+
+ with DAG("incremental_orders", schedule="@daily"):
+
+ @task(inlets=[orders], outlets=[orders])
+ def load_new_orders(**context):
+ asset_store = context["asset_store"] # single-inlet shorthand
+
+ # Read the last watermark, default to epoch if first run.
+ watermark = asset_store.get("watermark",
default="1970-01-01T00:00:00Z")
+
+ # Fetch only rows created after the watermark.
+ rows = fetch_orders_since(watermark)
+ if not rows:
+ return
+
+ upload_to_warehouse(rows)
+
+ # Advance the watermark to the latest row seen.
+ new_watermark = max(r["created_at"] for r in rows)
+ asset_store.set("watermark", new_watermark)
+
+On each run the task reads the watermark left by the previous run, fetches
only the new data, and then advances the watermark. Because asset store
persists across runs, the next run starts exactly where this one left off, even
across retries, manual re-runs, or Scheduler restarts.
+
+
+Lifetime and garbage collection
+--------------------------------
+
+Asset store rows persist indefinitely. They are **not** subject to the
``[state_store] default_retention_days`` time-based expiry that applies to task
store.
+
+The only automatic cleanup is an *orphan sweep*: when an asset is deactivated
(no ``asset_active`` record exists), its store rows are removed during the next
garbage-collection pass. Until that sweep runs, stale rows may exist in the
database but cannot be written to. The Execution API resolver filters to active
assets only.
+
+To remove asset store entries explicitly, call ``clear()`` from within a task,
or use the REST API.
+
+.. note::
+
+ ``[state_store] clear_on_success`` does **not** clear asset store. Asset
store is cross-run by design, so automatic task-level cleanup would destroy
information that the next run depends on. Always clear asset store explicitly
when it is no longer needed.
diff --git a/airflow-core/docs/core-concepts/index.rst
b/airflow-core/docs/core-concepts/index.rst
index eff828881cc..5e53609d8d8 100644
--- a/airflow-core/docs/core-concepts/index.rst
+++ b/airflow-core/docs/core-concepts/index.rst
@@ -55,6 +55,15 @@ Here you can find detailed documentation about each one of
the core concepts of
variables
params
+**Task and Asset Store**
+
+.. toctree::
+ :maxdepth: 2
+
+ task-and-asset-store
+ task-store
+ asset-store
+
**Debugging**
.. toctree::
diff --git a/airflow-core/docs/core-concepts/task-and-asset-store.rst
b/airflow-core/docs/core-concepts/task-and-asset-store.rst
new file mode 100644
index 00000000000..41c99d5bdb7
--- /dev/null
+++ b/airflow-core/docs/core-concepts/task-and-asset-store.rst
@@ -0,0 +1,80 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _concepts:task-and-asset-store-overview:
+
+Task and Asset Store Overview
+=============================
+
+.. versionadded:: 3.3
+
+Airflow has always modeled tasks as stateless, idempotent units of work. A
growing class of workloads, however, require some amount of data to be
persisted outside of a task's return value, like a submitted job ID that must
survive a worker crash, a watermark that advances run-by-run, or a row counter
exposed for observability. Task store and Asset store fill that gap without
touching the XCom or Variable systems.
+
+Task and Asset Store
+--------------------
+
+Task and Asset store provide two key/value stores to persist data like a job
ID, watermark, or row count. These two stores are differentiated by *what* they
are scoped to:
+
+.. list-table::
+ :header-rows: 1
+ :widths: 20 25 25 30
+
+ * - Store
+ - Scope
+ - Default lifetime
+ - Primary use case
+ * - **Task store**
+ - A single task Instance (dag_id + run_id + task_id + map_index)
+ - Configurable retention; cleared on task success when ``clear_on_success
= True``
+ - Survive retries, track in-flight jobs, checkpoint progress within a
run, resume progress from checkpoint set by a past run
+ * - **Asset store**
+ - An asset (independent of any particular run)
+ - Persists indefinitely; removed only when the asset is deactivated
+ - Cross-run watermarks, incremental-load cursors, per-asset metadata
+
+Both stores accept JSON-able values. Values can be stored using the default
metastore backend, or be offloaded via a :ref:`custom worker-side backend
<task-and-asset-store:worker-backends>`.
+
+When to use Task and Asset Store
+--------------------------------
+
+Use this table to choose the right mechanism for your use case.
+
+.. list-table::
+ :header-rows: 1
+ :widths: 22 78
+
+ * - Mechanism
+ - When to use it
+ * - **XCom**
+ - Pass data *between tasks* within a single Dag run (e.g. the output of
one task consumed by a downstream task) or across different multiple Dag runs
(referencing the data persisted from another run). XComs are cleared on retry,
and should NOT be used to persist data across task retries or across runs.
+ * - **Variables**
+ - Deployment-wide or installation-wide configuration that changes
infrequently and is set by operators rather than by tasks themselves.
+ * - **Task store**
+ - Data that must survive a worker crash or data that must survive across
retries within the **same run**. An external job ID written before a
long-running job completes is a perfect use case for task store.
+ * - **Asset store**
+ - Data that must persist **across asset events** or while asset
"watching" and is logically owned by an asset rather than a task. For example,
a watermark that advances each time a file lands in an object store.
+
+.. note::
+
+ If your current implementation already leverages an XCom-based pattern
successfully, there's **no need to migrate** to task store. Task store is meant
to solve problems that XCom was never designed for.
+
+Further reading
+---------------
+
+* :doc:`Task Store <task-store>`: full API reference and use-case examples
+* :doc:`Asset Store </core-concepts/asset-store>`: watermark pattern and API
reference
+* :doc:`Task and asset Store Configuration
</administration-and-deployment/task-and-asset-store>`: retention, GC, and
custom backends
diff --git a/airflow-core/docs/core-concepts/task-store.rst
b/airflow-core/docs/core-concepts/task-store.rst
new file mode 100644
index 00000000000..87e2e12dcf4
--- /dev/null
+++ b/airflow-core/docs/core-concepts/task-store.rst
@@ -0,0 +1,278 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ .. http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+.. _concepts:task-store:
+
+.. spelling:word-list::
+
+ intra
+ Intra
+ checkpointing
+
+Task Store
+==========
+
+.. versionadded:: 3.3
+
+Task store is a persistent key/value store scoped to a single task instance
(``dag_id`` + ``run_id`` + ``task_id`` + ``map_index``). It survives worker
crashes and task retries within the same Dag run, making it suitable for
storing external job IDs, intra-task checkpoints, and progress metadata.
+
+Data persisted via task store is accessed through the task context via
``context["task_store"]`` and exposes four methods: ``get``, ``set``,
``delete``, and ``clear``.
+
+
+Accessing task store
+--------------------
+
+Inside any ``@task``-decorated function or ``BaseOperator.execute()`` method,
task store is available through the ``context`` dictionary via the
``task_store`` key. From there, it can be used to retrieve, set, delete, or
clear data for a specific key-value pair. In this example, the ``job_id`` is
retrieved from task store, then updated, before being deleted. All data for
that task is then removed using the ``clear`` method.
+
+.. code-block:: python
+
+ from airflow.sdk import task
+ import random
+
+
+ @task
+ def my_task(**context):
+ # Retrieve task_store from context
+ task_store = context["task_store"]
+ my_value = task_store.get("my_key", default="my_default_key")
+
+ # Set the new value
+ new_value = f"It is {random.randint(1, 12 + 1)} o'clock"
+ task_store.set("my_key", new_value)
+
+ # Delete the value
+ task_store.delete("my_key")
+
+ # Clear all store entries for the task
+ task_store.clear()
+
+Reference
+---------
+
+``get(key, default)``
+~~~~~~~~~~~~~~~~~~~~~
+
+Returns the stored JSON value, or the ``default`` value if the key does not
exist.
+
+.. code-block:: python
+
+ value = task_store.get(
+ "job_id", default="123456789"
+ ) # returns the value associated with `job_id` or the default value
+
+``set(key, value, *, retention=None)``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Writes or overwrites a value for the specified key. Note, ``value`` can be any
JSON-compatible type, except for ``None``. This includes:
+
+* ``str``
+* ``int``
+* ``float``
+* ``bool``
+* ``list``
+* ``dict``
+
+The optional ``retention`` argument controls when the key expires:
+
+* ``timedelta(...)``: expire after the given duration from the time of the
write (e.g. ``timedelta(hours=6)``). The expiry timestamp is computed on the
worker before the value is sent to the API server.
+* ``NEVER_EXPIRE``: the key never expires and is skipped during garbage
collection, regardless of the global ``[state_store] default_retention_days``
setting.
+* ``None`` (default): fall back to the global ``[state_store]
default_retention_days`` config.
+
+.. important::
+
+ ``retention`` accepts only a :class:`~datetime.timedelta`, not a plain
integer number of days. Passing an integer raises a ``TypeError``.
+
+ .. code-block:: python
+
+ # correct
+ task_store.set("key", "val", retention=timedelta(days=7))
+
+ # wrong — raises TypeError
+ task_store.set("key", "val", retention=7)
+
+``NEVER_EXPIRE`` sentinel
+^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Import ``NEVER_EXPIRE`` from ``airflow.sdk``:
+
+.. code-block:: python
+
+ from airflow.sdk import NEVER_EXPIRE
+
+ task_store.set("job_id", job_id, retention=NEVER_EXPIRE)
+
+``delete(key)``
+~~~~~~~~~~~~~~~
+
+Deletes a single key. No-op if the key does not exist.
+
+.. code-block:: python
+
+ task_store.delete("job_id")
+
+``clear(all_map_indices=False)``
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Deletes *all* task store keys for this task instance.
+
+For :doc:`mapped tasks </authoring-and-scheduling/dynamic-task-mapping>`, the
default clears only the current map index. Pass ``all_map_indices=True`` to
wipe the store across **every** mapped instance of the task (fleet-wide reset).
+
+.. code-block:: python
+
+ # clear only this map index
+ task_store.clear()
+
+ # clear all map indices (fleet-wide)
+ task_store.clear(all_map_indices=True)
+
+
+Some Example Use Cases
+----------------------
+
+External job resumption
+~~~~~~~~~~~~~~~~~~~~~~~
+
+A common pattern for long-running external jobs: check whether a job ID is
already stored before submitting, and use ``NEVER_EXPIRE`` so the key outlives
+the default retention window.
+
+.. code-block:: python
+
+ from datetime import timedelta
+
+ from airflow.sdk import DAG, task
+ from airflow.sdk import NEVER_EXPIRE
+
+ with DAG("spark_job_dag", schedule=None):
+
+ @task
+ def run_spark_job(**context):
+ task_store = context["task_store"]
+
+ # Check for an already-submitted job from a previous attempt.
+ job_id = task_store.get("job_id")
+ if job_id is None:
+ job_id = spark_client.submit_job(...)
+ # Store with NEVER_EXPIRE so the key is not garbage-collected
before the job finishes
+ task_store.set("job_id", job_id, retention=NEVER_EXPIRE)
+
+ # Reattach to the job and wait for completion.
+ result = spark_client.wait_for_completion(job_id)
+ return result
+
+On a retry, the task finds the stored ``job_id`` and reattaches instead of
submitting a duplicate job. Another example of this sort of logic can be found
in `example_task_store.py
<https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/example_dags/example_task_store.py>`_.
+
+For ``BaseOperator`` subclasses, the
:class:`~airflow.sdk.bases.resumablemixin.ResumableJobMixin` encapsulates this
pattern. It persists the external job ID to task store after submission and, on
retry, reconnects to an active job or resubmits if the prior job reached a
terminal failure state.
+
+Intra-task checkpointing
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+For tasks that process paginated or batched data, store the last-completed
offset so a retry can resume mid-stream rather than restarting from the
beginning.
+
+.. code-block:: python
+
+ from airflow.sdk import DAG, task
+
+ with DAG("paginated_ingest", schedule="@daily"):
+
+ @task
+ def ingest_pages(**context):
+ # Retrieve the task_store
+ task_store = context["task_store"]
+ raw = task_store.get("last_page")
+
+ start_page = raw + 1 if raw is not None else 1
+
+ for page in range(start_page, total_pages + 1):
+ fetch_and_load(page)
+ task_store.set("last_page", page) # Update the task_store for
reuse later
+
+
+On a retry, the task reads ``last_page`` and skips pages that were already
processed.
+
+Progress metadata
+~~~~~~~~~~~~~~~~~
+
+Task store can expose in-progress metrics for observability — row counts,
status strings, or lightweight JSON payloads — without requiring XCom or an
external system.
+
+.. code-block:: python
+
+
+ from airflow.sdk import DAG, task
+
+ with DAG("row_ingest", schedule="@hourly"):
+
+ @task
+ def ingest_rows(**context):
+ task_store = context["task_store"]
+ total = 0
+
+ for batch in get_batches():
+ load(batch)
+ total += len(batch)
+ task_store.set(
+ "progress",
+ {"rows_loaded": total, "status": "running"},
+ )
+
+ task_store.set(
+ "progress",
+ {"rows_loaded": total, "status": "done"},
+ )
+
+The ``progress`` key is visible through the REST API and the Airflow UI while
the task is running.
+
+
+Sync vs. deferrable tasks
+-------------------------
+
+Task store behaves slightly differently depending on whether a task runs
synchronously or uses the deferral mechanism.
+
+Synchronous tasks
+~~~~~~~~~~~~~~~~~
+
+If the worker process crashes, the task instance is retried. Task store data
written before the crash is preserved, so the retry can pick up where the
previous attempt left off (see the `External job resumption`_ pattern above).
+
+Deferrable tasks
+~~~~~~~~~~~~~~~~
+
+Once a task defers, the Triggerer handles continuity across poke cycles. Use
task store in deferrable tasks only when you need to survive an
operator-initiated clear, not for normal poke continuity.
+
+
+Mapped tasks
+------------
+
+When a task is dynamically mapped (``task.expand(...)``), each map index has
its own task store namespace. ``clear()`` without arguments clears the store
only for the current index. ``clear(all_map_indices=True)`` wipes the store
across every index of the task.
+
+.. code-block:: python
+
+ # Inside a mapped task — clear only this index
+ task_store.clear()
+
+ # Wipe store for all indices of this task
+ task_store.clear(all_map_indices=True)
+
+
+Automatic cleanup (``clear_on_success``)
+----------------------------------------
+
+When ``[state_store] clear_on_success = True``, all task store keys for a task
instance are automatically deleted when the task moves to the ``success``
state. This is useful for reducing storage when post-success observability is
not needed.
+
+.. note::
+
+ ``clear_on_success`` clears **task store only**. Asset store is scoped to
the asset, not the task instance, and is never affected by this setting. Asset
store persists across runs and must be cleared explicitly.
+
+See :doc:`/administration-and-deployment/task-and-asset-store` for full
configuration details.
