Incremental Sync Strategies for GCP Billing Exports

The specific bottleneck this page solves is the late-arriving adjustment window: Google retroactively rewrites GCP billing rows — committed-use discount reallocations, tax corrections, credit settlements, and usage restatements — for up to ~30 days after a row first lands, and those rewrites target old usage_start_time days while arriving under a new export_time. A sync that filters incrementally on _PARTITIONTIME > last_run (the obvious choice, since the native export is partitioned on load time) will pull the new restated rows but write them as duplicates, or miss them entirely if it keys on usage day. The result is reconciliation drift that compounds invoice-over-invoice until someone full-scans the export table to rebuild from scratch. This page covers the GCP-specific incremental selection logic that sits inside the GCP BigQuery Billing Export Sync engine — the strategy layer that decides which rows to reload and how to merge them without double-counting.

# Root Cause & Failure Modes

The native export table gcp_billing_export_resource_v1_<BILLING_ACCOUNT_ID> is partitioned by _PARTITIONTIME (ingestion time), not by usage_start_time (when the cost was actually incurred). Those two clocks diverge by hours on the happy path and by weeks during settlement. Three naive strategies each break in a distinct way:

• Full reload every run. Correct, but every sync re-scans the entire export table. On a billing account doing seven figures a month, that is hundreds of GB to terabytes of scan per run — slot exhaustion and an on-demand bill that rivals the spend you are trying to optimize.
• WHERE DATE(usage_start_time) = yesterday append. Looks incremental, but Google’s restatements land in yesterday’s-yesterday partitions. The append never revisits them, so credits and committed-use discount reallocations that settle two weeks later are silently lost. Reported net cost stays permanently inflated.
• _PARTITIONTIME > last_run append. Catches restated rows because they re-enter under a fresh ingestion time — but it appends them next to the original regular rows for the same usage_start_time/sku/project key. You now have two rows where there should be one, and every SUM(cost) double-counts the corrected day.

The hard numbers that bound any correct design: the restatement window is roughly 30 days (treat 35 as a safe buffer), the export cadence is eventually-consistent (rows can lag hours), and MERGE on a composite key consumes slot time proportional to the touched partitions, not the table — so bounding the reload window is what keeps the job affordable. The fix is a composite watermark that tracks export_time, a bounded trailing reload window over the affected usage_start_time days, and a deterministic upsert keyed on the natural billing grain.

# Production Pipeline Architecture

This sync runs as a four-phase, idempotent execution model. It is the strategy detail behind the acquisition stage of Cloud Billing Data Ingestion & Parsing, and it assumes the detailed export has already been enabled per GCP Billing Export Configuration.

1. Watermark read. Load the last processed export_time from a metadata table, then rewind it by the 35-day restatement buffer to compute the lower bound of rows to consider.
2. Bounded selection. Select from the native export only rows whose export_time exceeds the rewound watermark, and project the distinct set of usage_start_time day-partitions those rows touch. This is the entire reload scope — typically a few dozen partitions, never the whole table.
3. Staging load. Materialize that bounded slice into a WRITE_TRUNCATE staging table partitioned identically to production. Truncation makes the load itself idempotent: a retried run overwrites the staging table rather than appending to it.
4. Deterministic MERGE. Upsert staging into the curated table on the composite billing grain (invoice_month, usage_start_time, sku_id, project_id, cost_type). WHEN MATCHED overwrites the restated measures; WHEN NOT MATCHED inserts new rows. The same input always produces the same output, so the whole pipeline is safe to re-run after any partial failure.

This is a deliberately different trade-off from the atomic partition-replacement (DELETE+INSERT) approach the parent engine documents: partition replacement is cheaper when whole days are reloaded wholesale, while the composite-key MERGE here is the better fit when you want row-level overwrite semantics and an explicit cost_type precedence. The watermark and reload-window mechanics feed directly into downstream rollups such as Time-Series Aggregation for Daily Cloud Cost Tracking, which can only stay correct if the days it aggregates are themselves restatement-stable.

# Step-by-Step Python Implementation

The module below discovers the bounded reload scope from the export_time watermark, stages exactly those rows, and runs a parameterized MERGE with cost_type precedence. It is import-complete, retries transient GCP API errors with exponential backoff, and persists the watermark only after the merge commits — so a crash mid-run reprocesses the same window on the next invocation rather than skipping it.

import os
import logging
from google.cloud import bigquery
from google.api_core.exceptions import GoogleAPIError, RetryError
from google.api_core.retry import Retry, if_transient_error

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

PROJECT_ID = os.getenv("GCP_PROJECT_ID")
BILLING_ACCOUNT_ID = os.getenv("BILLING_ACCOUNT_ID")  # e.g. 0X0X0X_0X0X0X_0X0X0X
EXPORT_DATASET = os.getenv("EXPORT_DATASET", "billing_export")
BQ_DATASET = "finops_billing"
BQ_PROD_TABLE = "gcp_billing_incremental"
BQ_STAGING_TABLE = "gcp_billing_staging"
WATERMARK_TABLE = "sync_watermarks"
RESTATEMENT_BUFFER_DAYS = 35  # rewind the watermark to cover Google's ~30-day restatement window

# Exponential backoff for transient BigQuery API errors (429/500/503).
bq_retry = Retry(initial=1.0, maximum=60.0, multiplier=2.0,
                 predicate=if_transient_error, timeout=300.0)


def export_table_fqn() -> str:
    suffix = BILLING_ACCOUNT_ID.replace("-", "_")
    return f"`{PROJECT_ID}.{EXPORT_DATASET}.gcp_billing_export_resource_v1_{suffix}`"


def read_watermark(client: bigquery.Client) -> None:
    """Stage only rows newer than (max export_time - buffer) into a truncated table.

    Rewinding the watermark by the restatement buffer is what pulls back the
    older usage_start_time days that Google has since rewritten.
    """
    staging_fqn = f"`{PROJECT_ID}.{BQ_DATASET}.{BQ_STAGING_TABLE}`"
    watermark_fqn = f"`{PROJECT_ID}.{BQ_DATASET}.{WATERMARK_TABLE}`"

    stage_sql = f"""
    CREATE OR REPLACE TABLE {staging_fqn}
    PARTITION BY DATE(usage_start_time) AS
    SELECT
        invoice.month                      AS invoice_month,
        usage_start_time,
        sku.id                             AS sku_id,
        project.id                         AS project_id,
        cost_type,
        cost,
        (SELECT SUM(c.amount) FROM UNNEST(credits) c) AS credits,
        usage.amount                       AS usage_amount,
        export_time
    FROM {export_table_fqn()}
    WHERE export_time > (
        SELECT TIMESTAMP_SUB(
            COALESCE(MAX(last_synced), TIMESTAMP('1970-01-01')),
            INTERVAL {RESTATEMENT_BUFFER_DAYS} DAY)
        FROM {watermark_fqn}
        WHERE sync_type = 'billing_export'
    )
    """
    job = client.query(stage_sql, retry=bq_retry)
    job.result()
    logger.info("Staged %s rows for the bounded reload window.", job.num_dml_affected_rows or "all")


def execute_merge(client: bigquery.Client) -> int:
    """Upsert staging into the curated table on the natural billing grain."""
    prod_fqn = f"`{PROJECT_ID}.{BQ_DATASET}.{BQ_PROD_TABLE}`"
    staging_fqn = f"`{PROJECT_ID}.{BQ_DATASET}.{BQ_STAGING_TABLE}`"
    merge_sql = f"""
    MERGE {prod_fqn} AS target
    USING {staging_fqn} AS source
    ON  target.invoice_month     = source.invoice_month
    AND target.usage_start_time  = source.usage_start_time
    AND target.sku_id            = source.sku_id
    AND target.project_id        = source.project_id
    AND target.cost_type         = source.cost_type
    WHEN MATCHED THEN UPDATE SET
        cost          = source.cost,
        credits       = source.credits,
        usage_amount  = source.usage_amount,
        _last_updated = CURRENT_TIMESTAMP()
    WHEN NOT MATCHED THEN INSERT (
        invoice_month, usage_start_time, sku_id, project_id, cost_type,
        cost, credits, usage_amount, _last_updated)
    VALUES (
        source.invoice_month, source.usage_start_time, source.sku_id,
        source.project_id, source.cost_type, source.cost, source.credits,
        source.usage_amount, CURRENT_TIMESTAMP())
    """
    job = client.query(merge_sql, retry=bq_retry)
    job.result()
    logger.info("MERGE complete. Rows affected: %s", job.num_dml_affected_rows)
    return job.num_dml_affected_rows or 0


def advance_watermark(client: bigquery.Client) -> None:
    """Persist the max export_time only after the merge has committed."""
    watermark_fqn = f"`{PROJECT_ID}.{BQ_DATASET}.{WATERMARK_TABLE}`"
    staging_fqn = f"`{PROJECT_ID}.{BQ_DATASET}.{BQ_STAGING_TABLE}`"
    sql = f"""
    MERGE {watermark_fqn} AS target
    USING (SELECT MAX(export_time) AS max_ts FROM {staging_fqn}) AS source
    ON target.sync_type = 'billing_export'
    WHEN MATCHED AND source.max_ts IS NOT NULL THEN
        UPDATE SET last_synced = source.max_ts
    WHEN NOT MATCHED AND source.max_ts IS NOT NULL THEN
        INSERT (sync_type, last_synced) VALUES ('billing_export', source.max_ts)
    """
    client.query(sql, retry=bq_retry).result()
    logger.info("Watermark advanced to staging max export_time.")


def main() -> None:
    logger.info("Starting incremental GCP billing export sync...")
    client = bigquery.Client(project=PROJECT_ID)
    try:
        read_watermark(client)
        execute_merge(client)
        advance_watermark(client)
        logger.info("Sync pipeline completed successfully.")
    except (GoogleAPIError, RetryError) as exc:
        logger.error("Pipeline failed on a GCP API error: %s", exc)
        raise


if __name__ == "__main__":
    main()

# Verification & Testing

Treat correctness as something you assert, not something you hope for. Three checks catch the failure modes above before they reach a report:

• Idempotency assertion. Run main() twice back-to-back against a fixed export snapshot. The second run’s MERGE should report a row count consistent with zero net change (only _last_updated touched). If the curated SUM(cost) moves between identical runs, your composite key is missing a dimension that distinguishes rows.
• Restatement replay. Pick a usage_start_time day Google has restated (its export_time is days after the usage date) and compare the curated row count and SUM(cost + credits) for that day against SELECT ... FROM <export> WHERE DATE(usage_start_time) = @day. They must match exactly — a divergence means the rewind buffer is too short.
• Bytes-processed budget. Capture job.total_bytes_processed from the staging query in dry-run mode (QueryJobConfig(dry_run=True)) and assert it stays under a partition-sized threshold. A sudden jump to whole-table magnitude signals partition pruning broke — usually a predicate that wraps export_time in a non-sargable function.

For unit coverage, point the client at the BigQuery emulator or a disposable scratch dataset, seed a regular row plus a later adjustment row for the same key, and assert the curated table holds the merged result rather than two rows.

# Common Pitfalls Checklist

• Keying the MERGE on usage day alone. Drop cost_type and sku_id from the ON clause and adjustment rows collapse into regular charges — fix by keying on the full natural grain (invoice_month, usage_start_time, sku_id, project_id, cost_type).
• Advancing the watermark before the merge commits. A crash between steps then skips a window forever — persist the watermark only after execute_merge returns.
• Rewind buffer shorter than the restatement window. Anything under ~30 days lets late credits escape — use 35 days and verify with the restatement-replay check.
• Reporting the cost column as spend. It is pre-credit; net cost is cost + SUM(credits.amount) — carry the summed credits field through the merge, as the code does.
• Filtering on _PARTITIONTIME to drive the upsert. That clock is ingestion time and double-counts restatements — drive selection on export_time and merge on usage-time grain.

# Frequently Asked Questions

# Related

• GCP BigQuery Billing Export Sync — the parent engine this strategy plugs into, covering IAM, curated-table layout, and atomic partition replacement.
• Cloud Billing Data Ingestion & Parsing — the four-stage acquisition→normalization→allocation→persistence model this sync implements for GCP.
• GCP Billing Export Configuration — the account-level setup that produces the detailed export this sync consumes.
• Time-Series Aggregation for Daily Cloud Cost Tracking — the daily rollup layer that depends on restatement-stable partitions produced here.
• Handling Billing API Rate Limits & Retries — the backoff and quota patterns behind the bq_retry decorator used throughout this module.

Up: GCP BigQuery Billing Export Sync · Home: Cloud Cost Optimization & FinOps Automation