Move remote Parquet tables under RemoteStore

Author: FrancescAlted

doc/guides/remote_tables.md

@@ -5,7 +5,7 @@ Fixed-width, `blosc2.utf8()`, batch-backed variable-length, list, struct/object,
 and dictionary columns are fetched on demand, including their null masks. A table
 inside a hierarchy can also be opened through `RemoteStore`.
 
-## Single-file Parquet (experimental)
+## Single-file Parquet
 
 `blosc2.open()` recognizes `.parquet` paths, including fsspec URLs, and returns
 a read-only `RemoteCTable`. Use `source_format="parquet"` for an extensionless
@@ -32,14 +32,28 @@ opens and cached reads need no connection to the source. Cached sources are
 assumed unchanged until `refresh()` is called. Use `lazy=False` to import the
 whole table eagerly.
 
+Parquet is a `RemoteStore` with one root CTable. Use
+`RemoteStore(url, allow_table_root=True)` and `store[""]` when a store operation
+needs to own the cache and traffic counters. A Parquet file accepts only the
+root selector (`""` or `"/"`); selecting a child path raises an error.
+`RemoteCTable(url)` and lazy `blosc2.open(url)` use the same owner. NONE retains
+no converted row groups, MEMORY shares the store budget, and DISK retains
+complete converted physical-column/row-group units. A small slice can therefore
+read a whole row group. Shared DISK caches support `read_cached_table()` and
+offline `trim_sparse_cache()` with the same aggregate allowance.
+Caches from the earlier Parquet prototype layout are not migrated; use a fresh
+`cache_dir` for this RemoteStore layout.
+
 HTTP servers must honor byte-range requests; fsspec raises a range-request error
 for servers that only return complete files. Download the file and import it
 locally when range access is unavailable.
 
-`table.save("reference.b2z")` creates a portable archive with data already in
+`table.save("reference.b2z")` creates a RemoteStore archive with data already in
 the cache. Opening it reuses that data and fetches missing groups on demand.
-Use `include_cache=False` to save only the table metadata. Older `.b2nd`
-references remain readable.
+Archives of local sources keep an absolute path to the local Parquet file.
+Use `include_cache=False` to save only the table metadata. `.b2z` exports use
+the RemoteStore version 1 manifest with `kind="parquet"`. Runtime filesystems, credentials,
+and storage options must be supplied again when reopening.
 
 Use `shared_cache=True` when separate processes need to share one `cache_dir`.
 They should use the same source and conversion options.

doc/reference/remotectable.rst

@@ -78,13 +78,13 @@ Parquet cache details
 ---------------------
 
 Parquet ``cache_dir`` stores each accessed physical field and row group as a
-native CTable directory beneath ``<source>.parquet--<hash>/<generation>.b2d/``.
-The generation manifest retains the table schema and row-group boundaries; a
-small index in ``cache_dir`` identifies its source revision. Warm opens use
-these local files without contacting the source. Cached sources are assumed
-immutable until ``refresh()`` is called. The generation path itself can be
-opened with :func:`blosc2.open` to recover the complete logical table and fetch
-uncached groups on demand.
+native CTable directory under the RemoteStore generation's ``parquet-groups``
+directory. The common manifest retains the footer, schema, conversion options,
+source marker, and row-group boundaries. Warm opens reuse this discovery and
+payload without contacting the source. Cached sources are assumed immutable
+until ``refresh()`` is called. New portable ``.b2z`` archives use the common
+RemoteStore manifest. Local Parquet sources use the same cache layout and may
+export ``.b2z`` references tied to the local source path.
 
 HTTP Parquet sources need byte-range support. A persistent cache also needs a
 source size and version marker, such as an ETag, modification time, or

doc/reference/remotestore.rst

@@ -3,7 +3,7 @@
 RemoteStore
 ===========
 
-``RemoteStore`` discovers a read-only B2Z, Zarr or HDF5 hierarchy and returns
+``RemoteStore`` discovers a read-only B2Z, Zarr, HDF5 or Parquet source and returns
 :ref:`RemoteArray` and :ref:`RemoteCTable` leaves. Groups and leaves share one
 source session: a B2Z archive, a native HDF5 index, or a Zarr store. Zarr listing
 remains lazy.
@@ -16,6 +16,11 @@ DISK accepts ``max_cache_bytes=None`` for unbounded retention.
 Sources must be immutable. Generic ``blosc2.open(..., lazy=True, dataset=...)``
 continues to open a single array.
 
+A Parquet file has one CTable at its root. Pass ``allow_table_root=True`` to
+retain a store handle, then use ``store[""]`` to access the table. Parquet has no
+child selectors. Converted physical-column/row-group caches share the store's
+policy, allowance, traffic counter, persistence, and refresh generation.
+
 For HDF5, ``hdf5_index=`` accepts a native index dictionary, local JSON path, or
 remote fsspec URL. An explicit index skips hierarchy discovery and must match the
 source URL and selected scope. See :doc:`../guides/remote_arrays` for the

src/blosc2/b2objects.py

@@ -128,10 +128,6 @@ def decode_b2object_payload(payload: dict[str, Any], *, carrier_path=None, carri
         if carrier is None:
             raise ValueError("A persisted RemoteArray requires its B2ND carrier")
         return blosc2.RemoteArray._from_payload(payload, carrier)
-    if kind == "remote_parquet":
-        from blosc2.remote_parquet import RemoteParquetCTable
-
-        return RemoteParquetCTable._from_payload(payload)
     if kind == "lazyexpr":
         return decode_structured_lazyexpr(payload, carrier_path=carrier_path)
     if kind == "lazyudf":

src/blosc2/core.py

@@ -769,6 +769,8 @@ def parse_container_url(
         return urlpath, dataset, "hdf5"
     if any(part.endswith(".zarr") for part in parts):
         return urlpath, dataset, "zarr"
+    if parsed.path.lower().endswith(".parquet"):
+        return urlpath, dataset, "parquet"
 
     return urlpath, dataset, None
 

src/blosc2/remote_ctable.py

@@ -11,6 +11,7 @@
 import operator
 import os
 
+import blosc2
 from blosc2.ctable import CTable
 from blosc2.ctable_storage import RemoteTableStorage
 from blosc2.remote_array import CACHE_POLICY_DEFAULT, RemoteMetadataMapping
@@ -110,40 +111,13 @@ def __new__(
         _filesystem_resolver=None,
         _batch_validator=None,
     ):
-        if source_format == "parquet" or (
+        parquet = source_format == "parquet" or (
             isinstance(urlpath, (str, os.PathLike))
             and os.fspath(urlpath).split("?", 1)[0].lower().endswith(".parquet")
-        ):
-            if source_format not in (None, "parquet"):
-                raise ValueError("source_format conflicts with the .parquet suffix")
-            from blosc2.remote_parquet import RemoteParquetCTable
-
-            return RemoteParquetCTable(
-                os.fspath(urlpath),
-                storage_options=storage_options,
-                parquet_options=parquet_options,
-                columns=columns,
-                max_rows=max_rows,
-                string_max_length=string_max_length,
-                null_storage=null_storage,
-                auto_null_sentinels=auto_null_sentinels,
-                separate_nested_cols=separate_nested_cols,
-                list_serializer=list_serializer,
-                blosc2_batch_size=blosc2_batch_size,
-                blosc2_items_per_block=blosc2_items_per_block,
-                batch_size=batch_size,
-                cparams=cparams,
-                dparams=dparams,
-                validate=validate,
-                max_cache_bytes=max_cache_bytes,
-                cache_policy=cache_policy,
-                cache_dir=cache_dir,
-                shared_cache=shared_cache,
-                max_concurrency=max_concurrency,
-                metadata_buffer_bytes=metadata_buffer_bytes,
-                row_buffer_bytes=row_buffer_bytes,
-            )
-        if (
+        )
+        if parquet and source_format not in (None, "parquet"):
+            raise ValueError("source_format conflicts with the .parquet suffix")
+        if not parquet and (
             source_format is not None
             or any(
                 value is not None
@@ -180,20 +154,61 @@ def __new__(
 
         from blosc2.remote_store import RemoteStore
 
-        store = RemoteStore(
-            urlpath,
-            dataset=dataset,
-            path=path,
-            storage_options=storage_options,
-            cache_policy=cache_policy,
-            max_cache_bytes=max_cache_bytes,
-            cache_dir=cache_dir,
-            hdf5_index=hdf5_index,
-            _allow_array_root=True,
-            _filesystem=_filesystem,
-            _filesystem_resolver=_filesystem_resolver,
-            _batch_validator=_batch_validator,
-        )
+        conversion = None
+        if parquet:
+            conversion = {
+                "parquet_options": parquet_options,
+                "columns": columns,
+                "max_rows": max_rows,
+                "string_max_length": string_max_length,
+                "null_storage": null_storage,
+                "auto_null_sentinels": auto_null_sentinels,
+                "separate_nested_cols": separate_nested_cols,
+                "list_serializer": list_serializer,
+                "blosc2_batch_size": blosc2_batch_size,
+                "blosc2_items_per_block": blosc2_items_per_block,
+                "batch_size": batch_size,
+                "cparams": cparams,
+                "dparams": dparams,
+                "validate": validate,
+            }
+
+        if shared_cache:
+            if cache_dir is None or (
+                cache_policy is not CACHE_POLICY_DEFAULT and cache_policy is not blosc2.CachePolicy.DISK
+            ):
+                raise ValueError("shared_cache=True requires a disk cache")
+            store = RemoteStore.with_sparse_cache(
+                urlpath,
+                cache_dir,
+                dataset=dataset,
+                path=path,
+                storage_options=storage_options,
+                max_cache_bytes=max_cache_bytes,
+                _filesystem=_filesystem,
+                _filesystem_resolver=_filesystem_resolver,
+                _batch_validator=_batch_validator,
+                _source_format="parquet" if parquet else None,
+                _parquet_conversion=conversion,
+            )
+        else:
+            store = RemoteStore(
+                urlpath,
+                dataset=dataset,
+                path=path,
+                storage_options=storage_options,
+                cache_policy=cache_policy,
+                max_cache_bytes=max_cache_bytes,
+                cache_dir=cache_dir,
+                hdf5_index=hdf5_index,
+                _allow_array_root=True,
+                _filesystem=_filesystem,
+                _filesystem_resolver=_filesystem_resolver,
+                _batch_validator=_batch_validator,
+                _source_format="parquet" if parquet else None,
+                _parquet_conversion=conversion,
+                _allow_local_source=parquet,
+            )
         try:
             _, full = store._resolve("")
             kind, diagnostic = store._owner.nodes[full]
@@ -211,12 +226,16 @@ def __init__(self, *args, **kwargs):
 
     @classmethod
     def open_reference(cls, path, *, storage_options=None, parquet_options=None):
-        """Reopen a saved Parquet table with replacement runtime options."""
-        from blosc2.remote_parquet import RemoteParquetCTable
+        """Reopen a saved Parquet RemoteStore archive."""
+        if parquet_options is not None:
+            raise TypeError("Parquet reader options are frozen in the saved reference")
+        from blosc2.remote_store import RemoteStore
 
-        return RemoteParquetCTable.open_reference(
-            path, storage_options=storage_options, parquet_options=parquet_options
-        )
+        table = RemoteStore._open_artifact(path, storage_options=storage_options)
+        if not isinstance(table, cls):
+            table.close()
+            raise ValueError("Reference does not contain a remote CTable")
+        return table
 
     @classmethod
     def with_sparse_cache(
@@ -239,6 +258,7 @@ def with_sparse_cache(
         _source_validator=None,
         _manifest_validator=None,
         _max_nodes=None,
+        source_format=None,
     ):
         """Attach a remote CTable to a sparse disk cache shared across processes.
 
@@ -247,37 +267,6 @@ def with_sparse_cache(
         ``max_cache_bytes=None`` for unlimited retention. For ordinary shared
         caching, prefer ``blosc2.open(url, cache_dir=..., shared_cache=True)``.
         """
-        if isinstance(urlpath, (str, os.PathLike)) and os.fspath(urlpath).split("?", 1)[0].lower().endswith(
-            ".parquet"
-        ):
-            if any(
-                value is not None
-                for value in (
-                    dataset,
-                    path,
-                    manifest,
-                    carrier,
-                    _filesystem,
-                    _filesystem_resolver,
-                    _batch_validator,
-                    _source_validator,
-                    _manifest_validator,
-                    _max_nodes,
-                )
-            ):
-                raise ValueError("Parquet sparse caching accepts a single file and cache directory")
-            from blosc2.remote_parquet import RemoteParquetCTable
-
-            return RemoteParquetCTable(
-                os.fspath(urlpath),
-                cache_dir=runtime_cache_path,
-                shared_cache=True,
-                max_cache_bytes=max_cache_bytes,
-                storage_options=storage_options,
-                max_concurrency=max_concurrency,
-                metadata_buffer_bytes=metadata_buffer_bytes,
-                row_buffer_bytes=row_buffer_bytes,
-            )
         settings = {
             name: _positive_integer(name, value)
             for name, value in {
@@ -304,6 +293,7 @@ def with_sparse_cache(
             _source_validator=_source_validator,
             _manifest_validator=_manifest_validator,
             _max_nodes=_max_nodes,
+            _source_format=source_format,
         )
         try:
             _, full = store._resolve("")
@@ -319,7 +309,14 @@ def with_sparse_cache(
     @classmethod
     def _from_owner(cls, owner, full_path, **settings):
         settings = {name: _positive_integer(name, value) for name, value in settings.items()}
-        storage = RemoteTableStorage(owner, full_path, **settings)
+        if owner.format == "parquet":
+            from blosc2.remote_parquet import ParquetTableStorage
+
+            storage = ParquetTableStorage(
+                owner, owner.parquet_schema, owner.parquet_physical, owner.parquet_length, **settings
+            )
+        else:
+            storage = RemoteTableStorage(owner, full_path, **settings)
         try:
             return cls._open_from_storage(storage)
         except BaseException:
@@ -408,10 +405,12 @@ def attrs(self):
     @property
     def source(self):
         storage = self._remote_storage()
+        from blosc2.remote_store import public_source_url
+
         return {
             "kind": storage._owner.format,
             "version": 1,
-            "urlpath": storage._owner.urlpath,
+            "urlpath": public_source_url(storage._owner.urlpath),
             "dataset": storage._root_key,
             "assume_immutable": True,
         }