Repository: geopandas/dask-geopandas

Issue:
Drop distributed as a required dependency?
Continuing discussion from https://github.com/conda-forge/dask-geopandas-feedstock/pull/15#discussion_r1294298427 on whether [`distributed`](https://distributed.dask.org/en/stable/) should be a required dependency of dask-geopandas. Currently it is listed as required in setup.py at https://github.com/geopandas/dask-geopandas/blame/2d0abad32e324a494b34738ab70feb6d78cc0942/setup.py#L16

Looking at the git history, it seems like distributed has been listed as required since 2020 - 7a49aefb44dda954494feb902c9e9378be7372cd.

Discussion:

Patch:
diff --git a/ci/envs/310-no-optional-deps.yaml b/ci/envs/310-no-optional-deps.yaml
index bc9d0a9..99fa647 100644
--- a/ci/envs/310-no-optional-deps.yaml
+++ b/ci/envs/310-no-optional-deps.yaml
@@ -5,7 +5,6 @@ dependencies:
   # required dependencies
   - python=3.10
   - dask
-  - distributed
   - geopandas
   - pyproj
   - packaging
diff --git a/dask_geopandas/io/file.py b/dask_geopandas/io/file.py
index 5bd634d..3870d84 100644
--- a/dask_geopandas/io/file.py
+++ b/dask_geopandas/io/file.py
@@ -114,7 +114,7 @@ def read_file(
         read_geometry = False
     meta = pyogrio.read_dataframe(
         path, layer=layer, columns=columns, read_geometry=read_geometry, max_features=5
-    )
+    ).head(0)
 
     # Define parts
     parts = []
diff --git a/pyproject.toml b/pyproject.toml
index ed8d02a..a36c32d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -35,7 +35,6 @@ dependencies = [
     "geopandas>=0.12",
     "shapely>=2.0",
     "dask[dataframe]>=2022.06.0",
-    "distributed>=2022.06.0",
     "packaging",
 ]
 


Test patch:
diff --git a/dask_geopandas/tests/io/test_arrow.py b/dask_geopandas/tests/io/test_arrow.py
index 0778897..a971c6e 100644
--- a/dask_geopandas/tests/io/test_arrow.py
+++ b/dask_geopandas/tests/io/test_arrow.py
@@ -214,3 +214,15 @@ def test_index(tmp_path, naturalearth_lowres):
     assert "hilbert_distance" not in result.columns
     assert result.index.name is None
     assert result.index.compute()[0] == 0
+
+
+def test_read_meta_is_empty(tmp_path, naturalearth_lowres):
+    df = geopandas.read_file(naturalearth_lowres)
+
+    basedir = tmp_path / "dataset"
+    basedir.mkdir()
+    df.iloc[:100].to_feather(basedir / "data.0.feather")
+    df.iloc[100:].to_feather(basedir / "data.1.feather")
+
+    result = dask_geopandas.read_feather(basedir)
+    assert len(result._meta) == 0
diff --git a/dask_geopandas/tests/io/test_backend_integration.py b/dask_geopandas/tests/io/test_backend_integration.py
new file mode 100644
index 0000000..4e25a1a
--- /dev/null
+++ b/dask_geopandas/tests/io/test_backend_integration.py
@@ -0,0 +1,68 @@
+import geopandas
+
+import dask_geopandas
+
+import pytest
+from geopandas.testing import assert_geodataframe_equal
+
+try:
+    import pyogrio  # noqa: F401
+
+    PYOGRIO = True
+except ImportError:
+    PYOGRIO = False
+
+BACKENDS = ["arrow", "file", "parquet"]
+
+
+@pytest.fixture(params=BACKENDS)
+def backend(request):
+    param = request.param
+    if not PYOGRIO and param == "file":
+        pytest.skip("Unable to import pyogrio for file backend")
+    return param
+
+
+def from_arrow_backend(path, tmp_path, npartitions):
+    df = geopandas.read_file(path)
+    basedir = tmp_path / "dataset"
+    basedir.mkdir()
+    ddf = dask_geopandas.from_geopandas(df, npartitions=npartitions)
+    for i, part in enumerate(ddf.partitions):
+        part.compute().to_feather(basedir / f"data.{i}.feather")
+    return dask_geopandas.read_feather(basedir)
+
+
+def from_file_backend(path, tmp_path, npartitions):
+    return dask_geopandas.read_file(path, npartitions=npartitions)
+
+
+def from_parquet_backend(path, tmp_path, npartitions):
+    ddf = dask_geopandas.from_geopandas(
+        geopandas.read_file(path), npartitions=npartitions
+    )
+    basedir = tmp_path / "dataset"
+    ddf.to_parquet(basedir)
+    return dask_geopandas.read_parquet(basedir)
+
+
+def get_from_backend(backend, data_path, tmp_path, npartitions=4):
+    if backend == "arrow":
+        ddf = from_arrow_backend(data_path, tmp_path, npartitions)
+    elif backend == "file":
+        ddf = from_file_backend(data_path, tmp_path, npartitions)
+    elif backend == "parquet":
+        ddf = from_parquet_backend(data_path, tmp_path, npartitions)
+    else:
+        raise ValueError()
+    return ddf
+
+
+def test_spatial_shuffle_integration(backend, naturalearth_lowres, tmp_path):
+    ddf = get_from_backend(backend, naturalearth_lowres, tmp_path)
+    new_idx = ddf.hilbert_distance()
+    expected = ddf.compute().set_index(new_idx.compute())
+
+    result = ddf.spatial_shuffle()
+    # Sort because the index is shuffled
+    assert_geodataframe_equal(result.compute().sort_index(), expected.sort_index())
diff --git a/dask_geopandas/tests/io/test_file.py b/dask_geopandas/tests/io/test_file.py
index 750e2c0..b69e53d 100644
--- a/dask_geopandas/tests/io/test_file.py
+++ b/dask_geopandas/tests/io/test_file.py
@@ -92,6 +92,12 @@ def test_read_file_columns(naturalearth_lowres):
     assert_geoseries_equal(result.compute(), df["geometry"])
 
 
+def test_read_file_meta_is_empty(naturalearth_lowres):
+    path = naturalearth_lowres
+    result = dask_geopandas.read_file(path, npartitions=4)
+    assert len(result._meta) == 0
+
+
 def test_read_file_layer(tmp_path):
     df_points = geopandas.GeoDataFrame(
         {
diff --git a/dask_geopandas/tests/io/test_parquet.py b/dask_geopandas/tests/io/test_parquet.py
index d135d6e..faa1653 100644
--- a/dask_geopandas/tests/io/test_parquet.py
+++ b/dask_geopandas/tests/io/test_parquet.py
@@ -248,3 +248,15 @@ def test_read_parquet_default_crs(tmp_path):
     result = dask_geopandas.read_parquet(filename)
     assert result.crs.equals(pyproj.CRS("OGC:CRS84"))
     assert result["other_geom"].crs.equals(pyproj.CRS("OGC:CRS84"))
+
+
+def test_read_parquet_meta_is_empty(tmp_path, naturalearth_lowres):
+    # basic roundtrip
+    df = geopandas.read_file(naturalearth_lowres)
+    ddf = dask_geopandas.from_geopandas(df, npartitions=4)
+
+    basedir = tmp_path / "dataset"
+    ddf.to_parquet(basedir)
+
+    result = dask_geopandas.read_parquet(basedir)
+    assert len(result._meta) == 0