Hoe we CDC naar Postgres hebben gepusht — en replicatie in een geoliede machine veranderden

Het beschikbaar maken van gegevens uit transactionele databases voor analytische databases is een essentieel onderdeel van elke moderne data-architectuur. Het is echter ook een voortdurende strijd tegen fragiele tooling, hoge kosten en complexe operaties. Toen we bij Snowflake begonnen met het bouwen van een Postgres-service, werd het oplossen van dit probleem natuurlijk onze belangrijkste prioriteit.

Dit artikel is een diepgaande analyse van de engineering achter data mirroring: hoe we Postgres-replicatie vanaf de grond opnieuw hebben bedacht.

Het optimaliseren van Postgres-replicatie

Postgres is een geweldige operationele database, maar het verhaal rondom change data capture (CDC) laat nog te wensen over. Veel pipelines blijken fragiel omdat replicatietools belast worden met het afhandelen van de complexe interactie tussen continue gegevens en schemawijzigingen, snapshots en storingen. Om een betrouwbare, kant-en-klare ervaring voor Snowflake Postgres te bouwen, moesten we Postgres-replicatie vanaf de grond opnieuw uitvinden.

Data mirroring is een nieuwe functie van Snowflake Postgres (momenteel in public preview) om zeer veerkrachtige datareplicatie naar Snowflake uit te voeren met lage kosten, minimale vertraging (lag) en transactionele consistentie. Onder de motorkap werkt het door wijzigingen direct vanuit Postgres in Apache Iceberg™-tabellen te pushen, in transactionele batches. Deze batches worden automatisch — transactioneel en serverless — toegepast op tabellen in Snowflake.

De eenvoud van "transactionele push naar het data lake, transactionele toepassing in Snowflake, geen extra infrastructuur" verandert replicatie van een chaotisch proces met veel complexe foutcondities in een eenvoudige, geoliede machine die eeuwig blijft draaien. Met één druk op de knop heb je je Postgres-tabellen in Snowflake.

Van pull naar push: Change Data Capture verplaatsen naar Postgres

Change data capture is het proces waarbij wijzigingen uit een transactionele database worden vastgelegd in een vorm die toestaat dat ze op een ander systeem opnieuw kunnen worden afgespeeld.

In Postgres is de primaire faciliteit hiervoor "logical decoding", wat verwijst naar het decoderen van WAL-records (Write Ahead Log) naar logische insert/update/delete-operaties op rij-niveau. Deze operaties worden als een stream over het netwerk aangeboden. Vanaf dat punt ligt de verantwoordelijkheid bij de client.

In de praktijk omvat replicatie veel meer stappen:

• Backfilling;
• Schemawijzigingen;
• Het afhandelen van create/add/remove/drop table-operaties;
• Nieuwe tabel-snapshots;
• Herstarten na een storing;
• Efficiënt samenvoegen van wijzigingen;
• Behoud van transactiegrenzen;
• Right-sizing en meer.

Zelfs de ingebouwde logische replicatie in Postgres handelt slechts enkele van deze aspecten af. Een van de problemen met de logical decoding-aanpak is dat het externe systeem dat de wijzigingen consumeert, niets weet over de status van Postgres. Het weet bijvoorbeeld niet wanneer schemawijzigingen plaatsvinden, hoe tabel-snapshots aansluiten op wijzigingen, of zelfs of Postgres nog actief is of dat het netwerk is uitgevallen.

De oplossing voor dit probleem is simpel: push de wijzigingen vanuit Postgres naar een data lake, en in ons geval naar Iceberg-tabellen (met gecomprimeerde Parquet). Object stores zoals Amazon S3 zijn zeer schaalbaar en betrouwbaar, en worden al constant gebruikt voor Postgres-backups. Het is daarom ook de juiste bestemming voor change data capture.

Mirroring maakt gebruik van een nieuwe Postgres-extensie genaamd snowflake_cdc. Deze pusht continu batches met wijzigingen naar per-tabel wijzigingslogs en een "meta log" op de achtergrond (via "base workers"). Het voordeel van het gebruik van een extensie is dat deze precies weet wat er in Postgres gebeurt. De extensie kan schemawijzigingen en complexe data manipulation language (DML) en data definition language (DDL) transacties zorgvuldig coördineren. Daarnaast kan hij snapshots maken terwijl hij wijzigingen pusht en de snapshots afstemmen op die wijzigingen.

Push-gebaseerde CDC vermijdt een hele klasse aan infrastructuurproblemen en ontkoppelt de producent en consument effectief via object storage.

Replicatie-tijdlijnen ontwarren

Bij het bouwen van een replicatiesysteem zoals data mirroring is de tijdlijn van de database een cruciaal aspect. Replicatieprocessen gaan om met de status van de database in het recente verleden.

Elke schrijfactie naar Postgres doorloopt effectief vier fasen, die elk een continu proces vertegenwoordigen dat op dezelfde tijdlijn opereert, maar op een ander moment:

1. Write (Schrijven): De schrijfactie wijzigt de tabel en voegt gegevens toe aan de WAL (in het "nu").
2. Decode (Decoderen): Vorige WAL-gegevens worden vertaald naar wijzigingen op rij-niveau.
3. Capture (Vastleggen): Vorige wijzigingen op rij-niveau worden in batches vastgelegd.
4. Apply (Toepassen): Vorige batchwijzigingen worden samengevoegd in de doeltabellen.

Het decoderingsproces vertrouwt op een speciale faciliteit in Postgres om catalogustabellen te lezen zoals ze waren op het moment van schrijven (een "historische snapshot"). Op die manier kunnen binaire WAL-records worden begrepen als logische rijwijzigingen, zelfs als een tabel op het moment van decoderen al is gewijzigd of verwijderd. In het geval van data mirroring gaan deze records naar tijdelijke bestanden.

Periodiek krijgt de decoder een signaal om de huidige batch te finaliseren en een bericht naar het capture-proces te sturen dat een batch klaarstaat. Het capture-proces voegt de gefinaliseerde bestanden toe aan de Iceberg-wijzigingslogs, schrijft een record in de meta log en houdt de gerepliceerde LSN (Log Sequence Number) bij. Schemawijzigingen volgen hetzelfde pad (write $\to$ decode $\to$ capture) en kunnen leiden tot nieuwe wijzigingslogs.

De nieuwe records in de meta log en wijzigingslog verschijnen in de Iceberg-tabellen. Het apply-proces in Snowflake fungeert vervolgens als een finite state machine die alle instructies in de meta log uitvoert. Wanneer de operatie een wijzigingsbatch is (het meest voorkomende geval), worden alle aangrenzende batches voor elke tabel samen verwerkt.

Deze aanpak zorgt ervoor dat schemawijzigingen correct worden gesequenced in de stroom van wijzigingen, zelfs als ze deel uitmaakten van een transactie die ook andere schrijfacties bevatte. Als er een onverwachte storing optreedt waardoor de WAL verloren gaat, kan Postgres automatisch nieuwe snapshots pushen en Snowflake instrueren deze te consumeren (wat in de praktijk zelden voorkomt dankzij het gebruik van failover slots).

Transacties als bouwsteen voor gedistribueerde systemen

Database-systemen kunnen een enorme hoeveelheid complexiteit met betrekking tot systeem- en hardwarestoringen verbergen via één simpele primitieve: transacties. Als een transactie faalt door een storing op een lager niveau, gebeurt er niets en kun je het opnieuw proberen. Als een transactie slaagt, kun je er zeker van zijn dat hetzelfde werk nooit nogmaals wordt uitgevoerd.

De frustratie die we voelen bij processen zoals ETL (extract, transform, load) of CDC komt doordat transacties plotseling wegvallen, waardoor we alle verschillende foutmodi zelf moeten afhandelen.

Ons eerste antwoord op dit probleem was Postgres for your data lake, de beheerde versie van onze open-source pg_lake-extensie. Deze geeft Postgres de unieke mogelijkheid om transacties uit te voeren over zowel Postgres-tabellen als Iceberg-tabellen. Waar ETL normaal gesproken externe tools vereist waarbij gebruikers moeten ontwerpen voor idempotentie en zeer nauwkeurige administratie moeten bijhouden, kun je nu simpelweg SQL gebruiken om gegevens uit een Postgres-tabel te verwijderen, in een Iceberg-tabel te plaatsen en te committen. Op dat moment zijn de gegevens querybaar in Snowflake.

Deze aanpak is veelzijdig, maar SQL leent zich niet voor end-to-end replicatie van updates met een hoge frequentie. Dat is de laag die data mirroring toevoegt. Onder de motorkap maakt data mirroring volledig gebruik van pg_lake en de Iceberg-implementatie van Snowflake. Het neemt batches met gegevens en schemawijzigingen uit Postgres-tabellen en pusht deze in meerdere wijzigingslogs in Iceberg binnen één transactie aan de Postgres-zijde. Snowflake voegt vervolgens meerdere batches tegelijk samen in een transactie aan de Snowflake-zijde. Dit betekent dat alle Snowflake-tabellen in één transactie worden bijgewerkt, precies tot een Postgres-transactiegrens, waardoor foreign keys en join-correctheid behouden blijven.

Deze transactionele replicatieaanpak schaalt naar zeer hoge doorvoersnelheden en vermijdt de meeste gebruikelijke storingen en race conditions van conventionele cross-systeem replicatie.

Hoge prestaties bij toepassing en live views

Transacties maken ook iets anders mogelijk: correctheid op schaal. Een veelgebruikte aanpak voor replicatie is om elke operatie om te zetten in een soort upsert. De reden hiervoor is dat het zeer moeilijk is om het consistentieprobleem tussen tabel-snapshots en wijzigingen op te lossen, evenals wijzigingen die na een storing meerdere keren kunnen worden afgespeeld.

De upsert-aanpak heeft echter verschillende nadelen:

• Er kunnen inconsistente tussenstaten ontstaan op de bestemming, vooral wanneer een tabel wordt toegevoegd.
• Inserts worden zeer duur om te repliceren omdat ze moeten worden afgezet tegen bestaande rijen in de doeltabel, wat kostbaar is bij kolomgebaseerde opslag (columnar storage).
• Het wordt zeer moeilijk of onmogelijk om recente wijzigzingen efficiënt te combineren met de doeltabel.

Omdat we replicatie maken tot een transactioneel proces dat vanuit Postgres wordt beheerd, hebben we deze beperkingen niet. We kunnen een stroom van perfecte deleties en inserties creëren die exact één keer worden toegepast, zonder het risico dat een insert al in een snapshot of een andere wijzigingsbatch is verschenen.

Dit betekent concreet:

• Replicatie van workloads met veel inserts (meestal de grootste tabellen) is extreem snel en kostenefficiënt, omdat inserts worden toegevoegd (appended) en nooit als upsert worden verwerkt.
• Recente wijzigingen kunnen efficiënt worden gecombineerd met de bestaande gegevens.

Dit brengt ons bij live views. Live views zijn een functie van data mirroring die niet-toegepaste wijzigingen in de per-tabel wijzigingslogs combineert met de gegevens in de doeltabellen. Het belangrijkste is dat filters en projecties in de query direct kunnen worden doorgegeven naar de opslaglaag en tabelscans op zowel de Parquet-bestanden in de wijzigingslog als de basistabel. Met andere woorden: live views zijn snel.

Door live views is het niet langer nodig om wijzigingen zeer frequent toe te passen om een lage vertraging (lag) te behouden. Zelfs bij infrequent toepassen blijft de vertraging van de live view ruim onder de minuut, terwijl je nog steeds hoogwaardige queries krijgt met slechts een minimale overhead.

Replicatie als een geoliede machine

De combinatie van push-gebaseerde CDC, zorgvuldig ontwerp rond tijdlijnen, transactionele grenzen aan beide zijden en live views zorgt ervoor dat mirroring replicatie verandert van een chaotisch proces in een Zwitsers uurwerk. Er zijn geen externe connectoren die achter kunnen lopen. Geen snapshots die kunnen conflicteren met wijzigingen. Geen upserts die vertragen naarmate tabellen groeien. Er is een Postgres-extensie die batches naar object storage pusht, en Snowflake die deze toepast — beide transactioneel, beide onafhankelijk, beide onbeperkt.

Je stelt het één keer in, en het draait voor altijd.

Data mirroring voor Snowflake Postgres

Historisch gezien leefden operationele en analytische workloads in verschillende werelden. Die werelden werden vaak aan elkaar geknoopt met fragiele pipelines en extra systemen die de kosten verhoogden en zorgden voor veel frustratie. Ons doel is om deze werelden te verenigen met solide fundamenten aan beide zijden via eenvoudige, maar goed ontworpen oplossingen.

Met Snowflake Postgres heb je een productie-waardige Postgres-omgeving waar je op kunt vertrouwen, en nu de flexibiliteit van twee manieren om je workloads te verenigen:

• Data mirroring: De public preview biedt altijd aanwezige, automatische replicatie van Postgres naar Snowflake. Je stelt het één keer in en je tabellen, inclusief schemawijzigingen, blijven continu synchroon.
• Postgres for your data lake: Nu algemeen beschikbaar; dit biedt flexibele, door de ontwikkelaar beheerde databewegingen tussen Postgres en Snowflake met behulp van open formaten zoals Iceberg. Je schrijft SQL, en data verplaatst zich wanneer en hoe jij dat wilt.