Release 0.1.34: %gpkg GeoPackage export, pushable with %dds (ref #309764)

.gitignore

@@ -14,3 +14,7 @@ debugger/bathing_water/
 .env
 .env.*
 !debugger/.env.example
+
+# Windows download metadata (WSL copies it next to downloaded files)
+*:Zone.Identifier
+debugger/output.gpkg

README.md

@@ -36,7 +36,7 @@ pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@staging"
 ```
 
 # staging's latest release (early access) — pin to the tag the "staging" badge above shows
-pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.33-staging"
+pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.34-staging"
 ```
 
 ## Usage
@@ -329,6 +329,72 @@ builds its session (the kernel environment's value only applies if that `.env` d
 After each push, `%dds` tells an integrated JupyterLab panel which files changed (comm target
 `eea_datalakehouse.dds_pushed`, one message per file).
 
+## GeoPackage export (`%gpkg`)
+
+`eea_datalakehouse.geopackage` writes the tables and views under a Dremio catalog path to one
+OGC GeoPackage (`.gpkg`) file: each spatial one as a feature layer, each other one as an
+attributes table. Written with `sqlite3`, `shapely` and `pyproj` — no GDAL.
+
+```python
+import eea_datalakehouse.notebook   # registers %gpkg with the other magics
+
+%catalog use("catalog.water_management_resources.bathing_water.bwd")
+%gpkg export()                                   # every table and view -> ./bwd.gpkg
+%gpkg export(tables=["stations", "areas"], limit=1000, srid=3035)
+%gpkg export(path="../wfd", filename="wfd_2025.gpkg", output_dir="exports", overwrite=False)
+%gpkg export(workers=8)                          # more tables streamed at once
+%gpkg help
+```
+
+- `path` defaults to the `%catalog use(...)` context and is resolved like any `%catalog` path;
+  it can also name a single table or view.
+- `tables` picks some, by name below `path` (`"stations"`, `"sub.areas"`); default all, at any
+  depth. Two with the same name become `sub_a_stations` / `sub_b_stations`.
+- `limit` caps the rows per table; `srid` (`3035`, `"EPSG:3035"`) reprojects every layer to that
+  CRS; `filename` / `output_dir` default to `<dataset>.gpkg` in the notebook's folder.
+- **Spatial or tabular** is decided from the data: a column whose values decode as WKB/EWKB,
+  WKT/EWKT, hex WKB or GeoJSON is the geometry (columns named `geom`, `geometry`, `wkt`,
+  `shape`, ... are tried first); failing that, latitude/longitude columns make points.
+- **CRS** per row: the SRID inside the geometry (EWKB/EWKT), else a CRS column (`srid`,
+  `sr_id`, `epsg`, `crs`, ...: `3035`, `"EPSG:3035"`, an EPSG URN or a CRS name), else EPSG:4326
+  for latitude/longitude. A layer takes `srid`, or its first rows' CRS; rows in another CRS are
+  reprojected to it. A layer with no CRS at all is written with the undefined SRS (-1), and the
+  report says so.
+- **CRS definitions** come from `geopackage/epsg_definitions.json` (`{code: {name, wkt}}`, shipped
+  with the package): both ends of every reprojection and the `gpkg_spatial_ref_sys` rows are built
+  from it, so the coordinates and the SRS that describes them agree. A code the file lacks falls
+  back to pyproj's EPSG database; codes and CRS names in the file (`900913`, `"ETRS89 / UTM zone
+  32N"`) are accepted for `srid` and in CRS columns.
+
+**Streaming.** Nothing is read whole. `workers` tables (default 4) are read at once, each over
+its own Arrow Flight gRPC channel (`Catalog.open_flight_channel`, one login shared), a batch at
+a time with no overall deadline on the stream. Each worker decodes, reprojects and encodes its
+batches; one thread owns the SQLite file and only inserts, fed through a bounded queue — a few
+batches per worker — so memory stays flat whatever the table size. The file is built under a
+temporary name with no journal or fsync in one transaction, and renamed into place only when
+every table succeeded; a failure in any table stops the others and leaves no partial file.
+Against a remote Dremio the parallel channels hide the network wait (about 2× faster with 4
+workers in a simulated 50 ms/batch test); without network wait, Python's per-row work caps the
+export at roughly 50–60k rows/s whatever the number of workers.
+
+**Progress.** `%gpkg export` prints a line as each table or view is written (`✓ 3/4 areas ←
+catalog...areas: features, POINT, SRS 3035, 1,500 rows, 0.1s`), or as one fails (`✗` with its
+name and the error — the export then stops and no file is written), and a total at the end.
+
+**Then push it to DDS.** The report `%gpkg export` returns is also the written file's path
+(`os.PathLike`), so `%dds push` takes it like any file — uploaded as
+`application/geopackage+sqlite3` to the same dataset's folder:
+
+```python
+report = %gpkg export(srid=3035)
+%dds push(DELIVERABLES, report)        # -> catalog/.../bwd/deliverables/bwd.gpkg
+```
+
+In Python: `GpkgController(DremioSource(catalog)).export(catalog_path, "out.gpkg", tables=...,
+limit=..., srid=..., workers=..., progress=print)` returns a `GpkgReport` listing each layer, its geometry
+source, type, SRS and row count. `Catalog.stream_rows(sql, idempotency_key=...)` streams any
+SELECT as Arrow batches the same way.
+
 ## Notebook facade (`%catalog` / `%ingest`)
 
 For interactive use in JupyterLab, `eea_datalakehouse.notebook` registers two line magics
@@ -447,6 +513,10 @@ context itself — raises `CatalogSessionError` if none is set yet:
 | `dds_documents/controller.py` | `DdsController` / `DdsFolder` — push files to a catalog node's DDS document folders (metadata, deliverables, ...) |
 | `dds_documents/paths.py` | catalog (`a.b.c`) ⇄ storage (`a/b/c`) path conversion |
 | `dds_documents/session.py` | `DdsSession` — what `%dds` dispatches onto |
+| `geopackage/controller.py` | `GpkgController` — Dremio tables/views to a GeoPackage, streamed over parallel Flight channels |
+| `geopackage/spatial.py` | spatial-or-tabular detection, geometry decoding, CRS lookup and reprojection |
+| `geopackage/writer.py` | `GeoPackageWriter` — the `.gpkg` file itself, with `sqlite3`, bulk-loaded |
+| `geopackage/session.py` | `GpkgSession` — what `%gpkg` dispatches onto |
 | `sdi/controller.py` | `SdiController` — `get_xml()`, `resolve_series()` |
 | `sdi/catalogue.py` | `SdiCatalogue` — read-only GeoNetwork REST client |
 | `sdi/session.py` | `SdiSession` — what `%sdi` dispatches onto |
@@ -535,7 +605,7 @@ To pin to one specific release instead, use the exact tag the live badges under
 %pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.30"
 
 # staging's latest release (early access)
-%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.33-staging"
+%pip install "git+https://github.com/eeadata/EEALakeHouse.python.git@v0.1.34-staging"
 ```
 
 Use the `%pip` magic rather than `!pip` — it installs into the kernel the

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "EEADataLakehouse"
-version = "0.1.33"
+version = "0.1.34"
 description = "Data preparation and validation utilities for the EEA data lakehouse pipeline."
 readme = "README.md"
 requires-python = ">=3.11"
@@ -50,6 +50,9 @@ Repository = "https://github.com/eeadata/EEALakeHouse.python"
 [tool.setuptools.packages.find]
 where = ["src"]
 
+[tool.setuptools.package-data]
+"eea_datalakehouse.geopackage" = ["epsg_definitions.json"]  # CRS definitions for reprojection
+
 [tool.ruff]
 target-version = "py311"
 line-length = 100
@@ -64,7 +67,7 @@ strict = true
 warn_return_any = false
 
 [[tool.mypy.overrides]]
-module = ["respx.*", "pyarrow.*"]  # neither ships type stubs
+module = ["respx.*", "pyarrow.*", "shapely.*"]  # none ships type stubs
 ignore_missing_imports = true
 
 [tool.pytest.ini_options]

src/eea_datalakehouse/catalog/client.py

@@ -32,8 +32,11 @@
 import signal
 import threading
 import weakref
+from collections.abc import Iterator
 from typing import Any, Literal
 
+import pyarrow as pa
+
 from . import foldertree, operations, retry_state
 from .foldertree import FolderTransferReport, FolderTree, ProgressCallback
 from .operations import CatalogEntry, TableInfo, VersionResult, ViewDefinition
@@ -307,6 +310,20 @@ def gettableitemsfrom(self, table_path: str, *, idempotency_key: str) -> TableIn
             self._executor, table_path, idempotency_key=idempotency_key
         )
 
+    def stream_rows(self, sql: str, *, idempotency_key: str) -> Iterator[pa.RecordBatch]:
+        """Run a SELECT and yield its rows as Arrow `RecordBatch` chunks as they
+        arrive, never the whole result at once (see `FlightSqlExecutor.
+        stream_batches`). Always Arrow Flight: like `datacopy`/`datamove`,
+        this moves data, and REST's poll-then-paginate shape doesn't scale to
+        a bulk read."""
+        return self._flight_executor.stream_batches(sql, idempotency_key=idempotency_key)
+
+    def open_flight_channel(self) -> FlightSqlExecutor:
+        """A Flight executor on its own gRPC channel, logged in as this
+        catalog — for reading several large results at once, one channel
+        each (see `FlightSqlExecutor.fork`). The caller closes it."""
+        return self._flight_executor.fork()
+
     def getentriesfrom(self, schema_path: str, *, idempotency_key: str) -> list[CatalogEntry]:
         return operations.getentriesfrom(
             self._executor, schema_path, idempotency_key=idempotency_key

src/eea_datalakehouse/catalog/sql.py

@@ -21,12 +21,16 @@
 from __future__ import annotations
 
 import os
+import queue
+import threading
 import time
+from collections.abc import Iterator
 from dataclasses import dataclass
 from typing import Any, Protocol
 from urllib.parse import urlsplit
 
 import httpx
+import pyarrow as pa
 
 from .errors import CatalogOperationError, EngineStartingError
 
@@ -285,6 +289,9 @@ def __init__(
         self._owns_client = flight_client is None
         self._client = flight_client
         self._auth_header: tuple[bytes, bytes] | None = None
+        # One executor may serve several threads (parallel endpoint reads,
+        # or a caller's own workers): connecting and the login happen once.
+        self._lock = threading.RLock()
 
     def __repr__(self) -> str:
         return f"FlightSqlExecutor(location={self._location!r}, username={self._username!r})"
@@ -295,17 +302,35 @@ def close(self) -> None:
             self._client = None
         self._auth_header = None
 
+    def fork(self) -> FlightSqlExecutor:
+        """A new executor with its **own gRPC channel**, reusing this one's
+        login (the bearer token from the handshake is valid on any channel).
+
+        One channel multiplexes every call over a single HTTP/2 connection;
+        several large result streams read at once go faster each on their own.
+        The caller closes the fork.
+        """
+        twin = FlightSqlExecutor(self._location, self._username, self._token, timeout=self._timeout)
+        with self._lock:
+            twin._auth_header = self._auth_header
+        return twin
+
     def _get_client(self):
         import pyarrow.flight as flight
 
-        if self._client is None:
-            self._client = flight.FlightClient(self._location)
-        return self._client
+        with self._lock:
+            if self._client is None:
+                self._client = flight.FlightClient(self._location)
+            return self._client
 
     def _authenticate(self, idempotency_key: str | None):
         """The basic-auth handshake's bearer-token header, done once and
         cached — reused by every call for this executor's whole lifetime,
         same reasoning as `_get_client()` reusing one `FlightClient`."""
+        with self._lock:
+            return self._authenticate_locked(idempotency_key)
+
+    def _authenticate_locked(self, idempotency_key: str | None):
         import pyarrow.flight as flight
 
         if self._auth_header is None:
@@ -388,3 +413,114 @@ def fetch_all(
                     idempotency_key=idempotency_key,
                 ) from exc
         return rows
+
+    def stream_batches(
+        self, sql: str, *, idempotency_key: str | None = None, max_queued: int = 4
+    ) -> Iterator[pa.RecordBatch]:
+        """Run a SELECT and yield each Arrow `RecordBatch` as it arrives.
+
+        Nothing is gathered into memory: a whole table can be read through
+        this a batch at a time. Planning the query (``GetFlightInfo``) keeps
+        this executor's timeout — that is where a cold engine makes Dremio
+        wait — but reading the result has no overall deadline, so a long
+        stream is never cut off part-way. When Dremio splits the result into
+        several endpoints, they are read at once, each in its own thread, at
+        most ``max_queued`` batches ahead per endpoint; the batches then come
+        in arrival order. Stopping early (``close()`` on this generator, or
+        breaking out of the loop) cancels the reads still running.
+        """
+        import pyarrow.flight as flight
+
+        client, options, info = self._get_flight_info(sql, idempotency_key)
+        stream_options = flight.FlightCallOptions(headers=options.headers)
+        tickets = [endpoint.ticket for endpoint in info.endpoints]
+        if len(tickets) <= 1:
+            for ticket in tickets:
+                yield from self._read_endpoint(client, ticket, stream_options, sql, idempotency_key)
+            return
+        yield from self._read_endpoints(
+            client, tickets, stream_options, sql, idempotency_key, max_queued
+        )
+
+    def _read_endpoint(
+        self,
+        client: Any,
+        ticket: Any,
+        options: Any,
+        sql: str,
+        idempotency_key: str | None,
+        stop: threading.Event | None = None,
+    ) -> Iterator[pa.RecordBatch]:
+        import pyarrow.flight as flight
+
+        reader = None
+        try:
+            reader = client.do_get(ticket, options)
+            for chunk in reader:
+                if stop is not None and stop.is_set():
+                    return
+                if chunk.data is not None and chunk.data.num_rows:
+                    yield chunk.data
+            reader = None  # read to the end: nothing to cancel
+        except flight.FlightTimedOutError as exc:
+            raise EngineStartingError(
+                f"Flight SQL stream timed out (likely a starting engine): {sql!r}",
+                idempotency_key=idempotency_key,
+            ) from exc
+        except flight.FlightServerError as exc:
+            raise CatalogOperationError(f"Flight SQL error while reading results: {exc}") from exc
+        finally:
+            if reader is not None:
+                reader.cancel()
+
+    def _read_endpoints(
+        self,
+        client: Any,
+        tickets: list[Any],
+        options: Any,
+        sql: str,
+        idempotency_key: str | None,
+        max_queued: int,
+    ) -> Iterator[pa.RecordBatch]:
+        """Every endpoint at once, one thread each, into one bounded queue."""
+        done = object()
+        stop = threading.Event()
+        batches: queue.Queue[Any] = queue.Queue(maxsize=max_queued * len(tickets))
+
+        def put(item: Any) -> bool:
+            while not stop.is_set():
+                try:
+                    batches.put(item, timeout=0.1)
+                    return True
+                except queue.Full:
+                    continue
+            return False
+
+        def read(ticket: Any) -> None:
+            try:
+                stream = self._read_endpoint(client, ticket, options, sql, idempotency_key, stop)
+                for batch in stream:
+                    if not put(batch):
+                        return
+            except BaseException as exc:  # noqa: BLE001 — handed to the consumer below
+                put(exc)
+            finally:
+                put(done)
+
+        threads = [threading.Thread(target=read, args=(t,), daemon=True) for t in tickets]
+        for thread in threads:
+            thread.start()
+        try:
+            running = len(threads)
+            while running:
+                item = batches.get()
+                if item is done:
+                    running -= 1
+                elif isinstance(item, BaseException):
+                    raise item
+                else:
+                    yield item
+        finally:
+            stop.set()
+            for thread in threads:
+                thread.join(timeout=5)

src/eea_datalakehouse/dds_documents/controller.py

@@ -37,6 +37,8 @@
 from .paths import path_segments, to_storage_path
 
 DEFAULT_CONTENT_TYPE = "application/octet-stream"
+# Not in every platform's MIME table; %gpkg export's files are pushed here.
+mimetypes.add_type("application/geopackage+sqlite3", ".gpkg")
 _INVALID_NAME_CHARS = set('<>:"|?*')
 _GLOB_CHARS = set("*?[")