Implementing Logical Replication in PostgreSQL: Multi-Master and Selective Data Sync

Introduction

Logical replication in PostgreSQL allows you to replicate specific tables, rows, or even columns between databases, enabling sophisticated data distribution patterns. Unlike physical streaming replication that copies the entire database cluster, logical replication provides fine-grained control over what data gets replicated and where.

Logical vs Physical Replication

Physical Streaming Replication

• Replicates entire database cluster
• Binary-level replication
• Read-only replicas
• Same PostgreSQL version required
• Lower overhead

Logical Replication

• Selective table/row replication
• Cross-version compatible
• Writable subscribers
• Higher overhead
• Flexible data distribution

Use Cases for Logical Replication

1. Selective Data Distribution: Replicate specific tables to different regions
2. Multi-Master Setups: Multiple writable databases with bidirectional sync
3. Cross-Version Upgrades: Replicate from old to new PostgreSQL versions
4. Data Aggregation: Consolidate data from multiple sources
5. GDPR Compliance: Replicate only non-sensitive columns

Prerequisites and Setup

Configuration Requirements

On Publisher (Source):

# postgresql.conf
wal_level = logical
max_replication_slots = 10
max_wal_senders = 10

On Subscriber (Target):

# postgresql.conf
max_replication_slots = 10
max_logical_replication_workers = 10
max_worker_processes = 16

Restart PostgreSQL

sudo systemctl restart postgresql

Network Configuration

Ensure databases can communicate:

# Test connection from subscriber to publisher
psql -h publisher.example.com -U replication_user -d source_db

Configure pg_hba.conf on publisher:

# Allow replication connections
host    all    replication_user    subscriber_ip/32    md5

Basic Logical Replication Setup

Step 1: Create Replication User

On Publisher:

CREATE ROLE replication_user WITH REPLICATION LOGIN PASSWORD 'secure_password';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO replication_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO replication_user;

Step 2: Create Source Tables

On Publisher:

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) NOT NULL,
    email VARCHAR(100) NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    amount DECIMAL(10,2),
    status VARCHAR(20),
    created_at TIMESTAMP DEFAULT NOW()
);

INSERT INTO users (username, email) VALUES 
    ('alice', '[email protected]'),
    ('bob', '[email protected]');

Step 3: Create Publication

On Publisher:

-- Publish all tables
CREATE PUBLICATION my_publication FOR ALL TABLES;

-- Or publish specific tables
CREATE PUBLICATION my_publication FOR TABLE users, orders;

-- Or with row filters (PostgreSQL 15+)
CREATE PUBLICATION active_users FOR TABLE users WHERE (status = 'active');

View publications:

SELECT * FROM pg_publication;
SELECT * FROM pg_publication_tables;

Step 4: Create Replica Tables

On Subscriber:

-- Tables must have identical structure
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) NOT NULL,
    email VARCHAR(100) NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    user_id INTEGER REFERENCES users(id),
    amount DECIMAL(10,2),
    status VARCHAR(20),
    created_at TIMESTAMP DEFAULT NOW()
);

Step 5: Create Subscription

On Subscriber:

CREATE SUBSCRIPTION my_subscription
    CONNECTION 'host=publisher.example.com port=5432 dbname=source_db user=replication_user password=secure_password'
    PUBLICATION my_publication;

Step 6: Verify Replication

On Publisher:

SELECT * FROM pg_stat_replication;
SELECT * FROM pg_replication_slots;

On Subscriber:

SELECT * FROM pg_stat_subscription;
SELECT * FROM users;  -- Should see replicated data

Advanced Configuration

Column-Level Replication (PostgreSQL 15+)

Replicate only specific columns:

-- On Publisher: Replicate only non-sensitive columns
CREATE TABLE customers (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100),
    credit_card VARCHAR(20),  -- Won't be replicated
    created_at TIMESTAMP
);

CREATE PUBLICATION customer_basic 
    FOR TABLE customers (id, name, email, created_at);

Row Filtering

Replicate only active records:

CREATE PUBLICATION active_data 
    FOR TABLE orders WHERE (status IN ('pending', 'processing'));

Regional data distribution:

CREATE PUBLICATION us_customers 
    FOR TABLE customers WHERE (country = 'US');

CREATE PUBLICATION eu_customers 
    FOR TABLE customers WHERE (country IN ('UK', 'DE', 'FR'));

Multiple Publications

-- Publisher: Create multiple publications
CREATE PUBLICATION oltp_data FOR TABLE users, orders;
CREATE PUBLICATION analytics_data FOR TABLE logs, metrics;

-- Subscriber: Subscribe to multiple publications
CREATE SUBSCRIPTION multi_sub
    CONNECTION 'host=publisher port=5432 dbname=mydb user=repuser password=pass'
    PUBLICATION oltp_data, analytics_data;

Bidirectional Replication (Multi-Master)

Setup Two-Way Sync

Database A Configuration:

-- Create publication
CREATE PUBLICATION db_a_pub FOR TABLE shared_table;

-- Subscribe to Database B
CREATE SUBSCRIPTION db_a_sub
    CONNECTION 'host=db-b.example.com dbname=mydb user=repuser'
    PUBLICATION db_b_pub
    WITH (origin = none);  -- Prevents replication loops

Database B Configuration:

-- Create publication
CREATE PUBLICATION db_b_pub FOR TABLE shared_table;

-- Subscribe to Database A
CREATE SUBSCRIPTION db_b_sub
    CONNECTION 'host=db-a.example.com dbname=mydb user=repuser'
    PUBLICATION db_a_pub
    WITH (origin = none);

Conflict Resolution

Logical replication uses "last write wins" by default:

-- Set replica identity to track conflicts
ALTER TABLE shared_table REPLICA IDENTITY FULL;

Conflict detection strategies:

1. Timestamp-based: Add updated_at column

CREATE TABLE shared_table (
    id SERIAL PRIMARY KEY,
    data TEXT,
    updated_at TIMESTAMP DEFAULT NOW()
);

CREATE OR REPLACE FUNCTION update_timestamp()
RETURNS TRIGGER AS $$
BEGIN
    NEW.updated_at = NOW();
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER update_shared_table_timestamp
    BEFORE UPDATE ON shared_table
    FOR EACH ROW
    EXECUTE FUNCTION update_timestamp();

2. Version numbering:

CREATE TABLE shared_table (
    id SERIAL PRIMARY KEY,
    data TEXT,
    version INTEGER DEFAULT 1
);

Initial Data Sync Options

Option 1: Automatic Copy (Default)

-- Subscriber automatically copies existing data
CREATE SUBSCRIPTION my_sub
    CONNECTION 'host=publisher dbname=mydb user=repuser'
    PUBLICATION my_pub
    WITH (copy_data = true);  -- Default

Option 2: Manual Initial Sync

For large datasets, use pg_dump:

# Dump specific tables from publisher
pg_dump -h publisher.example.com -U postgres -d source_db \
    -t users -t orders --no-owner --no-acl > initial_data.sql

# Load into subscriber
psql -h subscriber.example.com -U postgres -d target_db < initial_data.sql

# Create subscription without initial copy
psql -h subscriber.example.com -U postgres -d target_db -c "
    CREATE SUBSCRIPTION my_sub
    CONNECTION 'host=publisher dbname=source_db user=repuser'
    PUBLICATION my_pub
    WITH (copy_data = false);
"

Option 3: Parallel Initial Sync

-- Use multiple workers for faster initial sync
CREATE SUBSCRIPTION fast_sync
    CONNECTION 'host=publisher dbname=mydb user=repuser'
    PUBLICATION my_pub
    WITH (
        copy_data = true,
        streaming = on,
        synchronous_commit = off
    );

Monitoring Logical Replication

Publisher Monitoring

-- View replication slots
SELECT 
    slot_name,
    plugin,
    active,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS lag_size
FROM pg_replication_slots;

-- View active replication connections
SELECT 
    pid,
    usename,
    application_name,
    client_addr,
    state,
    sync_state,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), sent_lsn)) AS send_lag
FROM pg_stat_replication;

Subscriber Monitoring

-- View subscription status
SELECT 
    subname,
    pid,
    received_lsn,
    latest_end_lsn,
    pg_size_pretty(pg_wal_lsn_diff(latest_end_lsn, received_lsn)) AS lag
FROM pg_stat_subscription;

-- Check for replication errors
SELECT * FROM pg_stat_subscription WHERE last_msg_receipt_time < NOW() - INTERVAL '5 minutes';

Monitoring Script

#!/bin/bash
# logical-replication-monitor.sh

echo "=== Publisher Status ==="
psql -h publisher -d mydb -c "
    SELECT slot_name, active, 
           pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS lag
    FROM pg_replication_slots;"

echo ""
echo "=== Subscriber Status ==="
psql -h subscriber -d mydb -c "
    SELECT subname, pid, 
           pg_size_pretty(pg_wal_lsn_diff(latest_end_lsn, received_lsn)) AS lag
    FROM pg_stat_subscription;"

Troubleshooting

Issue 1: Subscription Not Receiving Data

Check subscription status:

SELECT subname, pid, subenabled, subconninfo FROM pg_subscription;

Enable subscription if disabled:

ALTER SUBSCRIPTION my_sub ENABLE;

Check for errors:

SELECT * FROM pg_stat_subscription;

Issue 2: Replication Lag Growing

Identify slow tables:

SELECT schemaname, tablename, n_tup_ins, n_tup_upd, n_tup_del
FROM pg_stat_user_tables
ORDER BY (n_tup_ins + n_tup_upd + n_tup_del) DESC;

Increase worker processes:

# postgresql.conf on subscriber
max_logical_replication_workers = 20
max_worker_processes = 30

Issue 3: Replication Slot Bloat

Check slot usage:

SELECT 
    slot_name,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
FROM pg_replication_slots;

Drop inactive slots:

SELECT pg_drop_replication_slot('inactive_slot_name');

Issue 4: Initial Sync Failure

Restart initial sync:

-- Drop and recreate subscription
DROP SUBSCRIPTION my_sub;

CREATE SUBSCRIPTION my_sub
    CONNECTION 'host=publisher dbname=mydb user=repuser'
    PUBLICATION my_pub
    WITH (copy_data = true);

Performance Optimization

1. Use Appropriate Replica Identity

-- Default: PRIMARY KEY only
ALTER TABLE my_table REPLICA IDENTITY DEFAULT;

-- Full: All columns (higher overhead)
ALTER TABLE my_table REPLICA IDENTITY FULL;

-- Index: Use specific unique index
CREATE UNIQUE INDEX replica_idx ON my_table(col1, col2);
ALTER TABLE my_table REPLICA IDENTITY USING INDEX replica_idx;

2. Disable Constraints During Initial Sync

-- Temporarily disable triggers for faster initial load
ALTER TABLE my_table DISABLE TRIGGER ALL;

-- After initial sync completes
ALTER TABLE my_table ENABLE TRIGGER ALL;

3. Parallel Apply (PostgreSQL 16+)

CREATE SUBSCRIPTION parallel_sub
    CONNECTION 'host=publisher dbname=mydb user=repuser'
    PUBLICATION my_pub
    WITH (streaming = parallel);

Best Practices

1. Always use replica identity: Ensure tables have PRIMARY KEY or UNIQUE constraint
2. Monitor replication lag: Set up alerts for lag > 100MB or 5 minutes
3. Use row filters carefully: Filters evaluated on publisher, affecting performance
4. Plan for conflicts: Implement conflict detection in multi-master setups
5. Test failover procedures: Practice switching between databases
6. Regular maintenance: Clean up old replication slots

Conclusion

Logical replication provides powerful data distribution capabilities for PostgreSQL. Key advantages:

• Selective table and row replication
• Cross-version compatibility
• Bidirectional multi-master configurations
• Fine-grained data privacy control

Start with simple single-direction replication, test thoroughly, then expand to more complex topologies as needed.