feat(pyspark): support pyspark 4.0

cpcloud · cpcloud · commit e9c35e6a3574 · 2026-02-01T11:06:33.000-05:00
BREAKING CHANGE: PySpark &lt;3.5 is no longer supported
diff --git a/.github/workflows/ibis-backends.yml b/.github/workflows/ibis-backends.yml
@@ -571,25 +571,24 @@ jobs:
       matrix:
         include:
           - python-version: "3.10"
-            pyspark-minor-version: "3.3"
+            pyspark-minor-version: "3.5"
+            scala-version: "2.12"
             tag: local
             deps:
-              - pyspark==3.3.4
-              - pandas==1.5.3
-              - numpy==1.23.5
+              - "'pyspark>=3.5,<4'"
           - python-version: "3.13"
-            pyspark-minor-version: "3.5"
+            pyspark-minor-version: "4.0"
+            scala-version: "2.13"
             tag: local
             deps:
-              - setuptools==75.1.0
-              - delta-spark==3.3.0
+              - delta-spark
           - python-version: "3.12"
-            pyspark-minor-version: "3.5"
+            pyspark-minor-version: "4.0"
             SPARK_REMOTE: "sc://localhost:15002"
+            scala-version: "2.13"
             tag: remote
             deps:
-              - setuptools==75.1.0
-              - delta-spark==3.3.0
+              - setuptools
               - googleapis-common-protos
               - grpcio
               - grpcio-status
@@ -622,17 +621,12 @@ jobs:
       - name: install uv
         uses: astral-sh/setup-uv@v7.2.1
 
-        # it requires a version of pandas that pyspark is not compatible with
-      - name: remove lonboard
-        if: matrix.pyspark-minor-version == '3.3'
-        run: uv remove --group docs --no-sync lonboard
-
       - name: install pyspark-specific dependencies
+        if: matrix.deps
         run: uv add --no-sync ${{ join(matrix.deps, ' ') }}
 
       - name: install iceberg
-        shell: bash
-        run: just download-iceberg-jar ${{ matrix.pyspark-minor-version }}
+        run: just download-iceberg-jar ${{ matrix.pyspark-minor-version }} ${{ matrix.scala-version }}
 
       - name: run spark connect tests
         if: matrix.tag == 'remote'
@@ -647,7 +641,6 @@ jobs:
         run: just ci-check "--extra pyspark --extra examples" -m pyspark
 
       - name: check that no untracked files were produced
-        shell: bash
         run: git checkout uv.lock pyproject.toml && ! git status --porcelain | grep -F .
 
       - name: upload code coverage
diff --git a/.gitignore b/.gitignore
@@ -149,3 +149,4 @@ jupyter_lite_config.json
 # pixi environments
 .pixi
 *.egg-info
+warehouse
diff --git a/compose.yaml b/compose.yaml
@@ -615,10 +615,15 @@ services:
       - risingwave
 
   spark-connect:
-    image: bitnamilegacy/spark:3.5.6
+    image: bitnamilegacy/spark:4.0.0
     ports:
       - 15002:15002
-    command: /opt/bitnami/spark/sbin/start-connect-server.sh --name ibis_testing --packages org.apache.spark:spark-connect_2.12:3.5.3,org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1,io.delta:delta-spark_2.12:3.3.0
+    command:
+      - /opt/bitnami/spark/sbin/start-connect-server.sh
+      - --name
+      - ibis_testing
+      - --packages
+      - org.apache.spark:spark-connect_2.13:4.0.0,org.apache.iceberg:iceberg-spark-runtime-4.0_2.13:1.10.1,io.delta:delta-spark_2.13:4.0.0
     healthcheck:
       test:
         - CMD-SHELL
diff --git a/docker/spark-connect/conf.properties b/docker/spark-connect/conf.properties
@@ -1,6 +1,5 @@
 spark.driver.extraJavaOptions=-Duser.timezone=GMT
 spark.executor.extraJavaOptions=-Duser.timezone=GMT
-spark.jars.packages=org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1
 spark.sql.catalog.local.type=hadoop
 spark.sql.catalog.local.warehouse=warehouse
 spark.sql.catalog.local=org.apache.iceberg.spark.SparkCatalog
diff --git a/ibis/backends/pyspark/__init__.py b/ibis/backends/pyspark/__init__.py
@@ -128,6 +128,7 @@ class Backend(
     name = "pyspark"
     compiler = sc.pyspark.compiler
     temporary_example = False
+    overwrite_example = True
 
     class Options(ibis.config.Config):
         """PySpark options.
diff --git a/ibis/backends/pyspark/datatypes.py b/ibis/backends/pyspark/datatypes.py
@@ -11,7 +11,6 @@
 
 # DayTimeIntervalType introduced in Spark 3.2 (at least) but didn't show up in
 # PySpark until version 3.3
-PYSPARK_33 = vparse(pyspark.__version__) >= vparse("3.3")
 PYSPARK_35 = vparse(pyspark.__version__) >= vparse("3.5")
 
 
@@ -35,13 +34,12 @@
 _to_pyspark_dtypes[dt.UUID] = pt.StringType
 
 
-if PYSPARK_33:
-    _pyspark_interval_units = {
-        pt.DayTimeIntervalType.SECOND: "s",
-        pt.DayTimeIntervalType.MINUTE: "m",
-        pt.DayTimeIntervalType.HOUR: "h",
-        pt.DayTimeIntervalType.DAY: "D",
-    }
+_pyspark_interval_units = {
+    pt.DayTimeIntervalType.SECOND: "s",
+    pt.DayTimeIntervalType.MINUTE: "m",
+    pt.DayTimeIntervalType.HOUR: "h",
+    pt.DayTimeIntervalType.DAY: "D",
+}
 
 
 class PySparkType(TypeMapper):
@@ -62,7 +60,7 @@ def to_ibis(cls, typ, nullable=True):
             fields = {f.name: cls.to_ibis(f.dataType) for f in typ.fields}
 
             return dt.Struct(fields, nullable=nullable)
-        elif PYSPARK_33 and isinstance(typ, pt.DayTimeIntervalType):
+        elif isinstance(typ, pt.DayTimeIntervalType):
             if (
                 typ.startField == typ.endField
                 and typ.startField in _pyspark_interval_units
@@ -71,7 +69,7 @@ def to_ibis(cls, typ, nullable=True):
                 return dt.Interval(unit, nullable=nullable)
             else:
                 raise com.IbisTypeError(f"{typ!r} couldn't be converted to Interval")
-        elif PYSPARK_35 and isinstance(typ, pt.TimestampNTZType):
+        elif isinstance(typ, pt.TimestampNTZType):
             return dt.Timestamp(nullable=nullable)
         elif isinstance(typ, pt.UserDefinedType):
             return cls.to_ibis(typ.sqlType(), nullable=nullable)
diff --git a/ibis/backends/pyspark/tests/conftest.py b/ibis/backends/pyspark/tests/conftest.py
@@ -14,6 +14,7 @@
 import pandas as pd
 import pytest
 from filelock import FileLock
+from packaging.version import parse as parse_version
 
 import ibis
 from ibis import util
@@ -277,8 +278,20 @@ def parquet_dir(self) -> str:
 
         @staticmethod
         def connect(*, tmpdir, worker_id, **kw):  # noqa: ARG004
+            import pyspark
             from pyspark.sql import SparkSession
 
+            spark_ver = parse_version(pyspark.__version__)
+
+            (scala_jarfile,) = (
+                Path(pyspark.__file__)
+                .parent.joinpath("jars")
+                .glob("scala-library-*.jar")
+            )
+            _, scala_version_str = scala_jarfile.stem.rsplit("-", 1)
+            scala_ver = parse_version(scala_version_str)
+            iceberg_jar = f"iceberg-spark-runtime-{spark_ver.major}.{spark_ver.minor}_{scala_ver.major}.{scala_ver.minor}:1.10.1"
+
             config = reduce(
                 lambda config, line: config.config(
                     *map(str.strip, line.strip().split("=", 1))
@@ -306,6 +319,7 @@ def connect(*, tmpdir, worker_id, **kw):  # noqa: ARG004
                         "spark.sql.execution.arrow.pyspark.enabled=false",
                         "spark.sql.shuffle.partitions=1",
                         "spark.storage.blockManagerSlaveTimeoutMs=4200s",
+                        f"spark.jars.packages=org.apache.iceberg:{iceberg_jar}",
                     ],
                 ),
                 SparkSession.builder.appName("ibis_testing"),
diff --git a/ibis/backends/pyspark/tests/test_client.py b/ibis/backends/pyspark/tests/test_client.py
@@ -1,12 +1,9 @@
 from __future__ import annotations
 
-import pytest
-
 import ibis
 from ibis.util import gen_name
 
 
-@pytest.mark.xfail_version(pyspark=["pyspark<3.4"], reason="no catalog support")
 def test_catalog_db_args(con):
     t = ibis.memtable({"epoch": [1712848119, 1712848121, 1712848155]})
 
@@ -61,7 +58,6 @@ def test_create_table_no_catalog(con):
     assert con.current_database != "default"
 
 
-@pytest.mark.xfail_version(pyspark=["pyspark<3.4"], reason="no catalog support")
 def test_create_table_with_partition_and_catalog(con):
     # Create a sample table with a partition column
     data = {
diff --git a/ibis/backends/sql/compilers/base.py b/ibis/backends/sql/compilers/base.py
@@ -899,7 +899,7 @@ def visit_Round(self, op, *, arg, digits):
     ### Dtype Dysmorphia
 
     def visit_TryCast(self, op, *, arg, to):
-        return sge.TryCast(this=arg, to=self.type_mapper.from_ibis(to))
+        return sge.TryCast(this=arg, to=self.type_mapper.from_ibis(to), safe=True)
 
     ### Comparator Conundrums
 
diff --git a/ibis/backends/tests/errors.py b/ibis/backends/tests/errors.py
@@ -52,6 +52,9 @@
     from pyspark.errors.exceptions.base import (
         ArithmeticException as PySparkArithmeticException,
     )
+    from pyspark.errors.exceptions.base import (
+        NumberFormatException as PySparkNumberFormatException,
+    )
     from pyspark.errors.exceptions.base import ParseException as PySparkParseException
     from pyspark.errors.exceptions.base import PySparkValueError
     from pyspark.errors.exceptions.base import PythonException as PySparkPythonException
@@ -66,7 +69,7 @@
         PySparkPythonException
     ) = PySparkUnsupportedOperationException = PySparkConnectGrpcException = (
         PySparkValueError
-    ) = None
+    ) = PySparkNumberFormatException = None
 
 try:
     from google.api_core.exceptions import BadRequest as GoogleBadRequest
diff --git a/ibis/backends/tests/test_aggregation.py b/ibis/backends/tests/test_aggregation.py
@@ -28,6 +28,7 @@
     PyDruidProgrammingError,
     PyODBCProgrammingError,
     PySparkAnalysisException,
+    PySparkNumberFormatException,
     PySparkPythonException,
     SingleStoreDBOperationalError,
     SnowflakeProgrammingError,
@@ -1191,7 +1192,9 @@ def test_median(alltypes, df):
     reason="doesn't support median of strings",
 )
 @pytest.mark.notyet(
-    ["pyspark"], raises=AssertionError, reason="pyspark returns null for string median"
+    ["pyspark"],
+    raises=(PySparkNumberFormatException, AssertionError),
+    reason="pyspark does not support string quantiles",
 )
 @pytest.mark.notyet(
     ["databricks"],
diff --git a/ibis/backends/tests/test_api.py b/ibis/backends/tests/test_api.py
@@ -37,7 +37,6 @@ def test_version(backend):
     reason="backend does not support catalogs",
     raises=AttributeError,
 )
-@pytest.mark.xfail_version(pyspark=["pyspark<3.4"])
 def test_catalog_consistency(backend, con):
     catalogs = con.list_catalogs()
     assert isinstance(catalogs, list)
diff --git a/ibis/backends/tests/test_client.py b/ibis/backends/tests/test_client.py
@@ -797,7 +797,6 @@ def test_insert_from_memtable(con, temp_table):
     raises=AttributeError,
     reason="doesn't support the common notion of a catalog",
 )
-@pytest.mark.xfail_version(pyspark=["pyspark<3.4"])
 def test_list_catalogs(con):
     # Every backend has its own databases
     test_catalogs = {
@@ -1870,7 +1869,6 @@ def test_cross_database_join(con_create_database, monkeypatch):
     raises=DatabricksServerOperationError,
 )
 @pytest.mark.notimpl(["athena"], reason="insert isn't implemented yet")
-@pytest.mark.xfail_version(pyspark=["pyspark<3.4"])
 def test_insert_into_table_missing_columns(con, temp_table):
     db = getattr(con, "current_database", None)
 
diff --git a/ibis/backends/tests/test_numeric.py b/ibis/backends/tests/test_numeric.py
@@ -1282,7 +1282,8 @@ def test_floating_mod(backend, alltypes, df):
         ),
     ],
 )
-@pytest.mark.notyet(["mysql", "singlestoredb", "pyspark"], raises=AssertionError)
+@pytest.mark.notyet(["mysql", "singlestoredb"], raises=AssertionError)
+@pytest.mark.notyet(["pyspark"], raises=(AssertionError, PySparkArithmeticException))
 @pytest.mark.notyet(["databricks"], raises=AssertionError, reason="returns NaNs")
 @pytest.mark.notyet(
     ["sqlite"], raises=AssertionError, reason="returns NULL when dividing by zero"
diff --git a/ibis/backends/tests/test_temporal.py b/ibis/backends/tests/test_temporal.py
@@ -34,14 +34,12 @@
     PyDruidProgrammingError,
     PyODBCDataError,
     PyODBCProgrammingError,
-    PySparkConnectGrpcException,
     SingleStoreDBOperationalError,
     SingleStoreDBProgrammingError,
     SnowflakeProgrammingError,
     TrinoUserError,
 )
 from ibis.common.annotations import ValidationError
-from ibis.conftest import IS_SPARK_REMOTE
 
 np = pytest.importorskip("numpy")
 pd = pytest.importorskip("pandas")
@@ -767,12 +765,6 @@ def convert_to_offset(x):
                     raises=PyAthenaOperationalError,
                     reason="not supported in hive",
                 ),
-                pytest.mark.notyet(
-                    ["pyspark"],
-                    condition=IS_SPARK_REMOTE,
-                    raises=PySparkConnectGrpcException,
-                    reason="arrow conversion breaks",
-                ),
                 pytest.mark.notyet(
                     ["databricks"],
                     raises=AssertionError,
@@ -827,12 +819,6 @@ def convert_to_offset(x):
                     raises=com.OperationNotDefinedError,
                     reason="Some wonkiness in sqlglot generation.",
                 ),
-                pytest.mark.notyet(
-                    ["pyspark"],
-                    condition=IS_SPARK_REMOTE,
-                    raises=PySparkConnectGrpcException,
-                    reason="arrow conversion breaks",
-                ),
                 pytest.mark.notyet(
                     ["databricks"],
                     raises=AssertionError,
@@ -1835,8 +1821,7 @@ def test_subsecond_cast_to_timestamp(con, dtype):
 @pytest.mark.notyet(
     ["pyspark"],
     reason="PySpark doesn't handle big timestamps",
-    condition=not IS_SPARK_REMOTE,
-    raises=pd.errors.OutOfBoundsDatetime,
+    raises=(pd.errors.OutOfBoundsDatetime, ArrowInvalid),
 )
 @pytest.mark.notyet(
     ["databricks"],
@@ -1901,8 +1886,10 @@ def test_timestamp_date_comparison(backend, alltypes, df, left_fn, right_fn):
 @pytest.mark.notimpl(
     ["clickhouse", "athena"], reason="returns incorrect results", raises=AssertionError
 )
-@pytest.mark.notimpl(
-    ["pyspark"], condition=not IS_SPARK_REMOTE, raises=pd.errors.OutOfBoundsDatetime
+@pytest.mark.notyet(
+    ["pyspark"],
+    reason="PySpark doesn't handle large timestamps",
+    raises=(pd.errors.OutOfBoundsDatetime, ArrowInvalid),
 )
 @pytest.mark.notimpl(["polars"], raises=PolarsInvalidOperationError)
 @pytest.mark.notyet(
diff --git a/justfile b/justfile
@@ -144,7 +144,7 @@ download-data owner="ibis-project" repo="testing-data" rev="master":
     fi
 
 # download the iceberg jar used for testing pyspark and iceberg integration
-download-iceberg-jar pyspark scala="2.12" iceberg="1.6.1":
+download-iceberg-jar pyspark-version="4.0" scala-version="2.13" iceberg-version="1.10.1":
     #!/usr/bin/env bash
     set -eo pipefail
 
@@ -155,10 +155,10 @@ download-iceberg-jar pyspark scala="2.12" iceberg="1.6.1":
     fi
     pyspark="$("${runner[@]}" -c "import pyspark; print(pyspark.__file__.rsplit('/', 1)[0])")"
     pushd "${pyspark}/jars"
-    jar="iceberg-spark-runtime-{{ pyspark }}_{{ scala }}-{{ iceberg }}.jar"
-    url="https://search.maven.org/remotecontent?filepath=org/apache/iceberg/iceberg-spark-runtime-{{ pyspark }}_{{ scala }}/{{ iceberg }}/${jar}"
-    curl -qSsL -o "${jar}" "${url}"
-    ls "${jar}"
+    jar="iceberg-spark-runtime-{{ pyspark-version }}_{{ scala-version }}-{{ iceberg-version }}.jar"
+    prefix=org/apache/iceberg/iceberg-spark-runtime-{{ pyspark-version }}_{{ scala-version }}
+    url="https://search.maven.org/remotecontent?filepath=${prefix}/{{ iceberg-version }}/${jar}"
+    curl --fail --fail-early --show-error --silent --location --output "${jar}" "${url}"
 
 # pull images
 pull *backends:
diff --git a/nix/pyproject-overrides.nix b/nix/pyproject-overrides.nix
diff --git a/pyproject.toml b/pyproject.toml
diff --git a/requirements-dev.txt b/requirements-dev.txt
diff --git a/uv.lock b/uv.lock
