PostgreSQL CDC

​

Overview

​

Configuration

​

Required Parameters

​

Optional Parameters

​

SSL Configuration

​

Record Formats

​

DB Records Only Format

{
  "data": {
    "id": 1,
    "name": "John",
    "email": "john@example.com",
    "_mage_created_at": "2021-12-31T23:00:00+00:00",
    "_mage_updated_at": null,
    "_mage_deleted_at": null
  },
  "metadata": {
    "schema": "public",
    "table": "users",
    "key_columns": ["id"],
    "operation": "INSERT",
    "lsn": "0/1234567"
  }
}

​

Event Types

​

Primary Key Detection

1. Schema Discovery: Queries INFORMATION_SCHEMA for table structure
2. Primary Key Detection: Identifies actual primary key columns from database constraints
3. Caching: Stores schema and primary key information for performance
4. Key Extraction: Extracts primary key column names from row events

{
  "metadata": {
    "key_columns": ["user_id", "tenant_id"]
  }
}

​

Examples

connector_type: postgres
host: localhost
port: 5432
database: mydb
user: repl_user
password: password
schemas: ["public"]
return_db_records_only: true

​

Prerequisites

​

PostgreSQL Server Configuration

-- Enable logical replication
ALTER SYSTEM SET wal_level = logical;
-- Restart PostgreSQL for changes to take effect

-- Verify WAL level
SHOW wal_level;
-- Should return: logical

​

Create Replication Slot

-- Create logical replication slot
SELECT pg_create_logical_replication_slot('mage_slot', 'pgoutput');

​

Create Publication

-- Create publication for all tables in a schema
CREATE PUBLICATION mage_pub FOR ALL TABLES;

-- Or create publication for specific tables
CREATE PUBLICATION mage_pub FOR TABLE users, orders, products;

-- Or create publication for specific schemas
CREATE PUBLICATION mage_pub FOR TABLES IN SCHEMA public, analytics;

​

User Permissions

-- Create replication user
CREATE USER repl_user WITH PASSWORD 'password';

-- Grant replication privileges
ALTER USER repl_user WITH REPLICATION;

-- Grant connect privileges
GRANT CONNECT ON DATABASE mydb TO repl_user;

-- Grant usage on schemas
GRANT USAGE ON SCHEMA public TO repl_user;
GRANT USAGE ON SCHEMA analytics TO repl_user;

-- Grant SELECT privileges for schema queries
GRANT SELECT ON ALL TABLES IN SCHEMA public TO repl_user;
GRANT SELECT ON ALL TABLES IN SCHEMA analytics TO repl_user;

-- Grant SELECT on information_schema for metadata queries
GRANT SELECT ON ALL TABLES IN SCHEMA information_schema TO repl_user;

​

Troubleshooting

SELECT * FROM pg_replication_slots WHERE slot_name = 'mage_slot';

SELECT * FROM pg_publication WHERE pubname = 'mage_pub';
SELECT * FROM pg_publication_tables WHERE pubname = 'mage_pub';

SELECT 
    slot_name,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS replication_lag
FROM pg_replication_slots
WHERE slot_name = 'mage_slot';

​

Integration with Generic IO Sink

• Automatic Column Type Mapping: Automatically maps _mage_* timestamp columns to appropriate database types (TIMESTAMP, DATETIME2, DateTime64, etc.) based on the target database
• Metadata Interpolation: Use metadata values from PostgreSQL CDC events in sink configurations for dynamic routing and table naming

​

Supported Databases

• BigQuery
• ClickHouse
• DuckDB
• MySQL
• MSSQL
• Postgres

​

Metadata Interpolation

• {schema}: Schema name from the event
• {table}: Table name from the event
• {key_columns}: List of primary key column names (e.g., ["id"] or ["user_id", "tenant_id"])

​

Example Configurations

connector_type: bigquery
profile: default
config:
  table_id: "{schema}.{table}"  # Dynamic table names from metadata
  unique_conflict_method: UPDATE
  unique_constraints: "{key_columns}"  # Uses primary keys from metadata

​

How Metadata Interpolation Works

1. Message Grouping: Messages are automatically grouped by their interpolated config values. For example, if you use table_name: "{schema}_{table}", messages from public.users will be grouped together and written to public_users table.
2. Key Columns Interpolation: When using {key_columns} in unique_constraints, the sink automatically:
   • Converts the list to a string representation during interpolation
   • Parses it back to a list (supports both "['id']" and '["id"]' formats)
   • Uses it for upsert operations based on unique_conflict_method

​

Best Practices

1. Use Replication Slots: Always use replication slots for reliable CDC tracking
2. Monitor WAL Growth: Monitor WAL size and ensure proper cleanup
3. Filter Events: Use schema/table filters to reduce processing overhead
4. Monitor Resources: Watch memory and CPU usage
5. Test Resume: Verify checkpoint functionality with LSN tracking
6. Secure Connections: Use SSL in production
7. Regular Backups: Backup checkpoint files
8. Schema Validation: Test with schema changes
9. Use Generic IO Sink: Leverage automatic type mapping and metadata interpolation for flexible data routing
10. Metadata Interpolation: Use {schema}, {table}, and {key_columns} for dynamic table routing and upsert configuration
11. Heartbeat Tables: Use heartbeat tables for managed services (like Aiven) to prevent WAL log filling

​

Timestamp Handling

• _mage_created_at: Set to event timestamp (datetime) for INSERT operations
• _mage_updated_at: Set to event timestamp (datetime) for UPDATE operations
• _mage_deleted_at: Set to event timestamp (datetime) for DELETE operations

• Proper type handling in downstream databases
• Automatic type conversion in Generic IO sink
• Consistent timezone handling across systems

​

Limitations

• PostgreSQL 10+: Requires PostgreSQL 10 or later for logical replication
• WAL Level: Requires wal_level = logical
• Network Dependency: Requires stable network connection
• Memory Usage: Schema caching uses memory
• WAL Retention: Depends on PostgreSQL WAL retention settings
• Timezone: All timestamps are in UTC timezone
• Replication Slot: Requires a dedicated replication slot per consumer
• Publication: Tables must be added to a publication to be replicated