This is an automated email from the ASF dual-hosted git repository. amoghrajesh pushed a commit to branch backport-4091ccf-v3-3-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 6c9832f55c653dd349e65f03791e607309fe4e81 Author: Amogh Desai <[email protected]> AuthorDate: Wed Aug 5 15:37:45 2026 +0530 Add support for pandas 3 based xcoms in airflow (#71103) * Keep DataFrame XComs working on pandas 3 pandas 3 exposes its public classes from the `pandas` namespace, so a DataFrame is now qualified as `pandas.DataFrame` rather than `pandas.core.frame.DataFrame`. The serializer was registered only under the old name, so pushing a DataFrame through XCom raised "cannot serialize object of type <class 'pandas.DataFrame'>". Both names are registered so values written by either version stay readable. pandas 3 also infers a str column where it used to infer object, and keeps its missing values as NA instead of stringifying them, which the amazon and salesforce tests asserted on. * Document pandas 3 impact on DataFrame XComs Deployments need to know that every component has to carry the pandas 3 support before pandas 3 reaches any worker, that a rollback strands the XComs written in the meantime, and that a pulled DataFrame now takes its dtypes from the reader's pandas version. * Name the pandas 3 newsfragment after its own pull request * Add regression test for the pandas 2/3 cross-version registry lookup * rename newsfragment file --------- Co-authored-by: Jarek Potiuk <[email protected]> (cherry picked from commit 4091ccf786b80772ba7f32d26ab8de01d79b7776) --- airflow-core/newsfragments/71103.significant.rst | 33 +++++++++++++++ .../unit/amazon/aws/transfers/test_sql_to_s3.py | 10 ++++- .../tests/unit/salesforce/hooks/test_salesforce.py | 9 +++- .../src/airflow/sdk/serde/serializers/pandas.py | 49 ++-------------------- task-sdk/tests/task_sdk/serde/test_serializers.py | 29 ++++++++++++- 5 files changed, 80 insertions(+), 50 deletions(-) diff --git a/airflow-core/newsfragments/71103.significant.rst b/airflow-core/newsfragments/71103.significant.rst new file mode 100644 index 00000000000..9dd43544b0d --- /dev/null +++ b/airflow-core/newsfragments/71103.significant.rst @@ -0,0 +1,33 @@ +pandas 3 changes how DataFrame XComs are stored and read back + +pandas 3 exposes its public classes from the ``pandas`` namespace, so a DataFrame is qualified as +``pandas.DataFrame`` instead of ``pandas.core.frame.DataFrame``. XComs record that name alongside the +serialized value, so the name written into the metadata database depends on the pandas version of the +component that pushed the value. Airflow registers both names, and a DataFrame written by either +pandas version can be read by either — no configuration change is needed, and existing XComs stay +readable. + +What you should do: + +* **Roll this Airflow version out to every component before pandas 3 reaches any of them** — workers + in particular. A component that predates this change cannot read a DataFrame XCom written under + pandas 3, and fails the pull with: + + .. code-block:: text + + ImportError: pandas.DataFrame was not found in allow list for deserialization imports. + To allow it, add it to allowed_deserialization_classes in the configuration + + The message points at configuration, but the allow list is not the cause and changing it does not + help. The rows are not corrupt: they become readable again as soon as the reader is upgraded. + +* **Treat a downgrade as a one-way door for those XComs.** Rolling back to an Airflow version without + this change strands any DataFrame XCom written while on pandas 3, with the same error, until you + roll forward again. + +* **Review Dags that inspect the dtypes of a pulled DataFrame.** The pandas version of the *reader* + determines what a pulled DataFrame looks like, not the version that wrote it. Under pandas 3, a + column of strings comes back with the ``str`` dtype rather than ``object``, and its missing values + come back as ``nan`` rather than ``None``. Values are unchanged, but downstream code that branches + on ``dtype == "object"``, checks cells with ``is None``, or compares against a reference frame with + ``DataFrame.equals()`` can behave differently after the upgrade. diff --git a/providers/amazon/tests/unit/amazon/aws/transfers/test_sql_to_s3.py b/providers/amazon/tests/unit/amazon/aws/transfers/test_sql_to_s3.py index 5fb50fff9fe..3a960d1fd97 100644 --- a/providers/amazon/tests/unit/amazon/aws/transfers/test_sql_to_s3.py +++ b/providers/amazon/tests/unit/amazon/aws/transfers/test_sql_to_s3.py @@ -24,12 +24,15 @@ from unittest import mock import pandas as pd import polars as pl import pytest +from packaging.version import Version from airflow.exceptions import AirflowProviderDeprecationWarning from airflow.models import Connection from airflow.providers.amazon.aws.transfers.sql_to_s3 import SqlToS3Operator from airflow.providers.common.compat.sdk import AirflowException +PANDAS_3_PLUS = Version(pd.__version__).major >= 3 + class TestSqlToS3Operator: @pytest.mark.parametrize( @@ -156,7 +159,12 @@ class TestSqlToS3Operator: ) dirty_df = pd.DataFrame({"strings": ["a", "b", None], "ints": [1, 2, None]}) op._fix_dtypes(df=dirty_df, file_format=op.file_format) - assert dirty_df["strings"].values[2] == params["null_string_result"] + if PANDAS_3_PLUS: + # pandas 3 infers a str column rather than object, and keeps its missing values as NA + # instead of the object None (csv) or the "None" it used to be stringified to (parquet) + assert pd.isna(dirty_df["strings"].values[2]) + else: + assert dirty_df["strings"].values[2] == params["null_string_result"] assert dirty_df["ints"].dtype.kind == "i" @mock.patch("airflow.providers.amazon.aws.transfers.sql_to_s3.S3Hook") diff --git a/providers/salesforce/tests/unit/salesforce/hooks/test_salesforce.py b/providers/salesforce/tests/unit/salesforce/hooks/test_salesforce.py index 39e2e41fde6..923d59a3956 100644 --- a/providers/salesforce/tests/unit/salesforce/hooks/test_salesforce.py +++ b/providers/salesforce/tests/unit/salesforce/hooks/test_salesforce.py @@ -24,12 +24,15 @@ from unittest.mock import Mock, patch import numpy as np import pandas as pd import pytest +from packaging.version import Version from requests import Session as request_session from simple_salesforce import Salesforce, api from airflow.models.connection import Connection from airflow.providers.salesforce.hooks.salesforce import SalesforceHook +PANDAS_3_PLUS = Version(pd.__version__).major >= 3 + class TestSalesforceHook: def setup_method(self): @@ -349,10 +352,12 @@ class TestSalesforceHook: data_frame = self.salesforce_hook.write_object_to_file(query_results=[], filename=filename, fmt="csv") mock_data_frame.return_value.to_csv.assert_called_once_with(filename, index=False) - # Note that the latest version of pandas dataframes (1.1.2) returns "nan" rather than "None" here + # Note that the latest version of pandas dataframes (1.1.2) returns "nan" rather than "None" here, + # and pandas 3 keeps missing values as NA instead of stringifying them to "nan" at all + missing = np.nan if PANDAS_3_PLUS else "nan" pd.testing.assert_frame_equal( data_frame, - pd.DataFrame({"test": [1, 2, 3], "dict": ["nan", "nan", str({"foo": "bar"})]}), + pd.DataFrame({"test": [1, 2, 3], "dict": [missing, missing, str({"foo": "bar"})]}), check_index_type=False, ) diff --git a/task-sdk/src/airflow/sdk/serde/serializers/pandas.py b/task-sdk/src/airflow/sdk/serde/serializers/pandas.py index a3562f5af47..aaa68580abe 100644 --- a/task-sdk/src/airflow/sdk/serde/serializers/pandas.py +++ b/task-sdk/src/airflow/sdk/serde/serializers/pandas.py @@ -22,49 +22,15 @@ from typing import TYPE_CHECKING from airflow.sdk.module_loading import qualname # lazy loading for performance reasons -# -# ``pandas.core.frame.DataFrame`` is what a DataFrame qualifies as under pandas 2. Under -# pandas 3 the public classes moved to the ``pandas`` namespace, so the same object -# qualifies as ``pandas.DataFrame``. That name is registered too, but *only* so that the -# registry dispatches here and this module can refuse it with an actionable message -- -# without it the caller gets serde's generic "cannot serialize object of type" instead. -# See ``_reject_pandas_3`` below. +# pandas 3 moved the public classes to the `pandas` namespace, so a DataFrame is qualified as +# `pandas.DataFrame` there and as `pandas.core.frame.DataFrame` on pandas 2. Both are registered so +# that XComs serialized by either version stay readable. serializers = [ - "pandas.core.frame.DataFrame", "pandas.DataFrame", + "pandas.core.frame.DataFrame", ] deserializers = serializers -# First pandas major version whose DataFrame XComs this version of Airflow does not support. -_FIRST_UNSUPPORTED_PANDAS_MAJOR = 3 - - -def _reject_pandas_3(pd) -> None: - """ - Refuse a DataFrame XCom when the installed pandas is 3 or newer. - - pandas is an optional dependency, so the ``<3`` constraint Airflow declares cannot be - relied on: a deployment may install pandas 3 directly, or pull it in through another - package. This is the runtime half of that constraint. - - The failure is deliberate rather than best-effort. Under pandas 3 a DataFrame round - trips through parquet with different dtypes than under pandas 2 -- an ``object`` column - comes back as ``str`` and missing values as ``nan`` rather than ``None`` -- so silently - accepting the value would hand tasks data that differs from what was written, with no - signal. Failing on the write is the louder and more recoverable half of that. - """ - major = int(pd.__version__.split(".")[0]) - if major < _FIRST_UNSUPPORTED_PANDAS_MAJOR: - return - raise RuntimeError( - f"DataFrame XComs are not supported with pandas {pd.__version__}: this version of " - f"Airflow supports pandas < {_FIRST_UNSUPPORTED_PANDAS_MAJOR} for XCom serialization. " - "Pin pandas < 3 in the environment that runs your tasks, or avoid passing DataFrames " - "through XCom. Moving to pandas 3 is a deliberate migration -- its DataFrames read " - "back with different dtypes -- and Airflow will enable it in a future release." - ) - - if TYPE_CHECKING: import pandas as pd @@ -81,8 +47,6 @@ def serialize(o: object) -> tuple[U, str, int, bool]: if not isinstance(o, pd.DataFrame): return "", "", 0, False - _reject_pandas_3(pd) - # for now, we *always* serialize into in memory # until we have a generic backend that manages # sinks @@ -102,11 +66,6 @@ def deserialize(cls: type, version: int, data: object) -> pd.DataFrame: if cls is not pd.DataFrame: raise TypeError(f"do not know how to deserialize {qualname(cls)}") - # Reading is refused too, not just writing: a payload written under pandas 2 comes back - # with pandas 3 dtypes, so the task would silently receive different data than was - # pushed. An error is the safer outcome. - _reject_pandas_3(pd) - if not isinstance(data, str): raise TypeError(f"serialized {qualname(cls)} has wrong data type {type(data)}") diff --git a/task-sdk/tests/task_sdk/serde/test_serializers.py b/task-sdk/tests/task_sdk/serde/test_serializers.py index 8ef21272194..d67436b664c 100644 --- a/task-sdk/tests/task_sdk/serde/test_serializers.py +++ b/task-sdk/tests/task_sdk/serde/test_serializers.py @@ -281,6 +281,25 @@ class TestSerializers: d = deserialize(e) assert i.equals(d) + @pytest.mark.parametrize( + "classname", + ["pandas.DataFrame", "pandas.core.frame.DataFrame"], + ) + def test_pandas_deserializes_regardless_of_writer_qualname(self, classname): + """ + A DataFrame XCom must deserialize under either pandas major's registry qualname. + + The installed pandas only ever produces its own qualname, so this forges the tag to + force the lookup through the other registry entry. + """ + import pandas as pd + + i = pd.DataFrame(data={"col1": [1, 2], "col2": [3, 4]}) + e = serialize(i) + e[CLASSNAME] = classname + d = deserialize(e) + assert i.equals(d) + def test_pandas_serializers(self): from airflow.sdk.serde.serializers.pandas import serialize @@ -317,12 +336,18 @@ class TestSerializers: @pytest.mark.parametrize( ("klass", "version", "data", "msg"), [ - (pd.DataFrame, 999, "", r"serialized 999 of pandas.core.frame.DataFrame > 1"), # version too new + # pandas 3 qualifies the class as pandas.DataFrame, pandas 2 as pandas.core.frame.DataFrame + ( + pd.DataFrame, + 999, + "", + r"serialized 999 of pandas(\.core\.frame)?\.DataFrame > 1", + ), # version too new ( pd.DataFrame, 1, 123, - r"serialized pandas.core.frame.DataFrame has wrong data type .*<class 'int'>", + r"serialized pandas(\.core\.frame)?\.DataFrame has wrong data type .*<class 'int'>", ), # bad payload type (str, 1, "", r"do not know how to deserialize builtins.str"), # bad class ],
