Skip to content

Commit 5149013

Browse files
committed
fixup! Give every mapped expansion a region and store its index as region_index
The state store stores the region index, and the object path is built from it. For an ordinary or mapped task that equals the public map index, but for an unmapped task inside a loop it is the iteration, so documenting the path component as map_index would tell a custom backend author that loop passes never appear in the path. Name it region_index in the state store docs, the common.io backend docs, the provider.yaml description and the generated provider info.
1 parent 6e36d1d commit 5149013

5 files changed

Lines changed: 6 additions & 5 deletions

File tree

‎airflow-core/docs/core-concepts/task-and-asset-state-store.rst‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,7 @@ Task and Asset state store provide two key/value stores to persist data like a j
3838
- Default lifetime
3939
- Primary use case
4040
* - **Task state store**
41-
- A single task Instance (``dag_id`` + ``run_id`` + ``task_id`` + ``region_id`` + ``map_index``)
41+
- A single task Instance (``dag_id`` + ``run_id`` + ``task_id`` + ``region_id`` + ``region_index``)
4242
- Configurable retention; cleared on task success when ``clear_on_success = True``
4343
- Survive retries, track in-flight jobs, checkpoint progress within a run, resume progress from checkpoint set by a past run
4444
* - **Asset state store**

‎airflow-core/src/airflow/serialization/dag_version_diff.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -256,6 +256,7 @@
256256
"retries",
257257
"retry_delay",
258258
"retry_exponential_backoff",
259+
"returns_dag_result",
259260
"run_as_user",
260261
"start_date",
261262
"start_from_trigger",

‎providers/common/io/docs/state_store_backend.rst‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ section to ``airflow.providers.common.io.state_store.backend.StateStoreObjectSto
2727
``state_store_objectstorage_path`` to the desired base location. The connection id is obtained from the
2828
user part of the URL, e.g. ``state_store_objectstorage_path = s3://conn_id@mybucket/task-state/``.
2929

30-
Task state is stored under ``<dag_id>/<run_id>/[region=<region_id>/]<task_id>/<map_index>/<key>`` and asset state under
30+
Task state is stored under ``<dag_id>/<run_id>/[region=<region_id>/]<task_id>/<region_index>/<key>`` and asset state under
3131
``assets/<asset_identifier>/<key>`` beneath the configured base path.
3232

3333
By default (``state_store_objectstorage_threshold = 0``) all serialized values are offloaded to object storage.
@@ -76,6 +76,6 @@ Using the local filesystem (useful for development)::
7676

7777
.. note::
7878

79-
Task state paths are keyed on ``(dag_id, run_id, region_id, task_id, map_index)`` and are stable across
79+
Task state paths are keyed on ``(dag_id, run_id, region_id, task_id, region_index)`` and are stable across
8080
task retries. This makes this backend suitable for operators that use
8181
:class:`~airflow.sdk.ResumableJobMixin` to reconnect to external jobs after a retry.

‎providers/common/io/provider.yaml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -122,7 +122,7 @@ config:
122122
description: |
123123
Base path on object storage for the task/asset state store backend, in URL format.
124124
When set, StateStoreObjectStorageBackend will persist task and asset state under this
125-
prefix, organized as <dag_id>/<run_id>/[region=<region_id>/]<task_id>/<map_index>/<key>
125+
prefix, organized as <dag_id>/<run_id>/[region=<region_id>/]<task_id>/<region_index>/<key>
126126
for tasks and assets/<asset_identifier>/<key> for assets.
127127
version_added: 1.8.0
128128
type: string

‎providers/common/io/src/airflow/providers/common/io/get_provider_info.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,7 @@ def get_provider_info():
8585
"default": "",
8686
},
8787
"state_store_objectstorage_path": {
88-
"description": "Base path on object storage for the task/asset state store backend, in URL format.\nWhen set, StateStoreObjectStorageBackend will persist task and asset state under this\nprefix, organized as <dag_id>/<run_id>/[region=<region_id>/]<task_id>/<map_index>/<key>\nfor tasks and assets/<asset_identifier>/<key> for assets.\n",
88+
"description": "Base path on object storage for the task/asset state store backend, in URL format.\nWhen set, StateStoreObjectStorageBackend will persist task and asset state under this\nprefix, organized as <dag_id>/<run_id>/[region=<region_id>/]<task_id>/<region_index>/<key>\nfor tasks and assets/<asset_identifier>/<key> for assets.\n",
8989
"version_added": "1.8.0",
9090
"type": "string",
9191
"example": "s3://conn_id@bucket/task-state/",

0 commit comments

Comments
 (0)