Airflow s3 xcom backend - Show the xcom values in airflow UI than the s3 path for the xcom

[bparamjeet]

Description

Currently while using s3 xcom backend , the result is stored in s3 path as expected but while viewing the xcom from airflow UI , the value shown is the s3 path but it would be great if we have the actual xcom value in the place.

Use case/motivation

Usecase : It will be helpful for developers to view the xcom

Related issues

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[potiuk]

Not sure which implementation you use - but xcom backend has orm_deserialize_value method https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/models/xcom/index.html#airflow.models.xcom.BaseXCom.orm_deserialize_value that is used to produce the representation of the xcom shown in webserver. You should likely override it in the way that it shows values for small varlues (if you can make quick decision based on your knowledge about Xcom) - or just indication that the value is too big if it is - (this makes very little sense and it will crash your webserver if you display very long xcom value in the webserver).

[potiuk]

Converting to discussion.

[mamdouhtawfik]

@potiuk I have the same use case, on airflow v2.8.1, when I override orm_deserialize_value I get a sqlalchemy error when I try to view xcom value, either in the Admin menu or the task specific tab, stack trace below:
r/xcomEntries [GET] Traceback (most recent call last): File "/usr/local/airflow/.local/lib/python3.11/site-packages/flask/app.py", line 2529, in wsgi_app response = self.full_dispatch_request() ^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/flask/app.py", line 1825, in full_dispatch_request rv = self.handle_user_exception(e) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/flask/app.py", line 1823, in full_dispatch_request rv = self.dispatch_request() ^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/flask/app.py", line 1799, in dispatch_request return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/connexion/decorators/decorator.py", line 68, in wrapper response = function(request) ^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/connexion/decorators/uri_parsing.py", line 149, in wrapper response = function(request) ^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/connexion/decorators/validation.py", line 399, in wrapper return function(request) ^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/connexion/decorators/response.py", line 112, in wrapper response = function(request) ^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/connexion/decorators/parameter.py", line 120, in wrapper return function(**kwargs) ^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/api_connexion/security.py", line 164, in decorated return _requires_access( ^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/api_connexion/security.py", line 92, in _requires_access return func(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/api_connexion/parameters.py", line 104, in wrapped_function return func(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/utils/session.py", line 79, in wrapper return func(*args, session=session, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/airflow/api_connexion/endpoints/xcom_endpoint.py", line 77, in get_xcom_entries return xcom_collection_schema.dump(XComCollection(xcom_entries=query, total_entries=total_entries)) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/marshmallow/schema.py", line 550, in dump result = self._serialize(processed_obj, many=many) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/marshmallow/schema.py", line 518, in _serialize value = field_obj.serialize(attr_name, obj, accessor=self.get_attribute) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/marshmallow/fields.py", line 342, in serialize return self._serialize(value, attr, obj, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/marshmallow/fields.py", line 776, in _serialize return [self.inner._serialize(each, attr, obj, **kwargs) for each in value] ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/marshmallow/fields.py", line 776, in <listcomp> return [self.inner._serialize(each, attr, obj, **kwargs) for each in value] ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/sqlalchemy/engine/result.py", line 382, in iterrows for row in self._fetchiter_impl(): File "/usr/local/airflow/.local/lib/python3.11/site-packages/sqlalchemy/orm/loading.py", line 151, in chunks rows = [proc(row) for row in fetch] ^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/sqlalchemy/orm/loading.py", line 151, in <listcomp> rows = [proc(row) for row in fetch] ^^^^^^^^^ File "/usr/local/airflow/.local/lib/python3.11/site-packages/sqlalchemy/orm/loading.py", line 940, in _instance session_identity_map._add_unpresent(state, identitykey) File "/usr/local/airflow/.local/lib/python3.11/site-packages/sqlalchemy/orm/identity.py", line 249, in _killed raise sa_exc.InvalidRequestError( sqlalchemy.exc.InvalidRequestError: Object <CustomXComS3Backend at 0x7fef84ab2350> cannot be converted to 'persistent' state, as this identity map is no longer valid. Has the owning Session been closed? (Background on this error at: https://sqlalche.me/e/14/lkrp)
I tried multiple approach to avoid this, nothing worked, once I comment orm_deserialize_value the error goes away, yet of course I don't get to see the xcom values in the UI. Any ideas on how to resolve this?

Side note: Testing such backend takes a long time as I don't have an easy way to restart the mwaa-local-runner other than closing and starting docker compose again.