XComObjectStorageBackend in Airflow 3.0.1 Returns File Paths Instead of Deserialized Objects in Mapped Tasks

[GeryDeKocliko]

Apache Airflow version

3.0.1

If "Other Airflow 2 version" selected, which one?

No response

What happened?

Hi,

After upgrading to Airflow 3.0.1, I observed that certain XCom values are no longer returned as deserialized Python objects. Instead, they are returned as file paths. This behavior differs from 3.0.0 versions, where XComs were automatically deserialized upon retrieval.

What you think should happen instead?

XComs should consistently return deserialized Python objects upon retrieval, maintaining the behavior observed in previous Airflow versions.

How to reproduce

Xcom Environment Configuration :

    AIRFLOW__CORE__XCOM_BACKEND: airflow.providers.common.io.xcom.backend.XComObjectStorageBackend
    AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH: file://airflow/xcoms
    AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD: 0

DAG :

from airflow.decorators import dag, task
from airflow.models.xcom import XCom

@task
def generate_data():
    print("XCom.__name__ --->", XCom.__name__)
    return [1, 2, 3]


@task
def filter_item_from_mapped_task(item):
    print("filter_item_from_mapped_task item --->", item)
    if item==1:
        return None
    return item


@task
def print_item(item):
    print("print_item item --->", item)


@dag(
    dag_id='test_bug_dag',
)
def test_bug_dag():
    data = generate_data()
    filtered_data = filter_item_from_mapped_task.expand(item=data)
    print_item.expand(item=filtered_data)

test_bug_dag()

Steps to Reproduce:
1. Deploy the above DAG in your Airflow environment configured with the specified XCom backend settings.
2. Trigger the DAG manually.
3. Observe the logs for each task execution.

Observed Behavior:
• Between generate_data and filter_item_from_mapped_task: The XCom value is deserialized as expected, and the item parameter receives the actual data (e.g., integers).
• Between filter_item_from_mapped_task and print_item: The item parameter receives a file URI instead of the deserialized object. For example: print_item item ---> file:///opt/airflow/airflow/xcoms/test_bug_dag/manual__2025-05-15T06:15:18.397306+00:00/filter_item_from_mapped_task/a1f9943d-b61b-4cae-9b25-527190b96567

Operating System

Mac OS 14.6

Versions of Apache Airflow Providers

apache-airflow-providers-common-io==1.5.4

Deployment

Docker-Compose

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

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

[GeryDeKocliko]

In the Airflow 3.0.1 release notes, there’s a single bullet point related to XComs: PR #49692.

Could this change be the source of the issue I’m experiencing, where XComs return file paths instead of deserialized objects in mapped tasks? @kaxil

[kaxil]

Unsure but we will verify.

cc @amoghrajesh @vatsrahul1001 Let's reproduce it and fix it if it is a regression in 3.0.1 over 3.0.0

[vatsrahul1001]

I can reproduce this with below config in 3.0.1

export AIRFLOW__CORE__XCOM_BACKEND=airflow.providers.common.io.xcom.backend.XComObjectStorageBackend
export AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_PATH=s3://aws_default@airflow-tutorial-data
export AIRFLOW__COMMON_IO__XCOM_OBJECTSTORAGE_THRESHOLD=0

In print_item task logs I see path is coming instead of xcom itself when using the xcom backend. This works fine if we do not use xcom backend.

I tried to repro this with 3.0.0, however, Dag was failing due to this issue in AF3 so not able to verify if this issue was already in 3.0.0 or not

[TJaniF]

@vatsrahul1001 This was not an issue in 3.0.0 as it was part of my webinar demo dag and worked fine until updated to 3.0.1

While looking into this I found another issue which might be related:

using explicit xcom pulling via context returns None for mapped tasks.

This dag:

from airflow.sdk import dag, task, chain


@dag
def custom_xcom_backend_test():

    @task
    def provide_map():
        return [1, 2, 3, 4]

    @task
    def push_xcom(map_num):
        print(map_num)
        return {f"key_{i}": i for i in range(1000)}

    _push_xcom = push_xcom.expand(map_num=provide_map())

    @task
    def pull_xcom(my_xcom):
        print(my_xcom)

        for i in my_xcom:
            print(i)

    @task
    def pull_xcom_explicit(**context):

        my_xcom_unmapped = context["ti"].xcom_pull(
            dag_id="custom_xcom_backend_test",
            task_ids=["provide_map"],
            key="return_value",
        )
        print("XCom from unmapped task:", my_xcom_unmapped)

        my_xcom = context["ti"].xcom_pull(
            dag_id="custom_xcom_backend_test",
            task_ids=["push_xcom"],
            key="return_value",
        )
        print("XCom from mapped task:", my_xcom)

    pull_xcom(_push_xcom)

    chain(
        _push_xcom,
        pull_xcom_explicit(),
    )


custom_xcom_backend_test()

Has the folllowing logs for pull_xcom_explicit

[2025-05-15, 19:10:45] INFO - XCom from unmapped task: [[1, 2, 3, 4]]: chan="stdout": source="task"
[2025-05-15, 19:10:45] INFO - XCom from mapped task: [None]: chan="stdout": source="task"
[2025-05-15, 19:10:45] INFO - Done. Returned value was: None: source="airflow.task.operators.airflow.providers.standard.decorators.python._PythonDecoratedOperator"

This happens both with a default and custom XCom backend and for 3.0.0 and 3.0.1.

[vatsrahul1001]

@TJaniF could you share your DAG for this issue?

Also can you create a new issue for this

using explicit xcom pulling via context returns None for mapped tasks.

[amoghrajesh]

Yeah its related but a different problem