Incremental Data Transfers From PostgreSQL to BigQuery

Moving data from PostgreSQL to BigQuery is a common requirement for applications that need analytics beyond what their transactional database can comfortably handle. PostgreSQL can continue serving application requests, while BigQuery handles reporting, dashboards, and analytical queries over larger datasets.

The simplest approach is to copy the entire PostgreSQL table into BigQuery on every run. This works for small datasets, but it becomes increasingly expensive and time-consuming as the tables grow. Most records may remain unchanged between runs, yet a full refresh transfers them again.

Incremental data transfers solve this problem by moving only records that are new or have changed since the previous successful run. Instead of repeatedly copying an entire table, the pipeline identifies changes, transfers them to BigQuery, and updates the analytical tables.

This article explains how to design an incremental transfer pipeline from PostgreSQL to BigQuery, implement a practical batch-loading example, handle updates and deletions, and make the process reliable enough for production workloads.

What Is an Incremental Data Transfer?

An incremental transfer extracts only the changes that have occurred since a previous extraction point.

Suppose a PostgreSQL table contains ten million orders, but only five thousand orders have been created or updated since the last pipeline run. A full refresh processes all ten million records, while an incremental process can focus on the five thousand changed records.

A typical pipeline follows this sequence:

Identify new and changed records in PostgreSQL.
Export the selected records to a staging file or data stream.
Load the extracted data into a BigQuery staging table.
Merge the staged records into the target table.
Record the successful extraction checkpoint.

The checkpoint records how far the pipeline has progressed. It must be managed carefully because an incorrect checkpoint can cause missed records or repeated processing.

Incremental loading is particularly useful for transaction tables, customer records, product catalogs, application events, and other datasets that grow or change over time.

Choosing an Incremental Extraction Strategy

The extraction strategy determines how the pipeline detects changes.

Using a modification timestamp

One common approach is to add a column such as updated_at to the PostgreSQL table.

CREATE TABLE orders (
    order_id BIGINT PRIMARY KEY,
    customer_id BIGINT NOT NULL,
    order_status TEXT NOT NULL,
    total_amount NUMERIC(12, 2) NOT NULL,
    updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
);

The application must update updated_at whenever a record changes. PostgreSQL does not automatically update this column merely because another column changes.

For example, an update statement can maintain the timestamp explicitly:

UPDATE orders
SET order_status = 'SHIPPED',
    updated_at = CURRENT_TIMESTAMP
WHERE order_id = 1001;

The pipeline can then select records modified after its previous checkpoint:

SELECT
    order_id,
    customer_id,
    order_status,
    total_amount,
    updated_at
FROM orders
WHERE updated_at >= $1
  AND updated_at < $2
ORDER BY updated_at, order_id;

Here, $1 is the lower extraction boundary and $2 is the upper boundary.

Using an exclusive upper boundary helps define a stable extraction window. The lower boundary may deliberately overlap with a previous run to account for late commits or timestamp precision issues. Because overlapping extraction can return records more than once, the destination must handle duplicates safely.

Timestamp-based extraction is straightforward, but it depends on correct timestamp maintenance and does not automatically detect hard deletes.

Using a monotonically increasing identifier

An increasing identifier can work when records are append-only.

SELECT *
FROM application_events
WHERE event_id > $1
ORDER BY event_id;

The checkpoint stores the highest successfully processed identifier.

This method is simple, but it does not detect updates to older records. It also requires care if identifiers are allocated before transactions commit. A higher identifier can become visible before a transaction with a lower identifier commits, so a checkpoint based only on the maximum identifier can miss late-committing records.

For mutable tables, a modification timestamp, change log, or change data capture mechanism is generally more suitable.

Using change data capture

Change data capture, or CDC, captures database changes rather than repeatedly querying tables for modified rows.

PostgreSQL logical replication and logical decoding can provide change streams that include inserts, updates, and deletes. A CDC pipeline can preserve these changes and apply them to BigQuery.

CDC is useful when low latency, delete detection, or high change volume makes timestamp polling difficult. However, it requires more infrastructure and careful management of replication slots, offsets, event ordering, and recovery.

For a simple batch pipeline, a reliable updated_at column may be sufficient. For a database with frequent updates and deletes, CDC deserves serious consideration.

Building a Batch Pipeline

The following example uses Python to extract records from PostgreSQL, write them to a CSV file, and load the file into a BigQuery staging table.

The example demonstrates the core transfer process. Production deployment still requires secure credential management, checkpoint storage, monitoring, and an orchestration mechanism.

The pipeline uses these components:

PostgreSQL as the source database.
Python with psycopg to query PostgreSQL.
Google Cloud BigQuery for analytical storage.
A staging table to receive each batch before it is merged into the destination.

Install the required packages:

pip install "psycopg[binary]" google-cloud-bigquery

Configure PostgreSQL connection settings through environment variables and configure Google Cloud authentication using an approved workload identity or application credential mechanism. Avoid placing passwords or service account keys directly in source code.

Step 1: Extract a bounded batch

Use a fixed extraction window instead of repeatedly querying until the current time. A fixed upper boundary makes the batch easier to retry.

import os
import csv
import psycopg
from datetime import datetime, timezone

def extract_orders(start_time, end_time, output_file):
    connection_string = os.environ["POSTGRES_CONNECTION_STRING"]

    with psycopg.connect(connection_string) as connection:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                SELECT
                    order_id,
                    customer_id,
                    order_status,
                    total_amount,
                    updated_at
                FROM orders
                WHERE updated_at >= %s
                  AND updated_at < %s
                ORDER BY updated_at, order_id
                """,
                (start_time, end_time),
            )

            rows = cursor.fetchall()
            column_names = [
                description.name
                for description in cursor.description
            ]

    with open(output_file, "w", newline="", encoding="utf-8") as file:
        writer = csv.writer(file)
        writer.writerow(column_names)
        writer.writerows(rows)

    return len(rows)

The query uses parameterized values instead of constructing SQL through string concatenation. This avoids SQL injection risks and handles timestamp values through the database driver.

The extraction window should be selected by the orchestrator. For example, a run might process records from the previous successful checkpoint up to a fixed time captured at the start of the run.

The function returns the number of extracted records. Record this value in the pipeline logs so operators can distinguish an empty batch from an extraction failure.

For large datasets, fetchall() can consume substantial memory. Use server-side cursors, pagination with a stable key, or a streaming export approach when batches exceed the available memory.

Step 2: Load the file into BigQuery

Create a staging table with a schema compatible with the extracted data.

For this example, assume the destination tables are in the analytics dataset and the staging table is named orders_staging.

from google.cloud import bigquery

def load_to_bigquery(project_id, dataset_id, table_id, csv_file):
    client = bigquery.Client(project=project_id)

    table_ref = f"{project_id}.{dataset_id}.{table_id}"

    job_config = bigquery.LoadJobConfig(
        source_format=bigquery.SourceFormat.CSV,
        skip_leading_rows=1,
        autodetect=True,
        write_disposition=bigquery.WriteDisposition.WRITE_TRUNCATE,
    )

    with open(csv_file, "rb") as file:
        job = client.load_table_from_file(
            file,
            table_ref,
            job_config=job_config,
        )

    job.result()

    return job.output_rows

The example uses WRITE_TRUNCATE because the staging table is intended to contain only the current batch. This is appropriate only when the pipeline owns that staging table and concurrent runs cannot overwrite one another's data.

For concurrent pipelines, use separate staging tables or a batch identifier and design the staging lifecycle accordingly.

Schema autodetection is convenient for an example, but explicit schemas are safer in production. Define types for identifiers, monetary values, timestamps, and nullable fields so that schema changes do not silently alter the destination representation.

Step 3: Merge the staged records into the destination

Loading the batch into a staging table does not update the final analytical table. A MERGE statement can insert new records and update existing records based on their primary key.

MERGE `my-project.analytics.orders` AS target
USING `my-project.analytics.orders_staging` AS source
ON target.order_id = source.order_id

WHEN MATCHED
  AND source.updated_at >= target.updated_at
THEN UPDATE SET
  customer_id = source.customer_id,
  order_status = source.order_status,
  total_amount = source.total_amount,
  updated_at = source.updated_at

WHEN NOT MATCHED THEN
  INSERT (
    order_id,
    customer_id,
    order_status,
    total_amount,
    updated_at
  )
  VALUES (
    source.order_id,
    source.customer_id,
    source.order_status,
    source.total_amount,
    source.updated_at
  );

Replace my-project with the actual Google Cloud project ID.

The merge uses order_id as the matching key. Existing rows are updated when the incoming timestamp is at least as recent as the target timestamp. New rows are inserted.

The timestamp comparison can help prevent an older batch from overwriting a newer version of a record. It is not a complete concurrency-control mechanism, however. If two records have the same timestamp but different contents, the pipeline needs a deterministic ordering rule or a more precise source change version.

Also ensure that the staging source contains at most one effective row per order_id. BigQuery MERGE operations can fail when multiple source rows match the same target row for an update or delete. Deduplicate the source using a deterministic rule before merging.

Managing Checkpoints Safely

The checkpoint determines which records the next batch should extract. It may be stored in a metadata table, an orchestration system, or another durable store.

A basic checkpoint record could contain:

Field	Purpose
Pipeline name	Identifies the source-to-destination workflow
Last successful boundary	Defines the lower extraction point
Current batch boundary	Records the upper extraction point
Batch status	Tracks whether extraction, loading, and merging completed
Processed row count	Helps with monitoring and reconciliation
Run identifier	Connects logs, staging data, and checkpoint updates

The checkpoint should advance only after the batch has been successfully applied to the destination.

A safe execution sequence is:

Read the last successful checkpoint.
Choose and persist a fixed upper boundary for the batch.
Extract records for that window.
Load the extracted records into staging.
Merge the staged records into the target table.
Mark the batch successful and advance the checkpoint.

If the merge fails, do not advance the checkpoint. Retry the same batch using the recorded boundaries.

This requires the pipeline to be idempotent: processing the same batch again should not create duplicate records or corrupt newer data. Primary-key-based merging and deterministic staging data help achieve this.

The checkpoint update and the BigQuery merge are not one shared transaction. The orchestration layer must account for failures between these operations. For example, if the merge succeeds but the checkpoint update fails, rerunning the batch should safely apply the same data again.

Handling Deletes and Late-Arriving Changes

A query that filters on updated_at detects inserts and updates only when the timestamp is maintained correctly. It does not detect a row that has been physically deleted from PostgreSQL because that row is no longer available to the extraction query.

There are several ways to address this:

Use CDC to capture delete events.
Maintain a soft-delete flag and update its timestamp when a record is deleted logically.
Write deletion events to a separate change table.
Run periodic reconciliation jobs to compare source and destination data.

A soft-delete design might use columns such as is_deleted and updated_at. The pipeline transfers the changed row, and the destination marks it as deleted or removes it according to the analytical data model.

Late-arriving transactions are another concern. A transaction may commit after a pipeline has already processed a timestamp boundary. Overlapping extraction windows can reduce this risk, but the destination must tolerate duplicate records.

For high-reliability pipelines, CDC with durable offsets is often preferable to relying solely on timestamps.

Common Mistakes and Troubleshooting

Records are missing from BigQuery

Check whether the source application updates updated_at on every relevant change. Confirm the extraction boundaries and the stored checkpoint.

Also investigate late commits and timezone handling. Use timezone-aware timestamps consistently and define whether extraction boundaries are inclusive or exclusive.

Duplicate rows appear in the staging table

Overlapping extraction windows can intentionally re-read records. This is acceptable when the merge logic handles repeated records correctly.

If the staging source contains several versions of the same key, deduplicate it before merging.

Older data overwrites newer data

Confirm that the merge compares source and target versions or timestamps. If timestamps do not uniquely identify change order, use a more reliable source version or CDC ordering metadata.

The pipeline consumes too much memory

Avoid loading an entire large result set into memory. Stream rows to storage or use a bulk export mechanism appropriate for the workload.

The schema changes unexpectedly

Replace schema autodetection with an explicit schema. Add validation for new or missing source columns and handle schema changes through a controlled migration process.

Best Practices for Production Pipelines

Use a stable extraction window and durable checkpoints.
Make every batch safe to retry.
Use primary keys and deterministic merge rules.
Keep staging tables isolated between concurrent runs.
Use explicit BigQuery schemas for important datasets.
Monitor extracted, loaded, inserted, updated, and rejected row counts.
Record batch identifiers and boundaries for troubleshooting.
Use secure credential management and least-privilege IAM roles.
Plan explicitly for deletes and late-arriving changes.
Reconcile source and destination data periodically.
Consider CDC when polling cannot meet latency or correctness requirements.

Advantages and Limitations

Approach	Advantages	Limitations
Full refresh	Simple to understand and implement	Reprocesses unchanged data
Timestamp-based incremental load	Straightforward for mutable tables	Depends on reliable timestamps and does not inherently detect deletes
Increasing identifier	Efficient for append-only datasets	Does not capture updates and can be unsafe with late commits
Change data capture	Captures inserts, updates, and deletes with change ordering	Requires additional infrastructure and operational monitoring
Staging plus `MERGE`	Supports repeatable inserts and updates	Requires careful key selection, deduplication, and retry handling

The right approach depends on data volume, change frequency, latency requirements, and the cost of missing a change. A timestamp-based batch process is often a good starting point, while high-volume or correctness-sensitive pipelines may need CDC.

Summary

Incremental transfers reduce unnecessary data movement by extracting only new and changed records from PostgreSQL and applying them to BigQuery.

A reliable implementation needs more than a filtered SQL query. It requires stable extraction boundaries, a staging process, deterministic merge logic, durable checkpoints, and a clear strategy for deletes and late-arriving changes.

Start with a bounded timestamp-based batch when the source schema supports it. As the pipeline grows, add monitoring, reconciliation, and CDC where needed. These controls make the transfer easier to recover, troubleshoot, and operate as part of a production analytics platform.