You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "amoghrajesh (via GitHub)" <gi...@apache.org> on 2026/06/06 10:21:49 UTC
[PR] Make task and asset store row size limits configurable [airflow]
amoghrajesh opened a new pull request, #68133:
URL: https://github.com/apache/airflow/pull/68133
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
### What's being solved?
The task and asset store had a hardcoded 64 KB size limit that in core API datamodels that could not be adjusted without code changes. Users with legitimate large-payload use case had no way to raise it, and the limit wasn't enforced consistently too, the core API validated it but the execution API (worker path) did not.
### Current behaviour
- Core API rejected values over 64 KB with a hardcoded limit.
- Execution API (worker writes) had no size check at all.
- No config key existed to adjust the limit.
### Proposed change
Adds `[state_store] max_value_storage_bytes` (default `65535`) to control the limit:
- **Core API** (task and asset store): rejects oversized values with HTTP 400. Set to `0` to disable the limit entirely. DB column limits then apply (~1 GB on PostgreSQL, 16 MB on MySQL).
- **Task SDK** (`task_store.set()` and `asset_store.set()`): logs a warning in task logs when the serialized value exceeds the limit, but allows the write through to not interrupt the execution mid run. The warning suggests configuring a custom `[state_store] backend` for large payloads.
### Testing
Running example:
Task Store:
```python
@task
def my_task(**context: Context):
task_state = context["task_store"]
value = "x" * 700000
task_state.set("str_value", value)
```
<img width="2555" height="903" alt="image" src="https://github.com/user-attachments/assets/737f21ca-55fb-4501-8dcd-772142270130" />
Asset store:
```python
with DAG(
dag_id="aip103_asset_producer",
start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
schedule=None,
catchup=False,
tags=["aip-103", "asset-state-test"],
):
@task(outlets=[aip103_test_asset])
def produce(**context):
print(f"Producer running for {aip103_test_asset.uri!r}")
w = "w" * 700000
context["asset_store"].set("my_asset_value_watermark", w)
produce()
```
<img width="2555" height="903" alt="image" src="https://github.com/user-attachments/assets/a0c25408-b9f9-482f-926b-e6ba2503e0ad" />
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3393382809
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_store.py:
##########
@@ -66,8 +65,12 @@ def value_is_json_representable(cls, v: JsonValue) -> JsonValue:
serialized = json.dumps(v, allow_nan=False)
except ValueError:
raise ValueError("value contains non-finite numbers; NaN and Inf are not JSON representable")
- if len(serialized) > _MAX_SERIALIZED_BYTES:
- raise ValueError(f"value exceeds maximum serialized size of {_MAX_SERIALIZED_BYTES} bytes")
+ limit = conf.getint("state_store", "max_value_storage_bytes")
Review Comment:
Kept it inline for now — the two checks have slightly different error messages and the logic is a single if-block each, so a shared helper would add indirection without much payoff. If a third call site appears later it's worth revisiting.
---
Drafted-by: Claude Code (Sonnet 4.6) (no human review before posting)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3393449711
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_store.py:
##########
@@ -305,6 +305,12 @@ def test_put_overwrites_expiry_on_existing_key(self, test_client, time_machine):
assert resp["value"] == "v2"
assert resp["expires_at"] is None
+ def test_put_value_over_limit_returns_422(self, test_client):
+ # default is 65536 bytes
Review Comment:
Fixed it. corrected the off-by-one (65536 -> 65535) and added a conf_vars parametrize case that sets `max_value_storage_bytes=0` and confirms a large value is accepted with 204.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3411640182
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -724,6 +739,18 @@ def set(self, key: str, value: JsonValue) -> None:
ref = backend.serialize_asset_store_to_ref(value=value, key=key, scope=scope)
stored = _wrap_external_ref(ref)
+ limit = conf.getint("state_store", "max_value_storage_bytes")
+ if limit > 0:
Review Comment:
Sure handled.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh merged PR #68133:
URL: https://github.com/apache/airflow/pull/68133
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3392246837
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3239,6 +3239,21 @@ state_store:
type: integer
example: "10000"
default: "0"
+ max_value_storage_bytes:
+ description: |
+ Only applicable to MetastoreStoreBackend. Maximum size in bytes that a single task or asset store
+ value written via the core or execution API can have. Values that exceed this limit are rejected
Review Comment:
The first paragraph says values written via the core or execution API are rejected at the API boundary, but the execution API models don't have a size check (the next paragraph says the worker write is allowed through). Worth rewording this to core API only. The description also doesn't mention that `0` disables the limit, and the last line has a typo, "use a custom state backends".
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_store.py:
##########
@@ -66,8 +65,12 @@ def value_is_json_representable(cls, v: JsonValue) -> JsonValue:
serialized = json.dumps(v, allow_nan=False)
except ValueError:
raise ValueError("value contains non-finite numbers; NaN and Inf are not JSON representable")
- if len(serialized) > _MAX_SERIALIZED_BYTES:
- raise ValueError(f"value exceeds maximum serialized size of {_MAX_SERIALIZED_BYTES} bytes")
+ limit = conf.getint("state_store", "max_value_storage_bytes")
+ if limit > 0 and len(serialized) > limit:
+ raise ValueError(
+ f"value exceeds max_value_storage_bytes ({limit}); "
+ "for large payloads configure a custom [state_store] backend"
Review Comment:
Following this advice won't get past the 422: the validator runs regardless of which `[state_store] backend` is configured, so a REST caller's only remedy is raising `max_value_storage_bytes` (or setting it to 0). That also makes the config description's "Only applicable to MetastoreStoreBackend" inaccurate. Should the check be skipped when a custom backend is configured, or should the message just point at the config option instead?
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -600,7 +601,21 @@ def set(self, key: str, value: JsonValue, *, retention: timedelta | None = None)
# wrap the value with a marker to indicate that it's stored externally, and include the ref to the external storage
stored = _wrap_external_ref(ref)
- SUPERVISOR_COMMS.send(SetTaskStore(ti_id=self._ti_id, key=key, value=stored, expires_at=expires_at))
+ msg = SetTaskStore(ti_id=self._ti_id, key=key, value=stored, expires_at=expires_at)
+
+ limit = conf.getint("state_store", "max_value_storage_bytes")
+ if limit > 0:
+ serialized_size = len(json.dumps(stored))
+ if serialized_size > limit:
+ log.warning(
+ "Task store value for key %r is %d bytes, which exceeds configured max_value_storage_bytes=%d. "
+ "Consider using a custom [state_store] backend to offload large payloads.",
Review Comment:
`[state_store] backend` is the server-side backend and doesn't help here. The backend that offloads large payloads from a task is `[workers] state_store_backend`, which is also what stops this warning firing since `stored` becomes a small ref. Same section name in the asset store warning below and in the API error messages.
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_store.py:
##########
@@ -305,6 +305,12 @@ def test_put_overwrites_expiry_on_existing_key(self, test_client, time_machine):
assert resp["value"] == "v2"
assert resp["expires_at"] is None
+ def test_put_value_over_limit_returns_422(self, test_client):
+ # default is 65536 bytes
Review Comment:
Default is 65535, not 65536. Also, all the API tests cover only the default limit; a `conf_vars` case (e.g. `max_value_storage_bytes=0` accepting a large value) would actually exercise the new configurability and the `limit > 0` gating.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3393437502
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_store.py:
##########
@@ -66,8 +65,12 @@ def value_is_json_representable(cls, v: JsonValue) -> JsonValue:
serialized = json.dumps(v, allow_nan=False)
except ValueError:
raise ValueError("value contains non-finite numbers; NaN and Inf are not JSON representable")
- if len(serialized) > _MAX_SERIALIZED_BYTES:
- raise ValueError(f"value exceeds maximum serialized size of {_MAX_SERIALIZED_BYTES} bytes")
+ limit = conf.getint("state_store", "max_value_storage_bytes")
+ if limit > 0 and len(serialized) > limit:
+ raise ValueError(
+ f"value exceeds max_value_storage_bytes ({limit}); "
+ "for large payloads configure a custom [state_store] backend"
Review Comment:
Fixed, the error message now says "raise [state_store] max_value_storage_bytes or set it to 0 to disable the limit". Same change applied to `TaskStorePatchBody` and `AssetStoreBody`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "jroachgolf84 (via GitHub)" <gi...@apache.org>.
jroachgolf84 commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3369676776
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_store.py:
##########
@@ -66,8 +65,12 @@ def value_is_json_representable(cls, v: JsonValue) -> JsonValue:
serialized = json.dumps(v, allow_nan=False)
except ValueError:
raise ValueError("value contains non-finite numbers; NaN and Inf are not JSON representable")
- if len(serialized) > _MAX_SERIALIZED_BYTES:
- raise ValueError(f"value exceeds maximum serialized size of {_MAX_SERIALIZED_BYTES} bytes")
+ limit = conf.getint("state_store", "max_value_storage_bytes")
Review Comment:
We don't want to throw this in a `validate_payload_size` function or something to reuse like that, correct? I could be way off, but thought I'd throw it out there.
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/asset_store.py:
##########
@@ -67,6 +66,10 @@ def value_is_json_representable(cls, v: JsonValue) -> JsonValue:
serialized = json.dumps(v, allow_nan=False)
except ValueError:
raise ValueError("value contains non-finite numbers; NaN and Inf are not JSON representable")
- if len(serialized) > _MAX_SERIALIZED_BYTES:
- raise ValueError(f"value exceeds maximum serialized size of {_MAX_SERIALIZED_BYTES} bytes")
+ limit = conf.getint("state_store", "max_value_storage_bytes")
+ if limit > 0 and len(serialized) > limit:
+ raise ValueError(
+ f"value exceeds max_value_storage_bytes ({limit}); "
+ "for large payloads configure a custom [state_store] backend"
+ )
Review Comment:
I'd lean towards putting a limit on this. I think we want to use it as a "state store" rather than a "data store". I'd be worried about folks using as a place to dump huge amounts of data, which I don't think is the intention.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3400606504
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3239,6 +3239,21 @@ state_store:
type: integer
example: "10000"
default: "0"
+ max_value_storage_bytes:
+ description: |
+ Only applicable to MetastoreStoreBackend. Maximum size in bytes that a single task or asset store
+ value written via the core or execution API can have. Values that exceed this limit are rejected
+ at the API boundary.
+
+ Workers writing via the execution API log a warning to use custom backends when this limit
+ is exceeded but the write is allowed to avoid interrupting a task mid execution.
+
+ The default of 65535 bytes (64 KB) is appropriate for coordination state values such as
+ job IDs, cursors, and small status maps. For larger payloads, use a custom state backends.
+ version_added: 3.3.0
+ type: integer
+ example: "1048576"
+ default: "65535"
Review Comment:
Handled in [comments from kaxil](https://github.com/apache/airflow/pull/68133/commits/4adbfb1701aaf1d7618e06cfdc92d6f6e4e7ec83)
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_store.py:
##########
@@ -305,6 +305,12 @@ def test_put_overwrites_expiry_on_existing_key(self, test_client, time_machine):
assert resp["value"] == "v2"
assert resp["expires_at"] is None
+ def test_put_value_over_limit_returns_422(self, test_client):
+ # default is 65536 bytes
Review Comment:
Handled in [comments from kaxil](https://github.com/apache/airflow/pull/68133/commits/4adbfb1701aaf1d7618e06cfdc92d6f6e4e7ec83)
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -600,7 +601,21 @@ def set(self, key: str, value: JsonValue, *, retention: timedelta | None = None)
# wrap the value with a marker to indicate that it's stored externally, and include the ref to the external storage
stored = _wrap_external_ref(ref)
- SUPERVISOR_COMMS.send(SetTaskStore(ti_id=self._ti_id, key=key, value=stored, expires_at=expires_at))
+ msg = SetTaskStore(ti_id=self._ti_id, key=key, value=stored, expires_at=expires_at)
+
+ limit = conf.getint("state_store", "max_value_storage_bytes")
+ if limit > 0:
+ serialized_size = len(json.dumps(stored))
+ if serialized_size > limit:
+ log.warning(
+ "Task store value for key %r is %d bytes, which exceeds configured max_value_storage_bytes=%d. "
+ "Consider using a custom [state_store] backend to offload large payloads.",
Review Comment:
Handled in [comments from kaxil](https://github.com/apache/airflow/pull/68133/commits/4adbfb1701aaf1d7618e06cfdc92d6f6e4e7ec83)
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_store.py:
##########
@@ -66,8 +65,12 @@ def value_is_json_representable(cls, v: JsonValue) -> JsonValue:
serialized = json.dumps(v, allow_nan=False)
except ValueError:
raise ValueError("value contains non-finite numbers; NaN and Inf are not JSON representable")
- if len(serialized) > _MAX_SERIALIZED_BYTES:
- raise ValueError(f"value exceeds maximum serialized size of {_MAX_SERIALIZED_BYTES} bytes")
+ limit = conf.getint("state_store", "max_value_storage_bytes")
+ if limit > 0 and len(serialized) > limit:
+ raise ValueError(
+ f"value exceeds max_value_storage_bytes ({limit}); "
+ "for large payloads configure a custom [state_store] backend"
Review Comment:
Handled in [comments from kaxil](https://github.com/apache/airflow/pull/68133/commits/4adbfb1701aaf1d7618e06cfdc92d6f6e4e7ec83)
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3239,6 +3239,21 @@ state_store:
type: integer
example: "10000"
default: "0"
+ max_value_storage_bytes:
+ description: |
+ Only applicable to MetastoreStoreBackend. Maximum size in bytes that a single task or asset store
+ value written via the core or execution API can have. Values that exceed this limit are rejected
Review Comment:
Handled in [comments from kaxil](https://github.com/apache/airflow/pull/68133/commits/4adbfb1701aaf1d7618e06cfdc92d6f6e4e7ec83)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3368872148
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/asset_store.py:
##########
@@ -67,6 +66,10 @@ def value_is_json_representable(cls, v: JsonValue) -> JsonValue:
serialized = json.dumps(v, allow_nan=False)
except ValueError:
raise ValueError("value contains non-finite numbers; NaN and Inf are not JSON representable")
- if len(serialized) > _MAX_SERIALIZED_BYTES:
- raise ValueError(f"value exceeds maximum serialized size of {_MAX_SERIALIZED_BYTES} bytes")
+ limit = conf.getint("state_store", "max_value_storage_bytes")
+ if limit > 0 and len(serialized) > limit:
+ raise ValueError(
+ f"value exceeds max_value_storage_bytes ({limit}); "
+ "for large payloads configure a custom [state_store] backend"
+ )
Review Comment:
xcom does not enforce a size limit in the model, and the ObjectStorage XCom backend has `xcom_objectstorage_threshold` which tiers large values to object storage rather than rejecting them.
For task state, we have a clearer use case than xcom did at design time itself which is coordination state (job IDs, cursors, retry counts), not data payloads. A configurable default cap lets us nudge operators toward a custom backend before they accidentally bombard the DB with large values which is something xcom learned from retroactively. The `65535` default can be raised as needed via [state_store] max_value_storage_bytes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3393427046
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3239,6 +3239,21 @@ state_store:
type: integer
example: "10000"
default: "0"
+ max_value_storage_bytes:
+ description: |
+ Only applicable to MetastoreStoreBackend. Maximum size in bytes that a single task or asset store
+ value written via the core or execution API can have. Values that exceed this limit are rejected
Review Comment:
Fixed — description now says core REST API only (422 on reject), notes that execution API writes are not blocked, adds "Set to 0 to disable the limit", and points at `[workers] state_store_backend` as the remedy for large payloads.
---
Drafted-by: Claude Code (Sonnet 4.6) (no human review before posting)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3393457820
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3239,6 +3239,21 @@ state_store:
type: integer
example: "10000"
default: "0"
+ max_value_storage_bytes:
+ description: |
+ Only applicable to MetastoreStoreBackend. Maximum size in bytes that a single task or asset store
+ value written via the core or execution API can have. Values that exceed this limit are rejected
+ at the API boundary.
+
+ Workers writing via the execution API log a warning to use custom backends when this limit
+ is exceeded but the write is allowed to avoid interrupting a task mid execution.
+
+ The default of 65535 bytes (64 KB) is appropriate for coordination state values such as
+ job IDs, cursors, and small status maps. For larger payloads, use a custom state backends.
+ version_added: 3.3.0
+ type: integer
+ example: "1048576"
+ default: "65535"
Review Comment:
There is no specific such reason, it just felt like the rightly suited size to use for such an intent to checkpoint with job id, cursor etc. Clarified to this now:
```
The default of 65535 bytes (64 KB) is a policy default suited for coordination state such as
job IDs, cursors, and small status maps — the underlying database column (MEDIUMTEXT on MySQL,
unbounded Text on Postgres) does not enforce this limit. For larger payloads, configure a custom
[workers] state_store_backend to offload values to external storage.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3393448370
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -600,7 +601,21 @@ def set(self, key: str, value: JsonValue, *, retention: timedelta | None = None)
# wrap the value with a marker to indicate that it's stored externally, and include the ref to the external storage
stored = _wrap_external_ref(ref)
- SUPERVISOR_COMMS.send(SetTaskStore(ti_id=self._ti_id, key=key, value=stored, expires_at=expires_at))
+ msg = SetTaskStore(ti_id=self._ti_id, key=key, value=stored, expires_at=expires_at)
+
+ limit = conf.getint("state_store", "max_value_storage_bytes")
+ if limit > 0:
+ serialized_size = len(json.dumps(stored))
+ if serialized_size > limit:
+ log.warning(
+ "Task store value for key %r is %d bytes, which exceeds configured max_value_storage_bytes=%d. "
+ "Consider using a custom [state_store] backend to offload large payloads.",
Review Comment:
Missed this cos we were rewording the config in a parallel PR, handled now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on PR #68133:
URL: https://github.com/apache/airflow/pull/68133#issuecomment-4666246371
@kaxil appreciate another round of review from you here? WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3368437280
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/asset_store.py:
##########
@@ -67,6 +66,10 @@ def value_is_json_representable(cls, v: JsonValue) -> JsonValue:
serialized = json.dumps(v, allow_nan=False)
except ValueError:
raise ValueError("value contains non-finite numbers; NaN and Inf are not JSON representable")
- if len(serialized) > _MAX_SERIALIZED_BYTES:
- raise ValueError(f"value exceeds maximum serialized size of {_MAX_SERIALIZED_BYTES} bytes")
+ limit = conf.getint("state_store", "max_value_storage_bytes")
+ if limit > 0 and len(serialized) > limit:
+ raise ValueError(
+ f"value exceeds max_value_storage_bytes ({limit}); "
+ "for large payloads configure a custom [state_store] backend"
+ )
Review Comment:
Why should we limit this? Do we do so for XCom or ObjectStorage?
Worth checking
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3393382809
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_store.py:
##########
@@ -66,8 +65,12 @@ def value_is_json_representable(cls, v: JsonValue) -> JsonValue:
serialized = json.dumps(v, allow_nan=False)
except ValueError:
raise ValueError("value contains non-finite numbers; NaN and Inf are not JSON representable")
- if len(serialized) > _MAX_SERIALIZED_BYTES:
- raise ValueError(f"value exceeds maximum serialized size of {_MAX_SERIALIZED_BYTES} bytes")
+ limit = conf.getint("state_store", "max_value_storage_bytes")
Review Comment:
Kept it inline for now — the two checks have slightly different error messages and the logic is a single if-block each, so a shared helper would add indirection without much payoff. If a third call site appears later it's worth revisiting.
---
Drafted-by: Claude Code (Sonnet 4.6) (no human review before posting)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3401039737
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
Review Comment:
Not-blocking, but I just realized that `task_store` and `asset_store` only recommend to store the metadata instead of large payload even with custom object store backend (e.g. we shouldn't store the model checkpoint in `task_store`).
Would it be better to introduce a "what not to store" section in `task-store.rst` as follow-up?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3393436982
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3239,6 +3239,21 @@ state_store:
type: integer
example: "10000"
default: "0"
+ max_value_storage_bytes:
+ description: |
+ Only applicable to MetastoreStoreBackend. Maximum size in bytes that a single task or asset store
+ value written via the core or execution API can have. Values that exceed this limit are rejected
Review Comment:
Sounds good, this was a bit outdated, handled it
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_store.py:
##########
@@ -66,8 +65,12 @@ def value_is_json_representable(cls, v: JsonValue) -> JsonValue:
serialized = json.dumps(v, allow_nan=False)
except ValueError:
raise ValueError("value contains non-finite numbers; NaN and Inf are not JSON representable")
- if len(serialized) > _MAX_SERIALIZED_BYTES:
- raise ValueError(f"value exceeds maximum serialized size of {_MAX_SERIALIZED_BYTES} bytes")
+ limit = conf.getint("state_store", "max_value_storage_bytes")
+ if limit > 0 and len(serialized) > limit:
+ raise ValueError(
+ f"value exceeds max_value_storage_bytes ({limit}); "
+ "for large payloads configure a custom [state_store] backend"
Review Comment:
Fixed — the error message now says "raise [state_store] max_value_storage_bytes or set it to 0 to disable the limit". Same change applied to `TaskStorePatchBody` and `AssetStoreBody`.
---
Drafted-by: Claude Code (Sonnet 4.6) (no human review before posting)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3411648738
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
Review Comment:
Let me look it up and if needed, will take it in a follow up, thanks. We have a comparison table as of now but maybe its ok to call it out too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3392279292
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3239,6 +3239,21 @@ state_store:
type: integer
example: "10000"
default: "0"
+ max_value_storage_bytes:
+ description: |
+ Only applicable to MetastoreStoreBackend. Maximum size in bytes that a single task or asset store
+ value written via the core or execution API can have. Values that exceed this limit are rejected
+ at the API boundary.
+
+ Workers writing via the execution API log a warning to use custom backends when this limit
+ is exceeded but the write is allowed to avoid interrupting a task mid execution.
+
+ The default of 65535 bytes (64 KB) is appropriate for coordination state values such as
+ job IDs, cursors, and small status maps. For larger payloads, use a custom state backends.
+ version_added: 3.3.0
+ type: integer
+ example: "1048576"
+ default: "65535"
Review Comment:
Is there a reason for 65535 specifically? The column is `Text().with_variant(MEDIUMTEXT, "mysql")`, so nothing DB-side enforces anything at 64 KB (MEDIUMTEXT holds 16 MB, Postgres `Text` is unbounded). 65535 is the MySQL plain-`TEXT` max, which this column does not use. Fine as a judgment call, but then the description should present it as a policy default rather than implying a storage limit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3393381776
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/task_store.py:
##########
@@ -66,8 +65,12 @@ def value_is_json_representable(cls, v: JsonValue) -> JsonValue:
serialized = json.dumps(v, allow_nan=False)
except ValueError:
raise ValueError("value contains non-finite numbers; NaN and Inf are not JSON representable")
- if len(serialized) > _MAX_SERIALIZED_BYTES:
- raise ValueError(f"value exceeds maximum serialized size of {_MAX_SERIALIZED_BYTES} bytes")
+ limit = conf.getint("state_store", "max_value_storage_bytes")
Review Comment:
The two validators have slightly different error messages and the duplication is two short if blocks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3401039737
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
Review Comment:
Not-blocking, but I just realized that `task_store` and `asset_store` only recommend to store the metadata instead of large payload even with custom object store backend e.g. the model checkpoint.
Would it be better to introduce a "what not to store" section in `task-store.rst` as follow-up?
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -724,6 +739,18 @@ def set(self, key: str, value: JsonValue) -> None:
ref = backend.serialize_asset_store_to_ref(value=value, key=key, scope=scope)
stored = _wrap_external_ref(ref)
+ limit = conf.getint("state_store", "max_value_storage_bytes")
+ if limit > 0:
Review Comment:
```suggestion
if ((limit := conf.getint("state_store", "max_value_storage_bytes")) > 0):
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make task and asset store row size limits configurable [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #68133:
URL: https://github.com/apache/airflow/pull/68133#discussion_r3393427046
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3239,6 +3239,21 @@ state_store:
type: integer
example: "10000"
default: "0"
+ max_value_storage_bytes:
+ description: |
+ Only applicable to MetastoreStoreBackend. Maximum size in bytes that a single task or asset store
+ value written via the core or execution API can have. Values that exceed this limit are rejected
Review Comment:
Fixed — description now says core REST API only (422 on reject), notes that execution API writes are not blocked, adds "Set to 0 to disable the limit", and points at `[workers] state_store_backend` as the remedy for large payloads.
---
Drafted-by: Claude Code (Sonnet 4.6) (no human review before posting)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org