Repository navigation
feat: parquet generation function - #1845
Draft
davidgamez wants to merge 18 commits into
Draft
davidgamez wants to merge 18 commits into
davidgamez wants to merge 18 commits into
Conversation
Contributor
There was a problem hiding this comment.
Copilot review overview
🟡 Changes recommended
Concurrency, enqueue, purge, publication, and SQL-safety defects can leave builds duplicated, inaccessible, or incorrectly advertised.
Review effort: Balanced
Findings: 7
Open (10)
What changed in this PR
Adds on-demand GTFS-to-Parquet generation, status APIs, retention cleanup, deployment infrastructure, and local development tooling.
Changes:
- Implements Parquet conversion, publishing, progress tracking, and tests.
- Adds Operations API endpoints and scheduled retention cleanup.
- Configures Cloud Functions, task queues, storage CORS, and local tooling.
| File | Description |
|---|---|
.gitignore |
Ignores local Parquet output. |
README.md |
Documents local feed browsing. |
docs/OperationsAPI.yaml |
Defines Parquet API endpoints and models. |
scripts/parquet-generate-local.sh |
Bootstraps local generation. |
infra/batch/main.tf |
Enables HEAD CORS requests. |
infra/functions-python/main.tf |
Deploys builder, queue, purge, and volume. |
infra/functions-python/vars.tf |
Adds Parquet infrastructure variables. |
functions-python/helpers/ephemeral_workdir.py |
Generalizes work-directory ownership. |
functions-python/helpers/task_execution/task_execution_tracker.py |
Adds exclusive task claims and leases. |
functions-python/helpers/tests/test_task_execution_tracker.py |
Tests task claim behavior. |
functions-python/helpers/utils.py |
Enqueues Parquet build tasks. |
functions-python/operations_api/.openapi-generator/FILES |
Registers generated Parquet API files. |
functions-python/operations_api/README.md |
Documents local Parquet development. |
functions-python/operations_api/src/main.py |
Mounts the Parquet router. |
functions-python/operations_api/src/feeds_operations/impl/parquet_api_impl.py |
Implements status and generation endpoints. |
functions-python/operations_api/tests/feeds_operations/impl/test_parquet_api_impl.py |
Tests endpoint state handling. |
functions-python/parquet_builder/README.md |
Documents builder behavior and usage. |
functions-python/parquet_builder/function_config.json |
Configures the Cloud Function. |
functions-python/parquet_builder/requirements.txt |
Adds runtime dependencies. |
functions-python/parquet_builder/requirements_dev.txt |
Adds test dependencies. |
functions-python/parquet_builder/src/converter.py |
Converts GTFS tables to Parquet. |
functions-python/parquet_builder/src/main.py |
Orchestrates cloud builds and publishing. |
functions-python/parquet_builder/src/progress.py |
Throttles progress updates. |
functions-python/parquet_builder/src/scripts/generate_local.py |
Implements local generation and serving. |
functions-python/parquet_builder/tests/test_converter.py |
Tests conversion contracts. |
functions-python/parquet_builder/tests/test_generate_local.py |
Tests local generation and range serving. |
functions-python/parquet_builder/tests/test_main.py |
Tests build orchestration. |
functions-python/parquet_builder/tests/test_progress.py |
Tests progress throttling. |
functions-python/pmtiles_builder/src/main.py |
Uses the shared work-directory helper. |
functions-python/pmtiles_builder/tests/test_workdir.py |
Updates work-directory tests. |
functions-python/tasks_executor/src/main.py |
Registers the purge task. |
functions-python/tasks_executor/src/tasks/parquet/__init__.py |
Adds the Parquet task package. |
functions-python/tasks_executor/src/tasks/parquet/purge_expired_parquet.py |
Implements retention cleanup. |
functions-python/tasks_executor/tests/tasks/parquet/test_purge_expired_parquet.py |
Tests retention cleanup. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Comment on lines
+156
to
+159
| tracker.mark_triggered( | ||
| dataset.stable_id, | ||
| metadata={"phase": "start", "done": 0, "total": 0, "detail": ""}, | ||
| ) |
| return False | ||
| try: | ||
| con.execute(f""" | ||
| CREATE VIEW "{table}" AS |
Comment on lines
+219
to
+220
| tracker.heartbeat(dataset_stable_id, metadata=state) | ||
| db_session.commit() |
Comment on lines
+532
to
+537
| try: | ||
| blob.make_public() | ||
| except Exception as error: | ||
| # Uniform bucket-level access would make this unnecessary and impossible at the | ||
| # same time; the objects are public by bucket policy in that case. | ||
| logger.warning("Could not make %s public: %s", blob.name, error) |
| completed_at, | ||
| COALESCE((metadata ->> 'retention_days')::int, :default_days) AS retention_days | ||
| FROM task_execution_log | ||
| WHERE task_name = :task_name |
| purged = [] | ||
| for dataset_stable_id in datasets: | ||
| try: | ||
| deleted_objects += _delete_prefix(bucket, dataset_stable_id) |
Comment on lines
+1696
to
+1701
| triggers_replace = { | ||
| function_name = google_cloudfunctions2_function.parquet_builder.name | ||
| region = var.gcp_region | ||
| project = var.project_id | ||
| size = var.parquet_builder_in_memory_size | ||
| } |
| project_id=project_id, | ||
| gcp_region=gcp_region, | ||
| queue_name=queue_name, | ||
| task_name=None if force else task_name, |
Comment on lines
+275
to
+281
| except Exception as error: | ||
| logger.exception("Parquet build failed for %s", dataset_stable_id) | ||
| # Releases the claim as well as recording why, so the dataset can be retried | ||
| # without waiting out the lease. | ||
| tracker.mark_failed(dataset_stable_id, error_message=str(error)) | ||
| db_session.commit() | ||
| raise |
| ```json | ||
| { "status": "absent", "feed_stable_id": "mdb-1210", "dataset_stable_id": "mdb-1210-202402121801" } | ||
| { "status": "preparing", "phase": "convert", "done": 12, "total": 32, "detail": "stop_times" } | ||
| { "status": "ready", "base_url": "https://.../parquet", "tables": [{"name": "stops"}] } |
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary:
Renders GTFS datasets as Parquet so clients can browse a feed without downloading the archive.
A new
parquet-builderCloud Function streams a dataset's archive from GCS, converts each table with DuckDB one at a time, and publishes{table}.parquetplus amanifest.jsonto the public bucket. Two Operations API endpoints drive it (/gtfs_feeds/{id}/parquet,/gtfs_datasets/{id}/parquet): POST to start, GET to poll. Files expire through the bucket's owndaysSinceCustomTimelifecycle rule rather than a purge job.The same source is deployed three times, with its own memory, in-memory volume and Cloud Tasks queue per size:
RLIMIT_ASsmlThe volume is carved out of the memory, not added to it, so the process budget is
memory - volume - 200 MiB. Routing is by the largest single uncompressed file because the volume holds one at a time; totals are the wrong signal. The bands come from the catalogue: of 4277 feeds the median archive is 0.2 MB and only 31 exceed 100 MB, while about ten have a member over 1 GB and the worst has one of 4.8 GB.Routing self-corrects in both directions through a single per-feed
config_value_feedoverride:{"size":"l","locked":true}stops both directions. A feed locked below what it needs fails and stays failed, whichcurrent_size_lockedon the attempts endpoint makes visible.task_execution_attemptrecords every attempt append-only — which worker ran, why, what failed, peak memory and the budget it ran against — exposed asGET /gtfs_feeds/{id}/execution_attempts(filterable by status, failure_kind, date) and aggregated by a newtask_execution_statstask.Two measurement bugs fixed along the way:
peak_vms_byteswas sampled after DuckDB had been torn down, so it recorded the quiet moment after a build rather than the busy one during it (now polled on a sampler thread); andlimit_gcp_memoryonly logged the budget it computed, so an attempt could not say whether its usage was comfortable (now returned and recorded).Expected behavior:
POST to either parquet endpoint returns
202with the current state; the GET reportsabsent→preparing(with progress) →ready, then servesbase_url, from whichmanifest.jsonand each table are fetchable by HTTP range request. A feed with no dataset yet returns404.A dataset too large for the worker it was routed to fails, moves up a rung and succeeds without intervention, with the whole trail visible in
execution_attempts. Measured on dev:s3-21s,m13-69s,l3-133s, each peaking at 67-81% of its address-space budget.Testing tips:
./scripts/api-tests.sh --folder functions-python/parquet_builder(andhelpers,operations_api).scripts/parquet-generate-local.sh./v1/operations/gtfs_datasets/{dataset_id}/parquetwith{"force": true}, poll the GET, then open{base_url}/manifest.jsonin a browser to confirm CORS and range reads.mdb-2014(4.07 GiBstop_times.txt) is the heaviest case; a small feed such asntd-70242exercisess.GET /v1/operations/gtfs_feeds/{id}/execution_attemptsshows routing, escalation and the current override per feed.functions-python/parquet_builder/README.mdand rebuild; add"locked": trueto the same value to confirm it then refuses to escalate.Please make sure these boxes are checked before submitting your pull request - thanks!
./scripts/api-tests.shto make sure you didn't break anything🤖 Generated with Claude Code