Can psycopg2 copy_expert read from an io.StringIO() buffer?

Hi all,

Bit of explanation first...

I'm working on a script to "heal" a logically replicated database after
a replication outage. We get these outages periodically, especially to
the several we have hosted in RDS. Replication either stops outright or
slows so much it might as well be stopped. When that happens, the only
fix I've found is to drop the subscription and then start it back up
with (copy_data = False).

Once started again replication proceeds from that point on at full
speed. However, we now have a nice sized whole in the data that needs
to be backfilled.

I've been doing this by joining the publisher and subscriber with
postgres_fdw and comparing tables. Left joins between publisher and
subscriber will show me the rows on the publisher that are not on the
subscriber, and then flip the tables to find rows on the subscriber that
are not on the publisher and need to be deleted. To find rows that need
updated are found with select * from publisher.table except select *
from subscriber.table.

I had been doing this in a plpy function both reading and writing to the
fdw tables. That works but is excruciatingly slow--one table we have
has 170,000,000 rows and takes about 3 hours to sync. A 325G database
takes about 6.5 hours total.

Enter psycopg2. My plan now is to only query the fdw tables to identify
the rows, then use psycopg2 connections to both databases to directly
insert/delete/update (by deleting from and then inserting to) the
subscriber table.

Working on inserts now, with this code (queries freshly ported over from
the plpy function):

subscriber_connection.set_session(autocommit=True)
csv_buf = io.StringIO()
size = size = 3**20 #3GB

copy_query = """
copy (
select a.* from %s.%s a
join %s.pkeys b on a.%s = b.%s
order by a.%s
) to stdout with csv
""" % (schema,table_name,schema,pkey,pkey_column,pkey_column)

insert_query = """
copy %s.%s from stdin with csv;
""" % (schema,table_name)

try:
publisher_copy_cursor.copy_expert(copy_query,csv_buf,size)
subscriber_copy_cursor.copy_expert(insert_query,csv_buf)
subscriber_copy_cursor.close()
except Exception as e:
debug_output("insert error! %s" % (e))

So, that most of that works pretty well. csv_buf is filled with pkeys
of the rows missing from the subscriber. I've verified that works from
the python shell.

On the subscriber I see the copy command hit the logs:

2022-11-03 14:27:15.357
EDT,"postgres","dr_metroarchive",43810,"172.26.27.10:39614",636407fc.ab22,30,"COPY",2022-11-03
14:27:08 EDT,4/0,0,LOG,00000,"duration: 53.782 ms statement: copy
metro.client_profile from stdin with csv;",,,,,,,,,""

But no rows ever actually get inserted.

The only thing I can think of so far is that copy_expert isn't reading
csv_buf. And now that I write that I wonder if it's because csv_buf is
in memory on the publisher and not on the subscriber.

If that's the case maybe this all boils down to "how do I pipe stdout
from the publisher to the subscriber across the subscriber cursor?"

That's a lot of email for one short question--apologies and thanks in
advance for any clue by fours!

Jeff Ross