Redesign applySchemaDeltas to only apply added columns to catalog

[Amogh-Bharadwaj]

Why

Let's say you have a Postgres mirror running and it is in CDC.
The pull batch size is 1.
On source for a table t1, the following operations are performed at the same time:

ALTER TABLE t1 ADD COLUMN good_column TIMESTAMP DEFAULT CURRENT_TIMESTAMP;
INSERT INTO t1 DEFAULT VALUES;
ALTER TABLE t1 ADD COLUMN lost_column TIMESTAMP DEFAULT CURRENT_TIMESTAMP;

The way PeerDB syncs the above operations is:

1. Receives the relation message for the addition of good_column and the INSERT.
   1.1) Checks the schema of this table as stored in catalog, sees that good_column is not present and marks good_column as a delta to be synced.
2. Does not receive the column addition for lost_column because there was no subsequent DML (Postgres behaviour).
3. Syncs the INSERT to destination and adds good_column to destination.
4. Performs applySchemaDelta which gets the schema of the table t1 from Postgres and stores this schema in catalog. Note that lost_column will be stored here as it is part of the source table at this time.

Now, when another insert is performed:

INSERT INTO t1 DEFAULT VALUES;

PeerDB now:

1. Receives the relation message for the addition of lost_column and the INSERT.
   1.1) Checks the schema of this table as stored in catalog, sees that lost_column is present and moves on.
2. Errors out when inserting to target tables with some message resembling column "lost_column" not found in destination.

Because we never identified lost_column as a delta to be added to destination.

What

This PR changes the implementation of applySchemaDeltas where we do not touch Postgres, and instead simply do:

updated columns for t1 in catalog = current columns of t1 in catalog + added columns in delta of this batch

This way, when the second INSERT comes along in the example above, PeerDB will see that lost_column is not present in catalog and will add it to the list of schema deltas to be synced.

Note: Refactors the table OID migration code so that we can share a common function to update table schemas in catalog.

• E2E tests pending
• Functional reproduction pending

[jgao54]

@Amogh-Bharadwaj love this change! -- i.e. rely on our states (of schema) as base instead of fetching from the source db to apply schema delta, since many things can result in schema change between SyncRecords runs.

[jgao54]

did a quick code search and i see a few other places that are using options.tableMapping (i.e. latest schema). wondering if we have any other logic that is relying on latest schema when it should be relying on catalog schema and if they introduce any subtle edge cases like this one

[Amogh-Bharadwaj]

Good call out, will take a look

[jgao54]

For MigratePostgresTableOIDs:

	// MIGRATION: Migrate Postgres table OIDs to catalog before starting/resuming the flow
	migrateCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
		StartToCloseTimeout: 1 * time.Hour,
		HeartbeatTimeout:    2 * time.Minute,
	})

iiuc, on pause-restart, we also sync what's in the source db to our catalog. so there may be a chance here too that if there are schema changes between pause/resume, we would end up syncing catalog to the latest schema, and ApplySchemaDelta could miss column additions.

Im wondering what is the motivation here for always syncing schema to latest on pause/restart, vs. just letting CDC do the catch-up and ApplySchemaDelta.

Note this doesn't block your PR, just an observation and maybe a follow-up item.

[Amogh-Bharadwaj]

we also sync what's in the source db to our catalog

Actually this activity doesn't touch Postgres, it moves the OIDs from the state object to catalog and the OIDs in state are populated during initial setup flow and can be extended by setup flows of table additions, but is independent of schema changes so we should be good?

P.S: This migration is what enables table cancellation addition (added recently), it can be removed after a certain period of time