Incrementally importing data to a PostgreSQL database

A good way to start would probably be to use the --inserts option to pg_dump. From the documentation (emphasis mine) :

Dump data as INSERT commands (rather than COPY). This will make restoration very slow; it is mainly useful for making dumps that can be loaded into non-PostgreSQL databases. However, since this option generates a separate command for each row, an error in reloading a row causes only that row to be lost rather than the entire table contents. Note that the restore might fail altogether if you have rearranged column order. The --column-inserts option is safe against column order changes, though even slower.

I don't have the means to test it right now with pg_restore, but this might be enough for your case.

You could also use the fact that from the version 9.5, PostgreSQL provides ON CONFLICT DO ... for INSERTs. Use a simple scripting language to add these to the dump and you should be fine. I haven't found an option for pg_dump to add those automatically, unfortunately.

You might google "sporadically connected database synchronization" to see related solutions.

It's not a neatly solved problem as far as I know - there are some common work-arounds, but I am not aware of a database-centric out-of-the-box solution.

The most common way of dealing with this is to use a message bus to move events between your machines. For instance, if your "source database" is just a data store, with no other logic, you might get rid of it, and use a message bus to say "event x has occurred", and point the endpoint of that message bus at your "destination machine", which then writes that to your database.

You might consider Apache ActiveMQ or read "Patterns of enterprise integration".

#!/bin/sh

PSQL=/opt/postgres-9.5/bin/psql

TARGET_HOST=localhost
TARGET_DB=mystuff
TARGET_SCHEMA_IMPORT=copied
TARGET_SCHEMA_FINAL=final

SOURCE_HOST=192.168.0.101
SOURCE_DB=slurpert
SOURCE_SCHEMA=public

########
create_local_stuff()
{
${PSQL} -h ${TARGET_HOST} -U postgres ${TARGET_DB} <<OMG0

CREATE SCHEMA IF NOT EXISTS  ${TARGET_SCHEMA_IMPORT};
CREATE SCHEMA IF NOT EXISTS  ${TARGET_SCHEMA_FINAL};
CREATE TABLE IF NOT EXISTS  ${TARGET_SCHEMA_FINAL}.topic
        ( topic_id INTEGER NOT NULL PRIMARY KEY
        , topic_date TIMESTAMP WITH TIME ZONE
        , topic_body text
        );

CREATE TABLE IF NOT EXISTS  ${TARGET_SCHEMA_IMPORT}.tmp_topic
        ( topic_id INTEGER NOT NULL PRIMARY KEY
        , topic_date TIMESTAMP WITH TIME ZONE
        , topic_body text
        );
OMG0
}
########
find_highest()
{
${PSQL} -q -t -h ${TARGET_HOST} -U postgres ${TARGET_DB} <<OMG1
SELECT MAX(topic_id) FROM ${TARGET_SCHEMA_IMPORT}.tmp_topic;
OMG1
}
########
fetch_new_data()
{
watermark=${1-0}

echo ${watermark}

${PSQL} -h ${SOURCE_HOST} -U postgres ${SOURCE_DB} <<OMG2

\COPY (SELECT topic_id, topic_date, topic_body FROM ${SOURCE_SCHEMA}.topic WHERE topic_id >${watermark}) TO '/tmp/topic.dat';
OMG2
}
########

insert_new_data()
{
${PSQL} -h ${TARGET_HOST} -U postgres ${TARGET_DB} <<OMG3

DELETE FROM ${TARGET_SCHEMA_IMPORT}.tmp_topic WHERE 1=1;

COPY ${TARGET_SCHEMA_IMPORT}.tmp_topic(topic_id, topic_date, topic_body) FROM '/tmp/topic.dat';

INSERT INTO ${TARGET_SCHEMA_FINAL}.topic(topic_id, topic_date, topic_body)
SELECT topic_id, topic_date, topic_body
FROM ${TARGET_SCHEMA_IMPORT}.tmp_topic src
WHERE NOT EXISTS (
        SELECT *
        FROM ${TARGET_SCHEMA_FINAL}.topic nx
        WHERE nx.topic_id = src.topic_id
        );
OMG3
}
########

delete_below_watermark()
{
watermark=${1-0}

echo ${watermark}

${PSQL} -h ${SOURCE_HOST} -U postgres ${SOURCE_DB} <<OMG4

-- delete not yet activated; COUNT(*) instead
-- DELETE
SELECT COUNT(*)
FROM ${SOURCE_SCHEMA}.topic WHERE topic_id <= ${watermark}
        ;
OMG4
}
######## Main

#create_local_stuff

watermark="`find_highest`"
echo 'Highest:' ${watermark}

fetch_new_data ${watermark}
insert_new_data

echo 'Delete below:' ${watermark}
delete_below_watermark ${watermark}

# Eof

This is just an example. Some notes:

• I assume a non-decreasing serial PK for the table; in most cases it could also be a timestamp
• for simplicity, all the queries are run as user postgres, you might need to change this
• the watermark method will guarantee that only new records will be transmitted, minimising bandwidth usage
• the method is atomic, if the script crashes, nothing is lost
• only one table is fetched here, but you could add more
• because I'm paranoid, I us a different name for the staging table and put it into a separate schema
• The whole script does two queries on the remote machine (one for fetch one for delete); you could combine these.
• but there is only one script (executing from the local=target machine) involved.
• The DELETE is not yet active; it only does a count(*)