Our Journey to PostgreSQL 12 (opens in new tab)

When I was a kid and learning to program, I wrote some shitty databases for fun. I learned about the trade-offs and that it was easy to write a database that out-performed RDBMSes in specific criteria. But I hadn't thought of making them extensible.

I have a pet project I'm working on, which is a generi distributed system where each component is a microservice. It turns out there's lots of these things built already, mainly by systems engineers for obscure things (Airflow, Rundeck, Stackstorm being some examples). I'll probably think about how I can redesign my project with composeable databases in mind. I don't expect I'll ever have a working product, but it'll be useful to think about this problem.

You're basically telling them to put in months of work to find out. Even if it's not too snarky, it's a ridiculous way to learn something that could be conveyed pretty well in a blog post or a chapter of a book.

Actually not really. A Unix operating system can do everything I described with regular-old data, and it's not a distributed operating system. It simply has extensible standard interfaces.

Do you need a distributed database to read a .txt file with cat, Emacs, and Firefox? No. Why not? Because there's an I/O standard they all use. Does that .txt file have to live on a single filesystem, or disk? No. Why not? Because the storage mediums all have a standard virtual filesystem interface.

There is no reason databases cannot do exactly the same thing. It's just that nobody has made them do it yet. They've stuck with the exact same paradigm, and that then drives how all of us build our systems, with this archaic 40 year old model that requires heavy-lifting maintenance windows and a lot of praying.

> This is not a feature of SQL, this is a feature of the database

Yet they always seem tied together eh? Somehow the conventions are stuck together, and that then affects how our systems work.

> Postgres supports both Perl and Python extensions that run in the RDBMS process

But I'm talking about not having to use the RDBMS process. If I have a text file on the disk, I can use a million different programs to access it. I don't have to run one program through another program just to open, read, write, and close the file with any program. Why don't we design our databases to work this way?

> Very few people use them because running in the RDBMS process means that you can break the RDBMS process in really bad ways

Yes, it does sound bad. That's why I'd prefer an indirect method rather than having to wedge access through the RDBMS

> So if you consider other processes that communicate with the database and then show views of that over other protocols, that describes most of the backend apps in the world.

Yep! We architect entire systems-of-systems just because the model for our data management in an RDBMS is too rigid. We're building spaceships to get to the grocery store because we haven't yet figured out roads and cars.

Would you mind mentioning some good options? I've always been interested in databases, but find it hard to know which ones to learn more about and when they'd actually be worth investing in (especially since it's hard to build knowledge from toy projects).

5.7 TB for an OLTP database is small?! I must be living in a different world. Obviously I know you can go that big, but I thought the number of use-cases would be limited.

its huge by database standard, i worked in large multinationals and dealt with some of the their largest databases

5.7 is enormous by database standard , there is no way you can get good query performance on a 5.7 tb database without solid physical partitioning and heavily optimized queries, and most normal companies even with 200-500 GB database use datamarts to have good performance without a super complex architectures and geniuses working fro you in db admin department

the more i think about it, the more i think that 5.7 TB would be unusably huge, and if you have this much data, most wont even bother to partition, the db will be broken into several (hundreds) smaller databases

And you have to account for indexes, temporary tables used for data analysis, etc. And most of it is probably not compressed. So with that perspective it isn't that much data at all.

That's an incredibly large amount per user? I have worked on a couple online dating sites, including one that was fairly popular (Let's Date - which stiffed me for my last invoice before they went belly up grrr). Unless you're storing images in the database, it's really hard to generate 600k for a dating workload - even with indexes.

The only thing I can imagine generating 600k per user is putting something like "hit tracking" in the database. Which I've done - yes it adds up - but it's also relatively easy to move to some other kind of store.

That's really a good way of looking at it. I though it sounded like a lot of data, well, 600kb is a lot of textual data, but who knows what they have stuffed into the database.

I worked for an e-commerce site, with a few million customers, even more orders, data-duplication all over the place, and still we where using a perhaps a 200GB of database storage.

What was the strategy for moving things over? By age? Monitoring queries and determining what data isn’t being queried? Something else?

Nope.

hi! Can you share how you do HA on postgres? Master/slave with monitoring and manual fall over or is that automatic? If so reliable? What tooling do you use? Thanks!

Fine, with enough replicas, you can sleep well at night. But how about the 3 years uptime without reboot? Can you really enjoy your morning coffee without thinking about it? :)

Netflix went full ephemeral storage for their Cassandra clusters since the beginning, at the time when they were just spinning disks. Years later, they still insist on doing this, and had to come up with creative solution to fix the uptime issue: https://netflixtechblog.medium.com/datastore-flash-upgrades-...

Even with prioritized IOPS I once had to resort to RAID0 and replicas to get needed performance under budget on EBS. Probably should have just bumped instance size and used local storage.

I believe it is possible to repoint the replicas with re-wind and then repoint to a new timeline. This is something we looked at at Heroku a long time back. It wasn't trivial to fix, but Heroku eventually improved some of this. This post drills into some of that - https://blog.keikooda.net/2017/10/18/battle-with-a-phantom-w...

I've done a few postgres upgrades, the first using pg_upgrade, and the last doing effectively what you did (it was even 9.x -> 12).

My experience was that it needs to rewrite all the data for some tables/indexes under some circumstances, and the db won't be available while that happens. So, unless your db can be down for the time it takes to rewrite all that data it isn't really an option. After having done the upgrade through streaming logical replication I'm not sure I would try pg_upgrade again.

I did the pg_upgrade style update a long time ago, so most details are fuzzy, but I remember setting up a string of replicas something like:

primary -> [read_replica, backups_replica]

read_replica -> [upgrade_replica]

upgrade_replica -> [read_replica_upgraded, backups_replica_upgraded]

This allowed us to do multiple practice rounds without putting any unnecessary load on the primary. I think we needed to re-initiate replication off 'upgrade_replica' after the upgrade, but we did the live update during low-load so the extra read load wasn't an issue.

In pg_upgrade documentation there is documented a way to use rsync to quickly replicate the upgraded contents of the new primary to replicas. So you would first move to upgraded base VMs running the old version streaming from old primary, which can be done one host at a time if need be.

The new cluster can then be pg_upgraded and rsynced all at once.

Good questions. As logical replication matures it may someday be possible to replicate among versions.

CockroachDB is pretty good at encapsulating the complexity of multi-master.

You'll have to accept that transactions can fail due to conflicts, so if they are interactive, you'll have to retry manually.

Edit: I'd like hear criticism, instead of just seeing disapproval.

What kind of data is this?

My best guesses are

- either it is SAAS in which case shard it should make sense to shard by customer

- or it is something-to-consumer (social networking?) on which case I guess you'll have to take a step back and see if you can sacrifice one of your current assumptions

... but I feel I'm missing something since what I am saying feels a bit trivial.

Sharding a matching engine is indeed pretty hard, and requires redundancy and very deliberate data modelling choices.

That does seem like a fun exercise :).

The general rules of the game are: You can only scale up throughput of queries/transactions that only access 1 shard (some percentage going to 2 shards can be ok). You can only scale down response time of large operations that span across shard since they are parallelized. You should only join distributed tables on their distribution column. You can join reference tables on any column.

The thing that comes to mind is to use a reference table for any user data that is used to find/score matches. Reference tables are replicated to every node and can be joined with distributed tables and each other using arbitrary join clauses, so joining by score or distance is not a problem, but you need to store the data multiple times.

One of the immediate benefits of reference tables is that reads can be load-balanced across the nodes, either by using a setting (citus.task_assignment_policy = 'round-robin') or using a distributed table as a routing/parallelization scheme.

    CREATE TABLE profiles (
        user_id bigint primary key,
        location geometry,
        profile jsonb
    );
    SELECT create_reference_table('profiles');
    
    CREATE TABLE users (
        user_id bigint primary key references profiles (user_id),
        name text,
        email text
    );
    SELECT create_distributed_table('users', 'user_id');
    
    -- replicate match_score function to all the nodes
    SELECT create_distributed_function('match_score(jsonb,jsonb)');
    
    -- look up profile of user 350, goes to 1 shard
    SELECT * FROM users u, profiles p WHERE u.user_id = p.user_id AND u.user_id = 350;
    
    -- find matches for user #240 within 5km, goes to 1 shard
    SELECT b.user_id, match_score(a.profile, b.profile) AS score
    FROM users u, profiles a, profiles b
    WHERE u.user_id = 240 AND u.user_id = a.user_id 
    AND match_score(a.profile,b.profile) > 0.9 AND st_distance(a.location,b.location) < 5000 
    ORDER BY score DESC LIMIT 10;

The advantage of having the distributed users table in the join is mainly that you divide the work in a way that keeps each worker node's cache relatively hot for a specific subset of users, though you'll still be scanning most of the data to find matches.

Where it gets a bit more interesting is if your dating site is opinionated / does not let you search, since you can then generate matches upfront in batches in parallel.

    CREATE TABLE match_candidates (
        user_id_a bigint references profiles (user_id),
        user_id_b bigint references profiles (user_id),
        score float,
        primary key (user_id_a, user_id_b)
    );
    SELECT create_distributed_table('match_candidates', 'user_id_a', colocate_with :='users');
    
    -- generate match candidates for all users in a distributed, parallel fashion
    -- will generate a match candidate in both directions, assuming score is commutative
    INSERT INTO match_candidates
    SELECT a.user_id, b.user_id, match_score(a.profile,b.profile) AS score
    FROM users u, profiles a, profiles b
    WHERE u.user_id = a.user_id 
    AND match_score(a.profile,b.profile) > 0.9 AND st_distance(a.location,b.location) < 5000 
    ORDER BY score DESC LIMIT 10;

For interests/matches, it might make sense to have some redundancy in order to achieve reads that go to 1 shard as much possible.

    CREATE TABLE interests (
        user_id_a bigint references profiles (user_id),
        user_id_b bigint references profiles (user_id),
        initiated_by_a bool,
        mutual bool,
        primary key (user_id_a, user_id_b)
    );
    SELECT create_distributed_table('interests', 'user_id_a', colocate_with :='users');
    
    -- 240 is interested in 350, insert into 2 shards (uses 2PC)
    BEGIN;
    INSERT INTO interests VALUES (240, 350, true, false);
    INSERT INTO interests VALUES (350, 240, false, false);
    END;
    
    -- people interested in #350, goes to 1 shard
    SELECT * FROM interests JOIN profiles ON (user_id_b = user_id) WHERE user_id_a = 350 AND NOT initiated_by_a;
    
    -- it's a match! update 2 shards (uses 2PC)
    BEGIN;
    UPDATE interests SET mutual = true WHERE user_id_a = 240 AND user_id_b = 350;
    UPDATE interests SET mutual = true WHERE user_id_a = 350 AND user_id_b = 240;
    END;
    
    -- people #240 is matched with, goes to 1 shard
    SELECT * FROM interests JOIN profiles ON (user_id_b = user_id) WHERE user_id_a = 240 AND mutual;

For data related to a specific match, you can perhaps use the smallest user ID as the distribution column to avoid the redundancy.

    CREATE TABLE messages (
        user_id_a bigint,
        user_id_b bigint,
        from_a bool,
        message_text text,
        message_time timestamptz default now(),
        message_id bigserial,
        primary key (user_id_a, user_id_b, message_id),
        foreign key (user_id_a, user_id_b) references interests (user_id_a, user_id_b) on delete cascade
    );
    SELECT create_distributed_table('messages', 'user_id_a', colocate_with :='interests');

    -- user 350 sends a message to 240, goes to 1 shard
    INSERT INTO messages VALUES (240, 350, false, 'hi #240!');
    
    -- user 240 sends a message to 350, goes to 1 shard
    INSERT INTO messages VALUES (240, 350, true, 'hi!');
    
    -- user 240 looks at chat with user 350, goes to 1 shard
    SELECT from_a, message_text, message_time
    FROM messages 
    WHERE user_id_a = 240 AND user_id_b = 350
    ORDER BY message_time DESC LIMIT 100;  

This exercise goes on for a while. You still get the benefits of PostgreSQL and ability to scale up throughput of common operations or scale down response time of batch operations, but it does require careful data model choices.

(Citus engineer who enjoys distributed systems puzzles)

It's not in the post, but I answered this in a separate thread. RDS doesn't let us provision as many IOPS as we need.

Apparently Aurora behaves differently, but I wasn't aware of that when we specced out the project.

Did you consider lowering those IOPS with application-level and/or distributed in-memory cache and/or pub-sub notifications to let your app nodes not pester the database so much? Reasonably performant hand-written SQL (no ORM!), review of query plans, maybe shift the hot path into functions/procs?

That's an impressive amount of IOPS. I have always been a EBS / pd-ssd guy, and mostly rely on memory to reduce the IOPS requirement. But as cloud providers typically charge a ridiculous amount of money for memory, a setup like yours with instance storage / local-ssd is an intriguing option.

That limit doesn’t apply to Aurora - did you consider that?

So.......... caching?

Disclaimer: I work at AWS, not on EC2.

Locally attached disks are not ephemeral to instance reboots/power failures. However, the disks are wiped after instance terminations. On the official EC2 product pages this is called "instance storage" not "ephemeral storage."

still, changes between 9.6 and 12 are _numerous_, both in features and performance: llvm based query compilation, CTE de-materialisation, proper procedures, and that's just off the top of my head.

i wish the process for upgrading postgres were easier/more dynamic. i'm sure plenty of people are still using versions 9.6 or earlier.

That's still a lot of trust that nothing else wipes the account.

But this post wasn't about backups, so there might be a whole lot excluded from the diagram.