content: from __future__ import annotations
  
  import numpy as np
  import pandas as pd
  import xarray as xr
  
  from numpy import nan
  from packaging.version import Version
  
  import datashader as ds
  from datashader.datatypes import RaggedArray
  import datashader.utils as du
  
  import pytest
  from datashader.tests.test_pandas import _pandas
  
  try:
      import spatialpandas as sp
      import spatialpandas.dask  # noqa (API import)
  except ImportError:
      sp = None
  
  from datashader.tests.test_pandas import (
      assert_eq_xr, assert_eq_ndarray, values
  )
  
  
  try:
      import dask.dataframe as dd
      from dask.context import config
      config.set(scheduler='synchronous')
  except ImportError:
      pytestmark = pytest.importorskip("dask")
  
  
  def _dask():
      return dd.from_pandas(_pandas(), npartitions=2)
  
  
  def _dask_cudf():
      import dask_cudf
  
      _dask = dd.from_pandas(_pandas(), npartitions=2)
      if Version(dask_cudf.__version__) >= Version("24.06"):
          return _dask.to_backend("cudf")
      else:
          return dask_cudf.from_dask_dataframe(_dask)
  
  
  _backends = [
      pytest.param(_dask, id="dask"),
      pytest.param(_dask_cudf, marks=pytest.mark.gpu, id="dask-cudf"),
  ]
  
  
  @pytest.fixture(params=_backends)
  def ddf(request):
      return request.param()
  
  
  @pytest.fixture(params=[1, 2, 4])
  def npartitions(request):
      return request.param
  
  
  def _dask_DataFrame(*args, **kwargs):
      if kwargs.pop("geo", False):
          df = sp.GeoDataFrame(*args, **kwargs)
      else:
          df = pd.DataFrame(*args, **kwargs)
      return dd.from_pandas(df, npartitions=2)
  
  
  def _dask_cudf_DataFrame(*args, **kwargs):
      import cudf
      import dask_cudf
  
      if kwargs.pop("geo", False):
          # As of dask-cudf version 24.06, dask-cudf is not
          # compatible with spatialpandas version 0.4.10
          pytest.skip("dask-cudf currently does not work with spatialpandas")
      cdf = cudf.DataFrame(pd.DataFrame(*args, **kwargs), nan_as_null=False)
      return dask_cudf.from_cudf(cdf, npartitions=2)
  
  
  _backends = [
      pytest.param(_dask_DataFrame, id="dask"),
      pytest.param(_dask_cudf_DataFrame, marks=pytest.mark.gpu, id="dask-cudf"),
  ]
  
  @pytest.fixture(params=_backends)
  def DataFrame(request):
      return request.param
  
  c = ds.Canvas(plot_width=2, plot_height=2, x_range=(0, 1), y_range=(0, 1))
  c_logx = ds.Canvas(plot_width=2, plot_height=2, x_range=(1, 10),
                     y_range=(0, 1), x_axis_type='log')
  c_logy = ds.Canvas(plot_width=2, plot_height=2, x_range=(0, 1),
                     y_range=(1, 10), y_axis_type='log')
  c_logxy = ds.Canvas(plot_width=2, plot_height=2, x_range=(1, 10),
                      y_range=(1, 10), x_axis_type='log', y_axis_type='log')
  
  axis = ds.core.LinearAxis()
  lincoords = axis.compute_index(axis.compute_scale_and_translate((0, 1), 2), 2)
  coords = [lincoords, lincoords]
  dims = ['y', 'x']
  
  
  def assert_eq(agg, b):
      assert agg.equals(b)
  
  
  def floats(n):
      """Returns contiguous list of floats from initial point"""
      while True:
          yield n
          n = n + np.spacing(n)
  
  
  def test_count(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      out = xr.DataArray(np.array([[5, 5], [5, 5]], dtype='i4'),
                         coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.count('i32')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.count('i64')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.count()), out)
      out = xr.DataArray(np.array([[4, 5], [5, 5]], dtype='i4'),
                         coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.count('f32')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.count('f64')), out)
  
  
  def test_any(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      out = xr.DataArray(np.array([[True, True], [True, True]]),
                         coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.any('i64')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.any('f64')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.any()), out)
      out = xr.DataArray(np.array([[True, True], [True, False]]),
                         coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.any('empty_bin')), out)
  
  
  def test_sum(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      out = xr.DataArray(
          values(_pandas().i32).reshape((2, 2, 5)).sum(axis=2, dtype='f8').T,
          coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.sum('i32')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.sum('i64')), out)
  
      out = xr.DataArray(
          np.nansum(values(_pandas().f64).reshape((2, 2, 5)), axis=2).T,
          coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.sum('f32')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.sum('f64')), out)
  
  
  def test_first(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      out = xr.DataArray([[0, 10], [5, 15]], coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.first('i32')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.first('i64')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.first('f32')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.first('f64')), out)
  
  
  def test_last(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      out = xr.DataArray([[4, 14], [9, 19]], coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.last('i32')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.last('i64')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.last('f32')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.last('f64')), out)
  
  
  def test_min(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      out = xr.DataArray(
          values(_pandas().i64).reshape((2, 2, 5)).min(axis=2).astype('f8').T,
          coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.min('i32')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.min('i64')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.min('f32')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.min('f64')), out)
  
  
  def test_max(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      out = xr.DataArray(
          values(_pandas().i64).reshape((2, 2, 5)).max(axis=2).astype('f8').T,
          coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.max('i32')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.max('i64')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.max('f32')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.max('f64')), out)
  
  
  def test_min_row_index(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      out = xr.DataArray([[0, 10], [5, 15]], coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds._min_row_index()), out)
  
  
  def test_max_row_index(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      out = xr.DataArray([[4, 14], [9, 19]], coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds._max_row_index()), out)
  
  
  def test_min_n(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[-3, -1, 0, 4, nan, nan], [-13, -11, 10, 12, 14, nan]],
                           [[-9, -7, -5, 6, 8, nan], [-19, -17, -15, 16, 18, nan]]])
      for n in range(1, 7):
          agg = c.points(ddf, 'x', 'y', ds.min_n('plusminus', n=n))
          out = solution[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data, c.points(ddf, 'x', 'y', ds.min('plusminus')).data)
  
  
  def test_max_n(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[4, 0, -1, -3, nan, nan], [14, 12, 10, -11, -13, nan]],
                           [[8, 6, -5, -7, -9, nan], [18, 16, -15, -17, -19, nan]]])
      for n in range(1, 7):
          agg = c.points(ddf, 'x', 'y', ds.max_n('plusminus', n=n))
          out = solution[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data, c.points(ddf, 'x', 'y', ds.max('plusminus')).data)
  
  
  def test_min_n_row_index(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[0, 1, 2, 3, 4, -1], [10, 11, 12, 13, 14, -1]],
                           [[5, 6, 7, 8, 9, -1], [15, 16, 17, 18, 19, -1]]])
      for n in range(1, 7):
          agg = c.points(ddf, 'x', 'y', ds._min_n_row_index(n=n))
          out = solution[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data, c.points(ddf, 'x', 'y', ds._min_row_index()).data)
  
  
  def test_max_n_row_index(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[4, 3, 2, 1, 0, -1], [14, 13, 12, 11, 10, -1]],
                           [[9, 8, 7, 6, 5, -1], [19, 18, 17, 16, 15, -1]]])
      for n in range(1, 7):
          agg = c.points(ddf, 'x', 'y', ds._max_n_row_index(n=n))
          out = solution[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data, c.points(ddf, 'x', 'y', ds._max_row_index()).data)
  
  
  def test_first_n(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[0, -1, -3, 4, nan, nan], [10, -11, 12, -13, 14, nan]],
                           [[-5, 6, -7, 8, -9, nan], [-15, 16, -17, 18, -19, nan]]])
      for n in range(1, 7):
          agg = c.points(ddf, 'x', 'y', ds.first_n('plusminus', n=n))
          out = solution[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data,
                                c.points(ddf, 'x', 'y', ds.first('plusminus')).data)
  
  
  def test_last_n(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[4, -3, -1, 0, nan, nan], [14, -13, 12, -11, 10, nan]],
                           [[-9, 8, -7, 6, -5, nan], [-19, 18, -17, 16, -15, nan]]])
      for n in range(1, 7):
          agg = c.points(ddf, 'x', 'y', ds.last_n('plusminus', n=n))
          out = solution[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data, c.points(ddf, 'x', 'y', ds.last('plusminus')).data)
  
  
  def test_categorical_count(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      sol = np.array([[[2, 1, 1, 1], [1, 1, 2, 1]], [[1, 2, 1, 1], [1, 1, 1, 2]]], dtype=np.uint32)
      assert_eq_ndarray(c.points(ddf, 'x', 'y', ds.by('cat2')).data, sol)
  
      # ds.summary(name=ds.by("cat2")) should give same result as ds.by("cat2"). Issue 1252
      dataset = c.points(ddf, 'x', 'y', ds.summary(name=ds.by('cat2')))
      assert_eq_ndarray(dataset["name"].data, sol)
  
  
  def test_categorical_min(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      sol_int = np.array([[[0, 1, 2, 3], [12, 13, 10, 11]], [[8, 5, 6, 7], [16, 17, 18, 15]]],
                         dtype=np.float64)
      sol_float = np.array([[[0, 1, nan, 3], [12, 13, 10, 11]], [[8, 5, 6, 7], [16, 17, 18, 15]]])
      assert_eq_ndarray(c.points(ddf, 'x', 'y', ds.by('cat2', ds.min('i32'))).data, sol_int)
      assert_eq_ndarray(c.points(ddf, 'x', 'y', ds.by('cat2', ds.min('i64'))).data, sol_int)
      assert_eq_ndarray(c.points(ddf, 'x', 'y', ds.by('cat2', ds.min('f32'))).data, sol_float)
      assert_eq_ndarray(c.points(ddf, 'x', 'y', ds.by('cat2', ds.min('f64'))).data, sol_float)
  
  
  def test_categorical_max(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      sol_int = np.array([[[4, 1, 2, 3], [12, 13, 14, 11]], [[8, 9, 6, 7], [16, 17, 18, 19]]],
                         dtype=np.float64)
      sol_float = np.array([[[4, 1, nan, 3], [12, 13, 14, 11]], [[8, 9, 6, 7], [16, 17, 18, 19]]])
      assert_eq_ndarray(c.points(ddf, 'x', 'y', ds.by('cat2', ds.max('i32'))).data, sol_int)
      assert_eq_ndarray(c.points(ddf, 'x', 'y', ds.by('cat2', ds.max('i64'))).data, sol_int)
      assert_eq_ndarray(c.points(ddf, 'x', 'y', ds.by('cat2', ds.max('f32'))).data, sol_float)
      assert_eq_ndarray(c.points(ddf, 'x', 'y', ds.by('cat2', ds.max('f64'))).data, sol_float)
  
  
  def test_categorical_min_n(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[[0, 4, nan], [1, nan, nan], [nan, nan, nan], [3, nan, nan]],
                            [[12, nan, nan], [13, nan, nan], [10, 14, nan], [11, nan, nan]]],
                           [[[8, nan, nan], [5, 9, nan], [6, nan, nan], [7, nan, nan]],
                            [[16, nan, nan], [17, nan, nan], [18, nan, nan], [15, 19, nan]]]])
      for n in range(1, 3):
          agg = c.points(ddf, 'x', 'y', ds.by('cat2', ds.min_n('f32', n=n)))
          out = solution[:, :, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[..., 0].data,
                                c.points(ddf, 'x', 'y', ds.by('cat2', ds.min('f32'))).data)
  
  
  def test_categorical_max_n(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[[4, 0, nan], [1, nan, nan], [nan, nan, nan], [3, nan, nan]],
                            [[12, nan, nan], [13, nan, nan], [14, 10, nan], [11, nan, nan]]],
                           [[[8, nan, nan], [9, 5, nan], [6, nan, nan], [7, nan, nan]],
                            [[16, nan, nan], [17, nan, nan], [18, nan, nan], [19, 15, nan]]]])
      for n in range(1, 3):
          agg = c.points(ddf, 'x', 'y', ds.by('cat2', ds.max_n('f32', n=n)))
          out = solution[:, :, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[..., 0].data,
                                c.points(ddf, 'x', 'y', ds.by('cat2', ds.max('f32'))).data)
  
  
  def test_categorical_min_row_index(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[0, 1, 2, 3], [12, 13, 10, 11]], [[8, 5, 6, 7], [16, 17, 18, 15]]])
      agg = c.points(ddf, 'x', 'y', ds.by('cat2', ds._min_row_index()))
      assert_eq_ndarray(agg.data, solution)
  
  
  def test_categorical_max_row_index(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[4, 1, 2, 3], [12, 13, 14, 11]], [[8, 9, 6, 7], [16, 17, 18, 19]]])
      agg = c.points(ddf, 'x', 'y', ds.by('cat2', ds._max_row_index()))
      assert_eq_ndarray(agg.data, solution)
  
  
  def test_categorical_min_n_row_index(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[[0, 4, -1], [1, -1, -1], [2, -1, -1], [3, -1, -1]],
                            [[12, -1, -1], [13, -1, -1], [10, 14, -1], [11, -1, -1]]],
                           [[[8, -1, -1], [5, 9, -1], [6, -1, -1], [7, -1, -1]],
                            [[16, -1, -1], [17, -1, -1], [18, -1, -1], [15, 19, -1]]]])
      for n in range(1, 3):
          agg = c.points(ddf, 'x', 'y', ds.by('cat2', ds._min_n_row_index(n=n)))
          out = solution[:, :, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[..., 0].data,
                                c.points(ddf, 'x', 'y', ds.by('cat2', ds._min_row_index())).data)
  
  
  def test_categorical_max_n_row_index(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[[4, 0, -1], [1, -1, -1], [2, -1, -1], [3, -1, -1]],
                            [[12, -1, -1], [13, -1, -1], [14, 10, -1], [11, -1, -1]]],
                           [[[8, -1, -1], [9, 5, -1], [6, -1, -1], [7, -1, -1]],
                            [[16, -1, -1], [17, -1, -1], [18, -1, -1], [19, 15, -1]]]])
      for n in range(1, 3):
          agg = c.points(ddf, 'x', 'y', ds.by('cat2', ds._max_n_row_index(n=n)))
          out = solution[:, :, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[..., 0].data,
                                c.points(ddf, 'x', 'y', ds.by('cat2', ds._max_row_index())).data)
  
  
  def test_categorical_first(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[0, -1, nan, -3],
                            [12, -13, 10, -11]],
                           [[8, -5, 6, -7],
                            [16, -17, 18, -15]]])
      for n in range(1, 3):
          agg = c.points(ddf, 'x', 'y', ds.by('cat2', ds.first("plusminus")))
          assert_eq_ndarray(agg.data, solution)
  
  
  def test_categorical_last(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[4, -1, nan, -3],
                            [12, -13, 14, -11]],
                           [[8, -9, 6, -7],
                            [16, -17, 18, -19]]])
      for n in range(1, 3):
          agg = c.points(ddf, 'x', 'y', ds.by('cat2', ds.last("plusminus")))
          assert_eq_ndarray(agg.data, solution)
  
  
  def test_categorical_first_n(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[[0, 4, nan], [-1, nan, nan], [nan, nan, nan], [-3, nan, nan]],
                            [[12, nan, nan], [-13, nan, nan], [10, 14, nan], [-11, nan, nan]]],
                           [[[8, nan, nan], [-5, -9, nan], [6, nan, nan], [-7, nan, nan]],
                            [[16, nan, nan], [-17, nan, nan], [18, nan, nan], [-15, -19, nan]]]])
      for n in range(1, 3):
          agg = c.points(ddf, 'x', 'y', ds.by('cat2', ds.first_n("plusminus", n=n)))
          out = solution[:, :, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[..., 0].data,
                                c.points(ddf, 'x', 'y', ds.by('cat2', ds.first("plusminus"))).data)
  
  
  def test_categorical_last_n(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      solution = np.array([[[[4, 0, nan], [-1, nan, nan], [nan, nan, nan], [-3, nan, nan]],
                            [[12, nan, nan], [-13, nan, nan], [14, 10, nan], [-11, nan, nan]]],
                           [[[8, nan, nan], [-9, -5, nan], [6, nan, nan], [-7, nan, nan]],
                            [[16, nan, nan], [-17, nan, nan], [18, nan, nan], [-19, -15, nan]]]])
      for n in range(1, 3):
          agg = c.points(ddf, 'x', 'y', ds.by('cat2', ds.last_n("plusminus", n=n)))
          out = solution[:, :, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[..., 0].data,
                                c.points(ddf, 'x', 'y', ds.by('cat2', ds.last("plusminus"))).data)
  
  
  def test_where_max(ddf, npartitions):
      # Important to test with npartitions > 2 to have multiple combination stages.
      # Identical results to equivalent pandas test.
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      out = xr.DataArray([[16, 6], [11, 1]], coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.max('i32'), 'reverse')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.max('i64'), 'reverse')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.max('f32'), 'reverse')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.max('f64'), 'reverse')), out)
  
      # Using row index.
      out = xr.DataArray([[4, 14], [9, 19]], coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.max('i32'))), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.max('i64'))), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.max('f64'))), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.max('f32'))), out)
  
  
  def test_where_min(ddf, npartitions):
      # Important to test with npartitions > 2 to have multiple combination stages.
      # Identical results to equivalent pandas test.
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      out = xr.DataArray([[20, 10], [15, 5]], coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.min('i32'), 'reverse')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.min('i64'), 'reverse')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.min('f32'), 'reverse')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.min('f64'), 'reverse')), out)
  
      # Using row index.
      out = xr.DataArray([[0, 10], [5, 15]], coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.min('i32'))), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.min('i64'))), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.min('f64'))), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.min('f32'))), out)
  
  
  def test_where_max_n(ddf, npartitions):
      # Important to test with npartitions > 2 to have multiple combination stages.
      # Identical results to equivalent pandas test.
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      sol_rowindex = np.array([[[ 4,  0,  1,  3, -1, -1],
                                [14, 12, 10, 11, 13, -1]],
                               [[ 8,  6,  5,  7,  9, -1],
                                [18, 16, 15, 17, 19, -1]]])
      sol_reverse = np.where(np.logical_or(sol_rowindex < 0, sol_rowindex == 6), np.nan,
                             20 - sol_rowindex)
  
      for n in range(1, 7):
          # Using row index.
          agg = c.points(ddf, 'x', 'y', ds.where(ds.max_n('plusminus', n=n)))
          out = sol_rowindex[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data,
                                c.points(ddf, 'x', 'y', ds.where(ds.max('plusminus'))).data)
  
          # Using another column
          agg = c.points(ddf, 'x', 'y', ds.where(ds.max_n('plusminus', n=n), 'reverse'))
          out = sol_reverse[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data,
                                c.points(ddf, 'x', 'y', ds.where(ds.max('plusminus'),
                                                                 'reverse')).data)
  
  
  def test_where_min_n(ddf, npartitions):
      # Important to test with npartitions > 2 to have multiple combination stages.
      # Identical results to equivalent pandas test.
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      sol_rowindex = np.array([[[3,  1,  0,  4, -1, -1],
                                [13, 11, 10, 12, 14, -1]],
                               [[ 9,  7,  5,  6,  8, -1],
                                [19, 17, 15, 16, 18, -1]]])
      sol_reverse = np.where(np.logical_or(sol_rowindex < 0, sol_rowindex == 6), np.nan,
                             20 - sol_rowindex)
  
      for n in range(1, 7):
          # Using row index.
          agg = c.points(ddf, 'x', 'y', ds.where(ds.min_n('plusminus', n=n)))
          out = sol_rowindex[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data,
                                c.points(ddf, 'x', 'y', ds.where(ds.min('plusminus'))).data)
  
          # Using another column
          agg = c.points(ddf, 'x', 'y', ds.where(ds.min_n('plusminus', n=n), 'reverse'))
          out = sol_reverse[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data,
                                c.points(ddf, 'x', 'y', ds.where(ds.min('plusminus'),
                                                                 'reverse')).data)
  
  
  def test_where_first(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      # Note reductions like ds.where(ds.first('i32'), 'reverse') are supported,
      # but the same results can be achieved using the simpler ds.first('reverse')
      out = xr.DataArray([[20, 10], [15, 5]], coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.first('i32'), 'reverse')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.first('i64'), 'reverse')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.first('f32'), 'reverse')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.first('f64'), 'reverse')), out)
  
      # Using row index.
      out = xr.DataArray([[0, 10], [5, 15]], coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.first('i32'))), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.first('i64'))), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.first('f64'))), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.first('f32'))), out)
  
  
  def test_where_last(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      # Note reductions like ds.where(ds.last('i32'), 'reverse') are supported,
      # but the same results can be achieved using the simpler ds.last('reverse')
      out = xr.DataArray([[16, 6], [11, 1]], coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.last('i32'), 'reverse')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.last('i64'), 'reverse')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.last('f32'), 'reverse')), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.last('f64'), 'reverse')), out)
  
      # Using row index.
      out = xr.DataArray([[4, 14], [9, 19]], coords=coords, dims=dims)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.last('i32'))), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.last('i64'))), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.last('f64'))), out)
      assert_eq_xr(c.points(ddf, 'x', 'y', ds.where(ds.last('f32'))), out)
  
  
  def test_where_first_n(ddf, npartitions):
      # Important to test with npartitions > 2 to have multiple combination stages.
      # Identical results to equivalent pandas test.
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      sol_rowindex = np.array([[[ 0,  1,  3,  4, -1, -1],
                                [10, 11, 12, 13, 14, -1]],
                               [[ 5,  6,  7,  8,  9, -1],
                                [15, 16, 17, 18, 19, -1]]])
      sol_reverse = np.where(np.logical_or(sol_rowindex < 0, sol_rowindex == 6), np.nan,
                             20 - sol_rowindex)
  
      for n in range(1, 7):
          # Using row index.
          agg = c.points(ddf, 'x', 'y', ds.where(ds.first_n('plusminus', n=n)))
          out = sol_rowindex[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data,
                                c.points(ddf, 'x', 'y', ds.where(ds.first('plusminus'))).data)
  
          # Using another column
          agg = c.points(ddf, 'x', 'y', ds.where(ds.first_n('plusminus', n=n), 'reverse'))
          out = sol_reverse[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data,
                                c.points(ddf, 'x', 'y', ds.where(ds.first('plusminus'),
                                                                 'reverse')).data)
  
  
  def test_where_last_n(ddf, npartitions):
      # Important to test with npartitions > 2 to have multiple combination stages.
      # Identical results to equivalent pandas test.
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      sol_rowindex = np.array([[[ 4,  3,  1,  0, -1, -1],
                                [14, 13, 12, 11, 10, -1]],
                               [[ 9,  8,  7,  6,  5, -1],
                                [19, 18, 17, 16, 15, -1]]])
      sol_reverse = np.where(np.logical_or(sol_rowindex < 0, sol_rowindex == 6), np.nan,
                             20 - sol_rowindex)
  
      for n in range(1, 7):
          # Using row index.
          agg = c.points(ddf, 'x', 'y', ds.where(ds.last_n('plusminus', n=n)))
          out = sol_rowindex[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data,
                                c.points(ddf, 'x', 'y', ds.where(ds.last('plusminus'))).data)
  
          # Using another column
          agg = c.points(ddf, 'x', 'y', ds.where(ds.last_n('plusminus', n=n), 'reverse'))
          out = sol_reverse[:, :, :n]
          assert_eq_ndarray(agg.data, out)
          if n == 1:
              assert_eq_ndarray(agg[:, :, 0].data,
                                c.points(ddf, 'x', 'y', ds.where(ds.last('plusminus'),
                                                                 'reverse')).data)
  
  def test_summary_by(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
  
      # summary(by)
      agg_summary = c.points(ddf, 'x', 'y', ds.summary(by=ds.by("cat")))
      agg_by = c.points(ddf, 'x', 'y', ds.by("cat"))
      assert_eq_xr(agg_summary["by"], agg_by)
  
      # summary(by, other_reduction)
      agg_summary = c.points(ddf, 'x', 'y', ds.summary(by=ds.by("cat"), max=ds.max("plusminus")))
      agg_max = c.points(ddf, 'x', 'y', ds.max("plusminus"))
      assert_eq_xr(agg_summary["by"], agg_by)
      assert_eq_xr(agg_summary["max"], agg_max)
  
      # summary(other_reduction, by)
      agg_summary = c.points(ddf, 'x', 'y', ds.summary(max=ds.max("plusminus"), by=ds.by("cat")))
      assert_eq_xr(agg_summary["by"], agg_by)
      assert_eq_xr(agg_summary["max"], agg_max)
  
      # summary(by, by)
      agg_summary = c.points(ddf, 'x', 'y', ds.summary(by=ds.by("cat"),
                                                       by_any=ds.by("cat", ds.any())))
      agg_by_any = c.points(ddf, 'x', 'y', ds.by("cat", ds.any()))
      assert_eq_xr(agg_summary["by"], agg_by)
      assert_eq_xr(agg_summary["by_any"], agg_by_any)
  
      # summary(by("cat1"), by("cat2"))
      agg_summary = c.points(ddf, 'x', 'y', ds.summary(by=ds.by("cat"), by2=ds.by("cat2")))
      agg_by2 = c.points(ddf, 'x', 'y', ds.by("cat2"))
      assert_eq_xr(agg_summary["by"], agg_by)
      assert_eq_xr(agg_summary["by2"], agg_by2)
  
  
  def test_summary_where_n(ddf, npartitions):
      # Important to test with npartitions > 2 to have multiple combination stages.
      # Identical results to equivalent pandas test.
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
      sol_min_n_rowindex = np.array([[[3,  1,  0,  4, -1],
                                      [13, 11, 10, 12, 14]],
                                     [[ 9,  7,  5,  6,  8],
                                      [19, 17, 15, 16, 18]]])
      sol_max_n_rowindex = np.array([[[ 4,  0,  1,  3, -1],
                                      [14, 12, 10, 11, 13]],
                                     [[ 8,  6,  5,  7,  9],
                                      [18, 16, 15, 17, 19]]])
      sol_max_n_reverse = np.where(np.logical_or(sol_max_n_rowindex < 0, sol_max_n_rowindex == 6),
                                   np.nan, 20 - sol_max_n_rowindex)
  
      agg = c.points(ddf, 'x', 'y', ds.summary(
          count=ds.count(),
          min_n=ds.where(ds.min_n('plusminus', 5)),
          max_n=ds.where(ds.max_n('plusminus', 5), 'reverse'),
      ))
      assert_eq_ndarray(agg.coords['n'], np.arange(5))
  
      assert agg['count'].dims == ('y', 'x')
      assert agg['min_n'].dims == ('y', 'x', 'n')
      assert agg['max_n'].dims == ('y', 'x', 'n')
  
      assert agg['count'].dtype == np.dtype('uint32')
      assert agg['min_n'].dtype == np.dtype('int64')
      assert agg['max_n'].dtype == np.dtype('float64')
  
      assert_eq_ndarray(agg['count'].data, [[5, 5], [5, 5]])
      assert_eq_ndarray(agg['min_n'].data, sol_min_n_rowindex)
      assert_eq_ndarray(agg['max_n'].data, sol_max_n_reverse)
  
  
  def test_mean(ddf, npartitions):
      ddf = ddf.repartition(npartitions=npartitions)
      assert ddf.npartitions == npartitions
  