refactor(pyspark): only allow timezoned timestamps with pyspark < 3.4

Author: cpcloud

ibis/backends/pyspark/datatypes.py

@@ -126,7 +126,7 @@ def from_ibis(cls, dtype):
             else:
                 if not SUPPORTS_TIMESTAMP_NTZ:
                     raise com.UnsupportedBackendType(
-                        "PySpark<3.5 doesn't properly support timestamps without a timezone"
+                        "PySpark<3.4 doesn't properly support timestamps without a timezone"
                     )
                 return pt.TimestampNTZType()
         else:

ibis/backends/pyspark/tests/test_ddl.py

@@ -181,9 +181,9 @@ def test_create_table_reserved_identifier(con, alltypes, keyword_t):
 
 
 @pytest.mark.xfail_version(
-    pyspark=["pyspark<3.5"],
+    pyspark=["pyspark<3.4"],
     raises=ValueError,
-    reason="PySparkAnalysisException is not available in PySpark <3.5",
+    reason="PySparkAnalysisException is not available in PySpark <3.4",
 )
 def test_create_database_exists(con):
     con.create_database(dbname := util.gen_name("dbname"))
@@ -197,9 +197,9 @@ def test_create_database_exists(con):
 
 
 @pytest.mark.xfail_version(
-    pyspark=["pyspark<3.5"],
+    pyspark=["pyspark<3.4"],
     raises=ValueError,
-    reason="PySparkAnalysisException is not available in PySpark <3.5",
+    reason="PySparkAnalysisException is not available in PySpark <3.4",
 )
 def test_drop_database_exists(con):
     con.create_database(dbname := util.gen_name("dbname"))

ibis/backends/tests/test_generic.py

@@ -1130,7 +1130,7 @@ def test_typeof(con):
 @pytest.mark.notimpl(["polars"], reason="incorrect answer")
 @pytest.mark.notyet(["impala"], reason="can't find table in subquery")
 @pytest.mark.notimpl(["datafusion", "druid"])
-@pytest.mark.xfail_version(pyspark=["pyspark<3.5"])
+@pytest.mark.xfail_version(pyspark=["pyspark<3.4"])
 @pytest.mark.notyet(["exasol"], raises=ExaQueryError, reason="not supported by exasol")
 @pytest.mark.notyet(
     ["risingwave"],

ibis/backends/tests/test_struct.py

@@ -253,8 +253,8 @@ def test_keyword_fields(con, nullable):
     reason="doesn't seem to support IN-style subqueries on structs",
 )
 @pytest.mark.xfail_version(
-    pyspark=["pyspark<3.5"],
-    reason="requires pyspark 3.5",
+    pyspark=["pyspark<3.4"],
+    reason="requires pyspark 3.4",
     raises=PySparkAnalysisException,
 )
 @pytest.mark.notimpl(

ibis/backends/tests/test_temporal.py

@@ -293,9 +293,42 @@ def test_timestamp_extract_week_of_year(backend, alltypes, df):
 @pytest.mark.parametrize(
     ("ibis_unit", "pandas_unit"),
     [
-        param("Y", "Y", id="year"),
-        param("Q", "Q", id="quarter"),
-        param("M", "M", id="month"),
+        param(
+            "Y",
+            "Y",
+            id="year",
+            marks=[
+                pytest.mark.xfail_version(
+                    pyspark=["pyspark<3.4"],
+                    reason="no support for timezoneless timestamps",
+                    raises=UserWarning,
+                ),
+            ],
+        ),
+        param(
+            "Q",
+            "Q",
+            id="quarter",
+            marks=[
+                pytest.mark.xfail_version(
+                    pyspark=["pyspark<3.4"],
+                    reason="no support for timezoneless timestamps",
+                    raises=UserWarning,
+                ),
+            ],
+        ),
+        param(
+            "M",
+            "M",
+            id="month",
+            marks=[
+                pytest.mark.xfail_version(
+                    pyspark=["pyspark<3.4"],
+                    reason="no support for timezoneless timestamps",
+                    raises=UserWarning,
+                ),
+            ],
+        ),
         param(
             "W",
             "W",
@@ -306,9 +339,24 @@ def test_timestamp_extract_week_of_year(backend, alltypes, df):
                     raises=AssertionError,
                     reason="implemented, but doesn't match other backends",
                 ),
+                pytest.mark.xfail_version(
+                    pyspark=["pyspark<3.4"],
+                    reason="no support for timezoneless timestamps",
+                    raises=UserWarning,
+                ),
+            ],
+        ),
+        param(
+            "D",
+            "D",
+            marks=[
+                pytest.mark.xfail_version(
+                    pyspark=["pyspark<3.4"],
+                    reason="no support for timezoneless timestamps",
+                    raises=UserWarning,
+                ),
             ],
         ),
-        param("D", "D"),
         param(
             "h",
             "h",
@@ -432,6 +480,11 @@ def test_timestamp_truncate(backend, alltypes, df, ibis_unit, pandas_unit):
         ),
     ],
 )
+@pytest.mark.xfail_version(
+    pyspark=["pyspark<3.4"],
+    reason="no support for timezoneless timestamps",
+    raises=UserWarning,
+)
 @pytest.mark.notimpl(["druid"], raises=com.OperationNotDefinedError)
 def test_date_truncate(backend, alltypes, df, unit):
     expr = alltypes.timestamp_col.date().truncate(unit).name("tmp")
@@ -501,10 +554,31 @@ def test_date_truncate(backend, alltypes, df, unit):
                     reason="Bind error: Invalid unit: week",
                 ),
                 sqlite_without_ymd_intervals,
+                pytest.mark.xfail_version(
+                    pyspark=["pyspark<3.4"], raises=AssertionError
+                ),
+            ],
+        ),
+        param(
+            "D",
+            pd.offsets.DateOffset,
+            marks=[
+                sqlite_without_ymd_intervals,
+                pytest.mark.xfail_version(
+                    pyspark=["pyspark<3.4"], raises=AssertionError
+                ),
+            ],
+        ),
+        param(
+            "h",
+            pd.Timedelta,
+            marks=[
+                sqlite_without_hms_intervals,
+                pytest.mark.xfail_version(
+                    pyspark=["pyspark<3.4"], raises=AssertionError
+                ),
             ],
         ),
-        param("D", pd.offsets.DateOffset, marks=sqlite_without_ymd_intervals),
-        param("h", pd.Timedelta, marks=sqlite_without_hms_intervals),
         param("m", pd.Timedelta, marks=sqlite_without_hms_intervals),
         param("s", pd.Timedelta, marks=sqlite_without_hms_intervals),
         param(
@@ -770,6 +844,10 @@ def convert_to_offset(x):
                     raises=PySparkConnectGrpcException,
                     reason="arrow conversion breaks",
                 ),
+                pytest.mark.xfail_version(
+                    pyspark=["pyspark<3.4"],
+                    reason="no support for timezoneless timestamps",
+                ),
                 pytest.mark.notyet(
                     ["databricks"],
                     raises=AssertionError,
@@ -830,6 +908,10 @@ def convert_to_offset(x):
                     raises=PySparkConnectGrpcException,
                     reason="arrow conversion breaks",
                 ),
+                pytest.mark.xfail_version(
+                    pyspark=["pyspark<3.4"],
+                    reason="no support for timezoneless timestamps",
+                ),
                 pytest.mark.notyet(
                     ["databricks"],
                     raises=AssertionError,
@@ -876,6 +958,9 @@ def test_temporal_binop(backend, con, alltypes, df, expr_fn, expected_fn):
                     reason="TableException: DAY_INTERVAL_TYPES precision is not supported: 5",
                 ),
                 sqlite_without_ymd_intervals,
+                pytest.mark.xfail_version(
+                    pyspark=["pyspark<3.4"], raises=AssertionError
+                ),
             ],
         ),
         param("5W", plus, id="weeks-plus", marks=sqlite_without_ymd_intervals),
@@ -940,7 +1025,8 @@ def test_timestamp_comparison_filter(backend, con, alltypes, df, func_name):
         comparison_fn(alltypes.timestamp_col.cast("timestamp('UTC')"), ts)
     )
 
-    col = df.timestamp_col.dt.tz_localize("UTC")
+    if getattr((col := df.timestamp_col).dtype, "tz", None) is None:
+        col = df.timestamp_col.dt.tz_localize("UTC")
     expected = df[comparison_fn(col, ts)]
     result = con.execute(expr)
 
@@ -973,7 +1059,8 @@ def test_timestamp_comparison_filter_numpy(backend, con, alltypes, df, func_name
 
     ts = pd.Timestamp(ts.item(), tz="UTC")
 
-    col = df.timestamp_col.dt.tz_localize("UTC")
+    if getattr((col := df.timestamp_col).dtype, "tz", None) is None:
+        col = df.timestamp_col.dt.tz_localize("UTC")
     expected = df[comparison_fn(col, ts)]
     result = con.execute(expr)