gcp-datastream-postgres-bigquery

Creates a Google Cloud Datastream pipeline that continuously replicates change data from a PostgreSQL database into a BigQuery dataset.

This module wires together three Nullstone datastores:

• a Postgres source (datastore/gcp/postgres),
• a BigQuery destination (datastore/gcp/bigquery), and
• the VPC network (network/gcp/vpc) the source database lives in,

and provisions everything Datastream needs to move data between them: a private connection, source/destination connection profiles, and the stream itself.

How it connects

Datastream runs in a Google-managed project, so it reaches your database over Private Service Connect (PSC) rather than directly. This module creates a network attachment in the database's private subnet and a Datastream private connection bound to it. The source database must therefore be reachable over private networking — when using the gcp-cloudsql-postgres module, provision it with enable_psc = true.

The Postgres connection profile connects to the database endpoint exposed by the postgres connection (db_endpoint), against the database named by var.postgres_database.

You do not supply credentials. This module generates the username and password, creates that user on the source database, and hands the same values to Datastream — so there is nothing to create by hand, no secret to manage, and nothing that can drift out of sync. The generated username is exported as the postgres_username output. See What this module creates on the source.

Prerequisite on the source database: logical decoding

Logical replication requires wal_level >= logical. On Cloud SQL for Postgres this is controlled by the cloudsql.logical_decoding database flag — set it to on through the gcp-cloudsql-postgres module's var.db_flags. That flag lives on the postgres block, not on this one, and changing it requires a database restart, so apply the postgres block first and let the instance come back up.

# .nullstone/config.yml, on the postgres block
vars:
  db_flags:
    cloudsql.logical_decoding: "on"

This module reads wal_level before it does anything else. If logical decoding is off, it fails with a message naming the flag instead of letting Datastream fail with an opaque error.

What this module creates on the source

Everything else Datastream needs on the source database is created by this module, so launching a fresh environment needs no out-of-band psql session:

Object	Detail
REPLICATION on the admin role	Creating a replication slot is denied without it
The replication user	LOGIN REPLICATION, with a generated username and password (output as postgres_username)
Read-only grants	USAGE on each schema, SELECT on its tables (existing and future), CONNECT on the database
var.replication_publication	Derived from var.replication_objects, so the two cannot drift
var.replication_slot	Logical slot using the pgoutput plugin

Because the module both creates the user and configures the connection profile, the credentials cannot be mistyped or drift apart — the failure mode that a hand-created password invites.

These are created through the database admin function that the postgres block deploys inside the VPC — the same mechanism gcp-postgres-access uses — because the Terraform runner cannot reach a private database directly. This module therefore requires the postgres block to run gcp-cloudsql-postgres 0.5.0 or later, and fails with a message saying so if that block does not export db_admin_function_url and db_admin_invoker.

Objects that already exist are adopted, not recreated, so an environment whose publication and slot were created by hand applies cleanly and re-applying is a no-op.

Note that the replication slot is genuinely dropped when this module is destroyed. Every other object is left in place, but an abandoned logical slot retains WAL indefinitely and would eventually fill the instance's disk. The stream is torn down before the slot, which is what lets the drop succeed.

The equivalent SQL

For reference when reviewing or debugging an environment, this is what the module performs on your behalf as the Cloud SQL admin user:

-- The admin role cannot create a replication slot without this.
-- Cloud SQL allows a cloudsqlsuperuser member to grant it, including to itself.
ALTER USER <admin-user> WITH REPLICATION;

-- The user Datastream authenticates as, with generated credentials
CREATE USER <generated> WITH REPLICATION LOGIN PASSWORD '<generated>';

-- Read-only access, repeated for every schema in var.replication_objects.
-- USAGE on the schema is required in addition to SELECT on its tables.
-- ALTER DEFAULT PRIVILEGES covers tables created later, and only applies to tables created
-- by the role named in FOR ROLE -- the module applies it for every role that owns a table in
-- the schema, so tables added by later migrations are covered too.
GRANT USAGE ON SCHEMA public TO <generated>;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO <generated>;
ALTER DEFAULT PRIVILEGES FOR ROLE <table-owner> IN SCHEMA public GRANT SELECT ON TABLES TO <generated>;

-- The publication mirrors var.replication_objects:
--   default (public schema)       -> FOR TABLES IN SCHEMA public
--   empty list (everything)       -> FOR ALL TABLES
--   a schema with explicit tables -> FOR TABLE public.invoices, public.payments
CREATE PUBLICATION nullstone_pub1 FOR TABLES IN SCHEMA public;

-- Logical replication slot using the pgoutput plugin
SELECT pg_create_logical_replication_slot('datastream_slot1', 'pgoutput');

Selecting what to replicate

var.replication_objects controls which schemas and tables are streamed:

• The default ([{ schema = "public" }]) replicates every table in the public schema.
• Set it to empty ([]) to replicate every table in every schema.
• List a schema with no tables to replicate all tables in that schema.
• List specific tables under a schema to replicate only those tables.

replication_objects = [
  { schema = "public" },
  { schema = "billing", tables = ["invoices", "payments"] },
]

Destination layout

The stream writes to the dataset from the bigquery connection using Datastream's merge mode (single_target_dataset). Each source table becomes a BigQuery table that is kept as a current-state replica — inserts, updates, and soft-deletes are merged in, so no "latest row" logic is needed when querying. var.data_freshness controls how often those merges run (and therefore how fresh the replica is).

The stream is created with backfill_all, so existing rows are backfilled once before ongoing change data is applied. By default it starts in the RUNNING state and replicates as soon as it is created.

Pausing replication

Set var.enabled = false to pause the stream (desired_state = "PAUSED") without destroying it or losing its configuration. Replication stops, but the stream, connection profiles, and private connection are all left in place. Set it back to true to resume.

Outputs

Output	Description
stream_id	The id of the Datastream stream.
stream_name	The fully-qualified resource name of the Datastream stream.
stream_state	The current state of the Datastream stream (e.g. RUNNING).
private_connection_id	The id of the Datastream private connection.
source_connection_profile_id	The id of the Postgres source connection profile.
destination_connection_profile_id	The id of the BigQuery destination connection profile.
network_attachment_id	The id of the network attachment Datastream uses to reach the source over PSC.