Avoid writing empty placeholder files for date partitions with no data

src/gfw/ops/pipelines/bq_to_parquet/__init__.py

@@ -1,3 +1,4 @@
 from .main import run
 
+
 __all__ = ["run"]

src/gfw/ops/pipelines/bq_to_parquet/destination.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import datetime
+
 from dataclasses import dataclass, field
 
 from cloudpathlib import GSPath

src/gfw/ops/pipelines/bq_to_parquet/job.py

@@ -3,10 +3,10 @@
 
 import datetime
 import logging
+
 from dataclasses import dataclass, field
-from enum import auto, StrEnum
+from enum import StrEnum, auto
 
-from google.api_core.exceptions import NotFound
 from google.cloud import bigquery
 
 
@@ -17,7 +17,6 @@ class ExportStatus(StrEnum):
     """Outcome of a single BQ extract job."""
 
     SUCCESS = auto()
-    NOT_FOUND = auto()
     FAILED = auto()
 
 
@@ -35,9 +34,6 @@ def wait(self) -> ExportJob:
             self.job.result()
             logger.info(f"Completed {self.job.job_id}")
             self.status = ExportStatus.SUCCESS
-        except NotFound:
-            logger.warning(f"Skipped {self.date}: not found")
-            self.status = ExportStatus.NOT_FOUND
         except Exception:
             logger.exception(f"Failed {self.job.job_id} for {self.date}")
             self.status = ExportStatus.FAILED
@@ -61,11 +57,6 @@ def succeeded(self) -> list[ExportJob]:
         """Jobs that completed successfully."""
         return self.filter(ExportStatus.SUCCESS)
 
-    @property
-    def skipped(self) -> list[ExportJob]:
-        """Jobs skipped because the source table or shard was not found."""
-        return self.filter(ExportStatus.NOT_FOUND)
-
     def filter(self, status: ExportStatus) -> list[ExportJob]:
         """Return jobs matching the given status."""
         return [job for job in self.jobs if job.status == status]

src/gfw/ops/pipelines/bq_to_parquet/main.py

@@ -3,6 +3,7 @@
 
 import datetime
 import logging
+
 from dataclasses import dataclass
 from typing import Any, Callable
 
@@ -58,7 +59,34 @@ def submit_job(self, date: datetime.date) -> ExportJob:
 
     def remaining_dates(self, dates: list[datetime.date]) -> list[datetime.date]:
         exported = self.destination.existing_dates(dates)
-        return [d for d in dates if d not in exported]
+        if exported:
+            logger.info(f"{len(exported)} date(s) already exported, skipping")
+            logger.debug(f"Already exported: {sorted(exported)}")
+
+        candidates = [d for d in dates if d not in exported]
+        if not candidates:
+            logger.info("No remaining dates to export")
+            return []
+
+        with_data = self.dates_with_data(candidates)
+        for date in candidates:
+            if date not in with_data:
+                logger.info(f"Skipping {date}: no data in source")
+
+        return [d for d in candidates if d in with_data]
+
+    def dates_with_data(self, dates: list[datetime.date]) -> set[datetime.date]:
+        """Return the subset of dates that have data in the source.
+
+        Runs the query the source describes using this exporter's own BQ
+        client; the source itself performs no I/O. Assumes dates is non-empty.
+        """
+        query = self.source.dates_with_data_query(dates)
+        job_config = bigquery.QueryJobConfig(query_parameters=query.parameters)
+        rows = self.bq_client.query(query.sql, job_config=job_config).result()
+        ids = {row.date_id for row in rows if row.date_id is not None}
+
+        return {d for d in dates if d.strftime("%Y%m%d") in ids}
 
 
 def run(
@@ -105,7 +133,8 @@ def run(
 
         sharded:
             Set to ``True`` for date-sharded tables (``table_YYYYMMDD``). Each shard is
-            addressed directly. Missing shards are skipped with a warning.
+            addressed directly. Dates with no matching shard are skipped before
+            submitting any job (see Exporter.dates_with_data()).
 
         partition_prefix:
             Prefix applied to partition key names in the hive output path.
@@ -162,6 +191,4 @@ def run(
             "Retrying this task will skip already-exported dates and resume from the failures."
         )
 
-    logger.info(
-        f"Done: {len(results.succeeded)} exported, {len(results.skipped)} skipped (not found)"
-    )
+    logger.info(f"Done: {len(results.succeeded)} exported")

src/gfw/ops/pipelines/bq_to_parquet/source.py

@@ -2,8 +2,23 @@
 from __future__ import annotations
 
 import datetime
+
 from dataclasses import dataclass
-from typing import ClassVar
+from typing import ClassVar, NamedTuple
+
+from google.cloud import bigquery
+
+
+class DatesWithDataQuery(NamedTuple):
+    """A metadata-only query whose rows identify dates that have data.
+
+    Executing ``sql`` (with ``parameters`` bound) is expected to return one row
+    per date that has data, each exposing that date as ``YYYYMMDD`` under a
+    ``date_id`` column.
+    """
+
+    sql: str
+    parameters: list[bigquery.ScalarQueryParameter]
 
 
 @dataclass(frozen=True)
@@ -25,16 +40,79 @@ def ref(self, date: datetime.date) -> str:
         """Return the BQ table reference for the given date."""
         return f"{self.table}{self.separator}{date.strftime('%Y%m%d')}"
 
+    def dates_with_data_query(self, dates: list[datetime.date]) -> DatesWithDataQuery:
+        """Return a query to find which of ``dates`` have data.
+
+        This describes the check as data (SQL + parameters) without touching
+        BigQuery -- callers (e.g. :class:`~.main.Exporter`, which owns the BQ
+        client) are responsible for running it.
+
+        Subclasses must implement this, same as ``separator``: there's no
+        safe generic default, since silently skipping the check would mean
+        every date is assumed to have data.
+        """
+        raise NotImplementedError
+
 
 @dataclass(frozen=True)
 class PartitionedSource(Source):
     """A time-partitioned BigQuery table. Addresses each day as ``table$YYYYMMDD``."""
 
     separator: ClassVar[str] = "$"
 
+    def dates_with_data_query(self, dates: list[datetime.date]) -> DatesWithDataQuery:
+        """Return a query for dates whose partition has at least one row.
+
+        Extracting an empty or non-existent partition via the ``table$YYYYMMDD``
+        decorator does not raise an error: BigQuery returns zero rows and still
+        writes a schema-only output file. Checking partition metadata up front
+        avoids submitting jobs -- and writing placeholder files -- for dates
+        with no data.
+        """
+        project, dataset, table_name = self.table.split(".")
+        sql = f"""
+            SELECT partition_id AS date_id
+            FROM `{project}.{dataset}.INFORMATION_SCHEMA.PARTITIONS`
+            WHERE table_name = @table_name
+              AND partition_id BETWEEN @min_id AND @max_id
+              AND total_rows > 0
+        """
+        parameters = [
+            bigquery.ScalarQueryParameter("table_name", "STRING", table_name),
+            bigquery.ScalarQueryParameter("min_id", "STRING", min(dates).strftime("%Y%m%d")),
+            bigquery.ScalarQueryParameter("max_id", "STRING", max(dates).strftime("%Y%m%d")),
+        ]
+        return DatesWithDataQuery(sql, parameters)
+
 
 @dataclass(frozen=True)
 class ShardedSource(Source):
     """A date-sharded BigQuery table. Addresses each shard as ``table_YYYYMMDD``."""
 
     separator: ClassVar[str] = "_"
+
+    def dates_with_data_query(self, dates: list[datetime.date]) -> DatesWithDataQuery:
+        """Return a query for dates whose shard table exists.
+
+        Checks table existence directly via ``INFORMATION_SCHEMA.TABLES``,
+        rather than submitting an extract job and inferring "the shard doesn't
+        exist" from a ``NotFound`` error: that exception is raised for *any*
+        missing resource (table, dataset, project), so catching it alone can't
+        tell a genuinely missing shard apart from, say, a typo'd dataset name
+        that would make every date look like a benign gap.
+        """
+        project, dataset, table_name = self.table.split(".")
+        sql = f"""
+            SELECT SUBSTR(table_name, -8) AS date_id
+            FROM `{project}.{dataset}.INFORMATION_SCHEMA.TABLES`
+            WHERE STARTS_WITH(table_name, @table_prefix)
+              AND SUBSTR(table_name, -8) BETWEEN @min_id AND @max_id
+        """
+        parameters = [
+            bigquery.ScalarQueryParameter(
+                "table_prefix", "STRING", f"{table_name}{self.separator}"
+            ),
+            bigquery.ScalarQueryParameter("min_id", "STRING", min(dates).strftime("%Y%m%d")),
+            bigquery.ScalarQueryParameter("max_id", "STRING", max(dates).strftime("%Y%m%d")),
+        ]
+        return DatesWithDataQuery(sql, parameters)