ashb commented on code in PR #46319:
URL: https://github.com/apache/airflow/pull/46319#discussion_r1937507968
##########
airflow/api_fastapi/execution_api/routes/xcoms.py:
##########
@@ -152,14 +170,30 @@ def set_xcom(
},
)
- if not has_xcom_access(key, token):
- raise HTTPException(
- status_code=status.HTTP_403_FORBIDDEN,
- detail={
- "reason": "access_denied",
- "message": f"Task does not have access to set XCom key
'{key}'",
- },
+ if mapped_length is not None:
+ task_map = TaskMap(
+ dag_id=dag_id,
+ task_id=task_id,
+ run_id=run_id,
+ map_index=map_index,
+ length=mapped_length,
+ keys=None,
)
+ max_map_length = conf.getint("core", "max_map_length", fallback=1024)
+ if task_map.length > max_map_length:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail={
+ "reason": "unmappable_return_value_length",
+ "message": "pushed value is too large to map as a
downstream's dependency",
+ },
+ )
+ session.add(task_map)
+
+ # else:
+ # TODO: Can/should we check if a client _hasn't_ provided this for an
upstream of a mapped task? That
+ # means loading the serialized dag and that seems like a relatively costly
operation for minimal benefit
+ # (the mapped task would fail in a moment as it can't be expanded anyway.)
Review Comment:
One thing we might be able to do is to restructure the serdag format to be
able to efficiently find this info via JSON queries _in_ the field in the DB,
meaning we could query for certain facets directly without having to load the
full DAG into python memory
--
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-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
