Fix last and first error method with Dask processes scheduler

flox/core.py

@@ -892,7 +892,8 @@ def reindex_(
             "Currently does not support reindexing with object arrays of tuples. "
             "These occur when grouping by multi-indexed variables in xarray."
         )
-    if fill_value is xrdtypes.NA or isnull(fill_value):
+    # Use '==' instead of 'is', as Dask serialization can break identity checks.
+    if fill_value == xrdtypes.NA or isnull(fill_value):
         new_dtype, fill_value = xrdtypes.maybe_promote(array.dtype)
     else:
         new_dtype = array.dtype
@@ -1380,7 +1381,8 @@ def _finalize_results(
             if fill_value is None:
                 raise ValueError("Filling is required but fill_value is None.")
             # This allows us to match xarray's type promotion rules
-            if fill_value is xrdtypes.NA:
+            # Use '==' instead of 'is', as Dask serialization can break identity checks.
+            if fill_value == xrdtypes.NA:
                 new_dtype, fill_value = xrdtypes.maybe_promote(finalized[agg.name].dtype)
                 finalized[agg.name] = finalized[agg.name].astype(new_dtype)
 

tests/test_xarray.py

@@ -798,3 +798,16 @@ def test_groupby_preserve_dtype(reduction):
     expected = getattr(np, reduction)(ds.test.data, axis=0).dtype
 
     assert actual == expected
+
+
+@requires_dask
+def test_resample_first_last_empty():
+    with xr.set_options(use_flox=True), dask.config.set(scheduler="processes"):
+        arr = xr.DataArray(
+            np.nan,
+            coords={
+                "date": pd.to_datetime(["2025-03-24", "2025-06-23"]),
+            },
+            dims=["date"],
+        ).chunk(date=(1, 1))
+        arr.resample(date="QE").last().compute()