Materialize: Directly consuming a Postgres logical replication stream

[petrosagg]

Overview

The general idea is to provide users with a direct way to combine materialize and postgres that doesn't require having a Kafka and Debezium instance in the middle. The intent is to greatly simplify both the UX and the implementation for this usecase.

When logical replication is up and running, the upstream server (the publisher) sends decoded WAL entries to one or more downstream servers (the subscribers). Postgres includes a built-in decoder which is used for its own replication (and we should also use in materialize) called pgoutput.

The simplest setup that will sync the whole database is just two SQL statements on the respective machines:

-- On the publisher
CREATE PUBLICATION mypub FOR ALL TABLES;
-- On the subscriber
CREATE SUBSCRIPTION mysub CONNECTION 'dbname=foo host=bar user=repuser' PUBLICATION mypub;

Publisher's side

As far as the publisher goes, there are two relevant objects that track the state of the replication:

1. A PUBLICATION, which is more or less a list of tables that are intended to be replicated. This list can be a subset of the available tables but in that case care must be taken to not end up with dangling foreign keys in the replica.
2. A replication slot. This object is created by the subscriber as a side-effect of running a CREATE SUBSCRIPTION statement on it. When this object is created a snapshot is taken (in the MVCC sense) and the publisher will not recycle WAL entries until the subscriber verifies that it has received up to a particular LSN. It's a bit like a cursor object.

Subscriber's side

On the subscriber's side CREATE SUBSCRIPTION lifts a lot of weight. It handles the creation of the slots, reports back consumed LSNs, and, more importantly, handles the initial data synchronization if the tables are not empty.

Logical replication after everything is synced is fairly straight forward. Data is received transaction by transaction in commit order from a replication slot and the subscriber just applies them. The question is how the initial data synchronization is done. Let's look at the implementation of CREATE SUBSCRIPTION that will inform the implementation on materialize.

Initial data copy

When a subscription if first created on the subscriber, it creates a replication slot on the publisher and immediately starts to read it. When a change affects a table that has not had its initial data synced yet, it is simply dropped. It is crucial to not leave the slot untouched until all the tables are synced because that would cause an ever-growing WAL log in the upstream server.

With the main slot up and running the subscriber will move on to do sync the existing data of each table. This process is done independently and in parallel for each table in a separate worker. This is the procedure for a particular table:

-- Initial sync phase
BEGIN READ ONLY ISOLATION LEVEL REPEATABLE READ
CREATE_REPLICATION_SLOT <foo> LOGICAL pgoutput USE_SNAPSHOT;
COPY `table` TO STDOUT;
COMMIT

-- Catchup with main sync worker phase
START_REPLICATION SLOT <foo> 

What happens above is that a transaction is started with its first statement being CREATE_REPLICATION_SLOT with the USE_SNAPSHOT option. This has the effect of creating a replication slot and then setting the snapshot under which the current transaction runs to be that exact starting point. This means that COPY will get everything in the table up until the point where we can start consuming the slot. Once the COPY is done the initial sync worker goes into a catchup phase where it starts consuming the temporary table specific slot to get any changes that happened during the COPY when its stream catches up with the main slot created for the whole subscription the initial sync worker hands off the table and exits.

The above is repeated for all the tables in a publication

Implementation in materialize

The good news is that everything related to logical replication, including initial data sync, can be done with regular SQL commands and without any special plugins loaded into the upstream server. One detail here is that we'd have to speak the streaming replication protocol which is not yet supported by tokio-postgres but a PR was put up recently.

Regarding the implementation of the initial data sync we could copy what postgres is doing which would provide the minimum WAL hold up on the publisher and the fastest sync since everything is happening in parallel, but it's also the most complex. It is probably a good idea to implement a simplified initial sync protocol (described below) in order to get this feature out and optimise it later.

Simplified initial sync protocol

For the simplified protocol we basically do the same procedure as the single table syncing above, but for all the tables. The downside is that we'll be copying tables serially and that the slot will hold up WAL entries for the entire duration of the COPY.

-- Initial sync
BEGIN READ ONLY ISOLATION LEVEL REPEATABLE READ
CREATE_REPLICATION_SLOT <foo> LOGICAL pgoutput USE_SNAPSHOT;

-- for every table in the publication
COPY `table_1` TO STDOUT;
COPY `table_2` TO STDOUT;
...
COPY `table_n` TO STDOUT;

COMMIT

-- Start streaming from the end of the COPY
START_REPLICATION SLOT <foo> 

Limitations

Schema

Logical replication does not transfer DDL statements to the subscriber database and it's up to the administrator to make sure that schema changes are applied correctly to both. In materialize we can get an initial schema dump when we first set up the source but it's unclear how to handle upstream schema changes. I believe this is an existing limitation with debezium as well.

Upstream backup restore

In the scenario where the upstream database restores a backup the replication slots will observe the LSN moving backwards. I think in this case materialize should re-sync everything from scratch, essentially also restoring from backup. In the initial implementation we can just detect it and put the source in an errored state to prevent strange results.

[rjnn]

Amazing writeup! I'm all for this simplified "read replica" deployment.

One note: I wasn't sure that REPEATABLE READ was sufficient - it wouldn't be theoretically because phantoms are allowed, but pg doesn't emit phantom reads so we're good.

[benesch]

This looks phenomenal to me!

One thing that jumps out here (I think you alluded to this on Slack, @petrosagg) is how this really doesn't fit with our 1 source = 1 relation model. A PostgreSQL subscription source is going to want to create multiple relations. Hrm.

[rjnn]

@benesch can we get a size estimate for implementing this? How long would it take to get a working version of direct-postgres-replication if we greenlit working on this?

[petrosagg]

@rjnn There is a PoC level implementation that I didn't mention in the write up but which I intend to experiment with this week which takes a lot of the hurdles of implementing the replication protocol out of the equation and will let us focus on correctness, timestamps, and how to lift the source <-> table 1-1 mapping.

Postgres provides a CLI utility called pg_recvlogical which connects to an upstream database, creates a replication slot, and starts streaming the data in stdout. I've looked at its source code and it seems easy to add the part where it COPYs the tables in the beginning.

This means I will have a program that given a postgres connection string streams into its stdout all the data of the database from the beginning of time and continues to do that indefinitely (it even supports reconnects).

[benesch]

I'd say at least a month of work to get something behind --experimental. @petrosagg what do you think?

[petrosagg]

That sounds reasonable but I will have a more informed opinion once I work on this some more this week.

[benesch]

In terms of syntax, the best idea I have is something like

CREATE MULTISOURCE materialize.public.pgrepl POSTGRES HOST 'localhost:5432' PUBLICATION pub

where that will automatically create a bunch of additional sources named after the tables in the publication. And those sources are locked to the existence of pgrepl and can't be individually dropped or otherwise modified.

[petrosagg]

If we add a new keyword we could just have the same CREATE SUBSCRIPTION
syntax as postgres.

Also, what would break if this mechanism created actual TABLE objects (like
the ones from CREATE TABLE) and put the data into those? Are sources
special in some way? I know they are not materialized by default, but in
this case we want to materialize everything. Is there some other desirable
property that we'd lose?

On Wed, Jan 20, 2021, 22:35 Nikhil Benesch [e-mail] wrote:

In terms of syntax, the best idea I have is something like

CREATE MULTISOURCE materialize.public.pgrepl POSTGRES HOST 'localhost:5432' PUBLICATION pub

where that will automatically create a bunch of additional sources named
after the tables in the publication. And those sources are locked to the
existence of pgrepl and can't be individually dropped or otherwise
modified.

—
You are receiving this because you were mentioned.
Reply to this email directly, view it on GitHub
https://github.com/MaterializeInc/materialize/issues/5370#issuecomment-763963954,
or unsubscribe
https://github.com/notifications/unsubscribe-auth/AAHFLHGGTCIPHOV4A6ZPYODS25EDVANCNFSM4WISSWAQ
.

[benesch]

So ideally whatever syntax we come up with will generalize nicely to other SQL databases too. The problem with adopting CREATE SUBSCRIPTION verbatim is that it's a bit too PostgreSQL specific.

The difference between a source and a table, besides the fact that sources can be unmaterialized, are essentially:

• tables permit INSERT
• reads and writes on tables are linearizable

And I think we probably want to prevent people from INSERTing into these tables!

[petrosagg]

Fair point on syntax.

tables permit INSERT

There are other ways of preventing INSERT that we could use. Choosing a different object type just for that sounds like a big hammer. For example when creating a subscription the tables could be inserted in a special schema that the users only have read-only access permissions for.

reads and writes on tables are linearizable

That's something that we want, right? Postgres is also linearizable (in single node or synchronous replication setups).

But more generally, a TABLE is closer to what a user would expect to see after syncing an upstream database, as this is what happens with normal replication.

[benesch]

Fair point on syntax.

tables permit INSERT

There are other ways of preventing INSERT that we could use. Choosing a different object type just for that sounds like a big hammer. For example when creating a subscription the tables could be inserted in a special schema that the users only have read-only access permissions for.

reads and writes on tables are linearizable

That's something that we want, right? Postgres is also linearizable (in single node or synchronous replication setups).

Yep, absolutely, but the linearizability support for tables only works because Materialize controls the timestamps at which data is inserted. We don't have a way to propagate timestamp information from an upstream source into Materialize, but it's something we want to support in general. There's some discussion on this point in #4660. The point is that someday sources will also be linearizable provided that upstream sources are linearizable. @mjibson is working on this presently.

But more generally, a TABLE is closer to what a user would expect to see after syncing an upstream database, as this is what happens with normal replication.

We've just got to do some user education on this point, I'm afraid. In Materialize, a TABLE is an object whose data is manipulable directly in Materialize, whereas a SOURCE is an object whose data is managed elsewhere and imported in Materialize. The PostgreSQL "tables" that get replicated into Materialize are logically "sources" in Materialize, because they represent data that comes from elsewhere.

[rjnn]

I think I agree with @benesch that it makes most sense for these to be SOURCES, not TABLES, from a conceptual sense.

[petrosagg]

Status update ##

Last week I cleaned up the WIP replication PR on rust-postgres and added support for logical replication and parsing of pgoutput's messages. The code is in this branch and I'll coordinate with the original author to open a new PR upstream soon.

Based on the above I wrote a small program that given a postgres connection string and a publication name, it will:

1. Discover all the tables that are referenced by the publication
2. Discover the schema of involved tables
3. Create a logical replication slot
4. Copy all the data from the beginning of time up until the replication creation LSN
5. Atomically switch to logical replication for all future changes

The code for that turned out to be relatively simple, you can check out this gist.

Next steps

The next step towards a PoC is to figure out how to integrate the output of the above program with materialize sources.

[AntonioL]

Sorry to bump into this thread as an newcomer. This sounds super intriguing and the approach looks to be sane, super excited about that!

I have one question though. I do not quite understand the need for the COPY. Could you expand on that @petrosagg ?

Per the postgresql documentation on logical replication (https://www.postgresql.org/docs/12/logical-replication-architecture.html)

"Logical replication starts by copying a snapshot of the data on the publisher database. Once that is done, changes on the publisher are sent to the subscriber as they occur in real time. The subscriber applies data in the order in which commits were made on the publisher so that transactional consistency is guaranteed for the publications within any single subscription."

If I understand it well you seem to imply that we need to copy so that you get a copy of the current content of the database but that is not needed as it is already given to you once you start consuming the replication stream.

[petrosagg]

Heh, I thought so as well when I first read the documentation :) The wording makes it imply that you're going to get the initial snapshot as part of the replication stream but this is actually not the case. Further down it mentions:

The initial data in existing subscribed tables are snapshotted and copied in a parallel instance of a special kind of apply process. This process will create its own temporary replication slot and copy the existing data.

This special process uses a COPY command for each table, I've outlined how postgres does it with links to the source code in my initial comment (see "Initial data copy" section).

[benesch]

Some rough notes from our meeting just now. The proposal is to support a new POSTGRES source type with the following syntax:

CREATE [MATERIALIZED] SOURCE src FROM POSTGRES 'host:port' PUBLICATION pub TABLE tbl

The semantics of this source would be equivalent to: subscribe to the named publication on the specified PostgreSQL server, filtering out events that are not for tbl. tbl must obviously exist as part of the pub. This gives maximum expressive power to the user using the catalog commands they're already familiar with. If they want to unsubscribe from that table directly, they can just DROP SOURCE. If they want to rename the source, they can use ALTER RENAME.

There are two problems with this syntax:

• It is annoying to have to type CREATE SOURCE once for every table in the publication.
• The naive implementation requires that Materialize subscribe to the PostgreSQL instance one _per_ source instance, which is going to be terrible for performance.

To solve the first problem, we'll support an additional CREATE SOURCES shorthand:

CREATE SOURCES FROM POSTGRES 'host:port' PUBLICATION pub
[TABLES (table_spec, ...)]

table_spec:

upstream_name [AS downstream_name] [MATERIALIZED | UNMATERIALIZED]

This will create one unmaterialized source for each table in pub by default, using some sensible mapping from PostgreSQL table names to Materialize table names. The TABLES clause allows filtering the tables included to some subset, renaming those tables into different locations in Materialize's catalog, and marking the created sources as either materialized or unmaterialized.

Then there is the problem of efficiency. The ultimate goal here is to build some sort of source deduplication framework, so that typing

CREATE MATERIALIZED SOURCE src1 FROM POSTGRES 'host:port' PUBLICATION pub TABLE tbl1
-- wait 2m
CREATE MATERIALIZED SOURCE src2 FROM POSTGRES 'host:port' PUBLICATION pub TABLE tbl2

ultimately results in just one connection to the PostgreSQL server, except for the brief catch up window required when src2 is created. Note that this problem is more general than PostgreSQL. Multiple Kafka sources that refer to the same Kafka topic have the same problem.

No one knows how to build that full framework for deduplicating sources, so we instead propose that we special case CREATE SOURCES in the short term. It will not be allowed to type CREATE SOURCE ... FROM POSTGRES at first; instead, you must go through CREATE SOURCES so that Materialize is apprised of all sources in a publication in one shot.

Note that there is a halfway solution between only supporting CREATE SOURCES and supporting full source deduplication via transactional DDL. We could plausibly support something like

BEGIN
CREATE SOURCE src1 FROM POSTGRES 'host:port' PUBLICATION pub TABLE tbl1
CREATE SOURCE src2 FROM POSTGRES 'host:port' PUBLICATION pub TABLE tbl2
CREATE MATERIALIZED VIEW v AS SELECT * FROM src1, src2 WHERE <cond>
COMMIT

well before we are capable of supporting a full source deduplication framework.

[benesch]

So, @petrosagg, as we were talking about, putting this in the catalog is going to be a bit tricky. The catalog must not be dependent on external state for its own integrity, so we need to actually write down the sources that were created. In other words, we can't just write down the CREATE SOURCES command, since that depends on the list of tables present in PostgreSQL, but instead need to write down the individual CREATE SOURCE commands.

Complicating the matters is the fact that the only way to store a catalog item is by writing down the SQL that defines it, i.e., the items table in the catalog looks like this:

(1, CREATE SOURCE ...)
(3, CREATE TABLE ...)
(2, CREATE VIEW ...)

So, we'll need to support planning the CREATE SOURCE ... FROM POSTGRES command, even though we don't actually want anyone to use it.

I think the way to do this will be as follows:

• Support CREATE SOURCE src FROM POSTGRES HOST 'host:port' pub TABLE tbl2 (col1 ty1, col2 ty2). Note the addition of the data about the types in the table. This will the the result of "purifying" a CREATE SOURCE command as it requires communication with PostgreSQL. Don't tell anyone about this syntax, but technically leave it accessible. We do this already with undocumented schema registry "seeds", if you look through the parser code.
• Don't support purifying CREATE SOURCE yet. Only support batch purifying CREATE SOURCES. That will essentially force everyone to use CREATE SOURCES for the time being.
• Replumb how SQL purification and planning work so that it is possible to purify/plan a batch of sources at once, to support the last point.
• Rework catalog boot so that we "recover" the CREATE SOURCES commands that were run, basically by just taking all the PostgreSQL sources that we see and grouping them by (host, publication).

All that means there will be a point in the coordinator where we have a batch of PostgreSQL sources that all refer to the same host/publication, and we can submit that to the dataflow layer as we like.

[awang]

See https://github.com/MaterializeInc/materialize/issues/5971 for a description of what we intend to ship in 0.8