refactor(parquet): use parquet and duckdb instead of postgresql

README.md

@@ -5,7 +5,7 @@
 [![CircleCI](https://circleci.com/gh/datagouv/api-tabular.svg?style=svg)](https://app.circleci.com/pipelines/github/datagouv/api-tabular)
 [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
 
-An API service that provides RESTful access to CSV or tabular data converted by [Hydra](https://github.com/datagouv/hydra). This service provides a REST API to access PostgreSQL database tables containing CSV data, offering HTTP querying capabilities, pagination, and data streaming for CSV or tabular resources.
+An API service that provides RESTful access to CSV or tabular data converted by [Hydra](https://github.com/datagouv/hydra). The **Tabular API** queries public Parquet files with [DuckDB](https://duckdb.org/) (filters, pagination, aggregations, CSV/JSON export), while resource metadata still comes from PostgreSQL via PostgREST (`tables_index`). The **Metrics API** continues to use PostgREST against PostgreSQL.
 
 This service is mainly used, developed and maintained by [data.gouv.fr](https://data.gouv.fr) - the France Open Data platform.
 The production API is deployed on data.gouv.fr infrastructure at [`https://tabular-api.data.gouv.fr/api`](https://tabular-api.data.gouv.fr/api). See the [product documentation](https://www.data.gouv.fr/dataservices/api-tabulaire-data-gouv-fr-beta/) (in French) for usage details and the [technical documentation](https://tabular-api.data.gouv.fr/api/doc) for API reference.
@@ -74,9 +74,11 @@ The production API is deployed on data.gouv.fr infrastructure at [`https://tabul
    Query the API using a `resource_id`. Several test resources are available in the fake database:
 
    - **`aaaaaaaa-1111-bbbb-2222-cccccccccccc`** - Main test resource with 1000 rows
-   - **`aaaaaaaa-5555-bbbb-6666-cccccccccccc`** - Resource with database indexes
-   - **`dddddddd-7777-eeee-8888-ffffffffffff`** - Resource allowed for aggregation
-   - **`aaaaaaaa-9999-bbbb-1010-cccccccccccc`** - Resource with indexes and aggregation allowed
+   - **`aaaaaaaa-5555-bbbb-6666-cccccccccccc`** - Smaller sample resource
+   - **`dddddddd-7777-eeee-8888-ffffffffffff`** - Smaller sample resource
+   - **`aaaaaaaa-9999-bbbb-1010-cccccccccccc`** - Smaller sample resource
+
+   Local Parquet fixtures live under `db/parquet/{resource_id}.parquet`. Set `PARQUET_BASE_URL` to that directory when running the Tabular API against the test stack (the pytest suite does this automatically).
 
 ### 🏭 Run with a real Hydra database
 
@@ -320,9 +322,8 @@ column_name__sort=desc
 ```
 
 #### Aggregation Operators
-> ⚠️ **WARNING**: Aggregation requests are disabled by default.
-> You can allow or disallow it for all resources with the `ALLOW_AGGREGATION` config. If `ALLOW_AGGREGATION` is set to `false`, you can specify allow exceptions by listing these in the `ALLOW_AGGREGATION_EXCEPTIONS` list.
-> You can get the current status and exceptions at `/api/aggregation-exceptions/` endpoint.
+
+Aggregation is available on all resources and columns (except JSON limitations below).
 
 ```
 # group by values
@@ -486,30 +487,34 @@ Configuration is handled through TOML files and environment variables. The defau
 
 | Option | Default | Description |
 |--------|---------|-------------|
-| `PGREST_ENDPOINT` | `http://localhost:8080` | PostgREST server URL |
+| `PGREST_ENDPOINT` | `http://localhost:8080` | PostgREST server URL (metadata for Tabular API; data for Metrics API) |
+| `S3_ENDPOINT` | `""` | S3 endpoint hosting public Parquet files (set via env in deploy) |
+| `S3_BUCKET` | `""` | S3 bucket name (set via env in deploy) |
+| `PARQUET_BASE_URL` | `""` | Optional override for Parquet base URL/path (tests use a local directory). When empty, URLs are `https://{S3_ENDPOINT}/{S3_BUCKET}/parquet/{resource_id}.parquet` |
 | `SERVER_NAME` | `localhost:8005` | Server name for URL generation |
 | `SCHEME` | `http` | URL scheme (http/https) |
 | `SENTRY_DSN` | `None` | Sentry DSN for error reporting (optional) |
 | `PAGE_SIZE_DEFAULT` | `20` | Default page size |
 | `PAGE_SIZE_MAX` | `50` | Maximum allowed page size |
-| `BATCH_SIZE` | `50000` | Batch size for streaming |
+| `BATCH_SIZE` | `50000` | Max rows for CSV/JSON export |
 | `DOC_PATH` | `/api/doc` | Swagger documentation path |
-| `ALLOW_AGGREGATION` | `False` | Whether aggregation queries are allowed (`column__groupby`). If False, it can still be explicitly allowed with `ALLOW_AGGREGATION_EXCEPTIONS` |
-| `ALLOW_AGGREGATION_EXCEPTIONS` | `["dddddddd-7777-eeee-8888-ffffffffffff", "aaaaaaaa-9999-bbbb-1010-cccccccccccc"]` | List of resource IDs allowed for aggregation |
+
+> Disk caching of Parquet files is not implemented yet; DuckDB reads remote (or local) Parquet via `httpfs` / filesystem. A local cache may be added later if latency requires it.
 
 ### Environment Variables
 
 You can override any configuration value using environment variables:
 
 ```shell
 export PGREST_ENDPOINT="http://my-postgrest:8080"
+export S3_ENDPOINT="s3.example.com"
+export S3_BUCKET="my-bucket"
 export PAGE_SIZE_DEFAULT=50
 export SENTRY_DSN="https://your-sentry-dsn"
 ```
-Once the containers are up and running, you can directly query PostgREST on:
+
+PostgREST remains available for metadata inspection and for the Metrics API:
 `<PGREST_ENDPOINT>/<table_name>?<filters>`
-like for example:
-`http://localhost:8080/eb7a008177131590c2f1a2ca0?decompte=eq.10`
 
 ### Custom Configuration File
 

api_tabular/__init__.py

@@ -55,6 +55,21 @@ def check(self):
         """Sanity check on config"""
         pass
 
+    def parquet_url(self, resource_id: str) -> str:
+        """Public Parquet URL (or local path override) for a resource."""
+        assert self.configuration is not None
+        base = (self.configuration.get("PARQUET_BASE_URL") or "").rstrip("/")
+        if not base:
+            endpoint = (self.configuration.get("S3_ENDPOINT") or "").strip()
+            bucket = (self.configuration.get("S3_BUCKET") or "").strip()
+            if not endpoint or not bucket:
+                raise ValueError(
+                    "Parquet location is not configured: set PARQUET_BASE_URL, "
+                    "or both S3_ENDPOINT and S3_BUCKET"
+                )
+            base = f"https://{endpoint}/{bucket}/parquet"
+        return f"{base}/{resource_id}.parquet"
+
     def __getattr__(self, __name):
         assert self.configuration is not None
         return self.configuration.get(__name)

api_tabular/config_default.toml

@@ -1,4 +1,9 @@
 PGREST_ENDPOINT = "http://localhost:8080"
+# Set via environment (no defaults in repo). Used when PARQUET_BASE_URL is empty.
+S3_ENDPOINT = ""
+S3_BUCKET = ""
+# Optional override for tests / local paths. When empty, URLs are built from S3_*.
+PARQUET_BASE_URL = ""
 SERVER_NAME = "localhost:8005"
 SCHEME = "http"
 SENTRY_DSN = ""
@@ -7,8 +12,3 @@ PAGE_SIZE_DEFAULT = 20
 PAGE_SIZE_MAX = 50
 BATCH_SIZE = 50000
 DOC_PATH = "/api/doc"
-ALLOW_AGGREGATION = false
-ALLOW_AGGREGATION_EXCEPTIONS = [
-    "dddddddd-7777-eeee-8888-ffffffffffff",  # without indexes
-    "aaaaaaaa-9999-bbbb-1010-cccccccccccc",  # with indexes
-]  # list of resource_ids

api_tabular/core/query.py

@@ -1,8 +1,6 @@
 import re
 from collections import defaultdict
 
-from api_tabular.core.utils import is_aggregation_allowed
-
 
 def build_sql_query_string(
     request_arg: list,
@@ -11,6 +9,7 @@ def build_sql_query_string(
     page_size: int | None = None,
     offset: int = 0,
 ) -> str:
+    """Build a PostgREST query string (used by the Metrics API)."""
     sql_query = []
     aggregators = defaultdict(list)
     sorted = False
@@ -39,11 +38,6 @@ def build_sql_query_string(
         else:
             raise ValueError(f"argument '{arg}' could not be parsed")
     if aggregators:
-        if resource_id and not is_aggregation_allowed(resource_id):
-            raise PermissionError(
-                f"Aggregation parameters `{'`, `'.join(aggregators.keys())}` "
-                f"are not allowed for resource '{resource_id}'"
-            )
         agg_query = "select="
         for operator in aggregators:
             if operator == "groupby":

api_tabular/core/swagger.py

@@ -2,8 +2,6 @@
 
 import yaml
 
-from api_tabular.core.utils import is_aggregation_allowed
-
 TYPE_POSSIBILITIES = {
     "string": [
         "isnull",
@@ -200,10 +198,6 @@ def swagger_parameters(resource_columns: dict, resource_id: str) -> list:
     # see cast for db here: https://github.com/datagouv/csv-detective/blob/master/csv_detective/output/dataframe.py
     for key, value in resource_columns.items():
         for op in OPERATORS_DESCRIPTIONS:
-            if not is_aggregation_allowed(resource_id) and OPERATORS_DESCRIPTIONS[op].get(
-                "is_aggregator"
-            ):
-                continue
             if op in TYPE_POSSIBILITIES[value["python_type"]]:
                 op_name = cast(str, OPERATORS_DESCRIPTIONS[op]["name"])
                 op_description = cast(str, OPERATORS_DESCRIPTIONS[op]["description"])

api_tabular/core/utils.py

@@ -5,10 +5,6 @@
 from api_tabular.core.error import QueryException
 
 
-def is_aggregation_allowed(resource_id: str) -> bool:
-    return config.ALLOW_AGGREGATION or resource_id in config.ALLOW_AGGREGATION_EXCEPTIONS
-
-
 def process_total(res: Response | ClientResponse) -> int:
     # the Content-Range looks like this: '0-49/21777'
     # see https://docs.postgrest.org/en/stable/references/api/pagination_count.html

api_tabular/tabular/app.py

@@ -15,7 +15,6 @@
 from api_tabular.core.utils import build_offset
 from api_tabular.core.version import get_app_version
 from api_tabular.tabular.utils import (
-    get_potential_indexes,
     get_resource,
     get_resource_data,
     stream_resource_data,
@@ -62,8 +61,6 @@ async def resource_profile(request):
     resource: dict = await get_resource(
         request.app["csession"], resource_id, ["profile:csv_detective"]
     )
-    indexes: set | None = await get_potential_indexes(request.app["csession"], resource_id)
-    resource["indexes"] = list(indexes) if isinstance(indexes, set) else None
     return web.json_response(resource)
 
 
@@ -73,10 +70,7 @@ async def resource_swagger(request):
     resource: dict = await get_resource(
         request.app["csession"], resource_id, ["profile:csv_detective"]
     )
-    indexes: set | None = await get_potential_indexes(request.app["csession"], resource_id)
     columns: dict[str, str] = resource["profile"]["columns"]
-    if indexes:
-        columns = {col: params for col, params in columns.items() if col in indexes}
     swagger_string = build_swagger_file(columns, resource_id)
     return web.Response(body=swagger_string)
 
@@ -102,9 +96,9 @@ async def resource_data(request):
 
     offset = build_offset(page, page_size)
 
-    sql_query = await try_build_query(request, query_string, resource_id, page_size, offset)
-    resource = await get_resource(request.app["csession"], resource_id, ["parsing_table"])
-    response, total = await get_resource_data(request.app["csession"], resource, sql_query)
+    query = await try_build_query(request, query_string, resource_id, page_size, offset)
+    await get_resource(request.app["csession"], resource_id, [])
+    response, total = await get_resource_data(resource_id, query)
 
     next = build_link_with_page(request, query_string, page + 1, page_size)
     prev = build_link_with_page(request, query_string, page - 1, page_size)
@@ -146,16 +140,6 @@ async def get_health(request):
     return await check_health(request, f"{config.PGREST_ENDPOINT}/migrations_csv")
 
 
-@routes.get(r"/api/aggregation-exceptions/")
-async def get_aggregation_exceptions(request):
-    """Return the list of resources for which aggregation queries are allowed"""
-    body = {
-        "allowed": config.ALLOW_AGGREGATION,
-        "exceptions": config.ALLOW_AGGREGATION_EXCEPTIONS,
-    }
-    return web.json_response(body)
-
-
 async def app_factory():
     async def on_startup(app):
         app["csession"] = ClientSession()