From 4ceea9d267daf858c9ff37b207397c5223a86fcd Mon Sep 17 00:00:00 2001 From: Matthias Mohr Date: Fri, 11 Sep 2026 16:04:10 +0200 Subject: [PATCH 1/3] Add DuckDB-based converter --- CHANGELOG.md | 14 ++ pixi.lock | 75 ++++++++++- pyproject.toml | 1 + tests/test_convert_duckdb.py | 178 +++++++++++++++++++++++++ tests/test_encoding_geoparquet.py | 68 ++++++++++ tests/test_hilbert.py | 89 +++++++++++++ vecorel_cli/conversion/base.py | 185 +++++++++++++++++++++++++- vecorel_cli/conversion/duckdb.py | 190 +++++++++++++++++++++++++++ vecorel_cli/encoding/base.py | 10 ++ vecorel_cli/encoding/geoparquet.py | 204 ++++++++++++++++++++++++++++- vecorel_cli/parquet/types.py | 34 +++++ vecorel_cli/vecorel/hilbert.py | 127 ++++++++++++++++++ 12 files changed, 1166 insertions(+), 9 deletions(-) create mode 100644 tests/test_convert_duckdb.py create mode 100644 tests/test_hilbert.py create mode 100644 vecorel_cli/conversion/duckdb.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c2fb424..bf2b168 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ## [Unreleased] +- Add an experimental `DuckDBBaseConverter` to convert large Parquet-based datasets + without loading them into memory. Its output matches the default converter + (geometry handling, Hilbert order, data types, metadata, file packaging). + `duckdb` is a new dependency. +- Converters record the collection id in the collection metadata and no longer add a + constant `collection` column. Previously files without constant columns were written + without any collection id. +- Converters drop rows that can never validate (missing required values, empty or + missing geometries), bounded by the new `max_dropped_share` (default 1%). +- Converters fail when both `sources` and `variants` are declared. +- Converters warn when no column is mapped to `id` and when the id column is not unique. +- Converters load all schemas upfront with retries, so a temporary network issue + no longer kills a long conversion at the very end. + ## [v0.2.17] - 2026-09-03 - Read GeoJSON as UTF-8, which the format mandates, instead of following the platform locale. diff --git a/pixi.lock b/pixi.lock index 2defc44..27c21a5 100644 --- a/pixi.lock +++ b/pixi.lock @@ -275,6 +275,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/linux-64/zstd-1.5.7-hb78ec9c_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/4a/57/3b7d4dd193ade4641c865bc2b93aeeb71162e81fc348b8dad020215601ed/build-1.4.2-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/4f/98/6549769f158126fa64fd6c1ac2eb59a18282146c939867a3eb31b7c1db07/duckdb-1.5.5-cp312-cp312-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -555,6 +556,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/osx-64/zstd-1.5.7-h3eecb57_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/4a/57/3b7d4dd193ade4641c865bc2b93aeeb71162e81fc348b8dad020215601ed/build-1.4.2-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/79/15/5ceb58ffb5bb8a62b3fd7abb39c41467cdf94850ece02e6d88664dfc75ce/duckdb-1.5.5-cp312-cp312-macosx_10_13_x86_64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -835,6 +837,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/osx-arm64/zstd-1.5.7-hbf9d68e_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/4a/57/3b7d4dd193ade4641c865bc2b93aeeb71162e81fc348b8dad020215601ed/build-1.4.2-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/bf/5c/bf02da0b354fe83cca4f95a4fbf762181af466f7d551ab2a093f7698882a/duckdb-1.5.5-cp312-cp312-macosx_11_0_arm64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -1112,6 +1115,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/win-64/zstd-1.5.7-h534d264_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/4a/57/3b7d4dd193ade4641c865bc2b93aeeb71162e81fc348b8dad020215601ed/build-1.4.2-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/af/b7/5753b41d3124838f868f9f523362812d9fc45409e9e4dd70dcbb0a25826e/duckdb-1.5.5-cp312-cp312-win_amd64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -1381,6 +1385,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/linux-64/zlib-ng-2.3.3-hceb46e0_1.conda - conda: https://conda.anaconda.org/conda-forge/linux-64/zstd-1.5.7-hb78ec9c_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/4f/98/6549769f158126fa64fd6c1ac2eb59a18282146c939867a3eb31b7c1db07/duckdb-1.5.5-cp312-cp312-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -1635,6 +1640,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/osx-64/zlib-ng-2.3.3-h8bce59a_1.conda - conda: https://conda.anaconda.org/conda-forge/osx-64/zstd-1.5.7-h3eecb57_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/79/15/5ceb58ffb5bb8a62b3fd7abb39c41467cdf94850ece02e6d88664dfc75ce/duckdb-1.5.5-cp312-cp312-macosx_10_13_x86_64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -1889,6 +1895,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/osx-arm64/zlib-ng-2.3.3-hed4e4f5_1.conda - conda: https://conda.anaconda.org/conda-forge/osx-arm64/zstd-1.5.7-hbf9d68e_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/bf/5c/bf02da0b354fe83cca4f95a4fbf762181af466f7d551ab2a093f7698882a/duckdb-1.5.5-cp312-cp312-macosx_11_0_arm64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -2141,6 +2148,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/win-64/zlib-ng-2.3.3-h0261ad2_1.conda - conda: https://conda.anaconda.org/conda-forge/win-64/zstd-1.5.7-h534d264_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/af/b7/5753b41d3124838f868f9f523362812d9fc45409e9e4dd70dcbb0a25826e/duckdb-1.5.5-cp312-cp312-win_amd64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -2376,6 +2384,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/linux-64/zlib-ng-2.3.3-hceb46e0_1.conda - conda: https://conda.anaconda.org/conda-forge/linux-64/zstd-1.5.7-hb78ec9c_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/4f/98/6549769f158126fa64fd6c1ac2eb59a18282146c939867a3eb31b7c1db07/duckdb-1.5.5-cp312-cp312-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -2597,6 +2606,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/osx-64/zlib-ng-2.3.3-h8bce59a_1.conda - conda: https://conda.anaconda.org/conda-forge/osx-64/zstd-1.5.7-h3eecb57_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/79/15/5ceb58ffb5bb8a62b3fd7abb39c41467cdf94850ece02e6d88664dfc75ce/duckdb-1.5.5-cp312-cp312-macosx_10_13_x86_64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -2818,6 +2828,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/osx-arm64/zlib-ng-2.3.3-hed4e4f5_1.conda - conda: https://conda.anaconda.org/conda-forge/osx-arm64/zstd-1.5.7-hbf9d68e_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/bf/5c/bf02da0b354fe83cca4f95a4fbf762181af466f7d551ab2a093f7698882a/duckdb-1.5.5-cp312-cp312-macosx_11_0_arm64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -3037,6 +3048,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/win-64/zlib-ng-2.3.3-h0261ad2_1.conda - conda: https://conda.anaconda.org/conda-forge/win-64/zstd-1.5.7-h534d264_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/af/b7/5753b41d3124838f868f9f523362812d9fc45409e9e4dd70dcbb0a25826e/duckdb-1.5.5-cp312-cp312-win_amd64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -3297,6 +3309,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/linux-64/zstd-1.5.7-hb78ec9c_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/4a/57/3b7d4dd193ade4641c865bc2b93aeeb71162e81fc348b8dad020215601ed/build-1.4.2-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/4f/98/6549769f158126fa64fd6c1ac2eb59a18282146c939867a3eb31b7c1db07/duckdb-1.5.5-cp312-cp312-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -3544,6 +3557,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/osx-64/zstd-1.5.7-h3eecb57_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/4a/57/3b7d4dd193ade4641c865bc2b93aeeb71162e81fc348b8dad020215601ed/build-1.4.2-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/79/15/5ceb58ffb5bb8a62b3fd7abb39c41467cdf94850ece02e6d88664dfc75ce/duckdb-1.5.5-cp312-cp312-macosx_10_13_x86_64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -3791,6 +3805,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/osx-arm64/zstd-1.5.7-hbf9d68e_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/4a/57/3b7d4dd193ade4641c865bc2b93aeeb71162e81fc348b8dad020215601ed/build-1.4.2-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/bf/5c/bf02da0b354fe83cca4f95a4fbf762181af466f7d551ab2a093f7698882a/duckdb-1.5.5-cp312-cp312-macosx_11_0_arm64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -4035,6 +4050,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/win-64/zstd-1.5.7-h534d264_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/4a/57/3b7d4dd193ade4641c865bc2b93aeeb71162e81fc348b8dad020215601ed/build-1.4.2-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/af/b7/5753b41d3124838f868f9f523362812d9fc45409e9e4dd70dcbb0a25826e/duckdb-1.5.5-cp312-cp312-win_amd64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -4298,6 +4314,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/linux-64/zlib-ng-2.3.3-hceb46e0_1.conda - conda: https://conda.anaconda.org/conda-forge/linux-64/zstd-1.5.7-hb78ec9c_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/4f/98/6549769f158126fa64fd6c1ac2eb59a18282146c939867a3eb31b7c1db07/duckdb-1.5.5-cp312-cp312-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -4546,6 +4563,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/osx-64/zlib-ng-2.3.3-h8bce59a_1.conda - conda: https://conda.anaconda.org/conda-forge/osx-64/zstd-1.5.7-h3eecb57_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/79/15/5ceb58ffb5bb8a62b3fd7abb39c41467cdf94850ece02e6d88664dfc75ce/duckdb-1.5.5-cp312-cp312-macosx_10_13_x86_64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -4794,6 +4812,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/osx-arm64/zlib-ng-2.3.3-hed4e4f5_1.conda - conda: https://conda.anaconda.org/conda-forge/osx-arm64/zstd-1.5.7-hbf9d68e_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/bf/5c/bf02da0b354fe83cca4f95a4fbf762181af466f7d551ab2a093f7698882a/duckdb-1.5.5-cp312-cp312-macosx_11_0_arm64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -5040,6 +5059,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/win-64/zlib-ng-2.3.3-h0261ad2_1.conda - conda: https://conda.anaconda.org/conda-forge/win-64/zstd-1.5.7-h534d264_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/af/b7/5753b41d3124838f868f9f523362812d9fc45409e9e4dd70dcbb0a25826e/duckdb-1.5.5-cp312-cp312-win_amd64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -5281,6 +5301,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/linux-64/zlib-ng-2.3.3-hceb46e0_1.conda - conda: https://conda.anaconda.org/conda-forge/linux-64/zstd-1.5.7-hb78ec9c_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/4f/98/6549769f158126fa64fd6c1ac2eb59a18282146c939867a3eb31b7c1db07/duckdb-1.5.5-cp312-cp312-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -5508,6 +5529,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/osx-64/zlib-ng-2.3.3-h8bce59a_1.conda - conda: https://conda.anaconda.org/conda-forge/osx-64/zstd-1.5.7-h3eecb57_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/79/15/5ceb58ffb5bb8a62b3fd7abb39c41467cdf94850ece02e6d88664dfc75ce/duckdb-1.5.5-cp312-cp312-macosx_10_13_x86_64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -5735,6 +5757,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/osx-arm64/zlib-ng-2.3.3-hed4e4f5_1.conda - conda: https://conda.anaconda.org/conda-forge/osx-arm64/zstd-1.5.7-hbf9d68e_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/bf/5c/bf02da0b354fe83cca4f95a4fbf762181af466f7d551ab2a093f7698882a/duckdb-1.5.5-cp312-cp312-macosx_11_0_arm64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -5960,6 +5983,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/win-64/zlib-ng-2.3.3-h0261ad2_1.conda - conda: https://conda.anaconda.org/conda-forge/win-64/zstd-1.5.7-h534d264_6.conda - pypi: https://files.pythonhosted.org/packages/ed/c9/d7977eaacb9df673210491da99e6a247e93df98c715fc43fd136ce1d3d33/arrow-1.4.0-py3-none-any.whl + - pypi: https://files.pythonhosted.org/packages/af/b7/5753b41d3124838f868f9f523362812d9fc45409e9e4dd70dcbb0a25826e/duckdb-1.5.5-cp312-cp312-win_amd64.whl - pypi: https://files.pythonhosted.org/packages/cf/58/8acf1b3e91c58313ce5cb67df61001fc9dcd21be4fadb76c1a2d540e09ed/fqdn-1.5.1-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/7b/55/e5326141505c5d5e34c5e0935d2908a74e4561eca44108fbfb9c13d2911a/isoduration-20.11.0-py3-none-any.whl - pypi: https://files.pythonhosted.org/packages/07/44/1747a046ee10035252cf224af8217c9168910ffd54d4aeb6bacf8ee40672/json_stream-2.5.0-py3-none-any.whl @@ -8083,6 +8107,54 @@ packages: - pkg:pypi/distlib?source=hash-mapping size: 275642 timestamp: 1752823081585 +- pypi: https://files.pythonhosted.org/packages/4f/98/6549769f158126fa64fd6c1ac2eb59a18282146c939867a3eb31b7c1db07/duckdb-1.5.5-cp312-cp312-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl + name: duckdb + version: 1.5.5 + sha256: 7a6d2d11859d82a936ebdcb30ce3d8a1cbb3e990bff05c12abb9b54c44fa7bd1 + requires_dist: + - ipython ; extra == 'all' + - fsspec ; extra == 'all' + - numpy ; extra == 'all' + - pandas ; extra == 'all' + - pyarrow ; extra == 'all' + - adbc-driver-manager ; extra == 'all' + requires_python: '>=3.10.0' +- pypi: https://files.pythonhosted.org/packages/79/15/5ceb58ffb5bb8a62b3fd7abb39c41467cdf94850ece02e6d88664dfc75ce/duckdb-1.5.5-cp312-cp312-macosx_10_13_x86_64.whl + name: duckdb + version: 1.5.5 + sha256: 33db46679b071f108d57139493dee2d37e1f5efcf5c5c039c2969eed11a6c8a7 + requires_dist: + - ipython ; extra == 'all' + - fsspec ; extra == 'all' + - numpy ; extra == 'all' + - pandas ; extra == 'all' + - pyarrow ; extra == 'all' + - adbc-driver-manager ; extra == 'all' + requires_python: '>=3.10.0' +- pypi: https://files.pythonhosted.org/packages/af/b7/5753b41d3124838f868f9f523362812d9fc45409e9e4dd70dcbb0a25826e/duckdb-1.5.5-cp312-cp312-win_amd64.whl + name: duckdb + version: 1.5.5 + sha256: ddfbdb096c11d51ee22492397d342c90a82e62c5d09961477895934d0a25372f + requires_dist: + - ipython ; extra == 'all' + - fsspec ; extra == 'all' + - numpy ; extra == 'all' + - pandas ; extra == 'all' + - pyarrow ; extra == 'all' + - adbc-driver-manager ; extra == 'all' + requires_python: '>=3.10.0' +- pypi: https://files.pythonhosted.org/packages/bf/5c/bf02da0b354fe83cca4f95a4fbf762181af466f7d551ab2a093f7698882a/duckdb-1.5.5-cp312-cp312-macosx_11_0_arm64.whl + name: duckdb + version: 1.5.5 + sha256: f0b88535a5d86fdd63dba6ea02ab68c003dfb9e4892b11256ef24c4da208baae + requires_dist: + - ipython ; extra == 'all' + - fsspec ; extra == 'all' + - numpy ; extra == 'all' + - pandas ; extra == 'all' + - pyarrow ; extra == 'all' + - adbc-driver-manager ; extra == 'all' + requires_python: '>=3.10.0' - conda: https://conda.anaconda.org/conda-forge/noarch/exceptiongroup-1.3.1-pyhd8ed1ab_0.conda sha256: ee6cf346d017d954255bbcbdb424cddea4d14e4ed7e9813e429db1d795d01144 md5: 8e662bd460bda79b1ea39194e3c4c9ab @@ -16734,7 +16806,7 @@ packages: - pypi: ./ name: vecorel-cli version: 0.2.17 - sha256: 38d5383deffa9ad6134f076e78728a12ed406c638397e71e9747fbcde7979329 + sha256: 74c8acca9c6b69fbcc6563458e87bcaa400c817342ba528a9e236d494ddbae7e requires_dist: - pyyaml>=6.0,<7.0 - click>=8.1,<9.0 @@ -16752,6 +16824,7 @@ packages: - semantic-version>=2.10.0,<3.0 - json-stream>=2.3.0,<3.0 - loguru==0.7.3 + - duckdb>=1.4,<2.0 - s3fs>=2025.7.0 ; extra == 's3' - gcsfs>=2025.7.0 ; extra == 'gcs' - s3fs>=2025.7.0 ; extra == 'cloud' diff --git a/pyproject.toml b/pyproject.toml index a2964a1..d2c7429 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,7 @@ dependencies = [ "semantic-version>=2.10.0,<3.0", "json-stream>=2.3.0,<3.0", "loguru==0.7.3", + "duckdb>=1.4,<2.0", ] [project.optional-dependencies] diff --git a/tests/test_convert_duckdb.py b/tests/test_convert_duckdb.py new file mode 100644 index 0000000..fc6169b --- /dev/null +++ b/tests/test_convert_duckdb.py @@ -0,0 +1,178 @@ +import json + +import geopandas as gpd +import numpy as np +import pyarrow.parquet as pq +import shapely + +from vecorel_cli.conversion.base import BaseConverter +from vecorel_cli.conversion.duckdb import DuckDBBaseConverter +from vecorel_cli.validate import ValidateData +from vecorel_cli.vecorel.hilbert import hilbert_keys_for_table + +# One shared converter configuration, so the two codepaths cannot drift apart. +# column_filters and column_migrations must stay out of it: they are Python +# callables in the GeoDataFrame-based codepath and SQL fragments in the +# DuckDB-based one. +CONFIG = { + "id": "test", + "short_name": "Test", + "title": "Test dataset", + "description": "Test dataset", + "license": "CC0-1.0", + "columns": { + "geometry": "geometry", + "id": "id", + "name": "name", + }, + "missing_schemas": { + "properties": { + "name": {"type": "string"}, + } + }, +} + +Converter = type("Converter", (DuckDBBaseConverter,), dict(CONFIG)) +PandasConverter = type("PandasConverter", (BaseConverter,), dict(CONFIG)) + + +def _source_file(folder): + """A source with everything the geometry handling must fix: + a multi-part geometry, an invalid bowtie, a Z polygon, and a point.""" + square = shapely.Polygon([(0, 0), (0, 1), (1, 1), (1, 0)]) + multi = shapely.MultiPolygon( + [ + shapely.Polygon([(2, 0), (2, 1), (3, 1), (3, 0)]), + shapely.Polygon([(4, 0), (4, 1), (5, 1), (5, 0)]), + ] + ) + bowtie = shapely.Polygon([(6, 0), (7, 1), (7, 0), (6, 1)]) + with_z = shapely.Polygon([(8, 0, 5), (8, 1, 5), (9, 1, 5), (9, 0, 5)]) + point = shapely.Point(10, 0) + + gdf = gpd.GeoDataFrame( + { + "id": ["square", "multi", "bowtie", "with_z", "point"], + "name": ["a", "b", "c", "d", "e"], + "geometry": [square, multi, bowtie, with_z, point], + }, + crs="EPSG:4326", + ) + path = folder / "source.parquet" + gdf.to_parquet(path) + return str(path) + + +def test_duckdb_converter(tmp_folder): + src = _source_file(tmp_folder) + dest = tmp_folder / "converted.parquet" + + Converter().convert(dest, input_files={src: "source.parquet"}) + + result = gpd.read_parquet(dest) + # multi is split in two, bowtie is repaired into two valid polygons, + # the point is dropped and the Z dimension is removed + assert sorted(result["id"]) == ["bowtie", "bowtie", "multi", "multi", "square", "with_z"] + assert set(result.geometry.geom_type) == {"Polygon"} + assert result.geometry.is_valid.all() + assert not result.geometry.has_z.any() + + with pq.ParquetFile(dest) as pf: + schema = pf.schema_arrow + field = schema.field("geometry") + assert str(field.type) == "binary" + assert not field.nullable + field = schema.field("id") + assert str(field.type) == "string" + assert not field.nullable + assert str(schema.field("name").type) == "string" + assert "bbox" in schema.names + + # sorted against the CRS-derived Hilbert grid + with pq.ParquetFile(dest) as pf: + table = pf.read() + keys = hilbert_keys_for_table(table, "geometry", (-180.0, -90.0, 180.0, 90.0)) + assert bool(np.all(keys[1:] >= keys[:-1])) + + validation = ValidateData().validate(dest, num=100, schema_map={}) + assert validation.errors == [] + + +def test_duckdb_converter_original_geometries(tmp_folder): + src = _source_file(tmp_folder) + dest = tmp_folder / "converted.parquet" + + Converter().convert(dest, input_files={src: "source.parquet"}, original_geometries=True) + + result = gpd.read_parquet(dest) + assert len(result) == 5 + assert set(result.geometry.geom_type) == {"Polygon", "MultiPolygon", "Point"} + + +def test_codepath_parity(tmp_folder): + """The GeoDataFrame-based and the DuckDB-based codepaths must produce + comparable files from the same source and converter configuration: + same schema, same rows in the same order, same key metadata, same packaging. + """ + src = _source_file(tmp_folder) + kwargs = { + "input_files": {src: "source.parquet"}, + "compression": "zstd", + "geoparquet_version": "1.1.0", + } + pandas_dest = tmp_folder / "pandas.parquet" + duckdb_dest = tmp_folder / "duckdb.parquet" + PandasConverter().convert(pandas_dest, **kwargs) + Converter().convert(duckdb_dest, **kwargs) + + with pq.ParquetFile(pandas_dest) as pf: + pandas_schema = pf.schema_arrow + pandas_table = pf.read() + pandas_groups = pf.metadata.num_row_groups + pandas_compression = pf.metadata.row_group(0).column(0).compression + with pq.ParquetFile(duckdb_dest) as pf: + duckdb_schema = pf.schema_arrow + duckdb_table = pf.read() + duckdb_groups = pf.metadata.num_row_groups + duckdb_compression = pf.metadata.row_group(0).column(0).compression + + # Same columns with the same types and nullability (the column order is + # allowed to differ) + assert sorted(pandas_schema.names) == sorted(duckdb_schema.names) + for name in pandas_schema.names: + f1, f2 = pandas_schema.field(name), duckdb_schema.field(name) + assert f1.type == f2.type, f"{name}: {f1.type} != {f2.type}" + assert f1.nullable == f2.nullable, f"{name}: nullability differs" + + # Same rows in the same (Hilbert) order; the geometries must describe the + # same shapes, but the WKB may differ in vertex order (different GEOS builds) + assert pandas_table.num_rows == duckdb_table.num_rows + assert pandas_table["id"].to_pylist() == duckdb_table["id"].to_pylist() + assert pandas_table["name"].to_pylist() == duckdb_table["name"].to_pylist() + pandas_geoms = shapely.from_wkb(pandas_table["geometry"].to_pylist()) + duckdb_geoms = shapely.from_wkb(duckdb_table["geometry"].to_pylist()) + for g1, g2 in zip(pandas_geoms, duckdb_geoms): + assert shapely.equals(g1, g2), f"{shapely.to_wkt(g1)} != {shapely.to_wkt(g2)}" + + # Same collection metadata and the same key GeoParquet metadata + pandas_collection = json.loads(pandas_schema.metadata[b"collection"]) + duckdb_collection = json.loads(duckdb_schema.metadata[b"collection"]) + assert pandas_collection == duckdb_collection + + pandas_geo = json.loads(pandas_schema.metadata[b"geo"]) + duckdb_geo = json.loads(duckdb_schema.metadata[b"geo"]) + for key in ("version", "primary_column"): + assert pandas_geo[key] == duckdb_geo[key] + pandas_column = pandas_geo["columns"]["geometry"] + duckdb_column = duckdb_geo["columns"]["geometry"] + for key in ("encoding", "covering"): + assert pandas_column[key] == duckdb_column[key] + + # Same packaging + assert pandas_compression == duckdb_compression + assert pandas_groups == duckdb_groups + + # Both validate + for dest in (pandas_dest, duckdb_dest): + validation = ValidateData().validate(dest, num=100, schema_map={}) + assert validation.errors == [] diff --git a/tests/test_encoding_geoparquet.py b/tests/test_encoding_geoparquet.py index 387b004..b51bfac 100644 --- a/tests/test_encoding_geoparquet.py +++ b/tests/test_encoding_geoparquet.py @@ -1,5 +1,10 @@ +import json from pathlib import Path +import pyarrow as pa +import pyarrow.parquet as pq +import pyarrow.types as pat + from vecorel_cli.encoding.geoparquet import GeoParquet from vecorel_cli.vecorel.collection import Collection @@ -46,3 +51,66 @@ def test_get_collection_returns_existing(tmp_folder): result = geojson.get_collection() assert isinstance(result, Collection) assert result == test_collection + + +def test_postprocess(tmp_parquet_file): + # Degrade a compliant file to what external tools such as DuckDB or GDAL may produce: + # large_string/large_binary columns, a naive microsecond timestamp, nullable columns, + # no bbox column, GeoParquet 1.0.0 + src = pq.read_table("tests/data-files/inspire.parquet") + fields = [] + arrays = [] + for i, field in enumerate(src.schema): + if field.name == "bbox": + continue + if pat.is_string(field.type): + dtype = pa.large_string() + elif pat.is_binary(field.type): + dtype = pa.large_binary() + else: + dtype = field.type + fields.append(pa.field(field.name, dtype, nullable=True)) + arrays.append(src.column(i).cast(dtype)) + fields.append(pa.field("determination_datetime", pa.timestamp("us"))) + arrays.append(pa.array([1672531200000000] * len(src), type=pa.timestamp("us"))) + + metadata = dict(src.schema.metadata) + geo = json.loads(metadata[b"geo"]) + geo["version"] = "1.0.0" + geo["columns"]["geometry"].pop("covering", None) + metadata[b"geo"] = json.dumps(geo).encode("utf-8") + degraded = pa.table(arrays, schema=pa.schema(fields, metadata=metadata)) + pq.write_table(degraded, tmp_parquet_file, store_schema=True) + + gp = GeoParquet(tmp_parquet_file) + assert gp.postprocess(geoparquet_version="1.1.0", compression="zstd") is True + # a second run detects the compliant file and doesn't rewrite it + assert gp.postprocess(geoparquet_version="1.1.0", compression="zstd") is False + + result = pq.ParquetFile(tmp_parquet_file) + schema = result.schema_arrow + + field = schema.field("geometry") + assert field.type == pa.binary() + assert not field.nullable + field = schema.field("id") + assert field.type == pa.string() + assert not field.nullable + assert schema.field("inspire:id").type == pa.string() + assert schema.field("determination_datetime").type == pa.timestamp("ms", tz="UTC") + + bbox = schema.field("bbox") + assert pat.is_struct(bbox.type) + assert bbox.type.field("xmin").type == pa.float64() + + metadata = result.metadata.metadata + geo = json.loads(metadata[b"geo"]) + assert geo["version"] == "1.1.0" + assert geo["columns"]["geometry"]["covering"]["bbox"]["xmin"] == ["bbox", "xmin"] + assert b"collection" in metadata + + # data is intact + data = pq.read_table(tmp_parquet_file) + assert data.num_rows == src.num_rows + assert data["id"].to_pylist() == src["id"].to_pylist() + assert data["geometry"].to_pylist() == src["geometry"].to_pylist() diff --git a/tests/test_hilbert.py b/tests/test_hilbert.py new file mode 100644 index 0000000..6e00c0e --- /dev/null +++ b/tests/test_hilbert.py @@ -0,0 +1,89 @@ +import json + +import numpy as np +import pyarrow as pa +import pyarrow.parquet as pq +import shapely + +from vecorel_cli.vecorel.hilbert import ( + bounds_array_for_table, + crs_total_bounds, + ensure_hilbert_sorted, + hilbert_keys_for_table, + hilbert_reference_bounds, +) + +# A bare projection definition without an area of use, +# like the unnamed "MGI / Austria Lambert" in Austria's 2018 edition +CUSTOM_CRS = "+proj=laea +lat_0=52 +lon_0=10 +x_0=4321000 +y_0=3210000 +units=m" + + +def test_hilbert_reference_bounds_from_crs(): + assert hilbert_reference_bounds("EPSG:4326") == (-180.0, -90.0, 180.0, 90.0) + + +def test_hilbert_reference_bounds_fallback(): + fallback = [1.0, 2.0, 3.0, 4.0] + assert hilbert_reference_bounds(CUSTOM_CRS, fallback) == (1.0, 2.0, 3.0, 4.0) + assert hilbert_reference_bounds(CUSTOM_CRS) is None + + +def _make_table(points, with_bbox=True, metadata=None): + geometries = [shapely.Point(x, y).buffer(0.1) for x, y in points] + wkb = [shapely.to_wkb(g) for g in geometries] + bounds = shapely.bounds(np.array(geometries, dtype=object)) + arrays = { + "id": pa.array([str(i) for i in range(len(points))], type=pa.string()), + "geometry": pa.array(wkb, type=pa.binary()), + } + if with_bbox: + arrays["bbox"] = pa.StructArray.from_arrays( + [bounds[:, 0], bounds[:, 1], bounds[:, 2], bounds[:, 3]], + names=["xmin", "ymin", "xmax", "ymax"], + ) + table = pa.table(arrays) + if metadata: + table = table.replace_schema_metadata(metadata) + return table + + +def test_bounds_array_bbox_and_wkb_paths_agree(): + points = [(0, 0), (10, 10), (5, 5)] + with_bbox = _make_table(points, with_bbox=True) + without_bbox = _make_table(points, with_bbox=False) + + np.testing.assert_allclose( + bounds_array_for_table(with_bbox, "geometry"), + bounds_array_for_table(without_bbox, "geometry"), + ) + + +def test_ensure_hilbert_sorted(tmp_folder): + path = str(tmp_folder / "unsorted.parquet") + # scattered, deliberately not in Hilbert order + points = [(30, 40), (-120, -30), (5, 5), (100, 60), (-60, 10)] + metadata = { + b"geo": json.dumps({"primary_column": "geometry", "columns": {"geometry": {}}}).encode(), + b"collection": b'{"schemas": {}}', + } + table = _make_table(points, metadata=metadata) + pq.write_table(table, path) + + total_bounds = crs_total_bounds("EPSG:4326") + keys = hilbert_keys_for_table(table, "geometry", total_bounds) + assert not bool(np.all(keys[1:] >= keys[:-1])), "Test data must start unsorted" + + assert ensure_hilbert_sorted(path, "geometry", total_bounds, "zstd") is True + + result = pq.read_table(path) + sorted_keys = hilbert_keys_for_table(result, "geometry", total_bounds) + assert bool(np.all(sorted_keys[1:] >= sorted_keys[:-1])) + # data is intact, schema stays narrow, metadata is preserved + assert sorted(result["id"].to_pylist()) == sorted(table["id"].to_pylist()) + assert result.schema.field("geometry").type == pa.binary() + assert result.schema.field("id").type == pa.string() + assert result.schema.metadata[b"geo"] == metadata[b"geo"] + assert result.schema.metadata[b"collection"] == metadata[b"collection"] + + # a second run is a no-op + assert ensure_hilbert_sorted(path, "geometry", total_bounds, "zstd") is False diff --git a/vecorel_cli/conversion/base.py b/vecorel_cli/conversion/base.py index 74e5e78..9c3a954 100644 --- a/vecorel_cli/conversion/base.py +++ b/vecorel_cli/conversion/base.py @@ -62,6 +62,10 @@ class BaseConverter(LoggerMixin): index_as_id: bool = False + # Rows lacking a value for a schema-required property or a geometry are + # dropped up to this share of all rows, else they are kept with a warning + max_dropped_share: float = 0.01 + def __init__(self, *args, **kwargs): super().__init__() @@ -71,6 +75,170 @@ def __init__(self, *args, **kwargs): if not key.startswith("_") and isinstance(item, (list, dict, set)): setattr(self, key, copy(item)) + def _require_one_source_of_urls(self): + """Fail when both `sources` and `variants` are declared. + + The converter takes `sources` when it is set and ignores the variants + entirely, so `--variant 2011` silently converts whatever `sources` + points at. A converter that inherits variants it does not want says so + with `variants = {}`. + """ + if self.sources and self.variants: + raise ValueError( + f"{type(self).__name__} declares both sources and variants; sources wins " + "and every --variant would convert the same file. Drop sources, or set " + "variants = {} when the inherited ones do not apply." + ) + + def _check_id_mapping(self): + """Warn before converting when nothing will end up as `id`. + + Vecorel requires the identifier, and nothing downstream enforces it: + the converter drops columns no mapping names, so a converter without + one simply writes a file without `id` and validates. The subtler half + is `index_as_id = True`, which fills the column and the same drop step + removes it again, because `columns` never named it. A converter with + no natural key sets both `index_as_id` and `"id": "id"`. + """ + targets = set() + for value in list(self.columns.values()) + list(self.column_additions or {}): + targets.update(value if isinstance(value, (list, tuple)) else [value]) + if "id" not in targets: + hint = ( + ' — `index_as_id = True` is set, so add \'"id": "id"\' to columns' + if self.index_as_id + else " — map a unique source column to it, or set index_as_id = True" + ' and add \'"id": "id"\' to columns' + ) + self.warning(f"{type(self).__name__} maps no column to 'id'{hint}") + + def _check_unique_ids(self, gdf, columns): + """Warn when the column that becomes `id` does not identify a feature. + + Vecorel asks for one identifier per feature, but nothing measures it: + a converter can map a group id shared by thousands of features, and + every converter that reads several files and sets `index_as_id` + repeats the same index once per file. This runs before geometries are + exploded, so it judges what the converter assigned rather than the + split parts of one source feature. + + A missing id is a different failure, dropped under a bounded rule + afterwards, so it is not counted here: a column that identifies all + features except the few that carry none is a perfectly good identifier. + """ + sources = [ + k for k, v in columns.items() if "id" in (v if isinstance(v, (list, tuple)) else [v]) + ] + column = next( + (c for c in sources if c in gdf.columns), "id" if "id" in gdf.columns else None + ) + if column is None: + # The mapping exists (convert() checks that) but the data does not + # carry it, and the unlisted-column drop then writes a file + # without `id` that validates. + self.warning( + f"{type(self).__name__}: none of the columns mapped to 'id' " + f"({', '.join(sources) or 'none'}) is in this source; it has " + f"{', '.join(sorted(gdf.columns)[:12])}" + ) + return + ids = gdf[column].dropna() + if ids.is_unique: + return + counts = ids.value_counts() + duplicated = int(len(ids) - len(counts)) + worst = int(counts.iloc[0]) + self.warning( + f"{type(self).__name__}: '{column}' is not unique — {duplicated:,} of {len(ids):,} " + f"rows repeat an id (one appears {worst:,} times), so it cannot be `id`. Map a column " + "that identifies a feature, build one from the source's key columns, or use the row " + "index (index_as_id) only when the conversion reads a single file." + ) + + def _drop_incomplete_rows(self, gdf, columns): + """Drop rows that can never validate: null values in a property the + schemas require, and empty or missing geometries (which also cannot + be tiled or Hilbert-sorted). Bounded by ``max_dropped_share``: above + that share the rows are kept (the converter needs fixing, and quietly + dropping large parts of a dataset would hide that), with a warning + that the output will not validate. + + The required properties come from the merged schemas (the core + specification, the declared extensions and the custom schemas), minus + the collection-only properties and the geometry, which is checked + separately below. + """ + collection = self.create_collection(self.id.strip()) + schemas = collection.merge_schemas({}) + collection_only = set(collection.get_collection_only_properties()) + required = [ + r for r in schemas.get("required", []) if r != "geometry" and r not in collection_only + ] + + # This runs before columns are renamed, so look up the source column + for key in required: + for src, dst in columns.items(): + targets = dst if isinstance(dst, (list, tuple)) else [dst] + if key in targets and src in gdf.columns: + nulls = gdf[src].isna() + if nulls.any(): + share = nulls.mean() + if share > self.max_dropped_share: + self.warning( + f"{int(nulls.sum())} of {len(gdf)} rows ({share:.1%}) have no " + f"{key} ({src}); the output will not validate — fix the converter" + ) + else: + self.warning( + f"Dropping {int(nulls.sum())} rows without a value for {key} ({src})" + ) + gdf = gdf[~nulls] + + if gdf.active_geometry_name is not None: + geom = gdf.geometry + blank = geom.isna() | geom.is_empty + if blank.any(): + share = blank.mean() + if share > self.max_dropped_share: + self.warning( + f"{int(blank.sum())} of {len(gdf)} rows ({share:.1%}) have an empty or " + f"missing geometry; the output will not validate — fix the converter" + ) + else: + self.warning( + f"Dropping {int(blank.sum())} rows with an empty or missing geometry" + ) + gdf = gdf[~blank] + + return gdf + + def _prewarm_schemas(self): + """Fetch every schema this conversion will need before doing any real + work, with retries. The schema hosts fail intermittently; without this, + a transient blip after a long source download kills the conversion at + the very last step. load_file caches per process, so a successful + pre-warm makes the write network-free.""" + import time + + from ..vecorel.util import load_file + + uris = set(self.extensions) + uris.add(Schemas.get_core_uri()) + attempts = 8 + for uri in sorted(uris): + for attempt in range(attempts): + try: + load_file(uri) + break + except Exception as e: + if attempt == attempts - 1: + raise RuntimeError( + f"Cannot load schema {uri} after {attempts} attempts: {e}" + ) from e + self.warning(f"Schema fetch failed ({uri}), retrying: {str(e)[:100]}") + # ~4 min of tolerance: schema host outages have outlasted a 30 s budget + time.sleep(min(2**attempt * 2, 60)) + def migrate(self, gdf) -> GeoDataFrame: return gdf @@ -316,6 +484,10 @@ def convert( if self.bbox is not None and len(self.bbox) != 4: raise ValueError("If provided, the bounding box must consist of 4 numbers") + self._check_id_mapping() + self._require_one_source_of_urls() + self._prewarm_schemas() + # Create output folder if it doesn't exist directory = os.path.dirname(output_file) if directory: @@ -360,10 +532,6 @@ def convert( gdf[key] = value columns[key] = key - # Add collection ID - columns["collection"] = "collection" - gdf["collection"] = cid - # 4. Run column migrations if self.column_migrations: self.info("Applying column migrations") @@ -375,6 +543,9 @@ def convert( gdf = self.post_migrate(gdf) + self._check_unique_ids(gdf, columns) + gdf = self._drop_incomplete_rows(gdf, columns) + if hash_before != self._hash_df(gdf.head()): self.info("GeoDataFrame after migrations and filters:") self.info(gdf.head().to_string()) @@ -430,7 +601,11 @@ def convert( self.info("Creating GeoParquet file: " + str(output_file)) columns = list(actual_columns.values()) pq = GeoParquet(output_file) - pq.set_collection(self.create_collection(cid)) + collection = self.create_collection(cid) + # Record the collection id at the collection level rather than as a + # constant column, like the DuckDB-based codepath does + collection["collection"] = cid + pq.set_collection(collection) pq.write( gdf, diff --git a/vecorel_cli/conversion/duckdb.py b/vecorel_cli/conversion/duckdb.py new file mode 100644 index 0000000..75c2df1 --- /dev/null +++ b/vecorel_cli/conversion/duckdb.py @@ -0,0 +1,190 @@ +import json +import os +from pathlib import Path +from typing import Optional + +import duckdb +import pyarrow.parquet as pq + +from ..encoding.geojson import VecorelJSONEncoder +from ..encoding.geoparquet import GeoParquet +from ..vecorel.hilbert import ensure_hilbert_sorted, hilbert_reference_bounds +from .base import BaseConverter + + +# This converter is experimental, use with caution. +# Results may not be fully compliant yet. +# Use this primarily for datasets that are too large to be processed by the default converter. +class DuckDBBaseConverter(BaseConverter): + def convert( + self, + output_file, + cache=None, + input_files=None, + variant=None, + compression=None, + compression_level: Optional[int] = None, + geoparquet_version=None, + original_geometries=False, + **kwargs, + ) -> str: + geoparquet_version = geoparquet_version or "1.1.0" + # Same packaging as the GeoDataFrame-based codepath + compression = compression or "zstd" + if compression == "zstd" and compression_level is None: + compression_level = 15 + row_group_size = GeoParquet.row_group_size + + self.variant = variant + cid = self.id.strip() + if self.bbox is not None and len(self.bbox) != 4: + raise ValueError("If provided, the bounding box must consist of 4 numbers") + + self._check_id_mapping() + self._require_one_source_of_urls() + self._prewarm_schemas() + + # Create output folder if it doesn't exist + directory = os.path.dirname(output_file) + if directory: + os.makedirs(directory, exist_ok=True) + + if input_files is not None and isinstance(input_files, dict) and len(input_files) > 0: + self.warning("Using user provided input file(s) instead of the pre-defined file(s)") + urls = input_files + else: + urls = self.get_urls() + if urls is None: + raise ValueError("No input files provided") + + self.info("Getting file(s) if not cached yet") + if cache: + request_args = {} + if self.avoid_range_request: + request_args["block_size"] = 0 + urls = self.download_files(urls, cache, **request_args) + elif self.avoid_range_request: + self.warning( + "avoid_range_request is set, but cache is not used, so this setting has no effect" + ) + + selections = [] + for k, v in self.columns.items(): + if k in self.column_migrations: + selections.append(f'{self.column_migrations.get(k)} as "{v}"') + else: + selections.append(f'"{k}" as "{v}"') + selection = ", ".join(selections) + + filters = [] + where = "" + if self.bbox is not None: + filters.append( + f"ST_Intersects(geometry, ST_MakeEnvelope({self.bbox[0]}, {self.bbox[1]}, {self.bbox[2]}, {self.bbox[3]}))" + ) + for k, v in self.column_filters.items(): + filters.append(v) + if len(filters) > 0: + where = f"WHERE {' AND '.join(filters)}" + + if isinstance(urls, str): + sources = f'"{urls}"' + else: + paths = [] + for url in urls: + if isinstance(url, tuple): + paths.append(f'"{url[0]}"') + else: + paths.append(f'"{url}"') + sources = "[" + ",".join(paths) + "]" + + collection = self.create_collection(cid) + collection.update(self.column_additions) + collection["collection"] = self.id + + if isinstance(output_file, Path): + output_file = str(output_file) + + collection_json = json.dumps(collection, cls=VecorelJSONEncoder).encode("utf-8") + + source_query = f""" + SELECT {selection} + FROM read_parquet({sources}, union_by_name=true) + {where} + """ + if original_geometries: + query = source_query + else: + # Mirror the geometry handling of the GeoDataFrame-based codepath: + # make geometries valid, split multi-part geometries, keep only + # valid polygons, and remove the Z dimension + query = f""" + WITH src AS ({source_query}), + valid AS ( + SELECT * REPLACE (ST_MakeValid(geometry) AS geometry) FROM src + ), + parts AS ( + SELECT * EXCLUDE (geometry), UNNEST(ST_Dump(geometry), recursive := true) + FROM valid + ) + SELECT * EXCLUDE (geom, path), ST_Force2D(geom) AS geometry + FROM parts + WHERE ST_GeometryType(geom) = 'POLYGON' AND ST_IsValid(geom) + """ + + con = duckdb.connect() + con.install_extension("spatial") + con.load_extension("spatial") + # No ORDER BY here: ST_Hilbert without bounds is meaningless (whole + # countries collapse into a handful of cells), and with bounds it uses + # a different reference grid than the rest of the pipeline. The + # canonical in-place Hilbert sort below runs after post-processing. + con.execute( + f""" + COPY ({query}) TO ? ( + FORMAT parquet, + ROW_GROUP_SIZE {row_group_size}, + compression ?, + KV_METADATA {{ + collection: ?, + }} + ) + """, + [output_file, compression, collection_json], + ) + + # Post-process the written Parquet file to a compliant GeoParquet file + # (canonical data types, nullability, bbox column, metadata) + gp = GeoParquet(output_file) + gp.set_collection(collection) + try: + gp.postprocess( + compression=compression, + compression_level=compression_level, + geoparquet_version=geoparquet_version, + ) + except Exception as e: + self.warning(f"GeoParquet post-processing failed: {e}") + + # Canonical spatial ordering against the CRS-derived Hilbert grid, + # the same grid the GeoDataFrame-based converter sorts against + with pq.ParquetFile(output_file) as pf: + meta = pf.schema_arrow.metadata or {} + if b"geo" in meta: + geo = json.loads(meta[b"geo"]) + primary = geo["primary_column"] + crs = geo["columns"][primary].get("crs") or "EPSG:4326" + bounds = hilbert_reference_bounds(crs, geo["columns"][primary].get("bbox")) + if bounds is None: + self.warning("CRS declares no area of use; skipping spatial ordering") + elif ensure_hilbert_sorted( + output_file, + primary, + bounds, + compression, + compression_level, + row_group_size=row_group_size, + ): + self.info("Sorted output into Hilbert order") + + return output_file diff --git a/vecorel_cli/encoding/base.py b/vecorel_cli/encoding/base.py index 05ab935..c6dbc85 100644 --- a/vecorel_cli/encoding/base.py +++ b/vecorel_cli/encoding/base.py @@ -83,6 +83,16 @@ def write( ) -> bool: raise NotImplementedError("Subclasses must implement this method") + def postprocess(self, schema_map: SchemaMapping = {}, **kwargs) -> bool: + """ + Rewrites an existing file (e.g. written by an external tool) into a + compliant Vecorel file, if needed. + + Checks the file first and only rewrites when something needs to change. + Returns True if the file was rewritten, False if it was compliant already. + """ + raise NotImplementedError("Not supported by encoding") + def read( self, num: Optional[int] = None, diff --git a/vecorel_cli/encoding/geoparquet.py b/vecorel_cli/encoding/geoparquet.py index 79370fc..6fe8a39 100644 --- a/vecorel_cli/encoding/geoparquet.py +++ b/vecorel_cli/encoding/geoparquet.py @@ -1,19 +1,28 @@ import json +import os from pathlib import Path +from tempfile import NamedTemporaryFile from typing import Optional, Union import pyarrow as pa import pyarrow.parquet as pq from geopandas import GeoDataFrame +from geopandas.array import from_wkb from geopandas.io.arrow import _arrow_to_geopandas -from pyarrow import NativeFile +from pyarrow import NativeFile, StructArray from pyarrow.fs import FSSpecHandler, PyFileSystem from yarl import URL from ..const import GEOPARQUET_DEFAULT_VERSION, GEOPARQUET_VERSIONS from ..encoding.geojson import VecorelJSONEncoder from ..parquet.geopandas import to_parquet -from ..parquet.types import get_geopandas_dtype, get_pyarrow_field, get_pyarrow_type_for_geopandas +from ..parquet.types import ( + get_geopandas_dtype, + get_pyarrow_field, + get_pyarrow_type, + get_pyarrow_type_for_geopandas, + normalize_pa_type, +) from ..validation.base import Validator from ..vecorel.typing import SchemaMapping from ..vecorel.util import get_fs, load_file @@ -111,7 +120,10 @@ def get_compression(self) -> Optional[str]: Get the compression method used in the file. Returns "mixed" if multiple compression methods are found. """ - metadata = self.get_parquet_metadata() + return self._detect_compression(self.get_parquet_metadata()) + + @staticmethod + def _detect_compression(metadata: pq.FileMetaData) -> Optional[str]: compressions = set() row_group = metadata.row_group(0) @@ -256,6 +268,192 @@ def write( return True + def postprocess( + self, + schema_map: SchemaMapping = {}, + compression: Optional[str] = None, + compression_level: Optional[int] = None, + geoparquet_version: Optional[str] = None, + **kwargs, # capture unknown arguments + ) -> bool: + """ + Rewrites an existing Parquet file (e.g. written by an external tool such as + DuckDB or GDAL) into a compliant Vecorel GeoParquet file: + + - Converts the column data types to the types defined in the Vecorel schemas, + otherwise normalizes them to the canonical Vecorel types + (e.g. large_string -> string, timestamps -> timestamp[ms, UTC]) + - Sets the nullability based on the required properties + - Adds a bbox covering column for GeoParquet > 1.0.0 + - Updates the GeoParquet metadata and embeds the collection metadata + + The collection metadata is taken from `set_collection` or read from the file. + The file is rewritten row group by row group, so it never needs to fit into memory. + If no compression is given, the compression of the existing file is used. + + Checks the file first and only rewrites when something needs to change, + so it is cheap to call on files that are already compliant. + Returns True if the file was rewritten, False if it was compliant already. + """ + if not isinstance(self.uri, Path): + raise ValueError("Post-processing is only supported for local files") + if geoparquet_version not in GEOPARQUET_VERSIONS: + geoparquet_version = GEOPARQUET_DEFAULT_VERSION + + tmp_path = None + try: + # The reader must be closed before the temp file can replace the original file, + # as Windows can't replace files that are still opened + with pq.ParquetFile(str(self.uri)) as pq_file: + if compression is None: + compression = self._detect_compression(pq_file.metadata) + if compression == "zstd" and compression_level is None: + compression_level = 15 + tmp_path = self._rewrite( + pq_file, schema_map, compression, compression_level, geoparquet_version + ) + if tmp_path is None: + return False + os.replace(tmp_path, self.uri) + except Exception: + if tmp_path and os.path.exists(tmp_path): + os.unlink(tmp_path) + raise + + # The file has changed, invalidate the cached metadata + self.pq_metadata = None + self.pq_schema = None + return True + + # Rewrites the Parquet file to a temp file and returns its path, + # or returns None if the file needs no changes + def _rewrite( + self, + pq_file: pq.ParquetFile, + schema_map: SchemaMapping, + compression: Optional[str], + compression_level: Optional[int], + geoparquet_version: str, + ) -> Optional[str]: + existing_schema = pq_file.schema_arrow + col_names = existing_schema.names + assert "geometry" in col_names, "Missing geometry column in the Parquet file" + + collection = self.get_collection() + schemas = collection.merge_schemas(schema_map) + has_multiple_collections = len(collection.get_schemas()) > 1 + props = schemas.get("properties", {}) + + required_columns = {"geometry"} + if "id" in col_names: + required_columns.add("id") + if not has_multiple_collections: + collection_only = collection.get_collection_only_properties(schema_map=schema_map) + required_columns |= { + r + for r in schemas.get("required", []) + if r in col_names and r not in collection_only + } + + add_bbox = geoparquet_version != "1.0.0" and "bbox" not in col_names + + # Update the GeoParquet and collection metadata + metadata = existing_schema.metadata or {} + metadata[b"collection"] = json.dumps(collection, cls=VecorelJSONEncoder).encode("utf-8") + if b"geo" in metadata: + geo = json.loads(metadata[b"geo"]) + geo["version"] = geoparquet_version + if geoparquet_version != "1.0.0" and (add_bbox or "bbox" in col_names): + primary_column = geo.get("primary_column", "geometry") + column = geo.get("columns", {}).get(primary_column) + if column is not None: + column["covering"] = { + "bbox": { + "xmin": ["bbox", "xmin"], + "ymin": ["bbox", "ymin"], + "xmax": ["bbox", "xmax"], + "ymax": ["bbox", "ymax"], + } + } + metadata[b"geo"] = json.dumps(geo).encode("utf-8") + + # Build a new Arrow schema with the data types from the Vecorel schemas, + # otherwise with normalized data types, and with adjusted nullability. + # The bbox covering column is kept as written (a float64 struct, like + # the GeoDataFrame-based codepath writes it), not as the schema's + # bounding-box type. + new_fields = [] + for field in existing_schema: + pa_type = None + prop_schema = props.get(field.name) if field.name != "bbox" else None + if prop_schema is not None: + try: + pa_type = get_pyarrow_type(prop_schema) + except Exception as e: + self.warning(f"{field.name}: Can't create data type from schema: {e}") + if pa_type is None: + pa_type = normalize_pa_type(field.type) + new_fields.append( + pa.field( + field.name, + pa_type, + nullable=field.nullable and field.name not in required_columns, + metadata=field.metadata, + ) + ) + + if add_bbox: + new_fields.append( + pa.field( + "bbox", + pa.struct( + [ + ("xmin", pa.float64()), + ("ymin", pa.float64()), + ("xmax", pa.float64()), + ("ymax", pa.float64()), + ] + ), + ) + ) + new_schema = pa.schema(new_fields, metadata=metadata) + + # Nothing to change? Then leave the file untouched. + if not add_bbox and new_schema.equals(existing_schema, check_metadata=True): + return None + + # Streamingly rewrite the file to a temp file + with NamedTemporaryFile("wb", delete=False, dir=self.uri.parent, suffix=".parquet") as tmp: + tmp_path = tmp.name + + writer = pq.ParquetWriter( + tmp_path, + new_schema, + compression=compression, + compression_level=compression_level, + use_dictionary=True, + write_statistics=True, + ) + try: + for rg in range(pq_file.num_row_groups): + tbl = pq_file.read_row_group(rg) + if add_bbox: + # array of shape (n, 4) with minx, miny, maxx, maxy + bounds = from_wkb(tbl["geometry"]).bounds + bbox_array = StructArray.from_arrays( + [bounds[:, 0], bounds[:, 1], bounds[:, 2], bounds[:, 3]], + names=["xmin", "ymin", "xmax", "ymax"], + ) + tbl = tbl.append_column("bbox", bbox_array) + # Ensure the table adheres to the new schema (types, nullability) + if tbl.schema != new_schema: + tbl = tbl.cast(new_schema, safe=False) + writer.write_table(tbl) + finally: + writer.close() + + return tmp_path + # kwargs: # if num = None => kwargs go into pq.read_table # if num is set => kwargs go into pg.ParquetFile diff --git a/vecorel_cli/parquet/types.py b/vecorel_cli/parquet/types.py index 4d3b04b..26ad874 100644 --- a/vecorel_cli/parquet/types.py +++ b/vecorel_cli/parquet/types.py @@ -183,6 +183,40 @@ def get_pyarrow_type(schema): return None +def normalize_pa_field(field: pa.Field) -> pa.Field: + return pa.field( + field.name, normalize_pa_type(field.type), nullable=field.nullable, metadata=field.metadata + ) + + +def normalize_pa_type(dtype: pa.DataType) -> pa.DataType: + """ + Convert Arrow data types to the canonical types required by the Vecorel SDL, + e.g. large_string -> string, large_binary -> binary, timestamps -> timestamp[ms, UTC]. + Non-canonical types can occur when reading files written by external tools + such as DuckDB or GDAL. + """ + if pat.is_large_string(dtype) or pat.is_string_view(dtype): + return pa.string() + if pat.is_large_binary(dtype) or pat.is_binary_view(dtype): + return pa.binary() + if pat.is_timestamp(dtype): + # naive timestamps are assumed to be in UTC + return pa.timestamp("ms", tz="UTC") + if ( + pat.is_list(dtype) + or pat.is_large_list(dtype) + or pat.is_list_view(dtype) + or pat.is_large_list_view(dtype) + ): + return pa.list_(normalize_pa_field(dtype.value_field)) + if pat.is_struct(dtype): + return pa.struct([normalize_pa_field(dtype.field(i)) for i in range(dtype.num_fields)]) + if pat.is_map(dtype): + return pa.map_(normalize_pa_type(dtype.key_type), normalize_pa_type(dtype.item_type)) + return dtype + + def get_pyarrow_type_for_geopandas(dtype): """ geopandas datatypes to pyarrow datatypes diff --git a/vecorel_cli/vecorel/hilbert.py b/vecorel_cli/vecorel/hilbert.py index da182e2..ad51439 100644 --- a/vecorel_cli/vecorel/hilbert.py +++ b/vecorel_cli/vecorel/hilbert.py @@ -111,6 +111,129 @@ def hilbert_distances_from_bounds( return _encode(level, x, y).astype(np.uint64, copy=False) +def hilbert_reference_bounds(crs, fallback=None): + """The CRS's own reference extent for the Hilbert grid, or ``fallback``. + + ``crs_total_bounds`` derives a dataset-independent grid from the CRS's area + of use, so parts converted separately sort into the same order. A CRS given + as a bare projection definition has no area of use to derive it from — + Austria's 2018 edition ships an unnamed "MGI / Austria Lambert" that pyproj + cannot match to an EPSG code — and there the data's own extent orders the + file just as well; only mergeability with separately converted parts is + lost. Returns None when neither is available, leaving the caller to decide + whether that is fatal. + """ + try: + return crs_total_bounds(crs) + except ValueError: + return tuple(fallback) if fallback else None + + +def bounds_array_for_table(table, primary_col: str) -> np.ndarray: + """Return an (N, 4) float64 array of [xmin, ymin, xmax, ymax] per feature. + + Uses the GeoParquet 1.1.0 covering ``bbox`` struct column when present + (zero-decode); otherwise falls back to decoding WKB. + """ + import pyarrow as pa + + if "bbox" in table.column_names and pa.types.is_struct(table.column("bbox").type): + arr = table.column("bbox").combine_chunks() + return np.column_stack( + [ + arr.field("xmin").to_numpy(zero_copy_only=False), + arr.field("ymin").to_numpy(zero_copy_only=False), + arr.field("xmax").to_numpy(zero_copy_only=False), + arr.field("ymax").to_numpy(zero_copy_only=False), + ] + ).astype(np.float64, copy=False) + + import shapely + + wkb_list = table.column(primary_col).combine_chunks().to_pylist() + geoms = shapely.from_wkb(wkb_list) + return shapely.bounds(geoms) + + +def hilbert_keys_for_table(table, primary_col: str, total_bounds) -> np.ndarray: + bounds = bounds_array_for_table(table, primary_col) + return hilbert_distances_from_bounds(bounds, total_bounds) + + +def ensure_hilbert_sorted( + path: str, + primary_col: str, + total_bounds, + compression: str, + compression_level=None, + row_group_size=None, +) -> bool: + """If the GeoParquet file at ``path`` is already Hilbert-sorted against + ``total_bounds``, leave it untouched and return False. Otherwise sort it + in place and return True. + + The whole file is loaded into memory once for the sort. Schema metadata + (``geo``, collection JSON, etc.) is preserved. + """ + import pyarrow as pa + import pyarrow.parquet as pq + + # cheap check first: the keys only need the bbox covering column + with pq.ParquetFile(path) as pf: + has_bbox = "bbox" in pf.schema_arrow.names + probe = pf.read(columns=["bbox"]) if has_bbox else pf.read() + hilberts = hilbert_keys_for_table(probe, primary_col, total_bounds) + # NB: hilberts is uint64; never use np.diff for monotonicity here — uint + # underflow makes any descent wrap to a huge positive and fool the check. + if hilberts.size <= 1 or bool(np.all(hilberts[1:] >= hilberts[:-1])): + return False + + with pq.ParquetFile(path) as pf: + table = pf.read() + metadata = pf.schema_arrow.metadata + narrow_schema = table.schema.with_metadata(metadata) + + # int32 offsets of plain binary/string columns overflow when a take() + # concatenates >2 GB of chunks (large WKB columns); widen them for the + # take, and cast each written batch back so the file keeps its original + # schema (Parquet's physical BYTE_ARRAY is identical either way). + fields = [] + widened = False + for f in table.schema: + if pa.types.is_binary(f.type): + fields.append(f.with_type(pa.large_binary())) + widened = True + elif pa.types.is_string(f.type): + fields.append(f.with_type(pa.large_string())) + widened = True + else: + fields.append(f) + if widened: + table = table.cast(pa.schema(fields, metadata=table.schema.metadata)) + + order = np.argsort(hilberts, kind="stable") + sorted_table = table.take(pa.array(order)) + sorted_table = sorted_table.replace_schema_metadata(metadata) + + write_kwargs = {"compression": compression} + if compression_level is not None: + write_kwargs["compression_level"] = compression_level + + # Write batch-wise against the ORIGINAL narrow schema: each batch is far + # below the int32 offset limit, so the down-cast is safe, the geo/collection + # metadata is preserved, and the rewritten file schema-matches untouched + # pre-sorted siblings during merges. + step = row_group_size or 131_072 + writer = pq.ParquetWriter(path, narrow_schema, **write_kwargs) + try: + for start in range(0, sorted_table.num_rows, step): + batch = sorted_table.slice(start, step) + writer.write_table(batch.cast(narrow_schema) if widened else batch) + finally: + writer.close() + return True + + def hilbert_sort_geodataframe(gdf, level: int = HILBERT_DEFAULT_LEVEL): """Return a copy of ``gdf`` sorted by Hilbert distance against the CRS's total bounds. The original index is reset. @@ -136,7 +259,11 @@ def hilbert_sort_geodataframe(gdf, level: int = HILBERT_DEFAULT_LEVEL): __all__ = [ + "bounds_array_for_table", "crs_total_bounds", + "ensure_hilbert_sorted", "hilbert_distances_from_bounds", + "hilbert_keys_for_table", + "hilbert_reference_bounds", "hilbert_sort_geodataframe", ] From 2c72a53b47488006ba6271ad331ba5de8fed6ab0 Mon Sep 17 00:00:00 2001 From: Matthias Mohr Date: Fri, 11 Sep 2026 16:34:34 +0200 Subject: [PATCH 2/3] Code review --- CHANGELOG.md | 3 +- tests/test_convert_duckdb.py | 20 +- tests/test_encoding_geoparquet.py | 15 +- vecorel_cli/conversion/base.py | 114 ++++------- vecorel_cli/conversion/duckdb.py | 319 ++++++++++++++++++++++++----- vecorel_cli/encoding/geoparquet.py | 159 ++++++++------ vecorel_cli/vecorel/hilbert.py | 73 +++++-- 7 files changed, 497 insertions(+), 206 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bf2b168..1be5503 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,7 +15,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. constant `collection` column. Previously files without constant columns were written without any collection id. - Converters drop rows that can never validate (missing required values, empty or - missing geometries), bounded by the new `max_dropped_share` (default 1%). + missing geometries). Missing required values are dropped only up to the new + `max_dropped_share` (default 1%), above it the conversion fails. - Converters fail when both `sources` and `variants` are declared. - Converters warn when no column is mapped to `id` and when the id column is not unique. - Converters load all schemas upfront with retries, so a temporary network issue diff --git a/tests/test_convert_duckdb.py b/tests/test_convert_duckdb.py index fc6169b..4cbaa7f 100644 --- a/tests/test_convert_duckdb.py +++ b/tests/test_convert_duckdb.py @@ -98,6 +98,19 @@ def test_duckdb_converter(tmp_folder): assert validation.errors == [] +def test_duckdb_converter_index_as_id(tmp_folder): + src = _source_file(tmp_folder) + dest = tmp_folder / "converted.parquet" + + IndexConverter = type("IndexConverter", (DuckDBBaseConverter,), {**CONFIG, "index_as_id": True}) + IndexConverter().convert(dest, input_files={src: "source.parquet"}) + + result = gpd.read_parquet(dest) + # row numbers are assigned before geometries are split, + # so the parts of one source feature share an id (like the default codepath) + assert sorted(result["id"]) == ["0", "1", "1", "2", "2", "3"] + + def test_duckdb_converter_original_geometries(tmp_folder): src = _source_file(tmp_folder) dest = tmp_folder / "converted.parquet" @@ -165,8 +178,11 @@ def test_codepath_parity(tmp_folder): assert pandas_geo[key] == duckdb_geo[key] pandas_column = pandas_geo["columns"]["geometry"] duckdb_column = duckdb_geo["columns"]["geometry"] - for key in ("encoding", "covering"): - assert pandas_column[key] == duckdb_column[key] + for key in ("encoding", "covering", "crs", "bbox"): + assert pandas_column.get(key) == duckdb_column.get(key), f"geo {key} differs" + assert sorted(pandas_column.get("geometry_types", [])) == sorted( + duckdb_column.get("geometry_types", []) + ) # Same packaging assert pandas_compression == duckdb_compression diff --git a/tests/test_encoding_geoparquet.py b/tests/test_encoding_geoparquet.py index b51bfac..c7a9c86 100644 --- a/tests/test_encoding_geoparquet.py +++ b/tests/test_encoding_geoparquet.py @@ -86,9 +86,12 @@ def test_postprocess(tmp_parquet_file): assert gp.postprocess(geoparquet_version="1.1.0", compression="zstd") is True # a second run detects the compliant file and doesn't rewrite it assert gp.postprocess(geoparquet_version="1.1.0", compression="zstd") is False + # unless a different compression is requested + assert gp.postprocess(geoparquet_version="1.1.0", compression="brotli") is True - result = pq.ParquetFile(tmp_parquet_file) - schema = result.schema_arrow + with pq.ParquetFile(tmp_parquet_file) as result: + schema = result.schema_arrow + metadata = result.metadata.metadata field = schema.field("geometry") assert field.type == pa.binary() @@ -103,7 +106,6 @@ def test_postprocess(tmp_parquet_file): assert pat.is_struct(bbox.type) assert bbox.type.field("xmin").type == pa.float64() - metadata = result.metadata.metadata geo = json.loads(metadata[b"geo"]) assert geo["version"] == "1.1.0" assert geo["columns"]["geometry"]["covering"]["bbox"]["xmin"] == ["bbox", "xmin"] @@ -114,3 +116,10 @@ def test_postprocess(tmp_parquet_file): assert data.num_rows == src.num_rows assert data["id"].to_pylist() == src["id"].to_pylist() assert data["geometry"].to_pylist() == src["geometry"].to_pylist() + + # a downgrade removes the covering metadata, which only exists since GeoParquet 1.1 + assert gp.postprocess(geoparquet_version="1.0.0", compression="brotli") is True + with pq.ParquetFile(tmp_parquet_file) as pf: + geo = json.loads(pf.metadata.metadata[b"geo"]) + assert geo["version"] == "1.0.0" + assert "covering" not in geo["columns"]["geometry"] diff --git a/vecorel_cli/conversion/base.py b/vecorel_cli/conversion/base.py index 9c3a954..9b2a703 100644 --- a/vecorel_cli/conversion/base.py +++ b/vecorel_cli/conversion/base.py @@ -76,13 +76,8 @@ def __init__(self, *args, **kwargs): setattr(self, key, copy(item)) def _require_one_source_of_urls(self): - """Fail when both `sources` and `variants` are declared. - - The converter takes `sources` when it is set and ignores the variants - entirely, so `--variant 2011` silently converts whatever `sources` - points at. A converter that inherits variants it does not want says so - with `variants = {}`. - """ + """Fail when both `sources` and `variants` are declared: + `sources` wins and every `--variant` would silently convert the same file.""" if self.sources and self.variants: raise ValueError( f"{type(self).__name__} declares both sources and variants; sources wins " @@ -91,15 +86,9 @@ def _require_one_source_of_urls(self): ) def _check_id_mapping(self): - """Warn before converting when nothing will end up as `id`. - - Vecorel requires the identifier, and nothing downstream enforces it: - the converter drops columns no mapping names, so a converter without - one simply writes a file without `id` and validates. The subtler half - is `index_as_id = True`, which fills the column and the same drop step - removes it again, because `columns` never named it. A converter with - no natural key sets both `index_as_id` and `"id": "id"`. - """ + """Warn before converting when nothing is mapped to the required `id` property. + Unmapped columns are dropped, which also removes the column filled by + `index_as_id` unless the converter maps `"id": "id"`.""" targets = set() for value in list(self.columns.values()) + list(self.column_additions or {}): targets.update(value if isinstance(value, (list, tuple)) else [value]) @@ -114,17 +103,9 @@ def _check_id_mapping(self): def _check_unique_ids(self, gdf, columns): """Warn when the column that becomes `id` does not identify a feature. - - Vecorel asks for one identifier per feature, but nothing measures it: - a converter can map a group id shared by thousands of features, and - every converter that reads several files and sets `index_as_id` - repeats the same index once per file. This runs before geometries are - exploded, so it judges what the converter assigned rather than the - split parts of one source feature. - - A missing id is a different failure, dropped under a bounded rule - afterwards, so it is not counted here: a column that identifies all - features except the few that carry none is a perfectly good identifier. + Runs before geometries are exploded, so it judges what the converter + assigned rather than the split parts of one source feature. Null ids + are not counted here; they are dropped under a bounded rule afterwards. """ sources = [ k for k, v in columns.items() if "id" in (v if isinstance(v, (list, tuple)) else [v]) @@ -133,9 +114,6 @@ def _check_unique_ids(self, gdf, columns): (c for c in sources if c in gdf.columns), "id" if "id" in gdf.columns else None ) if column is None: - # The mapping exists (convert() checks that) but the data does not - # carry it, and the unlisted-column drop then writes a file - # without `id` that validates. self.warning( f"{type(self).__name__}: none of the columns mapped to 'id' " f"({', '.join(sources) or 'none'}) is in this source; it has " @@ -156,18 +134,13 @@ def _check_unique_ids(self, gdf, columns): ) def _drop_incomplete_rows(self, gdf, columns): - """Drop rows that can never validate: null values in a property the - schemas require, and empty or missing geometries (which also cannot - be tiled or Hilbert-sorted). Bounded by ``max_dropped_share``: above - that share the rows are kept (the converter needs fixing, and quietly - dropping large parts of a dataset would hide that), with a warning - that the output will not validate. - - The required properties come from the merged schemas (the core - specification, the declared extensions and the custom schemas), minus - the collection-only properties and the geometry, which is checked - separately below. - """ + """Drop rows that can never validate. Rows with null values in a + schema-required property are dropped up to ``max_dropped_share``; + above it the conversion fails, as quietly dropping large parts of a + dataset would hide that the converter needs fixing (and the writer + rejects nulls in the non-nullable required fields anyway). Rows with + an empty or missing geometry are always dropped: they cannot survive + the geometry processing anyway and would break the Hilbert sort.""" collection = self.create_collection(self.id.strip()) schemas = collection.merge_schemas({}) collection_only = set(collection.get_collection_only_properties()) @@ -175,6 +148,10 @@ def _drop_incomplete_rows(self, gdf, columns): r for r in schemas.get("required", []) if r != "geometry" and r not in collection_only ] + # One combined mask, so max_dropped_share bounds the total share + invalid = pd.Series(False, index=gdf.index) + reasons = [] + # This runs before columns are renamed, so look up the source column for key in required: for src, dst in columns.items(): @@ -182,42 +159,43 @@ def _drop_incomplete_rows(self, gdf, columns): if key in targets and src in gdf.columns: nulls = gdf[src].isna() if nulls.any(): - share = nulls.mean() - if share > self.max_dropped_share: - self.warning( - f"{int(nulls.sum())} of {len(gdf)} rows ({share:.1%}) have no " - f"{key} ({src}); the output will not validate — fix the converter" - ) - else: - self.warning( - f"Dropping {int(nulls.sum())} rows without a value for {key} ({src})" - ) - gdf = gdf[~nulls] + reasons.append(f"{int(nulls.sum())} without a value for {key} ({src})") + invalid |= nulls + + if invalid.any(): + share = invalid.mean() + details = "; ".join(reasons) + if share > self.max_dropped_share: + raise ValueError( + f"{int(invalid.sum())} of {len(gdf)} rows ({share:.1%}) have no value for " + f"a required property ({details}); fix the converter instead of dropping them" + ) + self.warning( + f"Dropping {int(invalid.sum())} of {len(gdf)} rows that can never " + f"validate ({details})" + ) + gdf = gdf[~invalid] if gdf.active_geometry_name is not None: geom = gdf.geometry blank = geom.isna() | geom.is_empty if blank.any(): share = blank.mean() + message = ( + f"Dropping {int(blank.sum())} of {len(gdf)} rows with an empty " + f"or missing geometry" + ) if share > self.max_dropped_share: - self.warning( - f"{int(blank.sum())} of {len(gdf)} rows ({share:.1%}) have an empty or " - f"missing geometry; the output will not validate — fix the converter" - ) - else: - self.warning( - f"Dropping {int(blank.sum())} rows with an empty or missing geometry" - ) - gdf = gdf[~blank] + message += f" ({share:.1%}, exceeds max_dropped_share) — fix the converter" + self.warning(message) + gdf = gdf[~blank] return gdf def _prewarm_schemas(self): - """Fetch every schema this conversion will need before doing any real - work, with retries. The schema hosts fail intermittently; without this, - a transient blip after a long source download kills the conversion at - the very last step. load_file caches per process, so a successful - pre-warm makes the write network-free.""" + """Fetch every schema this conversion will need upfront, with retries, + so a transient schema-host blip cannot kill the conversion at the very + last step. load_file caches per process.""" import time from ..vecorel.util import load_file @@ -602,8 +580,6 @@ def convert( columns = list(actual_columns.values()) pq = GeoParquet(output_file) collection = self.create_collection(cid) - # Record the collection id at the collection level rather than as a - # constant column, like the DuckDB-based codepath does collection["collection"] = cid pq.set_collection(collection) diff --git a/vecorel_cli/conversion/duckdb.py b/vecorel_cli/conversion/duckdb.py index 75c2df1..a30c87d 100644 --- a/vecorel_cli/conversion/duckdb.py +++ b/vecorel_cli/conversion/duckdb.py @@ -1,17 +1,27 @@ import json import os from pathlib import Path +from tempfile import NamedTemporaryFile from typing import Optional import duckdb +import numpy as np +import pyarrow as pa import pyarrow.parquet as pq from ..encoding.geojson import VecorelJSONEncoder from ..encoding.geoparquet import GeoParquet -from ..vecorel.hilbert import ensure_hilbert_sorted, hilbert_reference_bounds +from ..vecorel.hilbert import hilbert_keys_for_table, hilbert_reference_bounds from .base import BaseConverter +# COPY doesn't support bound parameters for read_parquet file names, +# so paths are inlined as escaped single-quoted string literals +def _sql_path(path) -> str: + escaped = str(path).replace("'", "''") + return f"'{escaped}'" + + # This converter is experimental, use with caution. # Results may not be fully compliant yet. # Use this primarily for datasets that are too large to be processed by the default converter. @@ -44,7 +54,6 @@ def convert( self._require_one_source_of_urls() self._prewarm_schemas() - # Create output folder if it doesn't exist directory = os.path.dirname(output_file) if directory: os.makedirs(directory, exist_ok=True) @@ -68,50 +77,175 @@ def convert( "avoid_range_request is set, but cache is not used, so this setting has no effect" ) + if isinstance(urls, str): + sources = urls + else: + sources = [url[0] if isinstance(url, tuple) else url for url in urls] + + con = duckdb.connect() + con.install_extension("spatial") + con.load_extension("spatial") + + # Skip mapped columns that the source doesn't carry, like the + # GeoDataFrame-based codepath does + available = { + row[0] + for row in con.execute( + "DESCRIBE SELECT * FROM read_parquet(?, union_by_name=true)", [sources] + ).fetchall() + } + + # DuckDB drops the CRS from the GeoParquet metadata, so take it from the + # sources for the Hilbert grid and the output metadata. The sources are + # combined without reprojection, so they must all use the same CRS. + source_crs = None + for i, source in enumerate([sources] if isinstance(sources, str) else sources): + crs = None + row = con.execute( + "SELECT value FROM parquet_kv_metadata(?) WHERE key = 'geo'", [source] + ).fetchone() + if row: + source_geo = json.loads(bytes(row[0])) + primary_column = source_geo.get("primary_column", "") + crs = source_geo.get("columns", {}).get(primary_column, {}).get("crs") + if i == 0: + source_crs = crs + elif crs != source_crs: + raise ValueError( + f"The sources use different coordinate reference systems: {source} " + "differs from the first source. Reproject the sources to a common CRS." + ) selections = [] + selected_targets = [] for k, v in self.columns.items(): - if k in self.column_migrations: - selections.append(f'{self.column_migrations.get(k)} as "{v}"') - else: - selections.append(f'"{k}" as "{v}"') + targets = list(v) if isinstance(v, (list, tuple)) else [v] + if k == "id" and self.index_as_id: + # 0-based, like the index the GeoDataFrame-based codepath assigns + selections.append('(row_number() OVER () - 1) AS "id"') + selected_targets.append("id") + continue + if k not in available: + self.warning(f"Column '{k}' not found in dataset, removing from schema") + continue + expr = self.column_migrations.get(k, f'"{k}"') + for target in targets: + selections.append(f'{expr} as "{target}"') + selected_targets.append(target) + + collection = self.create_collection(cid) + collection["collection"] = cid + + # Constants pinned to the feature level become literal columns, all others + # end up in the collection metadata (like the dehydration step of the + # GeoDataFrame-based codepath) + addition_params = [] + if self.column_additions: + context = collection.get_collection_context() + for key, value in self.column_additions.items(): + if context.get(key) is False: + selections.append(f'? as "{key}"') + addition_params.append(value) + selected_targets.append(key) + else: + collection[key] = value selection = ", ".join(selections) filters = [] where = "" if self.bbox is not None: + # The filter runs against the source relation, so it must use the + # source column that is mapped to the geometry + geom_source = next( + ( + k + for k, v in self.columns.items() + if "geometry" in (v if isinstance(v, (list, tuple)) else [v]) + ), + "geometry", + ) filters.append( - f"ST_Intersects(geometry, ST_MakeEnvelope({self.bbox[0]}, {self.bbox[1]}, {self.bbox[2]}, {self.bbox[3]}))" + f'ST_Intersects("{geom_source}", ST_MakeEnvelope({self.bbox[0]}, {self.bbox[1]}, {self.bbox[2]}, {self.bbox[3]}))' ) for k, v in self.column_filters.items(): filters.append(v) if len(filters) > 0: where = f"WHERE {' AND '.join(filters)}" - if isinstance(urls, str): - sources = f'"{urls}"' - else: - paths = [] - for url in urls: - if isinstance(url, tuple): - paths.append(f'"{url[0]}"') - else: - paths.append(f'"{url}"') - sources = "[" + ",".join(paths) + "]" - - collection = self.create_collection(cid) - collection.update(self.column_additions) - collection["collection"] = self.id - if isinstance(output_file, Path): output_file = str(output_file) collection_json = json.dumps(collection, cls=VecorelJSONEncoder).encode("utf-8") + if isinstance(sources, str): + sources_sql = _sql_path(sources) + else: + sources_sql = "[" + ",".join(_sql_path(path) for path in sources) + "]" source_query = f""" SELECT {selection} - FROM read_parquet({sources}, union_by_name=true) + FROM read_parquet({sources_sql}, union_by_name=true) {where} """ + + # Same bounded null-value drop, empty-geometry drop and id uniqueness + # check as in the GeoDataFrame-based codepath, in one scan + schemas = collection.merge_schemas({}) + collection_only = set(collection.get_collection_only_properties()) + required = [ + r + for r in schemas.get("required", []) + if r != "geometry" and r not in collection_only and r in selected_targets + ] + stats = ["count(*)"] + null_cond = None + if required: + null_cond = " OR ".join(f'"{target}" IS NULL' for target in required) + stats.append(f"count(*) FILTER (WHERE {null_cond})") + # row numbers are unique by construction + check_ids = "id" in selected_targets and not self.index_as_id + if check_ids: + stats.append('count("id")') + stats.append('count(DISTINCT "id")') + blank_cond = None + if "geometry" in selected_targets: + blank_cond = '"geometry" IS NULL OR ST_IsEmpty("geometry")' + stats.append(f"count(*) FILTER (WHERE {blank_cond})") + if len(stats) > 1: + values = list( + con.execute( + f"SELECT {', '.join(stats)} FROM ({source_query})", addition_params + ).fetchone() + ) + total = values.pop(0) + invalid = values.pop(0) if null_cond else 0 + if check_ids: + non_null = values.pop(0) + distinct = values.pop(0) + if distinct < non_null: + self.warning( + f"{type(self).__name__}: 'id' is not unique — {non_null - distinct:,} " + f"of {non_null:,} rows repeat an id, so it cannot be `id`. Map a column " + "that identifies a feature, or build one from the source's key columns." + ) + blanks = values.pop(0) if blank_cond else 0 + if invalid: + share = invalid / total + if share > self.max_dropped_share: + raise ValueError( + f"{invalid} of {total} rows ({share:.1%}) have no value for a required " + f"property ({null_cond}); fix the converter instead of dropping them" + ) + self.warning( + f"Dropping {invalid} of {total} rows without a value for a " + f"required property ({null_cond})" + ) + source_query = f"SELECT * FROM ({source_query}) WHERE NOT ({null_cond})" + if blanks: + share = blanks / total + message = f"Dropping {blanks} of {total} rows with an empty or missing geometry" + if share > self.max_dropped_share: + message += f" ({share:.1%}, exceeds max_dropped_share) — fix the converter" + self.warning(message) + source_query = f"SELECT * FROM ({source_query}) WHERE NOT ({blank_cond})" if original_geometries: query = source_query else: @@ -132,13 +266,10 @@ def convert( WHERE ST_GeometryType(geom) = 'POLYGON' AND ST_IsValid(geom) """ - con = duckdb.connect() - con.install_extension("spatial") - con.load_extension("spatial") # No ORDER BY here: ST_Hilbert without bounds is meaningless (whole # countries collapse into a handful of cells), and with bounds it uses # a different reference grid than the rest of the pipeline. The - # canonical in-place Hilbert sort below runs after post-processing. + # canonical Hilbert sort below runs on the written file. con.execute( f""" COPY ({query}) TO ? ( @@ -150,41 +281,123 @@ def convert( }} ) """, - [output_file, compression, collection_json], + [*addition_params, output_file, compression, collection_json], ) - # Post-process the written Parquet file to a compliant GeoParquet file - # (canonical data types, nullability, bbox column, metadata) - gp = GeoParquet(output_file) - gp.set_collection(collection) - try: - gp.postprocess( - compression=compression, - compression_level=compression_level, - geoparquet_version=geoparquet_version, - ) - except Exception as e: - self.warning(f"GeoParquet post-processing failed: {e}") - - # Canonical spatial ordering against the CRS-derived Hilbert grid, - # the same grid the GeoDataFrame-based converter sorts against + # Sort against the same CRS-derived Hilbert grid as the + # GeoDataFrame-based codepath with pq.ParquetFile(output_file) as pf: meta = pf.schema_arrow.metadata or {} if b"geo" in meta: geo = json.loads(meta[b"geo"]) primary = geo["primary_column"] - crs = geo["columns"][primary].get("crs") or "EPSG:4326" + crs = geo["columns"][primary].get("crs") or source_crs or "EPSG:4326" bounds = hilbert_reference_bounds(crs, geo["columns"][primary].get("bbox")) if bounds is None: self.warning("CRS declares no area of use; skipping spatial ordering") - elif ensure_hilbert_sorted( - output_file, - primary, - bounds, - compression, - compression_level, - row_group_size=row_group_size, - ): - self.info("Sorted output into Hilbert order") + else: + keys_path, is_sorted = self._write_hilbert_keys(output_file, primary, bounds) + try: + if not is_sorted: + self._sort_output( + con, + output_file, + keys_path, + compression, + collection_json, + row_group_size, + ) + self.info("Sorted output into Hilbert order") + finally: + if os.path.exists(keys_path): + os.unlink(keys_path) + + gp = GeoParquet(output_file) + gp.set_collection(collection) + gp.postprocess( + compression=compression, + compression_level=compression_level, + geoparquet_version=geoparquet_version, + crs=source_crs, + ) return output_file + + # Streams the Hilbert keys per row group to a sidecar file and reports + # whether the file is already sorted, so memory stays bounded + def _write_hilbert_keys(self, output_file, primary, bounds): + directory = os.path.dirname(output_file) or "." + with NamedTemporaryFile("wb", delete=False, dir=directory, suffix=".parquet") as tmp: + keys_path = tmp.name + schema = pa.schema([("hilbert", pa.uint64()), ("ordinal", pa.uint64())]) + is_sorted = True + last = None + ordinal = 0 + try: + writer = pq.ParquetWriter(keys_path, schema) + try: + with pq.ParquetFile(output_file) as pf: + file_schema = pf.schema_arrow + has_bbox = "bbox" in file_schema.names and pa.types.is_struct( + file_schema.field("bbox").type + ) + columns = ["bbox"] if has_bbox else [primary] + for rg in range(pf.num_row_groups): + table = pf.read_row_group(rg, columns=columns) + keys = hilbert_keys_for_table(table, primary, bounds) + if keys.size: + if last is not None and keys[0] < last: + is_sorted = False + if not bool(np.all(keys[1:] >= keys[:-1])): + is_sorted = False + last = keys[-1] + # The ordinal makes ties keep their original order, like + # the stable sort of the GeoDataFrame-based codepath + ordinals = np.arange(ordinal, ordinal + keys.size, dtype=np.uint64) + ordinal += keys.size + writer.write_table( + pa.table({"hilbert": keys, "ordinal": ordinals}, schema=schema) + ) + finally: + writer.close() + except Exception: + if os.path.exists(keys_path): + os.unlink(keys_path) + raise + return keys_path, is_sorted + + # Rewrites the file in the order given by the Hilbert keys. + # DuckDB sorts externally (spilling to disk if needed), so this works for + # datasets that don't fit into memory. + def _sort_output( + self, con, output_file, keys_path, compression, collection_json, row_group_size + ): + directory = os.path.dirname(output_file) or "." + tmp_path = None + try: + with NamedTemporaryFile("wb", delete=False, dir=directory, suffix=".parquet") as tmp: + tmp_path = tmp.name + + con.execute( + f""" + COPY ( + SELECT d.* + FROM read_parquet({_sql_path(output_file)}) d + POSITIONAL JOIN read_parquet({_sql_path(keys_path)}) k + ORDER BY k.hilbert, k.ordinal + ) TO ? ( + FORMAT parquet, + ROW_GROUP_SIZE {row_group_size}, + compression ?, + KV_METADATA {{ + collection: ?, + }} + ) + """, + [tmp_path, compression, collection_json], + ) + os.replace(tmp_path, output_file) + except Exception: + if tmp_path and os.path.exists(tmp_path): + os.unlink(tmp_path) + raise diff --git a/vecorel_cli/encoding/geoparquet.py b/vecorel_cli/encoding/geoparquet.py index 6fe8a39..7f2906b 100644 --- a/vecorel_cli/encoding/geoparquet.py +++ b/vecorel_cli/encoding/geoparquet.py @@ -124,8 +124,10 @@ def get_compression(self) -> Optional[str]: @staticmethod def _detect_compression(metadata: pq.FileMetaData) -> Optional[str]: - compressions = set() + if metadata.num_row_groups == 0: + return None + compressions = set() row_group = metadata.row_group(0) for col_idx in range(row_group.num_columns): column = row_group.column(col_idx) @@ -274,6 +276,7 @@ def postprocess( compression: Optional[str] = None, compression_level: Optional[int] = None, geoparquet_version: Optional[str] = None, + crs=None, # the CRS to record in the GeoParquet metadata, e.g. from the source file **kwargs, # capture unknown arguments ) -> bool: """ @@ -305,12 +308,21 @@ def postprocess( # The reader must be closed before the temp file can replace the original file, # as Windows can't replace files that are still opened with pq.ParquetFile(str(self.uri)) as pq_file: + existing_compression = self._detect_compression(pq_file.metadata) if compression is None: - compression = self._detect_compression(pq_file.metadata) + compression = existing_compression + if compression == "mixed": # per-column codecs are not preserved + compression = "zstd" if compression == "zstd" and compression_level is None: compression_level = 15 tmp_path = self._rewrite( - pq_file, schema_map, compression, compression_level, geoparquet_version + pq_file, + schema_map, + compression, + compression_level, + geoparquet_version, + crs=crs, + compression_changed=compression != existing_compression, ) if tmp_path is None: return False @@ -334,6 +346,8 @@ def _rewrite( compression: Optional[str], compression_level: Optional[int], geoparquet_version: str, + crs=None, + compression_changed: bool = False, ) -> Optional[str]: existing_schema = pq_file.schema_arrow col_names = existing_schema.names @@ -344,44 +358,57 @@ def _rewrite( has_multiple_collections = len(collection.get_schemas()) > 1 props = schemas.get("properties", {}) - required_columns = {"geometry"} - if "id" in col_names: - required_columns.add("id") + # Must mirror the nullability rule of writer and validator: + # nullable = not required, and everything is nullable for files + # with multiple collections + required_columns = set() if not has_multiple_collections: - collection_only = collection.get_collection_only_properties(schema_map=schema_map) - required_columns |= { - r - for r in schemas.get("required", []) - if r in col_names and r not in collection_only - } - + required_columns = {"geometry"} + if "id" in col_names: + required_columns.add("id") + required_columns |= {r for r in schemas.get("required", []) if r in col_names} + + if "bbox" in col_names: + bbox_type = existing_schema.field("bbox").type + children = ( + {bbox_type.field(i).name for i in range(bbox_type.num_fields)} + if pa.types.is_struct(bbox_type) + else set() + ) + if children != {"xmin", "ymin", "xmax", "ymax"}: + raise ValueError( + f"The bbox column is not a GeoParquet covering struct, is {bbox_type}" + ) add_bbox = geoparquet_version != "1.0.0" and "bbox" not in col_names - # Update the GeoParquet and collection metadata metadata = existing_schema.metadata or {} + if b"geo" not in metadata: + # Creating GeoParquet metadata from scratch would require CRS and + # geometry information this file doesn't carry + raise ValueError("The Parquet file has no GeoParquet metadata") metadata[b"collection"] = json.dumps(collection, cls=VecorelJSONEncoder).encode("utf-8") - if b"geo" in metadata: - geo = json.loads(metadata[b"geo"]) - geo["version"] = geoparquet_version - if geoparquet_version != "1.0.0" and (add_bbox or "bbox" in col_names): - primary_column = geo.get("primary_column", "geometry") - column = geo.get("columns", {}).get(primary_column) - if column is not None: - column["covering"] = { - "bbox": { - "xmin": ["bbox", "xmin"], - "ymin": ["bbox", "ymin"], - "xmax": ["bbox", "xmax"], - "ymax": ["bbox", "ymax"], - } + geo = json.loads(metadata[b"geo"]) + geo["version"] = geoparquet_version + column = geo.get("columns", {}).get(geo.get("primary_column", "geometry")) + if column is not None: + if crs is not None: + column["crs"] = crs + if geoparquet_version == "1.0.0": + # covering metadata only exists since GeoParquet 1.1 + column.pop("covering", None) + elif add_bbox or "bbox" in col_names: + column["covering"] = { + "bbox": { + "xmin": ["bbox", "xmin"], + "ymin": ["bbox", "ymin"], + "xmax": ["bbox", "xmax"], + "ymax": ["bbox", "ymax"], } - metadata[b"geo"] = json.dumps(geo).encode("utf-8") + } + metadata[b"geo"] = json.dumps(geo).encode("utf-8") - # Build a new Arrow schema with the data types from the Vecorel schemas, - # otherwise with normalized data types, and with adjusted nullability. - # The bbox covering column is kept as written (a float64 struct, like - # the GeoDataFrame-based codepath writes it), not as the schema's - # bounding-box type. + # The bbox covering column is kept as written (a float64 struct, like the + # GeoDataFrame-based codepath writes it), not as the schema's bounding-box type new_fields = [] for field in existing_schema: pa_type = None @@ -397,7 +424,7 @@ def _rewrite( pa.field( field.name, pa_type, - nullable=field.nullable and field.name not in required_columns, + nullable=field.name not in required_columns, metadata=field.metadata, ) ) @@ -418,39 +445,47 @@ def _rewrite( ) new_schema = pa.schema(new_fields, metadata=metadata) - # Nothing to change? Then leave the file untouched. - if not add_bbox and new_schema.equals(existing_schema, check_metadata=True): + if ( + not add_bbox + and not compression_changed + and new_schema.equals(existing_schema, check_metadata=True) + ): return None - # Streamingly rewrite the file to a temp file with NamedTemporaryFile("wb", delete=False, dir=self.uri.parent, suffix=".parquet") as tmp: tmp_path = tmp.name - writer = pq.ParquetWriter( - tmp_path, - new_schema, - compression=compression, - compression_level=compression_level, - use_dictionary=True, - write_statistics=True, - ) try: - for rg in range(pq_file.num_row_groups): - tbl = pq_file.read_row_group(rg) - if add_bbox: - # array of shape (n, 4) with minx, miny, maxx, maxy - bounds = from_wkb(tbl["geometry"]).bounds - bbox_array = StructArray.from_arrays( - [bounds[:, 0], bounds[:, 1], bounds[:, 2], bounds[:, 3]], - names=["xmin", "ymin", "xmax", "ymax"], - ) - tbl = tbl.append_column("bbox", bbox_array) - # Ensure the table adheres to the new schema (types, nullability) - if tbl.schema != new_schema: - tbl = tbl.cast(new_schema, safe=False) - writer.write_table(tbl) - finally: - writer.close() + writer = pq.ParquetWriter( + tmp_path, + new_schema, + compression=compression, + compression_level=compression_level, + use_dictionary=True, + write_statistics=True, + ) + try: + for rg in range(pq_file.num_row_groups): + tbl = pq_file.read_row_group(rg) + if add_bbox: + # array of shape (n, 4) with minx, miny, maxx, maxy + bounds = from_wkb(tbl["geometry"]).bounds + bbox_array = StructArray.from_arrays( + [bounds[:, 0], bounds[:, 1], bounds[:, 2], bounds[:, 3]], + names=["xmin", "ymin", "xmax", "ymax"], + ) + tbl = tbl.append_column("bbox", bbox_array) + # The safe cast fails on lossy conversions (e.g. out-of-range integers), + # so invalid source data is reported instead of silently corrupted + if tbl.schema != new_schema: + tbl = tbl.cast(new_schema) + writer.write_table(tbl) + finally: + writer.close() + except Exception: + if os.path.exists(tmp_path): + os.unlink(tmp_path) + raise return tmp_path diff --git a/vecorel_cli/vecorel/hilbert.py b/vecorel_cli/vecorel/hilbert.py index ad51439..ff0c03b 100644 --- a/vecorel_cli/vecorel/hilbert.py +++ b/vecorel_cli/vecorel/hilbert.py @@ -126,7 +126,7 @@ def hilbert_reference_bounds(crs, fallback=None): try: return crs_total_bounds(crs) except ValueError: - return tuple(fallback) if fallback else None + return tuple(fallback) if fallback is not None else None def bounds_array_for_table(table, primary_col: str) -> np.ndarray: @@ -160,6 +160,35 @@ def hilbert_keys_for_table(table, primary_col: str, total_bounds) -> np.ndarray: return hilbert_distances_from_bounds(bounds, total_bounds) +def hilbert_keys_for_file(path: str, primary_col: str, total_bounds) -> np.ndarray: + """Hilbert keys for every row of a GeoParquet file, in row order. + + Computed row group by row group, so the file never needs to fit into + memory: only the bbox covering column is read when present (zero-decode), + otherwise the WKB of the primary geometry column. + """ + import pyarrow.parquet as pq + import pyarrow.types as pat + + keys = [] + with pq.ParquetFile(path) as pf: + schema = pf.schema_arrow + has_bbox = "bbox" in schema.names and pat.is_struct(schema.field("bbox").type) + columns = ["bbox"] if has_bbox else [primary_col] + for rg in range(pf.num_row_groups): + table = pf.read_row_group(rg, columns=columns) + keys.append(hilbert_keys_for_table(table, primary_col, total_bounds)) + if not keys: + return np.array([], dtype=np.uint64) + return np.concatenate(keys) + + +def is_hilbert_sorted(keys: np.ndarray) -> bool: + # NB: keys are uint64; never use np.diff for monotonicity here — uint + # underflow makes any descent wrap to a huge positive and fool the check. + return keys.size <= 1 or bool(np.all(keys[1:] >= keys[:-1])) + + def ensure_hilbert_sorted( path: str, primary_col: str, @@ -178,14 +207,9 @@ def ensure_hilbert_sorted( import pyarrow as pa import pyarrow.parquet as pq - # cheap check first: the keys only need the bbox covering column - with pq.ParquetFile(path) as pf: - has_bbox = "bbox" in pf.schema_arrow.names - probe = pf.read(columns=["bbox"]) if has_bbox else pf.read() - hilberts = hilbert_keys_for_table(probe, primary_col, total_bounds) - # NB: hilberts is uint64; never use np.diff for monotonicity here — uint - # underflow makes any descent wrap to a huge positive and fool the check. - if hilberts.size <= 1 or bool(np.all(hilberts[1:] >= hilberts[:-1])): + # cheap check first, streamed row group by row group + hilberts = hilbert_keys_for_file(path, primary_col, total_bounds) + if is_hilbert_sorted(hilberts): return False with pq.ParquetFile(path) as pf: @@ -222,15 +246,30 @@ def ensure_hilbert_sorted( # Write batch-wise against the ORIGINAL narrow schema: each batch is far # below the int32 offset limit, so the down-cast is safe, the geo/collection # metadata is preserved, and the rewritten file schema-matches untouched - # pre-sorted siblings during merges. + # pre-sorted siblings during merges. Write to a temp file first and replace + # atomically, so a failure doesn't destroy the previously valid file. + import os + from tempfile import NamedTemporaryFile + step = row_group_size or 131_072 - writer = pq.ParquetWriter(path, narrow_schema, **write_kwargs) + tmp_path = None try: - for start in range(0, sorted_table.num_rows, step): - batch = sorted_table.slice(start, step) - writer.write_table(batch.cast(narrow_schema) if widened else batch) - finally: - writer.close() + with NamedTemporaryFile( + "wb", delete=False, dir=os.path.dirname(path) or ".", suffix=".parquet" + ) as tmp: + tmp_path = tmp.name + writer = pq.ParquetWriter(tmp_path, narrow_schema, **write_kwargs) + try: + for start in range(0, sorted_table.num_rows, step): + batch = sorted_table.slice(start, step) + writer.write_table(batch.cast(narrow_schema) if widened else batch) + finally: + writer.close() + os.replace(tmp_path, path) + except Exception: + if tmp_path and os.path.exists(tmp_path): + os.unlink(tmp_path) + raise return True @@ -263,7 +302,9 @@ def hilbert_sort_geodataframe(gdf, level: int = HILBERT_DEFAULT_LEVEL): "crs_total_bounds", "ensure_hilbert_sorted", "hilbert_distances_from_bounds", + "hilbert_keys_for_file", "hilbert_keys_for_table", "hilbert_reference_bounds", "hilbert_sort_geodataframe", + "is_hilbert_sorted", ] From 479c7a5c250e3c79bd6d0c13051dce398dcbc656 Mon Sep 17 00:00:00 2001 From: Matthias Mohr Date: Fri, 11 Sep 2026 18:56:41 +0200 Subject: [PATCH 3/3] Further code review --- tests/test_convert_duckdb.py | 26 ++++++++++++++++++++++++++ vecorel_cli/conversion/base.py | 5 +++-- vecorel_cli/conversion/duckdb.py | 22 +++++++++++++++++++++- 3 files changed, 50 insertions(+), 3 deletions(-) diff --git a/tests/test_convert_duckdb.py b/tests/test_convert_duckdb.py index 4cbaa7f..74e4969 100644 --- a/tests/test_convert_duckdb.py +++ b/tests/test_convert_duckdb.py @@ -3,6 +3,7 @@ import geopandas as gpd import numpy as np import pyarrow.parquet as pq +import pytest import shapely from vecorel_cli.conversion.base import BaseConverter @@ -111,6 +112,31 @@ def test_duckdb_converter_index_as_id(tmp_folder): assert sorted(result["id"]) == ["0", "1", "1", "2", "2", "3"] +def test_duckdb_converter_source_crs(tmp_folder): + src1 = _source_file(tmp_folder) + dest = tmp_folder / "converted.parquet" + + # the same CRS, declared as a differently rendered PROJJSON object + table = pq.read_table(src1) + metadata = dict(table.schema.metadata) + geo = json.loads(metadata[b"geo"]) + crs = geo["columns"]["geometry"]["crs"] + crs.pop("scope", None) + crs.pop("area", None) + crs["$schema"] = "https://proj.org/schemas/v0.5/projjson.schema.json" + metadata[b"geo"] = json.dumps(geo).encode() + src2 = str(tmp_folder / "source2.parquet") + pq.write_table(table.replace_schema_metadata(metadata), src2) + + Converter().convert(dest, input_files={src1: "a.parquet", src2: "b.parquet"}) + assert len(gpd.read_parquet(dest)) == 12 + + src3 = str(tmp_folder / "source3.parquet") + gpd.read_parquet(src1).to_crs("EPSG:3857").to_parquet(src3) + with pytest.raises(ValueError, match="different coordinate reference"): + Converter().convert(dest, input_files={src1: "a.parquet", src3: "c.parquet"}) + + def test_duckdb_converter_original_geometries(tmp_folder): src = _source_file(tmp_folder) dest = tmp_folder / "converted.parquet" diff --git a/vecorel_cli/conversion/base.py b/vecorel_cli/conversion/base.py index 9b2a703..0dcb565 100644 --- a/vecorel_cli/conversion/base.py +++ b/vecorel_cli/conversion/base.py @@ -62,8 +62,9 @@ class BaseConverter(LoggerMixin): index_as_id: bool = False - # Rows lacking a value for a schema-required property or a geometry are - # dropped up to this share of all rows, else they are kept with a warning + # Rows with null values in schema-required properties are dropped up to this + # share of all rows; above it the conversion fails. Rows with an empty or + # missing geometry are always dropped, regardless of this share. max_dropped_share: float = 0.01 def __init__(self, *args, **kwargs): diff --git a/vecorel_cli/conversion/duckdb.py b/vecorel_cli/conversion/duckdb.py index a30c87d..a58339c 100644 --- a/vecorel_cli/conversion/duckdb.py +++ b/vecorel_cli/conversion/duckdb.py @@ -22,6 +22,19 @@ def _sql_path(path) -> str: return f"'{escaped}'" +def _normalize_crs(crs): + """The comparable pyproj CRS for a GeoParquet crs value, + which defaults to OGC:CRS84 when missing.""" + from pyproj import CRS + + return CRS.from_user_input(crs if crs is not None else "OGC:CRS84") + + +def _equal_crs(a, b) -> bool: + # GeoParquet coordinates are always x, y regardless of the CRS axis order + return a.equals(b, ignore_axis_order=True) + + # This converter is experimental, use with caution. # Results may not be fully compliant yet. # Use this primarily for datasets that are too large to be processed by the default converter. @@ -99,6 +112,7 @@ def convert( # sources for the Hilbert grid and the output metadata. The sources are # combined without reprojection, so they must all use the same CRS. source_crs = None + reference = None for i, source in enumerate([sources] if isinstance(sources, str) else sources): crs = None row = con.execute( @@ -110,7 +124,8 @@ def convert( crs = source_geo.get("columns", {}).get(primary_column, {}).get("crs") if i == 0: source_crs = crs - elif crs != source_crs: + reference = _normalize_crs(crs) + elif not _equal_crs(_normalize_crs(crs), reference): raise ValueError( f"The sources use different coordinate reference systems: {source} " "differs from the first source. Reproject the sources to a common CRS." @@ -142,6 +157,11 @@ def convert( if self.column_additions: context = collection.get_collection_context() for key, value in self.column_additions.items(): + # constants override equally named source columns + if key in selected_targets: + keep = [i for i, t in enumerate(selected_targets) if t != key] + selections = [selections[i] for i in keep] + selected_targets = [selected_targets[i] for i in keep] if context.get(key) is False: selections.append(f'? as "{key}"') addition_params.append(value)