Fix pyarrow and dask parquet issues (#92)

* Create test data files using pyarrow 5.0.0 and dask 2021.7.2

* Fix pyarrow 5.0.0 warnings

* Remove metadata access warnings pyarrow 5.0.0

* Changes to work with dask 2021.8.0 and pyarrow 5.0.0

* Changes to work with dask 2022.7.1

* Up to date, with test parquet files for pyarrow 5.0.0 and 8.0.0

* Reduce size of parquet test files and add them to MANIFEST

* Fix linting

* Increase test matrix

* Rename license_file to license_files

* Remove test warnings

* Set environment channels in CI test.yaml

* Improvements to build system when using pip

Author: ianthomas23

.github/workflows/build.yaml

@@ -28,8 +28,8 @@ jobs:
         shell: bash -l {0}
     env:
       CHANS_DEV: "-c pyviz/label/dev  -c conda-forge"
-      PKG_TEST_PYTHON: "--test-python=py37"
-      PYTHON_VERSION: "3.7"
+      PKG_TEST_PYTHON: "--test-python=py39"
+      PYTHON_VERSION: "3.9"
       CHANS: "-c pyviz"
       CONDA_UPLOAD_TOKEN: ${{ secrets.CONDA_UPLOAD_TOKEN }}
     steps:
@@ -42,7 +42,7 @@ jobs:
       - uses: conda-incubator/setup-miniconda@v2
         with:
           miniconda-version: "latest"
-          python-version: 3.8
+          python-version: 3.9
       - name: Set output
         id: vars
         run: echo ::set-output name=tag::${GITHUB_REF#refs/*/}
@@ -67,8 +67,8 @@ jobs:
         shell: bash -l {0}
     env:
       CHANS_DEV: "-c pyviz/label/dev -c conda-forge"
-      PKG_TEST_PYTHON: "--test-python=py37"
-      PYTHON_VERSION: "3.7"
+      PKG_TEST_PYTHON: "--test-python=py39"
+      PYTHON_VERSION: "3.9"
       CHANS: "-c pyviz"
       PPU: ${{ secrets.PPU }}
       PPP: ${{ secrets.PPP }}

.github/workflows/test.yaml

@@ -17,25 +17,14 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: ['ubuntu-latest']
-        python-version: [3.8]
-        exclude:
-          - os: macos-latest
-            python-version: 3.6
-          - os: macos-latest
-            python-version: 3.8
-            # Fiona for Python 3.6 on Windows has build issues
-            # See https://github.com/conda-forge/fiona-feedstock/issues/171 for more details
-          - os: windows-latest
-            python-version: 3.6
+        os: ['ubuntu-latest', 'macos-latest', 'windows-latest']
+        python-version: [3.7, 3.8, 3.9, '3.10']
     timeout-minutes: 60
     defaults:
       run:
         shell: bash -l {0}
     env:
       PYTHON_VERSION: ${{ matrix.python-version }}
-      CHANS_DEV: "-c conda-forge -c pyviz/label/dev"
-      CHANS: "-c pyviz"
       GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
     steps:
       - uses: actions/checkout@v2
@@ -53,16 +42,18 @@ jobs:
       - name: conda setup
         run: |
           conda config --set always_yes True
-          conda install -c pyviz "pyctdev>=0.5"
+          conda install -c pyviz pyctdev
           doit ecosystem_setup
-          doit env_create ${{ env.CHANS_DEV}} --python=${{ matrix.python-version }}
+          conda create -n test-environment python=${{ matrix.python-version }}
+          conda activate test-environment
+          conda config --env --append channels pyviz/label/dev --append channels conda-forge
+          conda install pyctdev
       - name: doit develop_install
         run: |
           eval "$(conda shell.bash hook)"
           conda activate test-environment
-          conda install ${{ env.CHANS_DEV }} "pip<21.2.1"
           conda list
-          doit develop_install ${{ env.CHANS_DEV }} -o tests
+          doit develop_install -o tests
       - name: doit env_capture
         run: |
           eval "$(conda shell.bash hook)"

.gitignore

@@ -103,7 +103,6 @@ venv.bak/
 # mypy
 .mypy_cache/
 
-*.parq
 .idea/
 spatialpandas/.version
 

MANIFEST.in

@@ -3,6 +3,7 @@ include NOTICE
 include README.md
 include CHANGELOG.md
 include spatialpandas/.version
+graft spatialpandas/tests/test_data
 global-exclude *.py[co]
 global-exclude *~
 global-exclude *.ipynb_checkpoints/*

setup.cfg

@@ -1,5 +1,9 @@
 [metadata]
-license_file = LICENSE
+license_files = LICENSE
 
 [wheel]
 universal = 1
+
+[tool:pyctdev.conda]
+namespace_map =
+    geopandas=geopandas-base

setup.py

@@ -7,16 +7,15 @@
         'codecov',
         'flake8',
         'hilbertcurve',
-        'geopandas-base',
+        'geopandas',
         'hypothesis',
-        'libstdcxx-ng >=12',
+        'keyring',
         'pytest-cov',
         'pytest',
+        'rfc3986',
         'scipy',
         'shapely',
         'twine',
-        'rfc3986',
-        'keyring',
     ],
     'examples': [
         'datashader',
@@ -28,15 +27,14 @@
 }
 
 install_requires = [
+    'dask[complete]',
     'fsspec',
     'numba',
-    'pandas >=0.25',
+    'pandas',
     'param',
     'pyarrow >=1.0',
     'python-snappy',
     'retrying',
-    'numpy',
-    'dask[complete] >=2.0'
 ]
 
 setup_args = dict(
@@ -52,7 +50,7 @@
     url='https://github.com/holoviz/spatialpandas',
     maintainer='Datashader developers',
     maintainer_email='dev@datashader.org',
-    python_requires='>=3.6',
+    python_requires='>=3.7',
     install_requires=install_requires,
     extras_require=extras_require,
     tests_require=extras_require['tests'],

spatialpandas/geometry/multipolygon.py

@@ -59,7 +59,7 @@ def to_shapely(self):
             shapely MultiPolygon shape
         """
         import shapely.geometry as sg
-        polygon_arrays = np.asarray(self.data.as_py())
+        polygon_arrays = np.asarray(self.data.as_py(), dtype=object)
 
         polygons = []
         for polygon_array in polygon_arrays:

spatialpandas/geometry/polygon.py

@@ -61,7 +61,7 @@ def to_shapely(self):
         """
         import shapely.geometry as sg
         ring_arrays = [np.asarray(line_coords).reshape(len(line_coords) // 2, 2)
-                       for line_coords in np.asarray(self.data.as_py())]
+                       for line_coords in np.asarray(self.data.as_py(), dtype=object)]
         rings = [sg.LinearRing(ring_array) for ring_array in ring_arrays]
         return sg.Polygon(shell=rings[0], holes=rings[1:])
 

spatialpandas/io/parquet.py

@@ -1,10 +1,9 @@
-import copy
 import json
 import pathlib
-from distutils.version import LooseVersion
 from functools import reduce
 from glob import has_magic
 from numbers import Number
+from packaging.version import Version
 from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
 
 import fsspec
@@ -31,7 +30,7 @@
 )
 
 # improve pandas compatibility, based on geopandas _compat.py
-PANDAS_GE_12 = str(pd.__version__) >= LooseVersion("1.2.0")
+PANDAS_GE_12 = Version(pd.__version__) >= Version("1.2.0")
 
 _geometry_dtypes = [
     PointDtype, MultiPointDtype, RingDtype, LineDtype,
@@ -63,16 +62,18 @@ def _load_parquet_pandas_metadata(
         pqds = pq.ParquetDataset(
             path,
             filesystem=filesystem,
-            validate_schema=False,
+            #validate_schema=False,
+            use_legacy_dataset=False,
             **engine_kwargs,
         )
-        common_metadata = pqds.common_metadata
-        if common_metadata is None:
-            # Get metadata for first piece
-            piece = pqds.pieces[0]
-            metadata = piece.get_metadata().metadata
-        else:
-            metadata = pqds.common_metadata.metadata
+        filename = pathlib.Path(pqds.files[0]).parent.joinpath("_common_metadata")
+        try:
+            common_metadata = pq.read_metadata(filename)
+        except FileNotFoundError:
+            # Common metadata doesn't exist, so get metadata for first piece instead
+            filename = pathlib.Path(pqds.files[0])
+            common_metadata = pq.read_metadata(filename)
+        metadata = common_metadata.metadata
     else:
         with filesystem.open(path) as f:
             pf = pq.ParquetFile(f)
@@ -198,7 +199,7 @@ def to_parquet_dask(
     **kwargs: Any,
 ) -> None:
     engine_kwargs = engine_kwargs or {}
-    
+
     if not isinstance(ddf, DaskGeoDataFrame):
         raise TypeError(f"Expected DaskGeoDataFrame not {type(ddf)}")
     filesystem = validate_coerce_filesystem(path, filesystem, storage_options)
@@ -207,48 +208,27 @@ def to_parquet_dask(
             path and filesystem.isdir(path):
         filesystem.rm(path, recursive=True)
 
-    dd_to_parquet(
-        ddf,
-        path,
-        engine="pyarrow",
-        compression=compression,
-        storage_options=storage_options,
-        **kwargs,
-    )
-
-    # Write partition bounding boxes to the _metadata file
+    # Determine partition bounding boxes to save to _metadata file
     partition_bounds = {}
     for series_name in ddf.columns:
         series = ddf[series_name]
         if isinstance(series.dtype, GeometryDtype):
-            if series._partition_bounds is None:
-                # Bounds are not already computed. Compute bounds from the parquet file
-                # that was just written.
-                filesystem.invalidate_cache(path)
-                series = read_parquet_dask(
-                    path,
-                    columns=[series_name],
-                    filesystem=filesystem,
-                    load_divisions=False,
-                    storage_options=storage_options,
-                )[series_name]
             partition_bounds[series_name] = series.partition_bounds.to_dict()
 
     spatial_metadata = {'partition_bounds': partition_bounds}
     b_spatial_metadata = json.dumps(spatial_metadata).encode('utf')
 
-    pqds = pq.ParquetDataset(
+    dd_to_parquet(
+        ddf,
         path,
-        filesystem=filesystem,
-        validate_schema=False,
+        engine="pyarrow",
+        compression=compression,
+        storage_options=storage_options,
+        custom_metadata={b'spatialpandas': b_spatial_metadata},
+        write_metadata_file=True,
         **engine_kwargs,
+        **kwargs,
     )
-    all_metadata = copy.copy(pqds.common_metadata.metadata)
-    all_metadata[b'spatialpandas'] = b_spatial_metadata
-    schema = pqds.common_metadata.schema.to_arrow_schema()
-    new_schema = schema.with_metadata(all_metadata)
-    with filesystem.open(pqds.common_metadata_path, 'wb') as f:
-        pq.write_metadata(new_schema, f)
 
 
 def read_parquet_dask(
@@ -293,7 +273,7 @@ def read_parquet_dask(
         build_sindex : boolean
             Whether to build partition level spatial indexes to speed up indexing.
         storage_options: Key/value pairs to be passed on to the file-system backend, if any.
-        engine_kwargs: pyarrow.parquet engine-related keyword arguments. 
+        engine_kwargs: pyarrow.parquet engine-related keyword arguments.
     Returns:
     DaskGeoDataFrame
     """
@@ -357,7 +337,8 @@ def _perform_read_parquet_dask(
         pa.parquet.ParquetDataset(
             path,
             filesystem=filesystem,
-            validate_schema=False,
+            #validate_schema=False,
+            use_legacy_dataset=False,
             **engine_kwargs,
         ) for path in paths
     ]
@@ -366,7 +347,7 @@ def _perform_read_parquet_dask(
     pieces = []
     for dataset in datasets:
         # Perform natural sort on pieces so that "part.10" comes after "part.2"
-        dataset_pieces = sorted(dataset.pieces, key=lambda piece: natural_sort_key(piece.path))
+        dataset_pieces = sorted(dataset.fragments, key=lambda piece: natural_sort_key(piece.path))
         pieces.extend(dataset_pieces)
 
     delayed_partitions = [
@@ -419,7 +400,7 @@ def _perform_read_parquet_dask(
         cols_no_index = None
 
     meta = dd_read_parquet(
-        paths[0],
+        datasets[0].files[0],
         columns=cols_no_index,
         filesystem=filesystem,
         engine='pyarrow',
@@ -514,10 +495,15 @@ def _perform_read_parquet_dask(
 
 def _load_partition_bounds(pqds):
     partition_bounds = None
-    if (pqds.common_metadata is not None and
-            b'spatialpandas' in pqds.common_metadata.metadata):
+    filename = pathlib.Path(pqds.files[0]).parent.joinpath("_common_metadata")
+    try:
+        common_metadata = pq.read_metadata(filename)
+    except FileNotFoundError:
+        common_metadata = None
+
+    if common_metadata is not None and b'spatialpandas' in common_metadata.metadata:
         spatial_metadata = json.loads(
-            pqds.common_metadata.metadata[b'spatialpandas'].decode('utf')
+            common_metadata.metadata[b'spatialpandas'].decode('utf')
         )
         if "partition_bounds" in spatial_metadata:
             partition_bounds = {}