Merge pull request #923 from davidhassell/parallel-read

Enable parallel netCDF dataset reads

Author: davidhassell

Changelog.rst

@@ -3,6 +3,13 @@ Version NEXTVERSION
 
 **2026-??-??**
 
+* New default backend for netCDF-4 in `cf.read` that allows parallel
+  reading: ``h5netcdf-pyfive``
+  (https://github.com/NCAS-CMS/cf-python/issues/912)
+* New optional backend for netCDF-3 in `cf.read` that allows parallel
+  reading: ``netcdf_file``
+  (https://github.com/NCAS-CMS/cf-python/issues/912)
+* Changed dependency: ``cfdm>=1.13.1.0, <1.13.2.0``
 * Fix for subspacing with cyclic `cf.wi` and `cf.wo` arguments
   (https://github.com/NCAS-CMS/cf-python/issues/887)
 

cf/__init__.py

@@ -153,10 +153,12 @@
     H5netcdfArray,
     NetCDFArray,
     NetCDF4Array,
+    PyfiveArray,
     PointTopologyArray,
     RaggedContiguousArray,
     RaggedIndexedArray,
     RaggedIndexedContiguousArray,
+    ScipyNetcdfFileArray,
     SubsampledArray,
     UMArray,
     ZarrArray,

cf/cfimplementation.py

@@ -35,9 +35,11 @@
     H5netcdfArray,
     NetCDF4Array,
     PointTopologyArray,
+    PyfiveArray,
     RaggedContiguousArray,
     RaggedIndexedArray,
     RaggedIndexedContiguousArray,
+    ScipyNetcdfFileArray,
     SubsampledArray,
     ZarrArray,
 )
@@ -147,7 +149,9 @@ def set_construct(self, parent, construct, axes=None, copy=True, **kwargs):
     GatheredArray=GatheredArray,
     H5netcdfArray=H5netcdfArray,
     NetCDF4Array=NetCDF4Array,
+    ScipyNetcdfFileArray=ScipyNetcdfFileArray,
     PointTopologyArray=PointTopologyArray,
+    PyfiveArray=PyfiveArray,
     Quantization=Quantization,
     RaggedContiguousArray=RaggedContiguousArray,
     RaggedIndexedArray=RaggedIndexedArray,
@@ -204,7 +208,9 @@ def implementation():
      'GatheredArray': cf.data.array.gatheredarray.GatheredArray,
      'H5netcdfArray': cf.data.array.h5netcdfarray.H5netcdfArray,
      'NetCDF4Array': cf.data.array.netcdf4array.NetCDF4Array,
+     'ScipyNetcdfFileArray': cf.data.array.scipynetcdffilearray.ScipyNetcdfFileArray,
      'PointTopologyArray': <class 'cf.data.array.pointtopologyarray.PointTopologyArray'>,
+     'PyfiveArray': cf.data.array.pyfivearray.PyfiveArray,
      'Quantization': cf.quantization.Quantization,
      'RaggedContiguousArray': cf.data.array.raggedcontiguousarray.RaggedContiguousArray,
      'RaggedIndexedArray': cf.data.array.raggedindexedarray.RaggedIndexedArray,

cf/data/array/__init__.py

@@ -6,7 +6,9 @@
 from .h5netcdfarray import H5netcdfArray
 from .netcdfarray import NetCDFArray
 from .netcdf4array import NetCDF4Array
+from .scipynetcdffilearray import ScipyNetcdfFileArray
 from .pointtopologyarray import PointTopologyArray
+from .pyfivearray import PyfiveArray
 from .raggedcontiguousarray import RaggedContiguousArray
 from .raggedindexedarray import RaggedIndexedArray
 from .raggedindexedcontiguousarray import RaggedIndexedContiguousArray

cf/data/array/h5netcdfarray.py

@@ -9,7 +9,7 @@ class H5netcdfArray(
     Container,
     cfdm.H5netcdfArray,
 ):
-    """A netCDF array accessed with `h5netcdf`.
+    """A netCDF array accessed with `h5netcdf` using the `h5py` backend.
 
     .. versionadded:: 3.16.3
 

cf/data/array/pyfivearray.py

@@ -0,0 +1,16 @@
+import cfdm
+
+from ...mixin_container import Container
+from .mixin import ActiveStorageMixin
+
+
+class PyfiveArray(
+    ActiveStorageMixin,
+    Container,
+    cfdm.PyfiveArray,
+):
+    """A netCDF array accessed with `pyfive`.
+
+    .. versionadded:: NEXTVERSION
+
+    """

cf/data/array/scipynetcdffilearray.py

@@ -0,0 +1,14 @@
+import cfdm
+
+from ...mixin_container import Container
+
+
+class ScipyNetcdfFileArray(
+    Container,
+    cfdm.ScipyNetcdfFileArray,
+):
+    """A netCDF-3 array accessed with `scipy.io.netcdf_file`.
+
+    .. versionadded:: NEXTVERSION
+
+    """

cf/data/collapse/collapse_active.py

@@ -186,35 +186,35 @@ def active_chunk_function(method, *args, **kwargs):
     #       with `cf.active_storage(True)`
     from activestorage import Active
 
-    filename = x.get_filename()
-    address = x.get_address()
-    max_requests = active_storage_max_requests()
+    info = is_log_level_info(logger)
+
+    max_requests = active_storage_max_requests().value
+    storage_options = None
+    address = None
+    dataset = x.get_variable(None)
+    if dataset is None:
+        # Dateaset is a string, not a variable object.
+        storage_options = x.get_storage_options()
+        address = x.get_address()
+        dataset = x.get_filename()
+
     active_kwargs = {
-        "uri": "/".join(filename.split("/")[3:]),
+        "dataset": dataset,
         "ncvar": address,
-        "storage_options": x.get_storage_options(),
+        "axis": axis,
+        "storage_options": storage_options,
         "active_storage_url": url,
-        "storage_type": "s3",
         "max_threads": max_requests,
     }
-    # WARNING: The "uri", "storage_options", and "storage_type" keys
-    #          of the `active_kwargs` dictionary are currently
-    #          formatted according to the whims of the `Active` class
-    #          (i.e. the pyfive branch of PyActiveStorage). Future
-    #          versions of `Active` will have a better API, that will
-    #          require improvements to `active_kwargs`.
 
     index = x.index()
-
-    details = (
-        f"{method!r} (file={filename}, address={address}, url={url}, "
-        f"Dask chunk={index})"
-    )
-
-    info = is_log_level_info(logger)
     if info:
         # Do some detailed logging
         start = time.time()
+        details = (
+            f"{method!r} (dataset={dataset!r}, ncvar={address}, "
+            f"Dask chunk={index})"
+        )
         logger.info(
             f"STARTED  active storage {details}: {datetime.datetime.now()}"
         )  # pragma: no cover
@@ -227,8 +227,7 @@ def active_chunk_function(method, *args, **kwargs):
     # reduction on the remote server
     #
     # WARNING: The `_version` API of `Active` is likely to change from
-    #          the current version (i.e. the pyfive branch of
-    #          PyActiveStorage)
+    #          the current version
     active._version = 2
 
     # ----------------------------------------------------------------

cf/data/utils.py

@@ -423,10 +423,12 @@ def collapse(
         kwargs["ddof"] = ddof
 
     # The applicable chunk function will have its own call to
-    # 'cfdm_to_memory', so we can set '_force_to_memory=False'. Also,
-    # setting _force_to_memory=False will ensure that any active
-    # storage operations are not compromised.
-    dx = d.to_dask_array(_force_to_memory=False)
+    # 'cfdm_to_memory', so we can set '_force_to_memory=False'.
+    # Setting _force_to_memory=False will also ensure that any active
+    # storage operations are not compromised. We can set
+    # _force_mask_hardness=False because collapse operations do not
+    # need to ever unset masked values.
+    dx = d.to_dask_array(_force_mask_hardness=False, _force_to_memory=False)
     dx = func(dx, **kwargs)
     d._set_dask(dx)
 

cf/flags.py

@@ -454,5 +454,5 @@ def sort(self):
 
         for attr in ("_flag_values", "_flag_meanings", "_flag_masks"):
             if hasattr(self, attr):
-                array = getattr(self, attr).view()
-                array[...] = array[indices]
+                array = getattr(self, attr)[indices]
+                setattr(self, attr, array)