SNOW-3923354 decouple pandas version from python connector and support pandas 3.x

.github/workflows/precommit.yml

@@ -238,15 +238,21 @@ jobs:
             .tox/coverage.xml
 
   test-local-testing:
-    name: Test Local Testing Module py-${{ matrix.os }}-${{ matrix.python-version }}
+    name: Test Local Testing Module py-${{ matrix.os }}-${{ matrix.python-version }}-pandas${{ matrix.pandas-major }}
     needs: build
     runs-on: ${{ matrix.os }}
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-latest]
-        python-version: ["3.14"]
-        cloud-provider: [azure]
+        include:
+          - os: ubuntu-latest
+            python-version: "3.14"
+            cloud-provider: azure
+            pandas-major: "2"
+          - os: ubuntu-latest
+            python-version: "3.14"
+            cloud-provider: azure
+            pandas-major: "3"
     steps:
       - name: Checkout Code
         uses: actions/checkout@v4
@@ -286,7 +292,7 @@ jobs:
       - name: Install tox
         run: uv pip install tox --system
       - name: Run tests
-        run: python -m tox -e "py${PYTHON_VERSION/\./}-local"
+        run: python -m tox -e "py${PYTHON_VERSION/\./}-local-pandas${{ matrix.pandas-major }}"
         env:
           PYTHON_VERSION: ${{ matrix.python-version }}
           cloud_provider: ${{ matrix.cloud-provider }}
@@ -302,7 +308,7 @@ jobs:
       - uses: actions/upload-artifact@v4
         with:
           include-hidden-files: true
-          name: coverage_${{ matrix.os }}-${{ matrix.python-version }}-local-testing
+          name: coverage_${{ matrix.os }}-${{ matrix.python-version }}-pandas${{ matrix.pandas-major }}-local-testing
           path: |
             .tox/.coverage
             .tox/coverage.xml

CHANGELOG.md

@@ -8,6 +8,26 @@
 
 - Added interval type support for Python UDFs and stored procedures. Use `datetime.timedelta` as the type annotation for day-time interval (`DayTimeIntervalType`) parameters and return values, and `YearMonthInterval` (a type annotation sentinel from `snowflake.snowpark.types`) for year-month interval (`YearMonthIntervalType`) parameters and return values.
 
+#### Dependency Updates
+
+- Added support for pandas 3, lifting the `pandas<3.0.0` cap added to the `[pandas]` extra in 1.53.0. `snowflake-snowpark-python[pandas]` now declares its own `pandas<4.0.0` and `pyarrow` requirements instead of depending on `snowflake-connector-python[pandas]`. pandas 2 remains fully supported. Upgrading Snowpark does not upgrade an existing pandas 2 install:
+  - pandas 3 requires Python 3.11 or later, so Python 3.10 environments continue to resolve pandas 2.
+  - `snowflake-snowpark-python[modin]` continues to resolve pandas 2, because the `[modin]` extra declares `pandas<=2.4`.
+
+#### Behavior Changes
+
+- On pandas 3, `DataFrame.to_pandas()` and `DataFrame.to_pandas_batches()` return text columns as the `str` dtype instead of `object`, so a SQL NULL in them arrives as `nan` instead of `None`. Test for NULL with `pandas.isna(value)`, not `value is None`. This also covers `VARIANT`, `OBJECT`, `ARRAY`, `MAP` and the geospatial types, and the `future.infer_string` option does not opt out of it.
+
+#### Bug Fixes
+
+- Fixed a bug where `Session.write_pandas`, `Session.write_arrow`, and `Session.create_dataframe` stored a `timedelta` at the column's own resolution instead of nanoseconds, so a microsecond column was written 1000 times too small. Duration columns are now converted to nanoseconds first. pandas 2 was affected too, and rows written by an earlier release need correcting.
+- Fixed a local testing bug where a SQL NULL became `nan` instead of `None` after a column was rebuilt through pandas, so `IS NULL` evaluated to `False` and `strict=True` UDF handlers received `nan`. This affected `parse_json`, `to_char`, `concat`, `concat_ws`, `strip_null_value`, `initcap`, `greatest`, `least`, and VARIANT/OBJECT/ARRAY/MAP subfield access, on pandas 2 as well as pandas 3.
+- The following local testing bugs appeared only when running with pandas 3:
+  - Fixed a bug where `ORDER BY` treated SQL NULL as `nan`.
+  - Fixed a bug where columns omitted from a MERGE insert kept `nan` instead of NULL under Copy-on-Write.
+  - Fixed a bug where `group_by` and `pivot` results lost their column and index names.
+  - Fixed a bug where `create_dataframe` raised `TypeError` for `VARIANT` columns.
+
 ## 1.54.0 (2026-07-29)
 
 ### Snowpark Python API updates

setup.py

@@ -43,8 +43,12 @@
     REQUIRED_PYTHON_VERSION = ">=3.10"
 
 PANDAS_REQUIREMENTS = [
-    f"snowflake-connector-python[pandas]{CONNECTOR_DEPENDENCY_VERSION}",
-    "pandas<3.0.0",
+    "pandas<4.0.0",
+    # Mirrors the pyarrow bounds in snowflake-connector-python's [pandas] extra
+    # (its setup.cfg). We no longer request that extra, but the connector's Arrow
+    # code still needs this range and warns at import when it is not met.
+    "pyarrow>=14.0.1,<24; python_version >= '3.14'",
+    "pyarrow>=14.0.1; python_version < '3.14'",
 ]
 MODIN_REQUIREMENTS = [
     *PANDAS_REQUIREMENTS,

src/snowflake/snowpark/mock/_connection.py

@@ -38,7 +38,11 @@
 from snowflake.snowpark.exceptions import SnowparkSessionException
 from snowflake.snowpark.mock._options import pandas
 from snowflake.snowpark.mock._plan import MockExecutionPlan, execute_mock_plan
-from snowflake.snowpark.mock._snowflake_data_type import ColumnEmulator, TableEmulator
+from snowflake.snowpark.mock._snowflake_data_type import (
+    ColumnEmulator,
+    TableEmulator,
+    isna_helper,
+)
 from snowflake.snowpark.mock._stage_registry import StageEntityRegistry
 from snowflake.snowpark.mock._telemetry import LocalTestOOBTelemetryService
 from snowflake.snowpark.mock._util import get_fully_qualified_name
@@ -653,7 +657,7 @@ def execute(
                     from snowflake.snowpark.mock import CUSTOM_JSON_ENCODER
 
                     for idx, row in res.iterrows():
-                        if row[col] is not None:
+                        if not isna_helper(row[col]):
                             # Snowflake sorts maps by key before serializing
                             if isinstance(row[col], dict):
                                 row[col] = dict(sorted(row[col].items()))

src/snowflake/snowpark/mock/_functions.py

@@ -1303,7 +1303,11 @@ def convert_char(data, _fmt):
                 raise_error=NotImplementedError,
             )
 
-    res = column.combine(fmt, convert_char)
+    res = ColumnEmulator(
+        [convert_char(v, f) for v, f in zip(column, fmt)],
+        index=column.index,
+        dtype=object,
+    ).__finalize__(column)
     res.sf_type = ColumnType(StringType(), column.sf_type.nullable)
     return res
 
@@ -1553,11 +1557,14 @@ def mock_parse_json(expr: ColumnEmulator):
     from snowflake.snowpark.mock import CUSTOM_JSON_DECODER
 
     if isinstance(expr.sf_type.datatype, StringType):
-        res = expr.apply(
-            lambda x: try_convert(
-                partial(json.loads, cls=CUSTOM_JSON_DECODER), False, x
-            )
-        )
+        res = ColumnEmulator(
+            [
+                try_convert(partial(json.loads, cls=CUSTOM_JSON_DECODER), False, x)
+                for x in expr
+            ],
+            index=expr.index,
+            dtype=object,
+        ).__finalize__(expr)
     else:
         res = expr.copy()
     res.sf_type = ColumnType(VariantType(), expr.sf_type.nullable)
@@ -1597,6 +1604,8 @@ def convert_variant_to_array(val):
 def mock_strip_null_value(expr: ColumnEmulator):
     return ColumnEmulator(
         [None if x == "null" else x for x in expr],
+        index=expr.index,
+        dtype=object,
         sf_type=ColumnType(expr.sf_type.datatype, True),
     )
 
@@ -1981,14 +1990,22 @@ def _greatest(x: CompareType, y: Any) -> Union[CompareType, float]:
 
 @patch("greatest")
 def mock_greatest(*exprs: ColumnEmulator):
-    result = reduce(lambda x, y: x.combine(y, _greatest), exprs)
+    result = ColumnEmulator(
+        [reduce(_greatest, row) for row in zip(*exprs)],
+        index=exprs[0].index,
+        dtype=object,
+    ).__finalize__(exprs[0])
     result.sf_type = exprs[0].sf_type
     return result
 
 
 @patch("least")
 def mock_least(*exprs: ColumnEmulator):
-    result = reduce(lambda x, y: x.combine(y, _least), exprs)
+    result = ColumnEmulator(
+        [reduce(_least, row) for row in zip(*exprs)],
+        index=exprs[0].index,
+        dtype=object,
+    ).__finalize__(exprs[0])
     result.sf_type = exprs[0].sf_type
     return result
 
@@ -2032,8 +2049,16 @@ def _initcap(value: Optional[str], delimiters: Optional[str]) -> str:
 
 
 @patch("initcap")
-def mock_initcap(values: ColumnEmulator, delimiters: ColumnEmulator):
-    result = values.combine(delimiters, _initcap)
+def mock_initcap(values: ColumnEmulator, delimiters: ColumnEmulator = None):
+    if delimiters is None:
+        data = [_initcap(v, None) for v in values]
+    else:
+        data = [_initcap(v, d) for v, d in zip(values, delimiters)]
+    result = ColumnEmulator(
+        data,
+        index=values.index,
+        dtype=object,
+    ).__finalize__(values)
     result.sf_type = values.sf_type
     return result
 
@@ -2149,8 +2174,14 @@ def mock_concat(*columns: ColumnEmulator) -> ColumnEmulator:
             ValueError("concat expects one or more column(s) to be passed in.")
         )
     pdf = pandas.concat(columns, axis=1)
-    result = pdf.T.apply(
-        lambda c: None if c.isnull().values.any() else c.astype(str).str.cat()
+    transposed = pdf.T
+    result = ColumnEmulator(
+        [
+            None if c.isnull().values.any() else c.astype(str).str.cat()
+            for _, c in transposed.items()
+        ],
+        index=transposed.columns,
+        dtype=object,
     )
     result.sf_type = ColumnType(StringType(), result.hasnans)
     return result
@@ -2165,10 +2196,14 @@ def mock_concat_ws(*columns: ColumnEmulator) -> ColumnEmulator:
             )
         )
     pdf = pandas.concat(columns, axis=1)
-    result = pdf.T.apply(
-        lambda c: None
-        if c.isnull().values.any()
-        else c[1:].astype(str).str.cat(sep=c[0])
+    transposed = pdf.T
+    result = ColumnEmulator(
+        [
+            None if c.isnull().values.any() else c[1:].astype(str).str.cat(sep=c[0])
+            for _, c in transposed.items()
+        ],
+        index=transposed.columns,
+        dtype=object,
     )
     result.sf_type = ColumnType(StringType(), result.hasnans)
     return result

src/snowflake/snowpark/mock/_plan.py

@@ -2005,9 +2005,9 @@ def flatten_object_cell_func(cell):
                         for unspecified_col in set(rows_to_insert.columns).difference(
                             inserted_columns
                         ):
-                            rows_to_insert[unspecified_col].replace(
-                                np.nan, None, inplace=True
-                            )
+                            rows_to_insert[unspecified_col] = rows_to_insert[
+                                unspecified_col
+                            ].replace(np.nan, None)
 
                     else:
                         if len(clause.values) != len(rows_to_insert.columns):
@@ -2966,14 +2966,20 @@ def get_bound(bound):
             and field in col[index]
             and col[index][field] is None
         ]
-        res = col.apply(lambda x: None if x is None or field not in x else x[field])
+        res = ColumnEmulator(
+            [None if x is None or field not in x else x[field] for x in col],
+            index=col.index,
+            dtype=object,
+        ).__finalize__(col)
         res.sf_type = ColumnType(VariantType(), col.sf_type.nullable)
         return res
     elif isinstance(exp, SubfieldInt):
         col = calculate_expression(exp.child, input_data, analyzer, expr_to_alias)
-        res = col.apply(
-            lambda x: None if x is None or exp.field >= len(x) else x[exp.field]
-        )
+        res = ColumnEmulator(
+            [None if x is None or exp.field >= len(x) else x[exp.field] for x in col],
+            index=col.index,
+            dtype=object,
+        ).__finalize__(col)
         res.sf_type = ColumnType(VariantType(), col.sf_type.nullable)
         return res
     elif isinstance(exp, SnowflakeUDF):

src/snowflake/snowpark/mock/_snowflake_data_type.py

@@ -446,7 +446,39 @@ def __setitem__(self, key, value):
             self.sf_types[key] = value.sf_type
             self._null_rows_idxs_map[key] = value._null_rows_idxs
 
+    def _object_dtype_sort_key(self, key, by):
+        """Wrap a ``sort_values(key=...)`` callable so it sees the untouched column.
+
+        pandas re-wraps each sort column as ``pandas.Series(ndarray)``, which under
+        pandas 3 infers ``str`` dtype, whose NA sentinel is ``nan`` -- so a NULL
+        stored as ``None`` reaches the comparator as ``nan``. Restore ``object``
+        dtype here rather than in the comparator: ``NaN`` sorts largest while NULL
+        obeys ``nulls_first``.
+        """
+        labels = list(by) if isinstance(by, (list, tuple)) else [by]
+
+        def wrapper(series):
+            # One sort column means we know the label outright; only a multi-column
+            # sort has to fall back on the name pandas set on the re-wrapped Series.
+            label = labels[0] if len(labels) == 1 else getattr(series, "name", None)
+            if label is not None:
+                try:
+                    original = PandasDataframeType.__getitem__(self, label)
+                except (KeyError, IndexError):
+                    original = None
+                if isinstance(original, PandasSeriesType) and len(original) == len(
+                    series
+                ):
+                    series = pd.Series(
+                        original.to_numpy(dtype=object), name=label, dtype=object
+                    )
+            return key(series)
+
+        return wrapper
+
     def sort_values(self, by, **kwargs):
+        if kwargs.get("key") is not None and kwargs.get("axis", 0) in (0, "index"):
+            kwargs["key"] = self._object_dtype_sort_key(kwargs["key"], by)
         result = super().sort_values(by, **kwargs)
         result.sf_types = self.sf_types
         return result
@@ -506,7 +538,7 @@ def broadcast_value(value: Any, len: int) -> "ColumnEmulator":
 
 
 class ColumnEmulator(PandasSeriesType):
-    _metadata = ["sf_type", "_null_rows_idxs"]
+    _metadata = ["_name", "sf_type", "_null_rows_idxs"]
 
     @property
     def _constructor(self):

src/snowflake/snowpark/session.py

@@ -3234,6 +3234,29 @@ def get_session_stage(
                 self._session_stage = full_qualified_stage_name
         return f"{STAGE_PREFIX}{self._session_stage}"
 
+    @staticmethod
+    def _normalize_arrow_duration_to_ns(table: "pyarrow.Table") -> "pyarrow.Table":
+        # Snowflake stores duration as an integer with no unit. Cast to ns
+        # so a us column is not written 1000x too small.
+        arrays = []
+        fields = []
+        changed = False
+        for i in range(table.num_columns):
+            field = table.schema.field(i)
+            column = table.column(i)
+            if pyarrow.types.is_duration(field.type) and field.type.unit != "ns":
+                column = column.cast(pyarrow.duration("ns"))
+                field = field.with_type(pyarrow.duration("ns"))
+                changed = True
+            arrays.append(column)
+            fields.append(field)
+        if not changed:
+            return table
+        # Carry the original schema metadata over: pyarrow.schema() drops it otherwise.
+        return pyarrow.Table.from_arrays(
+            arrays, schema=pyarrow.schema(fields, metadata=table.schema.metadata)
+        )
+
     @experimental(version="1.28.0")
     @publicapi
     def write_arrow(
@@ -3295,6 +3318,7 @@ def write_arrow(
                 set use_logical_type as True. Set to None to use Snowflakes default. For more information, see:
                 https://docs.snowflake.com/en/sql-reference/sql/create-file-format
         """
+        table = self._normalize_arrow_duration_to_ns(table)
         cursor = self._conn._conn.cursor()
 
         if quote_identifiers:
@@ -3610,6 +3634,19 @@ def write_pandas(
                     # TODO: Implement here write_pandas correctly.
                     success, ci_output = True, []
                 else:
+                    # write_pandas stores timedelta as a NUMBER of ticks with
+                    # no unit. Normalize to ns so pandas 3's us default does
+                    # not shrink values 1000x versus existing tables.
+                    timedelta_columns = [
+                        td_col
+                        for td_col in df.columns
+                        if pandas.api.types.is_timedelta64_dtype(df[td_col].dtype)
+                        and str(df[td_col].dtype) != "timedelta64[ns]"
+                    ]
+                    if timedelta_columns:
+                        df = df.copy(deep=False)
+                        for td_col in timedelta_columns:
+                            df[td_col] = df[td_col].astype("timedelta64[ns]")
                     success, _, _, ci_output = write_pandas(
                         self._conn._conn,
                         df,