PIPELINE-4277: Add GCS Parquet compaction pipeline

Dockerfile

@@ -15,7 +15,7 @@ ENV UV_COMPILE_BYTECODE=1
 COPY pyproject.toml requirements.txt README.md MANIFEST.in ./
 RUN uv pip install --system --upgrade pip && \
     uv pip install --system build && \
-    uv pip install --system --prefix=/install --prerelease=allow -r requirements.txt
+    uv pip install --system --prefix=/install -r requirements.txt
 
 COPY src ./src
 RUN uv pip install --system --prefix=/install --no-deps .
@@ -48,7 +48,7 @@ RUN apt-get update && \
 WORKDIR /opt/project
 
 COPY . .
-RUN uv pip install --system --prerelease=allow -e .[lint,dev,build] && \
+RUN uv pip install --system -e .[lint,dev,build] && \
     uv pip install --system -r requirements-test.txt
 
 # ---------------------------------------------------------------------------------------

Makefile

@@ -18,7 +18,7 @@ UV_VERSION := 0.10.9
 
 VENV:=uv venv
 PIP:=uv pip
-PIP_COMPILE:=uv pip compile --prerelease=allow
+PIP_COMPILE:=uv pip compile
 
 
 
@@ -81,7 +81,7 @@ install-test: upgrade-pip
 
 .PHONY: install  ## Install the package in editable mode & all dependencies for local development.
 install: upgrade-pip
-	${PIP} install --prerelease=allow -e .[lint,dev,build]
+	${PIP} install -e .[lint,dev,build]
 	make install-test
 
 .PHONY: test  ## Run all unit tests exporting coverage.xml report.

config/benchmark-parquet/scratch/nmea-parsed-1-day.yaml

@@ -0,0 +1,22 @@
+project: world-fishing-827
+start_date: "2023-02-07"
+end_date: "2023-02-08"
+queries:
+  - aggregation
+agg_column: tagblock_timestamp
+runs: 3
+location: US
+max_gb_billed: 100.0
+tables:
+  native:
+    table: world-fishing-827.scratch_tomas_ttl30d.pipe_nmea_parsed_2023_feb
+    column: tagblock_timestamp
+    type: native
+  external_uncompacted:
+    table: gfw-int-ais-staging-datalake.vessel_transmissions_v1.messages__PR_2627__feature_bq_to_parquet_dag
+    column: event_date
+    type: external
+  external_compacted:
+    table: gfw-int-ais-staging-datalake.vessel_transmissions_v1._messages_compacted__PR_2627__feature_bq_to_parquet_dag
+    column: event_date
+    type: external

config/benchmark-parquet/scratch/nmea-parsed-1-month.yaml

@@ -0,0 +1,30 @@
+project: world-fishing-827
+start_date: "2023-02-01"
+end_date: "2023-03-01"
+queries:
+  - aggregation
+agg_column: tagblock_timestamp
+runs: 3
+location: US
+max_gb_billed: 2000.0
+tables:
+  native:
+    table: world-fishing-827.scratch_tomas_ttl30d.pipe_nmea_parsed_2023_feb
+    column: tagblock_timestamp
+    type: native
+  external_uncompacted:
+    table: gfw-int-ais-staging-datalake.vessel_transmissions_v1.messages__PR_2627__feature_bq_to_parquet_dag
+    column: event_date
+    type: external
+  external_compacted:
+    table: gfw-int-ais-staging-datalake.vessel_transmissions_v1._messages_compacted__PR_2627__feature_bq_to_parquet_dag
+    column: event_date
+    type: external
+  external_uncompacted_cached:
+    table: gfw-int-ais-staging-datalake.vessel_transmissions_v1.messages__PR_2627__feature_bq_to_parquet_dag_cached
+    column: event_date
+    type: external
+  external_compacted_cached:
+    table: gfw-int-ais-staging-datalake.vessel_transmissions_v1._messages_compacted__PR_2627__feature_bq_to_parquet_dag_cached
+    column: event_date
+    type: external

config/compact-parquet/scratch/nmea-decoded-1-day.yaml

@@ -0,0 +1,10 @@
+gcs_output_path: gs://engineering-development-ttl30/tomas.link/parquet-migration/messages/
+gcs_staging_path: gs://engineering-development-ttl30/tomas.link/parquet-migration/_compacted_messages/
+event_source: wf827-pipe-nmea-parsed
+start_date: "2023-01-01"
+end_date: "2023-01-02"
+project: world-fishing-827
+partition_prefix: event_
+target_file_size_mb: 512
+# staging_path defaults to gs://engineering-development-ttl30/tomas.link/parquet-migration/_compact_messages_staging
+dry_run: false

pyproject.toml

@@ -31,12 +31,15 @@ classifiers = [
 requires-python = ">= 3.12"
 dependencies = [
     "cloudpathlib[gs]~=0.24",
+    "duckdb~=1.3",
+    "gcsfs~=2025.5",
     "gfw-common[bq]~=0.14",
     "google-cloud-bigquery~=3.0",
     "google-cloud-storage~=2.0",
     "pyyaml~=6.0",
     "rich~=14.0",
     "rich-argparse~=1.6",
+    "tenacity~=9.0",
 ]
 
 [project.urls]
@@ -148,6 +151,12 @@ addopts = "-v --cov=src --cov=init_project --cov-report=term-missing"
 branch = true
 parallel = true
 context = "${CONTEXT}"
+omit = [
+    "src/gfw/ops/pipelines/benchmark_parquet/main.py",  # Analysis, not used in production.
+    "src/gfw/ops/cli/commands/benchmark_parquet.py",    # Analysis, not used in production.
+    "**/__init__.py",
+
+]
 
 [tool.coverage.report]
 precision = 0

requirements.txt

@@ -1,31 +1,55 @@
 # This file was autogenerated by uv via the following command:
-#    uv pip compile --prerelease=allow -o requirements.txt pyproject.toml
+#    uv pip compile -o requirements.txt pyproject.toml
+aiohappyeyeballs==2.7.1
+    # via aiohttp
+aiohttp==3.14.1
+    # via gcsfs
+aiosignal==1.4.0
+    # via aiohttp
+attrs==26.1.0
+    # via aiohttp
 certifi==2026.6.17
     # via requests
-cffi==2.0.0
+cffi==2.1.0
     # via cryptography
-charset-normalizer==3.4.7
+charset-normalizer==3.4.9
     # via requests
 cloudpathlib==0.24.0
     # via gfw-ops (pyproject.toml)
-cryptography==47.0.0
+cryptography==49.0.0
+    # via google-auth
+decorator==5.3.1
+    # via gcsfs
+duckdb==1.5.4
+    # via gfw-ops (pyproject.toml)
+frozenlist==1.8.0
     # via
-    #   google-auth
-    #   pyopenssl
-gfw-common==0.14.0
+    #   aiohttp
+    #   aiosignal
+fsspec==2025.12.0
+    # via gcsfs
+gcsfs==2025.12.0
+    # via gfw-ops (pyproject.toml)
+gfw-common==0.14.1
     # via gfw-ops (pyproject.toml)
 google-api-core==2.31.0
     # via
     #   google-cloud-bigquery
     #   google-cloud-core
     #   google-cloud-storage
-google-auth==2.55.0
+    #   google-cloud-storage-control
+google-auth==2.55.2
     # via
+    #   gcsfs
     #   google-api-core
+    #   google-auth-oauthlib
     #   google-cloud-bigquery
     #   google-cloud-core
     #   google-cloud-storage
-google-cloud-bigquery==3.42.0
+    #   google-cloud-storage-control
+google-auth-oauthlib==1.4.0
+    # via gcsfs
+google-cloud-bigquery==3.42.2
     # via
     #   gfw-ops (pyproject.toml)
     #   gfw-common
@@ -37,6 +61,9 @@ google-cloud-storage==2.19.0
     # via
     #   gfw-ops (pyproject.toml)
     #   cloudpathlib
+    #   gcsfs
+google-cloud-storage-control==1.12.0
+    # via gcsfs
 google-crc32c==1.8.0
     # via
     #   google-cloud-storage
@@ -48,15 +75,23 @@ google-resumable-media==2.10.0
 googleapis-common-protos==1.75.0
     # via
     #   google-api-core
+    #   grpc-google-iam-v1
     #   grpcio-status
-grpcio==1.81.1
+grpc-google-iam-v1==0.14.4
+    # via google-cloud-storage-control
+grpcio==1.82.1
     # via
     #   google-api-core
+    #   google-cloud-storage-control
+    #   googleapis-common-protos
+    #   grpc-google-iam-v1
     #   grpcio-status
-grpcio-status==1.81.1
+grpcio-status==1.82.1
     # via google-api-core
 idna==3.18
-    # via requests
+    # via
+    #   requests
+    #   yarl
 jinja2==3.1.6
     # via gfw-common
 markdown-it-py==4.2.0
@@ -65,28 +100,40 @@ markupsafe==3.0.3
     # via jinja2
 mdurl==0.1.2
     # via markdown-it-py
+multidict==6.7.1
+    # via
+    #   aiohttp
+    #   yarl
+oauthlib==3.3.1
+    # via requests-oauthlib
 packaging==26.2
     # via google-cloud-bigquery
-proto-plus==1.28.0
-    # via google-api-core
-protobuf==6.33.6
+propcache==0.5.2
+    # via
+    #   aiohttp
+    #   yarl
+proto-plus==1.28.1
+    # via
+    #   google-api-core
+    #   google-cloud-storage-control
+protobuf==7.35.1
     # via
     #   google-api-core
+    #   google-cloud-storage-control
     #   googleapis-common-protos
+    #   grpc-google-iam-v1
     #   grpcio-status
     #   proto-plus
 pyarrow==18.1.0
     # via gfw-common
-pyasn1==0.6.3
+pyasn1==0.6.4
     # via pyasn1-modules
 pyasn1-modules==0.4.2
     # via google-auth
 pycparser==3.0
     # via cffi
 pygments==2.20.0
     # via rich
-pyopenssl==26.2.0
-    # via google-auth
 python-dateutil==2.9.0.post0
     # via
     #   gfw-common
@@ -97,9 +144,13 @@ pyyaml==6.0.3
     #   gfw-common
 requests==2.34.2
     # via
+    #   gcsfs
     #   google-api-core
     #   google-cloud-bigquery
     #   google-cloud-storage
+    #   requests-oauthlib
+requests-oauthlib==2.0.0
+    # via google-auth-oauthlib
 rich==14.3.4
     # via
     #   gfw-ops (pyproject.toml)
@@ -111,9 +162,14 @@ six==1.17.0
     # via python-dateutil
 sqlparse==0.5.0
     # via gfw-common
-typing-extensions==4.15.0
+tenacity==9.1.4
+    # via gfw-ops (pyproject.toml)
+typing-extensions==4.16.0
     # via
+    #   aiohttp
+    #   aiosignal
     #   grpcio
-    #   pyopenssl
 urllib3==2.7.0
     # via requests
+yarl==1.24.2
+    # via aiohttp

src/gfw/ops/cli/commands/benchmark_parquet.py

@@ -0,0 +1,70 @@
+"""CLI command for benchmarking BigQuery table query performance."""
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any
+
+from gfw.common.cli import Command, Option
+from gfw.common.cli.actions import NestedKeyValueAction
+
+from gfw.ops.pipelines import benchmark_parquet as pipeline
+
+
+_DESCRIPTION = "Benchmark SELECT * query performance across one or more BigQuery tables."
+
+HELP_PROJECT = "GCP project for billing."
+HELP_START_DATE = "Start of date range, inclusive (YYYY-MM-DD)."
+HELP_END_DATE = "End of date range, exclusive (YYYY-MM-DD)."
+HELP_TABLES = (
+    "Table configurations as dotted key=value pairs. "
+    "Each table is keyed by its display label. "
+    "Required fields: table, column. "
+    "Optional: type (native|external, default: external). "
+    "Example: --tables native.table=project.dataset.t native.column=ts native.type=native "
+    "external.table=project.dataset.ext external.column=event_date"
+)
+HELP_QUERIES = (
+    "Which queries to run. Valid values: select_star, aggregation. "
+    "Defaults to all. Example: --queries aggregation"
+)
+HELP_AGG_COLUMN = (
+    "Timestamp column name to aggregate by hour, applied to all tables. "
+    "Example: --agg-column tagblock_timestamp. "
+    "If omitted, the aggregation query is skipped."
+)
+HELP_RUNS = "Number of times each query is repeated. Results are averaged."
+HELP_LOCATION = "BigQuery job location."
+HELP_MAX_GB = "Safety cap on bytes billed per query in GB. Pass 0 to disable."
+
+
+class BenchmarkParquet(Command):
+    """Benchmark SELECT * query performance across one or more BigQuery tables."""
+
+    @property
+    def name(self) -> str:
+        """Command name."""
+        return "benchmark-parquet"
+
+    @property
+    def description(self) -> str:
+        """Command description."""
+        return _DESCRIPTION
+
+    @property
+    def options(self) -> list[Option]:
+        """Command options."""
+        return [
+            Option("--project", type=str, required=True, help=HELP_PROJECT),
+            Option("--start-date", type=str, required=True, help=HELP_START_DATE),
+            Option("--end-date", type=str, required=True, help=HELP_END_DATE),
+            Option("--tables", type=str, nargs="+", action=NestedKeyValueAction, help=HELP_TABLES),
+            Option("--queries", type=str, nargs="+", default=None, help=HELP_QUERIES),
+            Option("--agg-column", type=str, default=None, help=HELP_AGG_COLUMN),
+            Option("--runs", type=int, default=1, help=HELP_RUNS),
+            Option("--location", type=str, default="US", help=HELP_LOCATION),
+            Option("--max-gb-billed", type=float, default=100.0, help=HELP_MAX_GB),
+        ]
+
+    def run(self, config: SimpleNamespace, **kwargs: Any) -> None:
+        """Run the benchmark."""
+        pipeline.run(**vars(config), **kwargs)